Publisher responds with one single RESPONSE frame that has the COMPLETE flag set

Author: Cristinag

CMakeLists.txt

@@ -83,6 +83,9 @@ add_library(
   src/automata/ChannelResponder.cpp
   src/automata/ChannelResponder.h
   src/automata/RequestResponseRequester.cpp
+  src/automata/RequestResponseRequester.h
+  src/automata/RequestResponseResponder.cpp
+  src/automata/RequestResponseResponder.h
   src/automata/StreamSubscriptionRequesterBase.cpp
   src/automata/StreamSubscriptionRequesterBase.h
   src/automata/StreamSubscriptionResponderBase.cpp

TARGETS

@@ -39,6 +39,7 @@ cxx_library(
         'src/automata/ChannelRequester.cpp',
         'src/automata/ChannelResponder.cpp',
         'src/automata/RequestResponseRequester.cpp',
+        'src/automata/RequestResponseResponder.cpp',
         'src/automata/StreamSubscriptionRequesterBase.cpp',
         'src/automata/StreamSubscriptionResponderBase.cpp',
         'src/automata/StreamRequester.cpp',

src/automata/RequestResponseResponder.cpp

@@ -0,0 +1,82 @@
+// Copyright 2004-present Facebook. All Rights Reserved.
+
+#include "RequestResponseResponder.h"
+
+#include "src/ConnectionAutomaton.h"
+#include "src/Frame.h"
+#include "src/Payload.h"
+#include "src/ReactiveStreamsCompat.h"
+
+namespace reactivesocket {
+
+void RequestResponseResponderBase::onNext(Payload response) {
+  switch (state_) {
+    case State::RESPONDING: {
+      state_ = State::CLOSED;
+      Base::onNext(std::move(response), FrameFlags_COMPLETE);
+      connection_->endStream(streamId_, StreamCompletionSignal::GRACEFUL);
+      break;
+    }
+    case State::CLOSED:
+      break;
+  }
+}
+
+void RequestResponseResponderBase::onComplete() {
+  switch (state_) {
+    case State::RESPONDING: {
+      state_ = State::CLOSED;
+      connection_->outputFrameOrEnqueue(
+          Frame_RESPONSE::complete(streamId_).serializeOut());
+      connection_->endStream(streamId_, StreamCompletionSignal::GRACEFUL);
+    } break;
+    case State::CLOSED:
+      break;
+  }
+}
+
+
+void RequestResponseResponderBase::onError(folly::exception_wrapper ex) {
+  switch (state_) {
+    case State::RESPONDING: {
+      state_ = State::CLOSED;
+      auto msg = ex.what().toStdString();
+      connection_->outputFrameOrEnqueue(
+          Frame_ERROR::applicationError(streamId_, msg).serializeOut());
+      connection_->endStream(streamId_, StreamCompletionSignal::ERROR);
+    } break;
+    case State::CLOSED:
+      break;
+  }
+}
+
+void RequestResponseResponderBase::endStream(StreamCompletionSignal signal) {
+  switch (state_) {
+    case State::RESPONDING:
+      // Spontaneous ::endStream signal means an error.
+      DCHECK(StreamCompletionSignal::GRACEFUL != signal);
+      state_ = State::CLOSED;
+      break;
+    case State::CLOSED:
+      break;
+  }
+  Base::endStream(signal);
+}
+
+void RequestResponseResponderBase::onNextFrame(Frame_CANCEL&& frame) {
+  switch (state_) {
+    case State::RESPONDING:
+      state_ = State::CLOSED;
+      connection_->endStream(streamId_, StreamCompletionSignal::GRACEFUL);
+      break;
+    case State::CLOSED:
+      break;
+  }
+}
+
+std::ostream& RequestResponseResponderBase::logPrefix(std::ostream& os) {
+  return os << "RequestResponseResponder(" << &connection_ << ", "
+            << streamId_ << "): ";
+}
+
+}

src/automata/RequestResponseResponder.h

@@ -8,10 +8,11 @@
 #include "src/Frame.h"
 #include "src/Payload.h"
 #include "src/ReactiveStreamsCompat.h"
-#include "src/automata/StreamSubscriptionResponderBase.h"
 #include "src/mixins/ExecutorMixin.h"
 #include "src/mixins/LoggingMixin.h"
 #include "src/mixins/MemoryMixin.h"
+#include "src/mixins/MixinTerminator.h"
+#include "src/mixins/PublisherMixin.h"
 #include "src/mixins/SinkIfMixin.h"
 #include "src/mixins/StreamIfMixin.h"
 
@@ -25,24 +26,41 @@ enum class StreamCompletionSignal;
 
 /// Implementation of stream automaton that represents a RequestResponse
 /// responder
-class RequestResponseResponderBase : public StreamSubscriptionResponderBase {
-  using Base = StreamSubscriptionResponderBase;
+class RequestResponseResponderBase
+    : public LoggingMixin<PublisherMixin<Frame_RESPONSE, MixinTerminator>> {
+  using Base =  LoggingMixin<PublisherMixin<Frame_RESPONSE, MixinTerminator>>;
 
  public:
   using Base::Base;
 
+  /// @{
+  /// A Subscriber implementation exposed to the user of ReactiveSocket to
+  /// receive "response" payloads.
+  void onNext(Payload);
+
+  void onComplete();
+
+  void onError(folly::exception_wrapper);
+  /// @}
+
  protected:
   /// @{
+  void endStream(StreamCompletionSignal);
 
-  /// Don't need to intercept any frames, let them pass through.
+  /// Not all frames are intercepted, some just pass through.
   using Base::onNextFrame;
 
-  std::ostream& logPrefix(std::ostream& os) {
-    return os << "RequestResponseResponder(" << &connection_ << ", "
-              << streamId_ << "): ";
-  }
+  void onNextFrame(Frame_CANCEL&&);
+
+  std::ostream& logPrefix(std::ostream& os);
   /// @}
-};
+
+  /// State of the Subscription responder.
+  enum class State : uint8_t {
+    RESPONDING,
+    CLOSED,
+  } state_{State::RESPONDING};
+ };
 
 using RequestResponseResponder =
     SinkIfMixin<StreamIfMixin<LoggingMixin<ExecutorMixin<LoggingMixin<

src/mixins/LoggingMixin.h

@@ -69,6 +69,11 @@ class LoggingMixin : public Base {
     Base::onNext(std::move(payload));
   }
 
+  void onNext(Payload payload, FrameFlags flags) {
+    Base::logPrefix(LOG(INFO)) << "onNext(" << payload << ", " << flags << ")";
+    Base::onNext(std::move(payload), flags);
+  }
+
   void onComplete() {
     Base::logPrefix(LOG(INFO)) << "onComplete()";
     Base::onComplete();

src/mixins/PublisherMixin.h

@@ -29,8 +29,8 @@ class PublisherMixin : public Base {
     producingSubscription_.reset(&subscription);
   }
 
-  void onNext(Payload payload) {
-    ProducedFrame frame(Base::streamId_, FrameFlags_EMPTY, std::move(payload));
+  void onNext(Payload payload, FrameFlags flags = FrameFlags_EMPTY) {
+    ProducedFrame frame(Base::streamId_, flags, std::move(payload));
     Base::connection_->outputFrameOrEnqueue(frame.serializeOut());
   }
   /// @}

test/ReactiveSocketTest.cpp

@@ -447,11 +447,13 @@ TEST(ReactiveSocketTest, RequestResponse) {
   EXPECT_CALL(clientInput, onNext_(Equals(&originalPayload)))
       .InSequence(s)
       .WillOnce(Invoke([&](Payload&) { clientInputSub->cancel(); }));
+  // Client also receives onComplete() call since the response frame received
+  // had COMPELTE flag set
+  EXPECT_CALL(clientInput, onComplete_()).InSequence(s);
 
   EXPECT_CALL(serverOutputSub, cancel_()).InSequence(s).WillOnce(Invoke([&]() {
     serverOutput->onComplete();
   }));
-  EXPECT_CALL(clientInput, onComplete_()).InSequence(s);
 
   // Kick off the magic.
   clientSock->requestResponse(Payload(originalPayload->clone()), clientInput);