@Test
    void testHttpFetchScenario() throws Exception {
        AtomicInteger numParsed = new AtomicInteger();
        Map<String, Map<String, String>> result = Collections.synchronizedMap(new HashMap<>());
        StreamObserver<FetchAndParseReply> responseObserver = new StreamObserver<>() {
            @Override
            public void onNext(FetchAndParseReply fetchAndParseReply) {
                LOGGER.info("Parsed: {}", fetchAndParseReply.getFetchKey());
                numParsed.incrementAndGet();
                result.put(fetchAndParseReply.getFetchKey(), fetchAndParseReply.getFieldsMap());
            }

            @Override
            public void onError(Throwable throwable) {
                LOGGER.error("Error occurred", throwable);
            }

            @Override
            public void onCompleted() {
                LOGGER.info("Completed fetching.");
            }
        };
        StreamObserver<FetchAndParseRequest> request = tikaStub.fetchAndParseBiDirectionalStreaming(responseObserver);
        for (String file : files) {
            request.onNext(FetchAndParseRequest
                    .newBuilder()
                    .setFetcherId(httpFetcherId)
                    .setFetchKey(httpServerUrl + "/" + file)
                    .build());
        }
        request.onCompleted();

        Awaitility.await().atMost(Duration.ofSeconds(600)).until(() -> result.size() == files.size());

        Assertions.assertEquals(files.size(), numParsed.get());
    }
This test will start an HTTP server using jetty.
Then it will start Tika Pipes Grpc service.
Then it will, using a bidirectional stream of data, send urls to the
HTTP fetcher whilst simultaneously receiving parsed output as they parse.
tika-grpc/src/test/java/org/apache/tika/pipes/grpc/PipesBiDirectionalStreamingIntegrationTest.java