StreamsShellBolt extends AbstractProcessorBolt { public StreamsShellBolt(String command, int processTimeoutInMs) { if(command == null || command.isEmpty()) throw new IllegalArgumentException("process command can not be empty"); this.command = command; this.processTimeoutInMs = processTimeoutInMs; } StreamsShellBolt(String command, int processTimeoutInMs); StreamsShellBolt withOutputStreams(List<String> outputStreams); StreamsShellBolt withEnvMap(Map<String, String> envMap); void prepare(Map stormConf, TopologyContext context,
                        final OutputCollector collector); @Override void cleanup(); @Override void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer); @Override Map<String, Object> getComponentConfiguration(); static final Logger LOG; }
@Test public void testStreamsShellBoltTest() throws Exception { setUpExpectations(); copyFiles(readFile("/splitsentence.py") , new File("/tmp/splitsentence.py")); copyFiles(readFile("/streamline.py"), new File("/tmp/streamline.py")); String command = "python splitsentence.py"; StreamsShellBolt streamsShellBolt = new StreamsShellBolt(command, 60000); streamsShellBolt = streamsShellBolt.withOutputStreams(Arrays.asList("stream1")); streamsShellBolt.prepare(new HashMap(), mockContext, mockCollector); streamsShellBolt.execute(getNextTuple(1)); new Verifications() { { String streamId; Tuple anchor; List<List<Object>> tuples = new ArrayList<>(); mockCollector.emit(streamId = withCapture(), anchor = withCapture(), withCapture(tuples)); Assert.assertEquals("stream", streamId); Assert.assertEquals(4, tuples.size()); Map<String, Object> fieldsAndValues = ((StreamlineEvent) tuples.get(0).get(0)); Assert.assertEquals("THIS", fieldsAndValues.get("word")); fieldsAndValues = ((StreamlineEvent) tuples.get(1).get(0)); Assert.assertEquals("IS", fieldsAndValues.get("word")); fieldsAndValues = ((StreamlineEvent) tuples.get(2).get(0)); Assert.assertEquals("RANDOM", fieldsAndValues.get("word")); fieldsAndValues = ((StreamlineEvent) tuples.get(3).get(0)); Assert.assertEquals("SENTENCE1", fieldsAndValues.get("word")); } }; }