@Test public void testDeclareOutputFields() { FileReaderSpout spout = new FileReaderSpout(); TestDeclarer declarer = new TestDeclarer(); spout.declareOutputFields(declarer); final int numberOfOutputStreams = 2; Assert.assertEquals(numberOfOutputStreams, declarer.streamIdBuffer.size()); Assert.assertEquals(numberOfOutputStreams, declarer.schemaBuffer.size()); Assert.assertEquals(numberOfOutputStreams, declarer.directBuffer.size()); HashMap<String, Fields> expectedStreams = new HashMap<String, Fields>(); expectedStreams.put(null, new Fields("ts", "rawTuple")); expectedStreams.put(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); Assert.assertEquals(expectedStreams.keySet(), new HashSet<String>(declarer.streamIdBuffer)); for(int i = 0; i < numberOfOutputStreams; ++i) { Assert.assertEquals(expectedStreams.get(declarer.streamIdBuffer.get(i)).toList(), declarer.schemaBuffer .get(i).toList()); Assert.assertEquals(new Boolean(false), declarer.directBuffer.get(i)); } }
@Override public void declareOutputFields(OutputFieldsDeclarer declarer) { super.declareOutputFields(declarer); declarer.declareStream(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); }
FileReaderSpout extends AbstractOrderedFileInputSpout { @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { super.declareOutputFields(declarer); declarer.declareStream(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); } }
FileReaderSpout extends AbstractOrderedFileInputSpout { @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { super.declareOutputFields(declarer); declarer.declareStream(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); } FileReaderSpout();  FileReaderSpout(String streamID); }
FileReaderSpout extends AbstractOrderedFileInputSpout { @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { super.declareOutputFields(declarer); declarer.declareStream(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); } FileReaderSpout();  FileReaderSpout(String streamID); @Override void open(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, SpoutOutputCollector collector); @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override void activate(); @Override void deactivate(); @Override void ack(Object msgId); @Override void fail(Object msgId); @Override Map<String, Object> getComponentConfiguration(); }
FileReaderSpout extends AbstractOrderedFileInputSpout { @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { super.declareOutputFields(declarer); declarer.declareStream(TimestampMerger.FLUSH_STREAM_ID, new Fields("ts")); } FileReaderSpout();  FileReaderSpout(String streamID); @Override void open(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, SpoutOutputCollector collector); @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override void activate(); @Override void deactivate(); @Override void ack(Object msgId); @Override void fail(Object msgId); @Override Map<String, Object> getComponentConfiguration();  }