@Test public void testPrepare() { Map<?, ?> config = new HashMap<Object, Object>(); TopologyContext context = mock(TopologyContext.class); OutputCollector collector = mock(OutputCollector.class); checker.prepare(config, context, collector); verify(boltMockStatic).prepare(config, context, collector); verify(boltMockStatic, atMost(1)).prepare(config, context, collector); }
@Override public void prepare(@SuppressWarnings("rawtypes") Map arg0, TopologyContext arg1, OutputCollector arg2) { LinkedList<Integer> taskIds = new LinkedList<Integer>(); for(Entry<GlobalStreamId, Grouping> inputStream : arg1.getThisSources().entrySet()) { taskIds.addAll(arg1.getComponentTasks(inputStream.getKey().get_componentId())); } logger.debug("Detected producer tasks: {}", taskIds); if(this.tsIndex != -1) { assert (this.tsAttributeName == null && this.tsExtractor == null); this.merger = new StreamMerger<Tuple>(taskIds, this.tsIndex); } else if(this.tsAttributeName != null) { assert (this.tsExtractor == null); this.merger = new StreamMerger<Tuple>(taskIds, this.tsAttributeName); } else { assert (this.tsExtractor != null); this.merger = new StreamMerger<Tuple>(taskIds, this.tsExtractor); } this.wrappedBolt.prepare(arg0, arg1, arg2); }
TimestampMerger implements IRichBolt { @Override public void prepare(@SuppressWarnings("rawtypes") Map arg0, TopologyContext arg1, OutputCollector arg2) { LinkedList<Integer> taskIds = new LinkedList<Integer>(); for(Entry<GlobalStreamId, Grouping> inputStream : arg1.getThisSources().entrySet()) { taskIds.addAll(arg1.getComponentTasks(inputStream.getKey().get_componentId())); } logger.debug("Detected producer tasks: {}", taskIds); if(this.tsIndex != -1) { assert (this.tsAttributeName == null && this.tsExtractor == null); this.merger = new StreamMerger<Tuple>(taskIds, this.tsIndex); } else if(this.tsAttributeName != null) { assert (this.tsExtractor == null); this.merger = new StreamMerger<Tuple>(taskIds, this.tsAttributeName); } else { assert (this.tsExtractor != null); this.merger = new StreamMerger<Tuple>(taskIds, this.tsExtractor); } this.wrappedBolt.prepare(arg0, arg1, arg2); } }
TimestampMerger implements IRichBolt { @Override public void prepare(@SuppressWarnings("rawtypes") Map arg0, TopologyContext arg1, OutputCollector arg2) { LinkedList<Integer> taskIds = new LinkedList<Integer>(); for(Entry<GlobalStreamId, Grouping> inputStream : arg1.getThisSources().entrySet()) { taskIds.addAll(arg1.getComponentTasks(inputStream.getKey().get_componentId())); } logger.debug("Detected producer tasks: {}", taskIds); if(this.tsIndex != -1) { assert (this.tsAttributeName == null && this.tsExtractor == null); this.merger = new StreamMerger<Tuple>(taskIds, this.tsIndex); } else if(this.tsAttributeName != null) { assert (this.tsExtractor == null); this.merger = new StreamMerger<Tuple>(taskIds, this.tsAttributeName); } else { assert (this.tsExtractor != null); this.merger = new StreamMerger<Tuple>(taskIds, this.tsExtractor); } this.wrappedBolt.prepare(arg0, arg1, arg2); } TimestampMerger(IRichBolt wrappedBolt, int tsIndex);  TimestampMerger(IRichBolt wrappedBolt, String tsAttributeName);  TimestampMerger(IRichBolt wrappedBolt, TimeStampExtractor<Tuple> tsExtractor); }
TimestampMerger implements IRichBolt { @Override public void prepare(@SuppressWarnings("rawtypes") Map arg0, TopologyContext arg1, OutputCollector arg2) { LinkedList<Integer> taskIds = new LinkedList<Integer>(); for(Entry<GlobalStreamId, Grouping> inputStream : arg1.getThisSources().entrySet()) { taskIds.addAll(arg1.getComponentTasks(inputStream.getKey().get_componentId())); } logger.debug("Detected producer tasks: {}", taskIds); if(this.tsIndex != -1) { assert (this.tsAttributeName == null && this.tsExtractor == null); this.merger = new StreamMerger<Tuple>(taskIds, this.tsIndex); } else if(this.tsAttributeName != null) { assert (this.tsExtractor == null); this.merger = new StreamMerger<Tuple>(taskIds, this.tsAttributeName); } else { assert (this.tsExtractor != null); this.merger = new StreamMerger<Tuple>(taskIds, this.tsExtractor); } this.wrappedBolt.prepare(arg0, arg1, arg2); } TimestampMerger(IRichBolt wrappedBolt, int tsIndex);  TimestampMerger(IRichBolt wrappedBolt, String tsAttributeName);  TimestampMerger(IRichBolt wrappedBolt, TimeStampExtractor<Tuple> tsExtractor); @Override void prepare(@SuppressWarnings("rawtypes") Map arg0, TopologyContext arg1, OutputCollector arg2); @Override void execute(Tuple tuple); @Override void declareOutputFields(OutputFieldsDeclarer arg0); @Override Map<String, Object> getComponentConfiguration(); @Override void cleanup(); }
TimestampMerger implements IRichBolt { @Override public void prepare(@SuppressWarnings("rawtypes") Map arg0, TopologyContext arg1, OutputCollector arg2) { LinkedList<Integer> taskIds = new LinkedList<Integer>(); for(Entry<GlobalStreamId, Grouping> inputStream : arg1.getThisSources().entrySet()) { taskIds.addAll(arg1.getComponentTasks(inputStream.getKey().get_componentId())); } logger.debug("Detected producer tasks: {}", taskIds); if(this.tsIndex != -1) { assert (this.tsAttributeName == null && this.tsExtractor == null); this.merger = new StreamMerger<Tuple>(taskIds, this.tsIndex); } else if(this.tsAttributeName != null) { assert (this.tsExtractor == null); this.merger = new StreamMerger<Tuple>(taskIds, this.tsAttributeName); } else { assert (this.tsExtractor != null); this.merger = new StreamMerger<Tuple>(taskIds, this.tsExtractor); } this.wrappedBolt.prepare(arg0, arg1, arg2); } TimestampMerger(IRichBolt wrappedBolt, int tsIndex);  TimestampMerger(IRichBolt wrappedBolt, String tsAttributeName);  TimestampMerger(IRichBolt wrappedBolt, TimeStampExtractor<Tuple> tsExtractor); @Override void prepare(@SuppressWarnings("rawtypes") Map arg0, TopologyContext arg1, OutputCollector arg2); @Override void execute(Tuple tuple); @Override void declareOutputFields(OutputFieldsDeclarer arg0); @Override Map<String, Object> getComponentConfiguration(); @Override void cleanup(); final static String FLUSH_STREAM_ID; }