public void Audit(SendOptions sendOptions, TransportMessage transportMessage)
        {
            // Revert the original body if needed (if any mutators were applied, forward the original body as received)
            transportMessage.RevertToOriginalBodyIfNeeded();

            // Create a new transport message which will contain the appropriate headers
            var messageToForward = new TransportMessage(transportMessage.Id, transportMessage.Headers)
            {
                Body = transportMessage.Body,
                Recoverable = transportMessage.Recoverable,
                TimeToBeReceived = sendOptions.TimeToBeReceived.HasValue ? sendOptions.TimeToBeReceived.Value : transportMessage.TimeToBeReceived
            };

            messageToForward.Headers[Headers.ProcessingMachine] = RuntimeEnvironment.MachineName;
            messageToForward.Headers[Headers.ProcessingEndpoint] = EndpointName;

            if (transportMessage.ReplyToAddress != null)
            {
                messageToForward.Headers[Headers.OriginatingAddress] = transportMessage.ReplyToAddress.ToString();
            }

            // Send the newly created transport message to the queue
            MessageSender.Send(messageToForward, new SendOptions(sendOptions.Destination)
            {
                ReplyToAddress = Configure.PublicReturnAddress
            });
        }

        void SendMessage(TransportMessage message, SendOptions sendOptions, IModel channel)
        {

            var destination = sendOptions.Destination;

            string callbackAddress;

            if (sendOptions.GetType().FullName.EndsWith("ReplyOptions") &&
                message.Headers.TryGetValue(CallbackHeaderKey, out callbackAddress))
            {
                destination = Address.Parse(callbackAddress);
            }

            //set our callback address
            if (!string.IsNullOrEmpty(CallbackQueue))
            {
                message.Headers[CallbackHeaderKey] = CallbackQueue;     
            }
           
            var properties = channel.CreateBasicProperties();

            RabbitMqTransportMessageExtensions.FillRabbitMqProperties(message, sendOptions, properties);

            RoutingTopology.Send(channel, destination, message, properties);
        }

        public static DeliveryOptions ToDeliveryOptions(this Dictionary<string, string> options)
        {
            var operation = options["Operation"].ToLower();

            switch (operation)
            {
                case "publish":
                    return new PublishOptions(Type.GetType(options["EventType"]))
                    {
                        ReplyToAddress = Address.Parse(options["ReplyToAddress"])
                    };

                case "send":
                case "audit":
                    var sendOptions = new SendOptions(options["Destination"]);

                    ApplySendOptionSettings(sendOptions, options);

                    return sendOptions;

                case "reply":
                    var replyOptions = new ReplyOptions(Address.Parse(options["Destination"]), options["CorrelationId"])
                    {
                        ReplyToAddress = Address.Parse(options["ReplyToAddress"])
                    };
                    ApplySendOptionSettings(replyOptions, options);

                    return replyOptions;

                default:
                    throw new Exception("Unknown operation: " + operation);
            }


        }

        public static BrokeredMessage ToBrokeredMessage(this TransportMessage message, SendOptions options, SettingsHolder settings, bool expectDelay, Configure config)
        {
            var brokeredMessage = BrokeredMessageBodyConversion.InjectBody(message.Body);

            SetHeaders(message, options, settings, config, brokeredMessage);

            var timeToSend = DelayIfNeeded(options, expectDelay);
                        
            if (timeToSend.HasValue)
                brokeredMessage.ScheduledEnqueueTimeUtc = timeToSend.Value;

            TimeSpan? timeToLive = null;
            if (message.TimeToBeReceived < TimeSpan.MaxValue)
            {
                timeToLive = message.TimeToBeReceived;
            }
            else if (options.TimeToBeReceived.HasValue && options.TimeToBeReceived < TimeSpan.MaxValue)
            {
                timeToLive = options.TimeToBeReceived.Value;
            }

            if (timeToLive.HasValue)
            {
                if (timeToLive.Value <= TimeSpan.Zero) return null;

                brokeredMessage.TimeToLive = timeToLive.Value;
            }
            GuardMessageSize(brokeredMessage);

            return brokeredMessage;
        }

        public void Send(TransportMessage message, SendOptions sendOptions)
        {
            Address destination = null;
            try
            {
                destination = DetermineDestination(sendOptions);
            
                var connectionInfo = connectionStringProvider.GetForDestination(sendOptions.Destination);
                var queue = new TableBasedQueue(destination, connectionInfo.Schema);
                if (sendOptions.EnlistInReceiveTransaction)
                {
                    SqlTransaction currentTransaction;
                    if (connectionStore.TryGetTransaction(connectionInfo.ConnectionString, out currentTransaction))
                    {
                        queue.Send(message, sendOptions, currentTransaction.Connection, currentTransaction);
                    }
                    else
                    {
                        SqlConnection currentConnection;
                        if (connectionStore.TryGetConnection(connectionInfo.ConnectionString, out currentConnection))
                        {
                            queue.Send(message, sendOptions, currentConnection);
                        }
                        else
                        {
                            using (var connection = sqlConnectionFactory.OpenNewConnection(connectionInfo.ConnectionString))
                            {
                                queue.Send(message, sendOptions, connection);
                            }
                        }
                    }
                }
                else 
                {
                    // Suppress so that even if DTC is on, we won't escalate
                    using (var tx = new TransactionScope(TransactionScopeOption.Suppress))
                    {
                        using (var connection = sqlConnectionFactory.OpenNewConnection(connectionInfo.ConnectionString))
                        {
                            queue.Send(message, sendOptions, connection);
                        }

                        tx.Complete();
                    }
                }
            }
            catch (SqlException ex)
            {
                if (ex.Number == 208)
                {
                    ThrowQueueNotFoundException(destination, ex);
                }

                ThrowFailedToSendException(destination, ex);
            }
            catch (Exception ex)
            {
                ThrowFailedToSendException(destination, ex);
            }
        }

        public void RequiresImmediateDispatch_Should_Return_True_When_Immediate_Dispatch_Requested()
        {
            var options = new SendOptions();
            options.RequireImmediateDispatch();

            Assert.IsTrue(options.RequiredImmediateDispatch());
        }

        public void Defer(TransportMessage message, SendOptions sendOptions)
        {
            message.Headers[TimeoutManagerHeaders.RouteExpiredTimeoutTo] = sendOptions.Destination.ToString();

            DateTime deliverAt;

            if (sendOptions.DelayDeliveryWith.HasValue)
            {
                deliverAt = DateTime.UtcNow + sendOptions.DelayDeliveryWith.Value;
            }
            else
            {
                if (sendOptions.DeliverAt.HasValue)
                {
                    deliverAt = sendOptions.DeliverAt.Value;    
                }
                else
                {
                    throw new ArgumentException("A delivery time needs to be specified for Deferred messages");
                }
                
            }

            message.Headers[TimeoutManagerHeaders.Expire] = DateTimeExtensions.ToWireFormattedString(deliverAt);
            
            try
            {
                MessageSender.Send(message, new SendOptions(TimeoutManagerAddress));
            }
            catch (Exception ex)
            {
                Log.Error("There was a problem deferring the message. Make sure that DisableTimeoutManager was not called for your endpoint.", ex);
                throw;
            }
        }

        public async Task<ActionResult> SendTestMessage(SendTestMessage sendTestMessage)
        {
            try
            {
                var sendOptions = new SendOptions
                {
                    ContentType = sendTestMessage.ContentType,
                    Importance = sendTestMessage.Importance
                };

                var message = new TestMessage
                {
                    Text = sendTestMessage.MessageText
                };

                var bus = await _busManager.GetBus();
                var sentMessage = await bus.Send(message, sendOptions);
                return View("Index", new SendTestMessage
                {
                    MessageSent = true,
                    SentMessageId = sentMessage.MessageId
                });
            }
            catch (Exception ex)
            {
                sendTestMessage.ErrorsOccurred = true;
                sendTestMessage.ErrorMessage = ex.ToString();
            }
            return View("Index", sendTestMessage);
        }

 async Task RequestImmediateDispatch(IPipelineContext context)
 {
     #region RequestImmediateDispatch
     SendOptions options = new SendOptions();
     options.RequireImmediateDispatch();
     await context.Send(new MyMessage(), options);
     #endregion
 }

 public async Task<ActionResult> SendEnumMessage()
 {
     EnumMessage message = new EnumMessage();
     SendOptions sendOptions = new SendOptions();
     sendOptions.SetDestination("Samples.Callbacks.Receiver");
     Task<Status> statusTask = MvcApplication.BusSession.Request<Status>(message, sendOptions);
     return View("SendEnumMessage", await statusTask);
 }

        public void SendOptions_GetCorrelationId_Should_Return_Null_When_No_CorrelationId_Configured()
        {
            var options = new SendOptions();

            var correlationId = options.GetCorrelationId();

            Assert.IsNull(correlationId);
        }

 async void Simple(IEndpointInstance endpoint, SendOptions sendOptions, ILog log)
 {
     #region EnumCallback
     Message message = new Message();
     Status response = await endpoint.Request<Status>(message, sendOptions);
     log.Info("Callback received with response:" + response);
     #endregion
 }

        public void GetDeliveryDate_Should_Return_The_Configured_Delivery_Date()
        {
            var options = new SendOptions();
            DateTimeOffset deliveryDate = new DateTime(2012, 12, 12, 12, 12, 12);
            options.DoNotDeliverBefore(deliveryDate);

            Assert.AreEqual(deliveryDate, options.GetDeliveryDate());
        }

        public void IsRoutingToThisEndpoint_Should_Return_True_When_Routed_To_This_Endpoint()
        {
            var options = new SendOptions();

            options.RouteToThisEndpoint();

            Assert.IsTrue(options.IsRoutingToThisEndpoint());
        }

 public async Task<ActionResult> SendEnumMessage()
 {
     var message = new EnumMessage();
     var sendOptions = new SendOptions();
     sendOptions.SetDestination("Samples.Callbacks.Receiver");
     Task<Status> statusTask = MvcApplication.EndpointInstance.Request<Status>(message, sendOptions);
     return View("SendEnumMessage", await statusTask.ConfigureAwait(false));
 }

 public async Task<ActionResult> SendIntMessage()
 {
     IntMessage message = new IntMessage();
     SendOptions sendOptions = new SendOptions();
     sendOptions.SetDestination("Samples.Callbacks.Receiver");
     Task<int> statusTask = MvcApplication.Endpoint.Request<int>(message, sendOptions);
     return View("SendIntMessage", await statusTask);
 }

 public async Task<ActionResult> SendObjectMessage()
 {
     ObjectMessage message = new ObjectMessage();
     SendOptions sendOptions = new SendOptions();
     sendOptions.SetDestination("Samples.Callbacks.Receiver");
     Task<ObjectResponseMessage> responseTask = MvcApplication.Endpoint.Request<ObjectResponseMessage>(message, sendOptions);
     return View("SendObjectMessage", await responseTask);
 }

        public void IsRoutingToThisInstance_Should_Return_True_When_Routed_To_This_Instance()
        {
            var options = new SendOptions();

            options.RouteToThisInstance();

            Assert.IsTrue(options.IsRoutingToThisInstance());
        }

        public void GetDeliveryDelay_Should_Return_The_Configured_Delay_TimeSpan()
        {
            var options = new SendOptions();
            var delay = TimeSpan.FromMinutes(42);
            options.DelayDeliveryWith(delay);

            Assert.AreEqual(delay, options.GetDeliveryDelay());
        }

        public void SendOptions_GetDestination_Should_Return_Null_When_No_Destination_Configured()
        {
          var options = new SendOptions();

          var destination = options.GetDestination();

          Assert.IsNull(destination);
        }