folly/wangle/codec/CodecTest.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 #include <gtest/gtest.h>
  17
  18 #include <folly/wangle/codec/FixedLengthFrameDecoder.h>
  19 #include <folly/wangle/codec/LengthFieldBasedFrameDecoder.h>
  20 #include <folly/wangle/codec/LengthFieldPrepender.h>
  21 #include <folly/wangle/codec/LineBasedFrameDecoder.h>
  22
  23 using namespace folly;
  24 using namespace folly::wangle;
  25 using namespace folly::io;
  26
  27 class FrameTester
  28     : public InboundHandler<std::unique_ptr<IOBuf>> {
  29  public:
  30   explicit FrameTester(std::function<void(std::unique_ptr<IOBuf>)> test)
  31     : test_(test) {}
  32
  33   void read(Context* ctx, std::unique_ptr<IOBuf> buf) override {
  34     test_(std::move(buf));
  35   }
  36
  37   void readException(Context* ctx, exception_wrapper w) override {
  38     test_(nullptr);
  39   }
  40  private:
  41   std::function<void(std::unique_ptr<IOBuf>)> test_;
  42 };
  43
  44 class BytesReflector
  45     : public BytesToBytesHandler {
  46  public:
  47   Future<Unit> write(Context* ctx, std::unique_ptr<IOBuf> buf) override {
  48     IOBufQueue q_(IOBufQueue::cacheChainLength());
  49     q_.append(std::move(buf));
  50     ctx->fireRead(q_);
  51
  52     return makeFuture();
  53   }
  54 };
  55
  56 TEST(FixedLengthFrameDecoder, FailWhenLengthFieldEndOffset) {
  57   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
  58   int called = 0;
  59
  60   pipeline
  61     .addBack(FixedLengthFrameDecoder(10))
  62     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
  63         auto sz = buf->computeChainDataLength();
  64         called++;
  65         EXPECT_EQ(sz, 10);
  66       }))
  67     .finalize();
  68
  69   auto buf3 = IOBuf::create(3);
  70   buf3->append(3);
  71   auto buf11 = IOBuf::create(11);
  72   buf11->append(11);
  73   auto buf16 = IOBuf::create(16);
  74   buf16->append(16);
  75
  76   IOBufQueue q(IOBufQueue::cacheChainLength());
  77
  78   q.append(std::move(buf3));
  79   pipeline.read(q);
  80   EXPECT_EQ(called, 0);
  81
  82   q.append(std::move(buf11));
  83   pipeline.read(q);
  84   EXPECT_EQ(called, 1);
  85
  86   q.append(std::move(buf16));
  87   pipeline.read(q);
  88   EXPECT_EQ(called, 3);
  89 }
  90
  91 TEST(LengthFieldFramePipeline, SimpleTest) {
  92   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
  93   int called = 0;
  94
  95   pipeline
  96     .addBack(BytesReflector())
  97     .addBack(LengthFieldPrepender())
  98     .addBack(LengthFieldBasedFrameDecoder())
  99     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 100         auto sz = buf->computeChainDataLength();
 101         called++;
 102         EXPECT_EQ(sz, 2);
 103       }))
 104     .finalize();
 105
 106   auto buf = IOBuf::create(2);
 107   buf->append(2);
 108   pipeline.write(std::move(buf));
 109   EXPECT_EQ(called, 1);
 110 }
 111
 112 TEST(LengthFieldFramePipeline, LittleEndian) {
 113   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
 114   int called = 0;
 115
 116   pipeline
 117     .addBack(BytesReflector())
 118     .addBack(LengthFieldBasedFrameDecoder(4, 100, 0, 0, 4, false))
 119     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 120         auto sz = buf->computeChainDataLength();
 121         called++;
 122         EXPECT_EQ(sz, 1);
 123       }))
 124     .addBack(LengthFieldPrepender(4, 0, false, false))
 125     .finalize();
 126
 127   auto buf = IOBuf::create(1);
 128   buf->append(1);
 129   pipeline.write(std::move(buf));
 130   EXPECT_EQ(called, 1);
 131 }
 132
 133 TEST(LengthFieldFrameDecoder, Simple) {
 134   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
 135   int called = 0;
 136
 137   pipeline
 138     .addBack(LengthFieldBasedFrameDecoder())
 139     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 140         auto sz = buf->computeChainDataLength();
 141         called++;
 142         EXPECT_EQ(sz, 1);
 143       }))
 144     .finalize();
 145
 146   auto bufFrame = IOBuf::create(4);
 147   bufFrame->append(4);
 148   RWPrivateCursor c(bufFrame.get());
 149   c.writeBE((uint32_t)1);
 150   auto bufData = IOBuf::create(1);
 151   bufData->append(1);
 152
 153   IOBufQueue q(IOBufQueue::cacheChainLength());
 154
 155   q.append(std::move(bufFrame));
 156   pipeline.read(q);
 157   EXPECT_EQ(called, 0);
 158
 159   q.append(std::move(bufData));
 160   pipeline.read(q);
 161   EXPECT_EQ(called, 1);
 162 }
 163
 164 TEST(LengthFieldFrameDecoder, NoStrip) {
 165   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
 166   int called = 0;
 167
 168   pipeline
 169     .addBack(LengthFieldBasedFrameDecoder(2, 10, 0, 0, 0))
 170     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 171         auto sz = buf->computeChainDataLength();
 172         called++;
 173         EXPECT_EQ(sz, 3);
 174       }))
 175     .finalize();
 176
 177   auto bufFrame = IOBuf::create(2);
 178   bufFrame->append(2);
 179   RWPrivateCursor c(bufFrame.get());
 180   c.writeBE((uint16_t)1);
 181   auto bufData = IOBuf::create(1);
 182   bufData->append(1);
 183
 184   IOBufQueue q(IOBufQueue::cacheChainLength());
 185
 186   q.append(std::move(bufFrame));
 187   pipeline.read(q);
 188   EXPECT_EQ(called, 0);
 189
 190   q.append(std::move(bufData));
 191   pipeline.read(q);
 192   EXPECT_EQ(called, 1);
 193 }
 194
 195 TEST(LengthFieldFrameDecoder, Adjustment) {
 196   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
 197   int called = 0;
 198
 199   pipeline
 200     .addBack(LengthFieldBasedFrameDecoder(2, 10, 0, -2, 0))
 201     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 202         auto sz = buf->computeChainDataLength();
 203         called++;
 204         EXPECT_EQ(sz, 3);
 205       }))
 206     .finalize();
 207
 208   auto bufFrame = IOBuf::create(2);
 209   bufFrame->append(2);
 210   RWPrivateCursor c(bufFrame.get());
 211   c.writeBE((uint16_t)3); // includes frame size
 212   auto bufData = IOBuf::create(1);
 213   bufData->append(1);
 214
 215   IOBufQueue q(IOBufQueue::cacheChainLength());
 216
 217   q.append(std::move(bufFrame));
 218   pipeline.read(q);
 219   EXPECT_EQ(called, 0);
 220
 221   q.append(std::move(bufData));
 222   pipeline.read(q);
 223   EXPECT_EQ(called, 1);
 224 }
 225
 226 TEST(LengthFieldFrameDecoder, PreHeader) {
 227   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
 228   int called = 0;
 229
 230   pipeline
 231     .addBack(LengthFieldBasedFrameDecoder(2, 10, 2, 0, 0))
 232     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 233         auto sz = buf->computeChainDataLength();
 234         called++;
 235         EXPECT_EQ(sz, 5);
 236       }))
 237     .finalize();
 238
 239   auto bufFrame = IOBuf::create(4);
 240   bufFrame->append(4);
 241   RWPrivateCursor c(bufFrame.get());
 242   c.write((uint16_t)100); // header
 243   c.writeBE((uint16_t)1); // frame size
 244   auto bufData = IOBuf::create(1);
 245   bufData->append(1);
 246
 247   IOBufQueue q(IOBufQueue::cacheChainLength());
 248
 249   q.append(std::move(bufFrame));
 250   pipeline.read(q);
 251   EXPECT_EQ(called, 0);
 252
 253   q.append(std::move(bufData));
 254   pipeline.read(q);
 255   EXPECT_EQ(called, 1);
 256 }
 257
 258 TEST(LengthFieldFrameDecoder, PostHeader) {
 259   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
 260   int called = 0;
 261
 262   pipeline
 263     .addBack(LengthFieldBasedFrameDecoder(2, 10, 0, 2, 0))
 264     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 265         auto sz = buf->computeChainDataLength();
 266         called++;
 267         EXPECT_EQ(sz, 5);
 268       }))
 269     .finalize();
 270
 271   auto bufFrame = IOBuf::create(4);
 272   bufFrame->append(4);
 273   RWPrivateCursor c(bufFrame.get());
 274   c.writeBE((uint16_t)1); // frame size
 275   c.write((uint16_t)100); // header
 276   auto bufData = IOBuf::create(1);
 277   bufData->append(1);
 278
 279   IOBufQueue q(IOBufQueue::cacheChainLength());
 280
 281   q.append(std::move(bufFrame));
 282   pipeline.read(q);
 283   EXPECT_EQ(called, 0);
 284
 285   q.append(std::move(bufData));
 286   pipeline.read(q);
 287   EXPECT_EQ(called, 1);
 288 }
 289
 290 TEST(LengthFieldFrameDecoderStrip, PrePostHeader) {
 291   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
 292   int called = 0;
 293
 294   pipeline
 295     .addBack(LengthFieldBasedFrameDecoder(2, 10, 2, 2, 4))
 296     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 297         auto sz = buf->computeChainDataLength();
 298         called++;
 299         EXPECT_EQ(sz, 3);
 300       }))
 301     .finalize();
 302
 303   auto bufFrame = IOBuf::create(6);
 304   bufFrame->append(6);
 305   RWPrivateCursor c(bufFrame.get());
 306   c.write((uint16_t)100); // pre header
 307   c.writeBE((uint16_t)1); // frame size
 308   c.write((uint16_t)100); // post header
 309   auto bufData = IOBuf::create(1);
 310   bufData->append(1);
 311
 312   IOBufQueue q(IOBufQueue::cacheChainLength());
 313
 314   q.append(std::move(bufFrame));
 315   pipeline.read(q);
 316   EXPECT_EQ(called, 0);
 317
 318   q.append(std::move(bufData));
 319   pipeline.read(q);
 320   EXPECT_EQ(called, 1);
 321 }
 322
 323 TEST(LengthFieldFrameDecoder, StripPrePostHeaderFrameInclHeader) {
 324   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
 325   int called = 0;
 326
 327   pipeline
 328     .addBack(LengthFieldBasedFrameDecoder(2, 10, 2, -2, 4))
 329     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 330         auto sz = buf->computeChainDataLength();
 331         called++;
 332         EXPECT_EQ(sz, 3);
 333       }))
 334     .finalize();
 335
 336   auto bufFrame = IOBuf::create(6);
 337   bufFrame->append(6);
 338   RWPrivateCursor c(bufFrame.get());
 339   c.write((uint16_t)100); // pre header
 340   c.writeBE((uint16_t)5); // frame size
 341   c.write((uint16_t)100); // post header
 342   auto bufData = IOBuf::create(1);
 343   bufData->append(1);
 344
 345   IOBufQueue q(IOBufQueue::cacheChainLength());
 346
 347   q.append(std::move(bufFrame));
 348   pipeline.read(q);
 349   EXPECT_EQ(called, 0);
 350
 351   q.append(std::move(bufData));
 352   pipeline.read(q);
 353   EXPECT_EQ(called, 1);
 354 }
 355
 356 TEST(LengthFieldFrameDecoder, FailTestLengthFieldEndOffset) {
 357   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
 358   int called = 0;
 359
 360   pipeline
 361     .addBack(LengthFieldBasedFrameDecoder(4, 10, 4, -2, 4))
 362     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 363         ASSERT_EQ(nullptr, buf);
 364         called++;
 365       }))
 366     .finalize();
 367
 368   auto bufFrame = IOBuf::create(8);
 369   bufFrame->append(8);
 370   RWPrivateCursor c(bufFrame.get());
 371   c.writeBE((uint32_t)0); // frame size
 372   c.write((uint32_t)0); // crap
 373
 374   IOBufQueue q(IOBufQueue::cacheChainLength());
 375
 376   q.append(std::move(bufFrame));
 377   pipeline.read(q);
 378   EXPECT_EQ(called, 1);
 379 }
 380
 381 TEST(LengthFieldFrameDecoder, FailTestLengthFieldFrameSize) {
 382   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
 383   int called = 0;
 384
 385   pipeline
 386     .addBack(LengthFieldBasedFrameDecoder(4, 10, 0, 0, 4))
 387     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 388         ASSERT_EQ(nullptr, buf);
 389         called++;
 390       }))
 391     .finalize();
 392
 393   auto bufFrame = IOBuf::create(16);
 394   bufFrame->append(16);
 395   RWPrivateCursor c(bufFrame.get());
 396   c.writeBE((uint32_t)12); // frame size
 397   c.write((uint32_t)0); // nothing
 398   c.write((uint32_t)0); // nothing
 399   c.write((uint32_t)0); // nothing
 400
 401   IOBufQueue q(IOBufQueue::cacheChainLength());
 402
 403   q.append(std::move(bufFrame));
 404   pipeline.read(q);
 405   EXPECT_EQ(called, 1);
 406 }
 407
 408 TEST(LengthFieldFrameDecoder, FailTestLengthFieldInitialBytes) {
 409   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
 410   int called = 0;
 411
 412   pipeline
 413     .addBack(LengthFieldBasedFrameDecoder(4, 10, 0, 0, 10))
 414     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 415         ASSERT_EQ(nullptr, buf);
 416         called++;
 417       }))
 418     .finalize();
 419
 420   auto bufFrame = IOBuf::create(16);
 421   bufFrame->append(16);
 422   RWPrivateCursor c(bufFrame.get());
 423   c.writeBE((uint32_t)4); // frame size
 424   c.write((uint32_t)0); // nothing
 425   c.write((uint32_t)0); // nothing
 426   c.write((uint32_t)0); // nothing
 427
 428   IOBufQueue q(IOBufQueue::cacheChainLength());
 429
 430   q.append(std::move(bufFrame));
 431   pipeline.read(q);
 432   EXPECT_EQ(called, 1);
 433 }
 434
 435 TEST(LineBasedFrameDecoder, Simple) {
 436   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
 437   int called = 0;
 438
 439   pipeline
 440     .addBack(LineBasedFrameDecoder(10))
 441     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 442         auto sz = buf->computeChainDataLength();
 443         called++;
 444         EXPECT_EQ(sz, 3);
 445       }))
 446     .finalize();
 447
 448   auto buf = IOBuf::create(3);
 449   buf->append(3);
 450
 451   IOBufQueue q(IOBufQueue::cacheChainLength());
 452
 453   q.append(std::move(buf));
 454   pipeline.read(q);
 455   EXPECT_EQ(called, 0);
 456
 457   buf = IOBuf::create(1);
 458   buf->append(1);
 459   RWPrivateCursor c(buf.get());
 460   c.write<char>('\n');
 461   q.append(std::move(buf));
 462   pipeline.read(q);
 463   EXPECT_EQ(called, 1);
 464
 465   buf = IOBuf::create(4);
 466   buf->append(4);
 467   RWPrivateCursor c1(buf.get());
 468   c1.write(' ');
 469   c1.write(' ');
 470   c1.write(' ');
 471
 472   c1.write('\r');
 473   q.append(std::move(buf));
 474   pipeline.read(q);
 475   EXPECT_EQ(called, 1);
 476
 477   buf = IOBuf::create(1);
 478   buf->append(1);
 479   RWPrivateCursor c2(buf.get());
 480   c2.write('\n');
 481   q.append(std::move(buf));
 482   pipeline.read(q);
 483   EXPECT_EQ(called, 2);
 484 }
 485
 486 TEST(LineBasedFrameDecoder, SaveDelimiter) {
 487   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
 488   int called = 0;
 489
 490   pipeline
 491     .addBack(LineBasedFrameDecoder(10, false))
 492     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 493         auto sz = buf->computeChainDataLength();
 494         called++;
 495         EXPECT_EQ(sz, 4);
 496       }))
 497     .finalize();
 498
 499   auto buf = IOBuf::create(3);
 500   buf->append(3);
 501
 502   IOBufQueue q(IOBufQueue::cacheChainLength());
 503
 504   q.append(std::move(buf));
 505   pipeline.read(q);
 506   EXPECT_EQ(called, 0);
 507
 508   buf = IOBuf::create(1);
 509   buf->append(1);
 510   RWPrivateCursor c(buf.get());
 511   c.write<char>('\n');
 512   q.append(std::move(buf));
 513   pipeline.read(q);
 514   EXPECT_EQ(called, 1);
 515
 516   buf = IOBuf::create(3);
 517   buf->append(3);
 518   RWPrivateCursor c1(buf.get());
 519   c1.write(' ');
 520   c1.write(' ');
 521   c1.write('\r');
 522   q.append(std::move(buf));
 523   pipeline.read(q);
 524   EXPECT_EQ(called, 1);
 525
 526   buf = IOBuf::create(1);
 527   buf->append(1);
 528   RWPrivateCursor c2(buf.get());
 529   c2.write('\n');
 530   q.append(std::move(buf));
 531   pipeline.read(q);
 532   EXPECT_EQ(called, 2);
 533 }
 534
 535 TEST(LineBasedFrameDecoder, Fail) {
 536   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
 537   int called = 0;
 538
 539   pipeline
 540     .addBack(LineBasedFrameDecoder(10))
 541     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 542         ASSERT_EQ(nullptr, buf);
 543         called++;
 544       }))
 545     .finalize();
 546
 547   auto buf = IOBuf::create(11);
 548   buf->append(11);
 549
 550   IOBufQueue q(IOBufQueue::cacheChainLength());
 551
 552   q.append(std::move(buf));
 553   pipeline.read(q);
 554   EXPECT_EQ(called, 1);
 555
 556   buf = IOBuf::create(1);
 557   buf->append(1);
 558   q.append(std::move(buf));
 559   pipeline.read(q);
 560   EXPECT_EQ(called, 1);
 561
 562   buf = IOBuf::create(2);
 563   buf->append(2);
 564   RWPrivateCursor c(buf.get());
 565   c.write(' ');
 566   c.write<char>('\n');
 567   q.append(std::move(buf));
 568   pipeline.read(q);
 569   EXPECT_EQ(called, 1);
 570
 571   buf = IOBuf::create(12);
 572   buf->append(12);
 573   RWPrivateCursor c2(buf.get());
 574   for (int i = 0; i < 11; i++) {
 575     c2.write(' ');
 576   }
 577   c2.write<char>('\n');
 578   q.append(std::move(buf));
 579   pipeline.read(q);
 580   EXPECT_EQ(called, 2);
 581 }
 582
 583 TEST(LineBasedFrameDecoder, NewLineOnly) {
 584   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
 585   int called = 0;
 586
 587   pipeline
 588     .addBack(LineBasedFrameDecoder(
 589                10, true, LineBasedFrameDecoder::TerminatorType::NEWLINE))
 590     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 591         auto sz = buf->computeChainDataLength();
 592         called++;
 593         EXPECT_EQ(sz, 1);
 594       }))
 595     .finalize();
 596
 597   auto buf = IOBuf::create(2);
 598   buf->append(2);
 599   RWPrivateCursor c(buf.get());
 600   c.write<char>('\r');
 601   c.write<char>('\n');
 602
 603   IOBufQueue q(IOBufQueue::cacheChainLength());
 604
 605   q.append(std::move(buf));
 606   pipeline.read(q);
 607   EXPECT_EQ(called, 1);
 608 }
 609
 610 TEST(LineBasedFrameDecoder, CarriageNewLineOnly) {
 611   Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
 612   int called = 0;
 613
 614   pipeline
 615     .addBack(LineBasedFrameDecoder(
 616               10, true, LineBasedFrameDecoder::TerminatorType::CARRIAGENEWLINE))
 617     .addBack(FrameTester([&](std::unique_ptr<IOBuf> buf) {
 618         auto sz = buf->computeChainDataLength();
 619         called++;
 620         EXPECT_EQ(sz, 1);
 621       }))
 622     .finalize();
 623
 624   auto buf = IOBuf::create(3);
 625   buf->append(3);
 626   RWPrivateCursor c(buf.get());
 627   c.write<char>('\n');
 628   c.write<char>('\r');
 629   c.write<char>('\n');
 630
 631   IOBufQueue q(IOBufQueue::cacheChainLength());
 632
 633   q.append(std::move(buf));
 634   pipeline.read(q);
 635   EXPECT_EQ(called, 1);
 636 }