Hi,
Attached is a patch that adds support for passing all of the Message
properties via RabbitMQ (as per version 1.1 of System.Messaging).
This includes Recoverable (supporting persistence) and Priority (which
would support priority based delivery if RabbitMQ supported it) as
well as state and argument checking. This patch should be applied to
the "messaging-2008" branch.
Mike.
Index: Mono.Messaging/Mono.Messaging/ConnectionException.cs
===================================================================
--- Mono.Messaging/Mono.Messaging/ConnectionException.cs (revision 116599)
+++ Mono.Messaging/Mono.Messaging/ConnectionException.cs (working copy)
@@ -35,6 +35,12 @@
public class ConnectionException : MonoMessagingException {
private readonly QueueReference qRef;
+ public ConnectionException (QueueReference qRef, Exception e)
+ : base ("Unable to connect to Queue", e)
+ {
+ this.qRef = qRef;
+ }
+
public ConnectionException (QueueReference qRef)
: base ("Unable to connect to Queue")
{
Index: Mono.Messaging/Mono.Messaging/MessageBase.cs
===================================================================
--- Mono.Messaging/Mono.Messaging/MessageBase.cs (revision 116599)
+++ Mono.Messaging/Mono.Messaging/MessageBase.cs (working copy)
@@ -42,12 +42,12 @@
DateTime arrivedTime;
bool attachSenderId = true;
bool authenticated;
- string authenticationProviderName = "";
+ string authenticationProviderName = "Microsoft Base Cryptographic Provider version 1.0";
CryptographicProviderType authenticationProviderType
= CryptographicProviderType.RsaFull;
Stream bodyStream;
int bodyType;
- Guid connectorType;
+ Guid connectorType = Guid.Empty;
string correlationId;
IMessageQueue destinationQueue;
byte[] destinationSymmetricKey = new byte[0];
@@ -78,6 +78,7 @@
bool useEncryption;
bool useJournalQueue;
bool useTracing;
+ bool isDelivered;
public AcknowledgeTypes AcknowledgeType {
get { return acknowledgeType; }
@@ -85,7 +86,10 @@
}
public Acknowledgment Acknowledgment {
- get { return acknowledgment; }
+ get {
+ CheckDelivered ();
+ return acknowledgment;
+ }
}
public IMessageQueue AdministrationQueue {
@@ -99,7 +103,10 @@
}
public DateTime ArrivedTime {
- get { return arrivedTime; }
+ get {
+ CheckDelivered ();
+ return arrivedTime;
+ }
}
public bool AttachSenderId {
@@ -108,7 +115,10 @@
}
public bool Authenticated {
- get { return authenticated; }
+ get {
+ CheckDelivered ();
+ return authenticated;
+ }
}
public string AuthenticationProviderName {
@@ -142,17 +152,28 @@
}
public IMessageQueue DestinationQueue {
- get { return destinationQueue; }
+ get {
+ CheckDelivered ();
+ return destinationQueue;
+ }
}
public byte[] DestinationSymmetricKey {
get { return destinationSymmetricKey; }
- set { destinationSymmetricKey = value; }
+ set {
+ if (value == null)
+ throw new ArgumentException ("DestinationSymmetricKey can not be null");
+ destinationSymmetricKey = value;
+ }
}
public byte[] DigitalSignature {
get { return digitalSignature; }
- set { digitalSignature = value; }
+ set {
+ if (value == null)
+ throw new ArgumentException ("DigitalSignature can not be null");
+ digitalSignature = value;
+ }
}
public EncryptionAlgorithm EncryptionAlgorithm {
@@ -162,7 +183,11 @@
public byte[] Extension {
get { return extension; }
- set { extension = value; }
+ set {
+ if (value == null)
+ throw new ArgumentException ("Extension can not be null");
+ extension = value;
+ }
}
public HashAlgorithm HashAlgorithm {
@@ -171,15 +196,24 @@
}
public string Id {
- get { return id; }
+ get {
+ CheckDelivered ();
+ return id;
+ }
}
public bool IsFirstInTransaction {
- get { return isFirstInTransaction; }
+ get {
+ CheckDelivered ();
+ return isFirstInTransaction;
+ }
}
public bool IsLastInTransaction {
- get { return isLastInTransaction; }
+ get {
+ CheckDelivered ();
+ return isLastInTransaction;
+ }
}
public string Label {
@@ -188,7 +222,10 @@
}
public MessageType MessageType {
- get { return messageType; }
+ get {
+ CheckDelivered ();
+ return messageType;
+ }
}
public MessagePriority Priority {
@@ -212,19 +249,31 @@
}
public byte[] SenderId {
- get { return senderId; }
+ get {
+ CheckDelivered ();
+ return senderId;
+ }
}
public long SenderVersion {
- get { return senderVersion; }
+ get {
+ CheckDelivered ();
+ return senderVersion;
+ }
}
public DateTime SentTime {
- get { return sentTime; }
+ get {
+ CheckDelivered ();
+ return sentTime;
+ }
}
public string SourceMachine {
- get { return sourceMachine; }
+ get {
+ CheckDelivered ();
+ return sourceMachine;
+ }
}
public TimeSpan TimeToBeReceived {
@@ -238,7 +287,10 @@
}
public string TransactionId {
- get { return transactionId; }
+ get {
+ CheckDelivered ();
+ return transactionId;
+ }
}
public IMessageQueue TransactionStatusQueue {
@@ -271,6 +323,12 @@
set { useTracing = value; }
}
+ private void CheckDelivered ()
+ {
+ if (!isDelivered)
+ throw new InvalidOperationException ("Message has not been delivered");
+ }
+
public void SetDeliveryInfo (Acknowledgment acknowledgment,
DateTime arrivedTime,
IMessageQueue destinationQueue,
@@ -278,6 +336,7 @@
MessageType messageType,
byte[] senderId,
long senderVersion,
+ DateTime sentTime,
string sourceMachine,
string transactionId)
{
@@ -288,8 +347,10 @@
this.messageType = messageType;
this.senderId = senderId;
this.senderVersion = senderVersion;
+ this.sentTime = sentTime;
this.sourceMachine = sourceMachine;
this.transactionId = transactionId;
+ this.isDelivered = true;
}
}
}
Index: Mono.Messaging/Mono.Messaging/ChangeLog
===================================================================
--- Mono.Messaging/Mono.Messaging/ChangeLog (revision 116599)
+++ Mono.Messaging/Mono.Messaging/ChangeLog (working copy)
@@ -1,3 +1,11 @@
+2008-10-26 Michael Barker <[EMAIL PROTECTED]>
+
+ * ConnectionException.cs: Added support for InnerExceptions
+ * MessageBase.cs: Added check for delivered messages and throwing exceptions
+ for appropriate properties. Add null argument checks for some byte[]
+ properties.
+ * IMessage.cs: Added SentTime to the list of delivery properties.
+
2008-10-12 Michael Barker <[EMAIL PROTECTED]>
* IMessageEnumerator.cs: New
Index: Mono.Messaging/Mono.Messaging/IMessage.cs
===================================================================
--- Mono.Messaging/Mono.Messaging/IMessage.cs (revision 116599)
+++ Mono.Messaging/Mono.Messaging/IMessage.cs (working copy)
@@ -207,6 +207,7 @@
MessageType messageType,
byte[] senderId,
long senderVersion,
+ DateTime sentTime,
string sourceMachine,
string transactionId);
}
Index: Mono.Messaging.RabbitMQ/Test/Mono.Messaging.RabbitMQ/BasicMessagingTest.cs
===================================================================
--- Mono.Messaging.RabbitMQ/Test/Mono.Messaging.RabbitMQ/BasicMessagingTest.cs (revision 116599)
+++ Mono.Messaging.RabbitMQ/Test/Mono.Messaging.RabbitMQ/BasicMessagingTest.cs (working copy)
@@ -30,6 +30,7 @@
using System;
using System.Messaging;
+using System.Reflection;
using System.Threading;
using System.Text.RegularExpressions;
@@ -64,9 +65,181 @@
Assert.AreEqual (m.CorrelationId, m2.CorrelationId, "CorrelationId not set properly");
Assert.IsTrue (0 != m2.SenderVersion);
Assert.IsNotNull (m2.SourceMachine, "SourceMachine is null");
+ Assert.AreEqual (qName, m2.DestinationQueue.QueueName, "Destination Queue not set");
}
[Test]
+ public void CheckDefaults ()
+ {
+ Message m = new Message ("Test", new BinaryMessageFormatter ());
+ Assert.AreEqual (true, m.AttachSenderId, "AttachSenderId has incorrect default");
+ Assert.AreEqual ("Microsoft Base Cryptographic Provider version 1.0",
+ m.AuthenticationProviderName,
+ "AuthenticationProviderName has incorrect default");
+ Assert.AreEqual (0, m.Extension.Length, "Extension has incorrect default");
+ Assert.AreEqual ("", m.Label, "Label has incorrect default");
+ Assert.IsFalse (m.Recoverable, "Recoverable has incorrect default");
+ Assert.AreEqual (MessagePriority.Normal, m.Priority, "MessagePriority has incorrect default");
+ }
+
+ private static void CheckInvalidOperation (Message m, String property)
+ {
+ PropertyInfo pi = m.GetType ().GetProperty (property);
+ try {
+ Assert.IsNotNull (pi, "Property not defined: " + property);
+ pi.GetValue (m, null);
+ Assert.Fail (property);
+ } catch (InvalidOperationException e) {
+ } catch (TargetInvocationException e) {
+ Assert.AreEqual (typeof (InvalidOperationException),
+ e.InnerException.GetType ());
+ }
+ }
+
+ [Test]
+ public void CheckInvalidPropertyOperations ()
+ {
+ Message m = new Message ("Test", new BinaryMessageFormatter ());
+ CheckInvalidOperation (m, "Acknowledgment");
+ CheckInvalidOperation (m, "ArrivedTime");
+ CheckInvalidOperation (m, "Authenticated");
+ CheckInvalidOperation (m, "DestinationQueue");
+ CheckInvalidOperation (m, "Id");
+ CheckInvalidOperation (m, "IsFirstInTransaction");
+ CheckInvalidOperation (m, "IsLastInTransaction");
+ // TODO: Support 2.0 features.
+ //CheckInvalidOperation (m, "LookupId");
+ CheckInvalidOperation (m, "MessageType");
+ CheckInvalidOperation (m, "SenderId");
+ CheckInvalidOperation (m, "SenderVersion");
+ CheckInvalidOperation (m, "SentTime");
+ CheckInvalidOperation (m, "SourceMachine");
+ CheckInvalidOperation (m, "TransactionId");
+ }
+
+ private static void CheckArgumentInvalid (Message m, String property)
+ {
+ PropertyInfo pi = m.GetType ().GetProperty (property);
+ try {
+ Assert.IsNotNull (pi, "Property not defined: " + property);
+ pi.SetValue (m, null, null);
+ Assert.Fail (property);
+ } catch (InvalidOperationException e) {
+ } catch (TargetInvocationException e) {
+ Assert.AreEqual (typeof (ArgumentException),
+ e.InnerException.GetType ());
+ }
+ }
+
+ [Test]
+ public void CheckArgumentInvalidForProperties ()
+ {
+ Message m = new Message ("Stuff");
+ CheckArgumentInvalid (m, "DestinationSymmetricKey");
+ CheckArgumentInvalid (m, "DigitalSignature");
+ CheckArgumentInvalid (m, "Extension");
+ }
+
+ [Test]
+ public void SendReceiveBinaryMessageWithAllPropertiesSet ()
+ {
+ String qName = "testq";
+ MessageQueue mq = new MessageQueue (qName);
+ Assert.AreEqual (mq.QueueName, qName, "Queue name not set properly");
+
+ MessageQueue adminQ = new MessageQueue ("myAdmin");
+ MessageQueue responseQ = new MessageQueue ("myResponse");
+ Guid connectorType = Guid.NewGuid ();
+ String s = "Test: " + DateTime.Now;
+
+ Message m = new Message (s, new BinaryMessageFormatter ());
+ m.CorrelationId = "foo";
+ m.AcknowledgeType = AcknowledgeTypes.PositiveArrival;
+ m.AdministrationQueue = adminQ;
+ m.AppSpecific = 5;
+ m.AuthenticationProviderName = "Test Provider Name";
+ m.AuthenticationProviderType = CryptographicProviderType.None;
+ m.ConnectorType = connectorType;
+ m.DestinationSymmetricKey = new byte[] { 0x0A, 0x0B, 0x0C };
+ m.DigitalSignature = new byte[] { 0x0C, 0x0D, 0x0E };
+ m.EncryptionAlgorithm = EncryptionAlgorithm.Rc4;
+ m.Extension = new byte[] { 0x01, 0x02, 0x03 };
+ m.HashAlgorithm = HashAlgorithm.Sha;
+ m.Label = "MyLabel";
+ m.Priority = MessagePriority.AboveNormal;
+ m.Recoverable = true;
+ m.ResponseQueue = responseQ;
+ m.SenderCertificate = new byte[] { 0x04, 0x05, 0x06 };
+ m.TimeToBeReceived = new TimeSpan(0, 0, 5);
+ m.TimeToReachQueue = new TimeSpan(0, 0, 10);
+ m.UseAuthentication = true;
+ m.UseDeadLetterQueue = true;
+ m.UseEncryption = true;
+
+ mq.Send (m);
+
+ Message m2 = mq.Receive ();
+ m2.Formatter = new BinaryMessageFormatter ();
+ Assert.AreEqual (s, m2.Body);
+
+ Assert.AreEqual (AcknowledgeTypes.PositiveArrival, m2.AcknowledgeType,
+ "AcknowledgeType not passed correctly");
+ Assert.AreEqual (adminQ.QueueName, m2.AdministrationQueue.QueueName,
+ "AdministrationQueue not passed correctly");
+ Assert.AreEqual (5, m2.AppSpecific, "AppSpecific not passed correctly");
+ Assert.AreEqual (m.AuthenticationProviderName, m2.AuthenticationProviderName,
+ "AuthenticationProviderName not passed correctly");
+ Assert.AreEqual (m.AuthenticationProviderType, m2.AuthenticationProviderType,
+ "AuthenticationProviderType not passed correctly");
+ Assert.AreEqual (connectorType, m2.ConnectorType,
+ "ConnectorType not passed correctly");
+ Assert.AreEqual (m.CorrelationId, m2.CorrelationId,
+ "CorrelationId not passed correctly");
+ AreEqual (m.DestinationSymmetricKey, m2.DestinationSymmetricKey,
+ "DestinationSymmetricKey not passed correctly");
+ AreEqual (m.DigitalSignature, m2.DigitalSignature,
+ "DigitalSignature not passed properly");
+ Assert.AreEqual (EncryptionAlgorithm.Rc4, m2.EncryptionAlgorithm,
+ "EncryptionAlgorithm not passed properly");
+ AreEqual (m.Extension, m2.Extension, "Extension not passed properly");
+ Assert.AreEqual (m.HashAlgorithm, m2.HashAlgorithm,
+ "HashAlgorithm not passed properly");
+ Assert.AreEqual (m.Label, m2.Label, "Label not passed correctly");
+ Assert.AreEqual (MessagePriority.AboveNormal, m2.Priority,
+ "Priority not passed properly");
+ Assert.AreEqual (true, m2.Recoverable, "Recoverable not passed properly");
+ Assert.AreEqual (responseQ.QueueName, m2.ResponseQueue.QueueName,
+ "ResponseQueue not passed properly");
+ AreEqual (m.SenderCertificate, m2.SenderCertificate,
+ "SenderCertificate not passed properly");
+ Assert.AreEqual (m.TimeToBeReceived, m2.TimeToBeReceived,
+ "TimeToBeReceived not passed properly");
+ Assert.AreEqual (m.TimeToReachQueue, m2.TimeToReachQueue,
+ "TimeToReachQueue not passed properly");
+ Assert.IsTrue (m2.UseAuthentication,
+ "UseAuthentication not passed properly");
+ Assert.IsTrue (m2.UseDeadLetterQueue,
+ "UseDeadLetterQueue not passed properly");
+ Assert.IsTrue (m2.UseEncryption, "UseEncryption not pass properly");
+ //Assert.AreEqual ();
+
+ Assert.IsTrue (DateTime.MinValue == m.ArrivedTime);
+ Assert.IsNotNull (m2.Id, "Id is null");
+ Assert.IsTrue (Guid.Empty.ToString () != m2.Id, "Id is Empty");
+ Assert.IsTrue (DateTime.MinValue != m2.ArrivedTime, "Arrived Time is not set");
+ Assert.AreEqual (Acknowledgment.None, m2.Acknowledgment, "Acknowledgment");
+ Assert.IsTrue (0 != m2.SenderVersion);
+ Assert.IsNotNull (m2.SourceMachine, "SourceMachine is null");
+ }
+
+ private static void AreEqual(byte[] expected, byte[] actual, string message)
+ {
+ Assert.AreEqual (expected.Length, actual.Length, message);
+ for (int i = 0; i < expected.Length; i++)
+ Assert.AreEqual (expected[i], actual[i], message);
+ }
+
+ [Test]
public void SendReceiveXmlMessage ()
{
MessageQueue mq = new MessageQueue ("testq");
Index: Mono.Messaging.RabbitMQ/Mono.Messaging.RabbitMQ/RabbitMQMessageQueue.cs
===================================================================
--- Mono.Messaging.RabbitMQ/Mono.Messaging.RabbitMQ/RabbitMQMessageQueue.cs (revision 116599)
+++ Mono.Messaging.RabbitMQ/Mono.Messaging.RabbitMQ/RabbitMQMessageQueue.cs (working copy)
@@ -45,10 +45,6 @@
public class RabbitMQMessageQueue : IMessageQueue {
- private static readonly string SENDER_VERSION_KEY = "SenderVersion";
- private static readonly string SOURCE_MACHINE_KEY = "SourceMachine";
- private static readonly string BODY_TYPE_KEY = "BodyType";
-
private bool authenticate = false;
private short basePriority = 0;
private Guid category = Guid.Empty;
@@ -169,6 +165,7 @@
MessageType.Normal,
new byte[0],
senderVersion,
+ DateTime.UtcNow,
null,
null);
}
@@ -196,7 +193,7 @@
}
}
} catch (BrokerUnreachableException e) {
- throw new ConnectionException (QRef);
+ throw new ConnectionException (QRef, e);
}
}
@@ -216,9 +213,7 @@
if (result == null) {
throw new MonoMessagingException ("No Message Available");
} else {
- DebugUtil.DumpProperties(result, Console.Out, 0);
-
- IMessage m = MessageFactory.ReadMessage (result);
+ IMessage m = MessageFactory.ReadMessage (QRef, result);
return m;
}
}
Index: Mono.Messaging.RabbitMQ/Mono.Messaging.RabbitMQ/MessageFactory.cs
===================================================================
--- Mono.Messaging.RabbitMQ/Mono.Messaging.RabbitMQ/MessageFactory.cs (revision 116599)
+++ Mono.Messaging.RabbitMQ/Mono.Messaging.RabbitMQ/MessageFactory.cs (working copy)
@@ -49,56 +49,143 @@
private static readonly string SENDER_VERSION_KEY = "SenderVersion";
private static readonly string SOURCE_MACHINE_KEY = "SourceMachine";
private static readonly string BODY_TYPE_KEY = "BodyType";
+ private static readonly string ACKNOWLEDGE_TYPE_KEY = "AcknowledgeType";
+ private static readonly string ADMINISTRATION_QUEUE_KEY = "AdministrationQueue";
+ private static readonly string APP_SPECIFIC_KEY = "AppSpecific";
+ private static readonly string AUTHENTICATION_PROVIDER_NAME_KEY = "AuthenticationProviderName";
+ private static readonly string AUTHENTICATION_PROVIDER_TYPE_KEY = "AuthenticationProviderType";
+ private static readonly string CONNECTOR_TYPE_KEY = "ConnectorType";
+ private static readonly string DESTINATION_SYMMETRIC_KEY_KEY = "DestinationSymmetricKey";
+ private static readonly string DIGITAL_SIGNATURE_KEY = "DigitalSignature";
+ private static readonly string ENCRYPTION_ALGORITHM_KEY = "EncryptionAlgorithm";
+ private static readonly string EXTENSION_KEY = "Extension";
+ private static readonly string HASH_ALGORITHM_KEY = "HashAlgorithm";
+ private static readonly string LABEL_KEY = "Label";
+ private static readonly string SENDER_CERTIFICATE_KEY = "SenderCertificate";
+ private static readonly string TIME_TO_BE_RECEIVED_KEY = "TimeToBeReceived";
+ private static readonly string TIME_TO_REACH_QUEUE_KEY = "TimeToReachQueue";
+ private static readonly string USE_AUTHENTICATION_KEY = "UseAuthentication";
+ private static readonly string USE_DEAD_LETTER_QUEUE_KEY = "UseDeadLetterQueue";
+ private static readonly string USE_ENCRYPTION_KEY = "UseEncryption";
+ private static readonly int PERSISTENT_DELIVERY_MODE = 2;
+
public static IMessageBuilder WriteMessage (IModel ch, IMessage msg)
{
BasicMessageBuilder mb = new BasicMessageBuilder (ch);
mb.Properties.MessageId = msg.Id;
if (msg.CorrelationId != null)
mb.Properties.CorrelationId = msg.CorrelationId;
+ // TODO: Change to DateTime.UtcNow??
mb.Properties.Timestamp = MessageFactory.DateTimeToAmqpTimestamp (DateTime.Now);
Hashtable headers = new Hashtable ();
+
headers[SENDER_VERSION_KEY] = msg.SenderVersion;
headers[SOURCE_MACHINE_KEY] = (string) System.Environment.MachineName;
headers[BODY_TYPE_KEY] = msg.BodyType;
+ headers[ACKNOWLEDGE_TYPE_KEY] = (int) msg.AcknowledgeType;
+ if (msg.AdministrationQueue != null)
+ headers[ADMINISTRATION_QUEUE_KEY] = msg.AdministrationQueue.QRef.ToString ();
+ headers[APP_SPECIFIC_KEY] = msg.AppSpecific;
+ headers[AUTHENTICATION_PROVIDER_NAME_KEY] = msg.AuthenticationProviderName;
+ headers[AUTHENTICATION_PROVIDER_TYPE_KEY] = (int) msg.AuthenticationProviderType;
+ headers[CONNECTOR_TYPE_KEY] = msg.ConnectorType.ToString ();
+ headers[DESTINATION_SYMMETRIC_KEY_KEY] = msg.DestinationSymmetricKey;
+ headers[DIGITAL_SIGNATURE_KEY] = msg.DigitalSignature;
+ headers[ENCRYPTION_ALGORITHM_KEY] = (int) msg.EncryptionAlgorithm;
+ headers[EXTENSION_KEY] = msg.Extension;
+ headers[HASH_ALGORITHM_KEY] = (int) msg.HashAlgorithm;
+ SetValue (headers, LABEL_KEY, msg.Label);
+ mb.Properties.Priority = (byte) (int) msg.Priority;
+ mb.Properties.SetPersistent (msg.Recoverable);
+ if (msg.ResponseQueue != null)
+ mb.Properties.ReplyTo = msg.ResponseQueue.QRef.ToString ();
+ headers[SENDER_CERTIFICATE_KEY] = msg.SenderCertificate;
+ headers[TIME_TO_BE_RECEIVED_KEY] = msg.TimeToBeReceived.Ticks;
+ headers[TIME_TO_REACH_QUEUE_KEY] = msg.TimeToReachQueue.Ticks;
+ headers[USE_AUTHENTICATION_KEY] = msg.UseAuthentication;
+ headers[USE_DEAD_LETTER_QUEUE_KEY] = msg.UseDeadLetterQueue;
+ headers[USE_ENCRYPTION_KEY] = msg.UseEncryption;
mb.Properties.Headers = headers;
Stream s = msg.BodyStream;
s.Seek (0, SeekOrigin.Begin);
byte[] buf = new byte[s.Length];
- int numRead = msg.BodyStream.Read (buf, 0, buf.Length);
+ msg.BodyStream.Read (buf, 0, buf.Length);
mb.BodyStream.Write (buf, 0, buf.Length);
return mb;
- }
+ }
- public static IMessage ReadMessage (BasicDeliverEventArgs result)
+ private static void SetValue (Hashtable headers, string name, object val)
{
+ if (val != null)
+ headers[name] = val;
+ }
+
+ public static IMessage ReadMessage (QueueReference destination, BasicDeliverEventArgs result)
+ {
+ /*
+ if (destination == null)
+ throw new ArgumentException ("destination must not be null");
+ if (result == null)
+ throw new ArgumentException ("result must not be null");
+ */
MessageBase msg = new MessageBase ();
Stream s = new MemoryStream ();
s.Write (result.Body, 0, result.Body.Length);
Console.WriteLine ("Body.Length Out {0}", result.Body.Length);
DateTime arrivedTime = DateTime.Now;
- long senderVersion = (long) result.BasicProperties.Headers[SENDER_VERSION_KEY];
- string sourceMachine = GetString (result.BasicProperties, SOURCE_MACHINE_KEY);
+ IDictionary headers = result.BasicProperties.Headers;
+ long senderVersion = (long) headers[SENDER_VERSION_KEY];
+ string sourceMachine = GetString (headers, SOURCE_MACHINE_KEY);
DateTime sentTime = AmqpTimestampToDateTime (result.BasicProperties.Timestamp);
msg.SetDeliveryInfo (Acknowledgment.None,
arrivedTime,
- null,
+ new RabbitMQMessageQueue (destination),
result.BasicProperties.MessageId,
MessageType.Normal,
new byte[0],
senderVersion,
+ sentTime,
sourceMachine,
null);
msg.CorrelationId = result.BasicProperties.CorrelationId;
msg.BodyStream = s;
msg.BodyType = (int) result.BasicProperties.Headers[BODY_TYPE_KEY];
+ msg.AcknowledgeType = (AcknowledgeTypes)
+ Enum.ToObject (typeof (AcknowledgeTypes),
+ headers[ACKNOWLEDGE_TYPE_KEY]);
+ string adminQueuePath = GetString (headers, ADMINISTRATION_QUEUE_KEY);
+ if (adminQueuePath != null)
+ msg.AdministrationQueue = new RabbitMQMessageQueue (QueueReference.Parse (adminQueuePath));
+ msg.AppSpecific = (int) headers[APP_SPECIFIC_KEY];
+ msg.AuthenticationProviderName = GetString (headers, AUTHENTICATION_PROVIDER_NAME_KEY);
+ msg.AuthenticationProviderType = (CryptographicProviderType) Enum.ToObject (typeof (CryptographicProviderType), headers[AUTHENTICATION_PROVIDER_TYPE_KEY]);
+ string connectorType = GetString (headers, CONNECTOR_TYPE_KEY);
+ msg.ConnectorType = new Guid(connectorType);
+ msg.DestinationSymmetricKey = (byte[]) headers[DESTINATION_SYMMETRIC_KEY_KEY];
+ msg.DigitalSignature = (byte[]) headers[DIGITAL_SIGNATURE_KEY];
+ msg.EncryptionAlgorithm = (EncryptionAlgorithm) Enum.ToObject (typeof (EncryptionAlgorithm), headers[ENCRYPTION_ALGORITHM_KEY]);
+ msg.Extension = (byte[]) headers[EXTENSION_KEY];
+ msg.HashAlgorithm = (HashAlgorithm) Enum.ToObject (typeof (HashAlgorithm), headers[HASH_ALGORITHM_KEY]);
+ msg.Label = GetString (headers, LABEL_KEY);
+ msg.Priority = (MessagePriority) Enum.ToObject (typeof (MessagePriority), result.BasicProperties.Priority);
+ msg.Recoverable = result.BasicProperties.DeliveryMode == PERSISTENT_DELIVERY_MODE;
+ if (result.BasicProperties.ReplyTo != null)
+ msg.ResponseQueue = new RabbitMQMessageQueue (QueueReference.Parse (result.BasicProperties.ReplyTo));
+ msg.SenderCertificate = (byte[]) headers[SENDER_CERTIFICATE_KEY];
+ msg.TimeToBeReceived = new TimeSpan((long) headers[TIME_TO_BE_RECEIVED_KEY]);
+ msg.TimeToReachQueue = new TimeSpan((long) headers[TIME_TO_REACH_QUEUE_KEY]);
+ msg.UseAuthentication = (bool) headers[USE_AUTHENTICATION_KEY];
+ msg.UseDeadLetterQueue = (bool) headers[USE_DEAD_LETTER_QUEUE_KEY];
+ msg.UseEncryption = (bool) headers[USE_ENCRYPTION_KEY];
+
return msg;
}
- public static string GetString (IBasicProperties properties, String key)
+ public static string GetString (IDictionary properties, String key)
{
- byte[] b = (byte[]) properties.Headers[key];
+ byte[] b = (byte[]) properties[key];
if (b == null)
return null;
@@ -107,14 +194,14 @@
public static AmqpTimestamp DateTimeToAmqpTimestamp (DateTime t)
{
- DateTime epoch = new DateTime(1970, 1, 1, 0, 0, 0, 0, DateTimeKind.Utc);
+ DateTime epoch = new DateTime(1970, 1, 1, 0, 0, 0, 0);
TimeSpan ts = t.ToUniversalTime () - epoch;
return new AmqpTimestamp((long) ts.TotalSeconds);
}
public static DateTime AmqpTimestampToDateTime (AmqpTimestamp ats)
{
- DateTime epoch = new DateTime(1970, 1, 1, 0, 0, 0, 0, DateTimeKind.Utc);
+ DateTime epoch = new DateTime(1970, 1, 1, 0, 0, 0, 0);
return epoch.AddSeconds (ats.UnixTime).ToLocalTime ();
}
Index: Mono.Messaging.RabbitMQ/Mono.Messaging.RabbitMQ/ChangeLog
===================================================================
--- Mono.Messaging.RabbitMQ/Mono.Messaging.RabbitMQ/ChangeLog (revision 116599)
+++ Mono.Messaging.RabbitMQ/Mono.Messaging.RabbitMQ/ChangeLog (working copy)
@@ -1,3 +1,10 @@
+2008-10-26 Michael Barker <[EMAIL PROTECTED]>
+
+ * MessageFactory.cs: Support all properties defined in the 1.1 version of
+ System.Messaging, including Recoverable and Priority. Changed Timestamp
+ conversion methods to support .NET 1.1 compilation.
+ * RabbitMQMessageQueue.cs: Removed unnecessary constants.
+
2008-10-12 Michael Barker <[EMAIL PROTECTED]>
* RabbitMQMessageEnumerator.cs: New
Index: Mono.Messaging.RabbitMQ/Mono.Messaging.RabbitMQ/RabbitMQMessageEnumerator.cs
===================================================================
--- Mono.Messaging.RabbitMQ/Mono.Messaging.RabbitMQ/RabbitMQMessageEnumerator.cs (revision 116599)
+++ Mono.Messaging.RabbitMQ/Mono.Messaging.RabbitMQ/RabbitMQMessageEnumerator.cs (working copy)
@@ -142,7 +142,7 @@
private IMessage CreateMessage (BasicDeliverEventArgs result)
{
- return MessageFactory.ReadMessage (result);
+ return MessageFactory.ReadMessage (qRef, result);
}
}
_______________________________________________
Mono-devel-list mailing list
Mono-devel-list@lists.ximian.com
http://lists.ximian.com/mailman/listinfo/mono-devel-list
