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 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 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 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 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");
        }

		private void VerifyMessageHeaderIsPassed(Action<IOutboundMessage> setHeaderAction, Action<IInboundMessageHeaders> checkHeaderAction)
		{
			byte[] data;
			var serializer = new XmlMessageSerializer();

			OutboundMessage.Set(setHeaderAction);

			var message = new PingMessage();

			using (MemoryStream output = new MemoryStream())
			{
				serializer.Serialize(output, message);

				data = output.ToArray();
			}

			Trace.WriteLine(OutboundMessage.Headers.MessageType);

			Trace.WriteLine(Encoding.UTF8.GetString(data));

			using (MemoryStream input = new MemoryStream(data))
			{
				serializer.Deserialize(input);

				checkHeaderAction(CurrentMessage.Headers);
			}
		}

        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 async Task Should_be_received_properly()
        {
            object message = new PingMessage();
            await Bus.Publish(message);

            await _received;
        }

		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");
		}

		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 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 The_pipeline_should_have_insertable_items()
		{
			// two consumers, one for each type of message

			IndiscriminantConsumer<PingMessage> pingConsumer = new IndiscriminantConsumer<PingMessage>();
			IndiscriminantConsumer<PongMessage> pongConsumer = new IndiscriminantConsumer<PongMessage>();

			UnsubscribeAction pingToken = _pipeline.Subscribe(pingConsumer);
			UnsubscribeAction pongToken = _pipeline.Subscribe(pongConsumer);

			PipelineViewer.Trace(_pipeline);

			PingMessage pingMessage = new PingMessage();
			PongMessage pongMessage = new PongMessage();

			_pipeline.Dispatch(pingMessage, accept => true);
			_pipeline.Dispatch(pongMessage, accept => true);

			Assert.AreEqual(pingMessage, pingConsumer.Consumed);
			Assert.AreEqual(pongMessage, pongConsumer.Consumed);

			pingToken();
			pongToken();

			PipelineViewer.Trace(_pipeline);
		}

        public async Task Should_end_well()
        {
            var timer = Stopwatch.StartNew();

            Task[] publishers = new Task[100*1000];
            Parallel.For(0, 100, i =>
            {
                var offset = i * 1000;

                for (int j = 0; j < 1000; j++)
                {
                    var ping = new PingMessage();
                    var task = Bus.Publish(ping);
                    publishers[offset + j] = task;
                }
            });

            var published = timer.Elapsed;

            await Task.WhenAll(publishers);

            var confirmed = timer.Elapsed;

            Console.WriteLine("Published {0} messages in {1}ms ({2:F0}/s)", 100 * 1000, published.TotalMilliseconds, 100L * 1000L * 1000L / published.TotalMilliseconds);

            Console.WriteLine("Confirmed {0} messages in {1}ms ({2:F0}/s)", 100 * 1000, confirmed.TotalMilliseconds, 100L * 1000L * 1000L / confirmed.TotalMilliseconds);

            await _completed.Task;

            var completed = timer.Elapsed;

            Console.WriteLine("Completed {0} messages in {1}ms ({2:F0}/s)", 100 * 1000, completed.TotalMilliseconds, 100L * 1000L * 1000L / completed.TotalMilliseconds);
        }

 public async Task Setup()
 {
     var message = new PingMessage();
     await InputQueueSendEndpoint.Send(message, context =>
     {
         context.SetSessionId(message.CorrelationId.ToString());
     });
 }

		public void It_should_be_received_by_a_component()
		{
			RemoteBus.SubscribeConsumer<TestMessageConsumer<PingMessage>>();

			PingMessage message = new PingMessage();
			LocalBus.Publish(message);

			TestConsumerBase<PingMessage>.AnyShouldHaveReceivedMessage(message, _timeout);
		}

 public async Task Setup()
 {
     _pingMessage = new PingMessage();
     _pingMessage2 = new PingMessage();
     await InputQueueSendEndpoint.Send(_pingMessage, Pipe.Execute<SendContext<PingMessage>>(context =>
     {
         context.CorrelationId = _correlationId;
     }));
 }

		public void A_simple_bus_should_be_able_to_subscribe_and_publish()
		{
			TestMessageConsumer<PingMessage> consumer = new TestMessageConsumer<PingMessage>();
			LocalBus.Subscribe(consumer);

			PingMessage message = new PingMessage();
			LocalBus.Publish(message);

			consumer.ShouldHaveReceivedMessage(message, 500.Milliseconds());
		}

        public async Task Should_include_a_correlation_id()
        {
            var pingMessage = new PingMessage();
            await Bus.Publish(pingMessage);

            var context = await _handled;

            context.CorrelationId.HasValue.ShouldBe(true);
            context.CorrelationId.Value.ShouldBe(pingMessage.CorrelationId);
        }

        public void A_simple_bus_should_be_able_to_subscribe_and_publish()
        {
            var consumer = new TestMessageConsumer<PingMessage>();
            LocalBus.SubscribeInstance(consumer);

            var message = new PingMessage();
            LocalBus.Publish(message);

            consumer.ShouldHaveReceivedMessage(message, 8.Seconds());
        }

		public void It_should_be_received_by_one_subscribed_consumer()
		{
			TestMessageConsumer<PingMessage> consumer = new TestMessageConsumer<PingMessage>();
			RemoteBus.SubscribeInstance(consumer);

			PingMessage message = new PingMessage();
			LocalBus.Publish(message);

			consumer.ShouldHaveReceivedMessage(message, _timeout);
		}