Send client setup data (#108)

Author: yschimke

CMakeLists.txt

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

src/ConnectionAutomaton.cpp

@@ -22,13 +22,11 @@ ConnectionAutomaton::ConnectionAutomaton(
     std::unique_ptr<DuplexConnection> connection,
     StreamAutomatonFactory factory,
     Stats& stats,
-    bool isServer,
-    std::unique_ptr<KeepaliveTimer> keepAliveTimer)
+    bool isServer)
     : connection_(std::move(connection)),
       factory_(std::move(factory)),
       stats_(stats),
-      isServer_(isServer),
-      keepaliveTimer_(std::move(keepAliveTimer)) {
+      isServer_(isServer) {
   // We deliberately do not "open" input or output to avoid having c'tor on the
   // stack when processing any signals from the connection. See ::connect and
   // ::onSubscribe.
@@ -41,41 +39,18 @@ void ConnectionAutomaton::connect() {
   // subscription, which might deliver frames in-line.
   connection_->setInput(*this);
 
-  if (!isServer_) {
-    uint32_t keepaliveTime = keepaliveTimer_
-        ? keepaliveTimer_->keepaliveTime().count()
-        : std::numeric_limits<uint32_t>::max();
-
-    // TODO set correct version and allow clients to configure
-    Frame_SETUP frame(
-        FrameFlags_EMPTY,
-        0,
-        keepaliveTime,
-        std::numeric_limits<uint32_t>::max(),
-        "",
-        "",
-        Payload());
-    outputFrameOrEnqueue(frame.serializeOut());
-  }
   stats_.socketCreated();
-
-  if (keepaliveTimer_) {
-    keepaliveTimer_->start(this);
-  }
 }
 
 void ConnectionAutomaton::disconnect() {
   VLOG(6) << "disconnect";
 
-  if (keepaliveTimer_) {
-    keepaliveTimer_->stop();
-  }
-
   if (!connectionOutput_) {
     return;
   }
 
-  LOG_IF(WARNING, !pendingWrites_.empty()) << "disconnecting with pending writes (" << pendingWrites_.size() << ")";
+  LOG_IF(WARNING, !pendingWrites_.empty())
+      << "disconnecting with pending writes (" << pendingWrites_.size() << ")";
 
   // Send terminal signals to the DuplexConnection's input and output before
   // tearing it down. We must do this per DuplexConnection specification (see
@@ -307,7 +282,8 @@ void ConnectionAutomaton::onClose(ConnectionCloseListener listener) {
   closeListeners_.push_back(listener);
 }
 
-void ConnectionAutomaton::outputFrameOrEnqueue(std::unique_ptr<folly::IOBuf> frame) {
+void ConnectionAutomaton::outputFrameOrEnqueue(
+    std::unique_ptr<folly::IOBuf> frame) {
   if (!connectionOutput_) {
     return; // RS destructor has disconnected us from the DuplexConnection
   }
@@ -323,7 +299,7 @@ void ConnectionAutomaton::outputFrameOrEnqueue(std::unique_ptr<folly::IOBuf> fra
 }
 
 void ConnectionAutomaton::drainOutputFramesQueue() {
-// Drain the queue or the allowance.
+  // Drain the queue or the allowance.
   while (!pendingWrites_.empty() && writeAllowance_.tryAcquire()) {
     auto frame = std::move(pendingWrites_.front());
     pendingWrites_.pop_front();
@@ -332,13 +308,12 @@ void ConnectionAutomaton::drainOutputFramesQueue() {
 }
 
 void ConnectionAutomaton::outputFrame(
-        std::unique_ptr<folly::IOBuf> outputFrame) {
+    std::unique_ptr<folly::IOBuf> outputFrame) {
   std::stringstream ss;
   ss << FrameHeader::peekType(*outputFrame);
 
   stats_.frameWritten(ss.str());
 
   connectionOutput_.onNext(std::move(outputFrame));
 }
-
 }

src/ConnectionAutomaton.h

@@ -52,8 +52,7 @@ class ConnectionAutomaton :
       // TODO(stupaq): for testing only, can devirtualise if necessary
       StreamAutomatonFactory factory,
       Stats& stats,
-      bool client,
-      std::unique_ptr<KeepaliveTimer> keepaliveTimer);
+      bool client);
 
   /// Kicks off connection procedure.
   ///
@@ -165,7 +164,6 @@ class ConnectionAutomaton :
       pendingWrites_; // TODO(stupaq): two vectors?
   Stats& stats_;
   bool isServer_;
-  std::unique_ptr<KeepaliveTimer> keepaliveTimer_;
   std::vector<ConnectionCloseListener> closeListeners_;
 };
 }

src/ConnectionSetupPayload.h

@@ -0,0 +1,23 @@
+// Copyright 2004-present Facebook. All Rights Reserved.
+
+#pragma once
+
+#include <string>
+#include "Payload.h"
+
+namespace reactivesocket {
+class ConnectionSetupPayload {
+ public:
+  ConnectionSetupPayload(
+      std::string _metadataMimeType = "",
+      std::string _dataMimeType = "",
+      Payload _payload = Payload())
+      : metadataMimeType(std::move(_metadataMimeType)),
+        dataMimeType(std::move(_dataMimeType)),
+        payload(std::move(_payload)){};
+
+  std::string metadataMimeType;
+  std::string dataMimeType;
+  Payload payload;
+};
+}

src/ReactiveSocket.cpp

@@ -11,6 +11,7 @@
 #include <folly/MoveWrapper.h>
 
 #include "src/ConnectionAutomaton.h"
+#include "src/ConnectionSetupPayload.h"
 #include "src/DuplexConnection.h"
 #include "src/Frame.h"
 #include "src/Payload.h"
@@ -44,14 +45,15 @@ ReactiveSocket::ReactiveSocket(
               std::placeholders::_1,
               std::placeholders::_2),
           stats,
-          isServer,
-          std::move(keepaliveTimer))),
+          isServer)),
       handler_(std::move(handler)),
-      nextStreamId_(isServer ? 1 : 2) {}
+      nextStreamId_(isServer ? 1 : 2),
+      keepaliveTimer_(std::move(keepaliveTimer)) {}
 
 std::unique_ptr<ReactiveSocket> ReactiveSocket::fromClientConnection(
     std::unique_ptr<DuplexConnection> connection,
     std::unique_ptr<RequestHandler> handler,
+    ConnectionSetupPayload setupPayload,
     Stats& stats,
     std::unique_ptr<KeepaliveTimer> keepaliveTimer) {
   std::unique_ptr<ReactiveSocket> socket(new ReactiveSocket(
@@ -61,6 +63,27 @@ std::unique_ptr<ReactiveSocket> ReactiveSocket::fromClientConnection(
       stats,
       std::move(keepaliveTimer)));
   socket->connection_->connect();
+
+  uint32_t keepaliveTime = socket->keepaliveTimer_
+      ? socket->keepaliveTimer_->keepaliveTime().count()
+      : std::numeric_limits<uint32_t>::max();
+
+  // TODO set correct version
+  Frame_SETUP frame(
+      FrameFlags_EMPTY,
+      0,
+      keepaliveTime,
+      std::numeric_limits<uint32_t>::max(),
+      std::move(setupPayload.metadataMimeType),
+      std::move(setupPayload.dataMimeType),
+      std::move(setupPayload.payload));
+
+  socket->connection_->outputFrameOrEnqueue(frame.serializeOut());
+
+  if (socket->keepaliveTimer_) {
+    socket->keepaliveTimer_->start(socket->connection_.get());
+  }
+
   return socket;
 }
 
@@ -132,7 +155,8 @@ void ReactiveSocket::requestFireAndForget(Payload request) {
 }
 
 void ReactiveSocket::metadataPush(std::unique_ptr<folly::IOBuf> metadata) {
-  connection_->outputFrameOrEnqueue(Frame_METADATA_PUSH(std::move(metadata)).serializeOut());
+  connection_->outputFrameOrEnqueue(
+      Frame_METADATA_PUSH(std::move(metadata)).serializeOut());
 }
 
 bool ReactiveSocket::createResponder(

src/ReactiveSocket.h

@@ -7,6 +7,7 @@
 #include <functional>
 #include <memory>
 
+#include "src/ConnectionSetupPayload.h"
 #include "src/Payload.h"
 #include "src/ReactiveStreamsCompat.h"
 #include "src/Stats.h"
@@ -54,6 +55,7 @@ class ReactiveSocket {
   static std::unique_ptr<ReactiveSocket> fromClientConnection(
       std::unique_ptr<DuplexConnection> connection,
       std::unique_ptr<RequestHandler> handler,
+      ConnectionSetupPayload setupPayload = ConnectionSetupPayload(),
       Stats& stats = Stats::noop(),
       std::unique_ptr<KeepaliveTimer> keepaliveTimer =
           std::unique_ptr<KeepaliveTimer>(nullptr));
@@ -90,5 +92,6 @@ class ReactiveSocket {
   const std::shared_ptr<ConnectionAutomaton> connection_;
   std::unique_ptr<RequestHandler> handler_;
   StreamId nextStreamId_;
+  std::unique_ptr<KeepaliveTimer> keepaliveTimer_;
 };
 }

src/automata/ChannelRequester.cpp

@@ -68,7 +68,8 @@ void ChannelRequesterBase::onComplete() {
     case State::REQUESTED: {
       state_ = State::CLOSED;
       connection_->outputFrameOrEnqueue(
-          Frame_REQUEST_CHANNEL(streamId_, FrameFlags_COMPLETE, 0, Payload()).serializeOut());
+          Frame_REQUEST_CHANNEL(streamId_, FrameFlags_COMPLETE, 0, Payload())
+              .serializeOut());
       connection_->endStream(streamId_, StreamCompletionSignal::GRACEFUL);
     } break;
     case State::CLOSED:

src/automata/ChannelResponder.cpp

@@ -27,7 +27,8 @@ void ChannelResponderBase::onComplete() {
   switch (state_) {
     case State::RESPONDING: {
       state_ = State::CLOSED;
-      connection_->outputFrameOrEnqueue(Frame_RESPONSE::complete(streamId_).serializeOut());
+      connection_->outputFrameOrEnqueue(
+          Frame_RESPONSE::complete(streamId_).serializeOut());
       connection_->endStream(streamId_, StreamCompletionSignal::GRACEFUL);
     } break;
     case State::CLOSED:
@@ -40,7 +41,8 @@ void ChannelResponderBase::onError(folly::exception_wrapper ex) {
     case State::RESPONDING: {
       state_ = State::CLOSED;
       auto msg = ex.what().toStdString();
-      connection_->outputFrameOrEnqueue(Frame_ERROR::applicationError(streamId_, msg).serializeOut());
+      connection_->outputFrameOrEnqueue(
+          Frame_ERROR::applicationError(streamId_, msg).serializeOut());
       connection_->endStream(streamId_, StreamCompletionSignal::ERROR);
     } break;
     case State::CLOSED:
@@ -61,7 +63,8 @@ void ChannelResponderBase::cancel() {
   switch (state_) {
     case State::RESPONDING: {
       state_ = State::CLOSED;
-      connection_->outputFrameOrEnqueue(Frame_RESPONSE::complete(streamId_).serializeOut());
+      connection_->outputFrameOrEnqueue(
+          Frame_RESPONSE::complete(streamId_).serializeOut());
       connection_->endStream(streamId_, StreamCompletionSignal::GRACEFUL);
     } break;
     case State::CLOSED:

src/automata/StreamSubscriptionResponderBase.cpp

@@ -27,7 +27,8 @@ void StreamSubscriptionResponderBase::onComplete() {
   switch (state_) {
     case State::RESPONDING: {
       state_ = State::CLOSED;
-      connection_->outputFrameOrEnqueue(Frame_RESPONSE::complete(streamId_).serializeOut());
+      connection_->outputFrameOrEnqueue(
+          Frame_RESPONSE::complete(streamId_).serializeOut());
       connection_->endStream(streamId_, StreamCompletionSignal::GRACEFUL);
     } break;
     case State::CLOSED:
@@ -40,7 +41,8 @@ void StreamSubscriptionResponderBase::onError(folly::exception_wrapper ex) {
     case State::RESPONDING: {
       state_ = State::CLOSED;
       auto msg = ex.what().toStdString();
-      connection_->outputFrameOrEnqueue(Frame_ERROR::applicationError(streamId_, msg).serializeOut());
+      connection_->outputFrameOrEnqueue(
+          Frame_ERROR::applicationError(streamId_, msg).serializeOut());
       connection_->endStream(streamId_, StreamCompletionSignal::ERROR);
     } break;
     case State::CLOSED:

src/mixins/ConsumerMixin-inl.h

@@ -44,7 +44,8 @@ void ConsumerMixin<Frame, Base>::sendRequests() {
   toSync = pendingAllowance_.drainWithLimit(toSync);
   if (toSync > 0) {
     Base::connection_->outputFrameOrEnqueue(
-        Frame_REQUEST_N(Base::streamId_, static_cast<uint32_t>(toSync)).serializeOut());
+        Frame_REQUEST_N(Base::streamId_, static_cast<uint32_t>(toSync))
+            .serializeOut());
   }
 }