folly/wangle/channel/Pipeline.h

   1 /*
   2  * Copyright 2015 Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #pragma once
  18
  19 #include <folly/wangle/channel/HandlerContext.h>
  20 #include <folly/futures/Future.h>
  21 #include <folly/io/async/AsyncTransport.h>
  22 #include <folly/io/async/DelayedDestruction.h>
  23 #include <folly/ExceptionWrapper.h>
  24 #include <folly/Memory.h>
  25
  26 namespace folly { namespace wangle {
  27
  28 struct Nothing{};
  29
  30 /*
  31  * R is the inbound type, i.e. inbound calls start with pipeline.read(R)
  32  * W is the outbound type, i.e. outbound calls start with pipeline.write(W)
  33  *
  34  * Use Nothing for one of the types if your pipeline is unidirectional.
  35  * If R is Nothing, read(), readEOF(), and readException() will be disabled.
  36  * If W is Nothing, write() and close() will be disabled.
  37  */
  38 template <class R, class W = Nothing>
  39 class Pipeline : public DelayedDestruction {
  40  public:
  41   Pipeline();
  42   ~Pipeline();
  43
  44   std::shared_ptr<AsyncTransport> getTransport();
  45
  46   void setWriteFlags(WriteFlags flags);
  47   WriteFlags getWriteFlags();
  48
  49   void setReadBufferSettings(uint64_t minAvailable, uint64_t allocationSize);
  50   std::pair<uint64_t, uint64_t> getReadBufferSettings();
  51
  52   template <class T = R>
  53   typename std::enable_if<!std::is_same<T, Nothing>::value>::type
  54   read(R msg);
  55
  56   template <class T = R>
  57   typename std::enable_if<!std::is_same<T, Nothing>::value>::type
  58   readEOF();
  59
  60   template <class T = R>
  61   typename std::enable_if<!std::is_same<T, Nothing>::value>::type
  62   readException(exception_wrapper e);
  63
  64   template <class T = W>
  65   typename std::enable_if<!std::is_same<T, Nothing>::value, Future<void>>::type
  66   write(W msg);
  67
  68   template <class T = W>
  69   typename std::enable_if<!std::is_same<T, Nothing>::value, Future<void>>::type
  70   close();
  71
  72   template <class H>
  73   Pipeline& addBack(std::shared_ptr<H> handler);
  74
  75   template <class H>
  76   Pipeline& addBack(H&& handler);
  77
  78   template <class H>
  79   Pipeline& addBack(H* handler);
  80
  81   template <class H>
  82   Pipeline& addFront(std::shared_ptr<H> handler);
  83
  84   template <class H>
  85   Pipeline& addFront(H&& handler);
  86
  87   template <class H>
  88   Pipeline& addFront(H* handler);
  89
  90   template <class H>
  91   H* getHandler(int i);
  92
  93   void finalize();
  94
  95   // If one of the handlers owns the pipeline itself, use setOwner to ensure
  96   // that the pipeline doesn't try to detach the handler during destruction,
  97   // lest destruction ordering issues occur.
  98   // See thrift/lib/cpp2/async/Cpp2Channel.cpp for an example
  99   template <class H>
 100   bool setOwner(H* handler);
 101
 102   void attachTransport(std::shared_ptr<AsyncTransport> transport);
 103
 104   void detachTransport();
 105
 106  protected:
 107   explicit Pipeline(bool isStatic);
 108
 109   template <class Context>
 110   void addContextFront(Context* ctx);
 111
 112   void detachHandlers();
 113
 114  private:
 115   template <class Context>
 116   Pipeline& addHelper(std::shared_ptr<Context>&& ctx, bool front);
 117
 118   std::shared_ptr<AsyncTransport> transport_;
 119   WriteFlags writeFlags_{WriteFlags::NONE};
 120   std::pair<uint64_t, uint64_t> readBufferSettings_{2048, 2048};
 121
 122   bool isStatic_{false};
 123   std::shared_ptr<PipelineContext> owner_;
 124   std::vector<std::shared_ptr<PipelineContext>> ctxs_;
 125   std::vector<PipelineContext*> inCtxs_;
 126   std::vector<PipelineContext*> outCtxs_;
 127   InboundLink<R>* front_{nullptr};
 128   OutboundLink<W>* back_{nullptr};
 129 };
 130
 131 }}
 132
 133 namespace folly {
 134
 135 class AsyncSocket;
 136
 137 template <typename Pipeline>
 138 class PipelineFactory {
 139  public:
 140   virtual Pipeline* newPipeline(std::shared_ptr<AsyncSocket>) = 0;
 141   virtual ~PipelineFactory() {}
 142 };
 143
 144 }
 145
 146 #include <folly/wangle/channel/Pipeline-inl.h>