From 8ee9b4548437b10a930587f3dbd8f95b3af09d10 Mon Sep 17 00:00:00 2001 From: James Strachan Date: Tue, 28 Feb 2006 12:36:22 +0000 Subject: [PATCH] added support for transactions together with support for consumer.Receive(timeout) git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@381643 13f79535-47bb-0310-9956-ffa450edef68 --- .../src/OpenWire.Client/Connection.cs | 95 ++++-- .../src/OpenWire.Client/Core/Dispatcher.cs | 68 +++- .../OpenWire.Client/Core/ISynchronization.cs | 36 +++ .../Core/TransactionContext.cs | 110 +++++++ .../src/OpenWire.Client/IConnection.cs | 79 +++-- .../src/OpenWire.Client/IMessageConsumer.cs | 2 +- .../IO/ActiveMQDestinationMarshaller.cs | 4 +- .../IO/BaseCommandMarshaller.cs | 6 +- .../OpenWire.Client/IO/BrokerIdMarshaller.cs | 4 +- .../IO/BrokerInfoMarshaller.cs | 20 +- .../IO/ConnectionErrorMarshaller.cs | 8 +- .../IO/ConnectionIdMarshaller.cs | 4 +- .../IO/ConnectionInfoMarshaller.cs | 20 +- .../IO/ConsumerIdMarshaller.cs | 12 +- .../IO/ConsumerInfoMarshaller.cs | 48 +-- .../IO/ControlCommandMarshaller.cs | 4 +- .../IO/DataArrayResponseMarshaller.cs | 4 +- .../IO/DataResponseMarshaller.cs | 4 +- .../IO/DestinationInfoMarshaller.cs | 18 +- .../IO/DiscoveryEventMarshaller.cs | 8 +- .../IO/ExceptionResponseMarshaller.cs | 4 +- .../IO/IntegerResponseMarshaller.cs | 4 +- .../IO/JournalQueueAckMarshaller.cs | 8 +- .../IO/JournalTopicAckMarshaller.cs | 24 +- .../IO/JournalTraceMarshaller.cs | 4 +- .../IO/JournalTransactionMarshaller.cs | 10 +- .../IO/LocalTransactionIdMarshaller.cs | 8 +- .../IO/MessageAckMarshaller.cs | 26 +- .../IO/MessageDispatchMarshaller.cs | 16 +- .../MessageDispatchNotificationMarshaller.cs | 16 +- .../OpenWire.Client/IO/MessageIdMarshaller.cs | 12 +- .../OpenWire.Client/IO/MessageMarshaller.cs | 94 +++--- .../IO/ProducerIdMarshaller.cs | 12 +- .../IO/ProducerInfoMarshaller.cs | 12 +- .../IO/RemoveInfoMarshaller.cs | 4 +- .../IO/RemoveSubscriptionInfoMarshaller.cs | 12 +- .../OpenWire.Client/IO/ResponseMarshaller.cs | 4 +- .../OpenWire.Client/IO/SessionIdMarshaller.cs | 8 +- .../IO/SessionInfoMarshaller.cs | 4 +- .../IO/SubscriptionInfoMarshaller.cs | 16 +- .../IO/TransactionInfoMarshaller.cs | 12 +- .../IO/WireFormatInfoMarshaller.cs | 20 +- .../IO/XATransactionIdMarshaller.cs | 10 +- .../src/OpenWire.Client/ISession.cs | 15 + .../src/OpenWire.Client/MessageConsumer.cs | 109 ++++++- .../src/OpenWire.Client/MessageProducer.cs | 78 ++--- .../src/OpenWire.Client/Session.cs | 67 +++- .../tests/OpenWire.Client/ConsumerTest.cs | 102 +++--- .../tests/OpenWire.Client/TestSupport.cs | 8 +- .../tests/OpenWire.Client/TransactionTest.cs | 290 ++++++++++++++++++ 50 files changed, 1120 insertions(+), 443 deletions(-) create mode 100644 openwire-dotnet/src/OpenWire.Client/Core/ISynchronization.cs create mode 100644 openwire-dotnet/src/OpenWire.Client/Core/TransactionContext.cs create mode 100644 openwire-dotnet/tests/OpenWire.Client/TransactionTest.cs diff --git a/openwire-dotnet/src/OpenWire.Client/Connection.cs b/openwire-dotnet/src/OpenWire.Client/Connection.cs index a9785f9fec..96f123e9f6 100755 --- a/openwire-dotnet/src/OpenWire.Client/Connection.cs +++ b/openwire-dotnet/src/OpenWire.Client/Connection.cs @@ -13,16 +13,16 @@ namespace OpenWire.Client { private ITransport transport; private ConnectionInfo info; + private AcknowledgementMode acknowledgementMode = AcknowledgementMode.AutoAcknowledge; private BrokerInfo brokerInfo; // from broker private WireFormatInfo brokerWireFormatInfo; // from broker private IList sessions = new ArrayList(); - private bool transacted; + private IDictionary consumers = new Hashtable(); // TODO threadsafe private bool connected; private bool closed; - private AcknowledgementMode acknowledgementMode = AcknowledgementMode.AutoAcknowledge; private long sessionCounter; private long temporaryDestinationCounter; - private IDictionary consumers = new Hashtable(); // TODO threadsafe + private long localTransactionCounter; public Connection(ITransport transport, ConnectionInfo info) @@ -32,7 +32,7 @@ namespace OpenWire.Client this.transport.Command += new CommandHandler(OnCommand); this.transport.Start(); } - + /// /// Starts message delivery for this connection. /// @@ -53,16 +53,15 @@ namespace OpenWire.Client /// public ISession CreateSession() { - return CreateSession(transacted, acknowledgementMode); + return CreateSession(acknowledgementMode); } /// /// Creates a new session to work on this connection /// - public ISession CreateSession(bool transacted, AcknowledgementMode acknowledgementMode) + public ISession CreateSession(AcknowledgementMode acknowledgementMode) { - CheckConnected(); - SessionInfo info = CreateSessionInfo(transacted, acknowledgementMode); + SessionInfo info = CreateSessionInfo(acknowledgementMode); SyncRequest(info); Session session = new Session(this, info, acknowledgementMode); sessions.Add(session); @@ -71,10 +70,13 @@ namespace OpenWire.Client public void Dispose() { + /* foreach (Session session in sessions) { session.Dispose(); } + */ + DisposeOf(ConnectionId); sessions.Clear(); transport.Dispose(); closed = true; @@ -88,11 +90,6 @@ namespace OpenWire.Client set { this.transport = value; } } - public bool Transacted - { - get { return transacted; } - set { this.transacted = value; } - } public AcknowledgementMode AcknowledgementMode { @@ -112,13 +109,22 @@ namespace OpenWire.Client } } - public BrokerInfo BrokerInfo { + public ConnectionId ConnectionId + { + get { + return info.ConnectionId; + } + } + + public BrokerInfo BrokerInfo + { get { return brokerInfo; } } - public WireFormatInfo BrokerWireFormat { + public WireFormatInfo BrokerWireFormat + { get { return brokerWireFormatInfo; } @@ -131,6 +137,7 @@ namespace OpenWire.Client /// public Response SyncRequest(Command command) { + CheckConnected(); Response response = transport.Request(command); if (response is ExceptionResponse) { @@ -141,18 +148,17 @@ namespace OpenWire.Client return response; } - - protected SessionInfo CreateSessionInfo(bool transacted, AcknowledgementMode acknowledgementMode) + public void OneWay(Command command) { - SessionInfo answer = new SessionInfo(); - SessionId sessionId = new SessionId(); - sessionId.ConnectionId = info.ConnectionId.Value; - lock (this) - { - sessionId.Value = ++sessionCounter; - } - answer.SessionId = sessionId; - return answer; + CheckConnected(); + transport.Oneway(command); + } + + public void DisposeOf(DataStructure objectId) + { + RemoveInfo command = new RemoveInfo(); + command.ObjectId = objectId; + SyncRequest(command); } @@ -167,6 +173,20 @@ namespace OpenWire.Client } } + /// + /// Creates a new local transaction ID + /// + public LocalTransactionId CreateLocalTransactionId() + { + LocalTransactionId id= new LocalTransactionId(); + id.ConnectionId = ConnectionId; + lock (this) + { + id.Value = (++localTransactionCounter); + } + return id; + } + protected void CheckConnected() { if (closed) @@ -175,9 +195,9 @@ namespace OpenWire.Client } if (!connected) { + connected = true; // now lets send the connection and see if we get an ack/nak SyncRequest(info); - connected = true; } } @@ -224,10 +244,12 @@ namespace OpenWire.Client consumer.Dispatch(message); } } - else if (command is WireFormatInfo) { + else if (command is WireFormatInfo) + { this.brokerWireFormatInfo = (WireFormatInfo) command; } - else if (command is BrokerInfo) { + else if (command is BrokerInfo) + { this.brokerInfo = (BrokerInfo) command; } else @@ -235,6 +257,19 @@ namespace OpenWire.Client Console.WriteLine("ERROR:ÊUnknown command: " + command); } } - + + protected SessionInfo CreateSessionInfo(AcknowledgementMode acknowledgementMode) + { + SessionInfo answer = new SessionInfo(); + SessionId sessionId = new SessionId(); + sessionId.ConnectionId = info.ConnectionId.Value; + lock (this) + { + sessionId.Value = ++sessionCounter; + } + answer.SessionId = sessionId; + return answer; + } + } } diff --git a/openwire-dotnet/src/OpenWire.Client/Core/Dispatcher.cs b/openwire-dotnet/src/OpenWire.Client/Core/Dispatcher.cs index a31719f93a..d2131c1547 100644 --- a/openwire-dotnet/src/OpenWire.Client/Core/Dispatcher.cs +++ b/openwire-dotnet/src/OpenWire.Client/Core/Dispatcher.cs @@ -18,6 +18,7 @@ using System.Collections; using OpenWire.Client.Commands; using System; using OpenWire.Client; +using System.Threading; namespace OpenWire.Client.Core { @@ -26,14 +27,52 @@ namespace OpenWire.Client.Core /// public class Dispatcher { - Queue queue = Queue.Synchronized( new Queue() ); + Queue queue = new Queue(); + Object semaphore = new Object(); + ArrayList messagesToRedeliver = new ArrayList(); + /// + /// Whem we start a transaction we must redeliver any rolled back messages + /// + public void RedeliverRolledBackMessages() { + lock (semaphore) + { + Queue replacement = new Queue(queue.Count + messagesToRedeliver.Count); + foreach (ActiveMQMessage element in messagesToRedeliver) { + replacement.Enqueue(element); + } + messagesToRedeliver.Clear(); + + while (queue.Count > 0) + { + ActiveMQMessage element = (ActiveMQMessage) queue.Dequeue(); + replacement.Enqueue(element); + } + queue = replacement; + Monitor.PulseAll(semaphore); + } + } + + /// + /// Redeliver the given message, putting it at the head of the queue + /// + public void Redeliver(ActiveMQMessage message) + { + lock (semaphore) { + messagesToRedeliver.Add(message); + } + } + /// /// Method Enqueue /// public void Enqueue(ActiveMQMessage message) { - queue.Enqueue(message); + lock (semaphore) + { + queue.Enqueue(message); + Monitor.PulseAll(semaphore); + } } /// @@ -41,9 +80,9 @@ namespace OpenWire.Client.Core /// public IMessage DequeueNoWait() { - lock (queue) + lock (semaphore) { - if (queue.Peek() != null) + if (queue.Count > 0) { return (IMessage) queue.Dequeue(); } @@ -54,10 +93,20 @@ namespace OpenWire.Client.Core /// /// Method Dequeue /// - public IMessage Dequeue(long timeout) + public IMessage Dequeue(int timeout) { - // TODO - throw new Exception("Not implemented yet"); + lock (semaphore) + { + if (queue.Count == 0) + { + Monitor.Wait(semaphore, timeout); + } + if (queue.Count > 0) + { + return (IMessage) queue.Dequeue(); + } + } + return null; } /// @@ -65,7 +114,10 @@ namespace OpenWire.Client.Core /// public IMessage Dequeue() { - return (IMessage) queue.Dequeue(); + lock (semaphore) + { + return (IMessage) queue.Dequeue(); + } } } diff --git a/openwire-dotnet/src/OpenWire.Client/Core/ISynchronization.cs b/openwire-dotnet/src/OpenWire.Client/Core/ISynchronization.cs new file mode 100644 index 0000000000..1fb0fe059b --- /dev/null +++ b/openwire-dotnet/src/OpenWire.Client/Core/ISynchronization.cs @@ -0,0 +1,36 @@ +/* + * Copyright 2006 The Apache Software Foundation or its licensors, as + * applicable. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +namespace OpenWire.Client.Core +{ + public interface ISynchronization + { + /// + /// Called before a commit + /// + void BeforeCommit(); + + /// + /// Called after a commit + /// + void AfterCommit(); + + /// + /// Called after a transaction rollback + /// + void AfterRollback(); + } +} diff --git a/openwire-dotnet/src/OpenWire.Client/Core/TransactionContext.cs b/openwire-dotnet/src/OpenWire.Client/Core/TransactionContext.cs new file mode 100644 index 0000000000..7556c89e9b --- /dev/null +++ b/openwire-dotnet/src/OpenWire.Client/Core/TransactionContext.cs @@ -0,0 +1,110 @@ +/* + * Copyright 2006 The Apache Software Foundation or its licensors, as + * applicable. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +using System.Collections; +using OpenWire.Client.Commands; +using System; +using OpenWire.Client; + +namespace OpenWire.Client.Core +{ + public enum TransactionType + { + Begin = 0, Prepare = 1, CommitOnePhase = 2, CommitTwoPhase = 3, Rollback = 4, Recover=5, Forget = 6, End = 7 + } + + public class TransactionContext + { + private TransactionId transactionId; + private Session session; + private ArrayList synchronizations = new ArrayList(); + + public TransactionContext(Session session) { + this.session = session; + } + + public TransactionId TransactionId + { + get { return transactionId; } + } + + /// + /// Method AddSynchronization + /// + public void AddSynchronization(ISynchronization synchronization) + { + synchronizations.Add(synchronization); + } + + + public void Begin() + { + if (transactionId == null) + { + transactionId = session.Connection.CreateLocalTransactionId(); + + TransactionInfo info = new TransactionInfo(); + info.ConnectionId = session.Connection.ConnectionId; + info.TransactionId = transactionId; + info.Type = (int) TransactionType.Begin; + session.Connection.OneWay(info); + } + } + + + public void Rollback() + { + if (transactionId != null) + { + TransactionInfo info = new TransactionInfo(); + info.ConnectionId = session.Connection.ConnectionId; + info.TransactionId = transactionId; + info.Type = (int) TransactionType.Rollback; + + transactionId = null; + session.Connection.OneWay(info); + } + + foreach (ISynchronization synchronization in synchronizations) { + synchronization.AfterRollback(); + } + synchronizations.Clear(); + } + + public void Commit() + { + foreach (ISynchronization synchronization in synchronizations) { + synchronization.BeforeCommit(); + } + + if (transactionId != null) + { + TransactionInfo info = new TransactionInfo(); + info.ConnectionId = session.Connection.ConnectionId; + info.TransactionId = transactionId; + info.Type = (int) TransactionType.CommitOnePhase; + + transactionId = null; + session.Connection.OneWay(info); + } + + foreach (ISynchronization synchronization in synchronizations) { + synchronization.AfterCommit(); + } + synchronizations.Clear(); + } + } +} diff --git a/openwire-dotnet/src/OpenWire.Client/IConnection.cs b/openwire-dotnet/src/OpenWire.Client/IConnection.cs index e64893f12c..cecc5d4ff9 100755 --- a/openwire-dotnet/src/OpenWire.Client/IConnection.cs +++ b/openwire-dotnet/src/OpenWire.Client/IConnection.cs @@ -17,47 +17,46 @@ using System; using OpenWire.Client.Commands; -namespace OpenWire.Client { - - public enum AcknowledgementMode { - Unknown, AutoAcknowledge, ClientAcknowledge, Transactional - } - - +namespace OpenWire.Client +{ + + public enum AcknowledgementMode + { + Unknown, AutoAcknowledge, ClientAcknowledge, Transactional + } + + + /// + /// Represents a connection with a message broker + /// + public interface IConnection : IDisposable, IStartable + { + /// - /// Represents a connection with a message broker + /// Creates a new session to work on this connection /// - public interface IConnection : IDisposable, IStartable { - - /// - /// Creates a new session to work on this connection - /// - ISession CreateSession(); - - /// - /// Creates a new session to work on this connection - /// - ISession CreateSession(bool transacted, AcknowledgementMode acknowledgementMode); - - - // Properties - - bool Transacted { - get; - set; - } - - AcknowledgementMode AcknowledgementMode { - get; - set; - } - - String ClientId - { - get; - set; - } - - + ISession CreateSession(); + + /// + /// Creates a new session to work on this connection + /// + ISession CreateSession(AcknowledgementMode acknowledgementMode); + + + // Properties + + AcknowledgementMode AcknowledgementMode + { + get; + set; } + + String ClientId + { + get; + set; + } + + + } } diff --git a/openwire-dotnet/src/OpenWire.Client/IMessageConsumer.cs b/openwire-dotnet/src/OpenWire.Client/IMessageConsumer.cs index ac583a6f2a..8be6589e57 100755 --- a/openwire-dotnet/src/OpenWire.Client/IMessageConsumer.cs +++ b/openwire-dotnet/src/OpenWire.Client/IMessageConsumer.cs @@ -35,7 +35,7 @@ namespace OpenWire.Client /// /// If a message is available within the timeout duration it is returned otherwise this method returns null /// - IMessage Receive(long timeout); + IMessage Receive(int timeout); /// /// If a message is available immediately it is returned otherwise this method returns null diff --git a/openwire-dotnet/src/OpenWire.Client/IO/ActiveMQDestinationMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/ActiveMQDestinationMarshaller.cs index 43c6284b45..7afa70f9ec 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/ActiveMQDestinationMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/ActiveMQDestinationMarshaller.cs @@ -57,7 +57,7 @@ namespace OpenWire.Client.IO ActiveMQDestination info = (ActiveMQDestination)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += WriteString(info.PhysicalName, bs); + rc += WriteString(info.PhysicalName, bs); return rc + 0; } @@ -69,7 +69,7 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); ActiveMQDestination info = (ActiveMQDestination)o; - WriteString(info.PhysicalName, dataOut, bs); + WriteString(info.PhysicalName, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/BaseCommandMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/BaseCommandMarshaller.cs index 452759b1e1..e2b3b37990 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/BaseCommandMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/BaseCommandMarshaller.cs @@ -58,7 +58,7 @@ namespace OpenWire.Client.IO BaseCommand info = (BaseCommand)o; int rc = base.Marshal1(wireFormat, info, bs); - bs.WriteBoolean(info.ResponseRequired); + bs.WriteBoolean(info.ResponseRequired); return rc + 1; } @@ -70,8 +70,8 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); BaseCommand info = (BaseCommand)o; - DataStreamMarshaller.WriteShort(info.CommandId, dataOut); - bs.ReadBoolean(); + DataStreamMarshaller.WriteShort(info.CommandId, dataOut); + bs.ReadBoolean(); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/BrokerIdMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/BrokerIdMarshaller.cs index 52050e4026..6033641898 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/BrokerIdMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/BrokerIdMarshaller.cs @@ -68,7 +68,7 @@ namespace OpenWire.Client.IO BrokerId info = (BrokerId)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += WriteString(info.Value, bs); + rc += WriteString(info.Value, bs); return rc + 0; } @@ -80,7 +80,7 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); BrokerId info = (BrokerId)o; - WriteString(info.Value, dataOut, bs); + WriteString(info.Value, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/BrokerInfoMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/BrokerInfoMarshaller.cs index 40bce86a60..e871155871 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/BrokerInfoMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/BrokerInfoMarshaller.cs @@ -83,11 +83,11 @@ namespace OpenWire.Client.IO BrokerInfo info = (BrokerInfo)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1CachedObject(wireFormat, info.BrokerId, bs); - rc += WriteString(info.BrokerURL, bs); - rc += MarshalObjectArray(wireFormat, info.PeerBrokerInfos, bs); - rc += WriteString(info.BrokerName, bs); - bs.WriteBoolean(info.SlaveBroker); + rc += Marshal1CachedObject(wireFormat, info.BrokerId, bs); + rc += WriteString(info.BrokerURL, bs); + rc += MarshalObjectArray(wireFormat, info.PeerBrokerInfos, bs); + rc += WriteString(info.BrokerName, bs); + bs.WriteBoolean(info.SlaveBroker); return rc + 0; } @@ -99,11 +99,11 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); BrokerInfo info = (BrokerInfo)o; - Marshal2CachedObject(wireFormat, info.BrokerId, dataOut, bs); - WriteString(info.BrokerURL, dataOut, bs); - MarshalObjectArray(wireFormat, info.PeerBrokerInfos, dataOut, bs); - WriteString(info.BrokerName, dataOut, bs); - bs.ReadBoolean(); + Marshal2CachedObject(wireFormat, info.BrokerId, dataOut, bs); + WriteString(info.BrokerURL, dataOut, bs); + MarshalObjectArray(wireFormat, info.PeerBrokerInfos, dataOut, bs); + WriteString(info.BrokerName, dataOut, bs); + bs.ReadBoolean(); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/ConnectionErrorMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/ConnectionErrorMarshaller.cs index b5ff0256b7..136f21d4b1 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/ConnectionErrorMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/ConnectionErrorMarshaller.cs @@ -69,8 +69,8 @@ namespace OpenWire.Client.IO ConnectionError info = (ConnectionError)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += MarshalBrokerError(wireFormat, info.Exception, bs); - rc += Marshal1NestedObject(wireFormat, info.ConnectionId, bs); + rc += MarshalBrokerError(wireFormat, info.Exception, bs); + rc += Marshal1NestedObject(wireFormat, info.ConnectionId, bs); return rc + 0; } @@ -82,8 +82,8 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); ConnectionError info = (ConnectionError)o; - MarshalBrokerError(wireFormat, info.Exception, dataOut, bs); - Marshal2NestedObject(wireFormat, info.ConnectionId, dataOut, bs); + MarshalBrokerError(wireFormat, info.Exception, dataOut, bs); + Marshal2NestedObject(wireFormat, info.ConnectionId, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/ConnectionIdMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/ConnectionIdMarshaller.cs index 7d8d369cb1..777eab8519 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/ConnectionIdMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/ConnectionIdMarshaller.cs @@ -68,7 +68,7 @@ namespace OpenWire.Client.IO ConnectionId info = (ConnectionId)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += WriteString(info.Value, bs); + rc += WriteString(info.Value, bs); return rc + 0; } @@ -80,7 +80,7 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); ConnectionId info = (ConnectionId)o; - WriteString(info.Value, dataOut, bs); + WriteString(info.Value, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/ConnectionInfoMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/ConnectionInfoMarshaller.cs index ed252c244f..de03922be6 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/ConnectionInfoMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/ConnectionInfoMarshaller.cs @@ -83,11 +83,11 @@ namespace OpenWire.Client.IO ConnectionInfo info = (ConnectionInfo)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1CachedObject(wireFormat, info.ConnectionId, bs); - rc += WriteString(info.ClientId, bs); - rc += WriteString(info.Password, bs); - rc += WriteString(info.UserName, bs); - rc += MarshalObjectArray(wireFormat, info.BrokerPath, bs); + rc += Marshal1CachedObject(wireFormat, info.ConnectionId, bs); + rc += WriteString(info.ClientId, bs); + rc += WriteString(info.Password, bs); + rc += WriteString(info.UserName, bs); + rc += MarshalObjectArray(wireFormat, info.BrokerPath, bs); return rc + 0; } @@ -99,11 +99,11 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); ConnectionInfo info = (ConnectionInfo)o; - Marshal2CachedObject(wireFormat, info.ConnectionId, dataOut, bs); - WriteString(info.ClientId, dataOut, bs); - WriteString(info.Password, dataOut, bs); - WriteString(info.UserName, dataOut, bs); - MarshalObjectArray(wireFormat, info.BrokerPath, dataOut, bs); + Marshal2CachedObject(wireFormat, info.ConnectionId, dataOut, bs); + WriteString(info.ClientId, dataOut, bs); + WriteString(info.Password, dataOut, bs); + WriteString(info.UserName, dataOut, bs); + MarshalObjectArray(wireFormat, info.BrokerPath, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/ConsumerIdMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/ConsumerIdMarshaller.cs index 720a668718..e1a1f42c82 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/ConsumerIdMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/ConsumerIdMarshaller.cs @@ -70,9 +70,9 @@ namespace OpenWire.Client.IO ConsumerId info = (ConsumerId)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += WriteString(info.ConnectionId, bs); - rc += Marshal1Long(wireFormat, info.SessionId, bs); - rc += Marshal1Long(wireFormat, info.Value, bs); + rc += WriteString(info.ConnectionId, bs); + rc += Marshal1Long(wireFormat, info.SessionId, bs); + rc += Marshal1Long(wireFormat, info.Value, bs); return rc + 0; } @@ -84,9 +84,9 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); ConsumerId info = (ConsumerId)o; - WriteString(info.ConnectionId, dataOut, bs); - Marshal2Long(wireFormat, info.SessionId, dataOut, bs); - Marshal2Long(wireFormat, info.Value, dataOut, bs); + WriteString(info.ConnectionId, dataOut, bs); + Marshal2Long(wireFormat, info.SessionId, dataOut, bs); + Marshal2Long(wireFormat, info.Value, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/ConsumerInfoMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/ConsumerInfoMarshaller.cs index 8cfd215a60..7b86767e88 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/ConsumerInfoMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/ConsumerInfoMarshaller.cs @@ -91,17 +91,17 @@ namespace OpenWire.Client.IO ConsumerInfo info = (ConsumerInfo)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1CachedObject(wireFormat, info.ConsumerId, bs); - bs.WriteBoolean(info.Browser); - rc += Marshal1CachedObject(wireFormat, info.Destination, bs); - bs.WriteBoolean(info.DispatchAsync); - rc += WriteString(info.Selector, bs); - rc += WriteString(info.SubcriptionName, bs); - bs.WriteBoolean(info.NoLocal); - bs.WriteBoolean(info.Exclusive); - bs.WriteBoolean(info.Retroactive); - rc += MarshalObjectArray(wireFormat, info.BrokerPath, bs); - bs.WriteBoolean(info.NetworkSubscription); + rc += Marshal1CachedObject(wireFormat, info.ConsumerId, bs); + bs.WriteBoolean(info.Browser); + rc += Marshal1CachedObject(wireFormat, info.Destination, bs); + bs.WriteBoolean(info.DispatchAsync); + rc += WriteString(info.Selector, bs); + rc += WriteString(info.SubcriptionName, bs); + bs.WriteBoolean(info.NoLocal); + bs.WriteBoolean(info.Exclusive); + bs.WriteBoolean(info.Retroactive); + rc += MarshalObjectArray(wireFormat, info.BrokerPath, bs); + bs.WriteBoolean(info.NetworkSubscription); return rc + 2; } @@ -113,19 +113,19 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); ConsumerInfo info = (ConsumerInfo)o; - Marshal2CachedObject(wireFormat, info.ConsumerId, dataOut, bs); - bs.ReadBoolean(); - Marshal2CachedObject(wireFormat, info.Destination, dataOut, bs); - DataStreamMarshaller.WriteInt(info.PrefetchSize, dataOut); - bs.ReadBoolean(); - WriteString(info.Selector, dataOut, bs); - WriteString(info.SubcriptionName, dataOut, bs); - bs.ReadBoolean(); - bs.ReadBoolean(); - bs.ReadBoolean(); - DataStreamMarshaller.WriteByte(info.Priority, dataOut); - MarshalObjectArray(wireFormat, info.BrokerPath, dataOut, bs); - bs.ReadBoolean(); + Marshal2CachedObject(wireFormat, info.ConsumerId, dataOut, bs); + bs.ReadBoolean(); + Marshal2CachedObject(wireFormat, info.Destination, dataOut, bs); + DataStreamMarshaller.WriteInt(info.PrefetchSize, dataOut); + bs.ReadBoolean(); + WriteString(info.Selector, dataOut, bs); + WriteString(info.SubcriptionName, dataOut, bs); + bs.ReadBoolean(); + bs.ReadBoolean(); + bs.ReadBoolean(); + DataStreamMarshaller.WriteByte(info.Priority, dataOut); + MarshalObjectArray(wireFormat, info.BrokerPath, dataOut, bs); + bs.ReadBoolean(); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/ControlCommandMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/ControlCommandMarshaller.cs index 9231618f2a..b83583dce2 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/ControlCommandMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/ControlCommandMarshaller.cs @@ -68,7 +68,7 @@ namespace OpenWire.Client.IO ControlCommand info = (ControlCommand)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += WriteString(info.Command, bs); + rc += WriteString(info.Command, bs); return rc + 0; } @@ -80,7 +80,7 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); ControlCommand info = (ControlCommand)o; - WriteString(info.Command, dataOut, bs); + WriteString(info.Command, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/DataArrayResponseMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/DataArrayResponseMarshaller.cs index 58725018df..cacb34de9f 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/DataArrayResponseMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/DataArrayResponseMarshaller.cs @@ -79,7 +79,7 @@ namespace OpenWire.Client.IO DataArrayResponse info = (DataArrayResponse)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += MarshalObjectArray(wireFormat, info.Data, bs); + rc += MarshalObjectArray(wireFormat, info.Data, bs); return rc + 0; } @@ -91,7 +91,7 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); DataArrayResponse info = (DataArrayResponse)o; - MarshalObjectArray(wireFormat, info.Data, dataOut, bs); + MarshalObjectArray(wireFormat, info.Data, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/DataResponseMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/DataResponseMarshaller.cs index 36a8d189c0..505206aacf 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/DataResponseMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/DataResponseMarshaller.cs @@ -68,7 +68,7 @@ namespace OpenWire.Client.IO DataResponse info = (DataResponse)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1NestedObject(wireFormat, info.Data, bs); + rc += Marshal1NestedObject(wireFormat, info.Data, bs); return rc + 0; } @@ -80,7 +80,7 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); DataResponse info = (DataResponse)o; - Marshal2NestedObject(wireFormat, info.Data, dataOut, bs); + Marshal2NestedObject(wireFormat, info.Data, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/DestinationInfoMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/DestinationInfoMarshaller.cs index 5148ecf733..76237f3cf1 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/DestinationInfoMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/DestinationInfoMarshaller.cs @@ -83,10 +83,10 @@ namespace OpenWire.Client.IO DestinationInfo info = (DestinationInfo)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1CachedObject(wireFormat, info.ConnectionId, bs); - rc += Marshal1CachedObject(wireFormat, info.Destination, bs); - rc += Marshal1Long(wireFormat, info.Timeout, bs); - rc += MarshalObjectArray(wireFormat, info.BrokerPath, bs); + rc += Marshal1CachedObject(wireFormat, info.ConnectionId, bs); + rc += Marshal1CachedObject(wireFormat, info.Destination, bs); + rc += Marshal1Long(wireFormat, info.Timeout, bs); + rc += MarshalObjectArray(wireFormat, info.BrokerPath, bs); return rc + 1; } @@ -98,11 +98,11 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); DestinationInfo info = (DestinationInfo)o; - Marshal2CachedObject(wireFormat, info.ConnectionId, dataOut, bs); - Marshal2CachedObject(wireFormat, info.Destination, dataOut, bs); - DataStreamMarshaller.WriteByte(info.OperationType, dataOut); - Marshal2Long(wireFormat, info.Timeout, dataOut, bs); - MarshalObjectArray(wireFormat, info.BrokerPath, dataOut, bs); + Marshal2CachedObject(wireFormat, info.ConnectionId, dataOut, bs); + Marshal2CachedObject(wireFormat, info.Destination, dataOut, bs); + DataStreamMarshaller.WriteByte(info.OperationType, dataOut); + Marshal2Long(wireFormat, info.Timeout, dataOut, bs); + MarshalObjectArray(wireFormat, info.BrokerPath, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/DiscoveryEventMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/DiscoveryEventMarshaller.cs index f77755aea6..c5aa856753 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/DiscoveryEventMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/DiscoveryEventMarshaller.cs @@ -69,8 +69,8 @@ namespace OpenWire.Client.IO DiscoveryEvent info = (DiscoveryEvent)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += WriteString(info.ServiceName, bs); - rc += WriteString(info.BrokerName, bs); + rc += WriteString(info.ServiceName, bs); + rc += WriteString(info.BrokerName, bs); return rc + 0; } @@ -82,8 +82,8 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); DiscoveryEvent info = (DiscoveryEvent)o; - WriteString(info.ServiceName, dataOut, bs); - WriteString(info.BrokerName, dataOut, bs); + WriteString(info.ServiceName, dataOut, bs); + WriteString(info.BrokerName, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/ExceptionResponseMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/ExceptionResponseMarshaller.cs index 6478af3b77..614b3cdb5a 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/ExceptionResponseMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/ExceptionResponseMarshaller.cs @@ -68,7 +68,7 @@ namespace OpenWire.Client.IO ExceptionResponse info = (ExceptionResponse)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += MarshalBrokerError(wireFormat, info.Exception, bs); + rc += MarshalBrokerError(wireFormat, info.Exception, bs); return rc + 0; } @@ -80,7 +80,7 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); ExceptionResponse info = (ExceptionResponse)o; - MarshalBrokerError(wireFormat, info.Exception, dataOut, bs); + MarshalBrokerError(wireFormat, info.Exception, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/IntegerResponseMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/IntegerResponseMarshaller.cs index 05baaa5b10..76115e82a0 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/IntegerResponseMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/IntegerResponseMarshaller.cs @@ -68,7 +68,7 @@ namespace OpenWire.Client.IO IntegerResponse info = (IntegerResponse)o; int rc = base.Marshal1(wireFormat, info, bs); - + return rc + 1; } @@ -79,7 +79,7 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); IntegerResponse info = (IntegerResponse)o; - DataStreamMarshaller.WriteInt(info.Result, dataOut); + DataStreamMarshaller.WriteInt(info.Result, dataOut); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/JournalQueueAckMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/JournalQueueAckMarshaller.cs index 3afedd9b68..7115e2f521 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/JournalQueueAckMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/JournalQueueAckMarshaller.cs @@ -69,8 +69,8 @@ namespace OpenWire.Client.IO JournalQueueAck info = (JournalQueueAck)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1NestedObject(wireFormat, info.Destination, bs); - rc += Marshal1NestedObject(wireFormat, info.MessageAck, bs); + rc += Marshal1NestedObject(wireFormat, info.Destination, bs); + rc += Marshal1NestedObject(wireFormat, info.MessageAck, bs); return rc + 0; } @@ -82,8 +82,8 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); JournalQueueAck info = (JournalQueueAck)o; - Marshal2NestedObject(wireFormat, info.Destination, dataOut, bs); - Marshal2NestedObject(wireFormat, info.MessageAck, dataOut, bs); + Marshal2NestedObject(wireFormat, info.Destination, dataOut, bs); + Marshal2NestedObject(wireFormat, info.MessageAck, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/JournalTopicAckMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/JournalTopicAckMarshaller.cs index 513b60aea0..0a9fb0a3e2 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/JournalTopicAckMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/JournalTopicAckMarshaller.cs @@ -73,12 +73,12 @@ namespace OpenWire.Client.IO JournalTopicAck info = (JournalTopicAck)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1NestedObject(wireFormat, info.Destination, bs); - rc += Marshal1NestedObject(wireFormat, info.MessageId, bs); - rc += Marshal1Long(wireFormat, info.MessageSequenceId, bs); - rc += WriteString(info.SubscritionName, bs); - rc += WriteString(info.ClientId, bs); - rc += Marshal1NestedObject(wireFormat, info.TransactionId, bs); + rc += Marshal1NestedObject(wireFormat, info.Destination, bs); + rc += Marshal1NestedObject(wireFormat, info.MessageId, bs); + rc += Marshal1Long(wireFormat, info.MessageSequenceId, bs); + rc += WriteString(info.SubscritionName, bs); + rc += WriteString(info.ClientId, bs); + rc += Marshal1NestedObject(wireFormat, info.TransactionId, bs); return rc + 0; } @@ -90,12 +90,12 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); JournalTopicAck info = (JournalTopicAck)o; - Marshal2NestedObject(wireFormat, info.Destination, dataOut, bs); - Marshal2NestedObject(wireFormat, info.MessageId, dataOut, bs); - Marshal2Long(wireFormat, info.MessageSequenceId, dataOut, bs); - WriteString(info.SubscritionName, dataOut, bs); - WriteString(info.ClientId, dataOut, bs); - Marshal2NestedObject(wireFormat, info.TransactionId, dataOut, bs); + Marshal2NestedObject(wireFormat, info.Destination, dataOut, bs); + Marshal2NestedObject(wireFormat, info.MessageId, dataOut, bs); + Marshal2Long(wireFormat, info.MessageSequenceId, dataOut, bs); + WriteString(info.SubscritionName, dataOut, bs); + WriteString(info.ClientId, dataOut, bs); + Marshal2NestedObject(wireFormat, info.TransactionId, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/JournalTraceMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/JournalTraceMarshaller.cs index f81bf3c525..101ea3891c 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/JournalTraceMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/JournalTraceMarshaller.cs @@ -68,7 +68,7 @@ namespace OpenWire.Client.IO JournalTrace info = (JournalTrace)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += WriteString(info.Message, bs); + rc += WriteString(info.Message, bs); return rc + 0; } @@ -80,7 +80,7 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); JournalTrace info = (JournalTrace)o; - WriteString(info.Message, dataOut, bs); + WriteString(info.Message, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/JournalTransactionMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/JournalTransactionMarshaller.cs index 3db511646e..6e9e235696 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/JournalTransactionMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/JournalTransactionMarshaller.cs @@ -70,8 +70,8 @@ namespace OpenWire.Client.IO JournalTransaction info = (JournalTransaction)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1NestedObject(wireFormat, info.TransactionId, bs); - bs.WriteBoolean(info.WasPrepared); + rc += Marshal1NestedObject(wireFormat, info.TransactionId, bs); + bs.WriteBoolean(info.WasPrepared); return rc + 1; } @@ -83,9 +83,9 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); JournalTransaction info = (JournalTransaction)o; - Marshal2NestedObject(wireFormat, info.TransactionId, dataOut, bs); - DataStreamMarshaller.WriteByte(info.Type, dataOut); - bs.ReadBoolean(); + Marshal2NestedObject(wireFormat, info.TransactionId, dataOut, bs); + DataStreamMarshaller.WriteByte(info.Type, dataOut); + bs.ReadBoolean(); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/LocalTransactionIdMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/LocalTransactionIdMarshaller.cs index 6c170949a8..c15e53f268 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/LocalTransactionIdMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/LocalTransactionIdMarshaller.cs @@ -69,8 +69,8 @@ namespace OpenWire.Client.IO LocalTransactionId info = (LocalTransactionId)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1Long(wireFormat, info.Value, bs); - rc += Marshal1CachedObject(wireFormat, info.ConnectionId, bs); + rc += Marshal1Long(wireFormat, info.Value, bs); + rc += Marshal1CachedObject(wireFormat, info.ConnectionId, bs); return rc + 0; } @@ -82,8 +82,8 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); LocalTransactionId info = (LocalTransactionId)o; - Marshal2Long(wireFormat, info.Value, dataOut, bs); - Marshal2CachedObject(wireFormat, info.ConnectionId, dataOut, bs); + Marshal2Long(wireFormat, info.Value, dataOut, bs); + Marshal2CachedObject(wireFormat, info.ConnectionId, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/MessageAckMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/MessageAckMarshaller.cs index 351abf18fd..e77d6ee31a 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/MessageAckMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/MessageAckMarshaller.cs @@ -74,12 +74,12 @@ namespace OpenWire.Client.IO MessageAck info = (MessageAck)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1CachedObject(wireFormat, info.Destination, bs); - rc += Marshal1CachedObject(wireFormat, info.TransactionId, bs); - rc += Marshal1CachedObject(wireFormat, info.ConsumerId, bs); - rc += Marshal1NestedObject(wireFormat, info.FirstMessageId, bs); - rc += Marshal1NestedObject(wireFormat, info.LastMessageId, bs); - + rc += Marshal1CachedObject(wireFormat, info.Destination, bs); + rc += Marshal1CachedObject(wireFormat, info.TransactionId, bs); + rc += Marshal1CachedObject(wireFormat, info.ConsumerId, bs); + rc += Marshal1NestedObject(wireFormat, info.FirstMessageId, bs); + rc += Marshal1NestedObject(wireFormat, info.LastMessageId, bs); + return rc + 2; } @@ -90,13 +90,13 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); MessageAck info = (MessageAck)o; - Marshal2CachedObject(wireFormat, info.Destination, dataOut, bs); - Marshal2CachedObject(wireFormat, info.TransactionId, dataOut, bs); - Marshal2CachedObject(wireFormat, info.ConsumerId, dataOut, bs); - DataStreamMarshaller.WriteByte(info.AckType, dataOut); - Marshal2NestedObject(wireFormat, info.FirstMessageId, dataOut, bs); - Marshal2NestedObject(wireFormat, info.LastMessageId, dataOut, bs); - DataStreamMarshaller.WriteInt(info.MessageCount, dataOut); + Marshal2CachedObject(wireFormat, info.Destination, dataOut, bs); + Marshal2CachedObject(wireFormat, info.TransactionId, dataOut, bs); + Marshal2CachedObject(wireFormat, info.ConsumerId, dataOut, bs); + DataStreamMarshaller.WriteByte(info.AckType, dataOut); + Marshal2NestedObject(wireFormat, info.FirstMessageId, dataOut, bs); + Marshal2NestedObject(wireFormat, info.LastMessageId, dataOut, bs); + DataStreamMarshaller.WriteInt(info.MessageCount, dataOut); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/MessageDispatchMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/MessageDispatchMarshaller.cs index cd084c4a96..15d68cf8eb 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/MessageDispatchMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/MessageDispatchMarshaller.cs @@ -71,10 +71,10 @@ namespace OpenWire.Client.IO MessageDispatch info = (MessageDispatch)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1CachedObject(wireFormat, info.ConsumerId, bs); - rc += Marshal1CachedObject(wireFormat, info.Destination, bs); - rc += Marshal1NestedObject(wireFormat, info.Message, bs); - + rc += Marshal1CachedObject(wireFormat, info.ConsumerId, bs); + rc += Marshal1CachedObject(wireFormat, info.Destination, bs); + rc += Marshal1NestedObject(wireFormat, info.Message, bs); + return rc + 1; } @@ -85,10 +85,10 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); MessageDispatch info = (MessageDispatch)o; - Marshal2CachedObject(wireFormat, info.ConsumerId, dataOut, bs); - Marshal2CachedObject(wireFormat, info.Destination, dataOut, bs); - Marshal2NestedObject(wireFormat, info.Message, dataOut, bs); - DataStreamMarshaller.WriteInt(info.RedeliveryCounter, dataOut); + Marshal2CachedObject(wireFormat, info.ConsumerId, dataOut, bs); + Marshal2CachedObject(wireFormat, info.Destination, dataOut, bs); + Marshal2NestedObject(wireFormat, info.Message, dataOut, bs); + DataStreamMarshaller.WriteInt(info.RedeliveryCounter, dataOut); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/MessageDispatchNotificationMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/MessageDispatchNotificationMarshaller.cs index f4d5ecb1d0..05678ff6f7 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/MessageDispatchNotificationMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/MessageDispatchNotificationMarshaller.cs @@ -71,10 +71,10 @@ namespace OpenWire.Client.IO MessageDispatchNotification info = (MessageDispatchNotification)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1CachedObject(wireFormat, info.ConsumerId, bs); - rc += Marshal1CachedObject(wireFormat, info.Destination, bs); - rc += Marshal1Long(wireFormat, info.DeliverySequenceId, bs); - rc += Marshal1NestedObject(wireFormat, info.MessageId, bs); + rc += Marshal1CachedObject(wireFormat, info.ConsumerId, bs); + rc += Marshal1CachedObject(wireFormat, info.Destination, bs); + rc += Marshal1Long(wireFormat, info.DeliverySequenceId, bs); + rc += Marshal1NestedObject(wireFormat, info.MessageId, bs); return rc + 0; } @@ -86,10 +86,10 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); MessageDispatchNotification info = (MessageDispatchNotification)o; - Marshal2CachedObject(wireFormat, info.ConsumerId, dataOut, bs); - Marshal2CachedObject(wireFormat, info.Destination, dataOut, bs); - Marshal2Long(wireFormat, info.DeliverySequenceId, dataOut, bs); - Marshal2NestedObject(wireFormat, info.MessageId, dataOut, bs); + Marshal2CachedObject(wireFormat, info.ConsumerId, dataOut, bs); + Marshal2CachedObject(wireFormat, info.Destination, dataOut, bs); + Marshal2Long(wireFormat, info.DeliverySequenceId, dataOut, bs); + Marshal2NestedObject(wireFormat, info.MessageId, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/MessageIdMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/MessageIdMarshaller.cs index 8829779977..0d75b0d765 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/MessageIdMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/MessageIdMarshaller.cs @@ -70,9 +70,9 @@ namespace OpenWire.Client.IO MessageId info = (MessageId)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1CachedObject(wireFormat, info.ProducerId, bs); - rc += Marshal1Long(wireFormat, info.ProducerSequenceId, bs); - rc += Marshal1Long(wireFormat, info.BrokerSequenceId, bs); + rc += Marshal1CachedObject(wireFormat, info.ProducerId, bs); + rc += Marshal1Long(wireFormat, info.ProducerSequenceId, bs); + rc += Marshal1Long(wireFormat, info.BrokerSequenceId, bs); return rc + 0; } @@ -84,9 +84,9 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); MessageId info = (MessageId)o; - Marshal2CachedObject(wireFormat, info.ProducerId, dataOut, bs); - Marshal2Long(wireFormat, info.ProducerSequenceId, dataOut, bs); - Marshal2Long(wireFormat, info.BrokerSequenceId, dataOut, bs); + Marshal2CachedObject(wireFormat, info.ProducerId, dataOut, bs); + Marshal2Long(wireFormat, info.ProducerSequenceId, dataOut, bs); + Marshal2Long(wireFormat, info.BrokerSequenceId, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/MessageMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/MessageMarshaller.cs index 0c021f7bcf..547caa512e 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/MessageMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/MessageMarshaller.cs @@ -92,30 +92,30 @@ namespace OpenWire.Client.IO Message info = (Message)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1CachedObject(wireFormat, info.ProducerId, bs); - rc += Marshal1CachedObject(wireFormat, info.Destination, bs); - rc += Marshal1CachedObject(wireFormat, info.TransactionId, bs); - rc += Marshal1CachedObject(wireFormat, info.OriginalDestination, bs); - rc += Marshal1NestedObject(wireFormat, info.MessageId, bs); - rc += Marshal1CachedObject(wireFormat, info.OriginalTransactionId, bs); - rc += WriteString(info.GroupID, bs); - rc += WriteString(info.CorrelationId, bs); - bs.WriteBoolean(info.Persistent); - rc += Marshal1Long(wireFormat, info.Expiration, bs); - rc += Marshal1NestedObject(wireFormat, info.ReplyTo, bs); - rc += Marshal1Long(wireFormat, info.Timestamp, bs); - rc += WriteString(info.Type, bs); - bs.WriteBoolean(info.Content!=null); + rc += Marshal1CachedObject(wireFormat, info.ProducerId, bs); + rc += Marshal1CachedObject(wireFormat, info.Destination, bs); + rc += Marshal1CachedObject(wireFormat, info.TransactionId, bs); + rc += Marshal1CachedObject(wireFormat, info.OriginalDestination, bs); + rc += Marshal1NestedObject(wireFormat, info.MessageId, bs); + rc += Marshal1CachedObject(wireFormat, info.OriginalTransactionId, bs); + rc += WriteString(info.GroupID, bs); + rc += WriteString(info.CorrelationId, bs); + bs.WriteBoolean(info.Persistent); + rc += Marshal1Long(wireFormat, info.Expiration, bs); + rc += Marshal1NestedObject(wireFormat, info.ReplyTo, bs); + rc += Marshal1Long(wireFormat, info.Timestamp, bs); + rc += WriteString(info.Type, bs); + bs.WriteBoolean(info.Content!=null); rc += info.Content==null ? 0 : info.Content.Length+4; - bs.WriteBoolean(info.MarshalledProperties!=null); + bs.WriteBoolean(info.MarshalledProperties!=null); rc += info.MarshalledProperties==null ? 0 : info.MarshalledProperties.Length+4; - rc += Marshal1NestedObject(wireFormat, info.DataStructure, bs); - rc += Marshal1CachedObject(wireFormat, info.TargetConsumerId, bs); - bs.WriteBoolean(info.Compressed); - rc += MarshalObjectArray(wireFormat, info.BrokerPath, bs); - rc += Marshal1Long(wireFormat, info.Arrival, bs); - rc += WriteString(info.UserID, bs); - bs.WriteBoolean(info.RecievedByDFBridge); + rc += Marshal1NestedObject(wireFormat, info.DataStructure, bs); + rc += Marshal1CachedObject(wireFormat, info.TargetConsumerId, bs); + bs.WriteBoolean(info.Compressed); + rc += MarshalObjectArray(wireFormat, info.BrokerPath, bs); + rc += Marshal1Long(wireFormat, info.Arrival, bs); + rc += WriteString(info.UserID, bs); + bs.WriteBoolean(info.RecievedByDFBridge); return rc + 3; } @@ -127,37 +127,37 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); Message info = (Message)o; - Marshal2CachedObject(wireFormat, info.ProducerId, dataOut, bs); - Marshal2CachedObject(wireFormat, info.Destination, dataOut, bs); - Marshal2CachedObject(wireFormat, info.TransactionId, dataOut, bs); - Marshal2CachedObject(wireFormat, info.OriginalDestination, dataOut, bs); - Marshal2NestedObject(wireFormat, info.MessageId, dataOut, bs); - Marshal2CachedObject(wireFormat, info.OriginalTransactionId, dataOut, bs); - WriteString(info.GroupID, dataOut, bs); - DataStreamMarshaller.WriteInt(info.GroupSequence, dataOut); - WriteString(info.CorrelationId, dataOut, bs); - bs.ReadBoolean(); - Marshal2Long(wireFormat, info.Expiration, dataOut, bs); - DataStreamMarshaller.WriteByte(info.Priority, dataOut); - Marshal2NestedObject(wireFormat, info.ReplyTo, dataOut, bs); - Marshal2Long(wireFormat, info.Timestamp, dataOut, bs); - WriteString(info.Type, dataOut, bs); - if(bs.ReadBoolean()) { + Marshal2CachedObject(wireFormat, info.ProducerId, dataOut, bs); + Marshal2CachedObject(wireFormat, info.Destination, dataOut, bs); + Marshal2CachedObject(wireFormat, info.TransactionId, dataOut, bs); + Marshal2CachedObject(wireFormat, info.OriginalDestination, dataOut, bs); + Marshal2NestedObject(wireFormat, info.MessageId, dataOut, bs); + Marshal2CachedObject(wireFormat, info.OriginalTransactionId, dataOut, bs); + WriteString(info.GroupID, dataOut, bs); + DataStreamMarshaller.WriteInt(info.GroupSequence, dataOut); + WriteString(info.CorrelationId, dataOut, bs); + bs.ReadBoolean(); + Marshal2Long(wireFormat, info.Expiration, dataOut, bs); + DataStreamMarshaller.WriteByte(info.Priority, dataOut); + Marshal2NestedObject(wireFormat, info.ReplyTo, dataOut, bs); + Marshal2Long(wireFormat, info.Timestamp, dataOut, bs); + WriteString(info.Type, dataOut, bs); + if(bs.ReadBoolean()) { DataStreamMarshaller.WriteInt(info.Content.Length, dataOut); dataOut.Write(info.Content); } - if(bs.ReadBoolean()) { + if(bs.ReadBoolean()) { DataStreamMarshaller.WriteInt(info.MarshalledProperties.Length, dataOut); dataOut.Write(info.MarshalledProperties); } - Marshal2NestedObject(wireFormat, info.DataStructure, dataOut, bs); - Marshal2CachedObject(wireFormat, info.TargetConsumerId, dataOut, bs); - bs.ReadBoolean(); - DataStreamMarshaller.WriteInt(info.RedeliveryCounter, dataOut); - MarshalObjectArray(wireFormat, info.BrokerPath, dataOut, bs); - Marshal2Long(wireFormat, info.Arrival, dataOut, bs); - WriteString(info.UserID, dataOut, bs); - bs.ReadBoolean(); + Marshal2NestedObject(wireFormat, info.DataStructure, dataOut, bs); + Marshal2CachedObject(wireFormat, info.TargetConsumerId, dataOut, bs); + bs.ReadBoolean(); + DataStreamMarshaller.WriteInt(info.RedeliveryCounter, dataOut); + MarshalObjectArray(wireFormat, info.BrokerPath, dataOut, bs); + Marshal2Long(wireFormat, info.Arrival, dataOut, bs); + WriteString(info.UserID, dataOut, bs); + bs.ReadBoolean(); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/ProducerIdMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/ProducerIdMarshaller.cs index b75bf022c3..94b414ddf5 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/ProducerIdMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/ProducerIdMarshaller.cs @@ -70,9 +70,9 @@ namespace OpenWire.Client.IO ProducerId info = (ProducerId)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += WriteString(info.ConnectionId, bs); - rc += Marshal1Long(wireFormat, info.Value, bs); - rc += Marshal1Long(wireFormat, info.SessionId, bs); + rc += WriteString(info.ConnectionId, bs); + rc += Marshal1Long(wireFormat, info.Value, bs); + rc += Marshal1Long(wireFormat, info.SessionId, bs); return rc + 0; } @@ -84,9 +84,9 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); ProducerId info = (ProducerId)o; - WriteString(info.ConnectionId, dataOut, bs); - Marshal2Long(wireFormat, info.Value, dataOut, bs); - Marshal2Long(wireFormat, info.SessionId, dataOut, bs); + WriteString(info.ConnectionId, dataOut, bs); + Marshal2Long(wireFormat, info.Value, dataOut, bs); + Marshal2Long(wireFormat, info.SessionId, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/ProducerInfoMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/ProducerInfoMarshaller.cs index 99403c7454..36a4cfe94d 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/ProducerInfoMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/ProducerInfoMarshaller.cs @@ -81,9 +81,9 @@ namespace OpenWire.Client.IO ProducerInfo info = (ProducerInfo)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1CachedObject(wireFormat, info.ProducerId, bs); - rc += Marshal1CachedObject(wireFormat, info.Destination, bs); - rc += MarshalObjectArray(wireFormat, info.BrokerPath, bs); + rc += Marshal1CachedObject(wireFormat, info.ProducerId, bs); + rc += Marshal1CachedObject(wireFormat, info.Destination, bs); + rc += MarshalObjectArray(wireFormat, info.BrokerPath, bs); return rc + 0; } @@ -95,9 +95,9 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); ProducerInfo info = (ProducerInfo)o; - Marshal2CachedObject(wireFormat, info.ProducerId, dataOut, bs); - Marshal2CachedObject(wireFormat, info.Destination, dataOut, bs); - MarshalObjectArray(wireFormat, info.BrokerPath, dataOut, bs); + Marshal2CachedObject(wireFormat, info.ProducerId, dataOut, bs); + Marshal2CachedObject(wireFormat, info.Destination, dataOut, bs); + MarshalObjectArray(wireFormat, info.BrokerPath, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/RemoveInfoMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/RemoveInfoMarshaller.cs index ee6c9e6161..d597702960 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/RemoveInfoMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/RemoveInfoMarshaller.cs @@ -68,7 +68,7 @@ namespace OpenWire.Client.IO RemoveInfo info = (RemoveInfo)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1CachedObject(wireFormat, info.ObjectId, bs); + rc += Marshal1CachedObject(wireFormat, info.ObjectId, bs); return rc + 0; } @@ -80,7 +80,7 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); RemoveInfo info = (RemoveInfo)o; - Marshal2CachedObject(wireFormat, info.ObjectId, dataOut, bs); + Marshal2CachedObject(wireFormat, info.ObjectId, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/RemoveSubscriptionInfoMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/RemoveSubscriptionInfoMarshaller.cs index 14dd782428..060ff25ec2 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/RemoveSubscriptionInfoMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/RemoveSubscriptionInfoMarshaller.cs @@ -70,9 +70,9 @@ namespace OpenWire.Client.IO RemoveSubscriptionInfo info = (RemoveSubscriptionInfo)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1CachedObject(wireFormat, info.ConnectionId, bs); - rc += WriteString(info.SubcriptionName, bs); - rc += WriteString(info.ClientId, bs); + rc += Marshal1CachedObject(wireFormat, info.ConnectionId, bs); + rc += WriteString(info.SubcriptionName, bs); + rc += WriteString(info.ClientId, bs); return rc + 0; } @@ -84,9 +84,9 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); RemoveSubscriptionInfo info = (RemoveSubscriptionInfo)o; - Marshal2CachedObject(wireFormat, info.ConnectionId, dataOut, bs); - WriteString(info.SubcriptionName, dataOut, bs); - WriteString(info.ClientId, dataOut, bs); + Marshal2CachedObject(wireFormat, info.ConnectionId, dataOut, bs); + WriteString(info.SubcriptionName, dataOut, bs); + WriteString(info.ClientId, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/ResponseMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/ResponseMarshaller.cs index 012c2122c7..8ea4683fe9 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/ResponseMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/ResponseMarshaller.cs @@ -68,7 +68,7 @@ namespace OpenWire.Client.IO Response info = (Response)o; int rc = base.Marshal1(wireFormat, info, bs); - + return rc + 1; } @@ -79,7 +79,7 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); Response info = (Response)o; - DataStreamMarshaller.WriteShort(info.CorrelationId, dataOut); + DataStreamMarshaller.WriteShort(info.CorrelationId, dataOut); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/SessionIdMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/SessionIdMarshaller.cs index 3d00c517a9..f74166c8f1 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/SessionIdMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/SessionIdMarshaller.cs @@ -69,8 +69,8 @@ namespace OpenWire.Client.IO SessionId info = (SessionId)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += WriteString(info.ConnectionId, bs); - rc += Marshal1Long(wireFormat, info.Value, bs); + rc += WriteString(info.ConnectionId, bs); + rc += Marshal1Long(wireFormat, info.Value, bs); return rc + 0; } @@ -82,8 +82,8 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); SessionId info = (SessionId)o; - WriteString(info.ConnectionId, dataOut, bs); - Marshal2Long(wireFormat, info.Value, dataOut, bs); + WriteString(info.ConnectionId, dataOut, bs); + Marshal2Long(wireFormat, info.Value, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/SessionInfoMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/SessionInfoMarshaller.cs index 896d0ae710..f77051703d 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/SessionInfoMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/SessionInfoMarshaller.cs @@ -68,7 +68,7 @@ namespace OpenWire.Client.IO SessionInfo info = (SessionInfo)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1CachedObject(wireFormat, info.SessionId, bs); + rc += Marshal1CachedObject(wireFormat, info.SessionId, bs); return rc + 0; } @@ -80,7 +80,7 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); SessionInfo info = (SessionInfo)o; - Marshal2CachedObject(wireFormat, info.SessionId, dataOut, bs); + Marshal2CachedObject(wireFormat, info.SessionId, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/SubscriptionInfoMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/SubscriptionInfoMarshaller.cs index 71e46804ea..f8fc5034f3 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/SubscriptionInfoMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/SubscriptionInfoMarshaller.cs @@ -71,10 +71,10 @@ namespace OpenWire.Client.IO SubscriptionInfo info = (SubscriptionInfo)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += WriteString(info.ClientId, bs); - rc += Marshal1CachedObject(wireFormat, info.Destination, bs); - rc += WriteString(info.Selector, bs); - rc += WriteString(info.SubcriptionName, bs); + rc += WriteString(info.ClientId, bs); + rc += Marshal1CachedObject(wireFormat, info.Destination, bs); + rc += WriteString(info.Selector, bs); + rc += WriteString(info.SubcriptionName, bs); return rc + 0; } @@ -86,10 +86,10 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); SubscriptionInfo info = (SubscriptionInfo)o; - WriteString(info.ClientId, dataOut, bs); - Marshal2CachedObject(wireFormat, info.Destination, dataOut, bs); - WriteString(info.Selector, dataOut, bs); - WriteString(info.SubcriptionName, dataOut, bs); + WriteString(info.ClientId, dataOut, bs); + Marshal2CachedObject(wireFormat, info.Destination, dataOut, bs); + WriteString(info.Selector, dataOut, bs); + WriteString(info.SubcriptionName, dataOut, bs); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/TransactionInfoMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/TransactionInfoMarshaller.cs index 3cbf95b131..18120baeb9 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/TransactionInfoMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/TransactionInfoMarshaller.cs @@ -70,9 +70,9 @@ namespace OpenWire.Client.IO TransactionInfo info = (TransactionInfo)o; int rc = base.Marshal1(wireFormat, info, bs); - rc += Marshal1CachedObject(wireFormat, info.ConnectionId, bs); - rc += Marshal1CachedObject(wireFormat, info.TransactionId, bs); - + rc += Marshal1CachedObject(wireFormat, info.ConnectionId, bs); + rc += Marshal1CachedObject(wireFormat, info.TransactionId, bs); + return rc + 1; } @@ -83,9 +83,9 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); TransactionInfo info = (TransactionInfo)o; - Marshal2CachedObject(wireFormat, info.ConnectionId, dataOut, bs); - Marshal2CachedObject(wireFormat, info.TransactionId, dataOut, bs); - DataStreamMarshaller.WriteByte(info.Type, dataOut); + Marshal2CachedObject(wireFormat, info.ConnectionId, dataOut, bs); + Marshal2CachedObject(wireFormat, info.TransactionId, dataOut, bs); + DataStreamMarshaller.WriteByte(info.Type, dataOut); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/WireFormatInfoMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/WireFormatInfoMarshaller.cs index f50f908d61..8f93d48fe7 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/WireFormatInfoMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/WireFormatInfoMarshaller.cs @@ -73,10 +73,10 @@ namespace OpenWire.Client.IO WireFormatInfo info = (WireFormatInfo)o; int rc = base.Marshal1(wireFormat, info, bs); - bs.WriteBoolean(info.CacheEnabled); - bs.WriteBoolean(info.CompressionEnabled); - bs.WriteBoolean(info.StackTraceEnabled); - bs.WriteBoolean(info.TcpNoDelayEnabled); + bs.WriteBoolean(info.CacheEnabled); + bs.WriteBoolean(info.CompressionEnabled); + bs.WriteBoolean(info.StackTraceEnabled); + bs.WriteBoolean(info.TcpNoDelayEnabled); return rc + 9; } @@ -88,12 +88,12 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); WireFormatInfo info = (WireFormatInfo)o; - dataOut.Write(info.Magic, 0, 8); - DataStreamMarshaller.WriteInt(info.Version, dataOut); - bs.ReadBoolean(); - bs.ReadBoolean(); - bs.ReadBoolean(); - bs.ReadBoolean(); + dataOut.Write(info.Magic, 0, 8); + DataStreamMarshaller.WriteInt(info.Version, dataOut); + bs.ReadBoolean(); + bs.ReadBoolean(); + bs.ReadBoolean(); + bs.ReadBoolean(); } } diff --git a/openwire-dotnet/src/OpenWire.Client/IO/XATransactionIdMarshaller.cs b/openwire-dotnet/src/OpenWire.Client/IO/XATransactionIdMarshaller.cs index 545555e09e..a6cbaf6686 100644 --- a/openwire-dotnet/src/OpenWire.Client/IO/XATransactionIdMarshaller.cs +++ b/openwire-dotnet/src/OpenWire.Client/IO/XATransactionIdMarshaller.cs @@ -70,9 +70,9 @@ namespace OpenWire.Client.IO XATransactionId info = (XATransactionId)o; int rc = base.Marshal1(wireFormat, info, bs); - bs.WriteBoolean(info.GlobalTransactionId!=null); + bs.WriteBoolean(info.GlobalTransactionId!=null); rc += info.GlobalTransactionId==null ? 0 : info.GlobalTransactionId.Length+4; - bs.WriteBoolean(info.BranchQualifier!=null); + bs.WriteBoolean(info.BranchQualifier!=null); rc += info.BranchQualifier==null ? 0 : info.BranchQualifier.Length+4; return rc + 1; @@ -85,12 +85,12 @@ namespace OpenWire.Client.IO base.Marshal2(wireFormat, o, dataOut, bs); XATransactionId info = (XATransactionId)o; - DataStreamMarshaller.WriteInt(info.FormatId, dataOut); - if(bs.ReadBoolean()) { + DataStreamMarshaller.WriteInt(info.FormatId, dataOut); + if(bs.ReadBoolean()) { DataStreamMarshaller.WriteInt(info.GlobalTransactionId.Length, dataOut); dataOut.Write(info.GlobalTransactionId); } - if(bs.ReadBoolean()) { + if(bs.ReadBoolean()) { DataStreamMarshaller.WriteInt(info.BranchQualifier.Length, dataOut); dataOut.Write(info.BranchQualifier); } diff --git a/openwire-dotnet/src/OpenWire.Client/ISession.cs b/openwire-dotnet/src/OpenWire.Client/ISession.cs index 216b9bdd67..b19e487df8 100755 --- a/openwire-dotnet/src/OpenWire.Client/ISession.cs +++ b/openwire-dotnet/src/OpenWire.Client/ISession.cs @@ -104,6 +104,21 @@ namespace OpenWire.Client /// Creates a new binary message with the given body /// IBytesMessage CreateBytesMessage(byte[] body); + + + // Transaction methods + + /// + /// If this is a transactional session then commit all message + /// send and acknowledgements for producers and consumers in this session + /// + void Commit(); + + /// + /// If this is a transactional session then rollback all message + /// send and acknowledgements for producers and consumers in this session + /// + void Rollback(); } } diff --git a/openwire-dotnet/src/OpenWire.Client/MessageConsumer.cs b/openwire-dotnet/src/OpenWire.Client/MessageConsumer.cs index d3d928c770..59620acb67 100755 --- a/openwire-dotnet/src/OpenWire.Client/MessageConsumer.cs +++ b/openwire-dotnet/src/OpenWire.Client/MessageConsumer.cs @@ -22,10 +22,11 @@ using OpenWire.Client.Core; namespace OpenWire.Client { - public enum AckType { + public enum AckType + { DeliveredAck = 0, // Message delivered but not consumed - ConsumedAck = 1, // Message consumed, discard - PoisonAck = 2 // Message could not be processed due to poison pill but discard anyway + PoisonAck = 1, // Message could not be processed due to poison pill but discard anyway + ConsumedAck = 2 // Message consumed, discard } @@ -40,9 +41,12 @@ namespace OpenWire.Client private AcknowledgementMode acknowledgementMode; private bool closed; private Dispatcher dispatcher = new Dispatcher(); + private int maximumRedeliveryCount = 10; + private int redeliveryTimeout = 500; public event MessageListener Listener; + public MessageConsumer(Session session, ConsumerInfo info, AcknowledgementMode acknowledgementMode) { this.session = session; @@ -50,12 +54,29 @@ namespace OpenWire.Client this.acknowledgementMode = acknowledgementMode; } - public ConsumerId ConsumerId { + public ConsumerId ConsumerId + { get { return info.ConsumerId; } } - + + public int MaximumRedeliveryCount + { + get { return maximumRedeliveryCount; } + set { maximumRedeliveryCount = value; } + } + + public int RedeliveryTimeout + { + get { return redeliveryTimeout; } + set { redeliveryTimeout = value; } + } + + public void RedeliverRolledBackMessages() + { + dispatcher.RedeliverRolledBackMessages(); + } /// /// Method Dispatch @@ -65,7 +86,8 @@ namespace OpenWire.Client { dispatcher.Enqueue(message); - if (Listener != null) { + if (Listener != null) + { // lets dispatch to the thread pool for this connection for messages to be processed ThreadPool.QueueUserWorkItem(new WaitCallback(session.DispatchAsyncMessages)); } @@ -77,7 +99,7 @@ namespace OpenWire.Client return AutoAcknowledge(dispatcher.Dequeue()); } - public IMessage Receive(long timeout) + public IMessage Receive(int timeout) { CheckClosed(); return AutoAcknowledge(dispatcher.Dequeue(timeout)); @@ -102,12 +124,15 @@ namespace OpenWire.Client /// public void DispatchAsyncMessages() { - while (Listener != null) { + while (Listener != null) + { IMessage message = dispatcher.DequeueNoWait(); - if (message != null) { + if (message != null) + { Listener(message); } - else { + else + { break; } } @@ -145,7 +170,7 @@ namespace OpenWire.Client DoAcknowledge(message); } } - + protected void DoAcknowledge(Message message) { MessageAck ack = CreateMessageAck(message); @@ -163,10 +188,70 @@ namespace OpenWire.Client ack.FirstMessageId = message.MessageId; ack.LastMessageId = message.MessageId; ack.MessageCount = 1; - ack.TransactionId = message.TransactionId; + + if (session.Transacted) + { + session.DoStartTransaction(); + ack.TransactionId = session.TransactionContext.TransactionId; + session.TransactionContext.AddSynchronization(new MessageConsumerSynchronization(this, message)); + } return ack; } + public void AfterRollback(ActiveMQMessage message) + { + // lets redeliver the message again + message.RedeliveryCounter += 1; + if (message.RedeliveryCounter > MaximumRedeliveryCount) + { + // lets send back a poisoned pill + MessageAck ack = new MessageAck(); + ack.AckType = (int) AckType.PoisonAck; + ack.ConsumerId = info.ConsumerId; + ack.Destination = message.Destination; + ack.FirstMessageId = message.MessageId; + ack.LastMessageId = message.MessageId; + ack.MessageCount = 1; + session.Connection.OneWay(ack); + } + else + { + dispatcher.Redeliver(message); + + if (Listener != null) + { + // lets re-dispatch the message at some point in the future + Thread.Sleep(RedeliveryTimeout); + ThreadPool.QueueUserWorkItem(new WaitCallback(session.DispatchAsyncMessages)); + } + } + } + } + + // TODO maybe there's a cleaner way of creating stateful delegates to make this code neater + class MessageConsumerSynchronization : ISynchronization + { + private MessageConsumer consumer; + private Message message; + + public MessageConsumerSynchronization(MessageConsumer consumer, Message message) + { + this.message = message; + this.consumer = consumer; + } + + public void BeforeCommit() + { + } + + public void AfterCommit() + { + } + + public void AfterRollback() + { + consumer.AfterRollback((ActiveMQMessage) message); + } } } diff --git a/openwire-dotnet/src/OpenWire.Client/MessageProducer.cs b/openwire-dotnet/src/OpenWire.Client/MessageProducer.cs index 3496e3eb31..0761650946 100755 --- a/openwire-dotnet/src/OpenWire.Client/MessageProducer.cs +++ b/openwire-dotnet/src/OpenWire.Client/MessageProducer.cs @@ -29,11 +29,11 @@ namespace OpenWire.Client private ProducerInfo info; private long messageCounter; - bool persistent; - long timeToLive; - int priority; - bool disableMessageID; - bool disableMessageTimestamp; + bool persistent; + long timeToLive; + int priority; + bool disableMessageID; + bool disableMessageTimestamp; public MessageProducer(Session session, ProducerInfo info) { @@ -59,42 +59,48 @@ namespace OpenWire.Client activeMessage.ProducerId = info.ProducerId; activeMessage.Destination = ActiveMQDestination.Transform(destination); + if (session.Transacted) + { + session.DoStartTransaction(); + activeMessage.TransactionId = session.TransactionContext.TransactionId; + } + session.DoSend(destination, message); } public void Dispose() { - session.DisposeOf(info.ProducerId); + session.Connection.DisposeOf(info.ProducerId); } - - public bool Persistent - { - get { return persistent; } - set { this.persistent = value; } - } - - public long TimeToLive - { - get { return timeToLive; } - set { this.timeToLive = value; } -} - public int Priority - { - get { return priority; } - set { this.priority = value; } - } - - public bool DisableMessageID - { - get { return disableMessageID; } - set { this.disableMessageID = value; } - } - - public bool DisableMessageTimestamp - { - get { return disableMessageTimestamp; } - set { this.disableMessageTimestamp = value; } - } - + + public bool Persistent + { + get { return persistent; } + set { this.persistent = value; } + } + + public long TimeToLive + { + get { return timeToLive; } + set { this.timeToLive = value; } + } + public int Priority + { + get { return priority; } + set { this.priority = value; } + } + + public bool DisableMessageID + { + get { return disableMessageID; } + set { this.disableMessageID = value; } + } + + public bool DisableMessageTimestamp + { + get { return disableMessageTimestamp; } + set { this.disableMessageTimestamp = value; } + } + } } diff --git a/openwire-dotnet/src/OpenWire.Client/Session.cs b/openwire-dotnet/src/OpenWire.Client/Session.cs index 01aad0474c..7f3e2dc10f 100755 --- a/openwire-dotnet/src/OpenWire.Client/Session.cs +++ b/openwire-dotnet/src/OpenWire.Client/Session.cs @@ -33,17 +33,20 @@ namespace OpenWire.Client private long producerCounter; private int prefetchSize = 1000; private IDictionary consumers = new Hashtable(); + private TransactionContext transactionContext; public Session(Connection connection, SessionInfo info, AcknowledgementMode acknowledgementMode) { this.connection = connection; this.info = info; this.acknowledgementMode = acknowledgementMode; + transactionContext = new TransactionContext(this); } + public void Dispose() { - DisposeOf(info.SessionId); + connection.DisposeOf(info.SessionId); } public IMessageProducer CreateProducer() @@ -174,13 +177,52 @@ namespace OpenWire.Client return answer; } + public void Commit() + { + if (! Transacted) + { + throw new InvalidOperationException("You cannot perform a Commit() on a non-transacted session. Acknowlegement mode is: " + acknowledgementMode); + } + transactionContext.Commit(); + } + + public void Rollback() + { + if (! Transacted) + { + throw new InvalidOperationException("You cannot perform a Commit() on a non-transacted session. Acknowlegement mode is: " + acknowledgementMode); + } + transactionContext.Rollback(); + + // lets ensure all the consumers redeliver any rolled back messages + foreach (MessageConsumer consumer in consumers.Values) + { + consumer.RedeliverRolledBackMessages(); + } + } + + // Properties + public Connection Connection { - get { - return connection; - } + get { return connection; } + } + + public SessionId SessionId + { + get { return info.SessionId; } + } + + public bool Transacted + { + get { return acknowledgementMode == AcknowledgementMode.Transactional; } + } + + public TransactionContext TransactionContext + { + get { return transactionContext; } } // Implementation methods @@ -191,21 +233,22 @@ namespace OpenWire.Client connection.SyncRequest(command); } - public void DisposeOf(DataStructure objectId) + /// + /// Ensures that a transaction is started + /// + public void DoStartTransaction() { - // TODO dispose of all the session first? - RemoveInfo command = new RemoveInfo(); - command.ObjectId = objectId; - connection.SyncRequest(command); + if (Transacted) + { + transactionContext.Begin(); + } } public void DisposeOf(ConsumerId objectId) { consumers.Remove(objectId); connection.RemoveConsumer(objectId); - RemoveInfo command = new RemoveInfo(); - command.ObjectId = objectId; - connection.SyncRequest(command); + connection.DisposeOf(objectId); } public void DispatchAsyncMessages(object state) diff --git a/openwire-dotnet/tests/OpenWire.Client/ConsumerTest.cs b/openwire-dotnet/tests/OpenWire.Client/ConsumerTest.cs index 12cfc70d69..239485da49 100755 --- a/openwire-dotnet/tests/OpenWire.Client/ConsumerTest.cs +++ b/openwire-dotnet/tests/OpenWire.Client/ConsumerTest.cs @@ -27,65 +27,65 @@ namespace OpenWire.Client [TestFixture] public class ConsumerTest : TestSupport { - IConnectionFactory factory; - IConnection connection; - IDestination destination; + IConnectionFactory factory; + IConnection connection; + IDestination destination; - [SetUp] - protected void SetUp() - { - factory = new ConnectionFactory("localhost", 61616); - connection = factory.CreateConnection(); - } + [SetUp] + protected void SetUp() + { + factory = new ConnectionFactory("localhost", 61616); + connection = factory.CreateConnection(); + } - [TearDown] - protected void TearDown() - { - connection.Dispose(); - } + [TearDown] + protected void TearDown() + { + connection.Dispose(); + } [Test] - [Ignore("Not fully implemented yet.")] - public void testDurableConsumerSelectorChange() - { + [Ignore("Not fully implemented yet.")] + public void testDurableConsumerSelectorChange() + { - // Receive a message with the JMS API - connection.ClientId="test"; - connection.Start(); - - ISession session = connection.CreateSession(false, AcknowledgementMode.AutoAcknowledge); - destination = session.GetTopic("foo"); - IMessageProducer producer = session.CreateProducer(destination); - producer.Persistent = true; - IMessageConsumer consumer = session.CreateDurableConsumer((ITopic)destination, "test", "color='red'", false); + // Receive a message with the JMS API + connection.ClientId="test"; + connection.Start(); + + ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge); + destination = session.GetTopic("foo"); + IMessageProducer producer = session.CreateProducer(destination); + producer.Persistent = true; + IMessageConsumer consumer = session.CreateDurableConsumer((ITopic)destination, "test", "color='red'", false); - // Send the messages - ITextMessage message = session.CreateTextMessage("1st"); - //message.SetStringProperty("color", "red"); - producer.Send(message); - - IMessage m = consumer.Receive(1000); - Assert.IsNotNull(m); - Assert.AreEqual("1st", ((ITextMessage)m).Text ); + // Send the messages + ITextMessage message = session.CreateTextMessage("1st"); + //message.SetStringProperty("color", "red"); + producer.Send(message); + + IMessage m = consumer.Receive(1000); + Assert.IsNotNull(m); + Assert.AreEqual("1st", ((ITextMessage)m).Text ); - // Change the subscription. - consumer.Dispose(); - consumer = session.CreateDurableConsumer((ITopic)destination, "test", "color='blue'", false); - - message = session.CreateTextMessage("2nd"); - // message.setStringProperty("color", "red"); - producer.Send(message); - message = session.CreateTextMessage("3rd"); - // message.setStringProperty("color", "blue"); - producer.Send(message); + // Change the subscription. + consumer.Dispose(); + consumer = session.CreateDurableConsumer((ITopic)destination, "test", "color='blue'", false); + + message = session.CreateTextMessage("2nd"); + // message.setStringProperty("color", "red"); + producer.Send(message); + message = session.CreateTextMessage("3rd"); + // message.setStringProperty("color", "blue"); + producer.Send(message); - // Selector should skip the 2nd message. - m = consumer.Receive(1000); - Assert.IsNotNull(m); - Assert.AreEqual("3rd", ((ITextMessage)m).Text); - - Assert.IsNull(consumer.ReceiveNoWait()); - } + // Selector should skip the 2nd message. + m = consumer.Receive(1000); + Assert.IsNotNull(m); + Assert.AreEqual("3rd", ((ITextMessage)m).Text); + + Assert.IsNull(consumer.ReceiveNoWait()); + } } } diff --git a/openwire-dotnet/tests/OpenWire.Client/TestSupport.cs b/openwire-dotnet/tests/OpenWire.Client/TestSupport.cs index 05a23851ef..1e82ffa7c2 100644 --- a/openwire-dotnet/tests/OpenWire.Client/TestSupport.cs +++ b/openwire-dotnet/tests/OpenWire.Client/TestSupport.cs @@ -73,12 +73,18 @@ namespace OpenWire.Client protected virtual IDestination CreateDestination(ISession session) { - string name = "Test.DotNet." + GetType().Name; + string name = CreateDestinationName(); IDestination destination = session.GetQueue(name); Console.WriteLine("Using queue: " + destination); return destination; } + + protected virtual string CreateDestinationName() + { + string name = "Test.DotNet." + GetType().Name; + return name; + } protected virtual IMessage CreateMessage(ISession session) { diff --git a/openwire-dotnet/tests/OpenWire.Client/TransactionTest.cs b/openwire-dotnet/tests/OpenWire.Client/TransactionTest.cs new file mode 100644 index 0000000000..2b26bbb3d6 --- /dev/null +++ b/openwire-dotnet/tests/OpenWire.Client/TransactionTest.cs @@ -0,0 +1,290 @@ +/* + * Copyright 2006 The Apache Software Foundation or its licensors, as + * applicable. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +using System; +using System.IO; + +using NUnit.Framework; + +using OpenWire.Client; +using OpenWire.Client.Core; +using System.Collections; + +namespace OpenWire.Client +{ + [TestFixture] + public class TransactionTest : TestSupport + { + private static int destinationCounter; + + IDestination destination; + IConnection connection; + ISession session; + IMessageProducer producer; + IMessageConsumer consumer; + + [SetUp] + public void SetUp() + { + Connect(); + + // lets consume any outstanding messages from previous test runs + while (consumer.Receive(1000) != null) + { + } + session.Commit(); + } + + + + [TearDown] + public void TearDown() + { + Disconnect(); + } + + [Test] + public void TestSendRollback() + { + IMessage[] outbound = new IMessage[]{ + session.CreateTextMessage("First Message"), + session.CreateTextMessage("Second Message") + }; + + //sends a message + producer.Send(outbound[0]); + session.Commit(); + + //sends a message that gets rollbacked + producer.Send(session.CreateTextMessage("I'm going to get rolled back.")); + session.Rollback(); + + //sends a message + producer.Send(outbound[1]); + session.Commit(); + + //receives the first message + ArrayList messages = new ArrayList(); + Console.WriteLine("About to consume message 1"); + IMessage message = consumer.Receive(1000); + messages.Add(message); + Console.WriteLine("Received: " + message); + + //receives the second message + Console.WriteLine("About to consume message 2"); + message = consumer.Receive(4000); + messages.Add(message); + Console.WriteLine("Received: " + message); + + //validates that the rollbacked was not consumed + session.Commit(); + IMessage[] inbound = new IMessage[messages.Count]; + messages.CopyTo(inbound); + AssertTextMessagesEqual("Rollback did not work.", outbound, inbound); + } + + [Test] + public void TestSendSessionClose() + { + IMessage[] outbound = new IMessage[]{ + session.CreateTextMessage("First Message"), + session.CreateTextMessage("Second Message") + }; + + //sends a message + producer.Send(outbound[0]); + session.Commit(); + + //sends a message that gets rollbacked + producer.Send(session.CreateTextMessage("I'm going to get rolled back.")); + consumer.Dispose(); + session.Dispose(); + + Reconnect(); + + //sends a message + producer.Send(outbound[1]); + session.Commit(); + + //receives the first message + ArrayList messages = new ArrayList(); + Console.WriteLine("About to consume message 1"); + IMessage message = consumer.Receive(1000); + messages.Add(message); + Console.WriteLine("Received: " + message); + + //receives the second message + Console.WriteLine("About to consume message 2"); + message = consumer.Receive(4000); + messages.Add(message); + Console.WriteLine("Received: " + message); + + //validates that the rollbacked was not consumed + session.Commit(); + IMessage[] inbound = new IMessage[messages.Count]; + messages.CopyTo(inbound); + AssertTextMessagesEqual("Rollback did not work.", outbound, inbound); + } + + [Test] + public void TestReceiveRollback() + { + IMessage[] outbound = new IMessage[]{ + session.CreateTextMessage("First Message"), + session.CreateTextMessage("Second Message") + }; + + //sent both messages + producer.Send(outbound[0]); + producer.Send(outbound[1]); + session.Commit(); + + Console.WriteLine("Sent 0: " + outbound[0]); + Console.WriteLine("Sent 1: " + outbound[1]); + + ArrayList messages = new ArrayList(); + IMessage message = consumer.Receive(1000); + messages.Add(message); + Assert.AreEqual(outbound[0], message); + session.Commit(); + + // rollback so we can get that last message again. + message = consumer.Receive(1000); + Assert.IsNotNull(message); + Assert.AreEqual(outbound[1], message); + session.Rollback(); + + // Consume again.. the previous message should + // get redelivered. + message = consumer.Receive(5000); + Assert.IsNotNull(message, "Should have re-received the message again!"); + messages.Add(message); + session.Commit(); + + IMessage[] inbound = new IMessage[messages.Count]; + messages.CopyTo(inbound); + AssertTextMessagesEqual("Rollback did not work", outbound, inbound); + } + + + [Test] + public void TestReceiveTwoThenRollback() + { + IMessage[] outbound = new IMessage[]{ + session.CreateTextMessage("First Message"), + session.CreateTextMessage("Second Message") + }; + + producer.Send(outbound[0]); + producer.Send(outbound[1]); + session.Commit(); + + Console.WriteLine("Sent 0: " + outbound[0]); + Console.WriteLine("Sent 1: " + outbound[1]); + + ArrayList messages = new ArrayList(); + IMessage message = consumer.Receive(1000); + AssertTextMessageEqual("first mesage received before rollback", outbound[0], message); + + message = consumer.Receive(1000); + Assert.IsNotNull(message); + AssertTextMessageEqual("second message received before rollback", outbound[1], message); + session.Rollback(); + + // Consume again.. the previous message should + // get redelivered. + message = consumer.Receive(5000); + Assert.IsNotNull(message, "Should have re-received the first message again!"); + messages.Add(message); + AssertTextMessageEqual("first message received after rollback", outbound[0], message); + + message = consumer.Receive(5000); + Assert.IsNotNull(message, "Should have re-received the second message again!"); + messages.Add(message); + AssertTextMessageEqual("second message received after rollback", outbound[1], message); + + Assert.IsNull(consumer.ReceiveNoWait()); + session.Commit(); + + IMessage[] inbound = new IMessage[messages.Count]; + messages.CopyTo(inbound); + AssertTextMessagesEqual("Rollback did not work", outbound, inbound); + } + + protected override string CreateDestinationName() + { + // TODO - how can we get the test name? + return base.CreateDestinationName() + (++destinationCounter); + } + + protected void AssertTextMessagesEqual(String message, IMessage[] expected, IMessage[] actual) + { + Assert.AreEqual(expected.Length, actual.Length, "Incorrect number of messages received"); + + for (int i = 0; i < expected.Length; i++) + { + AssertTextMessageEqual(message + ". Index: " + i, expected[i], actual[i]); + } + } + + protected void AssertTextMessageEqual(String message, IMessage expected, IMessage actual) + { + Assert.IsTrue(expected is ITextMessage, "expected object not a text message"); + Assert.IsTrue(actual is ITextMessage, "actual object not a text message"); + + String expectedText = ((ITextMessage) expected).Text; + String actualText = ((ITextMessage) actual).Text; + + Assert.AreEqual(expectedText, actualText, message); + } + + protected void Connect() + { + IConnectionFactory factory = new ConnectionFactory("localhost", 61616); + + connection = factory.CreateConnection(); + + session = connection.CreateSession(AcknowledgementMode.Transactional); + + // reuse the same destination if we reconnect + if (destination == null) + { + destination = CreateDestination(session); + } + + consumer = session.CreateConsumer(destination); + + producer = session.CreateProducer(destination); + } + + + protected void Disconnect() + { + if (connection != null) + { + connection.Dispose(); + connection = null; + } + } + + protected void Reconnect() + { + Disconnect(); + Connect(); + } + + } +}