folly/wangle/channel/test/PipelineTest.cpp

   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 #include <folly/wangle/channel/Handler.h>
  18 #include <folly/wangle/channel/Pipeline.h>
  19 #include <folly/wangle/channel/StaticPipeline.h>
  20 #include <folly/wangle/channel/AsyncSocketHandler.h>
  21 #include <folly/wangle/channel/OutputBufferingHandler.h>
  22 #include <folly/wangle/channel/test/MockHandler.h>
  23 #include <gmock/gmock.h>
  24 #include <gtest/gtest.h>
  25
  26 using namespace folly;
  27 using namespace folly::wangle;
  28 using namespace testing;
  29
  30 typedef StrictMock<MockHandlerAdapter<int, int>> IntHandler;
  31
  32 ACTION(FireRead) {
  33   arg0->fireRead(arg1);
  34 }
  35
  36 ACTION(FireReadEOF) {
  37   arg0->fireReadEOF();
  38 }
  39
  40 ACTION(FireReadException) {
  41   arg0->fireReadException(arg1);
  42 }
  43
  44 ACTION(FireWrite) {
  45   arg0->fireWrite(arg1);
  46 }
  47
  48 ACTION(FireClose) {
  49   arg0->fireClose();
  50 }
  51
  52 // Test move only types, among other things
  53 TEST(PipelineTest, RealHandlersCompile) {
  54   EventBase eb;
  55   auto socket = AsyncSocket::newSocket(&eb);
  56   // static
  57   {
  58     StaticPipeline<IOBufQueue&, std::unique_ptr<IOBuf>,
  59       AsyncSocketHandler,
  60       OutputBufferingHandler>
  61     pipeline{AsyncSocketHandler(socket), OutputBufferingHandler()};
  62     EXPECT_TRUE(pipeline.getHandler<AsyncSocketHandler>(0));
  63     EXPECT_TRUE(pipeline.getHandler<OutputBufferingHandler>(1));
  64   }
  65   // dynamic
  66   {
  67     Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
  68     pipeline
  69       .addBack(AsyncSocketHandler(socket))
  70       .addBack(OutputBufferingHandler())
  71       .finalize();
  72     EXPECT_TRUE(pipeline.getHandler<AsyncSocketHandler>(0));
  73     EXPECT_TRUE(pipeline.getHandler<OutputBufferingHandler>(1));
  74   }
  75 }
  76
  77 // Test that handlers correctly fire the next handler when directed
  78 TEST(PipelineTest, FireActions) {
  79   IntHandler handler1;
  80   IntHandler handler2;
  81
  82   {
  83     InSequence sequence;
  84     EXPECT_CALL(handler2, attachPipeline(_));
  85     EXPECT_CALL(handler1, attachPipeline(_));
  86   }
  87
  88   StaticPipeline<int, int, IntHandler, IntHandler>
  89   pipeline(&handler1, &handler2);
  90
  91   EXPECT_CALL(handler1, read_(_, _)).WillOnce(FireRead());
  92   EXPECT_CALL(handler2, read_(_, _)).Times(1);
  93   pipeline.read(1);
  94
  95   EXPECT_CALL(handler1, readEOF(_)).WillOnce(FireReadEOF());
  96   EXPECT_CALL(handler2, readEOF(_)).Times(1);
  97   pipeline.readEOF();
  98
  99   EXPECT_CALL(handler1, readException(_, _)).WillOnce(FireReadException());
 100   EXPECT_CALL(handler2, readException(_, _)).Times(1);
 101   pipeline.readException(make_exception_wrapper<std::runtime_error>("blah"));
 102
 103   EXPECT_CALL(handler2, write_(_, _)).WillOnce(FireWrite());
 104   EXPECT_CALL(handler1, write_(_, _)).Times(1);
 105   EXPECT_NO_THROW(pipeline.write(1).value());
 106
 107   EXPECT_CALL(handler2, close_(_)).WillOnce(FireClose());
 108   EXPECT_CALL(handler1, close_(_)).Times(1);
 109   EXPECT_NO_THROW(pipeline.close().value());
 110
 111   {
 112     InSequence sequence;
 113     EXPECT_CALL(handler1, detachPipeline(_));
 114     EXPECT_CALL(handler2, detachPipeline(_));
 115   }
 116 }
 117
 118 // Test that nothing bad happens when actions reach the end of the pipeline
 119 // (a warning will be logged, however)
 120 TEST(PipelineTest, ReachEndOfPipeline) {
 121   IntHandler handler;
 122   EXPECT_CALL(handler, attachPipeline(_));
 123   StaticPipeline<int, int, IntHandler>
 124   pipeline(&handler);
 125
 126   EXPECT_CALL(handler, read_(_, _)).WillOnce(FireRead());
 127   pipeline.read(1);
 128
 129   EXPECT_CALL(handler, readEOF(_)).WillOnce(FireReadEOF());
 130   pipeline.readEOF();
 131
 132   EXPECT_CALL(handler, readException(_, _)).WillOnce(FireReadException());
 133   pipeline.readException(make_exception_wrapper<std::runtime_error>("blah"));
 134
 135   EXPECT_CALL(handler, write_(_, _)).WillOnce(FireWrite());
 136   EXPECT_NO_THROW(pipeline.write(1).value());
 137
 138   EXPECT_CALL(handler, close_(_)).WillOnce(FireClose());
 139   EXPECT_NO_THROW(pipeline.close().value());
 140
 141   EXPECT_CALL(handler, detachPipeline(_));
 142 }
 143
 144 // Test having the last read handler turn around and write
 145 TEST(PipelineTest, TurnAround) {
 146   IntHandler handler1;
 147   IntHandler handler2;
 148
 149   {
 150     InSequence sequence;
 151     EXPECT_CALL(handler2, attachPipeline(_));
 152     EXPECT_CALL(handler1, attachPipeline(_));
 153   }
 154
 155   StaticPipeline<int, int, IntHandler, IntHandler>
 156   pipeline(&handler1, &handler2);
 157
 158   EXPECT_CALL(handler1, read_(_, _)).WillOnce(FireRead());
 159   EXPECT_CALL(handler2, read_(_, _)).WillOnce(FireWrite());
 160   EXPECT_CALL(handler1, write_(_, _)).Times(1);
 161   pipeline.read(1);
 162
 163   {
 164     InSequence sequence;
 165     EXPECT_CALL(handler1, detachPipeline(_));
 166     EXPECT_CALL(handler2, detachPipeline(_));
 167   }
 168 }
 169
 170 TEST(PipelineTest, DynamicFireActions) {
 171   IntHandler handler1, handler2, handler3;
 172   EXPECT_CALL(handler2, attachPipeline(_));
 173   StaticPipeline<int, int, IntHandler>
 174   pipeline(&handler2);
 175
 176   {
 177     InSequence sequence;
 178     EXPECT_CALL(handler3, attachPipeline(_));
 179     EXPECT_CALL(handler1, attachPipeline(_));
 180   }
 181
 182   pipeline
 183     .addFront(&handler1)
 184     .addBack(&handler3)
 185     .finalize();
 186
 187   EXPECT_TRUE(pipeline.getHandler<IntHandler>(0));
 188   EXPECT_TRUE(pipeline.getHandler<IntHandler>(1));
 189   EXPECT_TRUE(pipeline.getHandler<IntHandler>(2));
 190
 191   EXPECT_CALL(handler1, read_(_, _)).WillOnce(FireRead());
 192   EXPECT_CALL(handler2, read_(_, _)).WillOnce(FireRead());
 193   EXPECT_CALL(handler3, read_(_, _)).Times(1);
 194   pipeline.read(1);
 195
 196   EXPECT_CALL(handler3, write_(_, _)).WillOnce(FireWrite());
 197   EXPECT_CALL(handler2, write_(_, _)).WillOnce(FireWrite());
 198   EXPECT_CALL(handler1, write_(_, _)).Times(1);
 199   EXPECT_NO_THROW(pipeline.write(1).value());
 200
 201   {
 202     InSequence sequence;
 203     EXPECT_CALL(handler1, detachPipeline(_));
 204     EXPECT_CALL(handler2, detachPipeline(_));
 205     EXPECT_CALL(handler3, detachPipeline(_));
 206   }
 207 }
 208
 209 TEST(PipelineTest, DynamicAttachDetachOrder) {
 210   IntHandler handler1, handler2;
 211   Pipeline<int, int> pipeline;
 212   {
 213     InSequence sequence;
 214     EXPECT_CALL(handler2, attachPipeline(_));
 215     EXPECT_CALL(handler1, attachPipeline(_));
 216   }
 217   pipeline
 218     .addBack(&handler1)
 219     .addBack(&handler2)
 220     .finalize();
 221   {
 222     InSequence sequence;
 223     EXPECT_CALL(handler1, detachPipeline(_));
 224     EXPECT_CALL(handler2, detachPipeline(_));
 225   }
 226 }
 227
 228 TEST(PipelineTest, GetContext) {
 229   IntHandler handler;
 230   EXPECT_CALL(handler, attachPipeline(_));
 231   StaticPipeline<int, int, IntHandler> pipeline(&handler);
 232   EXPECT_TRUE(handler.getContext());
 233   EXPECT_CALL(handler, detachPipeline(_));
 234 }
 235
 236 TEST(PipelineTest, HandlerInMultiplePipelines) {
 237   IntHandler handler;
 238   EXPECT_CALL(handler, attachPipeline(_)).Times(2);
 239   StaticPipeline<int, int, IntHandler> pipeline1(&handler);
 240   StaticPipeline<int, int, IntHandler> pipeline2(&handler);
 241   EXPECT_FALSE(handler.getContext());
 242   EXPECT_CALL(handler, detachPipeline(_)).Times(2);
 243 }
 244
 245 TEST(PipelineTest, HandlerInPipelineTwice) {
 246   IntHandler handler;
 247   EXPECT_CALL(handler, attachPipeline(_)).Times(2);
 248   StaticPipeline<int, int, IntHandler, IntHandler> pipeline(&handler, &handler);
 249   EXPECT_FALSE(handler.getContext());
 250   EXPECT_CALL(handler, detachPipeline(_)).Times(2);
 251 }
 252
 253 TEST(PipelineTest, NoDetachOnOwner) {
 254   IntHandler handler;
 255   EXPECT_CALL(handler, attachPipeline(_));
 256   StaticPipeline<int, int, IntHandler> pipeline(&handler);
 257   pipeline.setOwner(&handler);
 258 }
 259
 260 template <class Rin, class Rout = Rin, class Win = Rout, class Wout = Rin>
 261 class ConcreteHandler : public Handler<Rin, Rout, Win, Wout> {
 262   typedef typename Handler<Rin, Rout, Win, Wout>::Context Context;
 263  public:
 264   void read(Context* ctx, Rin msg) {}
 265   Future<void> write(Context* ctx, Win msg) { return makeFuture(); }
 266 };
 267
 268 typedef HandlerAdapter<std::string, std::string> StringHandler;
 269 typedef ConcreteHandler<int, std::string> IntToStringHandler;
 270 typedef ConcreteHandler<std::string, int> StringToIntHandler;
 271
 272 TEST(Pipeline, MissingInboundOrOutbound) {
 273   Pipeline<int, int> pipeline;
 274   pipeline
 275     .addBack(HandlerAdapter<std::string, std::string>{})
 276     .finalize();
 277   EXPECT_THROW(pipeline.read(0), std::invalid_argument);
 278   EXPECT_THROW(pipeline.readEOF(), std::invalid_argument);
 279   EXPECT_THROW(
 280       pipeline.readException(exception_wrapper(std::runtime_error("blah"))),
 281       std::invalid_argument);
 282   EXPECT_THROW(pipeline.write(0), std::invalid_argument);
 283   EXPECT_THROW(pipeline.close(), std::invalid_argument);
 284 }
 285
 286 TEST(Pipeline, DynamicConstruction) {
 287   {
 288     StaticPipeline<std::string, std::string, StringHandler, StringHandler>
 289     pipeline{StringHandler(), StringHandler()};
 290
 291     // Exercise both addFront and addBack. Final pipeline is
 292     // StI <-> ItS <-> StS <-> StS <-> StI <-> ItS
 293     EXPECT_NO_THROW(
 294       pipeline
 295         .addFront(IntToStringHandler{})
 296         .addFront(StringToIntHandler{})
 297         .addBack(StringToIntHandler{})
 298         .addBack(IntToStringHandler{})
 299         .finalize());
 300   }
 301 }
 302
 303 TEST(Pipeline, AttachTransport) {
 304   IntHandler handler;
 305   EXPECT_CALL(handler, attachPipeline(_));
 306   StaticPipeline<int, int, IntHandler>
 307   pipeline(&handler);
 308
 309   EventBase eb;
 310   auto socket = AsyncSocket::newSocket(&eb);
 311
 312   EXPECT_CALL(handler, attachTransport(_));
 313   pipeline.attachTransport(socket);
 314
 315   EXPECT_CALL(handler, detachTransport(_));
 316   pipeline.detachTransport();
 317
 318   EXPECT_CALL(handler, detachPipeline(_));
 319 }