Receive client setup (#110)

* receive setup frame

* cleanup

* request handler use value type

Author: yschimke

CMakeLists.txt

@@ -131,7 +131,8 @@ add_library(
   src/Stats.cpp
   src/folly/FollyKeepaliveTimer.cpp
   src/folly/FollyKeepaliveTimer.h
-  src/ConnectionSetupPayload.h)
+  src/ConnectionSetupPayload.h
+  src/ConnectionSetupPayload.cpp)
 
 target_link_libraries(
   ReactiveSocket

src/ConnectionAutomaton.cpp

@@ -180,30 +180,14 @@ void ConnectionAutomaton::onConnectionFrame(
         outputFrameOrEnqueue(Frame_ERROR::unexpectedFrame().serializeOut());
         disconnect();
       }
-    }
       return;
+    }
     case FrameType::SETUP: {
-      Frame_SETUP frame;
-      if (frame.deserializeFrom(std::move(payload))) {
-        if (frame.header_.flags_ & FrameFlags_LEASE) {
-          // TODO(yschimke) We don't have the correct lease and wait logic above
-          // yet
-          LOG(WARNING) << "ignoring setup frame with lease";
-          //          connectionOutput_.onNext(
-          //              Frame_ERROR::badSetupFrame("leases not supported")
-          //                  .serializeOut());
-          //          disconnect();
-        }
-      } else {
-        // TODO(yschimke) enable this later after clients upgraded
-        LOG(WARNING) << "ignoring bad setup frame";
-        //        connectionOutput_.onNext(
-        //            Frame_ERROR::badSetupFrame("bad setup
-        //            frame").serializeOut());
-        //        disconnect();
+      if (!factory_(0, std::move(payload))) {
+        assert(false);
       }
-    }
       return;
+    }
     case FrameType::METADATA_PUSH: {
       if (!factory_(0, std::move(payload))) {
         assert(false);
@@ -230,7 +214,8 @@ void ConnectionAutomaton::onTerminal(folly::exception_wrapper ex) {
     auto result = endStreamInternal(streams_.begin()->first, signal);
     (void)oldSize;
     (void)result;
-    // TODO(stupaq): what kind of a user action could violate these assertions?
+    // TODO(stupaq): what kind of a user action could violate these
+    // assertions?
     assert(result);
     assert(streams_.size() == oldSize - 1);
   }

src/ConnectionSetupPayload.cpp

@@ -0,0 +1,14 @@
+// Copyright 2004-present Facebook. All Rights Reserved.
+
+#include "ConnectionSetupPayload.h"
+#include <folly/String.h>
+
+namespace reactivesocket {
+std::ostream& operator<<(
+    std::ostream& os,
+    const ConnectionSetupPayload& setupPayload) {
+  return os << "[metadataMimeType: " << setupPayload.metadataMimeType
+            << " dataMimeType: " << setupPayload.dataMimeType
+            << " payload: " << setupPayload.payload << "]";
+}
+}

src/ConnectionSetupPayload.h

@@ -2,6 +2,7 @@
 
 #pragma once
 
+#include <folly/io/IOBuf.h>
 #include <string>
 #include "Payload.h"
 
@@ -20,4 +21,6 @@ class ConnectionSetupPayload {
   std::string dataMimeType;
   Payload payload;
 };
+
+std::ostream& operator<<(std::ostream&, const ConnectionSetupPayload&);
 }

src/NullRequestHandler.cpp

@@ -50,4 +50,7 @@ void NullRequestHandler::handleFireAndForgetRequest(Payload /*request*/) {}
 
 void NullRequestHandler::handleMetadataPush(
     std::unique_ptr<folly::IOBuf> /*request*/) {}
+
+void NullRequestHandler::handleSetupPayload(
+    ConnectionSetupPayload /*request*/) {}
 }

src/NullRequestHandler.h

@@ -2,6 +2,7 @@
 
 #pragma once
 
+#include "ConnectionSetupPayload.h"
 #include "src/RequestHandler.h"
 
 namespace reactivesocket {
@@ -37,6 +38,8 @@ class NullRequestHandler : public RequestHandler {
   void handleFireAndForgetRequest(Payload request) override;
 
   void handleMetadataPush(std::unique_ptr<folly::IOBuf> request) override;
+
+  void handleSetupPayload(ConnectionSetupPayload request) override;
 };
 
 using DefaultRequestHandler = NullRequestHandler;

src/ReactiveSocket.cpp

@@ -164,6 +164,34 @@ bool ReactiveSocket::createResponder(
     std::unique_ptr<folly::IOBuf> serializedFrame) {
   auto type = FrameHeader::peekType(*serializedFrame);
   switch (type) {
+    case FrameType::SETUP: {
+      Frame_SETUP frame;
+      if (frame.deserializeFrom(std::move(serializedFrame))) {
+        if (frame.header_.flags_ & FrameFlags_LEASE) {
+          // TODO(yschimke) We don't have the correct lease and wait logic above
+          // yet
+          LOG(WARNING) << "ignoring setup frame with lease";
+          //          connectionOutput_.onNext(
+          //              Frame_ERROR::badSetupFrame("leases not supported")
+          //                  .serializeOut());
+          //          disconnect();
+        }
+
+        handler_->handleSetupPayload(ConnectionSetupPayload(
+            std::move(frame.metadataMimeType_),
+            std::move(frame.dataMimeType_),
+            std::move(frame.payload_)));
+      } else {
+        // TODO(yschimke) enable this later after clients upgraded
+        LOG(WARNING) << "ignoring bad setup frame";
+        //        connectionOutput_.onNext(
+        //            Frame_ERROR::badSetupFrame("bad setup
+        //            frame").serializeOut());
+        //        disconnect();
+      }
+
+      break;
+    }
     case FrameType::REQUEST_CHANNEL: {
       Frame_REQUEST_CHANNEL frame;
       if (!frame.deserializeFrom(std::move(serializedFrame))) {

src/RequestHandler.h

@@ -2,6 +2,7 @@
 
 #pragma once
 
+#include "ConnectionSetupPayload.h"
 #include "src/Payload.h"
 #include "src/ReactiveStreamsCompat.h"
 
@@ -34,5 +35,9 @@ class RequestHandler {
 
   /// Handles a new metadata-push sent by the other end.
   virtual void handleMetadataPush(std::unique_ptr<folly::IOBuf> request) = 0;
+
+  /// Temporary home - this should eventually be an input to asking for a
+  /// RequestHandler so negotiation is possible
+  virtual void handleSetupPayload(ConnectionSetupPayload request) = 0;
 };
 }

tck-test/client.cpp

@@ -93,8 +93,7 @@ int main(int argc, char* argv[]) {
         folly::make_unique<DefaultRequestHandler>();
 
     reactiveSocket = ReactiveSocket::fromClientConnection(
-        std::move(framedConnection),
-        std::move(requestHandler));
+        std::move(framedConnection), std::move(requestHandler));
   });
 
   LOG(INFO) << "Test file parsed. Starting executing tests...";

test/MockRequestHandler.h

@@ -26,6 +26,7 @@ class MockRequestHandler : public RequestHandler {
   MOCK_METHOD1(
       handleMetadataPush_,
       void(std::unique_ptr<folly::IOBuf>& request));
+  MOCK_METHOD1(handleSetupPayload_, void(ConnectionSetupPayload& request));
 
   Subscriber<Payload>& handleRequestChannel(
       Payload request,
@@ -50,5 +51,9 @@ class MockRequestHandler : public RequestHandler {
   void handleMetadataPush(std::unique_ptr<folly::IOBuf> request) override {
     handleMetadataPush_(request);
   }
+
+  void handleSetupPayload(ConnectionSetupPayload request) override {
+    handleSetupPayload_(request);
+  }
 };
 }