Process a message

@param MessageInterface $message
@return mixed
@throws Exception\InvalidMessageException
437529
public function process(MessageInterface $message)
    {

        $event   = $this->getProcessEvent();
        $event->setWorker($this);
        $event->setMessage($message);

        $events  = $this->getEventManager();
        $results = $events->trigger(ProcessEvent::EVENT_PROCESSOR, $event, function ($result) {
            return ($result instanceof ProcessorInterface);
        });
        $processor = $results->last();
        if (!$processor instanceof ProcessorInterface) {
            throw new \RuntimeException(sprintf(
                '%s: no processor selected!',
                __METHOD__
            ));
        }

        $processor->setWorker($this);
        $event->setProcessor($processor);

        $result = $processor->process($message);
        $event->setResult($result);

        $events->trigger(ProcessEvent::EVENT_PROCESS_POST, $event);


        return $result;
    }
241663
protected function isValidMessage($message): Message
    {
        if ($message instanceof Message) {
            return $message;
        }

        if (\is_string($message) || $message instanceof Attachment) {
            return Message::create($message);
        }

        throw new MessengerException(sprintf(
            'message must be a string or an instance of %s or %s.',
            Message::class,
            Attachment::class
        ));
    }
144386
private function prepareRequest(MessageInterface $message, RecipientInterface $recipient) {
        $uri = new Uri($recipient->getEndpoint());
        return Pusher::createStandardRequest($message, $recipient)->withUri($uri);
    }
990567
public function store(OutboxMessage $message)
    {
        $messageId = $message->getMessageId();
        if (isset($this->messages[$messageId])) {
            throw new InvalidArgumentException("Outbox message with ID '$messageId' already exists in storage.");
        }

        $this->messages[$messageId] = new InMemoryStoredMessage($messageId, $message->getTransportOperations());
        $this->lastMessageId = $messageId;
    }
1198503
protected function headersAndBody(MessageInterface $message)
    {
        $string = '';

        foreach ($message->getHeaders() as $name => $values) {
            foreach ($values as $value) {
                $string .= $name.': '.$value."\n";
            }
        }

        $string .= "\n";
        $string .= (string) $message->getBody();

        return $string;
    }
946570
protected function headerToArray(MessageInterface $message, $header)
    {
        $lines = explode(Http::NEW_LINE, $header);

        foreach ($lines as $line) {
            $parts = explode(':', $line, 2);

            if (isset($parts[0]) && isset($parts[1])) {
                $key   = $parts[0];
                $value = substr($parts[1], 1);

                $message->addHeader($key, $value);
            }
        }
    }
754798
public function send(MessageInterface $message)
    {
        $messageEncoder = $this->getMessageEncoder();
        $rawMessage = $messageEncoder->encode($message);

        $request = array(
            sprintf("POST %s HTTP/1.1", $this->path),
            sprintf("Host: %s:%d", $this->host, $this->port),
            sprintf("Content-Length: %d", strlen($rawMessage)),
            "Content-Type: application/json",
            "Connection: Keep-Alive",
            "Accept: */*"
        );

        if (null !== $this->authentication) {
            $request[] = "Authorization: Basic " . base64_encode($this->authentication);
        }

        if ($messageEncoder instanceof CompressedJsonEncoder) {
            $request[] = "Content-Encoding: gzip";
        }

        $request[] = ""; // blank line to separate headers from body
        $request[] = $rawMessage;

        $request = implode($request, "\r\n");

        $byteCount = $this->socketClient->write($request);
        $headers = $this->readResponseHeaders();

        // if we don't have a HTTP/1.1 connection, or the server decided to close the connection
        // we should do so as well. next read/write-attempt will open a new socket in this case.
        if (strpos($headers, "HTTP/1.1") !== 0 || preg_match("!Connection:\s*Close!i", $headers)) {
            $this->socketClient->close();
        }

        if (!preg_match("!^HTTP/1.\d 202 Accepted!i", $headers)) {
            throw new RuntimeException(
                sprintf(
                    "Graylog-Server didn't answer properly, expected 'HTTP/1.x 202 Accepted', response is '%s'",
                    trim($headers)
                )
            );
        }

        return $byteCount;
    }
311684
public function processRequest(RequestInterface $request)
    {
        $this->response = $this->communicator->getResponse($request);

        return $this->response;
    }
161593
protected function getMethod(MessageInterface $message, string $prefix = null): Closure
    {
        $name = lcfirst($prefix . $message->getPayloadType()->getName());
        if (method_exists($this, $name) === false) {
            throw new MethodNotFound($message);
        }

        return (new ReflectionMethod($this, $name))->getClosure($this);
    }
396671
public function processEmail(Message $message)
    {
        // uncompress the message variables
        $body = json_decode($message->getBody());
        $message = $this->uncompressMessage($body->m);
        $variables = $message;
        if (property_exists($body, 'v')) {
            $variables = $this->uncompressMessage($body->v);
        }

        $this->send($message, $body->t, $variables);
    }
380586
public function receiveMessage($message, WorkflowEngine $workflowEngine)
    {
        if ($message instanceof WorkflowMessage) {
            if (MessageNameUtils::isProcessingCommand($message->messageName())) {
                $this->perform($workflowEngine, $message);
                return;
            }

            $this->assertTaskEntryExists($message->processTaskListPosition());

            $this->recordThat(TaskEntryMarkedAsDone::at($message->processTaskListPosition()));

            $this->perform($workflowEngine, $message);
            return;
        }

        if ($message instanceof LogMessage) {
            $this->assertTaskEntryExists($message->processTaskListPosition());

            $this->recordThat(LogMessageReceived::record($message));

            if ($message->isError()) {
                $this->recordThat(TaskEntryMarkedAsFailed::at($message->processTaskListPosition()));
            } elseif ($this->isSubProcess() && $this->syncLogMessages) {
                //We only sync non error messages, because errors are always synced and then they would be received twice
                $messageForParent = $message->reconnectToProcessTask($this->parentTaskListPosition);
                $workflowEngine->dispatch($messageForParent);
            }
        }
    }
366473
protected function createMessageBody(MessageInterface $message)
    {
        $body = [];
        if ($message instanceof BaseOptionedModel) {
            $body = $message->getOptions();
        }
        $body['message'] = $message->getText();

        return json_encode($body);
    }
1374219
private function persistHeaders(MessageInterface $message)
    {
        // Clone the response to not destroy any necessary headers when caching
        $headers = array_diff_key($message->getHeaders(), self::$noCache);

        // Cast the headers to a string
        foreach ($headers as &$value) {
            $value = implode(', ', $value);
        }

        return $headers;
    }
620439
private function dispatch(MessageInterface $message): bool
    {
        $type = $message->getType();

        if (!$this->hasHandlers($type)) {
            return true;
        }

        $result = true;

        foreach ($this->handlers[$type] as $handler) {
            $handlerResult = $this->handleMessage($handler, $message);

            if ($result && !$handlerResult) {
                $result = false;
            }
        }

        return $result;
    }
460679
private function handleMessage(HandlerInterface $handler, MessageInterface $message): bool
    {
        // check if handler implements Psr\Log\LoggerAwareInterface (you can use \Psr\Log\LoggerAwareTrait)
        if ($this->logger && method_exists($handler, 'setLogger')) {
            $handler->setLogger($this->logger);
        }

        try {
            $result = $handler->handle($message);

            $this->log(
                LogLevel::INFO,
                "End handle message #{$message->getId()} ({$message->getType()})",
                $this->messageLoggerContext($message)
            );
        } catch (Exception $e) {
            $this->log(
                LogLevel::ERROR,
                "Handler " . get_class($handler) . " throws exception - {$e->getMessage()}",
                ['error' => $e, 'message' => $this->messageLoggerContext($message), 'exception' => $e]
            );
            Debugger::log($e, Debugger::EXCEPTION);
            $result = false;
        }
        return $result;
    }
1045638
public function send(MessageInterface $message, &$failures = null)
    {
        return $this->mailer->send($message(), $failures);
    }
428699
public function buildNotifications(MessageInterface $message)
    {
        $notifications  = new \ArrayIterator();
        $recipientQueue = new \SplQueue();
        $recipientChunk = new \ArrayIterator();

        foreach ($message->getRecipientDeviceCollection() as $recipient) {
            $recipientChunk->append($recipient);

            if ($recipientChunk->count() >= $message->getMaxRecipientsPerMessage()) {
                $recipientQueue->enqueue($recipientChunk);
                $recipientChunk = new \ArrayIterator();
            }
        }

        if ($recipientChunk->count()) {
            $recipientQueue->enqueue($recipientChunk);
        }

        while (!$recipientQueue->isEmpty()) {
            $notification = $this->createNotification($recipientQueue->dequeue(), $message);
            $notifications->append($notification);
        }

        return $notifications;
    }