public void Should_allow_publish_request_with_custom_headers()
        {
            var pongReceived = new FutureMessage<PongMessage>();
            var pingReceived = new FutureMessage<PingMessage>();

            RemoteBus.SubscribeContextHandler<PingMessage>(x =>
            {
                pingReceived.Set(x.Message);
                var transactionIdFromHeader = new Guid(x.Headers["PingTransactionId"]);
                x.Respond(new PongMessage { TransactionId = transactionIdFromHeader });
            });
            LocalBus.ShouldHaveSubscriptionFor<PingMessage>();

            var ping = new PingMessage { TransactionId = Guid.NewGuid() };

            TimeSpan timeout = 8.Seconds();

            LocalBus.PublishRequest(ping, x =>
            {
                x.Handle<PongMessage>(message =>
                {
                    message.TransactionId.ShouldEqual(ping.TransactionId,
                        "The response correlationId did not match");
                    pongReceived.Set(message);
                });

                x.SetTimeout(timeout);
            }, ctx =>
            {
                ctx.SetHeader("PingTransactionId", ping.TransactionId.ToString());
            });

            pingReceived.IsAvailable(timeout).ShouldBeTrue("The ping was not received");
            pongReceived.IsAvailable(timeout).ShouldBeTrue("The pong was not received");
        }

		protected override void EstablishContext()
		{
			base.EstablishContext();

			_ping = new PingMessage();
			_future = new FutureMessage<PingMessage, Guid>(_ping.CorrelationId);

			_unsubscribe = RemoteBus.SubscribeHandler<PingMessage>(message => { _future.Set(message); });


			RemoteBus.ShouldHaveRemoteSubscriptionFor<PingMessage>();

			//LocalBus.ShouldHaveRemoteSubscriptionFor<PingMessage>();

			Trace.WriteLine("LocalBus");

			LocalBus.OutboundPipeline.Trace();

			Trace.WriteLine("RemoteBus");

			RemoteBus.OutboundPipeline.Trace();


			LocalBus.Publish(_ping);
		}

        public void Adding_many_dynamic_and_removing_should_retain_dynamics()
        {
            var dynamicA = new FutureMessage<A>();
            var dynamicB = new FutureMessage<B>();
            var dynamicC = new FutureMessage<C>();
            var dynamicD = new FutureMessage<D>();

            UnsubscribeAction subscriptionA = RemoteBus.SubscribeHandler<A>(dynamicA.Set);
            UnsubscribeAction subscriptionB = RemoteBus.SubscribeHandler<B>(dynamicB.Set);
            UnsubscribeAction subscriptionC = RemoteBus.SubscribeHandler<C>(dynamicC.Set);
            UnsubscribeAction subscriptionD = RemoteBus.SubscribeHandler<D>(dynamicD.Set);

            LocalBus.HasSubscription<D>(8.Seconds()).Any().ShouldBeTrue("No D subscription");
            try
            {
                subscriptionA().ShouldBeFalse("A static not remaining");
                subscriptionB().ShouldBeFalse("B static not remaining");
                subscriptionC().ShouldBeFalse("C static not remaining");

                LocalBus.Publish(new A());
                LocalBus.Publish(new B());
                LocalBus.Publish(new C());
                LocalBus.Publish(new D());

                _receivedA.IsAvailable(8.Seconds()).ShouldBeTrue("A not received");
                _receivedB.IsAvailable(8.Seconds()).ShouldBeTrue("B not received");
                _receivedC.IsAvailable(8.Seconds()).ShouldBeTrue("C not received");
                dynamicD.IsAvailable(8.Seconds()).ShouldBeTrue("D should have been received");
            }
            finally
            {
                subscriptionD();
            }
        }

		public void A_consumer_is_being_tested()
		{
			_test = TestFactory.ForHandler<InputMessage>()
				.New(x =>
					{
						x.Handler((bus,message) =>
							{
								bus.MessageContext<InputMessage>().Respond(new OutputMessage());
							});

					x.Send(new InputMessage(), c => c.SendResponseTo(_test.Scenario.Bus));
					x.Send(new InputMessage(), c => c.SendResponseTo(_test.Scenario.Bus));
				});

			_test.Execute();

			_test.Received.Any<InputMessage>().ShouldBeTrue();
			_test.Sent.Any<OutputMessage>().ShouldBeTrue();

			_future = new FutureMessage<ReceivedMessageTraceList>();
			_test.Scenario.Bus.GetMessageTrace(_test.Scenario.Bus.ControlBus.Endpoint, _future.Set);

			_future.IsAvailable(8.Seconds()).ShouldBeTrue();
			_list = _future.Message;
		}

		public void Should_use_a_clean_syntax_following_standard_conventions()
		{
			var pongReceived = new FutureMessage<PongMessage>();
			var pingReceived = new FutureMessage<PingMessage>();

			RemoteBus.SubscribeHandler<PingMessage>(x =>
				{
					pingReceived.Set(x);
					RemoteBus.MessageContext<PingMessage>().Respond(new PongMessage(x.CorrelationId));
				});

			var ping = new PingMessage();

			var timeout = 8.Seconds();

			LocalBus.PublishRequest(ping, x =>
				{
					x.Handle<PongMessage>(message =>
						{
							message.CorrelationId.ShouldEqual(ping.CorrelationId, "The response correlationId did not match");
							pongReceived.Set(message);
						});

					x.SetTimeout(timeout);
				});

			pingReceived.IsAvailable(timeout).ShouldBeTrue("The ping was not received");
			pongReceived.IsAvailable(timeout).ShouldBeTrue("The pong was not received");
		}

        public void A_response_should_be_sent_directly_if_a_reply_address_is_specified()
        {
            var ping = new PingMessage();

            var otherConsumer = new TestMessageConsumer<PongMessage>();
            RemoteBus.SubscribeInstance(otherConsumer);

            var consumer = new TestCorrelatedConsumer<PongMessage, Guid>(ping.CorrelationId);
            LocalBus.SubscribeInstance(consumer);

            var pong = new FutureMessage<PongMessage>();

            RemoteBus.SubscribeHandler<PingMessage>(message =>
                {
                    pong.Set(new PongMessage(message.CorrelationId));

                    RemoteBus.Context().Respond(pong.Message);
                });

            RemoteBus.ShouldHaveRemoteSubscriptionFor<PongMessage>();
            LocalBus.ShouldHaveRemoteSubscriptionFor<PongMessage>();
            LocalBus.ShouldHaveRemoteSubscriptionFor<PingMessage>();

            LocalBus.Publish(ping, context => context.SendResponseTo(LocalBus));

            Assert.IsTrue(pong.IsAvailable(8.Seconds()), "No pong generated");

            consumer.ShouldHaveReceivedMessage(pong.Message, 8.Seconds());
            otherConsumer.ShouldNotHaveReceivedMessage(pong.Message, 1.Seconds());
        }

		public void Should_support_the_asynchronous_programming_model()
		{
			var pongReceived = new FutureMessage<PongMessage>();
			var pingReceived = new FutureMessage<PingMessage>();
			var callbackCalled = new FutureMessage<IAsyncResult>();

			RemoteBus.SubscribeHandler<PingMessage>(x =>
				{
					pingReceived.Set(x);
					RemoteBus.MessageContext<PingMessage>().Respond(new PongMessage(x.CorrelationId));
				});

			var ping = new PingMessage();

			var timeout = 8.Seconds();

			LocalBus.BeginPublishRequest(ping, callbackCalled.Set, null, x =>
				{
					x.Handle<PongMessage>(message =>
						{
							message.CorrelationId.ShouldEqual(ping.CorrelationId, "The response correlationId did not match");
							pongReceived.Set(message);
						});

					x.SetTimeout(timeout);
				});

			callbackCalled.IsAvailable(timeout).ShouldBeTrue("The callback was not called");

			LocalBus.EndRequest(callbackCalled.Message);

			pingReceived.IsAvailable(timeout).ShouldBeTrue("The ping was not received");
			pongReceived.IsAvailable(timeout).ShouldBeTrue("The pong was not received");
		}

		public void The_retry_count_should_be_set_on_the_message()
		{
			FutureMessage<PingMessage> future = new FutureMessage<PingMessage>();

			bool first = true;

			LocalBus.Subscribe<PingMessage>(message =>
				{
					if(first)
					{
						Assert.AreEqual(0, CurrentMessage.Headers.RetryCount);

						CurrentMessage.RetryLater();

						first = false;
					}
					else
					{
						Assert.AreEqual(1, CurrentMessage.Headers.RetryCount);

						future.Set(message);
					}
				});

			LocalBus.Publish(new PingMessage());

			Assert.IsTrue(future.IsAvailable(5.Seconds()));
		}

        public void Should_support_send_as_well()
        {
            var pongReceived = new FutureMessage<PongMessage>();
            var pingReceived = new FutureMessage<PingMessage>();

            RemoteBus.SubscribeContextHandler<PingMessage>(x =>
                {
                    pingReceived.Set(x.Message);
                    x.Respond(new PongMessage {TransactionId = x.Message.TransactionId});
                });
            LocalBus.ShouldHaveSubscriptionFor<PingMessage>();

            var ping = new PingMessage();

            TimeSpan timeout = 8.Seconds();

            RemoteBus.Endpoint.SendRequest(ping, LocalBus, x =>
                {
                    x.Handle<PongMessage>(message =>
                        {
                            message.TransactionId.ShouldEqual(ping.TransactionId,
                                "The response correlationId did not match");
                            pongReceived.Set(message);
                        });

                    x.SetTimeout(timeout);
                });

            pingReceived.IsAvailable(timeout).ShouldBeTrue("The ping was not received");
            pongReceived.IsAvailable(timeout).ShouldBeTrue("The pong was not received");
        }

		public void A_response_should_be_published_if_no_reply_address_is_specified()
		{
			var ping = new PingMessage();

			var otherConsumer = new TestMessageConsumer<PongMessage>();
			RemoteBus.SubscribeInstance(otherConsumer);

			var consumer = new TestCorrelatedConsumer<PongMessage, Guid>(ping.CorrelationId);
			LocalBus.SubscribeInstance(consumer);

			var pong = new FutureMessage<PongMessage>();

			RemoteBus.SubscribeHandler<PingMessage>(message =>
				{
					pong.Set(new PongMessage(message.CorrelationId));

					RemoteBus.Context().Respond(pong.Message);
				});

			LocalBus.Publish(ping);

			pong.IsAvailable(8.Seconds()).ShouldBeTrue("No pong generated");

			consumer.ShouldHaveReceivedMessage(pong.Message, 8.Seconds());
			otherConsumer.ShouldHaveReceivedMessage(pong.Message, 8.Seconds());
		}

		public void The_retry_count_should_be_set_on_the_message()
		{
			var future = new FutureMessage<PingMessage>();

			bool first = true;

			LocalBus.SubscribeHandler<PingMessage>(message =>
				{
					if (first)
					{
						Assert.AreEqual(0, LocalBus.Context().RetryCount);

						LocalBus.MessageContext<PingMessage>().RetryLater();

						first = false;
					}
					else
					{
						Assert.AreEqual(1, LocalBus.Context().RetryCount);

						future.Set(message);
					}
				});

			LocalBus.ShouldHaveRemoteSubscriptionFor<PingMessage>();

			LocalBus.Publish(new PingMessage());

			Assert.IsTrue(future.IsAvailable(20.Seconds()));
		}

        public void Should_ignore_a_response_that_was_not_for_us()
        {
            var pongReceived = new FutureMessage<PongMessage>();
            var pingReceived = new FutureMessage<PingMessage>();
            var badResponse = new FutureMessage<PongMessage>();

            LocalBus.SubscribeHandler<PongMessage>(pongReceived.Set);

            RemoteBus.SubscribeContextHandler<PingMessage>(x =>
                {
                    pingReceived.Set(x.Message);
                    RemoteBus.Publish(new PongMessage {TransactionId = x.Message.TransactionId});
                });
            LocalBus.ShouldHaveSubscriptionFor<PingMessage>();

            var ping = new PingMessage();

            TimeSpan timeout = 8.Seconds();

            Assert.Throws<RequestTimeoutException>(() =>
                {
                    RemoteBus.Endpoint.SendRequest(ping, LocalBus, x =>
                        {
                            x.Handle<PongMessage>(badResponse.Set);

                            x.SetTimeout(timeout);
                        });
                });

            pingReceived.IsAvailable(timeout).ShouldBeTrue("The ping was not received");
            pongReceived.IsAvailable(timeout).ShouldBeTrue("The pong was not received");
            badResponse.IsAvailable(2.Seconds()).ShouldBeFalse("Should not have received a response");
        }

		public void A_response_should_be_published_if_no_reply_address_is_specified()
		{
			PingMessage ping = new PingMessage();

			TestMessageConsumer<PongMessage> otherConsumer = new TestMessageConsumer<PongMessage>();
			RemoteBus.Subscribe(otherConsumer);

			TestCorrelatedConsumer<PongMessage, Guid> consumer = new TestCorrelatedConsumer<PongMessage, Guid>(ping.CorrelationId);
			LocalBus.Subscribe(consumer);

			FutureMessage<PongMessage> pong = new FutureMessage<PongMessage>();

			RemoteBus.Subscribe<PingMessage>(message =>
				{
					pong.Set(new PongMessage(message.CorrelationId));

					CurrentMessage.Respond(pong.Message);
				});

			LocalBus.Publish(ping);

			Assert.IsTrue(pong.IsAvailable(3.Seconds()), "No pong generated");

			consumer.ShouldHaveReceivedMessage(pong.Message, 3.Seconds());
			otherConsumer.ShouldHaveReceivedMessage(pong.Message, 1.Seconds());
		}

		public void A_clean_method_of_a_request_reply_should_be_possible()
		{
			var ponged = new FutureMessage<PongMessage>();

			RemoteBus.SubscribeHandler<PingMessage>(x =>
				{
					// timing issue here it seems, but that's what Respond() is for, to RESPOND to messages
					// and not publish responses
					RemoteBus.ShouldHaveSubscriptionFor<PongMessage>();

					RemoteBus.Publish(new PongMessage(x.CorrelationId));
				});

			LocalBus.ShouldHaveSubscriptionFor<PingMessage>();

			var ping = new PingMessage();


			LocalBus.PublishRequest(ping, x =>
				{
					x.Handle<PongMessage>(message =>
						{
							message.CorrelationId.ShouldEqual(ping.CorrelationId);
							ponged.Set(message);
						});

					x.HandleTimeout(8.Seconds(), () => { });
				});

			ponged.IsAvailable(8.Seconds()).ShouldBeTrue("No ping response received");
		}

		protected override void EstablishContext()
		{
			base.EstablishContext();

			_future = new FutureMessage<PingMessage>();

			LocalBus.Subscribe<PingMessage>(message => _future.Set(message));

			_ping = new PingMessage();

			LocalBus.Publish(_ping);
		}

		private void MakeSagaDown()
		{
			MakeSagaSuspect();
			var fm = new FutureMessage<TimeoutExpired>();
			RemoteBus.Subscribe<TimeoutExpired>(fm.Set);
			Thread.Sleep(500);
			LocalBus.Publish(new TimeoutExpired {CorrelationId = _id, Tag = 2});
			fm.IsAvailable(1.Seconds()).ShouldBeTrue();

			var saga = Repository.Where(x => x.CorrelationId == _id).First();
			saga.CurrentState.ShouldEqual(HealthSaga.Down, "MakeSagaDown failed");
		}

		public void Subscring_to_an_endpoint_should_accept_and_dispatch_messages()
		{
            FutureMessage<PingMessage> fm=new FutureMessage<PingMessage>();
			bool workDid = false;

			LocalBus.Subscribe<PingMessage>(
				(msg)=> {workDid = true; fm.Set(msg); },
				delegate { return true; });

			RemoteBus.Publish(_message);
		    fm.IsAvailable(1.Seconds());
			Assert.That(workDid, Is.True, "Lazy Test!");
		}

        public void Dynamic_should_receive_the_message()
        {
            var receivedDynamic = new FutureMessage<A>();

            UnsubscribeAction subscription = RemoteBus.SubscribeHandler<A>(receivedDynamic.Set);
            using (subscription.Disposable())
            {
                LocalBus.Publish(new A());

                receivedDynamic.IsAvailable(8.Seconds()).ShouldBeTrue("Dynamic not received");
                _receivedA.IsAvailable(8.Seconds()).ShouldBeTrue("Static not received");
            }
        }

		public void MakeSagaSuspect()
		{
			LocalBus.Publish(new EndpointCameOnline(_id, LocalBus.ControlBus.Endpoint.Uri,LocalBus.Endpoint.Uri, 0));
			var fm = new FutureMessage<TimeoutExpired>();
			RemoteBus.Subscribe<TimeoutExpired>(fm.Set);
			Thread.Sleep(5000);
			LocalBus.Publish(new TimeoutExpired {CorrelationId = _id, Tag = 1});
			fm.IsAvailable(5.Seconds()).ShouldBeTrue();

			var saga = Repository.Where(x => x.CorrelationId == _id).FirstOrDefault();
			saga.ShouldNotBeNull();
			saga.CurrentState.ShouldEqual(HealthSaga.Suspect, "MakeSagaSuspect failed");
		}

		public void Configuring_a_service_bus_easily()
		{
			FutureMessage<PingMessage> received;
			using (IServiceBus bus = ServiceBusFactory.New(x => { x.ReceiveFrom("loopback://localhost/queue"); }))
			{
				received = new FutureMessage<PingMessage>();

				bus.SubscribeHandler<PingMessage>(received.Set);

				bus.Publish(new PingMessage());
				received.IsAvailable(8.Seconds()).ShouldBeTrue();
			}
		}