in activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java [572:724]
public void oneway(Object o) throws IOException {
Command command = (Command) o;
Exception error = null;
try {
synchronized (reconnectMutex) {
if (command != null && connectedTransport.get() == null) {
if (command.isShutdownInfo()) {
// Skipping send of ShutdownInfo command when not connected.
return;
} else if (command instanceof RemoveInfo || command.isMessageAck()) {
// Simulate response to RemoveInfo command or MessageAck (as it will be stale)
stateTracker.track(command);
if (command.isResponseRequired()) {
Response response = new Response();
response.setCorrelationId(command.getCommandId());
processCommand(response);
}
return;
} else if (command instanceof MessagePull) {
// Simulate response to MessagePull if timed as we can't honor that now.
MessagePull pullRequest = (MessagePull) command;
if (pullRequest.getTimeout() != 0) {
MessageDispatch dispatch = new MessageDispatch();
dispatch.setConsumerId(pullRequest.getConsumerId());
dispatch.setDestination(pullRequest.getDestination());
processCommand(dispatch);
}
return;
}
}
// Keep trying until the message is sent.
for (int i = 0; !disposed; i++) {
try {
// Wait for transport to be connected.
Transport transport = connectedTransport.get();
long start = System.currentTimeMillis();
boolean timedout = false;
while (transport == null && !disposed && connectionFailure == null
&& !Thread.currentThread().isInterrupted() && willReconnect()) {
LOG.trace("Waiting for transport to reconnect..: {}", command);
long end = System.currentTimeMillis();
if (command.isMessage() && timeout > 0 && (end - start > timeout)) {
timedout = true;
LOG.info("Failover timed out after {} ms", (end - start));
break;
}
try {
reconnectMutex.wait(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOG.debug("Interupted:", e);
}
transport = connectedTransport.get();
}
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 == true) {
error = new IOException("Failover timeout of " + timeout + " ms reached.");
} else if (!willReconnect()) {
error = new IOException("Reconnect attempts of " + maxReconnectAttempts + " exceeded");
} else {
error = new IOException("Unexpected failure.");
}
break;
}
Tracked tracked = null;
try {
tracked = stateTracker.track(command);
} catch (IOException ioe) {
LOG.debug("Cannot track the command {} {}", command, ioe);
}
// 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.
synchronized (requestMap) {
if (tracked != null && tracked.isWaitingForResponse()) {
requestMap.put(command.getCommandId(), tracked);
} else if (tracked == null && command.isResponseRequired()) {
requestMap.put(command.getCommandId(), command);
}
}
// Send the message.
try {
transport.oneway(command);
stateTracker.trackBack(command);
if (command.isShutdownInfo()) {
shuttingDown = true;
}
} catch (IOException e) {
// If the command was not tracked.. we will retry in
// this method
if (tracked == null && canReconnect()) {
// 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.isResponseRequired()) {
requestMap.remove(command.getCommandId());
}
// Rethrow the exception so it will handled by
// the outer catch
throw e;
} else {
// Handle the error but allow the method to return since the
// tracked commands are replayed on reconnect.
LOG.debug("Send oneway attempt: {} failed for command: {}", i, command);
handleTransportFailure(e);
}
}
return;
} catch (MaxFrameSizeExceededException e) {
LOG.debug("MaxFrameSizeExceededException for command: {}", command);
throw e;
} catch (IOException e) {
LOG.debug("Send oneway attempt: {} failed for command: {}", i, command);
handleTransportFailure(e);
}
}
}
} catch (InterruptedException e) {
// Some one may be trying to stop our thread.
Thread.currentThread().interrupt();
throw new InterruptedIOException();
}
if (!disposed) {
if (error != null) {
if (error instanceof IOException) {
throw (IOException) error;
}
throw IOExceptionSupport.create(error);
}
}
}