in src/Transport/Failover/FailoverTransport.cs [671:868]
public void Oneway(Command command)
{
Exception error = null;
lock(reconnectMutex)
{
if(command != null && ConnectedTransport == null)
{
if(command.IsShutdownInfo)
{
// Skipping send of ShutdownInfo command when not connected.
return;
}
else if(command.IsRemoveInfo || command.IsMessageAck)
{
stateTracker.Track(command);
// Simulate response to RemoveInfo command or a MessageAck
// since it would be stale at this point.
if(command.ResponseRequired)
{
OnCommand(this, new Response() { CorrelationId = command.CommandId }).GetAsyncResult();
}
return;
}
else if(command.IsMessagePull)
{
// Simulate response to MessagePull if timed as we can't honor that now.
MessagePull pullRequest = command as MessagePull;
if (pullRequest.Timeout != 0)
{
MessageDispatch dispatch = new MessageDispatch();
dispatch.ConsumerId = pullRequest.ConsumerId;
dispatch.Destination = pullRequest.Destination;
OnCommand(this, dispatch).GetAsyncResult();
}
return;
}
}
// Keep trying until the message is sent.
for(int i = 0; !disposed; i++)
{
try
{
// Any Ack that was being sent when the connection dropped is now
// stale so we don't send it here as it would cause an unmatched ack
// on the broker side and probably prevent a consumer from getting
// any new messages.
if(command.IsMessageAck && i > 0)
{
Tracer.Debug("Inflight MessageAck being dropped as stale.");
if(command.ResponseRequired)
{
OnCommand(this, new Response() { CorrelationId = command.CommandId }).GetAsyncResult();
}
return;
}
// Wait for transport to be connected.
ITransport transport = ConnectedTransport;
DateTime start = DateTime.Now;
bool timedout = false;
TimeSpan timewait = TimeSpan.FromMilliseconds(-1);
while(transport == null && !disposed && connectionFailure == null)
{
Tracer.Debug("Waiting for transport to reconnect.");
int elapsed = (int) (DateTime.Now - start).TotalMilliseconds;
if(this.timeout > 0 && elapsed > this.timeout)
{
timedout = true;
Tracer.DebugFormat("FailoverTransport.oneway - timed out after {0} mills", elapsed);
break;
}
if(this.timeout > 0)
{
// Set the timeout for waiting to be at most 100ms past the maximum timeout length.
int remainingTime = (this.timeout - elapsed) + 100;
timewait = TimeSpan.FromMilliseconds(remainingTime);
}
// Release so that the reconnect task can run
try
{
// Wait for something. The mutex will be pulsed if we connect, or are shut down.
Monitor.Wait(reconnectMutex, timewait);
}
catch(ThreadInterruptedException e)
{
Tracer.DebugFormat("Interrupted: {0}", e.Message);
}
transport = ConnectedTransport;
}
if(transport == null)
{
// Previous loop may have exited due to use being disposed.
if(disposed)
{
error = new IOException("Transport disposed.");
}
else if(connectionFailure != null)
{
error = connectionFailure;
}
else if(timedout)
{
error = new IOException("Failover oneway timed out after " + timeout + " milliseconds.");
}
else
{
error = new IOException("Unexpected failure.");
}
break;
}
// If it was a request and it was not being tracked by
// the state tracker, then hold it in the requestMap so
// that we can replay it later.
Tracked tracked = stateTracker.Track(command);
lock(((ICollection) requestMap).SyncRoot)
{
if(tracked != null && tracked.WaitingForResponse)
{
requestMap.Add(command.CommandId, tracked);
}
else if(tracked == null && command.ResponseRequired)
{
requestMap.Add(command.CommandId, command);
}
}
// Send the message.
try
{
transport.Oneway(command);
stateTracker.TrackBack(command);
}
catch(Exception e)
{
// If the command was not tracked.. we will retry in this method
// otherwise we need to trigger a reconnect before returning as
// the transport is failed.
if (tracked == null)
{
// since we will retry in this method.. take it
// out of the request map so that it is not
// sent 2 times on recovery
if(command.ResponseRequired)
{
lock(((ICollection) requestMap).SyncRoot)
{
requestMap.Remove(command.CommandId);
}
}
// Rethrow the exception so it will handled by
// the outer catch
throw;
}
else
{
if (Tracer.IsDebugEnabled)
{
Tracer.DebugFormat("Send Oneway attempt: {0} failed: Message = {1}",
i, e.Message);
Tracer.DebugFormat("Failed Message Was: {0}", command);
}
HandleTransportFailure(e);
}
}
return;
}
catch(Exception e)
{
if (Tracer.IsDebugEnabled)
{
Tracer.DebugFormat("Send Oneway attempt: {0} failed: Message = {1}",
i, e.Message);
Tracer.DebugFormat("Failed Message Was: {0}", command);
}
HandleTransportFailure(e);
}
}
}
if(!disposed)
{
if(error != null)
{
throw error;
}
}
}