private void processData(final QueryKey queryKey,
                             final Source source,
                             final long[] eventIds,
                             final DocRef pipelineRef,
                             final Pipeline pipeline,
                             final ErrorConsumer errorConsumer) {
        final ErrorReceiver errorReceiver = (severity, location, elementId, message, errorType, e) -> {
            final Optional<TaskTerminatedException> optional = TaskTerminatedException.unwrap(e);
            if (optional.isPresent()) {
                throw optional.get();
            }

            final StoredError storedError = new StoredError(severity, location, elementId, message, errorType);
            errorConsumer.add(storedError::toString);
            throw ProcessException.wrap(message, e);
        };

        errorReceiverProxy.setErrorReceiver(errorReceiver);
        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()) {
                // Include the XML Header and footer.
                segmentInputStream.include(0);
                segmentInputStream.include(segmentInputStream.count() - 1);

                // Include as many segments as we can.
                for (final long eventId : eventIds) {
                    segmentInputStream.include(eventId);
                    count++;
                }

                // Now try and extract the data.
                extract(queryKey, pipelineRef, pipeline, source, segmentInputStream, count);
            }
        } catch (final ExtractionException e) {
            throw e;
        } catch (final IOException | RuntimeException e) {
            final Optional<TaskTerminatedException> optional = TaskTerminatedException.unwrap(e);
            if (optional.isPresent()) {
                LOGGER.debug(e::getMessage, e);
            } else {
                // 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-search/stroom-search-extraction/src/main/java/stroom/search/extraction/ExtractionTaskHandler.java