ObservingHandlerTest.cpp File Reference

#include <wangle/channel/broadcast/test/Mocks.h>
#include <wangle/channel/test/MockHandler.h>
#include <wangle/codec/MessageToByteEncoder.h>

Go to the source code of this file.

Classes
class	ObservingHandlerTest
class	ObservingHandlerTest::MockIntToByteEncoder

Functions
	TEST_F (ObservingHandlerTest, Success)
	TEST_F (ObservingHandlerTest, ConnectError)
	TEST_F (ObservingHandlerTest, ConnectHandlerDeletion)
	TEST_F (ObservingHandlerTest, ConnectErrorHandlerDeletion)
	TEST_F (ObservingHandlerTest, BroadcastError)
	TEST_F (ObservingHandlerTest, ReadEOF)
	TEST_F (ObservingHandlerTest, ReadError)
	TEST_F (ObservingHandlerTest, WriteError)
	TEST_F (ObservingHandlerTest, WriteErrorHandlerDeletion)

Function Documentation

TEST_F	(	ObservingHandlerTest	,
		Success
	)

Definition at line 64 of file ObservingHandlerTest.cpp.

References testing::_, dummy(), EXPECT_CALL, testing::Invoke(), folly::makeFuture(), folly::makeMoveWrapper(), testing::Return(), and subscribe().

                                      {
  InSequence dummy;

  EXPECT_CALL(*prevHandler, transportActive(_))
      .WillOnce(Invoke([&](MockBytesToBytesHandler::Context* ctx) {
        ctx->fireTransportActive();
      }));
  // Verify that ingress is paused
  EXPECT_CALL(*prevHandler, transportInactive(_)).WillOnce(Return());
  EXPECT_CALL(pool, mockGetHandler(_))
      .WillOnce(Return(MoveWrapper<Future<BroadcastHandler<int, std::string>*>>(
          broadcastHandler.get())));
  EXPECT_CALL(*broadcastHandler, subscribe(_)).Times(1);
  // Verify that ingress is resumed
  EXPECT_CALL(*prevHandler, transportActive(_))
      .WillOnce(Invoke([&](MockBytesToBytesHandler::Context* ctx) {
        ctx->fireTransportActive();
      }));

  // Initialize the pipeline
  pipeline->transportActive();

  EXPECT_CALL(*observingHandler, mockWrite(_, 1))
      .WillOnce(Return(makeMoveWrapper(makeFuture())));
  EXPECT_CALL(*observingHandler, mockWrite(_, 2))
      .WillOnce(Return(makeMoveWrapper(makeFuture())));

  // Broadcast some data
  observingHandler->onNext(1);
  observingHandler->onNext(2);

  EXPECT_CALL(*observingHandler, mockClose(_))
      .WillOnce(Return(makeMoveWrapper(makeFuture())));

  // Finish the broadcast
  observingHandler->onCompleted();
}

TEST_F	(	ObservingHandlerTest	,
		ConnectError
	)

Definition at line 102 of file ObservingHandlerTest.cpp.

References testing::_, dummy(), EXPECT_CALL, testing::Invoke(), folly::makeFuture(), folly::makeMoveWrapper(), and testing::Return().

                                           {
  // Test when an error occurs while fetching the broadcast handler
  InSequence dummy;

  EXPECT_CALL(*prevHandler, transportActive(_))
      .WillOnce(Invoke([&](MockBytesToBytesHandler::Context* ctx) {
        ctx->fireTransportActive();
      }));
  // Verify that ingress is paused
  EXPECT_CALL(*prevHandler, transportInactive(_)).WillOnce(Return());
  // Inject error
  EXPECT_CALL(pool, mockGetHandler(_))
      .WillOnce(Return(MoveWrapper<Future<BroadcastHandler<int, std::string>*>>(
          make_exception_wrapper<std::exception>())));
  EXPECT_CALL(*observingHandler, mockClose(_))
      .WillOnce(Return(makeMoveWrapper(makeFuture())));

  // Initialize the pipeline
  pipeline->transportActive();
}

TEST_F	(	ObservingHandlerTest	,
		ConnectHandlerDeletion
	)

Definition at line 123 of file ObservingHandlerTest.cpp.

References testing::_, dummy(), EXPECT_CALL, folly::Promise< T >::getFuture(), testing::Invoke(), folly::makeMoveWrapper(), testing::Return(), folly::Promise< T >::setValue(), and subscribe().

                                                     {
  // Test when the handler goes away before the broadcast handler
  // is obtained
  InSequence dummy;

  EXPECT_CALL(*prevHandler, transportActive(_))
      .WillOnce(Invoke([&](MockBytesToBytesHandler::Context* ctx) {
        ctx->fireTransportActive();
      }));
  // Verify that ingress is paused
  EXPECT_CALL(*prevHandler, transportInactive(_)).WillOnce(Return());
  Promise<BroadcastHandler<int, std::string>*> promise;
  EXPECT_CALL(pool, mockGetHandler(_))
      .WillOnce(Return(makeMoveWrapper(promise.getFuture())));

  // Initialize the pipeline
  pipeline->transportActive();

  // Delete the handler and then fulfil the promise
  EXPECT_CALL(*broadcastHandler, subscribe(_)).Times(0);
  pipeline.reset();
  promise.setValue(broadcastHandler.get());
}

TEST_F	(	ObservingHandlerTest	,
		ConnectErrorHandlerDeletion
	)

Definition at line 147 of file ObservingHandlerTest.cpp.

References testing::_, dummy(), EXPECT_CALL, folly::Promise< T >::getFuture(), testing::Invoke(), folly::makeMoveWrapper(), testing::Return(), and folly::Promise< T >::setException().

                                                          {
  // Test when an error occurs while fetching the broadcast handler
  // after the handler is deleted
  InSequence dummy;

  EXPECT_CALL(*prevHandler, transportActive(_))
      .WillOnce(Invoke([&](MockBytesToBytesHandler::Context* ctx) {
        ctx->fireTransportActive();
      }));
  // Verify that ingress is paused
  EXPECT_CALL(*prevHandler, transportInactive(_)).WillOnce(Return());
  Promise<BroadcastHandler<int, std::string>*> promise;
  EXPECT_CALL(pool, mockGetHandler(_))
      .WillOnce(Return(makeMoveWrapper(promise.getFuture())));

  // Initialize the pipeline
  pipeline->transportActive();

  // Delete the handler and then inject an error
  pipeline.reset();
  promise.setException(std::exception());
}

TEST_F	(	ObservingHandlerTest	,
		BroadcastError
	)

Definition at line 170 of file ObservingHandlerTest.cpp.

References testing::_, dummy(), EXPECT_CALL, testing::Invoke(), folly::makeFuture(), folly::makeMoveWrapper(), testing::Return(), and subscribe().

                                             {
  InSequence dummy;

  EXPECT_CALL(*prevHandler, transportActive(_))
      .WillOnce(Invoke([&](MockBytesToBytesHandler::Context* ctx) {
        ctx->fireTransportActive();
      }));
  // Verify that ingress is paused
  EXPECT_CALL(*prevHandler, transportInactive(_)).WillOnce(Return());
  EXPECT_CALL(pool, mockGetHandler(_))
      .WillOnce(Return(MoveWrapper<Future<BroadcastHandler<int, std::string>*>>(
          broadcastHandler.get())));
  EXPECT_CALL(*broadcastHandler, subscribe(_)).Times(1);
  // Verify that ingress is resumed
  EXPECT_CALL(*prevHandler, transportActive(_))
      .WillOnce(Invoke([&](MockBytesToBytesHandler::Context* ctx) {
        ctx->fireTransportActive();
      }));

  // Initialize the pipeline
  pipeline->transportActive();

  EXPECT_CALL(*observingHandler, mockWrite(_, _))
      .WillOnce(Return(makeMoveWrapper(makeFuture())));

  // Broadcast some data
  observingHandler->onNext(1);

  EXPECT_CALL(*observingHandler, mockClose(_))
      .WillOnce(Return(makeMoveWrapper(makeFuture())));

  // Inject broadcast error
  observingHandler->onError(make_exception_wrapper<std::exception>());
}

TEST_F	(	ObservingHandlerTest	,
		ReadEOF
	)

Definition at line 205 of file ObservingHandlerTest.cpp.

References testing::_, dummy(), EXPECT_CALL, testing::Invoke(), testing::InvokeWithoutArgs(), folly::makeFuture(), folly::makeMoveWrapper(), testing::Return(), and subscribe().

                                      {
  InSequence dummy;

  EXPECT_CALL(*prevHandler, transportActive(_))
      .WillOnce(Invoke([&](MockBytesToBytesHandler::Context* ctx) {
        ctx->fireTransportActive();
      }));
  // Verify that ingress is paused
  EXPECT_CALL(*prevHandler, transportInactive(_)).WillOnce(Return());
  EXPECT_CALL(pool, mockGetHandler(_))
      .WillOnce(Return(MoveWrapper<Future<BroadcastHandler<int, std::string>*>>(
          broadcastHandler.get())));
  EXPECT_CALL(*broadcastHandler, subscribe(_)).Times(1);
  // Verify that ingress is resumed
  EXPECT_CALL(*prevHandler, transportActive(_))
      .WillOnce(Invoke([&](MockBytesToBytesHandler::Context* ctx) {
        ctx->fireTransportActive();
      }));

  // Initialize the pipeline
  pipeline->transportActive();

  EXPECT_CALL(*observingHandler, mockWrite(_, _))
      .WillOnce(Return(makeMoveWrapper(makeFuture())));

  // Broadcast some data
  observingHandler->onNext(1);

  EXPECT_CALL(*observingHandler, mockClose(_))
      .WillOnce(InvokeWithoutArgs([&] {
        // Delete the pipeline
        pipeline.reset();
        return makeMoveWrapper(makeFuture());
      }));
  EXPECT_CALL(*broadcastHandler, unsubscribe(_)).Times(1);

  // Client closes connection
  observingHandler->readEOF(nullptr);
}

TEST_F	(	ObservingHandlerTest	,
		ReadError
	)

Definition at line 245 of file ObservingHandlerTest.cpp.

References testing::_, dummy(), EXPECT_CALL, testing::Invoke(), testing::InvokeWithoutArgs(), folly::makeFuture(), folly::makeMoveWrapper(), testing::Return(), and subscribe().

                                        {
  InSequence dummy;

  EXPECT_CALL(*prevHandler, transportActive(_))
      .WillOnce(Invoke([&](MockBytesToBytesHandler::Context* ctx) {
        ctx->fireTransportActive();
      }));
  // Verify that ingress is paused
  EXPECT_CALL(*prevHandler, transportInactive(_)).WillOnce(Return());
  EXPECT_CALL(pool, mockGetHandler(_))
      .WillOnce(Return(MoveWrapper<Future<BroadcastHandler<int, std::string>*>>(
          broadcastHandler.get())));
  EXPECT_CALL(*broadcastHandler, subscribe(_)).Times(1);
  // Verify that ingress is resumed
  EXPECT_CALL(*prevHandler, transportActive(_))
      .WillOnce(Invoke([&](MockBytesToBytesHandler::Context* ctx) {
        ctx->fireTransportActive();
      }));

  // Initialize the pipeline
  pipeline->transportActive();

  EXPECT_CALL(*observingHandler, mockWrite(_, _))
      .WillOnce(Return(makeMoveWrapper(makeFuture())));

  // Broadcast some data
  observingHandler->onNext(1);

  EXPECT_CALL(*observingHandler, mockClose(_))
      .WillOnce(InvokeWithoutArgs([&] {
        // Delete the pipeline
        pipeline.reset();
        return makeMoveWrapper(makeFuture());
      }));
  EXPECT_CALL(*broadcastHandler, unsubscribe(_)).Times(1);

  // Inject read error
  observingHandler->readException(nullptr,
                                  make_exception_wrapper<std::exception>());
}

TEST_F	(	ObservingHandlerTest	,
		WriteError
	)

Definition at line 286 of file ObservingHandlerTest.cpp.

References testing::_, dummy(), EXPECT_CALL, testing::Invoke(), testing::InvokeWithoutArgs(), folly::makeFuture(), folly::makeMoveWrapper(), testing::Return(), and subscribe().

                                         {
  InSequence dummy;

  EXPECT_CALL(*prevHandler, transportActive(_))
      .WillOnce(Invoke([&](MockBytesToBytesHandler::Context* ctx) {
        ctx->fireTransportActive();
      }));
  // Verify that ingress is paused
  EXPECT_CALL(*prevHandler, transportInactive(_)).WillOnce(Return());
  EXPECT_CALL(pool, mockGetHandler(_))
      .WillOnce(Return(MoveWrapper<Future<BroadcastHandler<int, std::string>*>>(
          broadcastHandler.get())));
  EXPECT_CALL(*broadcastHandler, subscribe(_)).Times(1);
  // Verify that ingress is resumed
  EXPECT_CALL(*prevHandler, transportActive(_))
      .WillOnce(Invoke([&](MockBytesToBytesHandler::Context* ctx) {
        ctx->fireTransportActive();
      }));

  // Initialize the pipeline
  pipeline->transportActive();

  // Inject write error
  EXPECT_CALL(*observingHandler, mockWrite(_, _))
      .WillOnce(Return(
          MoveWrapper<Future<Unit>>(make_exception_wrapper<std::exception>())));
  EXPECT_CALL(*observingHandler, mockClose(_))
      .WillOnce(InvokeWithoutArgs([&] {
        // Delete the pipeline
        pipeline.reset();
        return makeMoveWrapper(makeFuture());
      }));
  EXPECT_CALL(*broadcastHandler, unsubscribe(_)).Times(1);

  // Broadcast some data
  observingHandler->onNext(1);
}

TEST_F	(	ObservingHandlerTest	,
		WriteErrorHandlerDeletion
	)

Definition at line 324 of file ObservingHandlerTest.cpp.

References testing::_, dummy(), EXPECT_CALL, folly::Promise< T >::getFuture(), testing::Invoke(), folly::makeMoveWrapper(), testing::Return(), folly::Promise< T >::setException(), and subscribe().

                                                        {
  // Test when write error occurs asynchronously after the handler
  // has been deleted.
  InSequence dummy;

  EXPECT_CALL(*prevHandler, transportActive(_))
      .WillOnce(Invoke([&](MockBytesToBytesHandler::Context* ctx) {
        ctx->fireTransportActive();
      }));
  // Verify that ingress is paused
  EXPECT_CALL(*prevHandler, transportInactive(_)).WillOnce(Return());
  EXPECT_CALL(pool, mockGetHandler(_))
      .WillOnce(Return(MoveWrapper<Future<BroadcastHandler<int, std::string>*>>(
          broadcastHandler.get())));
  EXPECT_CALL(*broadcastHandler, subscribe(_)).Times(1);
  // Verify that ingress is resumed
  EXPECT_CALL(*prevHandler, transportActive(_))
      .WillOnce(Invoke([&](MockBytesToBytesHandler::Context* ctx) {
        ctx->fireTransportActive();
      }));

  // Initialize the pipeline
  pipeline->transportActive();

  Promise<Unit> promise;
  EXPECT_CALL(*observingHandler, mockWrite(_, _))
      .WillOnce(Return(makeMoveWrapper(promise.getFuture())));

  // Broadcast some data
  observingHandler->onNext(1);

  // Delete the pipeline and then fail the write
  EXPECT_CALL(*broadcastHandler, unsubscribe(_)).Times(1);
  pipeline.reset();
  promise.setException(std::exception());
}