Partially remove dependency on folly async

Summary:
This change isolates the usage of folly async from Flipper. Is now self-contained in Flipper Folly schedulers.

Users of Flipper can decide not to use the types defined in that header and implement their own.

NOTE: changes are minimal, we are just replacing direct calls to folly event base with a scheduler which simply relays this on to folly.

Reviewed By: fabiomassimo

Differential Revision: D36626483

fbshipit-source-id: add0241caf4af0aa5c3b5c2e7efc2e725f5400ab

android/src/main/cpp/sonar.cpp

@@ -13,6 +13,7 @@
 #include <fb/fbjni.h>
 #endif
 
+#include <folly/futures/Future.h>
 #include <folly/io/async/AsyncSocketException.h>
 #include <folly/io/async/EventBase.h>
 #include <folly/io/async/EventBaseManager.h>
@@ -23,6 +24,7 @@
 #include <Flipper/FlipperClient.h>
 #include <Flipper/FlipperConnection.h>
 #include <Flipper/FlipperConnectionManager.h>
+#include <Flipper/FlipperFollyScheduler.h>
 #include <Flipper/FlipperResponder.h>
 #include <Flipper/FlipperSocket.h>
 #include <Flipper/FlipperSocketProvider.h>
@@ -42,6 +44,9 @@ void handleException(const std::exception& e) {
   __android_log_write(ANDROID_LOG_ERROR, "FLIPPER", message.c_str());
 }
 
+std::unique_ptr<facebook::flipper::Scheduler> sonarScheduler;
+std::unique_ptr<facebook::flipper::Scheduler> connectionScheduler;
+
 class JEventBase : public jni::HybridClass<JEventBase> {
  public:
   constexpr static auto kJavaDescriptor =
@@ -412,7 +417,7 @@ class JFlipperSocketProvider : public facebook::flipper::FlipperSocketProvider {
   virtual std::unique_ptr<facebook::flipper::FlipperSocket> create(
       facebook::flipper::FlipperConnectionEndpoint endpoint,
       std::unique_ptr<facebook::flipper::FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase) override {
+      facebook::flipper::Scheduler* scheduler) override {
     return std::make_unique<JFlipperWebSocket>(
         std::move(endpoint), std::move(payload));
     ;
@@ -420,7 +425,7 @@ class JFlipperSocketProvider : public facebook::flipper::FlipperSocketProvider {
   virtual std::unique_ptr<facebook::flipper::FlipperSocket> create(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase,
+      facebook::flipper::Scheduler* scheduler,
       ConnectionContextStore* connectionContextStore) override {
     return std::make_unique<JFlipperWebSocket>(
         std::move(endpoint), std::move(payload), connectionContextStore);
@@ -939,6 +944,11 @@ class JFlipperClient : public jni::HybridClass<JFlipperClient> {
       const std::string app,
       const std::string appId,
       const std::string privateAppDirectory) {
+    sonarScheduler =
+        std::make_unique<FollyScheduler>(callbackWorker->eventBase());
+    connectionScheduler =
+        std::make_unique<FollyScheduler>(connectionWorker->eventBase());
+
     FlipperClient::init(
         {{std::move(host),
           std::move(os),
@@ -947,8 +957,8 @@ class JFlipperClient : public jni::HybridClass<JFlipperClient> {
           std::move(app),
           std::move(appId),
           std::move(privateAppDirectory)},
-         callbackWorker->eventBase(),
+         sonarScheduler.get(),
-         connectionWorker->eventBase(),
+         connectionScheduler.get(),
          insecurePort,
          securePort,
          altInsecurePort,

iOS/FlipperKit/FlipperClient.mm

@@ -10,9 +10,8 @@
 #import "FlipperClient.h"
 #import <Flipper/FlipperCertificateProvider.h>
 #import <Flipper/FlipperClient.h>
+#import <Flipper/FlipperFollyScopedThreadScheduler.h>
 #import <Flipper/FlipperSocketProvider.h>
-#include <folly/io/async/EventBase.h>
-#include <folly/io/async/ScopedEventBaseThread.h>
 #include <memory>
 #import "FlipperClient+Testing.h"
 #import "FlipperCppWrapperPlugin.h"
@@ -32,8 +31,9 @@ using WrapperPlugin = facebook::flipper::FlipperCppWrapperPlugin;
 
 @implementation FlipperClient {
   facebook::flipper::FlipperClient* _cppClient;
-  folly::ScopedEventBaseThread sonarThread;
+  std::unique_ptr<facebook::flipper::Scheduler> sonarScheduler;
-  folly::ScopedEventBaseThread connectionThread;
+  std::unique_ptr<facebook::flipper::Scheduler> connectionScheduler;
+
   id<FlipperKitCertificateProvider> _certProvider;
 #if !TARGET_OS_OSX && !TARGET_OS_SIMULATOR && !TARGET_OS_MACCATALYST
   FKPortForwardingServer* _secureServer;
@@ -90,6 +90,10 @@ using WrapperPlugin = facebook::flipper::FlipperCppWrapperPlugin;
     deviceName = [[NSHost currentHost] localizedName];
 #endif
 
+    sonarScheduler =
+        std::make_unique<facebook::flipper::FollyScopedThreadScheduler>();
+    connectionScheduler =
+        std::make_unique<facebook::flipper::FollyScopedThreadScheduler>();
     static const std::string UNKNOWN = std::string("unknown");
     try {
       facebook::flipper::FlipperClient::init(
@@ -102,8 +106,8 @@ using WrapperPlugin = facebook::flipper::FlipperCppWrapperPlugin;
                [appId UTF8String] ?: UNKNOWN,
                [privateAppDirectory UTF8String],
            },
-           sonarThread.getEventBase(),
+           sonarScheduler.get(),
-           connectionThread.getEventBase(),
+           connectionScheduler.get(),
            [SKEnvironmentVariables getInsecurePort],
            [SKEnvironmentVariables getSecurePort],
            [SKEnvironmentVariables getAltInsecurePort],

iOS/FlipperKit/FlipperWebSocket.h

@@ -9,11 +9,11 @@
 
 #pragma once
 
+#import <Flipper/FlipperScheduler.h>
 #import <Flipper/FlipperSocket.h>
 #import <Flipper/FlipperSocketProvider.h>
 #import <Flipper/FlipperTransportTypes.h>
 #import <folly/dynamic.h>
-#import <folly/io/async/EventBase.h>
 #import <future>
 #import <memory>
 
@@ -67,14 +67,14 @@ class FlipperWebSocketProvider : public FlipperSocketProvider {
   virtual std::unique_ptr<FlipperSocket> create(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase) override {
+      Scheduler* scheduler) override {
     return std::make_unique<FlipperWebSocket>(
         std::move(endpoint), std::move(payload));
   }
   virtual std::unique_ptr<FlipperSocket> create(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase,
+      Scheduler* scheduler,
       ConnectionContextStore* connectionContextStore) override {
     return std::make_unique<FlipperWebSocket>(
         std::move(endpoint), std::move(payload), connectionContextStore);

xplat/Flipper/FlipperClient.cpp

@@ -8,6 +8,7 @@
 #include "FlipperClient.h"
 #include <fstream>
 #include <iostream>
+#include <sstream>
 #include <stdexcept>
 #include <vector>
 #include "ConnectionContextStore.h"

xplat/Flipper/FlipperClient.h

@@ -8,6 +8,7 @@
 #pragma once
 
 #include <map>
+#include <memory>
 #include <mutex>
 #include <vector>
 #include "FlipperCertificateProvider.h"

xplat/Flipper/FlipperConnectionManagerImpl.cpp

@@ -95,8 +95,8 @@ FlipperConnectionManagerImpl::FlipperConnectionManagerImpl(
       securePort(config.securePort),
       altInsecurePort(config.altInsecurePort),
       altSecurePort(config.altSecurePort),
-      flipperEventBase_(config.callbackWorker),
+      flipperScheduler_(config.callbackWorker),
-      connectionEventBase_(config.connectionWorker),
+      connectionScheduler_(config.connectionWorker),
       contextStore_(contextStore),
       implWrapper_(std::make_shared<FlipperConnectionManagerWrapper>(this)) {
   CHECK_THROW(config.callbackWorker, std::invalid_argument);
@@ -131,13 +131,10 @@ void FlipperConnectionManagerImpl::start() {
 
   auto step = flipperState_->start("Start connection thread");
 
-  folly::makeFuture()
+  flipperScheduler_->schedule([this, step]() {
-      .via(flipperEventBase_->getEventBase())
+    step->complete();
-      .delayed(std::chrono::milliseconds(0))
+    startSync();
-      .thenValue([this, step](auto&&) {
+  });
-        step->complete();
-        startSync();
-      });
 }
 
 void FlipperConnectionManagerImpl::startSync() {
@@ -210,7 +207,7 @@ bool FlipperConnectionManagerImpl::connectAndExchangeCertificate() {
   payload->medium = medium;
 
   auto newClient = FlipperSocketProvider::socketCreate(
-      endpoint, std::move(payload), connectionEventBase_);
+      endpoint, std::move(payload), flipperScheduler_);
   newClient->setEventHandler(ConnectionEvents(implWrapper_));
 
   auto connectingInsecurely = flipperState_->start("Connect insecurely");
@@ -259,7 +256,7 @@ bool FlipperConnectionManagerImpl::connectSecurely() {
   payload->csr_path = contextStore_->getCertificateDirectoryPath().c_str();
 
   auto newClient = FlipperSocketProvider::socketCreate(
-      endpoint, std::move(payload), connectionEventBase_, contextStore_.get());
+      endpoint, std::move(payload), connectionScheduler_, contextStore_.get());
   newClient->setEventHandler(ConnectionEvents(implWrapper_));
   newClient->setMessageHandler([this](const std::string& msg) {
     std::unique_ptr<FireAndForgetBasedFlipperResponder> responder;
@@ -294,10 +291,8 @@ void FlipperConnectionManagerImpl::reconnect() {
     log("Not started");
     return;
   }
-  folly::makeFuture()
+  flipperScheduler_->scheduleAfter(
-      .via(flipperEventBase_->getEventBase())
+      [this]() { startSync(); }, reconnectIntervalSeconds * 1000.0f);
-      .delayed(std::chrono::seconds(reconnectIntervalSeconds))
-      .thenValue([this](auto&&) { startSync(); });
 }
 
 void FlipperConnectionManagerImpl::stop() {
@@ -325,7 +320,7 @@ void FlipperConnectionManagerImpl::setCallbacks(Callbacks* callbacks) {
 }
 
 void FlipperConnectionManagerImpl::sendMessage(const folly::dynamic& message) {
-  flipperEventBase_->add([this, message]() {
+  flipperScheduler_->schedule([this, message]() {
     try {
       if (client_) {
         client_->send(message, []() {});
@@ -457,11 +452,11 @@ void FlipperConnectionManagerImpl::requestSignedCertificate() {
   auto gettingCert = flipperState_->start("Getting cert from desktop");
 
   certificateExchangeCompleted_ = false;
-  flipperEventBase_->add([this, message, gettingCert]() {
+  flipperScheduler_->schedule([this, message, gettingCert]() {
     client_->sendExpectResponse(
         folly::toJson(message),
         [this, gettingCert](const std::string& response, bool isError) {
-          flipperEventBase_->add([this, gettingCert, response, isError]() {
+          flipperScheduler_->schedule([this, gettingCert, response, isError]() {
             this->processSignedCertificateResponse(
                 gettingCert, response, isError);
           });
@@ -471,7 +466,7 @@ void FlipperConnectionManagerImpl::requestSignedCertificate() {
 }
 
 bool FlipperConnectionManagerImpl::isRunningInOwnThread() {
-  return flipperEventBase_->isInEventBaseThread();
+  return flipperScheduler_->isRunningInOwnThread();
 }
 
 } // namespace flipper

xplat/Flipper/FlipperConnectionManagerImpl.h

@@ -7,11 +7,10 @@
 
 #pragma once
 
-#include <folly/Executor.h>
-#include <folly/io/async/EventBase.h>
 #include <mutex>
 #include "FlipperConnectionManager.h"
 #include "FlipperInitConfig.h"
+#include "FlipperScheduler.h"
 #include "FlipperSocket.h"
 #include "FlipperState.h"
 
@@ -64,8 +63,8 @@ class FlipperConnectionManagerImpl : public FlipperConnectionManager {
   int altInsecurePort;
   int altSecurePort;
 
-  folly::EventBase* flipperEventBase_;
+  Scheduler* flipperScheduler_;
-  folly::EventBase* connectionEventBase_;
+  Scheduler* connectionScheduler_;
 
   std::unique_ptr<FlipperSocket> client_;
 

xplat/Flipper/FlipperInitConfig.h

@@ -7,8 +7,8 @@
 
 #pragma once
 
-#include <folly/io/async/EventBase.h>
 #include <map>
+#include "FlipperScheduler.h"
 
 namespace facebook {
 namespace flipper {
@@ -30,14 +30,14 @@ struct FlipperInitConfig {
   DeviceData deviceData;
 
   /**
-  EventBase on which client callbacks should be called.
+  Scheduler on which client callbacks should be called.
   */
-  folly::EventBase* callbackWorker;
+  Scheduler* callbackWorker;
 
   /**
-  EventBase to be used to maintain the network connection.
+  Scheduler to be used to maintain the network connection.
   */
-  folly::EventBase* connectionWorker;
+  Scheduler* connectionWorker;
 
   int insecurePort = 9089;
   int securePort = 9088;

xplat/Flipper/FlipperSocketProvider.cpp

@@ -21,19 +21,19 @@ std::unique_ptr<FlipperSocketProvider> FlipperSocketProvider::shelvedProvider_ =
 std::unique_ptr<FlipperSocket> FlipperSocketProvider::socketCreate(
     FlipperConnectionEndpoint endpoint,
     std::unique_ptr<FlipperSocketBasePayload> payload,
-    folly::EventBase* eventBase) {
+    Scheduler* scheduler) {
-  return provider_->create(std::move(endpoint), std::move(payload), eventBase);
+  return provider_->create(std::move(endpoint), std::move(payload), scheduler);
 }
 
 std::unique_ptr<FlipperSocket> FlipperSocketProvider::socketCreate(
     FlipperConnectionEndpoint endpoint,
     std::unique_ptr<FlipperSocketBasePayload> payload,
-    folly::EventBase* eventBase,
+    Scheduler* scheduler,
     ConnectionContextStore* connectionContextStore) {
   return provider_->create(
       std::move(endpoint),
       std::move(payload),
-      eventBase,
+      scheduler,
       connectionContextStore);
 }
 

xplat/Flipper/FlipperSocketProvider.h

@@ -7,8 +7,8 @@
 
 #pragma once
 
-#include <folly/io/async/EventBase.h>
 #include <memory>
+#include "FlipperScheduler.h"
 
 namespace facebook {
 namespace flipper {
@@ -32,35 +32,37 @@ class FlipperSocketProvider {
    @param endpoint Endpoint to connect to.
    @param payload Any configuration payload to establish a connection with
    the specified endpoint.
-   @param eventBase A folly event base used to execute connection operations.
+   @param scheduler An scheduler used to schedule and execute connection
+   operations.
    */
   virtual std::unique_ptr<FlipperSocket> create(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase) = 0;
+      Scheduler* scheduler) = 0;
   /**
    Create an instance of FlipperSocket.
    @param endpoint Endpoint to connect to.
    @param payload Any configuration payload to establish a connection with
    the specified endpoint.
-   @param eventBase A folly event base used to execute connection operations.
+   @param scheduler An scheduler used to schedule and execute connection
+   operations.
    @param connectionContextStore A connection context store used for obtaining
    the certificate used for secure connections.
    */
   virtual std::unique_ptr<FlipperSocket> create(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase,
+      Scheduler* scheduler,
       ConnectionContextStore* connectionContextStore) = 0;
 
   static std::unique_ptr<FlipperSocket> socketCreate(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase);
+      Scheduler* scheduler);
   static std::unique_ptr<FlipperSocket> socketCreate(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase,
+      Scheduler* scheduler,
       ConnectionContextStore* connectionContextStore);
 
   static void setDefaultProvider(

xplat/FlipperTests/FlipperConnectionManagerImplTerminationTests.cpp

@@ -6,10 +6,11 @@
  */
 
 #include <Flipper/FlipperConnectionManagerImpl.h>
+#include <Flipper/FlipperFollyScheduler.h>
 #include <FlipperTestLib/ConnectionContextStoreMock.h>
 #include <folly/Singleton.h>
-
 #include <gtest/gtest.h>
+#include <memory>
 
 namespace facebook {
 namespace flipper {
@@ -21,12 +22,16 @@ class FlipperConnectionManagerImplTerminationTest : public ::testing::Test {
  protected:
   std::shared_ptr<FlipperState> state;
   std::shared_ptr<ConnectionContextStore> contextStore;
+  std::unique_ptr<Scheduler> sonarScheduler;
+  std::unique_ptr<Scheduler> connectionScheduler;
   void SetUp() override {
     // Folly singletons must be registered before they are used.
     // Without this, test fails in phabricator.
     folly::SingletonVault::singleton()->registrationComplete();
     state = std::make_shared<FlipperState>();
     contextStore = std::make_shared<ConnectionContextStoreMock>();
+    sonarScheduler = std::make_unique<FollyScheduler>(new EventBase());
+    connectionScheduler = std::make_unique<FollyScheduler>(new EventBase());
   }
 };
 
@@ -35,7 +40,7 @@ TEST_F(
     testNullEventBaseGetsRejected) {
   try {
     auto instance = std::make_shared<FlipperConnectionManagerImpl>(
-        FlipperInitConfig{DeviceData{}, nullptr, new EventBase()},
+        FlipperInitConfig{DeviceData{}, nullptr, connectionScheduler.get()},
         state,
         contextStore);
     FAIL();
@@ -44,7 +49,7 @@ TEST_F(
   }
   try {
     auto instance = std::make_shared<FlipperConnectionManagerImpl>(
-        FlipperInitConfig{DeviceData{}, new EventBase(), nullptr},
+        FlipperInitConfig{DeviceData{}, sonarScheduler.get(), nullptr},
         state,
         contextStore);
     FAIL();
@@ -56,8 +61,8 @@ TEST_F(
 TEST_F(
     FlipperConnectionManagerImplTerminationTest,
     testNonStartedEventBaseDoesntHang) {
-  auto config =
+  auto config = FlipperInitConfig{
-      FlipperInitConfig{DeviceData{}, new EventBase(), new EventBase()};
+      DeviceData{}, sonarScheduler.get(), connectionScheduler.get()};
   auto instance = std::make_shared<FlipperConnectionManagerImpl>(
       config, state, contextStore);
   instance->start();
@@ -72,8 +77,11 @@ TEST_F(
       std::thread([flipperEventBase]() { flipperEventBase->loopForever(); });
   auto connectionThread = std::thread(
       [connectionEventBase]() { connectionEventBase->loopForever(); });
-  auto config =
+  auto localSonarScheduler = std::make_unique<FollyScheduler>(flipperEventBase);
-      FlipperInitConfig{DeviceData{}, flipperEventBase, connectionEventBase};
+  auto localConnectionScheduler =
+      std::make_unique<FollyScheduler>(connectionEventBase);
+  auto config = FlipperInitConfig{
+      DeviceData{}, localSonarScheduler.get(), localConnectionScheduler.get()};
   auto instance = std::make_shared<FlipperConnectionManagerImpl>(
       config, state, contextStore);
 

xplat/FlipperWebSocket/BaseClient.h

@@ -9,11 +9,11 @@
 
 #pragma once
 
+#include <Flipper/FlipperScheduler.h>
 #include <Flipper/FlipperSocket.h>
 #include <Flipper/FlipperSocketProvider.h>
 #include <Flipper/FlipperTransportTypes.h>
 #include <folly/dynamic.h>
-#include <folly/io/async/EventBase.h>
 #include <future>
 #include <memory>
 #include <mutex>
@@ -37,18 +37,18 @@ class BaseClient {
   BaseClient(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase)
+      Scheduler* scheduler)
       : endpoint_(std::move(endpoint)),
         payload_(std::move(payload)),
-        eventBase_(eventBase) {}
+        scheduler_(scheduler) {}
   BaseClient(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase,
+      Scheduler* scheduler,
       ConnectionContextStore* connectionContextStore)
       : endpoint_(std::move(endpoint)),
         payload_(std::move(payload)),
-        eventBase_(eventBase),
+        scheduler_(scheduler),
         connectionContextStore_(connectionContextStore) {}
 
   BaseClient(const BaseClient&) = delete;
@@ -84,7 +84,7 @@ class BaseClient {
  protected:
   FlipperConnectionEndpoint endpoint_;
   std::unique_ptr<FlipperSocketBasePayload> payload_;
-  folly::EventBase* eventBase_;
+  Scheduler* scheduler_;
   ConnectionContextStore* connectionContextStore_;
 
   SocketEventHandler eventHandler_;

xplat/FlipperWebSocket/FlipperWebSocket.cpp

@@ -14,8 +14,6 @@
 #include <Flipper/Log.h>
 #include <folly/String.h>
 #include <folly/futures/Future.h>
-#include <folly/io/async/AsyncSocketException.h>
-#include <folly/io/async/SSLContext.h>
 #include <folly/json.h>
 #include <cctype>
 #include <iomanip>
@@ -32,24 +30,24 @@ namespace flipper {
 FlipperWebSocket::FlipperWebSocket(
     FlipperConnectionEndpoint endpoint,
     std::unique_ptr<FlipperSocketBasePayload> payload,
-    folly::EventBase* eventBase)
+    Scheduler* scheduler)
     : FlipperWebSocket(
           std::move(endpoint),
           std::move(payload),
-          eventBase,
+          scheduler,
           nullptr) {}
 
 FlipperWebSocket::FlipperWebSocket(
     FlipperConnectionEndpoint endpoint,
     std::unique_ptr<FlipperSocketBasePayload> payload,
-    folly::EventBase* eventBase,
+    Scheduler* scheduler,
     ConnectionContextStore* connectionContextStore) {
   if (endpoint.secure) {
     socket_ = std::make_unique<WebSocketTLSClient>(
-        endpoint, std::move(payload), eventBase, connectionContextStore);
+        endpoint, std::move(payload), scheduler, connectionContextStore);
   } else {
     socket_ = std::make_unique<WebSocketClient>(
-        endpoint, std::move(payload), eventBase, connectionContextStore);
+        endpoint, std::move(payload), scheduler, connectionContextStore);
   }
 }
 

xplat/FlipperWebSocket/FlipperWebSocket.h

@@ -9,11 +9,11 @@
 
 #pragma once
 
+#include <Flipper/FlipperScheduler.h>
 #include <Flipper/FlipperSocket.h>
 #include <Flipper/FlipperSocketProvider.h>
 #include <Flipper/FlipperTransportTypes.h>
 #include <folly/dynamic.h>
-#include <folly/io/async/EventBase.h>
 #include <future>
 #include <memory>
 #include <mutex>
@@ -29,11 +29,11 @@ class FlipperWebSocket : public FlipperSocket {
   FlipperWebSocket(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase);
+      Scheduler* scheduler);
   FlipperWebSocket(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase,
+      Scheduler* scheduler,
       ConnectionContextStore* connectionContextStore);
 
   virtual ~FlipperWebSocket();
@@ -62,19 +62,19 @@ class FlipperWebSocketProvider : public FlipperSocketProvider {
   virtual std::unique_ptr<FlipperSocket> create(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase) override {
+      Scheduler* scheduler) override {
     return std::make_unique<FlipperWebSocket>(
-        std::move(endpoint), std::move(payload), eventBase);
+        std::move(endpoint), std::move(payload), scheduler);
   }
   virtual std::unique_ptr<FlipperSocket> create(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase,
+      Scheduler* scheduler,
       ConnectionContextStore* connectionContextStore) override {
     return std::make_unique<FlipperWebSocket>(
         std::move(endpoint),
         std::move(payload),
-        eventBase,
+        scheduler,
         connectionContextStore);
   }
 };

xplat/FlipperWebSocket/WebSocketClient.cpp

@@ -14,7 +14,6 @@
 #include <Flipper/Log.h>
 #include <folly/String.h>
 #include <folly/futures/Future.h>
-#include <folly/io/async/SSLContext.h>
 #include <folly/json.h>
 #include <websocketpp/common/memory.hpp>
 #include <websocketpp/common/thread.hpp>
@@ -31,22 +30,22 @@ namespace flipper {
 WebSocketClient::WebSocketClient(
     FlipperConnectionEndpoint endpoint,
     std::unique_ptr<FlipperSocketBasePayload> payload,
-    folly::EventBase* eventBase)
+    Scheduler* scheduler)
     : WebSocketClient(
           std::move(endpoint),
           std::move(payload),
-          eventBase,
+          scheduler,
           nullptr) {}
 
 WebSocketClient::WebSocketClient(
     FlipperConnectionEndpoint endpoint,
     std::unique_ptr<FlipperSocketBasePayload> payload,
-    folly::EventBase* eventBase,
+    Scheduler* scheduler,
     ConnectionContextStore* connectionContextStore)
     : BaseClient(
           std::move(endpoint),
           std::move(payload),
-          eventBase,
+          scheduler,
           connectionContextStore) {
   status_ = Status::Unconnected;
 
@@ -148,7 +147,7 @@ void WebSocketClient::disconnect() {
     thread_->join();
   }
   thread_ = nullptr;
-  eventBase_->add(
+  scheduler_->schedule(
       [eventHandler = eventHandler_]() { eventHandler(SocketEvent::CLOSE); });
 }
 
@@ -180,12 +179,12 @@ void WebSocketClient::send(
 void WebSocketClient::sendExpectResponse(
     const std::string& message,
     SocketSendExpectResponseHandler completion) {
-  connection_->set_message_handler(
+  connection_->set_message_handler([completion, scheduler = scheduler_](
-      [completion, eventBase = eventBase_](
+                                       websocketpp::connection_hdl hdl,
-          websocketpp::connection_hdl hdl, SocketClient::message_ptr msg) {
+                                       SocketClient::message_ptr msg) {
-        const std::string& payload = msg->get_payload();
+    const std::string& payload = msg->get_payload();
-        eventBase->add([completion, payload] { completion(payload, false); });
+    scheduler->schedule([completion, payload] { completion(payload, false); });
-      });
+  });
   websocketpp::lib::error_code ec;
   socket_.send(
       handle_,
@@ -205,7 +204,7 @@ void WebSocketClient::onOpen(SocketClient* c, websocketpp::connection_hdl hdl) {
   }
 
   status_ = Status::Initializing;
-  eventBase_->add(
+  scheduler_->schedule(
       [eventHandler = eventHandler_]() { eventHandler(SocketEvent::OPEN); });
 }
 
@@ -215,7 +214,7 @@ void WebSocketClient::onMessage(
     SocketClient::message_ptr msg) {
   const std::string& payload = msg->get_payload();
   if (messageHandler_) {
-    eventBase_->add([payload, messageHandler = messageHandler_]() {
+    scheduler_->schedule([payload, messageHandler = messageHandler_]() {
       messageHandler(payload);
     });
   }
@@ -230,7 +229,7 @@ void WebSocketClient::onFail(SocketClient* c, websocketpp::connection_hdl hdl) {
     connected_.set_value(false);
   }
   status_ = Status::Failed;
-  eventBase_->add(
+  scheduler_->schedule(
       [eventHandler = eventHandler_]() { eventHandler(SocketEvent::ERROR); });
 }
 
@@ -238,7 +237,7 @@ void WebSocketClient::onClose(
     SocketClient* c,
     websocketpp::connection_hdl hdl) {
   status_ = Status::Closed;
-  eventBase_->add(
+  scheduler_->schedule(
       [eventHandler = eventHandler_]() { eventHandler(SocketEvent::CLOSE); });
 }
 

xplat/FlipperWebSocket/WebSocketClient.h

@@ -9,11 +9,11 @@
 
 #pragma once
 
+#include <Flipper/FlipperScheduler.h>
 #include <Flipper/FlipperSocket.h>
 #include <Flipper/FlipperSocketProvider.h>
 #include <Flipper/FlipperTransportTypes.h>
 #include <folly/dynamic.h>
-#include <folly/io/async/EventBase.h>
 #include <future>
 #include <memory>
 #include <mutex>
@@ -35,11 +35,11 @@ class WebSocketClient : public BaseClient {
   WebSocketClient(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase);
+      Scheduler* scheduler);
   WebSocketClient(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase,
+      Scheduler* scheduler,
       ConnectionContextStore* connectionContextStore);
 
   WebSocketClient(const WebSocketClient&) = delete;

xplat/FlipperWebSocket/WebSocketTLSClient.cpp

@@ -33,22 +33,22 @@ namespace flipper {
 WebSocketTLSClient::WebSocketTLSClient(
     FlipperConnectionEndpoint endpoint,
     std::unique_ptr<FlipperSocketBasePayload> payload,
-    folly::EventBase* eventBase)
+    Scheduler* scheduler)
     : WebSocketTLSClient(
           std::move(endpoint),
           std::move(payload),
-          eventBase,
+          scheduler,
           nullptr) {}
 
 WebSocketTLSClient::WebSocketTLSClient(
     FlipperConnectionEndpoint endpoint,
     std::unique_ptr<FlipperSocketBasePayload> payload,
-    folly::EventBase* eventBase,
+    Scheduler* scheduler,
     ConnectionContextStore* connectionContextStore)
     : BaseClient(
           std::move(endpoint),
           std::move(payload),
-          eventBase,
+          scheduler,
           connectionContextStore) {
   status_ = Status::Unconnected;
 
@@ -157,7 +157,7 @@ void WebSocketTLSClient::disconnect() {
   }
 
   thread_ = nullptr;
-  eventBase_->add(
+  scheduler_->schedule(
       [eventHandler = eventHandler_]() { eventHandler(SocketEvent::CLOSE); });
 }
 
@@ -189,12 +189,12 @@ void WebSocketTLSClient::send(
 void WebSocketTLSClient::sendExpectResponse(
     const std::string& message,
     SocketSendExpectResponseHandler completion) {
-  connection_->set_message_handler(
+  connection_->set_message_handler([completion, scheduler = scheduler_](
-      [completion, eventBase = eventBase_](
+                                       websocketpp::connection_hdl hdl,
-          websocketpp::connection_hdl hdl, SocketTLSClient::message_ptr msg) {
+                                       SocketTLSClient::message_ptr msg) {
-        const std::string& payload = msg->get_payload();
+    const std::string& payload = msg->get_payload();
-        eventBase->add([completion, payload] { completion(payload, false); });
+    scheduler->schedule([completion, payload] { completion(payload, false); });
-      });
+  });
   websocketpp::lib::error_code ec;
   socket_.send(
       handle_,
@@ -216,7 +216,7 @@ void WebSocketTLSClient::onOpen(
   }
 
   status_ = Status::Initializing;
-  eventBase_->add(
+  scheduler_->schedule(
       [eventHandler = eventHandler_]() { eventHandler(SocketEvent::OPEN); });
 }
 
@@ -226,7 +226,7 @@ void WebSocketTLSClient::onMessage(
     SocketTLSClient::message_ptr msg) {
   const std::string& payload = msg->get_payload();
   if (messageHandler_) {
-    eventBase_->add([payload, messageHandler = messageHandler_]() {
+    scheduler_->schedule([payload, messageHandler = messageHandler_]() {
       messageHandler(payload);
     });
   }
@@ -257,7 +257,7 @@ void WebSocketTLSClient::onFail(
     }
   }
   status_ = Status::Failed;
-  eventBase_->add([eventHandler = eventHandler_, sslError]() {
+  scheduler_->schedule([eventHandler = eventHandler_, sslError]() {
     if (sslError) {
       eventHandler(SocketEvent::SSL_ERROR);
     } else {
@@ -270,7 +270,7 @@ void WebSocketTLSClient::onClose(
     SocketTLSClient* c,
     websocketpp::connection_hdl hdl) {
   status_ = Status::Closed;
-  eventBase_->add(
+  scheduler_->schedule(
       [eventHandler = eventHandler_]() { eventHandler(SocketEvent::CLOSE); });
 }
 

xplat/FlipperWebSocket/WebSocketTLSClient.h

@@ -13,7 +13,6 @@
 #include <Flipper/FlipperSocketProvider.h>
 #include <Flipper/FlipperTransportTypes.h>
 #include <folly/dynamic.h>
-#include <folly/io/async/EventBase.h>
 #include <future>
 #include <memory>
 #include <mutex>
@@ -39,11 +38,11 @@ class WebSocketTLSClient : public BaseClient {
   WebSocketTLSClient(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase);
+      Scheduler* scheduler);
   WebSocketTLSClient(
       FlipperConnectionEndpoint endpoint,
       std::unique_ptr<FlipperSocketBasePayload> payload,
-      folly::EventBase* eventBase,
+      Scheduler* scheduler,
       ConnectionContextStore* connectionContextStore);
 
   WebSocketTLSClient(const WebSocketTLSClient&) = delete;