@Test public void testTypeLiteralCanMarshalAnEmptyBatch() { TypeLiteral<DataChangeEvent<ID>> eventType = new TypeLiteral<DataChangeEvent<ID>>() { }; String empty = "{\"cursor\":{\"partition\":\"0\",\"offset\":\"95\"}}"; StreamBatchRecord<Object> sbr1 = support.lineToEventStreamBatchRecord(empty, eventType.type(), new LoggingStreamOffsetObserver()); assertEquals("95", sbr1.streamBatch().cursor().offset()); assertEquals("0", sbr1.streamBatch().cursor().partition()); assertTrue(sbr1.streamBatch().events().size() == 0); }
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);  }