telnet server

Summary: similar to https://github.com/netty/netty/tree/master/example/src/main/java/io/netty/example/telnet

Test Plan:
fbconfig folly/wangle/example/telnet; fbmake dbg
_bin/folly/wangle/example/telnet_server --port=8080
telnet localhost 8080

Still a couple sharp edges:
* No easy way to wait for ServerBootstrap termination.
* Pipelines always have to call attachReadCallback
* a bunch of missing methods in pipeline still, like channelActive

Reviewed By: hans@fb.com

Subscribers: doug, fugalh, folly-diffs@, jsedgwick, yfeldblum, chalfant

FB internal diff: D1959172

Signature: t1:1959172:1427993978:463f237036996451187e3ef3983cf2b4e89685ef

folly/wangle/bootstrap/BootstrapTest.cpp

@@ -32,7 +32,8 @@ typedef ClientBootstrap<BytesPipeline> TestClient;
 
 class TestClientPipelineFactory : public PipelineFactory<BytesPipeline> {
  public:
-  BytesPipeline* newPipeline(std::shared_ptr<AsyncSocket> sock) {
+  std::unique_ptr<BytesPipeline, folly::DelayedDestruction::Destructor>
+  newPipeline(std::shared_ptr<AsyncSocket> sock) {
     CHECK(sock->good());
 
     // We probably aren't connected immedately, check after a small delay
@@ -45,9 +46,12 @@ class TestClientPipelineFactory : public PipelineFactory<BytesPipeline> {
 
 class TestPipelineFactory : public PipelineFactory<BytesPipeline> {
  public:
-  BytesPipeline* newPipeline(std::shared_ptr<AsyncSocket> sock) {
+  std::unique_ptr<BytesPipeline, folly::DelayedDestruction::Destructor> newPipeline(
+    std::shared_ptr<AsyncSocket> sock) {
+
     pipelines++;
-    return new BytesPipeline();
+    return std::unique_ptr<BytesPipeline, folly::DelayedDestruction::Destructor>(
+      new BytesPipeline());
   }
   std::atomic<int> pipelines{0};
 };
@@ -279,8 +283,13 @@ template <typename HandlerPipeline>
 class TestHandlerPipelineFactory
     : public PipelineFactory<ServerBootstrap<BytesPipeline>::AcceptPipeline> {
  public:
-  ServerBootstrap<BytesPipeline>::AcceptPipeline* newPipeline(std::shared_ptr<AsyncSocket>) {
-    auto pipeline = new ServerBootstrap<BytesPipeline>::AcceptPipeline;
+  std::unique_ptr<ServerBootstrap<BytesPipeline>::AcceptPipeline,
+                  folly::DelayedDestruction::Destructor>
+  newPipeline(std::shared_ptr<AsyncSocket>) {
+
+    std::unique_ptr<ServerBootstrap<BytesPipeline>::AcceptPipeline,
+                    folly::DelayedDestruction::Destructor> pipeline(
+                      new ServerBootstrap<BytesPipeline>::AcceptPipeline);
     pipeline->addBack(HandlerPipeline());
     return pipeline;
   }

folly/wangle/bootstrap/ClientBootstrap.h

@@ -34,10 +34,10 @@ class ClientBootstrap {
   }
   ClientBootstrap* connect(SocketAddress address) {
     DCHECK(pipelineFactory_);
-    pipeline_.reset(
+    pipeline_=
       pipelineFactory_->newPipeline(
-        AsyncSocket::newSocket(EventBaseManager::get()->getEventBase(), address)
-      ));
+        AsyncSocket::newSocket(
+          EventBaseManager::get()->getEventBase(), address));
     return this;
   }
 

folly/wangle/bootstrap/ServerBootstrap-inl.h

@@ -32,27 +32,35 @@ class ServerAcceptor
   typedef std::unique_ptr<Pipeline,
                           folly::DelayedDestruction::Destructor> PipelinePtr;
 
-  class ServerConnection : public wangle::ManagedConnection {
+  class ServerConnection : public wangle::ManagedConnection,
+                           public wangle::PipelineManager {
    public:
     explicit ServerConnection(PipelinePtr pipeline)
-        : pipeline_(std::move(pipeline)) {}
-
-    ~ServerConnection() {
+        : pipeline_(std::move(pipeline)) {
+      pipeline_->setPipelineManager(this);
     }
 
-    void timeoutExpired() noexcept {
+    ~ServerConnection() {}
+
+    void timeoutExpired() noexcept override {
     }
 
-    void describe(std::ostream& os) const {}
-    bool isBusy() const {
+    void describe(std::ostream& os) const override {}
+    bool isBusy() const override {
       return false;
     }
-    void notifyPendingShutdown() {}
-    void closeWhenIdle() {}
-    void dropConnection() {
+    void notifyPendingShutdown() override {}
+    void closeWhenIdle() override {}
+    void dropConnection() override {
       delete this;
     }
-    void dumpConnectionState(uint8_t loglevel) {}
+    void dumpConnectionState(uint8_t loglevel) override {}
+
+    void deletePipeline(wangle::PipelineBase* p) override {
+      CHECK(p == pipeline_.get());
+      delete this;
+    }
+
    private:
     PipelinePtr pipeline_;
   };
@@ -178,8 +186,11 @@ class DefaultAcceptPipelineFactory
   typedef wangle::Pipeline<void*> AcceptPipeline;
 
  public:
-  AcceptPipeline* newPipeline(std::shared_ptr<AsyncSocket>) {
-    return new AcceptPipeline;
+  std::unique_ptr<AcceptPipeline, folly::DelayedDestruction::Destructor>
+    newPipeline(std::shared_ptr<AsyncSocket>) {
+
+    return std::unique_ptr<AcceptPipeline, folly::DelayedDestruction::Destructor>
+      (new AcceptPipeline);
   }
 };
 

folly/wangle/bootstrap/ServerBootstrap.h

@@ -284,6 +284,13 @@ class ServerBootstrap {
       }
       sockets_->clear();
     }
+    if (!stopped_) {
+      stopped_ = true;
+      // stopBaton_ may be null if ServerBootstrap has been std::move'd
+      if (stopBaton_) {
+        stopBaton_->post();
+      }
+    }
   }
 
   void join() {
@@ -295,6 +302,13 @@ class ServerBootstrap {
     }
   }
 
+  void waitForStop() {
+    if (!stopped_) {
+      CHECK(stopBaton_);
+      stopBaton_->wait();
+    }
+  }
+
   /*
    * Get the list of listening sockets
    */
@@ -328,6 +342,10 @@ class ServerBootstrap {
     std::make_shared<DefaultAcceptPipelineFactory>()};
   std::shared_ptr<ServerSocketFactory> socketFactory_{
     std::make_shared<AsyncServerSocketFactory>()};
+
+  std::unique_ptr<folly::Baton<>> stopBaton_{
+    folly::make_unique<folly::Baton<>>()};
+  bool stopped_{false};
 };
 
 } // namespace

folly/wangle/channel/AsyncSocketHandler.h

@@ -94,6 +94,7 @@ class AsyncSocketHandler
       detachReadCallback();
       socket_->closeNow();
     }
+    ctx->getPipeline()->deletePipeline();
     return folly::makeFuture();
   }
 

folly/wangle/channel/HandlerContext-inl.h

@@ -201,6 +201,10 @@ class ContextImpl
     }
   }
 
+  PipelineBase* getPipeline() override {
+    return this->pipeline_;
+  }
+
   std::shared_ptr<AsyncTransport> getTransport() override {
     return this->pipeline_->getTransport();
   }
@@ -306,6 +310,10 @@ class InboundContextImpl
     }
   }
 
+  PipelineBase* getPipeline() override {
+    return this->pipeline_;
+  }
+
   std::shared_ptr<AsyncTransport> getTransport() override {
     return this->pipeline_->getTransport();
   }
@@ -375,6 +383,10 @@ class OutboundContextImpl
     }
   }
 
+  PipelineBase* getPipeline() override {
+    return this->pipeline_;
+  }
+
   std::shared_ptr<AsyncTransport> getTransport() override {
     return this->pipeline_->getTransport();
   }

folly/wangle/channel/HandlerContext.h

@@ -22,6 +22,8 @@
 
 namespace folly { namespace wangle {
 
+class PipelineBase;
+
 template <class In, class Out>
 class HandlerContext {
  public:
@@ -34,6 +36,8 @@ class HandlerContext {
   virtual Future<void> fireWrite(Out msg) = 0;
   virtual Future<void> fireClose() = 0;
 
+  virtual PipelineBase* getPipeline() = 0;
+
   virtual std::shared_ptr<AsyncTransport> getTransport() = 0;
 
   virtual void setWriteFlags(WriteFlags flags) = 0;
@@ -64,6 +68,8 @@ class InboundHandlerContext {
   virtual void fireReadEOF() = 0;
   virtual void fireReadException(exception_wrapper e) = 0;
 
+  virtual PipelineBase* getPipeline() = 0;
+
   virtual std::shared_ptr<AsyncTransport> getTransport() = 0;
 
   // TODO Need get/set writeFlags, readBufferSettings? Probably not.
@@ -79,6 +85,8 @@ class OutboundHandlerContext {
   virtual Future<void> fireWrite(Out msg) = 0;
   virtual Future<void> fireClose() = 0;
 
+  virtual PipelineBase* getPipeline() = 0;
+
   virtual std::shared_ptr<AsyncTransport> getTransport() = 0;
 };
 

folly/wangle/channel/Pipeline.h

@@ -25,6 +25,28 @@
 
 namespace folly { namespace wangle {
 
+class PipelineManager {
+ public:
+  virtual ~PipelineManager() {}
+  virtual void deletePipeline(PipelineBase* pipeline) = 0;
+};
+
+class PipelineBase {
+ public:
+  void setPipelineManager(PipelineManager* manager) {
+    manager_ = manager;
+  }
+
+  void deletePipeline() {
+    if (manager_) {
+      manager_->deletePipeline(this);
+    }
+  }
+
+ private:
+  PipelineManager* manager_{nullptr};
+};
+
 struct Nothing{};
 
 /*
@@ -36,7 +58,7 @@ struct Nothing{};
  * If W is Nothing, write() and close() will be disabled.
  */
 template <class R, class W = Nothing>
-class Pipeline : public DelayedDestruction {
+class Pipeline : public PipelineBase, public DelayedDestruction {
  public:
   Pipeline();
   ~Pipeline();
@@ -137,7 +159,9 @@ class AsyncSocket;
 template <typename Pipeline>
 class PipelineFactory {
  public:
-  virtual Pipeline* newPipeline(std::shared_ptr<AsyncSocket>) = 0;
+  virtual std::unique_ptr<Pipeline, folly::DelayedDestruction::Destructor>
+  newPipeline(std::shared_ptr<AsyncSocket>) = 0;
+
   virtual ~PipelineFactory() {}
 };
 

folly/wangle/codec/StringCodec.h

@@ -23,15 +23,14 @@ namespace folly { namespace wangle {
 /*
  * StringCodec converts a pipeline from IOBufs to std::strings.
  */
-class StringCodec : public Handler<IOBufQueue&, std::string,
+class StringCodec : public Handler<std::unique_ptr<IOBuf>, std::string,
                                    std::string, std::unique_ptr<IOBuf>> {
  public:
   typedef typename Handler<
-   IOBufQueue&, std::string,
+   std::unique_ptr<IOBuf>, std::string,
    std::string, std::unique_ptr<IOBuf>>::Context Context;
 
-  void read(Context* ctx, IOBufQueue& q) override {
-    auto buf = q.pop_front();
+  void read(Context* ctx, std::unique_ptr<IOBuf> buf) override {
     buf->coalesce();
     std::string data((const char*)buf->data(), buf->length());
 

folly/wangle/service/ServiceTest.cpp

@@ -16,6 +16,7 @@
 #include <gtest/gtest.h>
 
 #include <folly/wangle/codec/StringCodec.h>
+#include <folly/wangle/codec/ByteToMessageCodec.h>
 #include <folly/wangle/service/ClientDispatcher.h>
 #include <folly/wangle/service/ServerDispatcher.h>
 #include <folly/wangle/service/Service.h>
@@ -26,6 +27,14 @@ using namespace wangle;
 
 typedef Pipeline<IOBufQueue&, std::string> ServicePipeline;
 
+class SimpleDecode : public ByteToMessageCodec {
+ public:
+  virtual std::unique_ptr<IOBuf> decode(
+    Context* ctx, IOBufQueue& buf, size_t&) {
+    return buf.move();
+  }
+};
+
 class EchoService : public Service<std::string, std::string> {
  public:
   virtual Future<std::string> operator()(std::string req) override {
@@ -45,10 +54,12 @@ class ServerPipelineFactory
     : public PipelineFactory<ServicePipeline> {
  public:
 
-  ServicePipeline* newPipeline(
-      std::shared_ptr<AsyncSocket> socket) override {
-    auto pipeline = new ServicePipeline();
+  std::unique_ptr<ServicePipeline, folly::DelayedDestruction::Destructor>
+  newPipeline(std::shared_ptr<AsyncSocket> socket) override {
+    std::unique_ptr<ServicePipeline, folly::DelayedDestruction::Destructor> pipeline(
+      new ServicePipeline());
     pipeline->addBack(AsyncSocketHandler(socket));
+    pipeline->addBack(SimpleDecode());
     pipeline->addBack(StringCodec());
     pipeline->addBack(SerialServerDispatcher<Req, Resp>(&service_));
     pipeline->finalize();
@@ -63,11 +74,14 @@ template <typename Req, typename Resp>
 class ClientPipelineFactory : public PipelineFactory<ServicePipeline> {
  public:
 
-  ServicePipeline* newPipeline(
-      std::shared_ptr<AsyncSocket> socket) override {
-    auto pipeline = new ServicePipeline();
+  std::unique_ptr<ServicePipeline, folly::DelayedDestruction::Destructor>
+  newPipeline(std::shared_ptr<AsyncSocket> socket) override {
+    std::unique_ptr<ServicePipeline, folly::DelayedDestruction::Destructor> pipeline(
+      new ServicePipeline());
     pipeline->addBack(AsyncSocketHandler(socket));
+    pipeline->addBack(SimpleDecode());
     pipeline->addBack(StringCodec());
+    pipeline->finalize();
     return pipeline;
    }
 };