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