@SuppressWarnings("rawtypes")
    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector)
    {
        Number timeout = (Number) conf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS);
        this.messageTimeoutMs = (int) TimeUnit.SECONDS.toMillis(timeout.longValue());
        this.collector = collector;
        this.emitIndex = 0;
        this.clientInfoList = Lists.newArrayList();
        int numTasks = context.getComponentTasks(context.getThisComponentId()).size();
        int myIndex = context.getThisTaskIndex();
        int numHosts = this.hostInfos.size();
        if (numTasks < numHosts)
        {
            for (HostInfo host : this.hostInfos)
            {
                this.clientInfoList.add(new KestrelClientInfo(host.getHost(), host.getPort()));
            }
        }
        else
        {
            HostInfo host = this.hostInfos.get(myIndex % numHosts);
            this.clientInfoList.add(new KestrelClientInfo(host.getHost(), host.getPort()));
        }

        this.emitBuffer = new LinkedList<EmitItem>();
    }
@ SuppressWarnings ( "rawtypes" ) @ Override public void open ( Map conf , TopologyContext context , SpoutOutputCollector collector ) { Number timeout = ( Number ) conf . get ( Config . TOPOLOGY_MESSAGE_TIMEOUT_SECS ) ; this . messageTimeoutMs = ( int ) TimeUnit . SECONDS . toMillis ( timeout . longValue ( ) ) ; this . collector = collector ; this . emitIndex = 0 ; this . clientInfoList = Lists . newArrayList ( ) ; int numTasks = context . getComponentTasks ( context . getThisComponentId ( ) ) . size ( ) ; int myIndex = context . getThisTaskIndex ( ) ; int numHosts = this . hostInfos . size ( ) ; if ( numTasks < numHosts ) { for ( HostInfo host : this . hostInfos ) { this . clientInfoList . add ( new KestrelClientInfo ( host . getHost ( ) , host . getPort ( ) ) ) ; } } else { HostInfo host = this . hostInfos . get ( myIndex % numHosts ) ; this . clientInfoList . add ( new KestrelClientInfo ( host . getHost ( ) , host . getPort ( ) ) ) ; } this . emitBuffer = new LinkedList < EmitItem > ( ) ; }
{@inheritDoc}