[STORM-2490] Lambda support

• Type: New Feature
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 2.0.0
• Component/s: storm-client
• Assignee: Xin Wang
• Reporter: Xin Wang
• Created: 24/Apr/17 13:35
• Updated: 27/Apr/17 01:37
• Resolved: 27/Apr/17 01:37

Description

In the past, If we want print tuples, we need to write the following code:

    class PritingBolt extends BaseBasicBolt{

        @Override
        public void execute(Tuple input, BasicOutputCollector collector) {
            System.out.println(input);
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            // nothing
        }
    }

builder.setBolt("bolt2", new PritingBolt());

Now, with this patch:

builder.setBolt("bolt2", tuple -> System.out.println(tuple));

The above is just an simplest demo. This patch provides some new methods in TopologyBuilder to allow you to use Java8 lambda expression:

setSpout(String id, SerializableSupplier<?> supplier)
setSpout(String id, SerializableSupplier<?> supplier, Number parallelism_hint)

//  receiving tuple, and emitting to downstream
setBolt(String id, SerializableBiConsumer<Tuple,BasicOutputCollector> biConsumer, String... fields)
setBolt(String id, SerializableBiConsumer<Tuple,BasicOutputCollector> biConsumer, Number parallelism_hint, String... fields)

// receiving tuple, and never emitting to downstream
setBolt(String id, SerializableConsumer<Tuple> consumer)
setBolt(String id, SerializableConsumer<Tuple> consumer, Number parallelism_hint)

Here is another example including the three interface usage:

        // example. spout1: generate random strings
        // bolt1: get the first part of a string
        // bolt2: output the tuple
        builder.setSpout("spout1", () -> UUID.randomUUID().toString());
        builder.setBolt("bolt1", (tuple, collector) -> {
            String[] parts = tuple.getStringByField("lambda").split("\\-");
            collector.emit(new Values(parts[0]));
        }, "field").shuffleGrouping("spout1");
        builder.setBolt("bolt2", tuple -> System.out.println(tuple)).shuffleGrouping("bolt1");

Issue Links

• links to: GitHub Pull Request #2089