@Test public void testDataChangeEventCanMarshalAGeneric() { TypeLiteral<DataChangeEvent<ID>> eventType = new TypeLiteral<DataChangeEvent<ID>>() { }; StreamBatchRecord<DataChangeEvent<ID>> sbr = support.lineToEventStreamBatchRecord(dataLine, eventType.type(), new LoggingStreamOffsetObserver()); assertEquals("95", sbr.streamBatch().cursor().offset()); assertEquals("0", sbr.streamBatch().cursor().partition()); DataChangeEvent<ID> event = sbr.streamBatch().events().get(0); assertEquals(event.getClass(), DataChangeEvent.class); assertEquals(event.data().getClass(), ID.class); assertEquals("1", event.data().id); assertEquals(DataChangeEvent.Op.C, event.op()); assertEquals("a2ab0b7c-ee58-48e5-b96a-d13bce73d857", event.metadata().eid()); assertEquals("2016-09-20T21:52Z", event.metadata().occurredAt().toString()); }
public <T> StreamBatchRecord<T> lineToEventStreamBatchRecord( String line, Type eventType, StreamOffsetObserver observer) { return new StreamBatchRecordReal<>(marshalEventStreamBatch(line, eventType), observer); }
JsonBatchSupport { public <T> StreamBatchRecord<T> lineToEventStreamBatchRecord( String line, Type eventType, StreamOffsetObserver observer) { return new StreamBatchRecordReal<>(marshalEventStreamBatch(line, eventType), observer); } }
JsonBatchSupport { public <T> StreamBatchRecord<T> lineToEventStreamBatchRecord( String line, Type eventType, StreamOffsetObserver observer) { return new StreamBatchRecordReal<>(marshalEventStreamBatch(line, eventType), observer); } JsonBatchSupport(JsonSupport jsonSupport); }
JsonBatchSupport { public <T> StreamBatchRecord<T> lineToEventStreamBatchRecord( String line, Type eventType, StreamOffsetObserver observer) { return new StreamBatchRecordReal<>(marshalEventStreamBatch(line, eventType), observer); } JsonBatchSupport(JsonSupport jsonSupport); StreamBatchRecord<T> lineToEventStreamBatchRecord(
      String line, Type eventType, StreamOffsetObserver observer); StreamBatchRecord<T> lineToSubscriptionStreamBatchRecord(
      String line, Type type, StreamOffsetObserver observer, String xNakadiStreamId,
      String subscriptionId); }
JsonBatchSupport { public <T> StreamBatchRecord<T> lineToEventStreamBatchRecord( String line, Type eventType, StreamOffsetObserver observer) { return new StreamBatchRecordReal<>(marshalEventStreamBatch(line, eventType), observer); } JsonBatchSupport(JsonSupport jsonSupport); StreamBatchRecord<T> lineToEventStreamBatchRecord(
      String line, Type eventType, StreamOffsetObserver observer); StreamBatchRecord<T> lineToSubscriptionStreamBatchRecord(
      String line, Type type, StreamOffsetObserver observer, String xNakadiStreamId,
      String subscriptionId);  }