[XProto] Move RequestQueue into Connection

This CL just moves some things around and has no
behavior changes.

1. Move response/event processing from X11EventSource to
   x11::Connection.  This is done because X11EventSource
   can be null in some tests, but we always have an
   x11::Connection.  Needed to fix tests for [1].  Also
   wanted for [2] which begins migrating events from
   Xlib to XProto.
2. Move some logic from x11::Future into x11::FutureBase.
   This is done to avoid including ui/gfx/x/connection.h
   from xproto_types.h which would otherwise add a circular
   include.

[1] https://chromium-review.googlesource.com/c/chromium/src/+/2220686
[2] https://chromium-review.googlesource.com/c/chromium/src/+/2220635

BUG=1066670
R=msisov
TBR=​sky

Change-Id: Ia0710595c6cb6337a8c9b52f31ceaafefaf00cbc
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2220524Reviewed-by: Thomas Anderson <thomasanderson@chromium.org>
Reviewed-by: Maksim Sisov <msisov@igalia.com>
Commit-Queue: Thomas Anderson <thomasanderson@chromium.org>
Cr-Commit-Position: refs/heads/master@{#773867}

ui/events/platform/x11/x11_event_source.cc

@@ -8,6 +8,7 @@
 #include <xcb/xcb.h>
 #include <xcb/xcbext.h>
 
+#include <memory>
 #include <type_traits>
 
 #include "base/logging.h"
@@ -39,27 +40,6 @@ namespace ui {
 
 namespace {
 
-// On the wire, sequence IDs are 16 bits.  In xcb, they're usually extended to
-// 32 and sometimes 64 bits.  In Xlib, they're extended to unsigned long, which
-// may be 32 or 64 bits depending on the platform.  This function is intended to
-// prevent bugs caused by comparing two differently sized sequences.  Also
-// handles rollover.  To use, compare the result of this function with 0.  For
-// example, to compare seq1 <= seq2, use CompareSequenceIds(seq1, seq2) <= 0.
-template <typename T, typename U>
-auto CompareSequenceIds(T t, U u) {
-  static_assert(std::is_unsigned<T>::value, "");
-  static_assert(std::is_unsigned<U>::value, "");
-  // Cast to the smaller of the two types so that comparisons will always work.
-  // If we casted to the larger type, then the smaller type will be zero-padded
-  // and may incorrectly compare less than the other value.
-  using SmallerType =
-      typename std::conditional<sizeof(T) <= sizeof(U), T, U>::type;
-  SmallerType t0 = static_cast<SmallerType>(t);
-  SmallerType u0 = static_cast<SmallerType>(u);
-  using SignedType = typename std::make_signed<SmallerType>::type;
-  return static_cast<SignedType>(t0 - u0);
-}
-
 bool InitializeXkb(XDisplay* display) {
   if (!display)
     return false;
@@ -129,18 +109,6 @@ x11::Bool IsPropertyNotifyForTimestamp(Display* display,
 
 }  // namespace
 
-X11EventSource::Request::Request(bool is_void,
-                                 unsigned int sequence,
-                                 ResponseCallback callback)
-    : is_void(is_void), sequence(sequence), callback(std::move(callback)) {}
-
-X11EventSource::Request::Request(Request&& other)
-    : is_void(other.is_void),
-      sequence(other.sequence),
-      callback(std::move(other.callback)) {}
-
-X11EventSource::Request::~Request() = default;
-
 #if defined(USE_GLIB)
 using X11EventWatcherImpl = X11EventWatcherGlib;
 #else
@@ -154,7 +122,6 @@ X11EventSource::X11EventSource(XDisplay* display)
       display_(display),
       dispatching_event_(nullptr),
       dummy_initialized_(false),
-      continue_stream_(true),
       distribution_(0, 999) {
   DCHECK(!instance_);
   instance_ = this;
@@ -187,59 +154,8 @@ X11EventSource* X11EventSource::GetInstance() {
 // X11EventSource, public
 
 void X11EventSource::DispatchXEvents() {
-  DCHECK(display_);
-
-  auto process_next_response = [&]() {
-    xcb_connection_t* connection = XGetXCBConnection(display_);
-    auto request = std::move(requests_.front());
-    requests_.pop();
-
-    void* raw_reply = nullptr;
-    xcb_generic_error_t* raw_error = nullptr;
-    xcb_poll_for_reply(connection, request.sequence, &raw_reply, &raw_error);
-    DCHECK(request.is_void || raw_reply || raw_error);
-
-    std::move(request.callback)
-        .Run(Reply{reinterpret_cast<uint8_t*>(raw_reply)}, Error{raw_error});
-  };
-
-  auto process_next_event = [&]() {
-    XEvent xevent;
-    XNextEvent(display_, &xevent);
-    ExtractCookieDataDispatchEvent(&xevent);
-  };
-
-  // Handle all pending events.
   continue_stream_ = true;
-  while (continue_stream_) {
-    bool has_next_response =
-        !requests_.empty() &&
-        CompareSequenceIds(XLastKnownRequestProcessed(display_),
-                           requests_.front().sequence) >= 0;
-    bool has_next_event = XPending(display_);
-
-    if (has_next_response && has_next_event) {
-      auto next_response_sequence = requests_.front().sequence;
-
-      XEvent event;
-      XPeekEvent(display_, &event);
-      auto next_event_sequence = event.xany.serial;
-
-      // All events have the sequence number of the last processed request
-      // included in them.  So if a reply and an event have the same sequence,
-      // the reply must have been received first.
-      if (CompareSequenceIds(next_event_sequence, next_response_sequence) <= 0)
-        process_next_response();
-      else
-        process_next_event();
-    } else if (has_next_response) {
-      process_next_response();
-    } else if (has_next_event) {
-      process_next_event();
-    } else {
-      break;
-    }
-  }
+  x11::Connection::Get()->Dispatch(this);
 }
 
 void X11EventSource::DispatchXEventNow(XEvent* event) {
@@ -254,8 +170,8 @@ Time X11EventSource::GetCurrentServerTime() {
     dummy_window_ = XCreateSimpleWindow(display_, DefaultRootWindow(display_),
                                         0, 0, 1, 1, 0, 0, 0);
     dummy_atom_ = gfx::GetAtom("CHROMIUM_TIMESTAMP");
-    dummy_window_events_.reset(
-        new XScopedEventSelector(dummy_window_, PropertyChangeMask));
+    dummy_window_events_ = std::make_unique<XScopedEventSelector>(
+        dummy_window_, PropertyChangeMask);
     dummy_initialized_ = true;
   }
 
@@ -526,13 +442,12 @@ void X11EventSource::OnDispatcherListChanged() {
   }
 }
 
-void X11EventSource::AddRequest(bool is_void,
-                                unsigned int sequence,
-                                ResponseCallback callback) {
-  DCHECK(requests_.empty() ||
-         CompareSequenceIds(requests_.back().sequence, sequence) < 0);
+bool X11EventSource::ShouldContinueStream() const {
+  return continue_stream_;
+}
 
-  requests_.emplace(is_void, sequence, std::move(callback));
+void X11EventSource::DispatchXEvent(XEvent* event) {
+  ExtractCookieDataDispatchEvent(event);
 }
 
 // ScopedXEventDispatcher implementation

ui/events/platform/x11/x11_event_source.h

@@ -8,7 +8,6 @@
 #include <stdint.h>
 
 #include <memory>
-#include <queue>
 #include <random>
 
 #include "base/auto_reset.h"
@@ -17,7 +16,6 @@
 #include "base/optional.h"
 #include "ui/events/events_export.h"
 #include "ui/events/platform/platform_event_source.h"
-#include "ui/gfx/x/request_queue.h"
 #include "ui/gfx/x/x11_types.h"
 
 using Time = unsigned long;
@@ -124,7 +122,7 @@ class EVENTS_EXPORT ScopedXEventDispatcher {
 // {Platform,X}EventDispatchers. Handles receiving, pre-process, translation
 // and post-processing of XEvents.
 class EVENTS_EXPORT X11EventSource : public PlatformEventSource,
-                                     x11::RequestQueue {
+                                     public x11::Connection::Delegate {
  public:
   explicit X11EventSource(XDisplay* display);
   ~X11EventSource() override;
@@ -198,16 +196,6 @@ class EVENTS_EXPORT X11EventSource : public PlatformEventSource,
  private:
   friend class ScopedXEventDispatcher;
 
-  struct Request {
-    Request(bool is_void, unsigned int sequence, ResponseCallback callback);
-    Request(Request&& other);
-    ~Request();
-
-    const bool is_void;
-    const unsigned int sequence;
-    ResponseCallback callback;
-  };
-
   // Tells XEventDispatchers, which can also have PlatformEventDispatchers, that
   // a translated event is going to be sent next, then dispatches the event and
   // notifies XEventDispatchers the event has been sent out and, most probably,
@@ -221,10 +209,9 @@ class EVENTS_EXPORT X11EventSource : public PlatformEventSource,
   void StopCurrentEventStream() override;
   void OnDispatcherListChanged() override;
 
-  // x11::RequestQueue
-  void AddRequest(bool is_void,
-                  unsigned int sequence,
-                  ResponseCallback callback) override;
+  // x11::Connection::Delegate:
+  bool ShouldContinueStream() const override;
+  void DispatchXEvent(XEvent* event) override;
 
   void RestoreOverridenXEventDispatcher();
 
@@ -262,8 +249,6 @@ class EVENTS_EXPORT X11EventSource : public PlatformEventSource,
   XEventDispatcher* overridden_dispatcher_ = nullptr;
   bool overridden_dispatcher_restored_ = false;
 
-  std::queue<Request> requests_;
-
   DISALLOW_COPY_AND_ASSIGN(X11EventSource);
 };
 

ui/gfx/x/BUILD.gn

@@ -103,8 +103,7 @@ component("xprotos") {
   sources = get_target_outputs(":gen_xprotos") + [
               "xproto_internal.h",
               "xproto_types.h",
-              "request_queue.h",
-              "request_queue.cc",
+              "xproto_types.cc",
               "xproto_util.h",
               "xproto_util.cc",
               "connection.h",

ui/gfx/x/connection.cc

@@ -18,6 +18,27 @@ namespace x11 {
 
 namespace {
 
+// On the wire, sequence IDs are 16 bits.  In xcb, they're usually extended to
+// 32 and sometimes 64 bits.  In Xlib, they're extended to unsigned long, which
+// may be 32 or 64 bits depending on the platform.  This function is intended to
+// prevent bugs caused by comparing two differently sized sequences.  Also
+// handles rollover.  To use, compare the result of this function with 0.  For
+// example, to compare seq1 <= seq2, use CompareSequenceIds(seq1, seq2) <= 0.
+template <typename T, typename U>
+auto CompareSequenceIds(T t, U u) {
+  static_assert(std::is_unsigned<T>::value, "");
+  static_assert(std::is_unsigned<U>::value, "");
+  // Cast to the smaller of the two types so that comparisons will always work.
+  // If we casted to the larger type, then the smaller type will be zero-padded
+  // and may incorrectly compare less than the other value.
+  using SmallerType =
+      typename std::conditional<sizeof(T) <= sizeof(U), T, U>::type;
+  SmallerType t0 = static_cast<SmallerType>(t);
+  SmallerType u0 = static_cast<SmallerType>(u);
+  using SignedType = typename std::make_signed<SmallerType>::type;
+  return static_cast<SignedType>(t0 - u0);
+}
+
 XDisplay* OpenNewXDisplay() {
   if (!XInitThreads())
     return nullptr;
@@ -67,4 +88,76 @@ xcb_connection_t* Connection::XcbConnection() {
   return XGetXCBConnection(display());
 }
 
+Connection::Request::Request(unsigned int sequence,
+                             FutureBase::ResponseCallback callback)
+    : sequence(sequence), callback(std::move(callback)) {}
+
+Connection::Request::Request(Request&& other)
+    : sequence(other.sequence), callback(std::move(other.callback)) {}
+
+Connection::Request::~Request() = default;
+
+void Connection::Dispatch(Delegate* delegate) {
+  DCHECK(display_);
+
+  auto process_next_response = [&]() {
+    xcb_connection_t* connection = XGetXCBConnection(display_);
+    auto request = std::move(requests_.front());
+    requests_.pop();
+
+    void* raw_reply = nullptr;
+    xcb_generic_error_t* raw_error = nullptr;
+    xcb_poll_for_reply(connection, request.sequence, &raw_reply, &raw_error);
+
+    std::move(request.callback)
+        .Run(FutureBase::RawReply{reinterpret_cast<uint8_t*>(raw_reply)},
+             FutureBase::RawError{raw_error});
+  };
+
+  auto process_next_event = [&]() {
+    XEvent xevent;
+    XNextEvent(display_, &xevent);
+    delegate->DispatchXEvent(&xevent);
+  };
+
+  // Handle all pending events.
+  while (delegate->ShouldContinueStream()) {
+    bool has_next_response =
+        !requests_.empty() &&
+        CompareSequenceIds(XLastKnownRequestProcessed(display_),
+                           requests_.front().sequence) >= 0;
+    bool has_next_event = XPending(display_);
+
+    if (has_next_response && has_next_event) {
+      auto next_response_sequence = requests_.front().sequence;
+
+      XEvent event;
+      XPeekEvent(display_, &event);
+      auto next_event_sequence = event.xany.serial;
+
+      // All events have the sequence number of the last processed request
+      // included in them.  So if a reply and an event have the same sequence,
+      // the reply must have been received first.
+      if (CompareSequenceIds(next_event_sequence, next_response_sequence) <= 0)
+        process_next_response();
+      else
+        process_next_event();
+    } else if (has_next_response) {
+      process_next_response();
+    } else if (has_next_event) {
+      process_next_event();
+    } else {
+      break;
+    }
+  }
+}
+
+void Connection::AddRequest(unsigned int sequence,
+                            FutureBase::ResponseCallback callback) {
+  DCHECK(requests_.empty() ||
+         CompareSequenceIds(requests_.back().sequence, sequence) < 0);
+
+  requests_.emplace(sequence, std::move(callback));
+}
+
 }  // namespace x11

ui/gfx/x/connection.h

@@ -5,6 +5,8 @@
 #ifndef UI_GFX_X_CONNECTION_H_
 #define UI_GFX_X_CONNECTION_H_
 
+#include <queue>
+
 #include "base/component_export.h"
 #include "ui/gfx/x/extension_manager.h"
 #include "ui/gfx/x/xproto.h"
@@ -15,6 +17,15 @@ namespace x11 {
 class COMPONENT_EXPORT(X11) Connection : public XProto,
                                          public ExtensionManager {
  public:
+  class Delegate {
+   public:
+    virtual bool ShouldContinueStream() const = 0;
+    virtual void DispatchXEvent(XEvent* event) = 0;
+
+   protected:
+    virtual ~Delegate() {}
+  };
+
   // Gets or creates the singeton connection.
   static Connection* Get();
 
@@ -35,10 +46,25 @@ class COMPONENT_EXPORT(X11) Connection : public XProto,
     return defualt_root_visual_;
   }
 
+  void Dispatch(Delegate* delegate);
+
  private:
+  friend class FutureBase;
+
+  struct Request {
+    Request(unsigned int sequence, FutureBase::ResponseCallback callback);
+    Request(Request&& other);
+    ~Request();
+
+    const unsigned int sequence;
+    FutureBase::ResponseCallback callback;
+  };
+
   explicit Connection(XDisplay* display);
   ~Connection();
 
+  void AddRequest(unsigned int sequence, FutureBase::ResponseCallback callback);
+
   XDisplay* const display_;
 
   uint32_t extended_max_request_length_ = 0;
@@ -47,6 +73,8 @@ class COMPONENT_EXPORT(X11) Connection : public XProto,
   const x11::Screen* default_screen_ = nullptr;
   const x11::Depth* default_root_depth_ = nullptr;
   const x11::VisualType* defualt_root_visual_ = nullptr;
+
+  std::queue<Request> requests_;
 };
 
 }  // namespace x11

ui/gfx/x/request_queue.cc

-// Copyright 2020 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 "ui/gfx/x/request_queue.h"
-
-#include "base/check_op.h"
-
-namespace x11 {
-
-// static
-RequestQueue* RequestQueue::instance_ = nullptr;
-
-RequestQueue::RequestQueue() {
-  DCHECK(!instance_);
-  instance_ = this;
-}
-
-RequestQueue::~RequestQueue() {
-  DCHECK_EQ(instance_, this);
-  instance_ = nullptr;
-}
-
-// static
-RequestQueue* RequestQueue::GetInstance() {
-  return instance_;
-}
-
-}  // namespace x11

ui/gfx/x/request_queue.h

-// Copyright 2020 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.
-
-#ifndef UI_GFX_X_REQUEST_QUEUE_H_
-#define UI_GFX_X_REQUEST_QUEUE_H_
-
-#include <xcb/xcb.h>
-
-#include <memory>
-
-#include "base/callback_forward.h"
-#include "base/component_export.h"
-#include "base/memory/free_deleter.h"
-
-namespace ui {
-class X11EventSource;
-}
-
-namespace x11 {
-
-// This interface allows //ui/gfx/x to call into //ui/events/platform/x11 which
-// is at a higher layer.  It should not be used by client code.
-class COMPONENT_EXPORT(X11) RequestQueue {
- private:
-  friend class ui::X11EventSource;
-  template <typename T>
-  friend class Future;
-
-  using Reply = std::unique_ptr<uint8_t, base::FreeDeleter>;
-  using Error = std::unique_ptr<xcb_generic_error_t, base::FreeDeleter>;
-  using ResponseCallback = base::OnceCallback<void(Reply reply, Error error)>;
-
-  RequestQueue();
-  virtual ~RequestQueue();
-
-  // Adds a request to the queue.  |is_void| indicates if a reply is generated
-  // for this request.  |sequence| is the ID of the request.  |callback| will
-  // be called upon request completion (or failure).
-  virtual void AddRequest(bool is_void,
-                          unsigned int sequence,
-                          ResponseCallback callback) = 0;
-
-  static RequestQueue* GetInstance();
-
-  static RequestQueue* instance_;
-};
-
-}  // namespace x11
-
-#endif  //  UI_GFX_X_REQUEST_QUEUE_H_

ui/gfx/x/xproto_internal.h

@@ -139,13 +139,12 @@ Future<Reply> SendRequest(x11::Connection* connection, WriteBuffer* buf) {
     return {nullptr, base::nullopt};
   }
 
-  XDisplay* display = connection->display();
-  xcb_connection_t* conn = XGetXCBConnection(display);
+  xcb_connection_t* conn = connection->XcbConnection();
   auto flags = XCB_REQUEST_CHECKED | XCB_REQUEST_RAW;
   auto sequence = xcb_send_request(conn, flags, &io[2], &xpr);
   if (xcb_connection_has_error(conn))
     return {nullptr, base::nullopt};
-  return {display, sequence};
+  return {connection, sequence};
 }
 
 // Helper function for xcbproto popcount.  Given an integral type, returns the

ui/gfx/x/xproto_types.cc

+// Copyright 2020 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 "ui/gfx/x/xproto_types.h"
+
+#include "ui/gfx/x/connection.h"
+
+namespace x11 {
+
+FutureBase::FutureBase(Connection* connection,
+                       base::Optional<unsigned int> sequence)
+    : connection_(connection), sequence_(sequence) {}
+
+// If a user-defined response-handler is not installed before this object goes
+// out of scope, a default response handler will be installed.  The default
+// handler throws away the reply and prints the error if there is one.
+FutureBase::~FutureBase() {
+  if (!sequence_)
+    return;
+
+  OnResponseImpl(base::BindOnce(
+      [](Connection* connection, RawReply reply, RawError error) {
+        if (!error)
+          return;
+
+        x11::LogErrorEventDescription(XErrorEvent({
+            .type = error->response_type,
+            .display = connection->display(),
+            .resourceid = error->resource_id,
+            .serial = error->full_sequence,
+            .error_code = error->error_code,
+            .request_code = error->major_code,
+            .minor_code = error->minor_code,
+        }));
+      },
+      connection_));
+}
+
+FutureBase::FutureBase(FutureBase&& future)
+    : connection_(future.connection_), sequence_(future.sequence_) {
+  future.connection_ = nullptr;
+  future.sequence_ = base::nullopt;
+}
+
+FutureBase& FutureBase::operator=(FutureBase&& future) {
+  connection_ = future.connection_;
+  sequence_ = future.sequence_;
+  future.connection_ = nullptr;
+  future.sequence_ = base::nullopt;
+  return *this;
+}
+
+void FutureBase::SyncImpl(Error** raw_error, uint8_t** raw_reply) {
+  if (!sequence_)
+    return;
+
+  *raw_reply = reinterpret_cast<uint8_t*>(
+      xcb_wait_for_reply(connection_->XcbConnection(), *sequence_, raw_error));
+  sequence_ = base::nullopt;
+}
+
+void FutureBase::OnResponseImpl(ResponseCallback callback) {
+  connection_->AddRequest(*sequence_, std::move(callback));
+}
+
+}  // namespace x11

ui/gfx/x/xproto_types.h

@@ -16,7 +16,6 @@
 #include "base/callback.h"
 #include "base/memory/free_deleter.h"
 #include "base/optional.h"
-#include "ui/gfx/x/request_queue.h"
 #include "ui/gfx/x/xproto_util.h"
 
 typedef struct _XDisplay XDisplay;
@@ -30,7 +29,7 @@ namespace detail {
 template <typename Reply>
 std::unique_ptr<Reply> ReadReply(const uint8_t* buffer);
 
-}
+}  // namespace detail
 
 using Error = xcb_generic_error_t;
 
@@ -70,66 +69,45 @@ struct Response<void> {
       : error(std::move(error)) {}
 };
 
+class COMPONENT_EXPORT(X11) FutureBase {
+ public:
+  using RawReply = std::unique_ptr<uint8_t, base::FreeDeleter>;
+  using RawError = std::unique_ptr<xcb_generic_error_t, base::FreeDeleter>;
+  using ResponseCallback =
+      base::OnceCallback<void(RawReply reply, RawError error)>;
+
+  FutureBase(const FutureBase&) = delete;
+  FutureBase& operator=(const FutureBase&) = delete;
+
+ protected:
+  FutureBase(Connection* connection, base::Optional<unsigned int> sequence);
+  ~FutureBase();
+
+  FutureBase(FutureBase&& future);
+  FutureBase& operator=(FutureBase&& future);
+
+  void SyncImpl(Error** raw_error, uint8_t** raw_reply);
+
+  void OnResponseImpl(ResponseCallback callback);
+
+ private:
+  Connection* connection_;
+  base::Optional<unsigned int> sequence_;
+};
+
 // An x11::Future wraps an asynchronous response from the X11 server.  The
 // response may be waited-for with Sync(), or asynchronously handled by
 // installing a response handler using OnResponse().
 template <typename Reply>
-class Future {
+class Future : public FutureBase {
  public:
   using Callback = base::OnceCallback<void(Response<Reply> response)>;
-  using RQ = RequestQueue;
-
-  // If a user-defined response-handler is not installed before this object goes
-  // out of scope, a default response handler will be installed.  The default
-  // handler throws away the reply and prints the error if there is one.
-  ~Future() {
-    if (!sequence_)
-      return;
-
-    EnqueueRequest(base::BindOnce(
-        [](XDisplay* display, RQ::Reply reply, RQ::Error error) {
-          if (!error)
-            return;
-
-          x11::LogErrorEventDescription(XErrorEvent({
-              .type = error->response_type,
-              .display = display,
-              .resourceid = error->resource_id,
-              .serial = error->full_sequence,
-              .error_code = error->error_code,
-              .request_code = error->major_code,
-              .minor_code = error->minor_code,
-          }));
-        },
-        display_));
-  }
-
-  Future(const Future&) = delete;
-  Future& operator=(const Future&) = delete;
-
-  Future(Future&& future)
-      : display_(future.display_), sequence_(future.sequence_) {
-    future.display_ = nullptr;
-    future.sequence_ = base::nullopt;
-  }
-  Future& operator=(Future&& future) {
-    display_ = future.display_;
-    sequence_ = future.sequence_;
-    future.display_ = nullptr;
-    future.sequence_ = base::nullopt;
-  }
-
-  xcb_connection_t* connection() { return XGetXCBConnection(display_); }
 
   // Blocks until we receive the response from the server. Returns the response.
   Response<Reply> Sync() {
-    if (!sequence_)
-      return {{}, {}};
-
     Error* raw_error = nullptr;
-    uint8_t* raw_reply = reinterpret_cast<uint8_t*>(
-        xcb_wait_for_reply(connection(), *sequence_, &raw_error));
-    sequence_ = base::nullopt;
+    uint8_t* raw_reply = nullptr;
+    SyncImpl(&raw_error, &raw_reply);
 
     std::unique_ptr<Reply> reply;
     if (raw_reply) {
@@ -146,47 +124,36 @@ class Future {
 
   // Installs |callback| to be run when the response is received.
   void OnResponse(Callback callback) {
-    if (!sequence_)
-      return;
-
     // This intermediate callback handles the conversion from |raw_reply| to a
     // real Reply object before feeding the result to |callback|.  This means
     // |callback| must be bound as the first argument of the intermediate
     // function.
-    auto wrapper = [](Callback callback, RQ::Reply raw_reply, RQ::Error error) {
+    auto wrapper = [](Callback callback, RawReply raw_reply, RawError error) {
       std::unique_ptr<Reply> reply =
           raw_reply ? detail::ReadReply<Reply>(raw_reply.get()) : nullptr;
       std::move(callback).Run({std::move(reply), std::move(error)});
     };
-    EnqueueRequest(base::BindOnce(wrapper, std::move(callback)));
-
-    sequence_ = base::nullopt;
+    OnResponseImpl(base::BindOnce(wrapper, std::move(callback)));
   }
 
  private:
   template <typename R>
   friend Future<R> SendRequest(Connection*, std::vector<uint8_t>*);
 
-  Future(XDisplay* display, base::Optional<unsigned int> sequence)
-      : display_(display), sequence_(sequence) {}
-
-  void EnqueueRequest(RQ::ResponseCallback callback) {
-    RQ::GetInstance()->AddRequest(std::is_void<Reply>::value, *sequence_,
-                                  std::move(callback));
-  }
-
-  XDisplay* display_;
-  base::Optional<unsigned int> sequence_;
+  Future(Connection* connection, base::Optional<unsigned int> sequence)
+      : FutureBase(connection, sequence) {}
 };
 
 // Sync() specialization for requests that don't generate replies.  The returned
 // response will only contain an error if there was one.
 template <>
 inline Response<void> Future<void>::Sync() {
-  if (!sequence_)
-    return Response<void>(nullptr);
+  Error* raw_error = nullptr;
+  uint8_t* raw_reply = nullptr;
+  SyncImpl(&raw_error, &raw_reply);
+
+  DCHECK(!raw_reply);
 
-  Error* raw_error = xcb_request_check(connection(), {*sequence_});
   std::unique_ptr<Error, base::FreeDeleter> error;
   if (raw_error)
     error.reset(raw_error);
@@ -198,18 +165,13 @@ inline Response<void> Future<void>::Sync() {
 // response argument to |callback| will only contain an error if there was one.
 template <>
 inline void Future<void>::OnResponse(Callback callback) {
-  if (!sequence_)
-    return;
-
   // See Future<Reply>::OnResponse() for an explanation of why
   // this wrapper is necessary.
-  auto wrapper = [](Callback callback, RQ::Reply reply, RQ::Error error) {
+  auto wrapper = [](Callback callback, RawReply reply, RawError error) {
     DCHECK(!reply);
     std::move(callback).Run(Response<void>{std::move(error)});
   };
-  EnqueueRequest(base::BindOnce(wrapper, std::move(callback)));
-
-  sequence_ = base::nullopt;
+  OnResponseImpl(base::BindOnce(wrapper, std::move(callback)));
 }
 
 }  // namespace x11