protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
 {
     Handler<PingMessage>(configurator, async context =>
     {
         throw new SerializationException("This is fine, forcing death");
     });
 }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            _consumer = new OneMessageConsumer(GetTask<MessageA>());


            configurator.Consumer(() => _consumer);
        }

 protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
 {
     configurator.Handler<PingMessage>(async context =>
     {
         await context.RespondAsync(new PongMessage(context.Message.CorrelationId));
     });
 }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            _bookingRequestSagaRepository = new InMemorySagaRepository<BookingRequestState>();
            _bookingRequestStateMachine = new BookingRequestStateMachine();

            configurator.StateMachineSaga(_bookingRequestStateMachine, _bookingRequestSagaRepository);
        }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            _count = 0;

            _received = GetTask<ConsumeContext<PingMessage>>();

            configurator.Handler<PingMessage>(async context =>
            {
                if (_timer == null)
                    _timer = Stopwatch.StartNew();

                if (_count++ < 2)
                {
                    Console.WriteLine("{0} now is not a good time", DateTime.UtcNow);
                    throw new IntentionalTestException("I'm so not ready for this jelly.");
                }

                _timer.Stop();

                Console.WriteLine("{0} okay, now is good (retried {1} times)", DateTime.UtcNow, context.Headers.Get("MT-Redelivery-Count", default(int?)));

                // okay, ready.
                _receivedTimeSpan = _timer.Elapsed;
                _received.TrySetResult(context);
            }, x => x.UseScheduledRedelivery(Retry.Intervals(1000,2000)));
        }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            _observer = GetObserver<A>();

            configurator.Observer(_observer);

            configurator.Observer(Observer.Create<ConsumeContext<A>>(m => Console.WriteLine(m.Message.Name)));
        }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            _completed = Handled<RoutingSlipCompleted>(configurator);

            var fetchAvatarActivity = GetActivityContext<FetchAvatarActivity>();

            _activityCompleted = Handled<RoutingSlipActivityCompleted>(configurator, context => context.Message.ActivityName.Equals(fetchAvatarActivity.Name));
        }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            base.ConfigureInputQueueEndpoint(configurator);

            var testActivity = GetActivityContext<TestActivity>();

            _handled = Handled<RoutingSlipActivityCompleted>(configurator, x => x.Message.ActivityName == testActivity.Name);
        }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            // TODO would be nice to support serialization per receiving endpoint

            // configurator.UseJsonSerializer();

            _requestReceived = Handler<A>(configurator, context => context.RespondAsync(new B()));
        }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            _observer = new ObservableObserver<ConsumeContext<A>>();

            _observer.GroupBy(x => x.Message.Name).Subscribe(value => Console.WriteLine("Key: {0}", value.Key));

            configurator.Observer(_observer);
        }

 protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
 {
     long value = 0;
     configurator.Handler<PingMessage>(async context =>
     {
         if (Interlocked.Increment(ref value) == 1000)
             _completed.TrySetResult(true);
     });
 }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            _first = Handler<FirstMessage>(configurator, async context =>
            {
                await context.ScheduleMessage(DateTime.Now, new SecondMessage());
            });

            _second = Handled<SecondMessage>(configurator);
        }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            configurator.Handler<JToken>(async context =>
            {
                await Console.Out.WriteLineAsync($"Received the token! {context.Message}");

                _completed.TrySetResult(context.Message);
            });
        }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            _count = 0;
            configurator.Handler<Interval>(async context =>
            {
                Interlocked.Increment(ref _count);
            });

            _done = Handled<Done>(configurator);
        }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            var sagaRepository = new InMemorySagaRepository<RegisterUserSaga>();
            configurator.Saga(sagaRepository);

            configurator.Handler<SendUserVerificationEmail>(async x =>
            {
                await Bus.Publish(new UserVerificationEmailSent(x.Message.CorrelationId, x.Message.Email));
            });
        }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            base.ConfigureInputQueueEndpoint(configurator);

            configurator.Turnout<ProcessFile>(_busFactoryConfigurator, x =>
            {
                x.SuperviseInterval = TimeSpan.FromSeconds(1);
                x.SetJobFactory(async context => await Task.Delay(context.Message.Size));
            });
        }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            configurator.UseRetry(Retry.None);

            Handler<PingMessage>(configurator, async context =>
            {
                _attempts++;
                throw new IntentionalTestException();
            });
        }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            base.ConfigureInputQueueEndpoint(configurator);

            configurator.UseTransform<A>(t =>
            {
                t.Set(x => x.Second, context => "World");
            });

            _received = Handled<A>(configurator);
        }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            _commandHandler = Handler<SecureCommand>(configurator, async context =>
            {
                ConsumeContext<UserCredentials> credentials;
                if (context.TryGetMessage(out credentials))
                    _credentials.SetResult(credentials.Message);
            });

            _credentialsHandler = Handled<UserCredentials>(configurator);
        }

        protected override void ConfigureInputQueueEndpoint(IInMemoryReceiveEndpointConfigurator configurator)
        {
            base.ConfigureInputQueueEndpoint(configurator);

            configurator.UseBsonSerializer();

            _handled = Handler<PingMessage>(configurator, async context =>
            {
                await context.RespondAsync(new PongMessage(context.Message.CorrelationId));
            });
        }