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