public void Should_property_adapt_itself_to_a_channel_network()
        {
            TraceLogger.Configure(LogLevel.Debug);
            ILogger log = Logger.GetLogger<Sending_a_message_through_a_wcf_channel>();
            log.Debug("Starting");

            var serviceUri = new Uri("net.pipe://localhost/Pipe");
            string pipeName = "Test";
            Channel<TestMessage> adapter = new ChannelAdapter<TestMessage>();
            using (var host = new WcfChannelHost<TestMessage>(adapter, serviceUri, pipeName))
            {
                log.Debug("Host started");

                var future = new Future<TestMessage>();

                using (adapter.Connect(x =>
                    {
                        x.AddConsumer(m =>
                            {
                                log.Debug(l => l.Write("Received: {0}", m.Value));
                                future.Complete(m);
                            });
                    }))
                {
                    var client = new WcfChannelProxy<TestMessage>(new SynchronousFiber(), serviceUri, pipeName);
                    log.Debug("Client started");

                    client.Send(new TestMessage("Hello!"));

                    future.WaitUntilCompleted(2.Seconds()).ShouldBeTrue();

                    log.Debug("Complete");
                }
            }
        }

        public static WcfChannelHost GetCurrentShelfHost(ChannelAdapter myChannel)
        {
            var pipeName = GetThisShelfPipeName();

            var address = GetBaseAddress(pipeName);
            return new WcfChannelHost(new ThreadPoolFiber(), myChannel, address, "shelf");
        }

        /// <summary>
        /// Creates a PollingFileSystemEventProducer
        /// </summary>		
        /// <param name="directory">The directory to watch</param>
        /// <param name="channel">The channel where events should be sent</param>
        /// <param name="scheduler">Event scheduler</param>
        /// <param name="fiber">Fiber to schedule on</param>
        /// <param name="checkInterval">The maximal time between events or polls on a given file</param>
        /// <param name="checkSubDirectory">Indicates if subdirectorys will be checked or ignored</param>
        public PollingFileSystemEventProducer(string directory, UntypedChannel channel, [NotNull] Scheduler scheduler, Fiber fiber, TimeSpan checkInterval, bool checkSubDirectory)
        {
            if (scheduler == null)
                throw new ArgumentNullException("scheduler");

            _directory = directory;
            _channel = channel;
            _fiber = fiber;
            _hashes = new Dictionary<string, Guid>();
            _scheduler = scheduler;
            _checkInterval = checkInterval;

            _scheduledAction = scheduler.Schedule(3.Seconds(), _fiber, HashFileSystem);

            var myChannel = new ChannelAdapter();

            _connection = myChannel.Connect(connectionConfigurator =>
            {
                connectionConfigurator.AddConsumerOf<FileSystemChanged>().UsingConsumer(HandleFileSystemChangedAndCreated);
                connectionConfigurator.AddConsumerOf<FileSystemCreated>().UsingConsumer(HandleFileSystemChangedAndCreated);
                connectionConfigurator.AddConsumerOf<FileSystemRenamed>().UsingConsumer(HandleFileSystemRenamed);
                connectionConfigurator.AddConsumerOf<FileSystemDeleted>().UsingConsumer(HandleFileSystemDeleted);
            });

            _fileSystemEventProducer = new FileSystemEventProducer(directory, myChannel, checkSubDirectory);
        }

        public void Start()
        {
            // file system watcher will fail if directory isn't there, ensure it is
            if (!Directory.Exists(_baseDirectory))
                Directory.CreateDirectory(_baseDirectory);

            _scheduler = new TimerScheduler(new PoolFiber());
            _channel = new ChannelAdapter();

            _producer = new PollingFileSystemEventProducer(_baseDirectory, _channel, _scheduler, new PoolFiber(),
                                                           2.Minutes());

            _connection = _channel.Connect(config =>
                {
                    config
                        .AddConsumerOf<FileSystemEvent>()
                        .BufferFor(3.Seconds())
                        .UseScheduler(_scheduler)
                        .Distinct(fsEvent => GetChangedDirectory(fsEvent.Path))
                        .UsingConsumer(fsEvents =>
                            {
                                fsEvents.Keys.Distinct().Each(key =>
                                    {
                                        if (key == _baseDirectory)
                                            return;

                                        _serviceChannel.Send(new ServiceFolderChanged(key));
                                    });
                            })
                        .HandleOnFiber(_fiber);
                });
        }

        public void A_file_is_created()
        {
            _baseDirectory = AppDomain.CurrentDomain.BaseDirectory;

            _filename = "test2.dat";
            _path = Path.Combine(_baseDirectory, _filename);

            System.IO.File.Delete(_path);

            _listener = new Future<FileCreated>();

            _channel = new ChannelAdapter();
            FiberFactory fiberFactory = () => new SynchronousFiber();
            _scheduler = new TimerScheduler(fiberFactory());
            _producer = new PollingFileSystemEventProducer(_baseDirectory, _channel, _scheduler, fiberFactory(),
                                                           20.Seconds());

            Thread.Sleep(5.Seconds());

            using (_channel.Connect(x => x.AddConsumerOf<FileCreated>().UsingConsumer(m => _listener.Complete(m))))
            {
                System.IO.File.Create(_path);

                _listener.WaitUntilCompleted(25.Seconds());
            }

            _producer.Dispose();
        }

        public static HostHost GetCurrentShelfHost(ChannelAdapter myChannel)
        {
            var pipeName = GetThisShelfPipeName();

            var address = GetBaseAddress(pipeName);
            return new HostHost(myChannel, address, "shelf");
        }

        public void Start()
        {
            // file system watcher will fail if directory isn't there, ensure it is
            if (!System.IO.Directory.Exists(_baseDir))
                System.IO.Directory.CreateDirectory(_baseDir);

            _channel = new ChannelAdapter();
            FiberFactory fiberFactory = () => new SynchronousFiber();
            _scheduler = new TimerScheduler(fiberFactory());
            _producer = new PollingFileSystemEventProducer(_baseDir, _channel, _scheduler, fiberFactory(),
                                                           2.Minutes());

            _channel.Connect(config => config
                                           .AddConsumerOf<FileSystemEvent>()
                                           .BufferFor(3.Seconds())
                                           .Distinct(fsEvent => GetChangedDirectory(fsEvent.Path))
                                           .UsingConsumer(fsEvents => fsEvents.Keys.ToList().ForEach(key =>
                                               {
                                                   if (key == _baseDir)
                                                       return;

                                                   _hostChannel.Send(new FileSystemChange
                                                       {
                                                           ShelfName = key
                                                       });
                                               })));
        }

		public void Sending_a_message_to_an_nhibernate_instance_channel()
		{
			_newValue = new Random().Next(1, 500000)/100m;

			using (ISession session = SessionFactory.OpenSession())
			using (ITransaction transaction = session.BeginTransaction())
			{
				session.CreateQuery("Delete TestInstance").ExecuteUpdate();

				session.Save(new TestInstance(27));

				transaction.Commit();
			}

			var input = new ChannelAdapter();
			using (input.Connect(x =>
				{
					x.AddConsumerOf<UpdateValue>()
						.UsingInstance()
						.Of<TestInstance>()
						.HandleOnCallingThread()
						.DistributedBy(msg => msg.Id)
						.PersistUsingNHibernate()
						.UsingSessionProvider(m => SessionFactory.OpenSession())
						.OnChannel(m => m.UpdateValueChannel)
						.CreateNewInstanceBy(m => new TestInstance(m.Id));
				}))
			{
				//
				input.Send(new UpdateValue(27, _newValue));
			}
		}

        public static HostHost GetCurrentServiceHost(ChannelAdapter myChannel, string serviceName)
        {
            var pipeName = GetThisShelfPipeName();

            var address = GetBaseAddress(pipeName);
            return new HostHost(myChannel, address, serviceName);
        }

        public void A_request_is_sent_via_wcf()
        {
            _pipeUri = new Uri("net.pipe://localhost/pipe");

            _response = new Future<Response<TestMessage>>();

            _client = new ChannelAdapter();
            _clientConnection = _client.Connect(x =>
                {
                    x.SendToWcfChannel(_pipeUri, _pipeName)
                        .HandleOnCallingThread();

                    x.AddConsumerOf<Response<TestMessage>>()
                        .UsingConsumer(_response.Complete);
                });

            _server = new ChannelAdapter();
            _serverConnection = _server.Connect(x =>
                {
                    x.ReceiveFromWcfChannel(_pipeUri, _pipeName);

                    x.AddConsumerOf<Request<TestMessage>>()
                        .UsingConsumer(request => request.Respond(request.Body));
                });
        }

		public void Sending_a_bid_request_should_get_a_response()
		{
			var response = new FutureChannel<Response<Status>>();

			UntypedChannel responseChannel = new ChannelAdapter();
			responseChannel.Connect(x => x.AddChannel(response));

			Auction.Request(new Ask(Id), responseChannel);

			response.WaitUntilCompleted(2.Seconds()).ShouldBeTrue("Timeout waiting for response");

			response.Value.Body.AuctionId.ShouldEqual(Id);


			// ThreadUtil.Sleep(2.Seconds());
			// go ahead and buy something

			var purchased = new FutureChannel<Response<Purchased>>();

			responseChannel.Connect(x => x.AddChannel(purchased));

			Auction.Request(new Buy
				{
					Quantity = 15,
					Token = response.Value.Body.Token
				}, responseChannel);

			purchased.WaitUntilCompleted(2.Seconds()).ShouldBeTrue("Timeout waiting for purchase");

			purchased.Value.Body.Quantity.ShouldEqual(15);
			purchased.Value.Body.Price.ShouldEqual(response.Value.Body.CurrentBid);
		}

		public void Should_properly_arrive_at_the_destination()
		{
			var serviceUri = new Uri("net.pipe://localhost/pipe");
			string pipeName = "test";

			var future = new Future<TestMessage>();
			var message = new TestMessage
				{
					Id = Guid.NewGuid(),
					Name = "Alpha",
				};

			UntypedChannel adapter = new ChannelAdapter();
			using (var remote = new WcfChannelHost(new SynchronousFiber(), adapter, serviceUri, pipeName))
			{
				using (adapter.Connect(x =>
					{
						x.AddConsumerOf<TestMessage>()
							.UsingConsumer(m => future.Complete(m));
					}))
				{
					var client = new WcfChannelProxy(new SynchronousFiber(), serviceUri, pipeName);

					client.Send(message);

					future.WaitUntilCompleted(2.Seconds()).ShouldBeTrue();
				}
			}

			future.Value.ShouldNotBeNull();
			future.Value.ShouldEqual(message);
			future.Value.ShouldNotBeTheSameAs(message);
		}

        public void EstablishContext()
        {
            using (var startEvent = new ManualResetEvent(false))
            {
                _srv = new TestService();

                _channelAdaptor = new ChannelAdapter();
                _hostChannel = WellknownAddresses.GetServiceCoordinatorHost(_channelAdaptor);

                using (_channelAdaptor.Connect(config => config.AddConsumerOf<ServiceStarted>().UsingConsumer(msg => startEvent.Set())))
                {

                    ServiceConfigurator<TestService> c = new ServiceConfigurator<TestService>();
                    c.WhenStarted(s => s.Start());
                    c.WhenStopped(s => s.Stop());
                    c.WhenPaused(s => { _wasPaused = true; });
                    c.WhenContinued(s => { _wasContinued = true; });
                    c.HowToBuildService(name => _srv);

                    _serviceController = c.Create(WellknownAddresses.GetServiceCoordinatorProxy());
                    _serviceController.Start();

                    startEvent.WaitOne(5.Seconds());

                    _serviceController.State.ShouldEqual(ServiceState.Started);
                }
            }
        }

        public void Sending_a_message_to_an_nhibernate_backed_state_machine()
        {
            TraceLogProvider.Configure(LogLevel.Debug);

            _newValue = new Random().Next(1, 500000)/100m;

            using (ISession session = SessionFactory.OpenSession())
            using (ITransaction transaction = session.BeginTransaction())
            {
                session.CreateQuery("Delete TestStateMachineInstance").ExecuteUpdate();

                transaction.Commit();
            }

            var input = new ChannelAdapter();
            using (input.Connect(x =>
                {
                    x.AddConsumersFor<TestStateMachineInstance>()
                        .BindUsing<TestStateMachineInstanceBinding, int>()
                        .ExecuteOnProducerThread()
                        .CreateNewInstanceUsing(id => new TestStateMachineInstance(id))
                        .PersistUsingNHibernate()
                        .UseSessionProvider(() => SessionFactory.OpenSession());
                }))
            {
                input.Flatten().Select(c => c.GetType()).ShouldEqual(new[]
                    {
                        typeof(ChannelAdapter),
                        typeof(BroadcastChannel),
                        typeof(TypedChannelAdapter<CreateOrder>),
                        typeof(InstanceChannel<CreateOrder>),
                        typeof(TypedChannelAdapter<UpdateOrder>),
                        typeof(InstanceChannel<UpdateOrder>),
                        typeof(TypedChannelAdapter<CompleteOrder>),
                        typeof(InstanceChannel<CompleteOrder>),
                    });

                var future = new Future<int>();
                TestStateMachineInstance.CompletedLatch = new CountdownLatch(1, future.Complete);
                //
                input.Send(new CreateOrder
                    {
                        Id = 27
                    });

                input.Send(new UpdateOrder
                    {
                        Id = 27,
                        Value = _newValue,
                    });

                input.Send(new CompleteOrder
                    {
                        Id = 27,
                    });

                future.WaitUntilCompleted(5.Seconds()).ShouldBeTrue();
            }
        }

		public void Run()
		{
			Stopwatch timer = Stopwatch.StartNew();

			const int channelCount = 10000;
			const int seedCount = 500;

			var channels = new UntypedChannel[channelCount];
			var connections = new ChannelConnection[channelCount];

			var complete = new Future<int>();

			var latch = new CountdownLatch(channelCount*seedCount, complete.Complete);

			for (int i = 0; i < channelCount; i++)
			{
				int channelNumber = i;
				channels[i] = new ChannelAdapter();
				connections[i] = channels[i].Connect(x =>
					{
						x.AddConsumerOf<AMessage>()
							.UsingConsumer(message =>
								{
									if (channelNumber < channels.Length - 1)
										channels[channelNumber + 1].Send(message);

									latch.CountDown();
								});
					});
			}

			var body = new AMessage();

			for (int i = 0; i < seedCount; i++)
			{
				channels[i].Send(body);

				for (int j = 0; j < i; j++)
					latch.CountDown();
			}

			bool completed = complete.WaitUntilCompleted(2.Minutes());

			timer.Stop();

			connections.Each(x => x.Dispose());
			

			if (!completed)
			{
				Console.WriteLine("Process did not complete");
				return;
			}

			Console.WriteLine("Processed {0} messages in with {1} channels in {2}ms", seedCount, channelCount,
			                  timer.ElapsedMilliseconds);

			Console.WriteLine("That's {0} messages per second!", ((long)seedCount*channelCount*1000)/timer.ElapsedMilliseconds);
		}

		public void Should_receive_the_request_message_type()
		{
			var responseChannel = new ChannelAdapter();

			_channel.Request<Simple>(responseChannel);

			_received.IsCompleted.ShouldBeTrue("Message was not received");
		}

		public void Should_properly_wrap_the_channel_as_synchronized()
		{
			Assert.IsNull(SynchronizationContext.Current);

			var fiber = new PoolFiber();

			var input = new ChannelAdapter();

			var context = new TestSynchronizationContext();

			var future = new Future<TestMessage>();

			SynchronizationContext.SetSynchronizationContext(context);

			Assert.IsNotNull(SynchronizationContext.Current);

			using (input.Connect(x =>
				{
					x.AddConsumerOf<TestMessage>()
						.OnCurrentSynchronizationContext()
						.UsingConsumer(message =>
							{
								Trace.WriteLine("Received on Thread: " + Thread.CurrentThread.ManagedThreadId);

								Assert.IsNotNull(SynchronizationContext.Current);
								Assert.AreEqual(context, SynchronizationContext.Current);

								future.Complete(message);
							});
				}))
			{
				Trace.WriteLine("Subscribed on Thread: " + Thread.CurrentThread.ManagedThreadId);

				SynchronizationContext.SetSynchronizationContext(null);

				input.Flatten().Select(c => c.GetType()).ShouldEqual(new[]
					{
						typeof(ChannelAdapter),
						typeof(BroadcastChannel),
						typeof(TypedChannelAdapter<TestMessage>),
						typeof(SynchronizedChannel<TestMessage>),
						typeof(ConsumerChannel<TestMessage>),
					});

				fiber.Add(() =>
					{
						Trace.WriteLine("Thread: " + Thread.CurrentThread.ManagedThreadId);
						Assert.IsNull(SynchronizationContext.Current);

						input.Send(new TestMessage());
					});

				Assert.IsNull(SynchronizationContext.Current);

				future.WaitUntilCompleted(2.Seconds()).ShouldBeTrue();
			}
		}

		void CreateChannelNetwork()
		{
			_connectionChannel = new ChannelAdapter<ConnectionContext>();
			_connectionChannelConnection = _connectionChannel.Connect(x =>
				{
					var channelProvider = new HttpConnectionChannelProvider(_connectionHandlers);
					var threadPoolChannel = new ThreadPoolChannel<ConnectionContext>(channelProvider, _concurrentConnectionLimit);

					x.AddChannel(threadPoolChannel);
				});
		}

		public void Should_allow_the_selective_consumer_channel_type()
		{
			var input = new ChannelAdapter();

			using (input.Connect(x =>
				{
					x.AddConsumerOf<TestMessage>()
						.UsingSelectiveConsumer(message => m => { });
				}))
			{
			}
		}

		public void Sending_a_message_to_an_nhibernate_backed_state_machine()
		{
			_newValue = new Random().Next(1, 500000)/100m;

			using (ISession session = SessionFactory.OpenSession())
			using (ITransaction transaction = session.BeginTransaction())
			{
				session.CreateQuery("Delete TestStateMachineInstance").ExecuteUpdate();

				transaction.Commit();
			}

			var input = new ChannelAdapter();
			using (input.Connect(x =>
				{
					x.AddConsumersFor<TestStateMachineInstance>()
						.BindUsing<TestStateMachineInstanceBinding, int>()
						.HandleOnCallingThread()
						.CreateNewInstanceBy(id => new TestStateMachineInstance(id))
						.PersistUsingNHibernate()
						.UseSessionProvider(() => SessionFactory.OpenSession());
				}))
			{
				_networkTypes = input.Flatten().Select(c => c.GetType());

				var future = new Future<int>();
				TestStateMachineInstance.CompletedLatch = new CountdownLatch(1, future.Complete);
				//
				input.Send(new UpdateOrder
					{
						Id = 47
					});

				input.Send(new CreateOrder
					{
						Id = 27
					});

				input.Send(new UpdateOrder
					{
						Id = 27,
						Value = _newValue,
					});

				input.Send(new CompleteOrder
					{
						Id = 27,
					});

				future.WaitUntilCompleted(5.Seconds()).ShouldBeTrue();
			}
		}