ConfigurableIngestTopology { void configureStreamGrouping(String prevComponent, Configuration boltConf, BoltDeclarer declarer) throws ConfigurationException { String streamType = boltConf.getString(STREAM_GROUPING_CONF_TYPE, STREAM_GROUPING_LOCAL_OR_SHUFFLE); String streamId = boltConf.getString(STREAM_ID, Utils.DEFAULT_STREAM_ID); if (StringUtils.equals(streamType, STREAM_GROUPING_FIELDS)) { configureStreamFieldsGrouping(prevComponent, streamId, boltConf, declarer); } else if (StringUtils.equals(streamType, STREAM_GROUPING_LOCAL_OR_SHUFFLE)) { declarer.localOrShuffleGrouping(prevComponent, streamId); } else { declarer.shuffleGrouping(prevComponent, streamId); } }  static void main(String[] args);  }
@Test public void testConfigureStreamGrouping_localOrShuffleGrouping( @Injectable String prevComponent, @Injectable Configuration boltConf, @Injectable BoltDeclarer declarer) throws Exception { new Expectations(topology) {{ boltConf.getString(STREAM_GROUPING_CONF_TYPE, STREAM_GROUPING_LOCAL_OR_SHUFFLE); result = STREAM_GROUPING_LOCAL_OR_SHUFFLE; boltConf.getString(STREAM_ID, Utils.DEFAULT_STREAM_ID); result = Utils.DEFAULT_STREAM_ID; declarer.localOrShuffleGrouping(prevComponent, Utils.DEFAULT_STREAM_ID); }}; topology.configureStreamGrouping(prevComponent, boltConf, declarer); }
@Test public void testConfigureStreamGrouping_default( @Injectable String prevComponent, @Injectable Configuration boltConf, @Injectable BoltDeclarer declarer) throws Exception { new Expectations(topology) {{ boltConf.getString(STREAM_GROUPING_CONF_TYPE, STREAM_GROUPING_LOCAL_OR_SHUFFLE); result = STREAM_GROUPING_SHUFFLE; boltConf.getString(STREAM_ID, Utils.DEFAULT_STREAM_ID); result = Utils.DEFAULT_STREAM_ID; declarer.shuffleGrouping(prevComponent, Utils.DEFAULT_STREAM_ID); }}; topology.configureStreamGrouping(prevComponent, boltConf, declarer); }