public static void main(String[] args) throws Exception {
        Path tikaConfigPath = Paths.get(args[0]);
        PipesIterator pipesIterator = PipesIterator.build(tikaConfigPath);
        long start = System.currentTimeMillis();
        try (AsyncProcessor processor = new AsyncProcessor(tikaConfigPath, pipesIterator)) {

            for (FetchEmitTuple t : pipesIterator) {
                boolean offered = processor.offer(t, TIMEOUT_MS);
                if (!offered) {
                    throw new TimeoutException("timed out waiting to add a fetch emit tuple");
                }
            }
            processor.finished();
            while (true) {
                if (processor.checkActive()) {
                    Thread.sleep(500);
                } else {
                    break;
                }
            }
            long elapsed = System.currentTimeMillis() - start;
            LOG.info("Successfully finished processing {} files in {} ms", processor.getTotalProcessed(), elapsed);
        }
    }
@Test
    public void testCrash() throws Exception {
        Path config = getPath("/configs/tika-config-broken.xml");
        assertThrows(TikaConfigException.class, () -> TikaAsyncCLI.main(new String[]{config.toAbsolutePath().toString()}));
    }