@Test public void testBusinessCanMarshalAGeneric() { TypeLiteral<BusinessEventMapped<UndefinedPayload>> eventType = new TypeLiteral<BusinessEventMapped<UndefinedPayload>>() { }; StreamBatchRecord<BusinessEventMapped<UndefinedPayload>> sbr = support.lineToEventStreamBatchRecord(businessLine, eventType.type(), observer); BusinessEventMapped<UndefinedPayload> event = sbr.streamBatch().events().get(0); assertEquals("36", sbr.streamBatch().cursor().offset()); assertEquals("0", sbr.streamBatch().cursor().partition()); EventMetadata metadata = event.metadata(); assertEquals("2016-09-20T21:52Z", metadata.occurredAt().toString()); assertEquals("a2ab0b7c-ee58-48e5-b96a-d13bce73d857", metadata.eid()); assertEquals("et-1", metadata.eventType()); assertEquals("0", metadata.partition()); assertEquals("2016-10-26T18:12:20.712Z", metadata.receivedAt().toString()); assertEquals("Nt0oU70k3UCNp2NKugrIF0QU", metadata.flowId()); assertEquals("a-compaction-key", metadata.partitionCompactionKey()); UndefinedPayload data = event.data(); assertEquals("1", data.id); assertEquals("2", data.foo); assertEquals("3", data.bar); }
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);  }