package com.musgame.bis.Chapter03.trident.topology;

import com.musgame.bis.Chapter03.trident.operator.*;
import com.musgame.bis.Chapter03.trident.spout.DiagnosisEventSpout;
import com.musgame.bis.Chapter03.trident.state.OutbreakTrendFactory;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.trident.Stream;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.operation.builtin.Count;
import org.apache.storm.tuple.Fields;

public class OutbreakDetectionTopology {

    public static StormTopology buildTopology() {
        TridentTopology topology = new TridentTopology();
        DiagnosisEventSpout spout = new DiagnosisEventSpout();
        Stream inputStream = topology.newStream("event", spout);

        inputStream.each(new Fields("event"), new DiseaseFilter())
                .each(new Fields("event"), new CityAssignment(), new Fields("city"))
                .each(new Fields("event", "city"), new HourAssignment(), new Fields("hour", "cityDiseaseHour"))
                .groupBy(new Fields("cityDiseaseHour"))
                .persistentAggregate(new OutbreakTrendFactory(), new Count(), new Fields("count")).newValuesStream()
                .each(new Fields("cityDiseaseHour", "count"), new OutbreakDetector(), new Fields("alert"))
                .each(new Fields("alert"), new DispatchAlert(), new Fields());
        return topology.build();
    }

    public static void main(String[] args) throws Exception {
        Config conf = new Config();
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("cdc", conf, buildTopology());
        Thread.sleep(200000);
        cluster.shutdown();
    }
}

@@ -819,16 +819,61 @@
 pout);%0A%0A
+        // Filter for critical events %E8%BF%87%E6%BB%A4%E5%85%B3%E9%94%AE%E4%BA%8B%E4%BB%B6%0A
         
@@ -930,16 +930,68 @@
 ilter())
+%0A%0A                // Locate the closest city %E6%89%BE%E5%88%B0%E6%9C%80%E8%BF%91%E7%9A%84%E5%9F%8E%E5%B8%82
 %0A       
@@ -1067,16 +1067,66 @@
 %22city%22))
+%0A%0A                // Derive the hour segment %E5%AF%BC%E5%87%BA%E5%B0%8F%E6%97%B6%E6%AE%B5
 %0A       
@@ -1182,32 +1182,56 @@
 ourAssignment(),
+%0A                       
  new Fields(%22hou
@@ -1246,39 +1246,105 @@
 yDiseaseHour%22))%0A
-       
+%0A                // Group occurrences in same city and hour %E5%9C%A8%E5%90%8C%E4%B8%80%E5%9F%8E%E5%B8%82%E6%8C%89%E5%B0%8F%E6%97%B6%E5%8F%91%E7%94%9F%E5%88%86%E7%BB%84%0A
          .groupB
@@ -1336,16 +1336,23 @@
         
+       
 .groupBy
@@ -1382,16 +1382,92 @@
 eHour%22))
+%0A%0A                // Count occurrences and persist the results %E8%AE%A1%E7%AE%97%E5%8F%91%E7%8E%B0%E6%AC%A1%E6%95%B0%E5%B9%B6%E6%8C%81%E4%B9%85%E4%BF%9D%E5%AD%98%E7%BB%93%E6%9E%9C
 %0A       
@@ -1557,16 +1557,33 @@
 count%22))
+%0A                
 .newValu
@@ -1592,16 +1592,60 @@
 Stream()
+%0A%0A                // Detect an outbreak %E6%A3%80%E6%B5%8B%E5%8F%91%E7%97%85
 %0A       
@@ -1702,53 +1702,145 @@
 t%22),
- new OutbreakDetector(), new Fields(%22alert%22))
+%0A                        new OutbreakDetector(),%0A                        new Fields(%22alert%22))%0A%0A                // Dispatch the alert %E5%8F%91%E5%87%BA%E8%AD%A6%E6%8A%A5
 %0A   

update example