@Override
public void emitDirect(
    int taskId, String streamId, Collection<Tuple> anchors, List<Object> tuple) {
  if (anchors != null) {
    ArrayList<org.apache.heron.api.tuple.Tuple> l =
        new ArrayList<org.apache.heron.api.tuple.Tuple>();
    for (Tuple t : anchors) {
      TupleImpl i = (TupleImpl) t;
      l.add(i.getDelegate());
    }
    delegate.emitDirect(taskId, streamId, l, tuple);
  } else {
    delegate.emitDirect(
        taskId, streamId, (Collection<org.apache.heron.api.tuple.Tuple>) null, tuple);
  }
}
 

private void sendData(AlertBolt alertBolt, Map<String, StreamDefinition> definitionMap, PolicyDefinition policyDefinition) {
    StreamDefinition definition = definitionMap.get("perfmon_latency_stream");
    long base = System.currentTimeMillis();
    for (int i = 0; i < 2; i++) {
        long time = base + i * 1000;

        Map<String, Object> mapdata = new HashMap<>();
        mapdata.put("host", "host-1");
        mapdata.put("timestamp", time);
        mapdata.put("metric", "perfmon_latency");
        mapdata.put("pool", "raptor");
        mapdata.put("value", 1000.0 + i * 1000.0);
        mapdata.put("colo", "phx");

        StreamEvent event = StreamEvent.builder().timestamep(time).attributes(mapdata, definition).build();
        PartitionedEvent pEvent = new PartitionedEvent(event, policyDefinition.getPartitionSpec().get(0), 1);

        GeneralTopologyContext mock = Mockito.mock(GeneralTopologyContext.class);

        Mockito.when(mock.getComponentId(1)).thenReturn("taskId");
        Mockito.when(mock.getComponentOutputFields("taskId", "test-stream-id")).thenReturn(new Fields(AlertConstants.FIELD_0));

        TupleImpl ti = new TupleImpl(mock, Collections.singletonList(pEvent), 1, "test-stream-id");
        alertBolt.execute(ti);
    }
}
 

private Tuple createTuple(AlertBolt bolt, String version) throws IOException {
    GeneralTopologyContext context = mock(GeneralTopologyContext.class);
    int taskId = 1;
    when(context.getComponentId(taskId)).thenReturn("comp1");
    when(context.getComponentOutputFields("comp1", TEST_STREAM)).thenReturn(new Fields("f0"));
    // case 1: bolt prepared but metadata not initialized (no bolt.onAlertBoltSpecChange)
    PartitionedEvent pe = new PartitionedEvent();
    pe.setPartitionKey(1);
    pe.setPartition(createPartition());
    StreamEvent streamEvent = new StreamEvent();
    streamEvent.setStreamId(TEST_STREAM);
    streamEvent.setTimestamp(System.currentTimeMillis());
    streamEvent.setMetaVersion(version);
    pe.setEvent(streamEvent);

    PartitionedEventSerializerImpl peSer = new PartitionedEventSerializerImpl(bolt);
    byte[] serializedEvent = peSer.serialize(pe);
    return new TupleImpl(context, Collections.singletonList(serializedEvent), taskId, TEST_STREAM);
}
 

@Override
public List<Integer> emit(String streamId, Collection<Tuple> anchors, List<Object> tuple) {
  if (anchors != null) {
    ArrayList<org.apache.heron.api.tuple.Tuple> l =
        new ArrayList<org.apache.heron.api.tuple.Tuple>();
    for (Tuple t : anchors) {
      TupleImpl i = (TupleImpl) t;
      l.add(i.getDelegate());
    }
    return delegate.emit(streamId, l, tuple);
  } else {
    return delegate.emit(streamId, (Collection<org.apache.heron.api.tuple.Tuple>) null, tuple);
  }
}
 

public static void main(String[] args) {
    // o o o o o o o o o o
    // 0 1 2 3 4 5 6 7 8 9
    TumblingCountWindows<Tuple> slideCountAssigner = TumblingCountWindows.create(3);
    for (int i = 0; i < 10; i++) {
        System.out.println("i=" + i);
        Collection<TimeWindow> windows = slideCountAssigner.assignWindows(new TupleImpl(), 0);
        for (TimeWindow window : windows) {
            System.out.println(window);
        }
        System.out.println();
    }
}
 

public static void main(String[] args) {
    // o o o o o o o o o o
    // 0 1 2 3 4 5 6 7 8 9
    SlidingCountWindows<Tuple> slideCountAssigner = SlidingCountWindows.create(3, 1);
    for (int i = 0; i < 10; i++) {
        System.out.println("i=" + i);
        Collection<TimeWindow> windows = slideCountAssigner.assignWindows(new TupleImpl(), 0);
        for (TimeWindow window : windows) {
            System.out.println(window);
        }
        System.out.println();
    }
}
 

private Tuple generateTestTuple(Object key, Object message) {
    TopologyBuilder builder = new TopologyBuilder();
    GeneralTopologyContext topologyContext = new GeneralTopologyContext(builder.createTopology(), new Config(), new HashMap(), new HashMap(), new HashMap(), "") {
        @Override
        public Fields getComponentOutputFields(String componentId, String streamId) {
            return new Fields("key", "message");
        }
    };
    return new TupleImpl(topologyContext, new Values(key, message), 1, "");
}
 

private Tuple generateTestTuple(Object message) {
    TopologyBuilder builder = new TopologyBuilder();
    GeneralTopologyContext topologyContext = new GeneralTopologyContext(builder.createTopology(), new Config(), new HashMap(), new HashMap(), new HashMap(), "") {
        @Override
        public Fields getComponentOutputFields(String componentId, String streamId) {
            return new Fields("message");
        }
    };
    return new TupleImpl(topologyContext, new Values(message), 1, "");
}
 

@Override
public void ack(Tuple input) {
  TupleImpl i = (TupleImpl) input;
  delegate.ack(i.getDelegate());
}
 

@Override
public void fail(Tuple input) {
  TupleImpl i = (TupleImpl) input;
  delegate.fail(i.getDelegate());
}
 

@Override
public void execute(org.apache.heron.api.tuple.Tuple tuple) {
  TupleImpl impl = new TupleImpl(tuple);
  delegate.execute(impl);
}