Allow messages larger than the default

Author: Emile Joubert

projects/wcf/RabbitMQ.ServiceModel/RabbitMQ.ServiceModel.csproj

@@ -83,6 +83,7 @@
     <Reference Include="System.Data" />
     <Reference Include="System.ServiceModel"><RequiredTargetFramework>3.0</RequiredTargetFramework></Reference>
     <Reference Include="System.Xml" />
+    <Reference Include="System.Runtime.Serialization" />
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\..\client\RabbitMQ.Client\RabbitMQ.Client.csproj">

projects/wcf/RabbitMQ.ServiceModel/src/serviceModel/RabbitMQBinding.cs

@@ -56,9 +56,10 @@ public sealed class RabbitMQBinding : Binding
     {
         private String m_host;
         private int m_port;
+        private long m_maxMessageSize;
         private IProtocol m_brokerProtocol;
         private CompositeDuplexBindingElement m_compositeDuplex;
-        private MessageEncodingBindingElement m_encoding;
+        private TextMessageEncodingBindingElement m_encoding;
         private bool m_isInitialized;
         private bool m_oneWayOnly;
         private ReliableSessionBindingElement m_session;
@@ -67,8 +68,8 @@ public sealed class RabbitMQBinding : Binding
         private RabbitMQTransportBindingElement m_transport;
 
         /// <summary>
-        /// Creates a new instance of the RabbitMQBinding class initialized 
-        /// to use the Protocols.DefaultProtocol. The broker must be set 
+        /// Creates a new instance of the RabbitMQBinding class initialized
+        /// to use the Protocols.DefaultProtocol. The broker must be set
         /// before use.
         /// </summary>
         public RabbitMQBinding()
@@ -79,7 +80,8 @@ public RabbitMQBinding()
         /// Uses the default protocol and the broker specified by the given
         /// Uri.
         /// </summary>
-        /// <param name="broker">The address of the broker to connect to</param>
+        /// <param name="hostname">The hostname of the broker to connect to</param>
+        /// <param name="port">The port of the broker to connect to</param>
         public RabbitMQBinding(String hostname, int port)
             : this(hostname, port, Protocols.DefaultProtocol)
         { }
@@ -88,7 +90,7 @@ public RabbitMQBinding(String hostname, int port)
         /// Uses the broker and protocol specified
         /// </summary>
         /// <param name="hostname">The hostname of the broker to connect to</param>
-        /// <param name="port">The port of the broker to connect to</param> 
+        /// <param name="port">The port of the broker to connect to</param>
         /// <param name="protocol">The protocol version to use</param>
         public RabbitMQBinding(String hostname, int port, IProtocol protocol)
             : this(protocol)
@@ -97,6 +99,19 @@ public RabbitMQBinding(String hostname, int port, IProtocol protocol)
             this.Port = port;
         }
 
+        /// <summary>
+        /// Uses the broker and protocol specified
+        /// </summary>
+        /// <param name="hostname">The hostname of the broker to connect to</param>
+        /// <param name="port">The port of the broker to connect to</param>
+        /// <param name="maxMessageSize">The largest allowable encoded message size</param>
+        /// <param name="protocol">The protocol version to use</param>
+        public RabbitMQBinding(String hostname, int port, long maxMessageSize, IProtocol protocol)
+            : this(hostname, port, protocol)
+        {
+            this.MaxMessageSize = maxMessageSize;
+        }
+
         /// <summary>
         /// Uses the specified protocol. The broker must be set before use.
         /// </summary>
@@ -117,6 +132,7 @@ public override BindingElementCollection CreateBindingElements()
             m_transport.HostName = this.HostName;
             m_transport.Port = this.Port;
             m_transport.BrokerProtocol = this.BrokerProtocol;
+            m_transport.MaxReceivedMessageSize = this.MaxMessageSize;
             BindingElementCollection elements = new BindingElementCollection();
 
             if (m_transactionsEnabled)
@@ -145,12 +161,12 @@ private void Initialize()
                     m_session = new ReliableSessionBindingElement();
                     m_compositeDuplex = new CompositeDuplexBindingElement();
                     m_transactionFlow = new TransactionFlowBindingElement();
-
+                    m_maxMessageSize = 8192L;
                     m_isInitialized = true;
                 }
             }
         }
-        
+
         /// <summary>
         /// Gets the scheme used by the binding, soap.amqp
         /// </summary>
@@ -179,6 +195,16 @@ public int Port
             set { m_port = value; }
         }
 
+        /// <summary>
+        /// Specifies the maximum encoded message size
+        /// </summary>
+        [ConfigurationProperty("maxmessagesize")]
+        public long MaxMessageSize
+        {
+            get { return m_maxMessageSize; }
+            set { m_maxMessageSize = value; }
+        }
+
         /// <summary>
         /// Specifies the version of the AMQP protocol that should be used to communicate with the broker
         /// </summary>
@@ -205,7 +231,7 @@ public ReliableSession ReliableSession
         }
 
         /// <summary>
-        /// Determines whether or not the TransactionFlowBindingElement will 
+        /// Determines whether or not the TransactionFlowBindingElement will
         /// be added to the channel stack
         /// </summary>
         public bool TransactionFlow

projects/wcf/RabbitMQ.ServiceModel/src/serviceModel/RabbitMQBindingConfigurationElement.cs

@@ -83,6 +83,7 @@ protected override void InitializeFrom(Binding binding)
             {
                 this.HostName = rabbind.HostName;
                 this.Port = rabbind.Port;
+                this.MaxMessageSize = rabbind.MaxMessageSize;
                 this.OneWayOnly = rabbind.OneWayOnly;
                 this.TransactionFlowEnabled = rabbind.TransactionFlow;
                 this.VirtualHost = rabbind.Transport.ConnectionFactory.VirtualHost;
@@ -113,6 +114,7 @@ protected override void OnApplyConfiguration(Binding binding)
             rabbind.Transport.ConnectionFactory.Password = this.Password;
             rabbind.Transport.ConnectionFactory.UserName = this.Username;
             rabbind.Transport.ConnectionFactory.VirtualHost = this.VirtualHost;
+            rabbind.Transport.MaxReceivedMessageSize = this.MaxMessageSize;
         }
 
         /// <summary>
@@ -191,6 +193,13 @@ public string ProtocolVersion
             }
         }
 
+        [ConfigurationProperty("maxmessagesize", DefaultValue = 8192L)]
+        public long MaxMessageSize
+        {
+            get { return (long)base["maxmessagesize"]; }
+            set { base["maxmessagesize"] = value; }
+        }
+
         private IProtocol GetProtocol() {
             IProtocol result = Protocols.Lookup(this.ProtocolVersion);
             if (result == null) {

projects/wcf/RabbitMQ.ServiceModel/src/serviceModel/RabbitMQInputChannel.cs

@@ -67,7 +67,8 @@ public RabbitMQInputChannel(BindingContext context, IModel model, EndpointAddres
             : base(context, address)
         {
             m_bindingElement = context.Binding.Elements.Find<RabbitMQTransportBindingElement>();
-            MessageEncodingBindingElement encoderElem = context.BindingParameters.Find<MessageEncodingBindingElement>();
+            TextMessageEncodingBindingElement encoderElem = context.BindingParameters.Find<TextMessageEncodingBindingElement>();
+            encoderElem.ReaderQuotas.MaxStringContentLength = (int)m_bindingElement.MaxReceivedMessageSize;
             if (encoderElem != null) {
                 m_encoder = encoderElem.CreateMessageEncoderFactory().Encoder;
             }

projects/wcf/RabbitMQ.ServiceModel/src/serviceModel/RabbitMQTransportBindingElement.cs

@@ -60,6 +60,7 @@ public sealed class RabbitMQTransportBindingElement : TransportBindingElement
         private IProtocol m_protocol;
         private String m_host;
         private int m_port;
+        private long m_maxReceivedMessageSize;
         private String m_username;
         private String m_password;
         private String m_vhost;
@@ -80,6 +81,7 @@ private RabbitMQTransportBindingElement(RabbitMQTransportBindingElement other)
             Username = other.Username;
             Password = other.Password;
             VirtualHost = other.VirtualHost;
+            MaxReceivedMessageSize = other.MaxReceivedMessageSize;
         }
 
 
@@ -190,6 +192,15 @@ public int Port
             }
         }
 
+        /// <summary>
+        /// The largest receivable encoded message
+        /// </summary>
+        public override long MaxReceivedMessageSize
+        {
+            get { return m_maxReceivedMessageSize; }
+            set { m_maxReceivedMessageSize = value; }
+        }
+
         /// <summary>
         /// The username  to use when authenticating with the broker
         /// </summary>

projects/wcf/RabbitMQ.ServiceModel/src/serviceModel/RabbitMQTransportElement.cs

@@ -212,6 +212,15 @@ public string VirtualHost
             set { base["virtualHost"] = value; }
         }
 
+        /// <summary>
+        /// The largest receivable encoded message
+        /// </summary>
+        public new long MaxReceivedMessageSize
+        {
+            get { return MaxReceivedMessageSize; }
+            set { MaxReceivedMessageSize = value; }
+        }
+
         protected override ConfigurationPropertyCollection Properties
         {
             get