StormStreamSelector implements OutputSelector<SplitStreamType<T>> { @Override public Iterable<String> select(SplitStreamType<T> value) { String sid = value.streamId; List<String> streamId = this.streams.get(sid); if (streamId == null) { streamId = new ArrayList<String>(1); streamId.add(sid); this.streams.put(sid, streamId); } return streamId; }  @Override Iterable<String> select(SplitStreamType<T> value);  }
@Test public void testSelector() { StormStreamSelector<Object> selector = new StormStreamSelector<Object>(); SplitStreamType<Object> tuple = new SplitStreamType<Object>(); Iterator<String> result; tuple.streamId = "stream1"; result = selector.select(tuple).iterator(); Assert.assertEquals("stream1", result.next()); Assert.assertFalse(result.hasNext()); tuple.streamId = "stream2"; result = selector.select(tuple).iterator(); Assert.assertEquals("stream2", result.next()); Assert.assertFalse(result.hasNext()); tuple.streamId = "stream1"; result = selector.select(tuple).iterator(); Assert.assertEquals("stream1", result.next()); Assert.assertFalse(result.hasNext()); }