@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<>();
    }
@Test
    public void stream_socketSource() throws IOException {
        // Given
        try (ServerSocket serverSocket = new ServerSocket(0)) {
            startServer(serverSocket);

            // When
            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();

            // Then
            Pipeline p = Pipeline.create();
            p.readFrom(socketSource)
             .writeTo(sinkList());
            hz().getJet().newJob(p).join();
            List<String> expected = IntStream.range(0, itemCount).mapToObj(i -> "line" + i).collect(toList());
            assertEquals(expected, new ArrayList<>(sinkList));
        }
    }