folly/wangle/channel/HandlerContext-inl.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 namespace folly { namespace wangle {
  20
  21 class PipelineContext {
  22  public:
  23   virtual ~PipelineContext() = default;
  24
  25   virtual void attachPipeline() = 0;
  26   virtual void detachPipeline() = 0;
  27
  28   template <class H, class HandlerContext>
  29   void attachContext(H* handler, HandlerContext* ctx) {
  30     if (++handler->attachCount_ == 1) {
  31       handler->ctx_ = ctx;
  32     } else {
  33       handler->ctx_ = nullptr;
  34     }
  35   }
  36
  37   virtual void setNextIn(PipelineContext* ctx) = 0;
  38   virtual void setNextOut(PipelineContext* ctx) = 0;
  39 };
  40
  41 template <class In>
  42 class InboundLink {
  43  public:
  44   virtual ~InboundLink() = default;
  45   virtual void read(In msg) = 0;
  46   virtual void readEOF() = 0;
  47   virtual void readException(exception_wrapper e) = 0;
  48   virtual void transportActive() = 0;
  49   virtual void transportInactive() = 0;
  50 };
  51
  52 template <class Out>
  53 class OutboundLink {
  54  public:
  55   virtual ~OutboundLink() = default;
  56   virtual Future<void> write(Out msg) = 0;
  57   virtual Future<void> close() = 0;
  58 };
  59
  60 template <class P, class H, class Context>
  61 class ContextImplBase : public PipelineContext {
  62  public:
  63   ~ContextImplBase() = default;
  64
  65   H* getHandler() {
  66     return handler_.get();
  67   }
  68
  69   void initialize(P* pipeline, std::shared_ptr<H> handler) {
  70     pipeline_ = pipeline;
  71     handler_ = std::move(handler);
  72   }
  73
  74   // PipelineContext overrides
  75   void attachPipeline() override {
  76     if (!attached_) {
  77       this->attachContext(handler_.get(), impl_);
  78       handler_->attachPipeline(impl_);
  79       attached_ = true;
  80     }
  81   }
  82
  83   void detachPipeline() override {
  84     handler_->detachPipeline(impl_);
  85     attached_ = false;
  86   }
  87
  88   void setNextIn(PipelineContext* ctx) override {
  89     auto nextIn = dynamic_cast<InboundLink<typename H::rout>*>(ctx);
  90     if (nextIn) {
  91       nextIn_ = nextIn;
  92     } else {
  93       throw std::invalid_argument("inbound type mismatch");
  94     }
  95   }
  96
  97   void setNextOut(PipelineContext* ctx) override {
  98     auto nextOut = dynamic_cast<OutboundLink<typename H::wout>*>(ctx);
  99     if (nextOut) {
 100       nextOut_ = nextOut;
 101     } else {
 102       throw std::invalid_argument("outbound type mismatch");
 103     }
 104   }
 105
 106  protected:
 107   Context* impl_;
 108   P* pipeline_;
 109   std::shared_ptr<H> handler_;
 110   InboundLink<typename H::rout>* nextIn_{nullptr};
 111   OutboundLink<typename H::wout>* nextOut_{nullptr};
 112
 113  private:
 114   bool attached_{false};
 115   using DestructorGuard = typename P::DestructorGuard;
 116 };
 117
 118 template <class P, class H>
 119 class ContextImpl
 120   : public HandlerContext<typename H::rout,
 121                           typename H::wout>,
 122     public InboundLink<typename H::rin>,
 123     public OutboundLink<typename H::win>,
 124     public ContextImplBase<P, H, HandlerContext<typename H::rout,
 125                                                 typename H::wout>> {
 126  public:
 127   typedef typename H::rin Rin;
 128   typedef typename H::rout Rout;
 129   typedef typename H::win Win;
 130   typedef typename H::wout Wout;
 131   static const HandlerDir dir = HandlerDir::BOTH;
 132
 133   explicit ContextImpl(P* pipeline, std::shared_ptr<H> handler) {
 134     this->impl_ = this;
 135     this->initialize(pipeline, std::move(handler));
 136   }
 137
 138   // For StaticPipeline
 139   ContextImpl() {
 140     this->impl_ = this;
 141   }
 142
 143   ~ContextImpl() = default;
 144
 145   // HandlerContext overrides
 146   void fireRead(Rout msg) override {
 147     DestructorGuard dg(this->pipeline_);
 148     if (this->nextIn_) {
 149       this->nextIn_->read(std::forward<Rout>(msg));
 150     } else {
 151       LOG(WARNING) << "read reached end of pipeline";
 152     }
 153   }
 154
 155   void fireReadEOF() override {
 156     DestructorGuard dg(this->pipeline_);
 157     if (this->nextIn_) {
 158       this->nextIn_->readEOF();
 159     } else {
 160       LOG(WARNING) << "readEOF reached end of pipeline";
 161     }
 162   }
 163
 164   void fireReadException(exception_wrapper e) override {
 165     DestructorGuard dg(this->pipeline_);
 166     if (this->nextIn_) {
 167       this->nextIn_->readException(std::move(e));
 168     } else {
 169       LOG(WARNING) << "readException reached end of pipeline";
 170     }
 171   }
 172
 173   void fireTransportActive() override {
 174     DestructorGuard dg(this->pipeline_);
 175     if (this->nextIn_) {
 176       this->nextIn_->transportActive();
 177     }
 178   }
 179
 180   void fireTransportInactive() override {
 181     DestructorGuard dg(this->pipeline_);
 182     if (this->nextIn_) {
 183       this->nextIn_->transportInactive();
 184     }
 185   }
 186
 187   Future<void> fireWrite(Wout msg) override {
 188     DestructorGuard dg(this->pipeline_);
 189     if (this->nextOut_) {
 190       return this->nextOut_->write(std::forward<Wout>(msg));
 191     } else {
 192       LOG(WARNING) << "write reached end of pipeline";
 193       return makeFuture();
 194     }
 195   }
 196
 197   Future<void> fireClose() override {
 198     DestructorGuard dg(this->pipeline_);
 199     if (this->nextOut_) {
 200       return this->nextOut_->close();
 201     } else {
 202       LOG(WARNING) << "close reached end of pipeline";
 203       return makeFuture();
 204     }
 205   }
 206
 207   PipelineBase* getPipeline() override {
 208     return this->pipeline_;
 209   }
 210
 211   void setWriteFlags(WriteFlags flags) override {
 212     this->pipeline_->setWriteFlags(flags);
 213   }
 214
 215   WriteFlags getWriteFlags() override {
 216     return this->pipeline_->getWriteFlags();
 217   }
 218
 219   void setReadBufferSettings(
 220       uint64_t minAvailable,
 221       uint64_t allocationSize) override {
 222     this->pipeline_->setReadBufferSettings(minAvailable, allocationSize);
 223   }
 224
 225   std::pair<uint64_t, uint64_t> getReadBufferSettings() override {
 226     return this->pipeline_->getReadBufferSettings();
 227   }
 228
 229   // InboundLink overrides
 230   void read(Rin msg) override {
 231     DestructorGuard dg(this->pipeline_);
 232     this->handler_->read(this, std::forward<Rin>(msg));
 233   }
 234
 235   void readEOF() override {
 236     DestructorGuard dg(this->pipeline_);
 237     this->handler_->readEOF(this);
 238   }
 239
 240   void readException(exception_wrapper e) override {
 241     DestructorGuard dg(this->pipeline_);
 242     this->handler_->readException(this, std::move(e));
 243   }
 244
 245   void transportActive() override {
 246     DestructorGuard dg(this->pipeline_);
 247     this->handler_->transportActive(this);
 248   }
 249
 250   void transportInactive() override {
 251     DestructorGuard dg(this->pipeline_);
 252     this->handler_->transportInactive(this);
 253   }
 254
 255   // OutboundLink overrides
 256   Future<void> write(Win msg) override {
 257     DestructorGuard dg(this->pipeline_);
 258     return this->handler_->write(this, std::forward<Win>(msg));
 259   }
 260
 261   Future<void> close() override {
 262     DestructorGuard dg(this->pipeline_);
 263     return this->handler_->close(this);
 264   }
 265
 266  private:
 267   using DestructorGuard = typename P::DestructorGuard;
 268 };
 269
 270 template <class P, class H>
 271 class InboundContextImpl
 272   : public InboundHandlerContext<typename H::rout>,
 273     public InboundLink<typename H::rin>,
 274     public ContextImplBase<P, H, InboundHandlerContext<typename H::rout>> {
 275  public:
 276   typedef typename H::rin Rin;
 277   typedef typename H::rout Rout;
 278   typedef typename H::win Win;
 279   typedef typename H::wout Wout;
 280   static const HandlerDir dir = HandlerDir::IN;
 281
 282   explicit InboundContextImpl(P* pipeline, std::shared_ptr<H> handler) {
 283     this->impl_ = this;
 284     this->initialize(pipeline, std::move(handler));
 285   }
 286
 287   // For StaticPipeline
 288   InboundContextImpl() {
 289     this->impl_ = this;
 290   }
 291
 292   ~InboundContextImpl() = default;
 293
 294   // InboundHandlerContext overrides
 295   void fireRead(Rout msg) override {
 296     DestructorGuard dg(this->pipeline_);
 297     if (this->nextIn_) {
 298       this->nextIn_->read(std::forward<Rout>(msg));
 299     } else {
 300       LOG(WARNING) << "read reached end of pipeline";
 301     }
 302   }
 303
 304   void fireReadEOF() override {
 305     DestructorGuard dg(this->pipeline_);
 306     if (this->nextIn_) {
 307       this->nextIn_->readEOF();
 308     } else {
 309       LOG(WARNING) << "readEOF reached end of pipeline";
 310     }
 311   }
 312
 313   void fireReadException(exception_wrapper e) override {
 314     DestructorGuard dg(this->pipeline_);
 315     if (this->nextIn_) {
 316       this->nextIn_->readException(std::move(e));
 317     } else {
 318       LOG(WARNING) << "readException reached end of pipeline";
 319     }
 320   }
 321
 322   void fireTransportActive() override {
 323     DestructorGuard dg(this->pipeline_);
 324     if (this->nextIn_) {
 325       this->nextIn_->transportActive();
 326     }
 327   }
 328
 329   void fireTransportInactive() override {
 330     DestructorGuard dg(this->pipeline_);
 331     if (this->nextIn_) {
 332       this->nextIn_->transportInactive();
 333     }
 334   }
 335
 336   PipelineBase* getPipeline() override {
 337     return this->pipeline_;
 338   }
 339
 340   // InboundLink overrides
 341   void read(Rin msg) override {
 342     DestructorGuard dg(this->pipeline_);
 343     this->handler_->read(this, std::forward<Rin>(msg));
 344   }
 345
 346   void readEOF() override {
 347     DestructorGuard dg(this->pipeline_);
 348     this->handler_->readEOF(this);
 349   }
 350
 351   void readException(exception_wrapper e) override {
 352     DestructorGuard dg(this->pipeline_);
 353     this->handler_->readException(this, std::move(e));
 354   }
 355
 356   void transportActive() override {
 357     DestructorGuard dg(this->pipeline_);
 358     this->handler_->transportActive(this);
 359   }
 360
 361   void transportInactive() override {
 362     DestructorGuard dg(this->pipeline_);
 363     this->handler_->transportInactive(this);
 364   }
 365
 366  private:
 367   using DestructorGuard = typename P::DestructorGuard;
 368 };
 369
 370 template <class P, class H>
 371 class OutboundContextImpl
 372   : public OutboundHandlerContext<typename H::wout>,
 373     public OutboundLink<typename H::win>,
 374     public ContextImplBase<P, H, OutboundHandlerContext<typename H::wout>> {
 375  public:
 376   typedef typename H::rin Rin;
 377   typedef typename H::rout Rout;
 378   typedef typename H::win Win;
 379   typedef typename H::wout Wout;
 380   static const HandlerDir dir = HandlerDir::OUT;
 381
 382   explicit OutboundContextImpl(P* pipeline, std::shared_ptr<H> handler) {
 383     this->impl_ = this;
 384     this->initialize(pipeline, std::move(handler));
 385   }
 386
 387   // For StaticPipeline
 388   OutboundContextImpl() {
 389     this->impl_ = this;
 390   }
 391
 392   ~OutboundContextImpl() = default;
 393
 394   // OutboundHandlerContext overrides
 395   Future<void> fireWrite(Wout msg) override {
 396     DestructorGuard dg(this->pipeline_);
 397     if (this->nextOut_) {
 398       return this->nextOut_->write(std::forward<Wout>(msg));
 399     } else {
 400       LOG(WARNING) << "write reached end of pipeline";
 401       return makeFuture();
 402     }
 403   }
 404
 405   Future<void> fireClose() override {
 406     DestructorGuard dg(this->pipeline_);
 407     if (this->nextOut_) {
 408       return this->nextOut_->close();
 409     } else {
 410       LOG(WARNING) << "close reached end of pipeline";
 411       return makeFuture();
 412     }
 413   }
 414
 415   PipelineBase* getPipeline() override {
 416     return this->pipeline_;
 417   }
 418
 419   // OutboundLink overrides
 420   Future<void> write(Win msg) override {
 421     DestructorGuard dg(this->pipeline_);
 422     return this->handler_->write(this, std::forward<Win>(msg));
 423   }
 424
 425   Future<void> close() override {
 426     DestructorGuard dg(this->pipeline_);
 427     return this->handler_->close(this);
 428   }
 429
 430  private:
 431   using DestructorGuard = typename P::DestructorGuard;
 432 };
 433
 434 template <class Handler, class Pipeline>
 435 struct ContextType {
 436   typedef typename std::conditional<
 437     Handler::dir == HandlerDir::BOTH,
 438     ContextImpl<Pipeline, Handler>,
 439     typename std::conditional<
 440       Handler::dir == HandlerDir::IN,
 441       InboundContextImpl<Pipeline, Handler>,
 442       OutboundContextImpl<Pipeline, Handler>
 443     >::type>::type
 444   type;
 445 };
 446
 447 }} // folly::wangle