private void processData(final Source source,
                             final DocRef pipelineRef,
                             final Pipeline pipeline) {
        long count = 0;

        try (final InputStreamProvider inputStreamProvider = source.get(0)) {
            // This is a valid stream so try and extract as many
            // segments as we are allowed.
            try (final SegmentInputStream segmentInputStream = inputStreamProvider.get()) {
                // Now try and extract the data.
                extract(pipelineRef, pipeline, source, segmentInputStream, count);
            }
        } catch (final TaskTerminatedException | ClosedByInterruptException e) {
            LOGGER.debug(e::getMessage, e);
        } catch (final ExtractionException e) {
            throw e;
        } catch (final IOException | RuntimeException e) {
            // Something went wrong extracting data from this stream.
            throw new ExtractionException("Unable to extract data from stream source with id: " +
                    source.getMeta().getId() + " - " + e.getMessage(), e);
        }
    }
Extract data from the segment list. Returns the total number of segments
that were successfully extracted.
stroom-analytics/stroom-analytics-impl/src/main/java/stroom/analytics/impl/AnalyticsStreamProcessor.java