Creates a callback for an individual stage

@return Closure The callback
@throws PipelineException Thrown if there was a problem creating a stage
772308
private function createStageCallback() : Closure
    {
        return function ($stages, $stage) {
            return function ($input) use ($stages, $stage) {
                if ($stage instanceof Closure) {
                    return $stage($input, $stages);
                } else {
                    if ($this->methodToCall === null) {
                        throw new PipelineException('Method must not be null');
                    }

                    if (!method_exists($stage, $this->methodToCall)) {
                        throw new PipelineException(get_class($stage) . "::{$this->methodToCall} does not exist");
                    }

                    return $stage->{$this->methodToCall}($input, $stages);
                }
            };
        };
    }
1028039
public function callMe()
    {
        $output = $this->dispatchStartEvent();

        $recursionMarker = $this->pool->recursionHandler->getMarker();
        $output .= $this->pool->render->renderSingeChildHr();
        $multiline = $this->parameters[static::PARAM_MULTILINE];

        // Iterate through.
        foreach ($this->parameters[static::PARAM_DATA] as $key => &$value) {
            // We will not output our recursion marker.
            // Meh, the only reason for the recursion marker
            // in arrays is because of the $GLOBAL array, which
            // we will only render once.
            if ($key === $recursionMarker) {
                continue;
            }

            /** @var Model $model */
            $model = $this->pool->createClass('Brainworxx\\Krexx\\Analyse\\Model');

            // Are we dealing with multiline code generation?
            if ($multiline === true) {
                // Here we tell the Codegen service that we need some
                // special handling.
                $model->setMultiLineCodeGen(Codegen::ITERATOR_TO_ARRAY);
            }

            // Handling string keys of the array.
            $this->handleKey($key, $model);
            // Handling of the value and add some output.
            $output .= $this->handleValue($value, $model);
        }

        return $output . $this->pool->render->renderSingeChildHr();
    }
977375
private synchronized PipelineInterpreter.State reloadAndSave() {
        // this classloader will hold all generated rule classes
        PipelineClassloader commonClassLoader = allowCodeGeneration ? new PipelineClassloader() : null;

        // read all rules and parse them
        Map<String, Rule> ruleNameMap = Maps.newHashMap();
        ruleService.loadAll().forEach(ruleDao -> {
            Rule rule;
            try {
                rule = pipelineRuleParser.parseRule(ruleDao.id(), ruleDao.source(), false, commonClassLoader);
            } catch (ParseException e) {
                rule = Rule.alwaysFalse("Failed to parse rule: " + ruleDao.id());
            }
            ruleNameMap.put(rule.name(), rule);
        });

        // read all pipelines and parse them
        ImmutableMap.Builder<String, Pipeline> pipelineIdMap = ImmutableMap.builder();
        pipelineService.loadAll().forEach(pipelineDao -> {
            Pipeline pipeline;
            try {
                pipeline = pipelineRuleParser.parsePipeline(pipelineDao.id(), pipelineDao.source());
            } catch (ParseException e) {
                pipeline = Pipeline.empty("Failed to parse pipeline" + pipelineDao.id());
            }
            //noinspection ConstantConditions
            pipelineIdMap.put(pipelineDao.id(), resolvePipeline(pipeline, ruleNameMap));
        });

        final ImmutableMap<String, Pipeline> currentPipelines = pipelineIdMap.build();

        // read all stream connections of those pipelines to allow processing messages through them
        final HashMultimap<String, Pipeline> connections = HashMultimap.create();
        for (PipelineConnections streamConnection : pipelineStreamConnectionsService.loadAll()) {
            streamConnection.pipelineIds().stream()
                    .map(currentPipelines::get)
                    .filter(Objects::nonNull)
                    .forEach(pipeline -> connections.put(streamConnection.streamId(), pipeline));
        }
        ImmutableSetMultimap<String, Pipeline> streamPipelineConnections = ImmutableSetMultimap.copyOf(connections);

        final PipelineInterpreter.State newState = stateFactory.newState(currentPipelines, streamPipelineConnections);
        latestState.set(newState);
        return newState;
    }
648705
public function handle($passable, Closure $callback)
    {
        $pipeline = array_reduce(array_reverse($this->pipes), function ($stack, $pipe)
        {
            return $this->createSlice($stack, $pipe);

        }, $this->prepareDestination($callback));

        return call_user_func($pipeline, $passable);
    }
791925
public function createAbstractMethodStubs() {
    if ($this->isAbstract() || $this->isInterface()) return $this;
    
    foreach ($this->getAllMethods(self::FULL_HIERARCHY & ~self::WITH_OWN) as $method) {
      //if ($this->needsImplementation($method)) {
      $this->createMethodStub($method);
      //} // performance: this will be already checked in createMethodStub 
    }
    
    return $this;
  }
821705
private function handlePipeline(MatchablePipelineCollectionInterface $pipelines): ?ResponseInterface
    {
        /** @var MatchablePipelineInterface $pipeline */
        foreach ($pipelines as $pipeline) {
            if ($pipeline->matchBy($this->matchableRequest)) {
                $stages = \array_merge($this->stages(), $pipeline->stages());
                $pipelineResponse = $this->processor->processStages($stages, $this->matchableRequest->request());

                if ($pipelineResponse instanceof ResponseInterface) {
                    return $pipelineResponse;
                }

                $body = $this->response->getBody();
                $body->write($pipelineResponse);

                return $this->response->withBody($body);
            }
        }

        return null;
    }
502862
def run_callback(self):
    
    pipeline_id = self.request.get('pipeline_id')
    if not pipeline_id:
      raise _CallbackTaskError('"pipeline_id" parameter missing.')

    pipeline_key = db.Key.from_path(_PipelineRecord.kind(), pipeline_id)
    pipeline_record = db.get(pipeline_key)
    if pipeline_record is None:
      raise _CallbackTaskError(
          'Pipeline ID "%s" for callback does not exist.' % pipeline_id)

    params = pipeline_record.params
    real_class_path = params['class_path']
    try:
      pipeline_func_class = mr_util.for_name(real_class_path)
    except ImportError, e:
      raise _CallbackTaskError(
          'Cannot load class named "%s" for pipeline ID "%s".'
          % (real_class_path, pipeline_id))

    if 'HTTP_X_APPENGINE_TASKNAME' not in self.request.environ:
      if pipeline_func_class.public_callbacks:
        pass
      elif pipeline_func_class.admin_callbacks:
        if not users.is_current_user_admin():
          raise _CallbackTaskError(
              'Unauthorized callback for admin-only pipeline ID "%s"'
              % pipeline_id)
      else:
        raise _CallbackTaskError(
            'External callback for internal-only pipeline ID "%s"'
            % pipeline_id)

    kwargs = {}
    for key in self.request.arguments():
      if key != 'pipeline_id':
        kwargs[str(key)] = self.request.get(key)

    def perform_callback():
      stage = pipeline_func_class.from_id(pipeline_id)
      if stage is None:
        raise _CallbackTaskError(
            'Pipeline ID "%s" deleted during callback' % pipeline_id)
      return stage._callback_internal(kwargs)

    # callback_xg_transaction is a 3-valued setting (None=no trans,
    # False=1-eg-trans, True=xg-trans)
    if pipeline_func_class._callback_xg_transaction is not None:
      transaction_options = db.create_transaction_options(
          xg=pipeline_func_class._callback_xg_transaction)
      callback_result = db.run_in_transaction_options(transaction_options,
                                                      perform_callback)
    else:
      callback_result = perform_callback()

    if callback_result is not None:
      status_code, content_type, content = callback_result
      self.response.set_status(status_code)
      self.response.headers['Content-Type'] = content_type
      self.response.out.write(content)
246750
public function create($pipelineType, $pipelineParams)
    {
        if (!isset($this->factories[$pipelineType])) {
            throw new \LogicException("No factory found for pipeline aggregation of type {$pipelineType}");
        }

        return $this->factories[$pipelineType]->create($pipelineParams);
    }
254096
public function redact() : Stage\Redact
    {
        $stage = new Stage\Redact($this);
        $this->addStage($stage);

        return $stage;
    }
907781
public static function begin($firstStageForm) : StagedForm
    {
        $firstStage = self::parseStage($firstStageForm);

        if (!($firstStage instanceof IndependentFormStage)) {
            throw InvalidArgumentException::format(
                    'Invalid form stage argument: first stage must be independent, callable with required parameter given'
            );
        }

        return new self($firstStage);
    }
889105
protected function getSlice()
	{
		return function ( $stack, $pipe )
		{
			return function ( $passable ) use ( $stack, $pipe )
			{
				try
				{
					$slice = $this->getSliceWithException();

					return call_user_func( $slice( $stack, $pipe ), $passable );
				}
				catch ( \Exception $e )
				{
					if ( is_callable( $this->exceptionHandler ) )
						return call_user_func( $this->exceptionHandler, $passable, $e );
					else
						throw new PipelineException( $e->getMessage(), 0, $e );
				}
				catch ( \Throwable $e )
				{
					if ( is_callable( $this->exceptionHandler ) )
						return call_user_func( $this->exceptionHandler, $passable, new FatalThrowableError( $e ) );
					else
						throw new PipelineException( $e->getMessage(), 0, $e );
				}
			};
		};
	}
1115302
public static function make(\Closure $callback = null)
    {
        self::$instance = new static();

        if ($callback instanceof \Closure) {
            return $callback(self::$instance);
        }

        return self::$instance;
    }
950
protected function dispatchToRouter() : Closure
    {
        return function ($request) {
            return $this->router->setContainer($this->container)->dispatch($request);
        };
    }
1141167
public function callMe()
    {
        $output = $this->dispatchStartEvent();

        /** @var \ReflectionClass $ref */
        $ref = $this->parameters[static::PARAM_REF];

        $doProtected = $this->pool->config->getSetting(Fallback::SETTING_ANALYSE_PROTECTED_METHODS) ||
            $this->pool->scope->isInScope();
        $doPrivate = $this->pool->config->getSetting(Fallback::SETTING_ANALYSE_PRIVATE_METHODS) ||
            $this->pool->scope->isInScope();
        $domId = $this->generateDomIdFromClassname($ref->getName(), $doProtected, $doPrivate);

        // We need to check, if we have a meta recursion here.
        if ($this->pool->recursionHandler->isInMetaHive($domId) === true) {
            // We have been here before.
            // We skip this one, and leave it to the js recursion handler!
            return $output .
                $this->pool->render->renderRecursion(
                    $this->dispatchEventWithModel(
                        'recursion',
                        $this->pool->createClass('Brainworxx\\Krexx\\Analyse\\Model')
                            ->setDomid($domId)
                            ->setNormal('Methods')
                            ->setName('Methods')
                            ->setType(static::TYPE_INTERNALS)
                    )
                );
        }

        return $output . $this->analyseMethods($ref, $domId, $doProtected, $doPrivate);
    }
493350
protected function defaultExceptionResolver(): Closure
    {
        return function (string $errorMessage) {
            return (new GenericException($errorMessage))
                ->setExtensions([$this->errorType => $this->errors])
                ->setCategory($this->errorType);
        };
    }
120053
public function start(Closure $callback)
    {
        $this->output = $callback;

        $this->cooldown();

        $this->process->start($callback);

        return $this;
    }
620086
Closure createClosure(String rootMethodName, final Object closureObj) throws Exception {
        if (!isClosureCommand(closureObj)) {
            throw new RuntimeException(format(ERROR_TYPE_MESSAGE, rootMethodName,
                    getClosureCommandType().getName()).getMessage());
        }
        Method closureMethod = closureObj.getClass().getMethod(INVOKE_METHOD);
        return new Closure(closureMethod, closureObj);
    }