src/main/java/com/rapportive/storm/spout/AMQPSpout.java
/**
     * Connects to the AMQP broker, declares the queue and subscribes to
     * incoming messages.
     */
Connects to the AMQP broker, declares the queue and subscribes to
incoming messages.
Connects to the AMQP broker declares the queue and subscribes to incoming messages .
@Override
    public void open(@SuppressWarnings("rawtypes") Map config, TopologyContext context, SpoutOutputCollector collector) {
        Long prefetchCount = (Long) config.get(CONFIG_PREFETCH_COUNT);
        if (prefetchCount == null) {
            log.info("Using default prefetch-count");
            prefetchCount = DEFAULT_PREFETCH_COUNT;
        } else if (prefetchCount < 1) {
            throw new IllegalArgumentException(CONFIG_PREFETCH_COUNT + " must be at least 1");
        }
        this.prefetchCount = prefetchCount.intValue();
        try {
            this.collector = collector;
            setupAMQP();
        } catch (IOException e) {
            log.error("AMQP setup failed", e);
        }
    }
@ Override public void open ( @ SuppressWarnings ( "rawtypes" ) Map config , TopologyContext context , SpoutOutputCollector collector ) { Long prefetchCount = ( Long ) config . get ( CONFIG_PREFETCH_COUNT ) ; if ( prefetchCount == null ) { log . info ( "Using default prefetch-count" ) ; prefetchCount = DEFAULT_PREFETCH_COUNT ; } else if ( prefetchCount < 1 ) { throw new IllegalArgumentException ( CONFIG_PREFETCH_COUNT + " must be at least 1" ) ; } this . prefetchCount = prefetchCount . intValue ( ) ; try { this . collector = collector ; setupAMQP ( ) ; } catch ( IOException e ) { log . error ( "AMQP setup failed" , e ) ; } }
Connects to the AMQP broker, declares the queue and subscribes to
 incoming messages.
Connects to the AMQP broker declares the queue and subscribes to incoming messages .