public void Should_support_delivery()
        {
            Vane<Message> vane = VaneFactory.New<Message>(x =>
                {
                    x.Consumer(() => new TestConsumer(), xc =>
                        {
                            xc.Consume<A>(c => c.Consume);
                            xc.Consume<B>(c => c.Consume);
                        });
                });

            var a = new A {Value = "Hello"};
            Payload<Message> payload = new MessagePayload<A>(a);

            vane.Execute(payload, CancellationToken.None);

            var b = new B {Value = "World"};
            payload = new MessagePayload<B>(b);

            vane.Execute(payload, CancellationToken.None);

            var visitor = new StringVaneVisitor();
            visitor.Visit(vane);
            Console.WriteLine(visitor.ToString());
        }

 public void ReceiveAcknowledgementFor(MessagePayload messagePayload)
 {
     ReceiveMessage(
         new MessagePayload()
             .AsAcknowlegement(messagePayload.GetPersistenceId())
             .SetToChannel(messagePayload.GetFromAddress().Channel));
 }

        public override void InputMessage(MessagePayload toInput)
        {
            if(!toInput.HasBody()) return;
            if(toInput.GetToAddress() != address) return;

            OnMessageProcessed(toInput);
        }

        void PersistMessage(MessagePayload toInput)
        {
            Logger.Debug("Caching message payload {0}", toInput.Id);

            toInput.SetPersistenceId(messageCache.Address, messageCache.UseType);
            messageCache.AddMessageAndIncrementSequence(toInput);
        }

 void PerformLogging(MessagePayload toInput)
 {
     Logger.Debug(
         "Attaching session {0} for message: {0}", 
         cache.GetCurrentSessionFor(GetServer()).Id, 
         toInput.Id);
 }

        public void HandleMessage(MessagePayload toHandle, List<MessagePayload> outgoingMessages)
        {
            if (!toHandle.IsLongPollRequest())
                return;

            outgoingMessages.AddRange(outgoingQueue.DequeueAll(toHandle.GetLongPollRequestServerPath()));
        }

        public override void InputMessage(MessagePayload toInput)
        {
            Logger.Debug("Recording message persistence source on payload {0}", toInput.Id);

            toInput.SetSourcePersistenceId(toInput.GetPersistenceId());
            OnMessageProcessed(toInput);
        }

 static void LogWithAddress(MessagePayload toHandle)
 {
     Logger.Debug("Handling sent message {0} for {1} from {2}",
         toHandle.Id,
         toHandle.GetToAddress().Channel,
         toHandle.GetToAddress().Server.Address);
 }

        MessagePayload CreateLongPollPayload(MessageServer toListenFor)
        {
            var payload = new MessagePayload();
            payload.SetLongPollRequest(toListenFor);

            return payload;
        }

 void CacheMessage(MessagePayload toInput)
 {
     if(messageCache.ContainsMessage(toInput))
         messageCache.UpdateMessage(toInput);
     else 
         messageCache.AddMessage(toInput);
 }

        public override void InputMessage(MessagePayload toInput)
        {
            Logger.Debug("Recording the last time sent on message payload {0}", toInput.Id);

            SetTimeOnMessage(toInput);
            OnMessageProcessed(toInput);
        }

        public override void InputMessage(MessagePayload toInput)
        {
            if (messageCache.ResetOn < toInput.GetSequenceOriginSetOn())
                messageCache.Reset(toInput.GetFirstSequence(), toInput.GetSequenceOriginSetOn());

            OnMessageProcessed(toInput);
        }

        public override void InputMessage(MessagePayload toInput)
        {
            SetCurentSenderAddress(toInput);
            SetCurrentReceiverAddress(toInput);
            SetCurrentCorrelationId(toInput);

            OnMessageProcessed(toInput);
        }

        public override void InputMessage(MessagePayload toInput)
        {
            Logger.Debug("Addressing message payload {0} to {1}", toInput.Id, toAddress);

            toInput.SetFromAddress(fromAddress);
            toInput.SetToAddress(toAddress);
            OnMessageProcessed(toInput);
        }

        public void Enqueue(MessagePayload toEnqueue)
        {
            Contract.Requires(toEnqueue != null);

            CreateQueueIfNonExistant(toEnqueue.GetToAddress().Server);

            queues[toEnqueue.GetToAddress().Server].Enqueue(toEnqueue);
        }

        int GetDelay(MessagePayload toGetDelayFor)
        {
            int unlimitedDelay = GetUnlimitedDelay(toGetDelayFor);

            return unlimitedDelay < Peak 
                ? unlimitedDelay 
                : Peak;
        }

        public override void InputMessage(MessagePayload toInput)
        {
            Logger.Debug("Recording message sequence origin on payload {0}", toInput.Id);

            toInput.SetFirstSequence(messageCache.GetFirstSequenceInCache());
            toInput.SetSequenceOriginSetOn(messageCache.FirstItemCachedOn);
            OnMessageProcessed(toInput);
        }

        public override void InputMessage(MessagePayload toInput)
        {
            Logger.Debug("Addressing message payload {0} to {1}", toInput.Id,  DirectReplyContext.GetCurrentClientAddress());

            toInput.SetFromAddress(fromAddress);
            toInput.SetToAddress(DirectReplyContext.GetCurrentClientAddress());
            OnMessageProcessed(toInput);
        }

 protected static void LogMessage(MessagePayload toInput)
 {
     Logger.Debug(
         "Repeating message: {0} on {1} with sequence {2}",
         toInput.Id,
         toInput.HasHeader<FromAddressHeader>() ? toInput.GetFromAddress().Channel : "n/a",
         toInput.HasSequence() ? toInput.GetSequence().ToString() : "n/a");
 }

        public void HandleMessage(MessagePayload toHandle, List<MessagePayload> outgoingMessages)
        {
            if (toHandle.IsLongPollRequest()) 
                return;

            LogMessage(toHandle);

            this.outgoingQueue.Enqueue(toHandle);
        }