tst/asynchirediscommanddispatcher_test.cpp

   1 /*
   2    Copyright (c) 2018-2019 Nokia.
   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 <type_traits>
  18 #include <memory>
  19 #include <cstring>
  20 #include <sys/epoll.h>
  21 #include <sys/timerfd.h>
  22 #include <sys/eventfd.h>
  23 #include <arpa/inet.h>
  24 #include <gtest/gtest.h>
  25 #include <async.h>
  26 #include "private/createlogger.hpp"
  27 #include "private/error.hpp"
  28 #include "private/logger.hpp"
  29 #include "private/redis/asynchirediscommanddispatcher.hpp"
  30 #include "private/redis/reply.hpp"
  31 #include "private/redis/contents.hpp"
  32 #include "private/timer.hpp"
  33 #include "private/tst/contentsbuildermock.hpp"
  34 #include "private/tst/enginemock.hpp"
  35 #include "private/tst/hiredissystemmock.hpp"
  36 #include "private/tst/hiredisepolladaptermock.hpp"
  37 #include "private/tst/redisreplybuilder.hpp"
  38
  39 using namespace shareddatalayer;
  40 using namespace shareddatalayer::redis;
  41 using namespace shareddatalayer::tst;
  42 using namespace testing;
  43
  44 namespace
  45 {
  46     class AsyncHiredisCommandDispatcherBaseTest: public testing::Test
  47     {
  48     public:
  49         std::shared_ptr<ContentsBuilderMock> contentsBuilderMock;
  50         StrictMock<EngineMock> engineMock;
  51         HiredisSystemMock hiredisSystemMock;
  52         std::shared_ptr<HiredisEpollAdapterMock> adapterMock;
  53         redisAsyncContext ac;
  54         int hiredisFd;
  55         std::unique_ptr<AsyncHiredisCommandDispatcher> dispatcher;
  56         void (*connected)(const redisAsyncContext*, int);
  57         void (*disconnected)(const redisAsyncContext*, int);
  58         Timer::Callback savedConnectionRetryTimerCallback;
  59         Timer::Duration expectedConnectionRetryTimerDuration;
  60         Timer::Duration expectedConnectionVerificationRetryTimerDuration;
  61         RedisReplyBuilder redisReplyBuilder;
  62         const AsyncConnection::Namespace defaultNamespace;
  63         std::shared_ptr<Logger> logger;
  64
  65         AsyncHiredisCommandDispatcherBaseTest():
  66             contentsBuilderMock(std::make_shared<ContentsBuilderMock>(AsyncConnection::SEPARATOR)),
  67             adapterMock(std::make_shared<HiredisEpollAdapterMock>(engineMock, hiredisSystemMock)),
  68             ac { },
  69             hiredisFd(3),
  70             connected(nullptr),
  71             disconnected(nullptr),
  72             expectedConnectionRetryTimerDuration(std::chrono::seconds(1)),
  73             expectedConnectionVerificationRetryTimerDuration(std::chrono::seconds(10)),
  74             redisReplyBuilder { },
  75             defaultNamespace("namespace"),
  76             logger(createLogger(SDL_LOG_PREFIX))
  77         {
  78         }
  79
  80         virtual ~AsyncHiredisCommandDispatcherBaseTest()
  81         {
  82         }
  83
  84         MOCK_METHOD0(connectAck, void());
  85
  86         MOCK_METHOD0(disconnectCallback, void());
  87
  88         MOCK_METHOD2(ack, void(const std::error_code&, const Reply&));
  89
  90         void expectationsUntilConnect()
  91         {
  92             expectationsUntilConnect(ac);
  93         }
  94
  95         void expectationsUntilConnect(redisAsyncContext& ac)
  96         {
  97             expectRedisAsyncConnect(ac);
  98         }
  99
 100         void expectRedisAsyncConnect()
 101         {
 102             expectRedisAsyncConnect(ac);
 103         }
 104
 105         void expectRedisAsyncConnect(redisAsyncContext& ac)
 106         {
 107             EXPECT_CALL(hiredisSystemMock, redisAsyncConnect(StrEq("host"), 6379U))
 108                 .Times(1)
 109                 .WillOnce(InvokeWithoutArgs([this, &ac]()
 110                                             {
 111                                                 ac.c.fd = hiredisFd;
 112                                                 return &ac;
 113                                             }));
 114         }
 115
 116         void expectRedisAsyncConnectReturnNullptr()
 117         {
 118             EXPECT_CALL(hiredisSystemMock, redisAsyncConnect(StrEq("host"), 6379U))
 119                 .Times(1)
 120                 .WillOnce(InvokeWithoutArgs([this]()
 121                                             {
 122                                                 ac.c.fd = hiredisFd;
 123                                                 return nullptr;
 124                                             }));
 125         }
 126
 127         void expectRedisAsyncSetConnectCallback()
 128         {
 129             expectRedisAsyncSetConnectCallback(ac);
 130         }
 131
 132         void expectRedisAsyncSetConnectCallback(redisAsyncContext& ac)
 133         {
 134             EXPECT_CALL(hiredisSystemMock, redisAsyncSetConnectCallback(&ac, _))
 135                 .Times(1)
 136                 .WillOnce(Invoke([this](const redisAsyncContext*, redisConnectCallback* cb)
 137                                  {
 138                                      connected = cb;
 139                                      return REDIS_OK;
 140                                  }));
 141         }
 142
 143         void expectRedisAsyncSetDisconnectCallback()
 144         {
 145             expectRedisAsyncSetDisconnectCallback(ac);
 146         }
 147
 148         void expectRedisAsyncSetDisconnectCallback(redisAsyncContext& ac)
 149         {
 150             EXPECT_CALL(hiredisSystemMock, redisAsyncSetDisconnectCallback(&ac, _))
 151                 .Times(1)
 152                 .WillOnce(Invoke([this](const redisAsyncContext*, redisDisconnectCallback* cb)
 153                                  {
 154                                      disconnected = cb;
 155                                      return REDIS_OK;
 156                                  }));
 157         }
 158
 159         void expectAdapterAttach()
 160         {
 161             expectAdapterAttach(ac);
 162         }
 163
 164         void expectAdapterAttach(redisAsyncContext& ac)
 165         {
 166             EXPECT_CALL(*adapterMock, attach(&ac))
 167                 .Times(1);
 168         }
 169
 170         void expectConnectAck()
 171         {
 172             EXPECT_CALL(*this, connectAck())
 173                 .Times(1);
 174         }
 175
 176         void expectDisconnectCallback()
 177         {
 178             EXPECT_CALL(*this, disconnectCallback())
 179                 .Times(1);
 180         }
 181
 182         void expectRedisAsyncFree()
 183         {
 184             EXPECT_CALL(hiredisSystemMock, redisAsyncFree(&ac))
 185                 .Times(1);
 186         }
 187
 188         void expectRedisAsyncDisconnect()
 189         {
 190             EXPECT_CALL(hiredisSystemMock, redisAsyncDisconnect(&ac))
 191                 .Times(1);
 192         }
 193
 194         void expectRedisAsyncCommandArgv(redisReply& rr)
 195         {
 196             EXPECT_CALL(hiredisSystemMock, redisAsyncCommandArgv(&ac, _, _, _, _, _))
 197                 .Times(1)
 198                 .WillOnce(Invoke([&rr](redisAsyncContext* ac, redisCallbackFn* cb, void* pd,
 199                                        int, const char**, const size_t*)
 200                                  {
 201                                      cb(ac, &rr, pd);
 202                                      return REDIS_OK;
 203                                  }));
 204         }
 205
 206         void expectCommandListQuery()
 207         {
 208             expectRedisAsyncCommandArgv(redisReplyBuilder.buildCommandListQueryReply());
 209         }
 210
 211         void expectCommandListQueryReturnError()
 212         {
 213             expectRedisAsyncCommandArgv(redisReplyBuilder.buildErrorReply("SomeErrorForCommandListQuery"));
 214         }
 215
 216         void verifyAckErrorReply(const Reply& reply)
 217         {
 218             EXPECT_EQ(Reply::Type::NIL, reply.getType());
 219             EXPECT_EQ(0, reply.getInteger());
 220             EXPECT_TRUE(reply.getString()->str.empty());
 221             EXPECT_EQ(static_cast<ReplyStringLength>(0), reply.getString()->len);
 222             EXPECT_TRUE(reply.getArray()->empty());
 223         }
 224
 225         void expectAckError()
 226         {
 227             EXPECT_CALL(*this, ack(Ne(std::error_code()), _))
 228                 .Times(1)
 229                 .WillOnce(Invoke([this](const std::error_code&, const Reply& reply)
 230                                  {
 231                                      verifyAckErrorReply(reply);
 232                                  }));
 233         }
 234
 235         void expectAckError(const std::error_code& ec)
 236         {
 237             EXPECT_CALL(*this, ack(ec, _))
 238                 .Times(1)
 239                 .WillOnce(Invoke([this](const std::error_code&, const Reply& reply)
 240                                  {
 241                                      verifyAckErrorReply(reply);
 242                                  }));
 243         }
 244
 245         void expectArmConnectionRetryTimer()
 246         {
 247             EXPECT_CALL(engineMock, armTimer(_, expectedConnectionRetryTimerDuration, _))
 248                 .Times(1)
 249                 .WillOnce(SaveArg<2>(&savedConnectionRetryTimerCallback));
 250         }
 251
 252         void expectArmConnectionVerificationRetryTimer()
 253         {
 254             EXPECT_CALL(engineMock, armTimer(_, expectedConnectionVerificationRetryTimerDuration, _))
 255                 .Times(1)
 256                 .WillOnce(SaveArg<2>(&savedConnectionRetryTimerCallback));
 257         }
 258
 259         void expectDisarmConnectionRetryTimer()
 260         {
 261             EXPECT_CALL(engineMock, disarmTimer(_))
 262                 .Times(1);
 263         }
 264     };
 265
 266     class AsyncHiredisCommandDispatcherDisconnectedTest: public AsyncHiredisCommandDispatcherBaseTest
 267     {
 268     public:
 269         AsyncHiredisCommandDispatcherDisconnectedTest()
 270         {
 271                 InSequence dummy;
 272                 expectationsUntilConnect();
 273                 expectAdapterAttach();
 274                 expectRedisAsyncSetConnectCallback();
 275                 expectRedisAsyncSetDisconnectCallback();
 276                 dispatcher.reset(new AsyncHiredisCommandDispatcher(engineMock,
 277                                                                    "host",
 278                                                                    htons(6379U),
 279                                                                    contentsBuilderMock,
 280                                                                    false,
 281                                                                    hiredisSystemMock,
 282                                                                    adapterMock,
 283                                                                    logger,
 284                                                                    false));
 285         }
 286
 287         ~AsyncHiredisCommandDispatcherDisconnectedTest()
 288         {
 289         }
 290     };
 291
 292     class AsyncHiredisCommandDispatcherWithPermanentCommandCallbacksTest: public AsyncHiredisCommandDispatcherBaseTest
 293     {
 294     public:
 295         AsyncHiredisCommandDispatcherWithPermanentCommandCallbacksTest()
 296         {
 297                 InSequence dummy;
 298                 expectationsUntilConnect();
 299                 expectAdapterAttach();
 300                 expectRedisAsyncSetConnectCallback();
 301                 expectRedisAsyncSetDisconnectCallback();
 302                 dispatcher.reset(new AsyncHiredisCommandDispatcher(engineMock,
 303                                                                    "host",
 304                                                                    htons(6379U),
 305                                                                    contentsBuilderMock,
 306                                                                    true,
 307                                                                    hiredisSystemMock,
 308                                                                    adapterMock,
 309                                                                    logger,
 310                                                                    false));
 311         }
 312
 313         ~AsyncHiredisCommandDispatcherWithPermanentCommandCallbacksTest()
 314         {
 315             expectRedisAsyncFree();
 316         }
 317     };
 318
 319     class AsyncHiredisCommandDispatcherConnectedTest: public AsyncHiredisCommandDispatcherDisconnectedTest
 320     {
 321     public:
 322         Contents contents;
 323         redisCallbackFn* savedCb;
 324         void* savedPd;
 325
 326         AsyncHiredisCommandDispatcherConnectedTest():
 327             contents { { "CMD", "key1", "value1", "key2", "value2" },
 328                        { 3, 4, 6, 4, 6 } },
 329             savedCb(nullptr),
 330             savedPd(nullptr)
 331         {
 332             expectCommandListQuery();
 333             connected(&ac, 0);
 334         }
 335
 336         ~AsyncHiredisCommandDispatcherConnectedTest()
 337         {
 338             expectRedisAsyncFree();
 339         }
 340
 341         void expectAck()
 342         {
 343             EXPECT_CALL(*this, ack(std::error_code(), _))
 344                 .Times(1);
 345         }
 346
 347         void expectReplyError(const std::string& msg)
 348         {
 349             EXPECT_CALL(hiredisSystemMock, redisAsyncCommandArgv(&ac, _, _, _, _, _))
 350                 .Times(1)
 351                 .WillOnce(Invoke([this, msg](redisAsyncContext* ac, redisCallbackFn* cb, void* pd,
 352                                     int, const char**, const size_t*)
 353                                  {
 354                                      cb(ac, &redisReplyBuilder.buildErrorReply(msg), pd);
 355                                      return REDIS_OK;
 356                                  }));
 357         }
 358
 359         void expectContextError(int code)
 360         {
 361             EXPECT_CALL(hiredisSystemMock, redisAsyncCommandArgv(&ac, _, _, _, _, _))
 362                 .Times(1)
 363                 .WillOnce(Invoke([code](redisAsyncContext* ac, redisCallbackFn* cb, void* pd, int, const char**, const size_t*)
 364                                  {
 365                                      ac->err = code;
 366                                      cb(ac, nullptr, pd);
 367                                      return REDIS_OK;
 368                                  }));
 369         }
 370
 371         void expectRedisAsyncFreeCallPendingCallback(redisCallbackFn* cb, void* pd)
 372         {
 373             EXPECT_CALL(hiredisSystemMock, redisAsyncFree(&ac))
 374                 .Times(1)
 375                 .WillOnce(Invoke([cb, pd](redisAsyncContext* ac)
 376                                  {
 377                                      cb(ac, nullptr, pd);
 378                                  }));
 379         }
 380
 381         void expectAckNotCalled()
 382         {
 383             EXPECT_CALL(*this, ack(_,_))
 384                 .Times(0);
 385         }
 386
 387         void expectRedisAsyncCommandArgv_SaveCb()
 388         {
 389             EXPECT_CALL(hiredisSystemMock, redisAsyncCommandArgv(&ac, _, _, _, _, _))
 390                 .Times(1)
 391                 .WillRepeatedly(Invoke([this](redisAsyncContext*, redisCallbackFn* cb, void* pd,
 392                                               int, const char**, const size_t*)
 393                                               {
 394                                                   savedCb = cb;
 395                                                   savedPd = pd;
 396                                                   return REDIS_OK;
 397                                               }));
 398         }
 399     };
 400
 401     using AsyncHiredisCommandDispatcherDeathTest = AsyncHiredisCommandDispatcherConnectedTest;
 402
 403     class AsyncHiredisCommandDispatcherForSentinelTest: public AsyncHiredisCommandDispatcherBaseTest
 404     {
 405     public:
 406         AsyncHiredisCommandDispatcherForSentinelTest()
 407         {
 408                 InSequence dummy;
 409                 expectationsUntilConnect();
 410                 expectAdapterAttach();
 411                 expectRedisAsyncSetConnectCallback();
 412                 expectRedisAsyncSetDisconnectCallback();
 413                 dispatcher.reset(new AsyncHiredisCommandDispatcher(engineMock,
 414                                                                    "host",
 415                                                                    htons(6379U),
 416                                                                    contentsBuilderMock,
 417                                                                    true,
 418                                                                    hiredisSystemMock,
 419                                                                    adapterMock,
 420                                                                    logger,
 421                                                                    true));
 422         }
 423
 424         ~AsyncHiredisCommandDispatcherForSentinelTest()
 425         {
 426             expectRedisAsyncFree();
 427         }
 428     };
 429 }
 430
 431 TEST_F(AsyncHiredisCommandDispatcherDisconnectedTest, IsNotCopyable)
 432 {
 433     EXPECT_FALSE(std::is_copy_constructible<AsyncHiredisCommandDispatcher>::value);
 434     EXPECT_FALSE(std::is_copy_assignable<AsyncHiredisCommandDispatcher>::value);
 435 }
 436
 437 TEST_F(AsyncHiredisCommandDispatcherDisconnectedTest, ImplementsAsyncRedisCommandDispatcher)
 438 {
 439     EXPECT_TRUE((std::is_base_of<AsyncCommandDispatcher, AsyncHiredisCommandDispatcher>::value));
 440 }
 441
 442 TEST_F(AsyncHiredisCommandDispatcherDisconnectedTest, CannotDispatchCommandsIfDisconnected)
 443 {
 444     Engine::Callback storedCallback;
 445     EXPECT_CALL(engineMock, postCallback(_))
 446         .Times(1)
 447         .WillOnce(SaveArg<0>(&storedCallback));
 448     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherDisconnectedTest::ack,
 449                                         this,
 450                                         std::placeholders::_1,
 451                                         std::placeholders::_2),
 452                               defaultNamespace,
 453                               { });
 454     expectAckError(std::error_code(AsyncRedisCommandDispatcherErrorCode::NOT_CONNECTED));
 455     storedCallback();
 456 }
 457
 458 TEST_F(AsyncHiredisCommandDispatcherDisconnectedTest, ContextErrorInConnectArmsRetryTimer)
 459 {
 460     InSequence dummy;
 461     expectationsUntilConnect();
 462     expectArmConnectionRetryTimer();
 463     ac.err = 123;
 464     dispatcher.reset(new AsyncHiredisCommandDispatcher(engineMock,
 465                                                        "host",
 466                                                        htons(6379U),
 467                                                        contentsBuilderMock,
 468                                                        false,
 469                                                        hiredisSystemMock,
 470                                                        adapterMock,
 471                                                        logger,
 472                                                        false));
 473     expectDisarmConnectionRetryTimer();
 474 }
 475
 476 TEST_F(AsyncHiredisCommandDispatcherBaseTest, NullRedisContextInConnectArmsRetryTimer)
 477 {
 478     InSequence dummy;
 479     expectRedisAsyncConnectReturnNullptr();
 480     expectArmConnectionRetryTimer();
 481     expectDisarmConnectionRetryTimer();
 482
 483     dispatcher.reset(new AsyncHiredisCommandDispatcher(engineMock,
 484                                                        "host",
 485                                                        htons(6379U),
 486                                                        contentsBuilderMock,
 487                                                        false,
 488                                                        hiredisSystemMock,
 489                                                        adapterMock,
 490                                                        logger,
 491                                                        false));
 492 }
 493
 494 TEST_F(AsyncHiredisCommandDispatcherDisconnectedTest, FailedCommandListQueryArmsRetryTimer)
 495 {
 496     InSequence dummy;
 497     expectCommandListQueryReturnError();
 498     expectArmConnectionVerificationRetryTimer();
 499     expectRedisAsyncFree();
 500     connected(&ac, 0);
 501     expectDisarmConnectionRetryTimer();
 502 }
 503
 504 TEST_F(AsyncHiredisCommandDispatcherDisconnectedTest, ErrorInConnectedCallbackArmsRetryTimer)
 505 {
 506     InSequence dummy;
 507     expectArmConnectionRetryTimer();
 508     connected(&ac, -1);
 509     expectDisarmConnectionRetryTimer();
 510 }
 511
 512 TEST_F(AsyncHiredisCommandDispatcherDisconnectedTest, ConnectionSucceedsWithRetryTimer)
 513 {
 514     InSequence dummy;
 515     expectArmConnectionRetryTimer();
 516
 517     connected(&ac, -1);
 518
 519     expectationsUntilConnect();
 520     expectAdapterAttach();
 521     expectRedisAsyncSetConnectCallback();
 522     expectRedisAsyncSetDisconnectCallback();
 523
 524     savedConnectionRetryTimerCallback();
 525
 526     expectCommandListQuery();
 527     expectConnectAck();
 528
 529     dispatcher->waitConnectedAsync(std::bind(&AsyncHiredisCommandDispatcherDisconnectedTest::connectAck, this));
 530     connected(&ac, 0);
 531
 532     expectRedisAsyncFree();
 533 }
 534
 535 TEST_F(AsyncHiredisCommandDispatcherDisconnectedTest, ConnectAckCalledOnceConnected)
 536 {
 537     InSequence dummy;
 538     expectCommandListQuery();
 539     expectConnectAck();
 540     dispatcher->waitConnectedAsync(std::bind(&AsyncHiredisCommandDispatcherDisconnectedTest::connectAck, this));
 541     connected(&ac, 0);
 542     expectRedisAsyncFree();
 543 }
 544
 545 TEST_F(AsyncHiredisCommandDispatcherDisconnectedTest, ConnectAckCalledIfConnected)
 546 {
 547     Engine::Callback storedCallback;
 548     EXPECT_CALL(engineMock, postCallback(_))
 549         .Times(1)
 550         .WillOnce(SaveArg<0>(&storedCallback));
 551     expectCommandListQuery();
 552     connected(&ac, 0);
 553     dispatcher->waitConnectedAsync(std::bind(&AsyncHiredisCommandDispatcherDisconnectedTest::connectAck, this));
 554     expectConnectAck();
 555     storedCallback();
 556     expectRedisAsyncFree();
 557 }
 558
 559 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, CanDispatchCommands)
 560 {
 561     EXPECT_CALL(hiredisSystemMock, redisAsyncCommandArgv(&ac, _, _, _, _, _))
 562         .Times(1)
 563         .WillOnce(Invoke([this](redisAsyncContext* ac, redisCallbackFn* cb, void* pd,
 564                                 int argc, const char** argv, const size_t* argvlen)
 565                          {
 566                              EXPECT_EQ((int)contents.stack.size(), argc);
 567                              EXPECT_EQ(contents.sizes[0], argvlen[0]);
 568                              EXPECT_EQ(contents.sizes[1], argvlen[1]);
 569                              EXPECT_EQ(contents.sizes[2], argvlen[2]);
 570                              EXPECT_EQ(contents.sizes[3], argvlen[3]);
 571                              EXPECT_EQ(contents.sizes[4], argvlen[4]);
 572                              EXPECT_FALSE(std::memcmp(argv[0], contents.stack[0].c_str(), contents.sizes[0]));
 573                              EXPECT_FALSE(std::memcmp(argv[1], contents.stack[1].c_str(), contents.sizes[1]));
 574                              EXPECT_FALSE(std::memcmp(argv[2], contents.stack[2].c_str(), contents.sizes[2]));
 575                              EXPECT_FALSE(std::memcmp(argv[3], contents.stack[3].c_str(), contents.sizes[3]));
 576                              EXPECT_FALSE(std::memcmp(argv[4], contents.stack[4].c_str(), contents.sizes[4]));
 577                              cb(ac, &redisReplyBuilder.buildNilReply(), pd);
 578                              return REDIS_OK;
 579                          }));
 580     expectAck();
 581     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 582                                         this,
 583                                         std::placeholders::_1,
 584                                         std::placeholders::_2),
 585                               defaultNamespace,
 586                               contents);
 587 }
 588
 589 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, CanParseNilReply)
 590 {
 591     expectRedisAsyncCommandArgv(redisReplyBuilder.buildNilReply());
 592     EXPECT_CALL(*this, ack(std::error_code(), _))
 593         .Times(1)
 594         .WillOnce(Invoke([](const std::error_code&, const Reply& reply)
 595                          {
 596                              EXPECT_EQ(Reply::Type::NIL, reply.getType());
 597                              EXPECT_EQ(0, reply.getInteger());
 598                              EXPECT_TRUE(reply.getString()->str.empty());
 599                              EXPECT_EQ(static_cast<ReplyStringLength>(0), reply.getString()->len);
 600                              EXPECT_TRUE(reply.getArray()->empty());
 601                          }));
 602     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 603                                         this,
 604                                         std::placeholders::_1,
 605                                         std::placeholders::_2),
 606                                         defaultNamespace,
 607                                         contents);
 608 }
 609
 610 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, CanParseIntegerReply)
 611 {
 612     expectRedisAsyncCommandArgv(redisReplyBuilder.buildIntegerReply());
 613     EXPECT_CALL(*this, ack(std::error_code(), _))
 614         .Times(1)
 615         .WillOnce(Invoke([this](const std::error_code&, const Reply& reply)
 616                          {
 617                              auto expected(redisReplyBuilder.buildIntegerReply());
 618                              EXPECT_EQ(Reply::Type::INTEGER, reply.getType());
 619                              EXPECT_EQ(expected.integer, reply.getInteger());
 620                          }));
 621     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 622                                         this,
 623                                         std::placeholders::_1,
 624                                         std::placeholders::_2),
 625                                         defaultNamespace,
 626                                         contents);
 627 }
 628
 629 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, CanParseStatusReply)
 630 {
 631     expectRedisAsyncCommandArgv(redisReplyBuilder.buildStatusReply());
 632     EXPECT_CALL(*this, ack(std::error_code(), _))
 633         .Times(1)
 634         .WillOnce(Invoke([this](const std::error_code&, const Reply& reply)
 635                          {
 636                              auto expected(redisReplyBuilder.buildStatusReply());
 637                              EXPECT_EQ(Reply::Type::STATUS, reply.getType());
 638                              EXPECT_EQ(expected.len, reply.getString()->len);
 639                              EXPECT_FALSE(std::memcmp(reply.getString()->str.c_str(), expected.str, expected.len));
 640                          }));
 641     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 642                                         this,
 643                                         std::placeholders::_1,
 644                                         std::placeholders::_2),
 645                                         defaultNamespace,
 646                                         contents);
 647 }
 648
 649 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, CanParseStringReply)
 650 {
 651     expectRedisAsyncCommandArgv(redisReplyBuilder.buildStringReply());
 652     EXPECT_CALL(*this, ack(std::error_code(), _))
 653         .Times(1)
 654         .WillOnce(Invoke([this](const std::error_code&, const Reply& reply)
 655                          {
 656                              auto expected(redisReplyBuilder.buildStringReply());
 657                              EXPECT_EQ(Reply::Type::STRING, reply.getType());
 658                              EXPECT_EQ(expected.len, reply.getString()->len);
 659                              EXPECT_FALSE(std::memcmp(reply.getString()->str.c_str(), expected.str, expected.len));
 660                          }));
 661     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 662                                         this,
 663                                         std::placeholders::_1,
 664                                         std::placeholders::_2),
 665                                         defaultNamespace,
 666                                         contents);
 667 }
 668
 669 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, CanParseArrayReply)
 670 {
 671     expectRedisAsyncCommandArgv(redisReplyBuilder.buildArrayReply());
 672     EXPECT_CALL(*this, ack(std::error_code(), _))
 673         .Times(1)
 674         .WillOnce(Invoke([](const std::error_code&, const Reply& reply)
 675                          {
 676                              auto array(reply.getArray());
 677                              EXPECT_EQ(Reply::Type::ARRAY, reply.getType());
 678                              EXPECT_EQ(Reply::Type::STRING, (*array)[0]->getType());
 679                              EXPECT_EQ(Reply::Type::NIL, (*array)[1]->getType());
 680                          }));
 681     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 682                                         this,
 683                                         std::placeholders::_1,
 684                                         std::placeholders::_2),
 685                                         defaultNamespace,
 686                                         contents);
 687 }
 688
 689 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, CanHandleDispatchHiredisBufferErrors)
 690 {
 691     EXPECT_CALL(hiredisSystemMock, redisAsyncCommandArgv(&ac, _, _, _, _, _))
 692         .Times(1)
 693         .WillOnce(Invoke([](redisAsyncContext* ac, redisCallbackFn*, void*, int, const char**, const size_t*)
 694                          {
 695                              ac->err = REDIS_ERR;
 696                              return REDIS_ERR;
 697                          }));
 698     Engine::Callback storedCallback;
 699     EXPECT_CALL(engineMock, postCallback(_))
 700         .Times(1)
 701         .WillOnce(SaveArg<0>(&storedCallback));
 702     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 703                                         this,
 704                                         std::placeholders::_1,
 705                                         std::placeholders::_2),
 706                                         defaultNamespace,
 707                                         contents);
 708     expectAckError();
 709     storedCallback();
 710 }
 711
 712 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, CanHandleDispatchHiredisCbErrors)
 713 {
 714     EXPECT_CALL(hiredisSystemMock, redisAsyncCommandArgv(&ac, _, _, _, _, _))
 715         .Times(1)
 716         .WillOnce(Invoke([](redisAsyncContext* ac, redisCallbackFn* cb, void* pd, int, const char**, const size_t*)
 717                          {
 718                              cb(ac, nullptr, pd);
 719                              return REDIS_OK;
 720                          }));
 721     expectAckError();
 722     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 723                                         this,
 724                                         std::placeholders::_1,
 725                                         std::placeholders::_2),
 726                                         defaultNamespace,
 727                                         contents);
 728 }
 729
 730 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, DatasetStillBeingLoadedInMemoryIsRecognizedFromReply)
 731 {
 732     expectReplyError("LOADING Redis is loading the dataset in memory");
 733     EXPECT_CALL(*this, ack(std::error_code(AsyncRedisCommandDispatcherErrorCode::DATASET_LOADING), _))
 734         .Times(1);
 735     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 736                                         this,
 737                                         std::placeholders::_1,
 738                                         std::placeholders::_2),
 739                                         defaultNamespace,
 740                                         contents);
 741 }
 742
 743 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, ProtocolErrorIsRecognizedFromReply)
 744 {
 745     expectReplyError("ERR Protocol error: invalid bulk length");
 746     EXPECT_CALL(*this, ack(std::error_code(AsyncRedisCommandDispatcherErrorCode::PROTOCOL_ERROR), _))
 747         .Times(1);
 748     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 749                                         this,
 750                                         std::placeholders::_1,
 751                                         std::placeholders::_2),
 752                                         defaultNamespace,
 753                                         contents);
 754 }
 755
 756 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, UnrecognizedReplyErrorIsConvertedToUnknownError)
 757 {
 758     expectReplyError("something sinister");
 759     EXPECT_CALL(*this, ack(std::error_code(AsyncRedisCommandDispatcherErrorCode::UNKNOWN_ERROR), _))
 760         .Times(1);
 761     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 762                                         this,
 763                                         std::placeholders::_1,
 764                                         std::placeholders::_2),
 765                                         defaultNamespace,
 766                                         contents);
 767 }
 768
 769 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, IOErrorInContext)
 770 {
 771     EXPECT_CALL(hiredisSystemMock, redisAsyncCommandArgv(&ac, _, _, _, _, _))
 772         .Times(1)
 773         .WillOnce(Invoke([](redisAsyncContext* ac, redisCallbackFn* cb, void* pd, int, const char**, const size_t*)
 774                          {
 775                              ac->err = REDIS_ERR_IO;
 776                              errno = EINVAL;
 777                              cb(ac, nullptr, pd);
 778                              return REDIS_OK;
 779                          }));
 780     EXPECT_CALL(*this, ack(std::error_code(AsyncRedisCommandDispatcherErrorCode::IO_ERROR), _))
 781         .Times(1);
 782     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 783                                         this,
 784                                         std::placeholders::_1,
 785                                         std::placeholders::_2),
 786                                         defaultNamespace,
 787                                         contents);
 788 }
 789
 790 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, IOErrorInContextWithECONNRESETerrnoValue)
 791 {
 792     EXPECT_CALL(hiredisSystemMock, redisAsyncCommandArgv(&ac, _, _, _, _, _))
 793         .Times(1)
 794         .WillOnce(Invoke([](redisAsyncContext* ac, redisCallbackFn* cb, void* pd, int, const char**, const size_t*)
 795                          {
 796                              ac->err = REDIS_ERR_IO;
 797                              errno = ECONNRESET;
 798                              cb(ac, nullptr, pd);
 799                              return REDIS_OK;
 800                          }));
 801     EXPECT_CALL(*this, ack(std::error_code(AsyncRedisCommandDispatcherErrorCode::CONNECTION_LOST), _))
 802         .Times(1);
 803     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 804                                         this,
 805                                         std::placeholders::_1,
 806                                         std::placeholders::_2),
 807                                         defaultNamespace,
 808                                         contents);
 809 }
 810
 811 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, EofErrorInContext)
 812 {
 813     expectContextError(REDIS_ERR_EOF);
 814     EXPECT_CALL(*this, ack(std::error_code(AsyncRedisCommandDispatcherErrorCode::CONNECTION_LOST), _))
 815         .Times(1);
 816     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 817                                         this,
 818                                         std::placeholders::_1,
 819                                         std::placeholders::_2),
 820                                         defaultNamespace,
 821                                         contents);
 822 }
 823
 824 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, ProtocolErrorInContext)
 825 {
 826     expectContextError(REDIS_ERR_PROTOCOL);
 827     EXPECT_CALL(*this, ack(std::error_code(AsyncRedisCommandDispatcherErrorCode::PROTOCOL_ERROR), _))
 828         .Times(1);
 829     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 830                                         this,
 831                                         std::placeholders::_1,
 832                                         std::placeholders::_2),
 833                                         defaultNamespace,
 834                                         contents);
 835 }
 836
 837 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, OomErrorInContext)
 838 {
 839     expectContextError(REDIS_ERR_OOM);
 840     EXPECT_CALL(*this, ack(std::error_code(AsyncRedisCommandDispatcherErrorCode::OUT_OF_MEMORY), _))
 841         .Times(1);
 842     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 843                                         this,
 844                                         std::placeholders::_1,
 845                                         std::placeholders::_2),
 846                                         defaultNamespace,
 847                                         contents);
 848 }
 849
 850 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, UnrecognizedContextErrorIsConvertedToUnknownError)
 851 {
 852     expectContextError(REDIS_ERR_OTHER);
 853     EXPECT_CALL(*this, ack(std::error_code(AsyncRedisCommandDispatcherErrorCode::UNKNOWN_ERROR), _))
 854         .Times(1);
 855     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 856                                         this,
 857                                         std::placeholders::_1,
 858                                         std::placeholders::_2),
 859                                         defaultNamespace,
 860                                         contents);
 861 }
 862
 863 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, PendingClientCallbacksAreNotCalledAfterDisabled)
 864 {
 865     InSequence dummy;
 866     expectRedisAsyncCommandArgv_SaveCb();
 867     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 868                                         this,
 869                                         std::placeholders::_1,
 870                                         std::placeholders::_2),
 871                                         defaultNamespace,
 872                                         contents);
 873     expectAck();
 874     savedCb(&ac, &redisReplyBuilder.buildStringReply(), savedPd);
 875     dispatcher->disableCommandCallbacks();
 876     expectRedisAsyncCommandArgv_SaveCb();
 877     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 878                                         this,
 879                                         std::placeholders::_1,
 880                                         std::placeholders::_2),
 881                                         defaultNamespace,
 882                                         contents);
 883     expectAckNotCalled();
 884     savedCb(&ac, &redisReplyBuilder.buildStringReply(), savedPd);
 885 }
 886
 887 TEST_F(AsyncHiredisCommandDispatcherConnectedTest, RegisteredClientDisconnectCallbackIsCalled)
 888 {
 889     InSequence dummy;
 890     dispatcher->registerDisconnectCb(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::disconnectCallback,
 891                                              this));
 892     expectDisconnectCallback();
 893     expectArmConnectionRetryTimer();
 894     disconnected(&ac, 0);
 895     expectDisarmConnectionRetryTimer();
 896     expectCommandListQuery();
 897     connected(&ac, 0); // restore connection to meet destructor expectations
 898 }
 899
 900 TEST_F(AsyncHiredisCommandDispatcherWithPermanentCommandCallbacksTest, CanHandleMultipleRepliesForSameRedisCommand)
 901 {
 902     InSequence dummy;
 903     redisCallbackFn* savedCb;
 904     void* savedPd;
 905     Contents contents({ { "cmd", "key", "value" }, { 3, 3, 5 } });
 906     expectCommandListQuery();
 907     connected(&ac, 0);
 908     EXPECT_CALL(hiredisSystemMock, redisAsyncCommandArgv(&ac, _, _, _, _, _))
 909         .Times(1)
 910         .WillOnce(Invoke([&savedCb, &savedPd](redisAsyncContext*, redisCallbackFn* cb, void* pd,
 911                                               int, const char**, const size_t*)
 912                          {
 913                              savedCb = cb;
 914                              savedPd = pd;
 915                              return REDIS_OK;
 916                          }));
 917     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 918                                         this,
 919                                         std::placeholders::_1,
 920                                         std::placeholders::_2),
 921                                         defaultNamespace,
 922                                         contents);
 923     EXPECT_CALL(*this, ack(std::error_code(), _))
 924             .Times(3);
 925     redisReply rr;
 926     rr.type = REDIS_REPLY_NIL;
 927     savedCb(&ac, &rr, savedPd);
 928     savedCb(&ac, &rr, savedPd);
 929     savedCb(&ac, &rr, savedPd);
 930 }
 931
 932 TEST_F(AsyncHiredisCommandDispatcherDeathTest, CbRemovedAfterHiredisCb)
 933 {
 934     redisCallbackFn* savedCb;
 935     void* savedPd;
 936     EXPECT_CALL(hiredisSystemMock, redisAsyncCommandArgv(&ac, _, _, _, _, _))
 937         .Times(1)
 938         .WillOnce(Invoke([this, &savedCb, &savedPd](redisAsyncContext* ac, redisCallbackFn* cb, void* pd,
 939                                                     int, const char**, const size_t*)
 940                          {
 941                              savedCb = cb;
 942                              savedPd = pd;
 943                              cb(ac, &redisReplyBuilder.buildNilReply(), pd);
 944                              return REDIS_OK;
 945                          }));
 946     expectAck();
 947     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherDeathTest::ack,
 948                                         this,
 949                                         std::placeholders::_1,
 950                                         std::placeholders::_2),
 951                                         defaultNamespace,
 952                                         contents);
 953     EXPECT_EXIT(savedCb(&ac, &redisReplyBuilder.buildNilReply(), savedPd), KilledBySignal(SIGABRT), "");
 954 }
 955
 956 TEST_F(AsyncHiredisCommandDispatcherDeathTest, TooManyRepliesAborts)
 957 {
 958     InSequence dummy;
 959     redisCallbackFn* savedCb;
 960     void* savedPd;
 961     Contents contents({ { "cmd", "key", "value" }, { 3, 3, 5 } });
 962     EXPECT_CALL(hiredisSystemMock, redisAsyncCommandArgv(&ac, _, _, _, _, _))
 963         .Times(1)
 964         .WillOnce(Invoke([&savedCb, &savedPd](redisAsyncContext*, redisCallbackFn* cb, void* pd,
 965                                               int, const char**, const size_t*)
 966                          {
 967                              savedCb = cb;
 968                              savedPd = pd;
 969                              return REDIS_OK;
 970                          }));
 971     expectAck();
 972     dispatcher->dispatchAsync(std::bind(&AsyncHiredisCommandDispatcherConnectedTest::ack,
 973                                         this,
 974                                         std::placeholders::_1,
 975                                         std::placeholders::_2),
 976                                         defaultNamespace,
 977                                         contents);
 978     savedCb(&ac, &redisReplyBuilder.buildNilReply(), savedPd);
 979     EXPECT_EXIT(savedCb(&ac, &redisReplyBuilder.buildNilReply(), savedPd), KilledBySignal(SIGABRT), "");
 980 }
 981
 982 TEST_F(AsyncHiredisCommandDispatcherForSentinelTest, CommandListInquiryIsNotSent)
 983 {
 984     EXPECT_CALL(hiredisSystemMock, redisAsyncCommandArgv(_, _, _, _, _, _))
 985         .Times(0);
 986     connected(&ac, 0);
 987 }