SourceBuilder { @Nonnull public static <C> SourceBuilder<C>.TimestampedStream<Void> timestampedStream( @Nonnull String name, @Nonnull FunctionEx<? super Processor.Context, ? extends C> createFn ) { return new SourceBuilder<C>(name, createFn).new TimestampedStream<Void>(); } private  SourceBuilder(
            @Nonnull String name,
            @Nonnull FunctionEx<? super Context, ? extends C> createFn
    ); @Nonnull static SourceBuilder<C>.Batch<Void> batch(
            @Nonnull String name,
            @Nonnull FunctionEx<? super Processor.Context, ? extends C> createFn
    ); @Nonnull static SourceBuilder<C>.Stream<Void> stream(
            @Nonnull String name,
            @Nonnull FunctionEx<? super Processor.Context, ? extends C> createFn
    ); @Nonnull static SourceBuilder<C>.TimestampedStream<Void> timestampedStream(
            @Nonnull String name,
            @Nonnull FunctionEx<? super Processor.Context, ? extends C> createFn
    );  }
@Test public void stream_socketSource_withTimestamps_andLateness() throws IOException { try (ServerSocket serverSocket = new ServerSocket(0)) { startServer(serverSocket); int localPort = serverSocket.getLocalPort(); FunctionEx<String, Long> timestampFn = line -> Long.valueOf(line.substring(LINE_PREFIX.length())); int lateness = 10; long lastExpectedTs = itemCount - lateness; StreamSource<String> socketSource = SourceBuilder .timestampedStream("socket-source-with-timestamps", ctx -> socketReader(localPort)) .<String>fillBufferFn((in, buf) -> { String line = in.readLine(); if (line != null) { long ts = timestampFn.apply(line); buf.add(line, ts); if (ts >= lastExpectedTs) { System.out.println(line); } } }) .destroyFn(BufferedReader::close) .build(); Pipeline p = Pipeline.create(); p.readFrom(socketSource) .withNativeTimestamps(lateness) .window(tumbling(1)) .aggregate(AggregateOperations.counting()) .writeTo(sinkList()); jet().newJob(p); List<WindowResult<Long>> expected = LongStream.range(1, itemCount - lateness) .mapToObj(i -> new WindowResult<>(i - 1, i, 1L)) .collect(toList()); assertTrueEventually(() -> assertEquals(expected, new ArrayList<>(sinkList)), 10); } }
@Test public void test_faultTolerance_snapshotWithUserDefinedObject() { StreamSource<WrappedInt> source = SourceBuilder .timestampedStream("src", ctx -> new WrappedNumberGeneratorContext()) .<WrappedInt>fillBufferFn((src, buffer) -> { for (int i = 0; i < 100; i++) { buffer.add(src.current, src.current.value); src.current = new WrappedInt(src.current.value + 1); } Thread.sleep(100); }) .createSnapshotFn(src -> { System.out.println("Will save " + src.current.value + " to snapshot"); return src; }) .restoreSnapshotFn((src, states) -> { assert states.size() == 1; src.restore(states.get(0)); System.out.println("Restored " + src.current.value + " from snapshot"); }) .build(); testFaultTolerance(source); }