Add current device consumption to InputConsumerNoResampling

Added current device consumption to InputConsumerNoResampling in
consumeBatchedInputEvents' body

Bug: 329776327
Flag: EXEMPT refactor
Test: TEST=libinput_tests; m $TEST && $ANDROID_HOST_OUT/nativetest64/$TEST/$TEST --gtest_filter="InputConsumerTest*"
Change-Id: Ibfcb28f019543a88b3ffada137199b1c3933d542

1 files changed, 33 insertions, 20 deletions

diff --git a/libs/input/InputConsumerNoResampling.cpp b/libs/input/InputConsumerNoResampling.cpp
index de55828a39..f33afe75fc 100644
--- a/libs/input/InputConsumerNoResampling.cpp
+++ b/libs/input/InputConsumerNoResampling.cpp
@@ -227,7 +227,7 @@ InputConsumerNoResampling::InputConsumerNoResampling(const std::shared_ptr<Input
 
 InputConsumerNoResampling::~InputConsumerNoResampling() {
     ensureCalledOnLooperThread(__func__);
-    consumeBatchedInputEvents(std::nullopt);
+    consumeBatchedInputEvents(/*requestedFrameTime=*/std::nullopt);
     while (!mOutboundQueue.empty()) {
         processOutboundEvents();
         // This is our last chance to ack the events. If we don't ack them here, we will get an ANR,
@@ -253,8 +253,7 @@ int InputConsumerNoResampling::handleReceiveCallback(int events) {
 
     int handledEvents = 0;
     if (events & ALOOPER_EVENT_INPUT) {
-        std::vector<InputMessage> messages = readAllMessages();
-        handleMessages(std::move(messages));
+        handleMessages(readAllMessages());
         handledEvents |= ALOOPER_EVENT_INPUT;
     }
 
@@ -362,10 +361,8 @@ void InputConsumerNoResampling::handleMessages(std::vector<InputMessage>&& messa
                 // add it to batch
                 mBatches[deviceId].emplace(msg);
             } else {
-                // consume all pending batches for this event immediately
-                // TODO(b/329776327): figure out if this could be smarter by limiting the
-                // consumption only to the current device.
-                consumeBatchedInputEvents(std::nullopt);
+                // consume all pending batches for this device immediately
+                consumeBatchedInputEvents(deviceId, /*requestedFrameTime=*/std::nullopt);
                 handleMessage(msg);
             }
         } else {
@@ -483,13 +480,13 @@ void InputConsumerNoResampling::handleMessage(const InputMessage& msg) const {
 }
 
 std::pair<std::unique_ptr<MotionEvent>, std::optional<uint32_t>>
-InputConsumerNoResampling::createBatchedMotionEvent(const nsecs_t frameTime,
+InputConsumerNoResampling::createBatchedMotionEvent(const nsecs_t requestedFrameTime,
                                                     std::queue<InputMessage>& messages) {
     std::unique_ptr<MotionEvent> motionEvent;
     std::optional<uint32_t> firstSeqForBatch;
     const nanoseconds resampleLatency =
             (mResampler != nullptr) ? mResampler->getResampleLatency() : nanoseconds{0};
-    const nanoseconds adjustedFrameTime = nanoseconds{frameTime} - resampleLatency;
+    const nanoseconds adjustedFrameTime = nanoseconds{requestedFrameTime} - resampleLatency;
 
     while (!messages.empty() &&
            (messages.front().body.motion.eventTime <= adjustedFrameTime.count())) {
@@ -511,36 +508,52 @@ InputConsumerNoResampling::createBatchedMotionEvent(const nsecs_t frameTime,
         if (!messages.empty()) {
             futureSample = &messages.front();
         }
-        mResampler->resampleMotionEvent(nanoseconds{frameTime}, *motionEvent, futureSample);
+        mResampler->resampleMotionEvent(nanoseconds{requestedFrameTime}, *motionEvent,
+                                        futureSample);
     }
     return std::make_pair(std::move(motionEvent), firstSeqForBatch);
 }
 
 bool InputConsumerNoResampling::consumeBatchedInputEvents(
-        std::optional<nsecs_t> requestedFrameTime) {
+        std::optional<DeviceId> deviceId, std::optional<nsecs_t> requestedFrameTime) {
     ensureCalledOnLooperThread(__func__);
     // When batching is not enabled, we want to consume all events. That's equivalent to having an
-    // infinite frameTime.
-    const nsecs_t frameTime = requestedFrameTime.value_or(std::numeric_limits<nsecs_t>::max());
+    // infinite requestedFrameTime.
+    requestedFrameTime = requestedFrameTime.value_or(std::numeric_limits<nsecs_t>::max());
     bool producedEvents = false;
-    for (auto& [_, messages] : mBatches) {
-        auto [motion, firstSeqForBatch] = createBatchedMotionEvent(frameTime, messages);
+
+    for (auto deviceIdIter = (deviceId.has_value()) ? (mBatches.find(*deviceId))
+                                                    : (mBatches.begin());
+         deviceIdIter != mBatches.cend(); ++deviceIdIter) {
+        std::queue<InputMessage>& messages = deviceIdIter->second;
+        auto [motion, firstSeqForBatch] = createBatchedMotionEvent(*requestedFrameTime, messages);
         if (motion != nullptr) {
             LOG_ALWAYS_FATAL_IF(!firstSeqForBatch.has_value());
             mCallbacks.onMotionEvent(std::move(motion), *firstSeqForBatch);
             producedEvents = true;
         } else {
-            // This is OK, it just means that the frameTime is too old (all events that we have
-            // pending are in the future of the frametime). Maybe print a
-            // warning? If there are multiple devices active though, this might be normal and can
-            // just be ignored, unless none of them resulted in any consumption (in that case, this
-            // function would already return "false" so we could just leave it up to the caller).
+            // This is OK, it just means that the requestedFrameTime is too old (all events that we
+            // have pending are in the future of the requestedFrameTime). Maybe print a warning? If
+            // there are multiple devices active though, this might be normal and can just be
+            // ignored, unless none of them resulted in any consumption (in that case, this function
+            // would already return "false" so we could just leave it up to the caller).
+        }
+
+        if (deviceId.has_value()) {
+            // We already consumed events for this device. Break here to prevent iterating over the
+            // other devices.
+            break;
         }
     }
     std::erase_if(mBatches, [](const auto& pair) { return pair.second.empty(); });
     return producedEvents;
 }
 
+bool InputConsumerNoResampling::consumeBatchedInputEvents(
+        std::optional<nsecs_t> requestedFrameTime) {
+    return consumeBatchedInputEvents(/*deviceId=*/std::nullopt, requestedFrameTime);
+}
+
 void InputConsumerNoResampling::ensureCalledOnLooperThread(const char* func) const {
     sp<Looper> callingThreadLooper = Looper::getForThread();
     if (callingThreadLooper != mLooper->getLooper()) {