@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);
  }
}
 

/**
 * Specify the timestamp extractor implementation.
 *
 * @param timestampExtractor the {@link TimestampExtractor} implementation
 */
@SuppressWarnings("HiddenField")
public BaseWindowedBolt withTimestampExtractor(TimestampExtractor timestampExtractor) {
  if (this.timestampExtractor != null) {
    throw new IllegalArgumentException(
        "Window is already configured with a timestamp extractor: " + timestampExtractor);
  }

  this.timestampExtractor = new org.apache.heron.api.windowing.TimestampExtractor() {

    @Override
    public long extractTimestamp(Tuple tuple) {
      return timestampExtractor.extractTimestamp(new TupleImpl(tuple));
    }
  };
  return this;
}
 

@Test
public void testAck() throws Exception {
    setupExpectationsForTuple();
    setupExpectationsForTopologyContextNoEmit();

    EventCorrelatingOutputCollector sut = getSystemUnderTest();

    Tuple anchor = new TupleImpl(mockedTopologyContext, new Values(PARENT_STREAMLINE_EVENT), TASK_0,
            Utils.DEFAULT_STREAM_ID);

    sut.ack(anchor);

    new Verifications() {{
        mockedOutputCollector.ack(anchor); times = 1;
    }};
}
 

@Test
public void testFail() throws Exception {
    setupExpectationsForTuple();
    setupExpectationsForTopologyContextNoEmit();

    EventCorrelatingOutputCollector sut = getSystemUnderTest();

    Tuple anchor = new TupleImpl(mockedTopologyContext, new Values(PARENT_STREAMLINE_EVENT), TASK_0,
            Utils.DEFAULT_STREAM_ID);

    sut.fail(anchor);

    new Verifications() {{
        mockedOutputCollector.fail(anchor); times = 1;
    }};
}
 

@Test
public void testResetTimeout() throws Exception {
    setupExpectationsForTuple();
    setupExpectationsForTopologyContextNoEmit();

    EventCorrelatingOutputCollector sut = getSystemUnderTest();

    Tuple anchor = new TupleImpl(mockedTopologyContext, new Values(PARENT_STREAMLINE_EVENT), TASK_0,
            Utils.DEFAULT_STREAM_ID);

    sut.resetTimeout(anchor);

    new Verifications() {{
        mockedOutputCollector.resetTimeout(anchor); times = 1;
    }};
}
 

private static ArrayList<Tuple> makeStreamLineEventStream (String streamName, String[] fieldNames, Object[][] records) {

        MockTopologyContext mockContext = new MockTopologyContext(new String[]{StreamlineEvent.STREAMLINE_EVENT} );
        ArrayList<Tuple> result = new ArrayList<>(records.length);

        // convert each record into a HashMap using fieldNames as keys
        for (Object[] record : records) {
            HashMap<String,Object> recordMap = new HashMap<>( fieldNames.length );
            for (int i = 0; i < fieldNames.length; i++) {
                recordMap.put(fieldNames[i], record[i]);
            }
            StreamlineEvent streamLineEvent = StreamlineEventImpl.builder()
                    .fieldsAndValues(recordMap)
                    .dataSourceId("multiple sources")
                    .build();
            ArrayList<Object> tupleValues = new ArrayList<>(1);
            tupleValues.add(streamLineEvent);
            TupleImpl tuple = new TupleImpl(mockContext, tupleValues, 0, streamName);
            result.add( tuple );
        }

        return result;
    }
 

@Test
public void testSimpleRuleStringLiteral() throws Exception {
    StreamlineEvent event = StreamlineEventImpl.builder()
            .fieldsAndValues(ImmutableMap.<String, Object>of("foo", "Normal", "bar", "abc", "baz", 200))
            .dataSourceId("dsrcid")
            .build();
    Tuple tuple = new TupleImpl(mockContext, new Values(event), 1, "inputstream");

    doTest(readFile("/simple-rule-string-literal.json"), tuple);
    new Verifications() {
        {
            String streamId;
            Tuple anchor;
            List<List<Object>> tuples = new ArrayList<>();
            mockCollector.emit(streamId = withCapture(), anchor = withCapture(), withCapture(tuples));
            times=0;
        }
    };
}
 

@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);
  }
}
 

@Override
public org.apache.heron.api.windowing.TimestampExtractor getTimestampExtractor() {

  return (this.delegate.getTimestampExtractor() == null) ? null
      : new org.apache.heron.api.windowing.TimestampExtractor() {
        @Override
        public long extractTimestamp(org.apache.heron.api.tuple.Tuple tuple) {
          return delegate.getTimestampExtractor().extractTimestamp(new TupleImpl(tuple));
        }
      };
}
 

@Override
public void execute(Tuple input) {
    TupleImpl tuple = (TupleImpl) input;
    try {
        String line = tuple.getLong(0) + "," + tuple.getLong(1) + "," + tuple.getLong(2) + "\n";
        writer.write(line);
        System.err.print(line);
    } catch (IOException e) {
        log.error("Problem writing to file", e);
    }

}
 

private Tuple getTuple(int i) {
    StreamlineEvent event = StreamlineEventImpl.builder()
            .fieldsAndValues(ImmutableMap.of("foo", i, "bar", 100, "baz", 200))
            .dataSourceId("dsrcid")
            .build();
    return new TupleImpl(mockContext, new Values(event), 1, "inputstream");
}
 

private Tuple getWeather(String city, long temperature, long humidity) {
    StreamlineEvent event = StreamlineEventImpl.builder()
            .fieldsAndValues(ImmutableMap.of("city", city,
                    "temperature", temperature,
                    "humidity", humidity))
            .dataSourceId("dsrcid")
            .build();
    return new TupleImpl(mockContext, new Values(event), 1, "inputstream");
}
 

private Tuple getTupleNestedAndArray() {
    Map<String, Object> map = new HashMap<>();
    List<Person> person = Arrays.asList(new Person("a", 1), new Person("b", 2));
    Map<Integer, Integer> squares = new HashMap<>();
    squares.put(1, 1);
    squares.put(2, 4);
    squares.put(3, 9);
    map.put("person", person);
    map.put("squares", squares);
    StreamlineEvent event = StreamlineEventImpl.builder().fieldsAndValues(map).dataSourceId("dsrcid").build();
    return new TupleImpl(mockContext, new Values(event), 1, "inputstream");
}
 

private Tuple getTuple() {
    Map<String, Object> map = new HashMap<>();
    map.put("intfield", 2);
    map.put("stringfield1", "hello");
    map.put("stringfield2", "world");
    map.put("stringfield3", " space ");
    map.put("stringfield4", "aaaa");
    map.put("negativefield", -1.0);
    map.put("doublefield", 1.41);
    StreamlineEvent event = StreamlineEventImpl.builder().fieldsAndValues(map).dataSourceId("dsrcid").build();
    return new TupleImpl(mockContext, new Values(event), 1, "inputstream");
}
 

private Tuple getNestedTuple(int i) {
    StreamlineEvent event = StreamlineEventImpl.builder().fieldsAndValues(
            ImmutableMap.of("user", Collections.singletonMap("screen_name", "a" + i % 2), "id", i,
                    "retweeted", true)
    ).dataSourceId("dsrcid").build();

    return new TupleImpl(mockContext, new Values(event), 1, "inputstream");
}
 

public static Tuple makeTuple(Object... contents) {
    Tuple tuple = mock(TupleImpl.class);
    return pushInto(tuple, contents);
}
 

public static Tuple makeRawTuple(String component, String stream, Object... contents) {
    Tuple tuple = mock(TupleImpl.class);
    when(tuple.getSourceComponent()).thenReturn(component);
    when(tuple.getSourceStreamId()).thenReturn(stream);
    return pushInto(tuple, contents);
}
 

@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);
}
 

private Tuple getNextTuple(int i) {
    StreamlineEvent event = StreamlineEventImpl.builder().fieldsAndValues(
            ImmutableMap.of("empid", i, "salary", i * 10, "deptid", i/5)
    ).dataSourceId("dsrcid").build();
    return new TupleImpl(mockContext, new Values(event), 1, "inputstream");
}
 

private Tuple getTupleForSum(int i) {
    StreamlineEvent event = StreamlineEventImpl.builder().fieldsAndValues(
            ImmutableMap.of("id", 1, "intField", i, "longField", (long) i, "doubleField", (double)i)
    ).dataSourceId("dsrcid").build();
    return new TupleImpl(mockContext, new Values(event), 1, "inputstream");
}
 

private Tuple getNextTuple(int i) {
    StreamlineEvent event = StreamlineEventImpl.builder().fieldsAndValues(
            ImmutableMap.of("sentence", "THIS IS RANDOM SENTENCE"+ i)
    ).dataSourceId("dsrcid").build();
    return new TupleImpl(mockContext, new Values(event), 1, "inputstream");
}