private void processNestedStreams(final Pipeline pipeline,
                                      final Meta meta,
                                      final Source source,
                                      final TaskContext taskContext) {
        boolean startedProcessing = false;

        // Get the stream providers.
        metaHolder.setMeta(meta);

        try {
            final StreamLocationFactory streamLocationFactory = new StreamLocationFactory();
            locationFactory.setLocationFactory(streamLocationFactory);

            // Loop over the stream boundaries and process each
            // sequentially.
            final long count = source.count();
            for (long index = 0; index < count && !taskContext.isTerminated(); index++) {
                try (final InputStreamProvider inputStreamProvider = source.get(index)) {
                    InputStream inputStream;

                    // If the task requires specific events to be processed then
                    // add them.
                    final String data = processorTask.getData();
                    if (data != null && !data.isEmpty()) {
                        final List<InclusiveRange> ranges = InclusiveRanges.rangesFromString(data);
                        final SegmentInputStream raSegmentInputStream = inputStreamProvider.get();
                        raSegmentInputStream.include(0);
                        for (final InclusiveRange range : ranges) {
                            for (long i = range.getMin(); i <= range.getMax(); i++) {
                                raSegmentInputStream.include(i);
                            }
                        }
                        raSegmentInputStream.include(raSegmentInputStream.count() - 1);
                        inputStream = raSegmentInputStream;

                    } else {
                        // Get the stream.
                        inputStream = inputStreamProvider.get();
                    }

                    // Get the appropriate encoding for the stream type.
                    final String encoding = feedProperties.getEncoding(
                            meta.getFeedName(), meta.getTypeName(), null);

                    // We want to get a preview of the input stream so we can
                    // skip it if it is effectively empty.
                    final PreviewInputStream previewInputStream = new PreviewInputStream(inputStream);
                    String preview = previewInputStream.previewAsString(PREVIEW_SIZE, encoding);
                    // Remove whitespace from the preview.
                    preview = preview.trim();

                    // If there are still characters in the preview then
                    // continue.
                    if (preview.length() >= MIN_STREAM_SIZE) {
                        // Try and remove XML declaration for cases where the
                        // input is blank except for an XML declaration.
                        preview = XML_DECL_PATTERN.matcher(preview).replaceFirst("");
                        // Remove whitespace from the preview.
                        preview = preview.trim();

                        // Skip the input stream if it is empty. replaces:
                        // inputStream.size >= MIN_STREAM_SIZE
                        if (preview.length() >= MIN_STREAM_SIZE) {
                            // Start processing if we haven't already.
                            if (!startedProcessing) {
                                startedProcessing = true;
                                pipeline.startProcessing();
                            }

                            metaHolder.setInputStreamProvider(inputStreamProvider);
                            metaHolder.setPartIndex(index);
                            streamLocationFactory.setPartIndex(index);

                            // Process the boundary.
                            try {
                                pipeline.process(previewInputStream, encoding);
                            } catch (final Throwable e) {
                                handleProcessingException(e, meta);
                            }

                            // Reset the error statistics for the next stream.
                            if (errorReceiverProxy.getErrorReceiver() instanceof ErrorStatistics errorStatistics) {
                                errorStatistics.reset();
                            }
                        }
                    }
                }
            }
        } catch (final Throwable e) {
            handleProcessingException(e, meta);
        } finally {
            try {
                if (startedProcessing) {
                    pipeline.endProcessing();
                }
            } catch (final Throwable e) {
                handleProcessingException(e, meta);
            }
        }
    }
Processes a source and writes the result to a target.
stroom-core/src/main/java/stroom/core/dataprocess/AbstractProcessorTaskExecutor.java