Add a stage to the Pipeline.

@param callable|string|PipelineInterface|array $stages
775883
public function pipe($stages): void
    {
        if ($stages instanceof PipelineInterface) {
            $stages = $stages->stages();
        }

        if (!\is_array($stages)) {
            $this->stages[] = $stages;

            return;
        }

        foreach ($stages as $stage) {
            $this->pipe($stage);
        }
    }
998540
private function set_stage_hooks( $hooks ) {
		$this->current_stage_hooks     = $hooks;
		$pseudo_hook                   = "{$hooks[0]}:before";
		$this->loggers[ $pseudo_hook ] = new Logger( array( 'hook' => $pseudo_hook ) );
		$this->loggers[ $pseudo_hook ]->start();
	}
870369
public function reindex($stage = null) {

		// Make sure the current data object is not orphaned.

		if($this->owner->ParentID > 0) {
			$parent = $this->owner->Parent();
			if(is_null($parent) || ($parent === false)) {
				return;
			}
		}

		// Make sure the extension requirements have been met before enabling the custom site tree search index,
		// since this may impact performance.

		if((ClassInfo::baseDataClass($this->owner) === 'SiteTree')
			&& SiteTree::has_extension('SiteTreePermissionIndexExtension')
			&& SiteTree::has_extension('SolrSearchPermissionIndexExtension')
			&& ClassInfo::exists('QueuedJob')
		) {

			// Queue a job to handle the recursive indexing.
			$indexing = new SolrSiteTreePermissionIndexJob($this->owner, $stage);
			singleton('QueuedJobService')->queueJob($indexing);
		}
		else {

			// When the requirements haven't been met, trigger a normal index.
			$this->searchService->index($this->owner, $stage);
		}
	}
622194
protected function sortExtensions(/*# string */ $stage)
    {
        $sorted = &$this->sorted;

        // if not sorted yet
        if (!isset($sorted[$stage])) {
            $sorted[$stage] = [];

            // current stage
            if (isset($this->extensions[$stage])) {
                foreach ($this->extensions[$stage] as $p => $e) {
                    $sorted[$stage][$p] = $e;
                }
            }

            // process the special STAGE_ALL
            $all = ExtensionStage::STAGE_ALL;
            if (isset($this->extensions[$all])) {
                foreach ($this->extensions[$all] as $p => $e) {
                    $sorted[$stage][$p] = isset($sorted[$stage][$p]) ?
                        (array_merge($sorted[$stage][$p], $e)) :
                        $e;
                }
            }

            // sort extensions by priority
            if (count($sorted[$stage])) {
                ksort($sorted[$stage]);
            }
        }
    }
854447
public function addExtensions(array $extensions)
	{
		if ($extensions)
		{
			$keys = array_map('snake_case', array_keys($extensions));

			$extensions = array_combine($keys, array_values($extensions));
		}

		$this->extensions = array_merge($this->extensions, $extensions);
	}
727494
public function purge(string $pattern, $stages): void
    {
        $pipeline = new HttpPipeline(RequestMethodInterface::METHOD_PURGE, $this->prefix.$pattern, $this->routerStages);
        $pipeline->pipe($stages);
        $this->pipelineCollection->add($pipeline);
    }
412224
final public function pipe(callable $stage, callable $fallback = null): PipeChainInterface
    {
        $this->stack->attach($stage, $fallback);

        return $this;
    }
384864
public function trace(string $pattern, $stages): void
    {
        $pipeline = new HttpPipeline(RequestMethodInterface::METHOD_TRACE, $this->prefix.$pattern, $this->routerStages);
        $pipeline->pipe($stages);
        $this->pipelineCollection->add($pipeline);
    }
734864
public function addStepWithEvents($event, $callable, $arguments = [])
    {
        $this->addStep('fireEvent', [$event.'.before']);
        $this->addStep($callable, $arguments);
        $this->addStep('fireEvent', [$event.'.after']);
    }
366054
public function processStages(array $stages, $payload)
    {
        // Take the stage to handle
        $currentStage = \array_shift($stages);

        // If is a string get the instance in the container
        if (\is_string($currentStage) && $this->container->has($currentStage)) {
            $currentStage = $this->container->get($currentStage);
        }

        // If the current stage is a RequestHandler use it and return the Response
        if ($currentStage instanceof RequestHandlerInterface) {
            return $currentStage->handle($payload);
        }

        // Process the current stage, and proceed to the stack
        if (false !== \current($stages) && \is_callable($currentStage)) {
            return $this->processStages($stages, $currentStage($payload));
        }

        // If there's not next stage in the stack, return the result for this one
        if (\is_callable($currentStage)) {
            return $currentStage($payload);
        }

        throw new UnprocessableStageException($currentStage, 'The stage can\'t be handled');
    }
1201533
protected function createDeletedVersion($stages = [])
    {
        // Skip if suppressed by parent delete
        if (!$this->getDeleteWritesVersion()) {
            return;
        }
        // Prepare manipulation
        $baseTable = $this->owner->baseTable();
        $now = DBDatetime::now()->Rfc2822();
        // Ensure all fixed_fields are specified
        $manipulation = [
            $baseTable => [
                'fields' => [
                    'ID' => $this->owner->ID,
                    'LastEdited' => $now,
                    'Created' => $this->owner->Created ?: $now,
                    'ClassName' => $this->owner->ClassName,
                ],
            ],
        ];
        // Prepare "deleted" augment write
        $this->augmentWriteVersioned(
            $manipulation,
            $this->owner->baseClass(),
            $baseTable,
            $this->owner->ID,
            $stages,
            true
        );
        unset($manipulation[$baseTable]);
        $this->owner->extend('augmentWriteDeletedVersion', $manipulation, $stages);
        DB::manipulate($manipulation);
    }
1018726
protected function withAddedStep(string $stage, callable $step, bool $replace = false)
    {
        $clone = clone $this;

        if ($replace) {
            $clone->stages[$stage] = [];
        }

        $clone->stages[$stage][] = $step;

        return $clone;
    }
1247730
public void setStageKeys(java.util.Collection<StageKey> stageKeys) {
        if (stageKeys == null) {
            this.stageKeys = null;
            return;
        }

        this.stageKeys = new java.util.ArrayList<StageKey>(stageKeys);
    }
195279
public function connect(string $pattern, $stages): void
    {
        $pipeline = new HttpPipeline(RequestMethodInterface::METHOD_CONNECT, $this->prefix.$pattern, $this->routerStages);
        $pipeline->pipe($stages);
        $this->pipelineCollection->add($pipeline);
    }
681369
public function get(string $pattern, $stages): void
    {
        $pipeline = new HttpPipeline(RequestMethodInterface::METHOD_GET, $this->prefix.$pattern, $this->routerStages);
        $pipeline->pipe($stages);
        $this->pipelineCollection->add($pipeline);
    }
673050
protected function processStages(array $stages, $payload, array $extra = [])
    {
        foreach ($stages as $key => $stage) {
            if ($this->isStageSelectable($key)) {
                $payload = $this->processStage($stage, $payload, $extra);
            }
        }

        return $payload;
    }
808667
private function assertThereIsAtLeastOneConnectorPerIntermediaryStage(
        $stageNumber,
        $stageCount,
        $stageName,
        $stageConnectors
    ) {
        if ($stageNumber < $stageCount && count($stageConnectors) == 0) {
            throw new PipelineBuildingException("No stage connector found for stage '$stageName'.");
        }
    }