src/main/java/io/paradoxical/rabbitmq/ListenerBase.java
/**
     * Process a message off the wire. Set up MDC context, deal with stopping sematnics, etc.
     * delegate actual work of processing message elsewhere though
     *
     * @param channel
     * @param queue
     * @param message
     * @throws IOException
     */
Process a message off the wire. Set up MDC context, deal with stopping sematnics, etc.
delegate actual work of processing message elsewhere though


Process a message off the wire . Set up MDC context deal with stopping sematnics etc . delegate actual work of processing message elsewhere though
protected void deliveryHandler(
            final Channel channel,
            final Queue queue,
            final RabbitRawMessage message,
            final FutureEventProcessor<T> callback) throws IOException {
        DelegatedTimer.Context firehoseTimer = null;
        try {
            if (!running) {
                logger.warn("Message received on dead channel, attempting nack but not expecting success");
                nack(message.getEnvelope(), channel, true, null);
                return;
            }
        }
        catch (Throwable nackException) {
            if (!running) {
                logger.warn("Stopped listener cannot respond on channel", nackException);
            }
            else {
                logger.error("Unhandled exception during dequeuing. Attempting nack!", nackException);
            }
        }
        try {
            messagesInProgress.incrementAndGet();
            logger.debug("Got message on listener tag=" + message.getConsumerTag() + " message-delivery-tag=" + message.getEnvelope().getDeliveryTag());
            firehoseTimer = getFirehoseTimer().time();
            getInFlightFirehoseCounter().inc();
            process(message, channel, queue, callback);
        }
        catch (Throwable processingException) {
            if (!running) {
                logger.warn("Stopped listener cannot respond on channel", processingException);
            }
            else {
                logger.error("Unhandled exception during dequeuing. Attempting nack!", processingException);
            }
            try {
                nack(message.getEnvelope(), channel, !message.getEnvelope().isRedeliver(), null);
                logger.warn("Unhandled exception was able to successfully nack message but message may not have be properly processed.");
            }
            catch (Throwable nackException) {
                if (!running) {
                    logger.warn("Channel closed, unable to re-nack");
                    return;
                }
                logger.error("Nack unable to proceed! " +
                             "Message may be stuck in dead state and the consumer connection should be restarted. tag=" +
                             message.getConsumerTag(), nackException);
            }
        }
        finally {
            messagesInProgress.decrementAndGet();
            drainingSemaphore.release();
            if (running) {
                if (firehoseTimer != null) {
                    firehoseTimer.stop();
                    getInFlightFirehoseCounter().dec();
                }
                logger.debug("Completing message - message-delivery-tag=" + message.getEnvelope().getDeliveryTag());
            }
        }
    }
protected void deliveryHandler ( final Channel channel , final Queue queue , final RabbitRawMessage message , final FutureEventProcessor < T > callback ) throws IOException { DelegatedTimer . Context firehoseTimer = null ; try { if ( ! running ) { logger . warn ( "Message received on dead channel, attempting nack but not expecting success" ) ; nack ( message . getEnvelope ( ) , channel , true , null ) ; return ; } } catch ( Throwable nackException ) { if ( ! running ) { logger . warn ( "Stopped listener cannot respond on channel" , nackException ) ; } else { logger . error ( "Unhandled exception during dequeuing. Attempting nack!" , nackException ) ; } } try { messagesInProgress . incrementAndGet ( ) ; logger . debug ( "Got message on listener tag=" + message . getConsumerTag ( ) + " message-delivery-tag=" + message . getEnvelope ( ) . getDeliveryTag ( ) ) ; firehoseTimer = getFirehoseTimer ( ) . time ( ) ; getInFlightFirehoseCounter ( ) . inc ( ) ; process ( message , channel , queue , callback ) ; } catch ( Throwable processingException ) { if ( ! running ) { logger . warn ( "Stopped listener cannot respond on channel" , processingException ) ; } else { logger . error ( "Unhandled exception during dequeuing. Attempting nack!" , processingException ) ; } try { nack ( message . getEnvelope ( ) , channel , ! message . getEnvelope ( ) . isRedeliver ( ) , null ) ; logger . warn ( "Unhandled exception was able to successfully nack message but message may not have be properly processed." ) ; } catch ( Throwable nackException ) { if ( ! running ) { logger . warn ( "Channel closed, unable to re-nack" ) ; return ; } logger . error ( "Nack unable to proceed! " + "Message may be stuck in dead state and the consumer connection should be restarted. tag=" + message . getConsumerTag ( ) , nackException ) ; } } finally { messagesInProgress . decrementAndGet ( ) ; drainingSemaphore . release ( ) ; if ( running ) { if ( firehoseTimer != null ) { firehoseTimer . stop ( ) ; getInFlightFirehoseCounter ( ) . dec ( ) ; } logger . debug ( "Completing message - message-delivery-tag=" + message . getEnvelope ( ) . getDeliveryTag ( ) ) ; } } }
Process a message off the wire.
Process a message off the wire .