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 class PipelineManager {
  29  public:
  30   virtual ~PipelineManager() {}
  31   virtual void deletePipeline(PipelineBase* pipeline) = 0;
  32 };
  33
  34 class PipelineBase : public DelayedDestruction {
  35  public:
  36   virtual ~PipelineBase() {}
  37
  38   void setPipelineManager(PipelineManager* manager) {
  39     manager_ = manager;
  40   }
  41
  42   void deletePipeline() {
  43     if (manager_) {
  44       manager_->deletePipeline(this);
  45     }
  46   }
  47
  48   void setTransport(std::shared_ptr<AsyncTransport> transport) {
  49     transport_ = transport;
  50   }
  51
  52   std::shared_ptr<AsyncTransport> getTransport() {
  53     return transport_;
  54   }
  55
  56  private:
  57   PipelineManager* manager_{nullptr};
  58   std::shared_ptr<AsyncTransport> transport_;
  59 };
  60
  61 struct Nothing{};
  62
  63 /*
  64  * R is the inbound type, i.e. inbound calls start with pipeline.read(R)
  65  * W is the outbound type, i.e. outbound calls start with pipeline.write(W)
  66  *
  67  * Use Nothing for one of the types if your pipeline is unidirectional.
  68  * If R is Nothing, read(), readEOF(), and readException() will be disabled.
  69  * If W is Nothing, write() and close() will be disabled.
  70  */
  71 template <class R, class W = Nothing>
  72 class Pipeline : public PipelineBase {
  73  public:
  74   Pipeline();
  75   ~Pipeline();
  76
  77   void setWriteFlags(WriteFlags flags);
  78   WriteFlags getWriteFlags();
  79
  80   void setReadBufferSettings(uint64_t minAvailable, uint64_t allocationSize);
  81   std::pair<uint64_t, uint64_t> getReadBufferSettings();
  82
  83   template <class T = R>
  84   typename std::enable_if<!std::is_same<T, Nothing>::value>::type
  85   read(R msg);
  86
  87   template <class T = R>
  88   typename std::enable_if<!std::is_same<T, Nothing>::value>::type
  89   readEOF();
  90
  91   template <class T = R>
  92   typename std::enable_if<!std::is_same<T, Nothing>::value>::type
  93   readException(exception_wrapper e);
  94
  95   template <class T = R>
  96   typename std::enable_if<!std::is_same<T, Nothing>::value>::type
  97   transportActive();
  98
  99   template <class T = R>
 100   typename std::enable_if<!std::is_same<T, Nothing>::value>::type
 101   transportInactive();
 102
 103   template <class T = W>
 104   typename std::enable_if<!std::is_same<T, Nothing>::value, Future<void>>::type
 105   write(W msg);
 106
 107   template <class T = W>
 108   typename std::enable_if<!std::is_same<T, Nothing>::value, Future<void>>::type
 109   close();
 110
 111   template <class H>
 112   Pipeline& addBack(std::shared_ptr<H> handler);
 113
 114   template <class H>
 115   Pipeline& addBack(H&& handler);
 116
 117   template <class H>
 118   Pipeline& addBack(H* handler);
 119
 120   template <class H>
 121   Pipeline& addFront(std::shared_ptr<H> handler);
 122
 123   template <class H>
 124   Pipeline& addFront(H&& handler);
 125
 126   template <class H>
 127   Pipeline& addFront(H* handler);
 128
 129   template <class H>
 130   H* getHandler(int i);
 131
 132   void finalize();
 133
 134   // If one of the handlers owns the pipeline itself, use setOwner to ensure
 135   // that the pipeline doesn't try to detach the handler during destruction,
 136   // lest destruction ordering issues occur.
 137   // See thrift/lib/cpp2/async/Cpp2Channel.cpp for an example
 138   template <class H>
 139   bool setOwner(H* handler);
 140
 141  protected:
 142   explicit Pipeline(bool isStatic);
 143
 144   template <class Context>
 145   void addContextFront(Context* ctx);
 146
 147   void detachHandlers();
 148
 149  private:
 150   template <class Context>
 151   Pipeline& addHelper(std::shared_ptr<Context>&& ctx, bool front);
 152
 153   WriteFlags writeFlags_{WriteFlags::NONE};
 154   std::pair<uint64_t, uint64_t> readBufferSettings_{2048, 2048};
 155
 156   bool isStatic_{false};
 157   std::shared_ptr<PipelineContext> owner_;
 158   std::vector<std::shared_ptr<PipelineContext>> ctxs_;
 159   std::vector<PipelineContext*> inCtxs_;
 160   std::vector<PipelineContext*> outCtxs_;
 161   InboundLink<R>* front_{nullptr};
 162   OutboundLink<W>* back_{nullptr};
 163 };
 164
 165 }}
 166
 167 namespace folly {
 168
 169 class AsyncSocket;
 170
 171 template <typename Pipeline>
 172 class PipelineFactory {
 173  public:
 174   virtual std::unique_ptr<Pipeline, folly::DelayedDestruction::Destructor>
 175   newPipeline(std::shared_ptr<AsyncSocket>) = 0;
 176
 177   virtual ~PipelineFactory() {}
 178 };
 179
 180 }
 181
 182 #include <folly/wangle/channel/Pipeline-inl.h>