Fail resumption if it happens in different thread (#627)

rsocket · Aug 9, 2017 · 0c81960 · 0c81960
1 parent 13b4ba9
commit 0c81960
Show file tree

Hide file tree

Showing 4 changed files with 10 additions and 0 deletions.
diff --git a/rsocket/RSocketServer.cpp b/rsocket/RSocketServer.cpp
@@ -160,6 +160,9 @@ void RSocketServer::onRSocketResume(
   }
   auto serverState = std::move(result.value());
   CHECK(serverState);
+  if (!serverState->rSocketStateMachine_->isInEventBaseThread()) {
+    throw RSocketException("Trying to RESUME in a different worker thread");
+  }
   serverState->rSocketStateMachine_->resumeServer(
       std::move(frameTransport), resumeParams);
 }

diff --git a/rsocket/internal/SetupResumeAcceptor.cpp b/rsocket/internal/SetupResumeAcceptor.cpp
@@ -170,6 +170,7 @@ void SetupResumeAcceptor::processFrame(
       } catch (const std::exception& exn) {
         folly::exception_wrapper ew{std::current_exception(), exn};
         auto errFrame = Frame_ERROR::rejectedResume(ew.what().toStdString());
+        VLOG(3) << "Out: " << errFrame;
         transport->outputFrameOrEnqueue(
             serializer->serializeOut(std::move(errFrame)));
         close(std::move(transport), std::move(ew));

diff --git a/rsocket/statemachine/RSocketStateMachine.cpp b/rsocket/statemachine/RSocketStateMachine.cpp
@@ -880,6 +880,10 @@ void RSocketStateMachine::connectClientSendSetup(
   connect(std::move(frameTransport), true, ProtocolVersion::Unknown);
 }
 
+bool RSocketStateMachine::isInEventBaseThread() {
+  return dynamic_cast<folly::EventBase*>(&executor_)->isInEventBaseThread();
+}
+
 void RSocketStateMachine::writeNewStream(
     StreamId streamId,
     StreamType streamType,

diff --git a/rsocket/statemachine/RSocketStateMachine.h b/rsocket/statemachine/RSocketStateMachine.h
@@ -211,6 +211,8 @@ class RSocketStateMachine final
     return connectionEvents_;
   }
 
+  bool isInEventBaseThread();
+
  private:
 
   bool connect(