You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by jg...@apache.org on 2011/03/09 23:15:40 UTC
svn commit: r1080031 -
/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Connection.cs
Author: jgomes
Date: Wed Mar 9 22:15:40 2011
New Revision: 1080031
URL: http://svn.apache.org/viewvc?rev=1080031&view=rev
Log:
Add synchronizing lock when checking connected status to avoid run-time clashes between threads that start up simultaneously.
Modified:
activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Connection.cs
Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Connection.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Connection.cs?rev=1080031&r1=1080030&r2=1080031&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Connection.cs (original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Connection.cs Wed Mar 9 22:15:40 2011
@@ -45,7 +45,7 @@ namespace Apache.NMS.ActiveMQ
private bool sendAcksAsync = false;
private bool dispatchAsync = true;
private int producerWindowSize = 0;
- private bool messagePrioritySupported=true;
+ private bool messagePrioritySupported = true;
private bool userSpecifiedClientID;
private readonly Uri brokerUri;
@@ -484,52 +484,52 @@ namespace Apache.NMS.ActiveMQ
}
}
- internal void addDispatcher( ConsumerId id, IDispatcher dispatcher )
+ internal void addDispatcher(ConsumerId id, IDispatcher dispatcher)
{
if(!this.closing.Value)
{
- this.dispatchers.Add( id, dispatcher );
+ this.dispatchers.Add(id, dispatcher);
}
}
- internal void removeDispatcher( ConsumerId id )
+ internal void removeDispatcher(ConsumerId id)
{
if(!this.closing.Value)
{
- this.dispatchers.Remove( id );
+ this.dispatchers.Remove(id);
}
}
- internal void addProducer( ProducerId id, MessageProducer producer )
+ internal void addProducer(ProducerId id, MessageProducer producer)
{
if(!this.closing.Value)
{
- this.producers.Add( id, producer );
+ this.producers.Add(id, producer);
}
}
- internal void removeProducer( ProducerId id )
+ internal void removeProducer(ProducerId id)
{
if(!this.closing.Value)
{
- this.producers.Remove( id );
+ this.producers.Remove(id);
}
}
public void Close()
{
- if(!this.closed.Value && !transportFailed.Value)
- {
- this.Stop();
- }
-
+ if(!this.closed.Value && !transportFailed.Value)
+ {
+ this.Stop();
+ }
+
lock(myLock)
{
if(this.closed.Value)
{
return;
}
-
+
try
{
Tracer.Info("Connection.Close(): Closing Connection Now.");
@@ -565,11 +565,11 @@ namespace Apache.NMS.ActiveMQ
}
finally
{
- if(executor != null)
- {
- executor.Shutdown();
- }
-
+ if(executor != null)
+ {
+ executor.Shutdown();
+ }
+
this.transport = null;
this.closed.Value = true;
this.connected.Value = false;
@@ -636,7 +636,7 @@ namespace Apache.NMS.ActiveMQ
Response response = transport.Request(command, requestTimeout);
if(response is ExceptionResponse)
{
- ExceptionResponse exceptionResponse = (ExceptionResponse) response;
+ ExceptionResponse exceptionResponse = (ExceptionResponse)response;
BrokerError brokerError = exceptionResponse.Exception;
throw new BrokerException(brokerError);
}
@@ -695,25 +695,28 @@ namespace Apache.NMS.ActiveMQ
internal void CheckConnected()
{
- if(closed.Value)
- {
- throw new ConnectionClosedException();
- }
-
- if(!connected.Value)
+ lock(myLock)
{
- if(!this.userSpecifiedClientID)
+ if(closed.Value)
{
- this.info.ClientId = this.clientIdGenerator.GenerateId();
+ throw new ConnectionClosedException();
}
- connected.Value = true;
- // now lets send the connection and see if we get an ack/nak
- if(null == SyncRequest(info))
+ if(!connected.Value)
{
- closed.Value = true;
- connected.Value = false;
- throw new ConnectionClosedException();
+ if(!this.userSpecifiedClientID)
+ {
+ this.info.ClientId = this.clientIdGenerator.GenerateId();
+ }
+
+ connected.Value = true;
+ // now lets send the connection and see if we get an ack/nak
+ if(null == SyncRequest(info))
+ {
+ closed.Value = true;
+ connected.Value = false;
+ throw new ConnectionClosedException();
+ }
}
}
}
@@ -728,19 +731,19 @@ namespace Apache.NMS.ActiveMQ
if(command.IsMessageDispatch)
{
WaitForTransportInterruptionProcessingToComplete();
- DispatchMessage((MessageDispatch) command);
+ DispatchMessage((MessageDispatch)command);
}
else if(command.IsKeepAliveInfo)
{
- OnKeepAliveCommand(commandTransport, (KeepAliveInfo) command);
+ OnKeepAliveCommand(commandTransport, (KeepAliveInfo)command);
}
else if(command.IsWireFormatInfo)
{
- this.brokerWireFormatInfo = (WireFormatInfo) command;
+ this.brokerWireFormatInfo = (WireFormatInfo)command;
}
else if(command.IsBrokerInfo)
{
- this.brokerInfo = (BrokerInfo) command;
+ this.brokerInfo = (BrokerInfo)command;
this.brokerInfoReceived.countDown();
}
else if(command.IsShutdownInfo)
@@ -752,7 +755,7 @@ namespace Apache.NMS.ActiveMQ
}
else if(command.IsProducerAck)
{
- ProducerAck ack = (ProducerAck) command as ProducerAck;
+ ProducerAck ack = (ProducerAck)command as ProducerAck;
if(ack.ProducerId != null)
{
MessageProducer producer = producers[ack.ProducerId] as MessageProducer;
@@ -771,7 +774,7 @@ namespace Apache.NMS.ActiveMQ
{
if(!closing.Value && !closed.Value)
{
- ConnectionError connectionError = (ConnectionError) command;
+ ConnectionError connectionError = (ConnectionError)command;
BrokerError brokerError = connectionError.Exception;
string message = "Broker connection error.";
string cause = "";
@@ -800,7 +803,7 @@ namespace Apache.NMS.ActiveMQ
{
if(dispatchers.Contains(dispatch.ConsumerId))
{
- IDispatcher dispatcher = (IDispatcher) dispatchers[dispatch.ConsumerId];
+ IDispatcher dispatcher = (IDispatcher)dispatchers[dispatch.ConsumerId];
// Can be null when a consumer has sent a MessagePull and there was
// no available message at the broker to dispatch or when signalled
@@ -1062,7 +1065,7 @@ namespace Apache.NMS.ActiveMQ
DestinationInfo command = new DestinationInfo();
command.ConnectionId = this.ConnectionId;
command.OperationType = DestinationInfo.REMOVE_OPERATION_TYPE; // 1 is remove
- command.Destination = (ActiveMQDestination) destination;
+ command.Destination = (ActiveMQDestination)destination;
this.Oneway(command);
}