public Pipeline create(final PipelineData pipelineData,
                           final Terminator terminator,
                           final SteppingController controller) {
        final ElementRegistry pipelineElementRegistry = pipelineElementRegistryFactory.get();

        // If we are stepping then we don't want to use the cache.
        // Create an instance of each element.
        final Map<String, Element> elementInstances = new HashMap<>();
        final Map<Element, PipelineElementType> elementTypeMap = new HashMap<>();
        final Map<String, Set<String>> linkSets = new HashMap<>();

        for (final PipelineElement element : pipelineData.getElements().getAdd()) {
            LOGGER.debug("create() - loading element {}", element);

            final Class<Element> elementClass = pipelineElementRegistry.getElementClass(element.getType());

            if (elementClass == null) {
                throw new PipelineFactoryException("Unable to load elementClass for type " + element.getType());
            }

            final Element elementInstance = elementFactory.getElementInstance(elementClass);

            if (elementInstance == null) {
                throw new PipelineFactoryException("Unable to load elementInstance for class " + elementClass);
            }

            // Set the id on the pipeline element for use in tracing
            // errors, intercepting input/output etc.
            elementInstance.setElementId(element.getId());

            // Set an object to provide quick processing termination if needed.
            elementInstance.setTerminator(terminator);

            // Set the properties on this instance.
            setProperties(pipelineElementRegistry, element.getId(), element.getType(), elementInstance, pipelineData,
                    controller);

            // Set the pipeline references on this instance.
            setPipelineReferences(pipelineElementRegistry, element.getId(), element.getType(), elementInstance,
                    pipelineData);

            // Store the instance.
            elementInstances.put(element.getId(), elementInstance);
            elementTypeMap.put(elementInstance, pipelineElementRegistry.getElementType(element.getType()));

            // Record links.
            final Set<String> linkSet = new HashSet<>();
            linkSets.put(element.getId(), linkSet);
            for (final PipelineLink link : pipelineData.getLinks().getAdd()) {
                if (link.getFrom().equals(element.getId())) {
                    linkSet.add(link.getTo());
                }
            }
        }

        // Get the source element.
        final SourceElement sourceElement = (SourceElement) elementInstances.get("Source");
        if (sourceElement == null) {
            throw new PipelineFactoryException("The pipeline has no source element");
        }

        // Get the split depth to use with the stepping controller.
        int controllerSplitDepth = 1;
        if (controller != null) {
            controllerSplitDepth =
                    getSplitDepth(elementInstances, elementTypeMap, linkSets, sourceElement.getElementId());
            controllerSplitDepth = Math.max(controllerSplitDepth, 1);
        }

        // Link the instances.
        link(elementInstances,
                elementTypeMap,
                linkSets,
                controller,
                sourceElement,
                sourceElement.getElementId(),
                controllerSplitDepth);

        // We need to create a root element that will be a target for the input
        // stream.
        TakesInput root = sourceElement;

        // Perform the last bit of setup if we are stepping.
        if (controller != null) {
            // If we still haven't added a record detector then add one to the
            // input.
            if (controller.getRecordDetector() == null) {
                final InputStreamRecordDetectorElement recordDetector = new InputStreamRecordDetectorElement();
                controller.setRecordDetector(recordDetector);
                recordDetector.addTarget(sourceElement);
                root = recordDetector;
            }

            controller.getRecordDetector().setController(controller);
        }

        return new PipelineImpl(processorFactory, elementInstances, root, controller != null);
    }
Convenience method for non stepping mode. Used only for testing.
stroom-pipeline/src/main/java/stroom/pipeline/factory/PipelineFactory.java