add-direct-opt

Sources/Contour/Configurator/AppConfigConfigurator.cs

@@ -240,6 +240,11 @@ public IBusConfigurator Configure(string endpointName, IBusConfigurator cfg)
                     configurator.Delayed();
                 }
 
+                if (outgoingElement.Direct)
+                {
+                    configurator.Direct();
+                }
+
                 // Connection string
                 var connectionString = endpointConfig.ConnectionString;
                 if (!string.IsNullOrEmpty(outgoingElement.ConnectionString))
@@ -342,6 +347,11 @@ public IBusConfigurator Configure(string endpointName, IBusConfigurator cfg)
                     configurator.Delayed();
                 }
 
+                if (incomingElement.Direct)
+                {
+                    configurator.Direct(incomingElement.DirectId);
+                }
+
                 // Connection string
                 var connectionString = endpointConfig.ConnectionString;
                 if (!string.IsNullOrEmpty(incomingElement.ConnectionString))

Sources/Contour/Configurator/Configuration/IIncoming.cs

@@ -25,5 +25,9 @@ public interface IIncoming : IMessage
         int? QueueMaxLengthBytes { get; }
 
         bool Delayed { get; }
+
+        bool Direct { get; }
+
+        string DirectId { get; }
     }
 }

Sources/Contour/Configurator/Configuration/IOutgoing.cs

@@ -20,6 +20,8 @@ public interface IOutgoing : IMessage
 
         bool Delayed { get; }
 
+        bool Direct { get; }
+
         TimeSpan? ConfirmTimeout { get; }
     }
 }

Sources/Contour/Configurator/IncomingElement.cs

@@ -117,6 +117,12 @@ public int? QueueMaxLengthBytes
         [ConfigurationProperty("delayed")]
         public bool Delayed => (bool)base["delayed"];
 
+        [ConfigurationProperty("direct")]
+        public bool Direct => (bool)base["direct"];
+
+        [ConfigurationProperty("directId")]
+        public string DirectId => (string)base["directId"];
+
         IQos IIncoming.Qos => this.Qos;
     }
 }

Sources/Contour/Configurator/OutgoingElement.cs

@@ -94,6 +94,9 @@ public bool? ReuseConnection
         [ConfigurationProperty("delayed", DefaultValue = false)]
         public bool Delayed => (bool)base["delayed"];
 
+        [ConfigurationProperty("direct")]
+        public bool Direct => (bool)base["direct"];
+
         /// <summary>
         /// Gets confirmation timeout on publishing.
         /// </summary>

Sources/Contour/Headers.cs

@@ -53,6 +53,10 @@ public class Headers
         /// </summary>
         public static readonly string Delay = "x-delay";
 
+        public static readonly string DirectId = "direct-id";
+
+        public static readonly string MatchHeaders = "x-match";
+
         /// <summary>
         /// Время жизни сообщений в очереди.
         /// </summary>
@@ -219,7 +223,12 @@ public static void ApplyDelay(Dictionary<string, object> headers, TimeSpan? dela
                 headers[Delay] = delay.Value.TotalMilliseconds;
             }
         }
-
+
+        public static void ApplyDirectId(Dictionary<string, object> headers, string directId)
+        {
+            headers[DirectId] = directId;
+        }
+
         /// <summary>
         /// Применяет к коллекции заголовков установку заголовка <c>Ttl</c>.
         /// </summary>

Sources/Contour/Receiving/IReceiverConfigurator.cs

@@ -92,6 +92,8 @@ public interface IReceiverConfigurator
         /// </summary>
         /// <returns>Конфигуратор получателя.</returns>
         IReceiverConfigurator Delayed();
+
+        IReceiverConfigurator Direct(string id);
     }
 
     /// <summary>

Sources/Contour/Receiving/ReceiverConfiguration.cs

@@ -205,5 +205,13 @@ public IReceiverConfigurator Delayed()
 
             return this;
         }
+
+        public IReceiverConfigurator Direct(string id)
+        {
+            this.Options.Direct = true;
+            this.Options.DirectId = id;
+
+            return this;
+        }
     }
 }

Sources/Contour/Receiving/ReceiverOptions.cs

@@ -76,6 +76,10 @@ public ReceiverOptions(BusOptions parent)
 
         public bool Delayed { get; set; }
 
+        public bool Direct { get; set; }
+
+        public string DirectId { get; set; }
+
         /// <summary>
         /// Создает новый экземпляр настроек как копию существующего.
         /// </summary>

Sources/Contour/Receiving/TypedReceiverConfigurationDecorator.cs

@@ -274,5 +274,10 @@ public IReceiverConfigurator Delayed()
 
             return this;
         }
+
+        public IReceiverConfigurator Direct(string id)
+        {
+            return this.configuration.Direct(id);
+        }
     }
 }

Sources/Contour/Sending/AbstractSender.cs

@@ -319,6 +319,11 @@ private IDictionary<string, object> ApplyOptions(PublishingOptions options)
             {
                 Headers.ApplyDelay(outputHeaders, options.Delay);
             }
+
+            if (this.Configuration.Options.Direct)
+            {
+                Headers.ApplyDirectId(outputHeaders, options.DirectId);
+            }
 
             Maybe<TimeSpan?> ttl = BusOptions.Pick(options.Ttl, this.Configuration.Options.GetTtl());
             Headers.ApplyTtl(outputHeaders, ttl);

Sources/Contour/Sending/ISenderConfigurator.cs

@@ -34,6 +34,8 @@ public interface ISenderConfigurator
         /// <returns>Sender configurator.</returns>
         ISenderConfiguration Delayed();
 
+        ISenderConfigurator Direct();
+
         /// <summary>
         /// Устанавливает псевдоним метки отправляемого сообщения.
         /// </summary>

Sources/Contour/Sending/PublishingOptions.cs

@@ -37,6 +37,8 @@ public class PublishingOptions
         /// </summary>
         public TimeSpan? Delay { get; set; }
 
+        public string DirectId { get; set; }
+
         /// <summary>
         /// Дополнительные заголовки
         /// </summary>

Sources/Contour/Sending/SenderConfiguration.cs

@@ -101,6 +101,13 @@ public ISenderConfiguration Delayed()
             return this;
         }
 
+        public ISenderConfigurator Direct()
+        {
+            this.Options.Direct = true;
+
+            return this;
+        }
+
         /// <summary>
         /// Проверяет корректность конфигурации.
         /// </summary>

Sources/Contour/Sending/SenderOptions.cs

@@ -56,6 +56,8 @@ public SenderOptions(BusOptions parent)
         /// </summary>
         public bool Delayed {  get; set; }
 
+        public bool Direct {  get; set; }
+
         /// <summary>
         /// Время ожидания подтверждения получения сообщения, null если требуется бесконечное ожидание.
         /// </summary>

Sources/Contour/Transport/RabbitMQ/BusConfigurationEx.cs

@@ -60,7 +60,9 @@ public static IBusConfigurator UseRabbitMq(this IBusConfigurator busConfigurator
                                          Headers.Ttl,
                                          Headers.QueueMaxLength,
                                          Headers.QueueMaxLengthBytes,
-                                         Headers.Delay
+                                         Headers.Delay,
+                                         Headers.DirectId,
+                                         Headers.MatchHeaders
                                      };
             var messageHeaderStorage = new MessageHeaderStorage(blockedHeaders);
             c.UseIncomingMessageHeaderStorage(messageHeaderStorage);

Sources/Contour/Transport/RabbitMQ/Internal/RabbitBusDefaults.cs

@@ -2,6 +2,8 @@
 using Contour.Receiving;
 using Contour.Sending;
 using Contour.Transport.RabbitMQ.Topology;
+using System.Collections.Generic;
+using Contour.Helpers;
 
 namespace Contour.Transport.RabbitMQ.Internal
 {
@@ -35,9 +37,19 @@ private static IRouteResolver RouteResolverBuilderImpl(IRouteResolverBuilder bui
         {
             string label = builder.Sender.Label.Name;
             ExchangeBuilder exchangeBuilder = Exchange.Named(label).Durable;
-            exchangeBuilder = builder.Sender.Options.Delayed
-                ? exchangeBuilder.DelayedFanout 
-                : exchangeBuilder.Fanout;
+
+            if (builder.Sender.Options.Delayed)
+            {
+                exchangeBuilder = builder.Sender.Options.Direct
+                    ? exchangeBuilder.DelayedHeaders
+                    : exchangeBuilder.DelayedFanout;
+            }
+            else
+            {
+                exchangeBuilder = builder.Sender.Options.Direct
+                    ? exchangeBuilder.Headers
+                    : exchangeBuilder.Fanout;
+            }
 
             Exchange exchange = builder.Topology.Declare(exchangeBuilder);
 
@@ -59,9 +71,22 @@ private static ISubscriptionEndpoint SubscriptionEndpointBuilderImpl(ISubscripti
 
             string queueName = builder.Endpoint.Address + "." + label;
 
-            var queueBuilder = Queue
-                .Named(queueName)
-                .Durable;
+            QueueBuilder queueBuilder;
+
+            if (builder.Receiver.Options.Direct)
+            {
+                queueName += "." + builder.Receiver.Options.DirectId;
+                queueBuilder = Queue
+                    .Named(queueName)
+                    .AutoDelete
+                    .Exclusive;
+            }
+            else
+            {
+                queueBuilder = Queue
+                    .Named(queueName)
+                    .Durable;
+            }
 
             var options = builder.Receiver.Options;
 
@@ -77,13 +102,33 @@ private static ISubscriptionEndpoint SubscriptionEndpointBuilderImpl(ISubscripti
 
             Queue queue = builder.Topology.Declare(queueBuilder);
             ExchangeBuilder exchangeBuilder = Exchange.Named(label).Durable;
-            exchangeBuilder = builder.Receiver.Options.Delayed
-                ? exchangeBuilder.DelayedFanout
-                : exchangeBuilder.Fanout;
+
+            if (builder.Receiver.Options.Delayed)
+            {
+                exchangeBuilder = builder.Receiver.Options.Direct
+                    ? exchangeBuilder.DelayedHeaders
+                    : exchangeBuilder.DelayedFanout;
+            }
+            else
+            {
+                exchangeBuilder = builder.Receiver.Options.Direct
+                    ? exchangeBuilder.Headers
+                    : exchangeBuilder.Fanout;
+            }
 
             Exchange exchange = builder.Topology.Declare(exchangeBuilder);
 
-            builder.Topology.Bind(exchange, queue);
+            Dictionary<string, object> arguments = null;
+            if (builder.Receiver.Options.Direct)
+            {
+                arguments = new Dictionary<string, object>()
+                {
+                    { Headers.DirectId, builder.Receiver.Options.DirectId },
+                    { Headers.MatchHeaders, "all" },
+                };
+            }
+
+            builder.Topology.Bind(exchange, queue, "", arguments);
 
             return builder.ListenTo(queue, exchange);
         }

Sources/Contour/Transport/RabbitMQ/Internal/RabbitChannel.cs

@@ -99,11 +99,11 @@ public void Accept(RabbitDelivery delivery)
         /// <param name="routingKey">
         /// The routing key.
         /// </param>
-        public void Bind(Queue queue, Exchange exchange, string routingKey)
+        public void Bind(Queue queue, Exchange exchange, string routingKey, IDictionary<string, object> arguments = null)
         {
             try
             {
-                this.SafeNativeInvoke(n => n.QueueBind(queue.Name, exchange.Name, routingKey));
+                this.SafeNativeInvoke(n => n.QueueBind(queue.Name, exchange.Name, routingKey, arguments));
             }
             catch (Exception e)
             {

Sources/Contour/Transport/RabbitMQ/Topology/ExchangeBuilder.cs

@@ -96,6 +96,15 @@ public ExchangeBuilder Topic
             }
         }
 
+        public ExchangeBuilder Headers
+        {
+            get
+            {
+                this.Instance.Type = ExchangeType.Headers;
+                return this;
+            }
+        }
+
         public ExchangeBuilder DelayedDirect
         {
             get
@@ -125,6 +134,16 @@ public ExchangeBuilder DelayedTopic
                 return this;
             }
         }
+
+        public ExchangeBuilder DelayedHeaders
+        {
+            get
+            {
+                this.Instance.Type = DelayedExchangeType;
+                this.Instance.Arguments[DelayedExchangeSubtypeArgumentName] = ExchangeType.Headers;
+                return this;
+            }
+        }
 
         #endregion
 

Sources/Contour/Transport/RabbitMQ/Topology/Queue.cs

@@ -54,13 +54,13 @@ internal Queue(string name)
         /// </summary>
         public TimeSpan? Ttl { get; internal set; }
 
-        /// <summary>
-        /// Максимальное количество сообщений в очереди.
+        /// <summary>
+        /// Максимальное количество сообщений в очереди.
         /// </summary>
         public int? Limit { get; internal set; }
 
-        /// <summary>
-        /// Максимальное количество байт, которое занимают сообщения в очереди.
+        /// <summary>
+        /// Максимальное количество байт, которое занимают сообщения в очереди.
         /// </summary>
         public int? MaxLengthBytes { get; internal set; }
 

Sources/Contour/Transport/RabbitMQ/Topology/QueueBuilder.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 
 namespace Contour.Transport.RabbitMQ.Topology
 {

Sources/Contour/Transport/RabbitMQ/Topology/TopologyBuilder.cs

@@ -3,6 +3,7 @@
 using Contour.Receiving;
 using Contour.Topology;
 using Contour.Transport.RabbitMQ.Internal;
+using System.Collections.Generic;
 
 namespace Contour.Transport.RabbitMQ.Topology
 {
@@ -49,11 +50,11 @@ public TopologyBuilder(IChannelProvider<IChannel> channelProvider)
         /// <param name="routingKey">
         /// Ключ маршрутизации, используется для определения очереди, в которую должно быть отправлено сообщение.
         /// </param>
-        public void Bind(Exchange exchange, Queue queue, string routingKey = "")
+        public void Bind(Exchange exchange, Queue queue, string routingKey = "", IDictionary<string, object> arguments = null)
         {
             using (var channel = (RabbitChannel)this.channelProvider.OpenChannel(CancellationToken.None))
             {
-                channel.Bind(queue, exchange, routingKey);
+                channel.Bind(queue, exchange, routingKey, arguments);
             }
         }