SourceBuilder { @Nonnull public static <C> SourceBuilder<C>.Batch<Void> batch( @Nonnull String name, @Nonnull FunctionEx<? super Processor.Context, ? extends C> createFn ) { return new SourceBuilder<C>(name, createFn).new Batch<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 batch_fileSource() throws Exception { File textFile = createTestFile(); BatchSource<String> fileSource = SourceBuilder .batch("file-source", x -> fileReader(textFile)) .<String>fillBufferFn((in, buf) -> { String line = in.readLine(); if (line != null) { buf.add(line); } else { buf.close(); } }) .destroyFn(BufferedReader::close) .build(); Pipeline p = Pipeline.create(); p.readFrom(fileSource) .writeTo(sinkList()); jet().newJob(p).join(); assertEquals( IntStream.range(0, itemCount).mapToObj(i -> "line" + i).collect(toList()), new ArrayList<>(sinkList) ); }
@Test public void batch_fileSource_distributed() throws Exception { File textFile = createTestFile(); BatchSource<String> fileSource = SourceBuilder .batch("distributed-file-source", ctx -> fileReader(textFile)) .<String>fillBufferFn((in, buf) -> { String line = in.readLine(); if (line != null) { buf.add(line); } else { buf.close(); } }) .destroyFn(BufferedReader::close) .distributed(PREFERRED_LOCAL_PARALLELISM) .build(); Pipeline p = Pipeline.create(); p.readFrom(fileSource) .writeTo(sinkList()); jet().newJob(p).join(); Map<String, Integer> actual = sinkToBag(); Map<String, Integer> expected = IntStream.range(0, itemCount) .boxed() .collect(Collectors.toMap(i -> "line" + i, i -> PREFERRED_LOCAL_PARALLELISM * MEMBER_COUNT)); assertEquals(expected, actual); }
@Test public void stream_socketSource() throws IOException { try (ServerSocket serverSocket = new ServerSocket(0)) { startServer(serverSocket); int localPort = serverSocket.getLocalPort(); BatchSource<String> socketSource = SourceBuilder .batch("socket-source", ctx -> socketReader(localPort)) .<String>fillBufferFn((in, buf) -> { String line = in.readLine(); if (line != null) { buf.add(line); } else { buf.close(); } }) .destroyFn(BufferedReader::close) .build(); Pipeline p = Pipeline.create(); p.readFrom(socketSource) .writeTo(sinkList()); jet().newJob(p).join(); List<String> expected = IntStream.range(0, itemCount).mapToObj(i -> "line" + i).collect(toList()); assertEquals(expected, new ArrayList<>(sinkList)); } }
@Test public void stream_socketSource_distributed() throws IOException { try (ServerSocket serverSocket = new ServerSocket(0)) { startServer(serverSocket); int localPort = serverSocket.getLocalPort(); BatchSource<String> socketSource = SourceBuilder .batch("distributed-socket-source", ctx -> socketReader(localPort)) .<String>fillBufferFn((in, buf) -> { String line = in.readLine(); if (line != null) { buf.add(line); } else { buf.close(); } }) .destroyFn(BufferedReader::close) .distributed(PREFERRED_LOCAL_PARALLELISM) .build(); Pipeline p = Pipeline.create(); p.readFrom(socketSource) .writeTo(sinkList()); jet().newJob(p).join(); Map<String, Integer> expected = IntStream.range(0, itemCount) .boxed() .collect(Collectors.toMap(i -> "line" + i, i -> PREFERRED_LOCAL_PARALLELISM * MEMBER_COUNT)); assertEquals(expected, sinkToBag()); } }
@Test public void stream_socketSource_withTimestamps() throws IOException { try (ServerSocket serverSocket = new ServerSocket(0)) { startServer(serverSocket); int localPort = serverSocket.getLocalPort(); ToLongFunctionEx<String> timestampFn = line -> Long.valueOf(line.substring(LINE_PREFIX.length())); BatchSource<String> socketSource = SourceBuilder .batch("socket-source-with-timestamps", ctx -> socketReader(localPort)) .<String>fillBufferFn((in, buf) -> { String line = in.readLine(); if (line != null) { buf.add(line); } else { buf.close(); } }) .destroyFn(BufferedReader::close) .build(); Pipeline p = Pipeline.create(); p.readFrom(socketSource) .addTimestamps(timestampFn, 0) .window(tumbling(1)) .aggregate(AggregateOperations.counting()) .writeTo(sinkList()); jet().newJob(p).join(); List<WindowResult<Long>> expected = LongStream .range(1, itemCount + 1) .mapToObj(i -> new WindowResult<>(i - 1, i, 1L)) .collect(toList()); assertEquals(expected, new ArrayList<>(sinkList)); } }
@Test public void stream_distributed_socketSource_withTimestamps() throws IOException { try (ServerSocket serverSocket = new ServerSocket(0)) { startServer(serverSocket); int localPort = serverSocket.getLocalPort(); ToLongFunctionEx<String> timestampFn = line -> Long.valueOf(line.substring(LINE_PREFIX.length())); BatchSource<String> socketSource = SourceBuilder .batch("socket-source-with-timestamps", ctx -> socketReader(localPort)) .<String>fillBufferFn((in, buf) -> { String line = in.readLine(); if (line != null) { buf.add(line); } else { buf.close(); } }) .destroyFn(BufferedReader::close) .distributed(PREFERRED_LOCAL_PARALLELISM) .build(); Pipeline p = Pipeline.create(); p.readFrom(socketSource) .addTimestamps(timestampFn, 1000) .window(tumbling(1)) .aggregate(AggregateOperations.counting()) .writeTo(sinkList()); jet().newJob(p).join(); List<WindowResult<Long>> expected = LongStream.range(1, itemCount + 1) .mapToObj(i -> new WindowResult<>(i - 1, i, (long) PREFERRED_LOCAL_PARALLELISM * MEMBER_COUNT)) .collect(toList()); assertEquals(expected, new ArrayList<>(sinkList)); } }