// Path: src/main/java/storm/benchmark/lib/bolt/RollingBolt.java
// public abstract class RollingBolt extends BaseBasicBolt {
// 
//   private static final long serialVersionUID = 5996017217250945842L;
//   public static final int DEFAULT_NUM_WINDOW_CHUNKS = 5;
//   public static final int DEFAULT_SLIDING_WINDOW_IN_SECONDS = DEFAULT_NUM_WINDOW_CHUNKS * 60;
//   public static final int DEFAULT_EMIT_FREQUENCY_IN_SECONDS = DEFAULT_SLIDING_WINDOW_IN_SECONDS / DEFAULT_NUM_WINDOW_CHUNKS;
// 
//   protected final int windowLengthInSeconds;
//   protected final int emitFrequencyInSeconds;
// 
//   public RollingBolt(int windowLengthInSeconds, int emitFrequencyInSeconds) {
//     this.windowLengthInSeconds = windowLengthInSeconds;
//     this.emitFrequencyInSeconds = emitFrequencyInSeconds;
//   }
// 
//   protected int getWindowChunks() {
//     return windowLengthInSeconds / emitFrequencyInSeconds;
//   }
// 
//   @Override
//   public void execute(Tuple tuple, BasicOutputCollector collector) {
//     if (TupleHelpers.isTickTuple(tuple)) {
//       emitCurrentWindow(collector);
//     } else {
//       updateCurrentWindow(tuple);
//     }
//   }
// 
//   public abstract void emitCurrentWindow(BasicOutputCollector collector);
// 
//   public abstract void updateCurrentWindow(Tuple tuple);
// 
//   @Override
//   public Map<String, Object> getComponentConfiguration() {
//     Map<String, Object> conf = new HashMap<String, Object>();
//     conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, emitFrequencyInSeconds);
//     return conf;
//   }
// }
// 
// Path: src/main/java/storm/benchmark/lib/bolt/RollingCountBolt.java
// public class RollingCountBolt extends RollingBolt {
// 
//   private static final long serialVersionUID = -903093673694769540L;
//   private static final Logger LOG = Logger.getLogger(RollingCountBolt.class);
//   public static final String FIELDS_OBJ = "obj";
//   public static final String FIELDS_CNT = "count";
// 
//   private final SlidingWindow<Object, Long> window;
// 
//   public RollingCountBolt() {
//     this(DEFAULT_SLIDING_WINDOW_IN_SECONDS, DEFAULT_EMIT_FREQUENCY_IN_SECONDS);
//   }
// 
//   public RollingCountBolt(int winLen, int emitFreq) {
//     super(winLen, emitFreq);
//     window = new SlidingWindow<Object, Long>(new LongSummer(), getWindowChunks());
//   }
// 
//   @Override
//   public void emitCurrentWindow(BasicOutputCollector collector) {
//     emitCurrentWindowCounts(collector);
//   }
// 
//   @Override
//   public void updateCurrentWindow(Tuple tuple) {
//     countObj(tuple);
//   }
// 
//   private void emitCurrentWindowCounts(BasicOutputCollector collector) {
//     Map<Object, Long> counts = window.reduceThenAdvanceWindow();
//     for (Entry<Object, Long> entry : counts.entrySet()) {
//       Object obj = entry.getKey();
//       Long count = entry.getValue();
//       LOG.info(String.format("get %d %s in last %d seconds", count, obj, windowLengthInSeconds));
//       collector.emit(new Values(obj, count));
//     }
//   }
// 
//   private void countObj(Tuple tuple) {
//     Object obj = tuple.getValue(0);
//     window.add(obj, (long) 1);
//   }
// 
//   @Override
//   public void declareOutputFields(OutputFieldsDeclarer declarer) {
//     declarer.declare(new Fields(FIELDS_OBJ, FIELDS_CNT));
//   }
// }
// 
// Path: src/test/java/storm/benchmark/util/MockTupleHelpers.java
// public final class MockTupleHelpers {
//   public static final String ANY_COMPONENT_ID = "any_component_id";
//   public static final String ANY_STREAM_ID = "any_stream_id";
// 
//   private MockTupleHelpers() {
//   }
// 
//   public static Tuple mockTickTuple() {
//     return mockTuple(Constants.SYSTEM_COMPONENT_ID, Constants.SYSTEM_TICK_STREAM_ID);
//   }
// 
//   public static Tuple mockAnyTuple() {
//     return mockTuple(ANY_COMPONENT_ID, ANY_STREAM_ID);
//   }
// 
//   public static Tuple mockTuple(String componentId, String streamId) {
//     Tuple tuple = mock(Tuple.class);
//     when(tuple.getSourceComponent()).thenReturn(componentId);
//     when(tuple.getSourceStreamId()).thenReturn(streamId);
//     return tuple;
//   }
// }


import java.util.Map;
import static org.fest.assertions.api.Assertions.assertThat;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.*;
import backtype.storm.Config;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import storm.benchmark.lib.bolt.RollingBolt;
import storm.benchmark.lib.bolt.RollingCountBolt;
import storm.benchmark.lib.bolt.UniqueVisitorBolt;
import storm.benchmark.util.MockTupleHelpers;
/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License
 */

package storm.benchmark.lib.bolt.common;




public class RollingBoltTest {
  private static final String ANY_NON_SYSTEM_COMPONENT_ID = "irrelevant_component_id";
  private static final String ANY_NON_SYSTEM_STREAM_ID = "irrelevant_stream_id";

  private Tuple mockNormalTuple(Object obj) {
// Path: src/main/java/storm/benchmark/lib/bolt/RollingBolt.java
// public abstract class RollingBolt extends BaseBasicBolt {
// 
//   private static final long serialVersionUID = 5996017217250945842L;
//   public static final int DEFAULT_NUM_WINDOW_CHUNKS = 5;
//   public static final int DEFAULT_SLIDING_WINDOW_IN_SECONDS = DEFAULT_NUM_WINDOW_CHUNKS * 60;
//   public static final int DEFAULT_EMIT_FREQUENCY_IN_SECONDS = DEFAULT_SLIDING_WINDOW_IN_SECONDS / DEFAULT_NUM_WINDOW_CHUNKS;
// 
//   protected final int windowLengthInSeconds;
//   protected final int emitFrequencyInSeconds;
// 
//   public RollingBolt(int windowLengthInSeconds, int emitFrequencyInSeconds) {
//     this.windowLengthInSeconds = windowLengthInSeconds;
//     this.emitFrequencyInSeconds = emitFrequencyInSeconds;
//   }
// 
//   protected int getWindowChunks() {
//     return windowLengthInSeconds / emitFrequencyInSeconds;
//   }
// 
//   @Override
//   public void execute(Tuple tuple, BasicOutputCollector collector) {
//     if (TupleHelpers.isTickTuple(tuple)) {
//       emitCurrentWindow(collector);
//     } else {
//       updateCurrentWindow(tuple);
//     }
//   }
// 
//   public abstract void emitCurrentWindow(BasicOutputCollector collector);
// 
//   public abstract void updateCurrentWindow(Tuple tuple);
// 
//   @Override
//   public Map<String, Object> getComponentConfiguration() {
//     Map<String, Object> conf = new HashMap<String, Object>();
//     conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, emitFrequencyInSeconds);
//     return conf;
//   }
// }
// 
// Path: src/main/java/storm/benchmark/lib/bolt/RollingCountBolt.java
// public class RollingCountBolt extends RollingBolt {
// 
//   private static final long serialVersionUID = -903093673694769540L;
//   private static final Logger LOG = Logger.getLogger(RollingCountBolt.class);
//   public static final String FIELDS_OBJ = "obj";
//   public static final String FIELDS_CNT = "count";
// 
//   private final SlidingWindow<Object, Long> window;
// 
//   public RollingCountBolt() {
//     this(DEFAULT_SLIDING_WINDOW_IN_SECONDS, DEFAULT_EMIT_FREQUENCY_IN_SECONDS);
//   }
// 
//   public RollingCountBolt(int winLen, int emitFreq) {
//     super(winLen, emitFreq);
//     window = new SlidingWindow<Object, Long>(new LongSummer(), getWindowChunks());
//   }
// 
//   @Override
//   public void emitCurrentWindow(BasicOutputCollector collector) {
//     emitCurrentWindowCounts(collector);
//   }
// 
//   @Override
//   public void updateCurrentWindow(Tuple tuple) {
//     countObj(tuple);
//   }
// 
//   private void emitCurrentWindowCounts(BasicOutputCollector collector) {
//     Map<Object, Long> counts = window.reduceThenAdvanceWindow();
//     for (Entry<Object, Long> entry : counts.entrySet()) {
//       Object obj = entry.getKey();
//       Long count = entry.getValue();
//       LOG.info(String.format("get %d %s in last %d seconds", count, obj, windowLengthInSeconds));
//       collector.emit(new Values(obj, count));
//     }
//   }
// 
//   private void countObj(Tuple tuple) {
//     Object obj = tuple.getValue(0);
//     window.add(obj, (long) 1);
//   }
// 
//   @Override
//   public void declareOutputFields(OutputFieldsDeclarer declarer) {
//     declarer.declare(new Fields(FIELDS_OBJ, FIELDS_CNT));
//   }
// }
// 
// Path: src/test/java/storm/benchmark/util/MockTupleHelpers.java
// public final class MockTupleHelpers {
//   public static final String ANY_COMPONENT_ID = "any_component_id";
//   public static final String ANY_STREAM_ID = "any_stream_id";
// 
//   private MockTupleHelpers() {
//   }
// 
//   public static Tuple mockTickTuple() {
//     return mockTuple(Constants.SYSTEM_COMPONENT_ID, Constants.SYSTEM_TICK_STREAM_ID);
//   }
// 
//   public static Tuple mockAnyTuple() {
//     return mockTuple(ANY_COMPONENT_ID, ANY_STREAM_ID);
//   }
// 
//   public static Tuple mockTuple(String componentId, String streamId) {
//     Tuple tuple = mock(Tuple.class);
//     when(tuple.getSourceComponent()).thenReturn(componentId);
//     when(tuple.getSourceStreamId()).thenReturn(streamId);
//     return tuple;
//   }
// }


// Path: src/test/java/storm/benchmark/lib/bolt/common/RollingBoltTest.java
import java.util.Map;
import static org.fest.assertions.api.Assertions.assertThat;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.*;
import backtype.storm.Config;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import storm.benchmark.lib.bolt.RollingBolt;
import storm.benchmark.lib.bolt.RollingCountBolt;
import storm.benchmark.lib.bolt.UniqueVisitorBolt;
import storm.benchmark.util.MockTupleHelpers;

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License
 */

package storm.benchmark.lib.bolt.common;




public class RollingBoltTest {
  private static final String ANY_NON_SYSTEM_COMPONENT_ID = "irrelevant_component_id";
  private static final String ANY_NON_SYSTEM_STREAM_ID = "irrelevant_stream_id";

  private Tuple mockNormalTuple(Object obj) {
    Tuple tuple = MockTupleHelpers.mockTuple(ANY_NON_SYSTEM_COMPONENT_ID, ANY_NON_SYSTEM_STREAM_ID);