protected void startOperation (final Message message, final Session session)
throws IOException,
ClassNotFoundException
{
Preconditions.checkArgument (message.specification instanceof AmqpMessage);
final AmqpMessage amqpMessage = (AmqpMessage) message.specification;
CompletionToken token;
IResult<Boolean> resultBool;
IResult<String> resultString;
String queue;
String exchange;
boolean durable;
boolean autoDelete;
boolean passive;
boolean autoAck;
boolean exclusive;
String consumer;
String routingKey;
byte[] dataBytes;
final AmqpDriver driver = super.getDriver (AmqpDriver.class);
switch (amqpMessage) {
case ACCESS :
AmqpStub.logger.trace ("Received initiation message"); // $NON-NLS-1$
break;
case ABORTED :
AmqpStub.logger.trace ("Received termination message"); // $NON-NLS-1$
break;
case DECL_EXCHANGE_REQUEST :
final AmqpPayloads.DeclareExchangeRequest declExchange = (DeclareExchangeRequest) message.payload;
token = declExchange.getToken ();
exchange = declExchange.getExchange ();
final ExchangeType type = declExchange.getType ();
durable = declExchange.getDurable ();
autoDelete = declExchange.getAutoDelete ();
passive = declExchange.getPassive ();
AmqpStub.logger.trace ("AmqpStub - Received request for DECLARE EXCHANGE "); // $NON-NLS-1$
// NOTE: execute operation
final DriverOperationFinishedHandler exchHandler = new DriverOperationFinishedHandler (token, session);
resultBool = driver.declareExchange (token.getClientId (), exchange, AmqpExchangeType.valueOf (type.toString ().toUpperCase ()), durable, autoDelete, passive, exchHandler);
exchHandler.setDetails (AmqpOperations.DECLARE_EXCHANGE, resultBool);
break;
case DECL_QUEUE_REQUEST :
final AmqpPayloads.DeclareQueueRequest declQueue = (DeclareQueueRequest) message.payload;
token = declQueue.getToken ();
queue = declQueue.getQueue ();
exclusive = declQueue.getExclusive ();
durable = declQueue.getDurable ();
autoDelete = declQueue.getAutoDelete ();
passive = declQueue.getPassive ();
AmqpStub.logger.trace ("AmqpStub - Received request for DECLARE QUEUE"); // $NON-NLS-1$
// NOTE: execute operation
final DriverOperationFinishedHandler queueHandler = new DriverOperationFinishedHandler (token, session);
resultBool = driver.declareQueue (token.getClientId (), queue, exclusive, durable, autoDelete, passive, queueHandler);
queueHandler.setDetails (AmqpOperations.DECLARE_QUEUE, resultBool);
break;
case BIND_QUEUE_REQUEST :
final AmqpPayloads.BindQueueRequest bindQueue = (BindQueueRequest) message.payload;
token = bindQueue.getToken ();
exchange = bindQueue.getExchange ();
queue = bindQueue.getQueue ();
routingKey = bindQueue.getRoutingKey ();
AmqpStub.logger.trace ("AmqpStub - Received request for BIND QUEUE"); // $NON-NLS-1$
// NOTE: execute operation
final DriverOperationFinishedHandler bindHandler = new DriverOperationFinishedHandler (token, session);
resultBool = driver.bindQueue (token.getClientId (), exchange, queue, routingKey, bindHandler);
bindHandler.setDetails (AmqpOperations.BIND_QUEUE, resultBool);
break;
case PUBLISH_REQUEST :
final AmqpPayloads.PublishRequest publish = (PublishRequest) message.payload;
token = publish.getToken ();
dataBytes = publish.getData ().toByteArray ();
durable = publish.getDurable ();
exchange = publish.getExchange ();
final boolean immediate = publish.getImmediate ();
final boolean mandatory = publish.getMandatory ();
routingKey = publish.getRoutingKey ();
String correlationId = null;
String replyTo = null;
final Envelope envelope = publish.getEnvelope ();
if (publish.hasCorrelationId ()) {
correlationId = publish.getCorrelationId ();
}
if (publish.hasReplyTo ()) {
replyTo = publish.getReplyTo ();
}
final AmqpOutboundMessage mssg = new AmqpOutboundMessage (exchange, routingKey, dataBytes, mandatory, immediate, durable, replyTo, envelope.getContentEncoding (), envelope.getContentType (), correlationId, null);
AmqpStub.logger.trace ("AmqpStub - Received request for PUBLISH"); // $NON-NLS-1$
// NOTE: execute operation
final DriverOperationFinishedHandler pubHandler = new DriverOperationFinishedHandler (token, session);
resultBool = driver.basicPublish (token.getClientId (), mssg, pubHandler);
pubHandler.setDetails (AmqpOperations.PUBLISH, resultBool);
break;
case CONSUME_REQUEST :
final AmqpPayloads.ConsumeRequest cop = (ConsumeRequest) message.payload;
token = cop.getToken ();
queue = cop.getQueue ();
consumer = cop.getConsumer ();
exclusive = cop.getExclusive ();
autoAck = cop.getAutoAck ();
AmqpStub.logger.trace ("AmqpStub - Received request for CONSUME"); // $NON-NLS-1$
// NOTE: execute operation
final DriverOperationFinishedHandler consHandler = new DriverOperationFinishedHandler (token, session);
final IAmqpConsumer consumeCallback = new ConsumerHandler (session);
resultString = driver.basicConsume (queue, consumer, exclusive, autoAck, consumeCallback, consHandler);
consHandler.setDetails (AmqpOperations.CONSUME, resultString);
break;
case GET_REQUEST :
final AmqpPayloads.GetRequest gop = (GetRequest) message.payload;
token = gop.getToken ();
queue = gop.getQueue ();
autoAck = gop.getAutoAck ();
AmqpStub.logger.trace ("AmqpStub - Received request for GET"); // $NON-NLS-1$
// NOTE: execute operation
final DriverOperationFinishedHandler getHandler = new DriverOperationFinishedHandler (token, session);
resultBool = driver.basicGet (token.getClientId (), queue, autoAck, getHandler);
getHandler.setDetails (AmqpOperations.GET, resultBool);
break;
case CANCEL_REQUEST :
final AmqpPayloads.CancelRequest clop = (CancelRequest) message.payload;
token = clop.getToken ();
consumer = clop.getConsumer ();
AmqpStub.logger.trace ("AmqpStub - Received request for CANCEL"); // $NON-NLS-1$
// NOTE: execute operation
final DriverOperationFinishedHandler cancelHandler = new DriverOperationFinishedHandler (token, session);
resultBool = driver.basicCancel (consumer, cancelHandler);
cancelHandler.setDetails (AmqpOperations.CANCEL, resultBool);
break;
case ACK :
final AmqpPayloads.Ack aop = (Ack) message.payload;
token = aop.getToken ();
final long delivery = aop.getDelivery ();
final boolean multiple = aop.getMultiple ();
AmqpStub.logger.trace ("AmqpStub - Received ACK "); // $NON-NLS-1$
// NOTE: execute operation
final DriverOperationFinishedHandler ackHandler = new DriverOperationFinishedHandler (token, session);
resultBool = driver.basicAck (token.getClientId (), delivery, multiple, ackHandler);
ackHandler.setDetails (AmqpOperations.ACK, resultBool);
break;
default:
final DriverOperationFinishedHandler errHandler = new DriverOperationFinishedHandler (null, session);
driver.handleUnsupportedOperationError (amqpMessage.toString (), errHandler);
AmqpStub.logger.error ("Unknown amqp message: " + amqpMessage.toString ()); // $NON-NLS-1$
break;
}
}