// Path: Chapter07/src/main/java/com/packtpub/storm/trident/operator/MessageTypeFilter.java
// public class MessageTypeFilter extends BaseFilter {
//     private static final long serialVersionUID = 1L;
// //    private static final Logger LOG = LoggerFactory.getLogger(MessageTypeFilter.class);
// 
//     @Override
//     public boolean isKeep(TridentTuple tuple) {
//         FixMessageDto message = (FixMessageDto) tuple.getValue(0);
//         if (message.msgType.equals("8") && message.price > 0) {
//             return true;
//         }
//         return false;
//     }
// }
// 
// Path: Chapter07/src/main/java/com/packtpub/storm/trident/spout/FixEventSpout.java
// @SuppressWarnings("rawtypes")
// public class FixEventSpout implements ITridentSpout<Long> {
//     private static final long serialVersionUID = 1L;
//     SpoutOutputCollector collector;
//     BatchCoordinator<Long> coordinator = new DefaultCoordinator();
//     Emitter<Long> emitter = new FixEventEmitter();
// 
//     @Override
//     public BatchCoordinator<Long> getCoordinator(String txStateId, Map conf, TopologyContext context) {
//         return coordinator;
//     }
// 
//     @Override
//     public Emitter<Long> getEmitter(String txStateId, Map conf, TopologyContext context) {
//         return emitter;
//     }
// 
//     @Override
//     public Map getComponentConfiguration() {
//         return null;
//     }
// 
//     @Override
//     public Fields getOutputFields() {
//         return new Fields("message");
//     }
// }
// 
// Path: Chapter07/src/main/java/com/packtpub/storm/trident/state/DruidStateFactory.java
// @SuppressWarnings("rawtypes")
// public class DruidStateFactory implements StateFactory {
//     private static final long serialVersionUID = 1L;
//     private static final Logger LOG = LoggerFactory.getLogger(DruidStateFactory.class);
//     private static RealtimeNode rn = null;
// 
//     private static synchronized void startRealtime() {
//         if (rn == null) {
//             final Lifecycle lifecycle = new Lifecycle();
//             rn = RealtimeNode.builder().build();
//             lifecycle.addManagedInstance(rn);
//             rn.registerJacksonSubtype(new NamedType(StormFirehoseFactory.class, "storm"));
//             Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
//                 @Override
//                 public void run() {
//                     LOG.info("Running shutdown hook");
//                     lifecycle.stop();
//                 }
//             }));
// 
//             try {
//                 lifecycle.start();
//             } catch (Throwable t) {
//                 LOG.info("Throwable caught at startup, committing seppuku", t);
//                 t.printStackTrace();
//                 System.exit(2);
//             }
//         }
//     }
// 
//     @Override
//     public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
//         DruidStateFactory.startRealtime();
//         return new DruidState(partitionIndex);
//     }
// }
// 
// Path: Chapter08/src/main/java/com/packtpub/storm/trident/state/DruidStateUpdater.java
// public class DruidStateUpdater implements StateUpdater<DruidState> {
//     private static final long serialVersionUID = 1L;
// 
//     // private static final Logger LOG =
//     // LoggerFactory.getLogger(DruidStateUpdater.class);
// 
//     @SuppressWarnings("rawtypes")
//     @Override
//     public void prepare(Map conf, TridentOperationContext context) {
//     }
// 
//     @Override
//     public void cleanup() {
//     }
// 
//     @Override
//     public void updateState(DruidState state, List<TridentTuple> tuples, TridentCollector collector) {
//         state.aggregateTuples(tuples);
//     }
// }


import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;
import com.metamx.druid.log.LogLevelAdjuster;
import com.packtpub.storm.trident.operator.MessageTypeFilter;
import com.packtpub.storm.trident.spout.FixEventSpout;
import com.packtpub.storm.trident.state.DruidStateFactory;
import com.packtpub.storm.trident.state.DruidStateUpdater;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.trident.Stream;
import storm.trident.TridentTopology;
package com.packtpub.storm.trident.topology;


public class FinancialAnalyticsTopology {
    private static final Logger LOG = LoggerFactory.getLogger(FinancialAnalyticsTopology.class);

    public static StormTopology buildTopology() {
        LOG.info("Building topology.");
// Path: Chapter07/src/main/java/com/packtpub/storm/trident/operator/MessageTypeFilter.java
// public class MessageTypeFilter extends BaseFilter {
//     private static final long serialVersionUID = 1L;
// //    private static final Logger LOG = LoggerFactory.getLogger(MessageTypeFilter.class);
// 
//     @Override
//     public boolean isKeep(TridentTuple tuple) {
//         FixMessageDto message = (FixMessageDto) tuple.getValue(0);
//         if (message.msgType.equals("8") && message.price > 0) {
//             return true;
//         }
//         return false;
//     }
// }
// 
// Path: Chapter07/src/main/java/com/packtpub/storm/trident/spout/FixEventSpout.java
// @SuppressWarnings("rawtypes")
// public class FixEventSpout implements ITridentSpout<Long> {
//     private static final long serialVersionUID = 1L;
//     SpoutOutputCollector collector;
//     BatchCoordinator<Long> coordinator = new DefaultCoordinator();
//     Emitter<Long> emitter = new FixEventEmitter();
// 
//     @Override
//     public BatchCoordinator<Long> getCoordinator(String txStateId, Map conf, TopologyContext context) {
//         return coordinator;
//     }
// 
//     @Override
//     public Emitter<Long> getEmitter(String txStateId, Map conf, TopologyContext context) {
//         return emitter;
//     }
// 
//     @Override
//     public Map getComponentConfiguration() {
//         return null;
//     }
// 
//     @Override
//     public Fields getOutputFields() {
//         return new Fields("message");
//     }
// }
// 
// Path: Chapter07/src/main/java/com/packtpub/storm/trident/state/DruidStateFactory.java
// @SuppressWarnings("rawtypes")
// public class DruidStateFactory implements StateFactory {
//     private static final long serialVersionUID = 1L;
//     private static final Logger LOG = LoggerFactory.getLogger(DruidStateFactory.class);
//     private static RealtimeNode rn = null;
// 
//     private static synchronized void startRealtime() {
//         if (rn == null) {
//             final Lifecycle lifecycle = new Lifecycle();
//             rn = RealtimeNode.builder().build();
//             lifecycle.addManagedInstance(rn);
//             rn.registerJacksonSubtype(new NamedType(StormFirehoseFactory.class, "storm"));
//             Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
//                 @Override
//                 public void run() {
//                     LOG.info("Running shutdown hook");
//                     lifecycle.stop();
//                 }
//             }));
// 
//             try {
//                 lifecycle.start();
//             } catch (Throwable t) {
//                 LOG.info("Throwable caught at startup, committing seppuku", t);
//                 t.printStackTrace();
//                 System.exit(2);
//             }
//         }
//     }
// 
//     @Override
//     public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
//         DruidStateFactory.startRealtime();
//         return new DruidState(partitionIndex);
//     }
// }
// 
// Path: Chapter08/src/main/java/com/packtpub/storm/trident/state/DruidStateUpdater.java
// public class DruidStateUpdater implements StateUpdater<DruidState> {
//     private static final long serialVersionUID = 1L;
// 
//     // private static final Logger LOG =
//     // LoggerFactory.getLogger(DruidStateUpdater.class);
// 
//     @SuppressWarnings("rawtypes")
//     @Override
//     public void prepare(Map conf, TridentOperationContext context) {
//     }
// 
//     @Override
//     public void cleanup() {
//     }
// 
//     @Override
//     public void updateState(DruidState state, List<TridentTuple> tuples, TridentCollector collector) {
//         state.aggregateTuples(tuples);
//     }
// }


// Path: Chapter07/src/main/java/com/packtpub/storm/trident/topology/FinancialAnalyticsTopology.java
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;
import com.metamx.druid.log.LogLevelAdjuster;
import com.packtpub.storm.trident.operator.MessageTypeFilter;
import com.packtpub.storm.trident.spout.FixEventSpout;
import com.packtpub.storm.trident.state.DruidStateFactory;
import com.packtpub.storm.trident.state.DruidStateUpdater;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.trident.Stream;
import storm.trident.TridentTopology;

package com.packtpub.storm.trident.topology;


public class FinancialAnalyticsTopology {
    private static final Logger LOG = LoggerFactory.getLogger(FinancialAnalyticsTopology.class);

    public static StormTopology buildTopology() {
        LOG.info("Building topology.");
        TridentTopology topology = new TridentTopology();