public void create_from_graph_and_run_through_the_channel()
        {
            using (var graph = new ChannelGraph())
            {
                var node = graph.ChannelFor<BusSettings>(x => x.Outbound);

                node.Uri = new Uri("memory://foo");

                var transport = new InMemoryTransport();
                transport.OpenChannels(graph);
                node.Channel.ShouldNotBeNull();

                var envelope = new Envelope();
                envelope.CorrelationId = Guid.NewGuid().ToString();
                envelope.Headers["Foo"] = "Bar";
                envelope.Data = new byte[] {1, 2, 3, 4, 5};

                var receiver = new RecordingReceiver();

                node.StartReceiving(receiver);

                node.Channel.Send(envelope.Data, envelope.Headers);

                Wait.Until(() => receiver.Received.Any(), timeoutInMilliseconds: 2000);

                var received = receiver.Received.Single();

                received.CorrelationId.ShouldEqual(envelope.CorrelationId);
                received.ContentType.ShouldEqual(envelope.ContentType);
                received.Data.ShouldEqual(envelope.Data);

            }
        }

        public void SetUp()
        {
            theContext = new TestContinuationContext();
            theEnvelope = ObjectMother.Envelope();

            new NoSubscriberHandler().Execute(theEnvelope, theContext);
        }

        public void recovers_delayed_messages_when_started()
        {
            using (var queues = new PersistentQueues(new RecordingLogger(), new DelayedMessageCache<MessageId>(), new LightningQueueSettings()))
            {
                queues.ClearAll();
                queues.Start(new []{ new LightningUri("lq.tcp://localhost:2425/the_queue") });

                var envelope = new Envelope();
                envelope.Data = new byte[0];
                envelope.ExecutionTime = DateTime.UtcNow;
                var delayedMessage = new MessagePayload
                {
                    Data = envelope.Data,
                    Headers = envelope.Headers.ToNameValues()
                };

                using (var scope = new TransactionScope())
                {
                    queues.ManagerFor(2425, true)
                        .EnqueueDirectlyTo(LightningQueuesTransport.DelayedQueueName, null, delayedMessage);
                    scope.Complete();
                }
            }

            var cache = new DelayedMessageCache<MessageId>();
            using (var queues = new PersistentQueues(new RecordingLogger(), cache, new LightningQueueSettings()))
            {
                queues.Start(new []{ new LightningUri("lq.tcp://localhost:2425/the_queue") });

                cache.AllMessagesBefore(DateTime.UtcNow.AddSeconds(1)).ShouldNotBeEmpty();
            }
        }

        // virtual for testing
        public string Send(Envelope envelope)
        {
            envelope.Headers[Envelope.MessageTypeKey] = envelope.Message.GetType().FullName;

            _modifiers.Each(x => x.Modify(envelope));

            var channels = _router.FindDestinationChannels(envelope).ToArray();

            if (!channels.Any())
            {
                throw new Exception("No channels match this message ({0})".ToFormat(envelope));
            }

            channels.Each(x => {
                try
                {
                    sendToChannel(envelope, x);
                }
                catch (Exception e)
                {
                    _logger.Error(envelope.CorrelationId, "Failed trying to send message {0} to channel {1}".ToFormat(envelope, x.Uri), e);
                    throw;
                }
            });

            return envelope.CorrelationId;
        }

        public void Execute(Envelope envelope, ContinuationContext context)
        {
            context.Outgoing.SendFailureAcknowledgement(envelope, "Moved message {0} to the Error Queue.\n{1}".ToFormat(envelope.CorrelationId, _exception));

            var report = new ErrorReport(envelope, _exception);
            envelope.Callback.MoveToErrors(report);
        }

        public void Receive(byte[] data, IHeaders headers, IMessageCallback callback)
        {
            var envelope = new Envelope(data, headers, callback);
            Received.Add(envelope);

            envelope.Callback.MarkSuccessful();
        }

        public static IEnumerable<EnvelopeToken> DequeueDelayedEnvelopes(DateTime currentTime)
        {
            var delayed = _delayedLock.Read(() => {
                return _delayed.Where(x => new Envelope(x.Headers).ExecutionTime.Value <= currentTime).ToArray();
            });

            var list = new List<EnvelopeToken>();

            foreach (EnvelopeToken token in delayed)
            {
                _delayedLock.Write(() => {
                    try
                    {
                        _delayed.Remove(token);

                        var envelope = new Envelope(token.Headers);
                        _queues[envelope.ReceivedAt].Enqueue(token);

                        list.Add(token);
                    }
                    catch (Exception ex)
                    {
                        Debug.WriteLine(ex);
                    }
                });

            }

            return list;
        }

 public MonitoringControlHandler(ILogger logger, Envelope envelope, ChannelGraph graph, IPersistentTaskController controller)
 {
     _logger = logger;
     _envelope = envelope;
     _graph = graph;
     _controller = controller;
 }

        public void SetUp()
        {
            _envelope = ObjectMother.Envelope();
            _message = new object();
            _context = new TestContinuationContext();

            new RespondWithMessageContinuation(_message).Execute(_envelope, _context);
        }

 public void SetUp()
 {
     theMessage = new Message1();
     theOriginalEnvelope = new Envelope
     {
         ReplyUri = "lq://foo".ToUri()
     };
 }

        public void attempts()
        {
            var envelope = new Envelope();
            envelope.Attempts.ShouldEqual(0);

            envelope.Attempts++;

            envelope.Attempts.ShouldEqual(1);
        }

 public ExceptionHandlerBehavior(IActionBehavior behavior, HandlerChain chain, Envelope envelope, IInvocationContext context, ILogger logger, IFubuRequest request)
 {
     _behavior = behavior;
     _chain = chain;
     _envelope = envelope;
     _context = context;
     _logger = logger;
     _request = request;
 }

        public virtual void SendSubscriptionChangedToPeer(TransportNode node)
        {
            var envelope = new Envelope
            {
                Message = new SubscriptionsChanged(),
                Destination = node.Addresses.FirstOrDefault()
            };

            _sender.Send(envelope);
        }

        public void matches_negative_with_no_execution_time_header()
        {
            var systemTime = SystemTime.Default();
            var envelope = new Envelope();

            var handler = new DelayedEnvelopeHandler(systemTime);

            envelope.IsDelayed(systemTime.UtcNow()).ShouldBeFalse();
            handler.Matches(envelope).ShouldBeFalse();
        }

        public void SetUp()
        {
            theException = new EnvelopeDeserializationException("foo");
            theContext = new TestContinuationContext();

            theEnvelope = ObjectMother.EnvelopeWithSerializationError();

            new DeserializationFailureContinuation(theException)
                .Execute(theEnvelope, theContext);
        }

        public object Message; // leave it like this please

        #endregion Fields

        #region Constructors

        public ErrorReport(Envelope envelope, Exception ex)
        {
            Headers = envelope.Headers.ToNameValues();
            ExceptionText = ex.ToString();
            ExceptionMessage = ex.Message;
            ExceptionType = ex.GetType().FullName;
            Explanation = ExceptionDetected;
            RawData = envelope.Data;
            Message = envelope.Message;
        }

        public void default_values_for_original_and_parent_id_are_null()
        {
            var parent = new Envelope
            {
                CorrelationId = Guid.NewGuid().ToString()
            };

            parent.OriginalId.ShouldBeNull();
            parent.ParentId.ShouldBeNull();
        }

        public void content_type()
        {
            var envelope = new Envelope();
            envelope.ContentType.ShouldEqual(null);

            envelope.ContentType = "text/xml";

            envelope.Headers[Envelope.ContentTypeKey].ShouldEqual("text/xml");
            envelope.ContentType.ShouldEqual("text/xml");
        }

        public void SetUp()
        {
            theEnvelope = ObjectMother.Envelope();
            theException = new NotImplementedException();

            theLogger = MockRepository.GenerateMock<ILogger>();

            theContext = new TestContinuationContext();

            new MoveToErrorQueue(theException).Execute(theEnvelope, theContext);
        }

        public override IHeaders Send(Envelope envelope, IEnvelopeSerializer serializer, Uri replyUri = null)
        {
            if (replyUri != null)
            {
                envelope.ReplyUri = replyUri;
            }

            LastEnvelope = envelope;

            return envelope.Headers;
        }