public void start() throws Exception {
        HealthStatusManager healthStatusManager = new HealthStatusManager();
        ServerCredentials creds;
        if (secure) {
            TlsServerCredentials.Builder channelCredBuilder = TlsServerCredentials.newBuilder();
            channelCredBuilder.keyManager(certChain, privateKey, privateKeyPassword);
            if (trustCertCollection != null && trustCertCollection.exists()) {
                channelCredBuilder.trustManager(trustCertCollection);
                if (clientAuthRequired) {
                    channelCredBuilder.clientAuth(TlsServerCredentials.ClientAuth.REQUIRE);
                }
            }
            creds = channelCredBuilder.build();
        } else {
            creds = InsecureServerCredentials.create();
        }
        if (tikaConfigXml == null) {
            // Create a default tika config
            tikaConfigXml = File.createTempFile("tika-config", ".xml");
            try (FileWriter fw = new FileWriter(tikaConfigXml, StandardCharsets.UTF_8)) {
                TikaConfigSerializer.serialize(new TikaConfig(), TikaConfigSerializer.Mode.STATIC_FULL, fw, StandardCharsets.UTF_8);
            }
        }
        File tikaConfigFile = new File(tikaConfigXml.getAbsolutePath());
        healthStatusManager.setStatus(TikaGrpcServer.class.getSimpleName(), ServingStatus.SERVING);
        server = Grpc
                .newServerBuilderForPort(port, creds)
                .addService(new TikaGrpcServerImpl(tikaConfigFile.getAbsolutePath()))
                .addService(healthStatusManager.getHealthService())
                .addService(ProtoReflectionService.newInstance())
                .build()
                .start();
        LOGGER.info("Server started, listening on " + port);
        Runtime
                .getRuntime()
                .addShutdownHook(new Thread(() -> {
                    // Use stderr here since the logger may have been reset by its JVM shutdown hook.
                    System.err.println("*** shutting down gRPC server since JVM is shutting down");
                    healthStatusManager.clearStatus(TikaGrpcServer.class.getSimpleName());
                    try {
                        TikaGrpcServer.this.stop();
                    } catch (InterruptedException e) {
                        e.printStackTrace(System.err);
                    }
                    System.err.println("*** server shut down");
                }));
    }
@Test
    public void testBiStream(Resources resources) throws Exception {
        String serverName = InProcessServerBuilder.generateName();

        Server server = InProcessServerBuilder
                .forName(serverName)
                .directExecutor()
                .addService(new TikaGrpcServerImpl(tikaConfigXml.getAbsolutePath()))
                .build()
                .start();
        resources.register(server, Duration.ofSeconds(10));

        ManagedChannel channel = InProcessChannelBuilder
                .forName(serverName)
                .directExecutor()
                .build();
        resources.register(channel, Duration.ofSeconds(10));
        TikaGrpc.TikaBlockingStub blockingStub = TikaGrpc.newBlockingStub(channel);
        TikaGrpc.TikaStub tikaStub = TikaGrpc.newStub(channel);

        String fetcherId = createFetcherId(1);
        String targetFolder = new File("target").getAbsolutePath();
        SaveFetcherReply reply = blockingStub.saveFetcher(SaveFetcherRequest
                .newBuilder()
                .setFetcherId(fetcherId)
                .setFetcherClass(FileSystemFetcher.class.getName())
                .setFetcherConfigJson(OBJECT_MAPPER.writeValueAsString(ImmutableMap
                        .builder()
                        .put("basePath", targetFolder)
                        .put("extractFileSystemMetadata", true)
                        .build()))
                .build());

        assertEquals(fetcherId, reply.getFetcherId());

        List<FetchAndParseReply> successes = Collections.synchronizedList(new ArrayList<>());
        List<FetchAndParseReply> errors = Collections.synchronizedList(new ArrayList<>());
        AtomicBoolean finished = new AtomicBoolean(false);

        StreamObserver<FetchAndParseReply> replyStreamObserver = new StreamObserver<>() {
            @Override
            public void onNext(FetchAndParseReply fetchAndParseReply) {
                LOG.debug("Fetched {} with metadata {}", fetchAndParseReply.getFetchKey(), fetchAndParseReply.getFieldsMap());
                if (PipesResult.STATUS.FETCH_EXCEPTION.name().equals(fetchAndParseReply.getStatus())) {
                    errors.add(fetchAndParseReply);
                } else {
                    successes.add(fetchAndParseReply);
                }
            }

            @Override
            public void onError(Throwable throwable) {
                fail(throwable);
            }

            @Override
            public void onCompleted() {
                LOG.info("Stream completed");
                finished.set(true);
            }
        };

        StreamObserver<FetchAndParseRequest> requestStreamObserver = tikaStub.fetchAndParseBiDirectionalStreaming(replyStreamObserver);

        File testDocumentFolder = new File("target/" + DateTimeFormatter
                .ofPattern("yyyy_MM_dd_HH_mm_ssSSS", Locale.getDefault())
                .format(LocalDateTime.now(ZoneId.systemDefault())) + "-" + UUID.randomUUID());
        assertTrue(testDocumentFolder.mkdir());
        try {
            for (int i = 0; i < NUM_TEST_DOCS; ++i) {
                File testFile = new File(testDocumentFolder, "test-" + i + ".html");
                FileUtils.writeStringToFile(testFile, "<html><body>test " + i + "</body></html>", StandardCharsets.UTF_8);
            }
            File[] testDocuments = testDocumentFolder.listFiles();
            assertNotNull(testDocuments);
            for (File testDocument : testDocuments) {
                requestStreamObserver.onNext(FetchAndParseRequest
                        .newBuilder()
                        .setFetcherId(fetcherId)
                        .setFetchKey(testDocument.getAbsolutePath())
                        .build());
            }
            // Now test error condition
            requestStreamObserver.onNext(FetchAndParseRequest
                    .newBuilder()
                    .setFetcherId(fetcherId)
                    .setFetchKey("does not exist")
                    .build());
            requestStreamObserver.onCompleted();
            assertEquals(NUM_TEST_DOCS, successes.size());
            assertEquals(1, errors.size());
            assertTrue(finished.get());
        } finally {
            FileUtils.deleteDirectory(testDocumentFolder);
        }
    }