Pipes command message to the next filter

@param CommandMessage $message The command message

@return void
969772
public function pipe(CommandMessage $message): void
    {
        /** @var CommandFilter $filter */
        $filter = $this->stack->pop();
        $filter->process($message, [$this, 'pipe']);
    }
841353
public function message($to, $message)
    {
        $this->logger->info(sprintf('Sending a message ["%s"] to %s', $message, $to));

        return call_user_func_array([$this->wrapped, 'message'], func_get_args());
    }
1319880
public function call(Message $message) : Message
    {
        $aggregate = $message->getHeaders()->containsKey(CqrsMessagingModule::INTEGRATION_MESSAGING_CQRS_AGGREGATE_HEADER)
                            ? $message->getHeaders()->get(CqrsMessagingModule::INTEGRATION_MESSAGING_CQRS_AGGREGATE_HEADER)
                            : null;
        $methodInvoker = MethodInvoker::createWith(
            $aggregate ? $aggregate : $message->getHeaders()->get(CqrsMessagingModule::INTEGRATION_MESSAGING_CQRS_AGGREGATE_CLASS_NAME_HEADER),
            $message->getHeaders()->get(CqrsMessagingModule::INTEGRATION_MESSAGING_CQRS_AGGREGATE_METHOD_HEADER),
            $this->messageToParameterConverters
        );

        $resultMessage = MessageBuilder::fromMessage($message);
        try {
            $result = $methodInvoker->processMessage($message);

            if (!$aggregate) {
                $resultMessage = $resultMessage
                    ->setHeader(CqrsMessagingModule::INTEGRATION_MESSAGING_CQRS_AGGREGATE_HEADER, $result);
            }
        } catch (\Throwable $e) {
            throw MessageHandlingException::fromOtherException($e, $message);
        }

        if (!is_null($result)) {
            $resultMessage = $resultMessage
                ->setPayload($result);
        }

        return $resultMessage
                ->build();
    }
978842
public function setupMessage(Message $message){

        $this->addRecipients($message);

        $this->setSubject($message);

        $this->callCustomBuilder($message);

    }
777073
public function onPipeMessage($serv, $fromWorkerId, $message)
    {
        parent::onPipeMessage($serv, $fromWorkerId, $message);
        $data = unserialize($message);
        switch ($data['type']) {
            case Macro::MSG_TYPR_ASYN:
                $this->asynPoolManager->distribute($data['message']);
                break;
        }
    }
1069376
public function logMessage(Message $message) {
        foreach($this->logger as $logger) {
            $logger->logMessage($message);
        }
    }
1293338
private function tryLogMessage($message)
    {
        $messageId = null;

        if ($message instanceof RemoteMessage) $messageId = $message->header()->uuid();
        elseif ($message instanceof ProcessingMessage) $messageId = $message->uuid();

        if (! $messageId) return;

        $entry = $this->messageLogger->getEntryForMessageId($messageId);

        if ($entry) return;

        $this->messageLogger->logIncomingMessage($message);
    }
853570
public function publish($message)
    {
        /* @var Connection $conn */
        list($conn) = (yield $this->take());

        $conn->writeCmd(Command::publish($this->topic, $message));
        $this->stats["messagesPublished"]++;

        $timeout = NsqConfig::getPublishTimeout();
        $onTimeout = $this->onPublishTimeout($conn);
        $timerId = $this->getPublishTimeoutTimerId($conn);

        Timer::after($timeout, $onTimeout, $timerId);

        yield setContext("conn", $conn);
        yield $this;
    }
726668
protected function runStrategyCommand($method, $message)
    {
        if (!$this->getOption($method)) {
            return;
        }

        $this->explainer->line($message);
        $this->results[] = $this->strategy->$method();
    }
1154451
protected function flush($message) {
        switch ($this->output) {
            case self::PRINT_AS_OUTPUT:
                IO::out($message);
                break;
            case self::PRINT_AS_MESSAGE:
                IO::err($message);
                break;
        }
    }
65458
protected function debug($message)
    {
        if ($this->debug) {
            if ($this->debug_handler) {
                call_user_func_array(
                    $this->debug_handler, array(&$this, $message)
                );
            } else {
                echo "DEBUG: $message\n";
            }
        }
    }
336548
public function endsWith($sub, $message = null)
    {
        $this->setRule(__FUNCTION__, function($val, $args)
                       {
                           $sub = $args[0];
                           return (strlen($val) === 0 || substr($val, -strlen($sub)) === $sub);
                       }, $message, array($sub));
        return $this;
    }
1190194
public function warning($message): void
	{
		$message = $this->logMessToArray($message);
		$this->log(array_merge(['SEVERITY' => 'WARNING'], $message));
	}
453942
public function addMessage($message)
    {
        collect((array) $message)->each(function ($item) {
            collect($item)->flatten()->each(function ($flattened) {
                $this->messageList->push($flattened);
            });
        });
    }
513889
public function pipeOutputTo(Command $cmd) : void
    {
        if (!$this->canPipe()) {
            throw new RuntimeException('Can\'t pipe output');
        }
        $this->pipeline[] = $cmd;
    }
645223
public function violate($message)
    {
        if (!empty($this->messages)){
            $message = $this->messages[0];
        } else {
            $this->setMessage($message);
        }
    }
927835
public function stash(string $message = null, bool $includeUntracked = false, bool $keepIndex = false)
    {
        $stashCommand = StashCommand::getInstance($this);
        $command = $stashCommand->save($message, $includeUntracked, $keepIndex);
        $this->caller->execute($command);
    }