Provided batched input delivery with a BufferedInputRouter

content/browser/renderer_host/input/input_queue.cc

+// Copyright (c) 2013 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "content/browser/renderer_host/input/input_queue.h"
+
+#include "base/bind.h"
+#include "content/browser/renderer_host/input/event_ack_handler.h"
+#include "content/browser/renderer_host/input/input_queue_client.h"
+#include "content/common/input/event_packet.h"
+#include "content/common/input/input_event.h"
+
+namespace content {
+namespace {
+
+// TODO(jdduke):
+//   INPUT_EVENT_NEEDS_ACK_AND_HAS_FOLLOWUP: Delay flush, hold future packets.
+//   INPUT_EVENT_NEEDS_ACK: Signal flush immediately, dispatch ack in-order.
+//   INPUT_EVENT_ONE_WAY: Signal flush immediately, ignore ack.
+InputEventType RestrictingEventType(const EventPacket& packet) {
+  InputEventType restricting_type = INPUT_EVENT_INVALID;
+  for (std::vector<InputEvent>::const_iterator event_it = packet.events.begin();
+       event_it != packet.events.end();
+       ++event_it) {
+    if (event_it->type > restricting_type)
+      restricting_type = event_it->type;
+  }
+  return restricting_type;
+}
+
+bool HasFollowupEvents(const EventPacket& packet) {
+  return RestrictingEventType(packet) == INPUT_EVENT_NEEDS_ACK_AND_HAS_FOLLOWUP;
+}
+
+bool HasAckEvents(const EventPacket& packet) {
+  return RestrictingEventType(packet) >= INPUT_EVENT_NEEDS_ACK;
+}
+
+bool IsOneWay(const EventPacket& packet) {
+  return RestrictingEventType(packet) <= INPUT_EVENT_ONE_WAY;
+}
+
+}
+
+// Utilty class for storing input events and their ack handlers, and dispatching
+// input event responses.
+class InputQueue::EventFrame : EventAckHandler::EventInjector {
+ public:
+  EventFrame() : weak_factory_(this) {}
+  virtual ~EventFrame() {}
+
+  void DeliverTo(InputQueueClient* client, int64 id) {
+    packet_.id = id;
+    client->Deliver(packet_);
+  }
+
+  void QueueEvent(const InputEvent& event, EventAckHandler* handler) {
+    packet_.events.push_back(event);
+    handlers_.push_back(handler);
+  }
+
+  // Validate |acked_packet| and dispatch the acked events upon successful
+  // validation.  Returns |false| if the validation failed.
+  AckResult ValidateAndDispatchAcks(const EventPacket& acked_packet) {
+    if (!Validate(acked_packet))
+      return ACK_INVALID;
+
+    std::vector<InputEvent> events;
+    std::swap(events, packet_.events);
+
+    std::vector<EventAckHandler*> handlers;
+    std::swap(handlers, handlers_);
+
+    // The EventFrame could be deleted as a result of the event ack; use a local
+    // weak ref to ensure proper shutdown.
+    base::WeakPtr<EventFrame> weak_ref_this = weak_factory_.GetWeakPtr();
+
+    const std::vector<InputEvent>& acked_events = acked_packet.events;
+    for (size_t event_idx = 0; event_idx < acked_events.size(); ++event_idx) {
+      const InputEvent& acked_event = acked_events[event_idx];
+
+      // All events marked undeliverable should be re-enqueued.
+      switch (acked_event.state) {
+        case INPUT_EVENT_IMPL_THREAD_COULD_NOT_DELIVER:
+        case INPUT_EVENT_MAIN_THREAD_COULD_NOT_DELIVER:
+          QueueEvent(events[event_idx], handlers[event_idx]);
+          continue;
+        default:
+          break;
+      }
+
+      // NULL handlers are acceptable.
+      if (!handlers[event_idx])
+        continue;
+
+      // Update the original event's state, and use it for the ack.
+      InputEvent& event = events[event_idx];
+      event.state = acked_event.state;
+
+      switch (acked_events[event_idx].type) {
+        case INPUT_EVENT_NEEDS_ACK_AND_HAS_FOLLOWUP:
+          handlers[event_idx]->OnInputEventAck(event, this);
+          break;
+        case INPUT_EVENT_NEEDS_ACK:
+          handlers[event_idx]->OnInputEventAck(event);
+          break;
+        default:
+          break;
+      }
+
+      // Exit immediately if the ack handler triggered an early shutdown.
+      if (!weak_ref_this.get())
+        return ACK_SHUTDOWN;
+    }
+    return ACK_OK;
+  }
+
+  bool IsEmpty() const { return packet_.events.empty(); }
+  size_t Size() const { return packet_.events.size(); }
+
+protected:
+
+  // EventInjector
+  virtual void InjectEvent(const InputEvent& event,
+                           EventAckHandler* handler) OVERRIDE {
+    QueueEvent(event, handler);
+  }
+
+  // Perform a sanity check of |acked_packet| against the queued events.
+  // The packet should contain the same number of events, and all events and
+  // the event messages should be of the same general type.
+  bool Validate(const EventPacket& acked_packet) {
+    if (acked_packet.id != packet_.id)
+      return false;
+
+    const std::vector<InputEvent>& source_events = packet_.events;
+    const std::vector<InputEvent>& acked_events = acked_packet.events;
+    if (source_events.size() != acked_events.size())
+      return false;
+
+    for (size_t event_idx = 0; event_idx < acked_events.size(); ++event_idx) {
+      const InputEvent& source_event = source_events[event_idx];
+      const InputEvent& acked_event = acked_events[event_idx];
+
+      if (acked_event.type != source_event.type)
+        return false;
+
+      if (acked_event.message.type() != source_event.message.type())
+        return false;
+    }
+
+    return true;
+  }
+
+private:
+  EventPacket packet_;
+  std::vector<EventAckHandler*> handlers_;
+  base::WeakPtrFactory<EventFrame> weak_factory_;
+};
+
+InputQueue::InputQueue(InputQueueClient* client)
+    : client_(client),
+      last_packet_id_(0),
+      flush_requested_(false),
+      flush_signalled_(false),
+      current_frame_(new EventFrame()),
+      pending_frame_(new EventFrame()) {
+  DCHECK(client_);
+}
+
+InputQueue::~InputQueue() {}
+
+void InputQueue::QueueEvent(const InputEvent& event,
+                            EventAckHandler* ack_handler) {
+  pending_frame_->QueueEvent(event, ack_handler);
+  RequestFlushIfNecessary();
+}
+
+void InputQueue::FlushEventsInCurrentFrame() {
+  // Ignore repeated flush attempts.
+  if (!flush_requested_)
+    return;
+
+  if (IsFlushing()) {
+    NOTREACHED() << "InputQueue::Flush() - A flush is already in-progress.";
+    return;
+  }
+
+  flush_requested_ = false;
+  flush_signalled_ = false;
+  current_frame_.swap(pending_frame_);
+
+  TryFinishFlush();
+}
+
+InputQueue::AckResult InputQueue::OnEventPacketAck(
+    const EventPacket& acked_packet) {
+  if (!IsFlushing())
+    return ACK_UNEXPECTED;
+
+  AckResult ack_result = current_frame_->ValidateAndDispatchAcks(acked_packet);
+  if (ack_result == ACK_OK)
+    TryFinishFlush();
+
+  return ack_result;
+}
+
+void InputQueue::OnEventPacketAckDelayed() {
+  if (!IsFlushing())
+    return;
+
+  SignalFlushedIfNecessary();
+  // Don't request another flush until the ack is received.
+}
+
+size_t InputQueue::QueuedEventCount() const {
+  return current_frame_->Size() + pending_frame_->Size();
+}
+
+void InputQueue::TryFinishFlush() {
+  if (!IsFlushing()) {
+    SignalFlushedIfNecessary();
+    RequestFlushIfNecessary();
+    return;
+  }
+
+  // Provide a unique id for every delivered packet.
+  current_frame_->DeliverTo(client_, ++last_packet_id_);
+}
+
+void InputQueue::RequestFlushIfNecessary() {
+  if (flush_requested_)
+    return;
+
+  // No events to flush.
+  if (pending_frame_->IsEmpty())
+    return;
+
+  // We're still flushing; wait until it finishes before an additional request.
+  if (IsFlushing())
+    return;
+
+  flush_requested_ = true;
+  client_->SetNeedsFlush();
+}
+
+void InputQueue::SignalFlushedIfNecessary() {
+  if (flush_signalled_)
+    return;
+
+  flush_signalled_ = true;
+  client_->DidFlush();
+}
+
+bool InputQueue::IsFlushing() const {
+  return !current_frame_->IsEmpty();
+}
+
+} // namespace content