RSocket Test Bootstrap (#348)

Getting started with unit tests for the new RSocket APIs.

The first test just starts a server and lets it shut down while leaving the contextual scope.

Author: benjchristensen

CMakeLists.txt

@@ -376,15 +376,9 @@ add_dependencies(tcpresumeserver gmock)
 add_subdirectory(experimental/yarpl)
 
 ########################################
-# Examples
+# RSocket Experimental
 ########################################
 
-add_library(
-        reactivesocket_examples_util
-        examples/util/ExampleSubscriber.cpp
-        examples/util/ExampleSubscriber.h
-)
-
 add_library(
         rsocket_experimental
         experimental/rsocket/RSocket.h
@@ -406,7 +400,7 @@ add_library(
         experimental/rsocket-src/transports/TcpConnectionAcceptor.cpp
         experimental/rsocket/transports/TcpConnectionFactory.h
         experimental/rsocket-src/transports/TcpConnectionFactory.cpp
-        )
+)
 
 add_dependencies(rsocket_experimental ReactiveStreams)
 
@@ -418,6 +412,39 @@ target_include_directories(rsocket_experimental PUBLIC "${PROJECT_SOURCE_DIR}/ex
 #include_directories(${CMAKE_CURRENT_BINARY_DIR}/experimental)
 
 
+add_executable(
+        rsocket_tests
+        experimental/rsocket-test/RSocketClientServerTest.cpp
+        experimental/rsocket-test/handlers/HelloStreamRequestHandler.h
+        experimental/rsocket-test/handlers/HelloStreamRequestHandler.cpp
+        experimental/rsocket-test/handlers/HelloStreamSubscription.h
+        experimental/rsocket-test/handlers/HelloStreamSubscription.cpp
+)
+
+target_link_libraries(
+        rsocket_tests
+        rsocket_experimental
+        ReactiveSocket
+        ${FOLLY_LIBRARIES}
+        ${GMOCK_LIBS}
+        ${GFLAGS_LIBRARY}
+        ${GLOG_LIBRARY}
+        ${CMAKE_THREAD_LIBS_INIT})
+
+add_dependencies(rsocket_tests gmock rsocket_experimental)
+
+add_test(NAME RSocketTests COMMAND rsocket_tests)
+
+########################################
+# Examples
+########################################
+
+add_library(
+        reactivesocket_examples_util
+        examples/util/ExampleSubscriber.cpp
+        examples/util/ExampleSubscriber.h
+)
+
 target_link_libraries(
         reactivesocket_examples_util
         rsocket_experimental

experimental/rsocket-test/RSocketClientServerTest.cpp

@@ -0,0 +1,33 @@
+// Copyright 2004-present Facebook. All Rights Reserved.
+
+#include <utility>
+
+#include <folly/io/IOBuf.h>
+#include <gmock/gmock.h>
+
+#include "handlers/HelloStreamRequestHandler.h"
+#include "rsocket/RSocket.h"
+#include "rsocket/RSocket.h"
+#include "rsocket/transports/TcpConnectionAcceptor.h"
+
+using namespace rsocket;
+using namespace rsocket::tests;
+
+TEST(RSocketClientServer, StartAndShutdown) {
+  {
+    TcpConnectionAcceptor::Options opts;
+    opts.port = 9889;
+    opts.threads = 2;
+
+    // RSocket server accepting on TCP
+    auto rs = RSocket::createServer(
+        std::make_unique<TcpConnectionAcceptor>(std::move(opts)));
+
+    // global request handler
+    auto handler = std::make_shared<HelloStreamRequestHandler>();
+    // start server
+    rs->start([handler](auto r) { return handler; });
+  }
+  // everything should cleanly shut down here
+  // TODO anything to assert here?
+}

experimental/rsocket-test/handlers/HelloStreamRequestHandler.cpp

@@ -0,0 +1,34 @@
+// Copyright 2004-present Facebook. All Rights Reserved.
+
+#include "HelloStreamRequestHandler.h"
+#include <string>
+#include "HelloStreamSubscription.h"
+
+using namespace ::reactivesocket;
+
+namespace rsocket {
+namespace tests {
+/// Handles a new inbound Stream requested by the other end.
+void HelloStreamRequestHandler::handleRequestStream(
+    Payload request,
+    StreamId streamId,
+    const std::shared_ptr<Subscriber<Payload>>& response) noexcept {
+  LOG(INFO) << "HelloStreamRequestHandler.handleRequestStream " << request;
+
+  // string from payload data
+  const char* p = reinterpret_cast<const char*>(request.data->data());
+  auto requestString = std::string(p, request.data->length());
+
+  response->onSubscribe(
+      std::make_shared<HelloStreamSubscription>(response, requestString, 10));
+}
+
+std::shared_ptr<StreamState> HelloStreamRequestHandler::handleSetupPayload(
+    ReactiveSocket& socket,
+    ConnectionSetupPayload request) noexcept {
+  LOG(INFO) << "HelloStreamRequestHandler.handleSetupPayload " << request;
+  // TODO what should this do?
+  return nullptr;
+}
+}
+}

experimental/rsocket-test/handlers/HelloStreamRequestHandler.h

@@ -0,0 +1,30 @@
+// Copyright 2004-present Facebook. All Rights Reserved.
+
+#pragma once
+
+#include <folly/ExceptionWrapper.h>
+#include "src/NullRequestHandler.h"
+#include "src/Payload.h"
+#include "src/ReactiveStreamsCompat.h"
+#include "src/StandardReactiveSocket.h"
+#include "src/SubscriptionBase.h"
+
+namespace rsocket {
+namespace tests {
+
+class HelloStreamRequestHandler : public reactivesocket::DefaultRequestHandler {
+ public:
+  /// Handles a new inbound Stream requested by the other end.
+  void handleRequestStream(
+      reactivesocket::Payload request,
+      reactivesocket::StreamId streamId,
+      const std::shared_ptr<
+          reactivesocket::Subscriber<reactivesocket::Payload>>&
+          response) noexcept override;
+
+  std::shared_ptr<reactivesocket::StreamState> handleSetupPayload(
+      reactivesocket::ReactiveSocket&,
+      reactivesocket::ConnectionSetupPayload request) noexcept override;
+};
+}
+}

experimental/rsocket-test/handlers/HelloStreamSubscription.cpp

@@ -0,0 +1,54 @@
+// Copyright 2004-present Facebook. All Rights Reserved.
+
+#include "HelloStreamSubscription.h"
+
+#include <sstream>
+#include <string>
+
+using namespace reactivesocket;
+
+namespace rsocket {
+namespace tests {
+
+// Emit a stream of ints starting at 0 until number of ints
+// emitted matches 'numberToEmit' at which point onComplete()
+// will be emitted.
+//
+// On each invocation will restrict emission to number of requested.
+//
+// This method has no concurrency since SubscriptionBase
+// schedules this on an Executor sequentially
+void HelloStreamSubscription::requestImpl(size_t n) noexcept {
+  LOG(INFO) << "requested=" << n << " currentElem=" << currentElem_
+            << " numberToEmit=" << numberToEmit_;
+
+  if (numberToEmit_ == 0) {
+    subscriber_->onComplete();
+    return;
+  }
+  for (size_t i = 0; i < n; i++) {
+    if (cancelled_) {
+      LOG(INFO) << "emission stopped by cancellation";
+      return;
+    }
+    std::stringstream ss;
+    ss << "Hello " << name_ << " " << currentElem_ << "!";
+    std::string s = ss.str();
+    subscriber_->onNext(Payload(s));
+    // currentElem is used to track progress across requestImpl invocations
+    currentElem_++;
+    // break the loop and complete the stream if numberToEmit_ is matched
+    if (currentElem_ == numberToEmit_) {
+      subscriber_->onComplete();
+      return;
+    }
+  }
+}
+
+void HelloStreamSubscription::cancelImpl() noexcept {
+  LOG(INFO) << "cancellation received";
+  // simple cancellation token (nothing to shut down, just stop next loop)
+  cancelled_ = true;
+}
+}
+}

experimental/rsocket-test/handlers/HelloStreamSubscription.h

@@ -0,0 +1,40 @@
+// Copyright 2004-present Facebook. All Rights Reserved.
+
+#pragma once
+
+#include <string>
+#include "src/DuplexConnection.h"
+#include "src/Payload.h"
+#include "src/SubscriptionBase.h"
+
+namespace rsocket {
+namespace tests {
+
+/// Emits a stream of ints
+class HelloStreamSubscription : public reactivesocket::SubscriptionBase {
+ public:
+  explicit HelloStreamSubscription(
+      std::shared_ptr<reactivesocket::Subscriber<reactivesocket::Payload>>
+          subscriber,
+      std::string name,
+      size_t numberToEmit = 2)
+      : ExecutorBase(reactivesocket::defaultExecutor()),
+        subscriber_(std::move(subscriber)),
+        name_(std::move(name)),
+        numberToEmit_(numberToEmit),
+        cancelled_(false) {}
+
+ private:
+  // Subscription methods
+  void requestImpl(size_t n) noexcept override;
+  void cancelImpl() noexcept override;
+
+  std::shared_ptr<reactivesocket::Subscriber<reactivesocket::Payload>>
+      subscriber_;
+  std::string name_;
+  size_t numberToEmit_;
+  size_t currentElem_ = 0;
+  std::atomic_bool cancelled_;
+};
+}
+}