src/main/java/io/paradoxical/rabbitmq/ListenerBase.java
/**
     * Main processing of a message.
     *
     * @param rabbitRawMessage
     * @param channel
     * @param queue
     * @param callback
     * @throws IOException
     */
Main processing of a message.
Main processing of a message .
private void process(final RabbitRawMessage rabbitRawMessage, final Channel channel, final Queue queue, final FutureEventProcessor<T> callback) throws IOException {
        Message<T> message = null;
        try {
            message = getMessageFromRaw(rabbitRawMessage, queue);
            T item = message.getItem();
            DelegatedTimer.Context eventTypeTimer = getEventTypeTimer(item).time();
            getEventTypeInFlightCounter(item).inc();
            try {
                delegateMessage(message, channel, callback);
            }
            finally {
                eventTypeTimer.stop();
                getEventTypeInFlightCounter(item).dec();
            }
        }
        catch (Throwable ex) {
            logger.error(String.format("Error during dequeueing. Tag %s", rabbitRawMessage.getEnvelope().getDeliveryTag()),
                         ex);
            if (message != null) {
                requeue(message, channel);
            }
            else {
                nack(rabbitRawMessage.getEnvelope(), channel, !rabbitRawMessage.getEnvelope().isRedeliver(), null);
            }
        }
        finally {
            cleanMessageContext();
        }
    }
private void process ( final RabbitRawMessage rabbitRawMessage , final Channel channel , final Queue queue , final FutureEventProcessor < T > callback ) throws IOException { Message < T > message = null ; try { message = getMessageFromRaw ( rabbitRawMessage , queue ) ; T item = message . getItem ( ) ; DelegatedTimer . Context eventTypeTimer = getEventTypeTimer ( item ) . time ( ) ; getEventTypeInFlightCounter ( item ) . inc ( ) ; try { delegateMessage ( message , channel , callback ) ; } finally { eventTypeTimer . stop ( ) ; getEventTypeInFlightCounter ( item ) . dec ( ) ; } } catch ( Throwable ex ) { logger . error ( String . format ( "Error during dequeueing. Tag %s" , rabbitRawMessage . getEnvelope ( ) . getDeliveryTag ( ) ) , ex ) ; if ( message != null ) { requeue ( message , channel ) ; } else { nack ( rabbitRawMessage . getEnvelope ( ) , channel , ! rabbitRawMessage . getEnvelope ( ) . isRedeliver ( ) , null ) ; } } finally { cleanMessageContext ( ) ; } }
Main processing of a message.
Main processing of a message .
// we got a message object, leverage the requeue semantics by retry if possible // redeliver if it wasn't already delivered once, dont know anything else about the message