src/main/java/com/rapportive/storm/spout/AMQPSpout.java
/**
     * Emits the next message from the queue as a tuple.
     *
     * <p>If no message is ready to emit, this will wait a short time
     * ({@link #WAIT_FOR_NEXT_MESSAGE}) for one to arrive on the queue,
     * to avoid a tight loop in the spout worker.</p>
     */
Emits the next message from the queue as a tuple.
If no message is ready to emit, this will wait a short time
( #WAIT_FOR_NEXT_MESSAGE) for one to arrive on the queue,
to avoid a tight loop in the spout worker.
Emits the next message from the queue as a tuple . If no message is ready to emit this will wait a short time ( #WAIT_FOR_NEXT_MESSAGE ) for one to arrive on the queue to avoid a tight loop in the spout worker .
@Override
    public void nextTuple() {
        if (amqpConsumer != null) {
            try {
                final QueueingConsumer.Delivery delivery = amqpConsumer.nextDelivery(WAIT_FOR_NEXT_MESSAGE);
                if (delivery == null) return;
                final long deliveryTag = delivery.getEnvelope().getDeliveryTag();
                final byte[] message = delivery.getBody();
                collector.emit(serialisationScheme.deserialize(message), deliveryTag);
                /*
                 * TODO what to do about malformed messages? Skip?
                 * Avoid infinite retry!
                 * Maybe we should output them on a separate stream.
                 */
            } catch (ShutdownSignalException e) {
                log.warn("AMQP connection dropped, will attempt to reconnect...");
                Utils.sleep(WAIT_AFTER_SHUTDOWN_SIGNAL);
                reconnect();
            } catch (InterruptedException e) {
            }
        }
    }
@ Override public void nextTuple ( ) { if ( amqpConsumer != null ) { try { final QueueingConsumer . Delivery delivery = amqpConsumer . nextDelivery ( WAIT_FOR_NEXT_MESSAGE ) ; if ( delivery == null ) return ; final long deliveryTag = delivery . getEnvelope ( ) . getDeliveryTag ( ) ; final byte [ ] message = delivery . getBody ( ) ; collector . emit ( serialisationScheme . deserialize ( message ) , deliveryTag ) ; /*
                 * TODO what to do about malformed messages? Skip?
                 * Avoid infinite retry!
                 * Maybe we should output them on a separate stream.
                 */ } catch ( ShutdownSignalException e ) { log . warn ( "AMQP connection dropped, will attempt to reconnect..." ) ; Utils . sleep ( WAIT_AFTER_SHUTDOWN_SIGNAL ) ; reconnect ( ) ; } catch ( InterruptedException e ) { } } }
Emits the next message from the queue as a tuple.
Emits the next message from the queue as a tuple .
// interrupted while waiting for message, big deal