Don't use private access in FlipperRSocketResponder

Summary:
Changing FlipperRSocketResponder to only use public parts of FlipperConnectionManagerImpl.

This means we can test it by injecting a FCM mock, and it can use its public interface.

Reviewed By: passy

Differential Revision: D14000078

fbshipit-source-id: c0431a888b0ca041807631c81b99fb8b947274d6

xplat/Flipper/FlipperConnectionManager.h

@@ -45,6 +45,13 @@ class FlipperConnectionManager {
    The callbacks should be set before a connection is established.
    */
   virtual void setCallbacks(Callbacks* callbacks) = 0;
+
+  /**
+   Called by ws server when a message has been received.
+  */
+  virtual void onMessageReceived(
+      const folly::dynamic& message,
+      std::unique_ptr<FlipperResponder> responder) = 0;
 };
 
 class FlipperConnectionManager::Callbacks {

xplat/Flipper/FlipperConnectionManagerImpl.cpp

@@ -193,7 +193,7 @@ void FlipperConnectionManagerImpl::connectSecurely() {
               std::move(address),
               std::move(sslContext)),
           std::move(parameters),
-          std::make_shared<FlipperRSocketResponder>(this),
+          std::make_shared<FlipperRSocketResponder>(this, connectionEventBase_),
           std::chrono::seconds(connectionKeepaliveSeconds), // keepaliveInterval
           nullptr, // stats
           std::make_shared<ConnectionEvents>(this))
@@ -241,6 +241,12 @@ void FlipperConnectionManagerImpl::sendMessage(const folly::dynamic& message) {
   });
 }
 
+void FlipperConnectionManagerImpl::onMessageReceived(
+    const folly::dynamic& message,
+    std::unique_ptr<FlipperResponder> responder) {
+  callbacks_->onMessageReceived(message, std::move(responder));
+}
+
 bool FlipperConnectionManagerImpl::isCertificateExchangeNeeded() {
   if (failedConnectionAttempts_ >= 2) {
     return true;

xplat/Flipper/FlipperConnectionManagerImpl.h

@@ -25,7 +25,6 @@ rsocket::Payload toRSocketPayload(folly::dynamic data);
 
 class FlipperConnectionManagerImpl : public FlipperConnectionManager {
   friend ConnectionEvents;
-  friend FlipperRSocketResponder;
 
  public:
   FlipperConnectionManagerImpl(FlipperInitConfig config, std::shared_ptr<FlipperState> state, std::shared_ptr<ConnectionContextStore> contextStore);
@@ -42,6 +41,10 @@ class FlipperConnectionManagerImpl : public FlipperConnectionManager {
 
   void sendMessage(const folly::dynamic& message) override;
 
+  void onMessageReceived(
+      const folly::dynamic& message,
+      std::unique_ptr<FlipperResponder> responder) override;
+
   void reconnect();
 
  private:

xplat/Flipper/FlipperRSocketResponder.cpp

@@ -31,7 +31,7 @@ void FlipperRSocketResponder::handleFireAndForget(
         std::make_unique<FireAndForgetBasedFlipperResponder>(websocket_, id);
   }
 
-  websocket_->callbacks_->onMessageReceived(
+  websocket_->onMessageReceived(
       folly::parseJson(payload), std::move(responder));
 }
 
@@ -44,7 +44,7 @@ FlipperRSocketResponder::handleRequestResponse(
   auto dynamicSingle = yarpl::single::Single<folly::dynamic>::create(
       [payload = std::move(requestString), this](auto observer) {
         auto responder = std::make_unique<FlipperResponderImpl>(observer);
-        websocket_->callbacks_->onMessageReceived(
+        websocket_->onMessageReceived(
             folly::parseJson(payload), std::move(responder));
       });
 
@@ -53,19 +53,18 @@ FlipperRSocketResponder::handleRequestResponse(
         observer->onSubscribe(yarpl::single::SingleSubscriptions::empty());
         dynamicSingle->subscribe(
             [observer, this](folly::dynamic d) {
-              websocket_->connectionEventBase_->runInEventBaseThread(
-                  [observer, d]() {
-                    try {
-                      observer->onSuccess(toRSocketPayload(d));
+              eventBase_->runInEventBaseThread([observer, d]() {
+                try {
+                  observer->onSuccess(toRSocketPayload(d));
 
-                    } catch (std::exception& e) {
-                      log(e.what());
-                      observer->onError(e);
-                    }
-                  });
+                } catch (std::exception& e) {
+                  log(e.what());
+                  observer->onError(e);
+                }
+              });
             },
             [observer, this](folly::exception_wrapper e) {
-              websocket_->connectionEventBase_->runInEventBaseThread(
+              eventBase_->runInEventBaseThread(
                   [observer, e]() { observer->onError(e); });
             });
       });

xplat/Flipper/FlipperRSocketResponder.h

@@ -9,15 +9,18 @@
 namespace facebook {
 namespace flipper {
 
-class FlipperConnectionManagerImpl;
+class FlipperConnectionManager;
 
 class FlipperRSocketResponder : public rsocket::RSocketResponder {
  private:
-  FlipperConnectionManagerImpl* websocket_;
+  FlipperConnectionManager* websocket_;
+  folly::EventBase* eventBase_;
 
  public:
-  FlipperRSocketResponder(FlipperConnectionManagerImpl* websocket)
-      : websocket_(websocket){};
+  FlipperRSocketResponder(
+      FlipperConnectionManager* websocket,
+      folly::EventBase* eventBase)
+      : websocket_(websocket), eventBase_(eventBase){};
 
   void handleFireAndForget(
       rsocket::Payload request,