Merge pull request #297 from elenacliu/hopping

[ISSUE #304] add 2 window examples

Author: ni-ze

example/src/main/java/org/apache/rocketmq/streams/examples/pojo/Order.java

@@ -0,0 +1,46 @@
+package org.apache.rocketmq.streams.examples.pojo;
+
+public class Order {
+    private String type;        // drink, food, 
+    private Integer price;          // order price
+    private String customer;    // customer name
+
+    public Order() {
+
+    }
+
+    public Order(String type, Integer price, String customer) {
+        this.type = type;
+        this.price = price;
+        this.customer = customer;
+    }
+
+    public String getType() {
+        return type;
+    }
+
+    public String getCustomer() {
+        return customer;
+    }
+
+    public Integer getPrice() {
+        return price;
+    }
+
+    public void setType(String type) {
+        this.type = type;
+    }
+
+    public void setCustomer(String customer) {
+        this.customer = customer;
+    }
+
+    public void setPrice(Integer price) {
+        this.price = price;
+    }
+
+    @Override
+    public String toString() {
+        return "Order{" + "type=" + type + ", price=" + price + ", customer=" + customer + "}";
+    }
+}

example/src/main/java/org/apache/rocketmq/streams/examples/window/WindowOrderCount.java

@@ -0,0 +1,83 @@
+package org.apache.rocketmq.streams.examples.window;
+
+import java.util.Properties;
+
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.streams.core.RocketMQStream;
+import org.apache.rocketmq.streams.core.function.SelectAction;
+import org.apache.rocketmq.streams.core.metadata.StreamConfig;
+import org.apache.rocketmq.streams.core.rstream.StreamBuilder;
+import org.apache.rocketmq.streams.core.topology.TopologyBuilder;
+import org.apache.rocketmq.streams.core.util.Pair;
+import org.apache.rocketmq.streams.core.window.Time;
+import org.apache.rocketmq.streams.core.window.WindowBuilder;
+import org.apache.rocketmq.streams.examples.pojo.Order;
+
+import com.alibaba.fastjson.JSON;
+
+public class WindowOrderCount {
+    public static void main(String[] args) {
+        StreamBuilder builder = getOrder2();
+
+        TopologyBuilder topologyBuilder = builder.build();
+        Properties properties = new Properties();
+        properties.putIfAbsent(MixAll.NAMESRV_ADDR_PROPERTY, "127.0.0.1:9876");
+        properties.put(StreamConfig.ALLOW_LATENESS_MILLISECOND, 2000);
+
+
+        RocketMQStream rocketMQStream = new RocketMQStream(topologyBuilder, properties);
+
+        Runtime.getRuntime().addShutdownHook(new Thread("ordercount-shutdown-hook") {
+            @Override
+            public void run() {
+                rocketMQStream.stop();
+            }
+        });
+
+        rocketMQStream.start();
+    }
+
+    private static StreamBuilder getOrder1() {
+    /**
+     * Get the count of drink/food orders in last 30 seconds every 10 seconds
+     **/
+        StreamBuilder builder = new StreamBuilder("windowOrderCount");
+
+        builder.source("order", source -> {
+                Order order = JSON.parseObject(source, Order.class);
+                System.out.println(order.toString());
+                return new Pair<>(null, order);
+            })
+            .keyBy(Order::getType)
+            .window(WindowBuilder.slidingWindow(Time.seconds(30), Time.seconds(10)))
+            .count()
+            .toRStream()
+            .print();
+
+        return builder;
+    }
+
+    private static StreamBuilder getOrder2() {
+    /**
+     * Get how much the customers pay for drink/food every 100 seconds
+     **/
+        StreamBuilder builder = new StreamBuilder("windowOrderCount");
+        builder.source("order", source -> {
+            Order order = JSON.parseObject(source, Order.class);
+            System.out.println(order.toString());
+            return new Pair<>(null, order);
+        })
+            .keyBy(new SelectAction<String, Order>() {
+                @Override
+                public String select(Order order) {
+                    return order.getCustomer() + "@" + order.getType();
+                }
+            })
+            .window(WindowBuilder.tumblingWindow(Time.seconds(100)))
+            .sum(Order::getPrice)
+            .toRStream()
+            .print();
+
+        return builder;
+    }
+}