import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
@Override
public void execute(TupleWindow inputWindow) {
    int sum = 0;
    List<Tuple> tuplesInWindow = inputWindow.get();
    LOG.debug("Events in current window: " + tuplesInWindow.size());
    if (tuplesInWindow.size() > 0) {
        /*
        * Since this is a tumbling window calculation,
        * we use all the tuples in the window to compute the avg.
        */
        for (Tuple tuple : tuplesInWindow) {
            sum += (Integer) tuple.getValue(0);
        }
        collector.emit(new Values(sum / tuplesInWindow.size()));
    }
}
 

import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
@Override
public void execute(TupleWindow tw) {
    HashMap<String, String> ads = new HashMap<String, String>();
    HashMap<String, String> web = new HashMap<String, String>();

    for (Tuple tuple : tw.get()) {
        if (tuple.getSourceComponent().equals("ad_split")) {
            ads.put((String) tuple.getValues().get(0), (String) tuple.getValues().get(1));
        }
        if (tuple.getSourceComponent().equals("web_split")) {
            web.put((String) tuple.getValues().get(0), (String) tuple.getValues().get(1));
        }
        //System.out.println(tuple.toString());
    }
    for (String userID : ads.keySet()) {
        if (web.get(userID) != null) {
            collector.emit(new Values(ads.get(userID), web.get(userID)));
        }
    }

}
 

import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
/**
 * Process the tuple window and optionally emit new tuples based on the tuples in the input window.
 *
 * @param inputWindow
 */
@Override
public void execute(TupleWindow inputWindow) {
    ++windowId;
    LOG.debug("Window activated, window id {}, number of tuples in window {}", windowId, inputWindow.get().size());

    Map<String, Tuple> eventIdToTupleMap = new HashMap<>();
    try {
        StreamlineEvent event;
        for (Tuple input : inputWindow.get()) {
            if ((event = getStreamlineEventFromTuple(input)) != null) {
                LOG.debug("++++++++ Executing tuple [{}] which contains StreamlineEvent [{}]", input, event);
                eventIdToTupleMap.put(event.getId(), input);
                processAndEmit(event, eventIdToTupleMap);
            }
        }
        // force evaluation of the last group by
        processAndEmit(GROUP_BY_TRIGGER_EVENT, eventIdToTupleMap);

        // current group is processed and result emitted
        eventIdToTupleMap.clear();
    } catch (Exception e) {
        collector.reportError(e);
        LOG.error("", e);
    }
}
 

import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
private boolean doTest(String rulesJson, int expectedExecuteCount, Function<Integer, Tuple> tupleGen) throws Exception {
    RulesProcessor rulesProcessor = Utils.createObjectFromJson(rulesJson, RulesProcessor.class);
    Window windowConfig = rulesProcessor.getRules().get(0).getWindow();
    final CountDownLatch latch = new CountDownLatch(expectedExecuteCount);
    WindowRulesBolt wb = new WindowRulesBolt(rulesJson, RuleProcessorRuntime.ScriptType.SQL) {
        @Override
        public void execute(TupleWindow inputWindow) {
            super.execute(inputWindow);
            latch.countDown();
        }
    };
    wb.withWindowConfig(windowConfig);
    WindowedBoltExecutor wbe = new WindowedBoltExecutor(wb);
    Map<String, Object> conf = wb.getComponentConfiguration();
    conf.put("topology.message.timeout.secs", 30);
    wbe.prepare(conf, mockContext, mockCollector);
    Thread.sleep(100);
    for (int i = 1; i <= 20; i++) {
        wbe.execute(tupleGen.apply(i));
    }
    // wait for up to 5 secs for the bolt's execute to finish
    return latch.await(5, TimeUnit.SECONDS);
}
 

import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
@Override
public void execute(TupleWindow inputWindow) {
  int sum = 0;
  List<Tuple> tuplesInWindow = inputWindow.get();
  LOG.fine("Events in current window: " + tuplesInWindow.size());
  if (tuplesInWindow.size() > 0) {
            /*
            * Since this is a tumbling window calculation,
            * we use all the tuples in the window to compute the avg.
            */
    for (Tuple tuple : tuplesInWindow) {
      sum += (int) tuple.getValue(0);
    }
    collector.emit(new Values(sum / tuplesInWindow.size()));
  }
}
 

import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
@Override
public void execute(TupleWindow inputWindow) {
    int sum = 0;
    List<Tuple> tuplesInWindow = inputWindow.get();
    LOG.debug("Events in current window: " + tuplesInWindow.size());
    if (tuplesInWindow.size() > 0) {
        /*
        * Since this is a tumbling window calculation,
        * we use all the tuples in the window to compute the avg.
        */
        for (Tuple tuple : tuplesInWindow) {
            sum += (int) tuple.getValue(0);
        }
        collector.emit(new Values(sum / tuplesInWindow.size()));
    }
}
 

import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
@Override
    public void execute(TupleWindow inputWindow) {
            /*
             * The inputWindow gives a view of
             * (a) all the events in the window
             * (b) events that expired since last activation of the window
             * (c) events that newly arrived since last activation of the window
             */
        List<Tuple> tuplesInWindow = inputWindow.get();
        List<Tuple> newTuples = inputWindow.getNew();
        List<Tuple> expiredTuples = inputWindow.getExpired();

        LOG.debug("Events in current window: " + tuplesInWindow.size());
        if (expiredTuples.size() > 0) {
            System.out.println(newTuples.size());
            System.out.println(expiredTuples.size());
        }
            /*
             * Instead of iterating over all the tuples in the window to compute
             * the sum, the values for the new events are added and old events are
             * subtracted. Similar optimizations might be possible in other
             * windowing computations.
             */
        for (Tuple tuple : newTuples) {
            sum = (Integer) tuple.getValue(0);
        }
//        for (Tuple tuple : expiredTuples) {
//            sum -= (int) tuple.getValue(0);
//        }
        collector.emit(new Values(sum));
    }
 

import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
@Test
public void testNestedKeys_trivial() throws Exception {
    ArrayList<Tuple> userStream = makeStreamLineEventStream("users", userFields, users);
    TupleWindow window = makeTupleWindow(userStream);
    WindowedQueryBolt bolt = new WindowedQueryBolt("users", SL_PREFIX + "userId")
            .selectStreamLine("name,users:city, users:city as cityagain");
    MockTopologyContext context = new MockTopologyContext(new String[]{StreamlineEvent.STREAMLINE_EVENT});
    MockCollector collector = new MockCollector();
    bolt.prepare(null, context, collector);
    bolt.execute(window);
    printResults_StreamLine(collector);
    Assert.assertEquals( userStream.size(), collector.actualResults.size() );
}
 

import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
@Test
public void testNestedKeys_StreamLine() throws Exception {
    ArrayList<Tuple> userStream = makeStreamLineEventStream("users", userFields, users);
    ArrayList<Tuple> cityStream = makeStreamLineEventStream("cities", cityFields, cities);
    TupleWindow window = makeTupleWindow(userStream, cityStream);
    WindowedQueryBolt bolt = new WindowedQueryBolt("users", "city")
            .join("cities", "cityName", "users")
            .selectStreamLine("name, users:city as city, cities:country");
    MockTopologyContext context = new MockTopologyContext(new String[]{StreamlineEvent.STREAMLINE_EVENT});
    MockCollector collector = new MockCollector();
    bolt.prepare(null, context, collector);
    bolt.execute(window);
    printResults_StreamLine(collector);
    Assert.assertEquals( cityStream.size(), collector.actualResults.size() );
}
 

import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
private static TupleWindow makeTupleWindow(ArrayList<Tuple>... streams) {
    ArrayList<Tuple> combined = null;
    for (int i = 0; i < streams.length; i++) {
        if(i==0) {
            combined = new ArrayList<>(streams[0]);
        } else {
            combined.addAll(streams[i]);
        }
    }
    return new TupleWindowImpl(combined, null, null);
}
 

import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
@Override
public void execute(TupleWindow inputWindow) {
    for (Tuple tuple : inputWindow.get()) {
        sum += tuple.getIntegerByField("value");
    }
    state.put("sum", sum);
    collector.emit(new Values(sum));
}
 

import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
@Override
public void execute(TupleWindow tupleWindow) {
    List<Tuple> tuplesInWindow = tupleWindow.get();
    List<Tuple> newTuples = tupleWindow.getNew();
    List<Tuple> expiredTuples = tupleWindow.getExpired();
    LOG.debug("Events in current window: " + tuplesInWindow.size());
    if(expiredTuples.size() > 0){
        System.out.println(newTuples.size());
        System.out.println(expiredTuples.size());
    }

    String OriginTabName = JoinCondition.originTabName;  // 获取被连接表表名
    String JoinTabName = "";  // 获取连接表表名
    Iterator<String> iterator = JoinCondition.linkTablemap.keySet().iterator();
    String JoinOP = "";  // join的方式，是Left，Right或者Inner
    String compareCol = ""; // 获取连接的条件，如JData_Action_201602.sku_id = JData_Action_201603.sku_id，则该项为sku_id
    while (iterator.hasNext()){
        String A = iterator.next();
        JoinTabName = A.split("\\|")[0];
        JoinOP = A.split("\\|")[1];
        compareCol = JoinCondition.linkTablemap.get(A).getTcItemRight().getColName();

        System.out.println(A + "  " +OriginTabName+ "  " + JoinTabName + "  "+ JoinOP+ "  " + compareCol);
    }
    for(Tuple tuple: newTuples){
        System.out.println(tuple.getValue(0));
        if(tuple.getValue(0).toString().equals(OriginTabName)){
            List_originTab.add(tuple);

        }
        else if(tuple.getValue(0).toString().equals(JoinTabName)){
            List_joinTab.add(tuple);

        }
    }

    if(JoinOP.equals("Left")) {
        Left_Join(compareCol);
    }
    else if(JoinOP.equals("Right")){
        Right_Join(compareCol);
    }
    else if(JoinOP.equals("Inner")){
        Inner_Join(compareCol);
    }

}
 

import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
@Override
public void execute(TupleWindow inputWindow) {

    CacheManager cacheManager = new CacheManager();

    List<Long> timeLine = new ArrayList<>();
    for(Tuple tuple: inputWindow.get()) {
        int type = tuple.getIntegerByField("type");
        JSONObject data = (JSONObject)tuple.getValueByField("data");
        switch (type){
            case Constants.TYPE_CALLER_INFO:
                CallerInformation info1 = data.toJavaObject(CallerInformation.class);
                String key1 = info1.getRequest_id();
                cacheManager.putServiceCallersCache(info1.getService_id(), key1, info1.getCaller_ip());
                cacheManager.putServiceRequestsCache(info1.getService_id(), info1.getRequest_id());
                break;
            case Constants.TYPE_RESP_TIME:
                ResponseTime info2 = data.toJavaObject(ResponseTime.class);
                String key2 = info2.getRequest_id();
                cacheManager.putServiceCostsCache(info2.getService_id(), key2, info2.getTotal_cost());
                cacheManager.putServiceRequestsCache(info2.getService_id(), info2.getRequest_id());
                timeLine.add(info2.getBegin_time());
                break;
            case Constants.TYPE_RESP_PARAM:
                ResponseParameters info3 = data.toJavaObject(ResponseParameters.class);
                String key3 = info3.getRequest_id();
                cacheManager.putServiceCodesCache(info3.getService_id(), key3, info3.getResponse_code());
                cacheManager.putServiceRequestsCache(info3.getService_id(), info3.getRequest_id());
                break;
            case Constants.TYPE_SERVICE_INFO:
                ServiceInformation info4 = data.toJavaObject(ServiceInformation.class);
                cacheManager.putServerServicesCache(info4.getServer_ip(), info4.getService_id());
                cacheManager.putServiceRequestsCache(info4.getService_id(), info4.getRequest_id());
                break;
            default:
                logger.warn("error type:{}",type);
                break;
        }
    }
    //save to redis:
    cacheManager.saveRedis();
    Collections.sort(timeLine);
    Long start = timeLine.get(0);
    Long end = timeLine.get(timeLine.size()-1);
    // emit the results
    collector.emit(new Values(start, end));
}
 

import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
@Override
public void execute(TupleWindow tupleWindow) {
    processorBolt.execute(tupleWindow);
}
 

import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
@Override
public void execute(TupleWindow inputWindow) {
  collector.emit(new Values(inputWindow.get().size()));
}
 

import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
@Override
public void execute(TupleWindow inputWindow) {

}
 

import org.apache.storm.windowing.TupleWindow; //导入依赖的package包/类
/**
 * Process the tuple window and optionally emit new tuples based on the tuples in the input
 * window.
 */
void execute(TupleWindow inputWindow);