@param array $options
@param       $message

@throws \Kerox\Messenger\Exception\MessengerException

@return array
184881
private function isValidOptions(array $options, $message): array
    {
        $allowedOptionsKeys = $this->getAllowedOptionsKeys();
        foreach ($options as $key => $value) {
            if (!\in_array($key, $allowedOptionsKeys, true)) {
                throw new InvalidOptionException(sprintf(
                    'Only "%s" are allowed keys for options.',
                    implode(', ', $allowedOptionsKeys)
                ));
            }

            if ($key === self::OPTION_MESSAGING_TYPE) {
                $this->isValidMessagingType($value);
            } elseif ($key === self::OPTION_NOTIFICATION_TYPE) {
                $this->isValidNotificationType($value);
            } elseif ($key === self::OPTION_TAG) {
                $this->isValidTag($value, $message);
            }
        }

        return $options;
    }
1002260
private function getAttachmentsAsArrays(Message $message): array
    {
        $attachments = [];

        foreach ($message->getAttachments() as $attachment) {
            $attachments[] = $attachment->toArray();
        }

        return $attachments;
    }
699555
public function appendMessage($message, $options = null, $internal_date = null) {
        return imap_append($this->client->getConnection(), $this->path, $message, $options, $internal_date);
    }
580195
protected function readMessageDefaults(\SimpleXMLElement $message): array
    {
        // init
        $defaults = [];

        /* @var \SimpleXMLElement $defaultValueNode */
        foreach ($message->defaults[0] ?? [] as $defaultValueNode) {
            $attributes = $defaultValueNode->attributes();
            $id = (string)$attributes->id;
            $defaults[$id] = (string)$attributes->value;
        }

        return $defaults;
    }
485598
public function encode(Message $message, int $options = 0): string
    {
        $ctx = new Context($options);
        $qdCount = $anCount = $nsCount = $arCount = 0;

        foreach ($message->getQuestionRecords() as $record) {
            if (!$this->encodeQuestionRecord($ctx, $record)) {
                goto done;
            }

            $qdCount++;
        }

        foreach ($message->getAnswerRecords() as $record) {
            if (!$this->encodeResourceRecord($ctx, $record)) {
                goto done;
            }

            $anCount++;
        }

        foreach ($message->getAuthorityRecords() as $record) {
            if (!$this->encodeResourceRecord($ctx, $record)) {
                goto done;
            }

            $nsCount++;
        }

        foreach ($message->getAdditionalRecords() as $record) {
            if (!$this->encodeResourceRecord($ctx, $record)) {
                goto done;
            }

            $arCount++;
        }

        done:
        $packet = $this->encodeHeader($ctx, $message, $qdCount, $anCount, $nsCount, $arCount) . $ctx->data;

        if (!($options & EncodingOptions::FORMAT_TCP)) {
            \assert(
                \strlen($packet) <= 512,
                new \Error('UDP packet exceeds 512 byte limit: got ' . \strlen($packet) . ' bytes')
            );

            return $packet;
        }

        \assert(
            \strlen($packet) <= 65535,
            new \Error('TCP packet exceeds 65535 byte limit: got ' . \strlen($packet) . ' bytes')
        );

        return \pack('n', \strlen($packet)) . $packet;
    }
1375066
public function validate($message): array
    {
        if (\is_array($message)) {
            $message = CloudMessage::fromArray($message);
        }

        if (!($message instanceof Message)) {
            throw new InvalidArgumentException(
                'Unsupported message type. Use an array or a class implementing %s'.Message::class
            );
        }

        try {
            $response = $this->messagingApi->validateMessage($message);
        } catch (NotFound $e) {
            throw (new InvalidMessage($e->getMessage(), $e->getCode()))
                ->withResponse($e->response());
        }

        return JSON::decode((string) $response->getBody(), true);
    }
980055
public function startDelete($message, array $options = null) {
		return $this->startEvent($message, ProfileInterface::TYPE_DELETE, $options);
	}
747683
protected function applyMessageOptions(Message $message)
    {
        if (empty($this->config['message']['options']) || !is_array($this->config['message']['options'])) {
            return $message;
        }

        foreach ($this->config['message']['options'] as $key => $value) {
            $method = 'set'.ucfirst($key);

            if (is_callable(array($message, $method))) {
                if (!is_array($value)) {
                    $value = array($value);
                }
                call_user_func_array(array($message, $method), $value);
            }
        }

        return $message;
    }
144399
protected function parseMessage($message)
    {
        $headerAndBody = explode("\r\n\r\n", $message, 2);
        $headers = explode("\r\n", $headerAndBody[0]);

        return [$this->parseHeader($headers), $headerAndBody[1]];

    }
296130
public function checkMessages(array $options = [])
    {
        $this->buildBody($options);

        $rawMessages = json_decode($this->getRequest()->getBody()->getContents());

        return $this->makeMessages($rawMessages->data);
    }
471175
public function handle(Message $message, array $options=[]) : PromiseInterface
    {
        $promise = new Promise(function () use (&$promise, $message, $options) {
            $promise->resolve(call_user_func(
                $this->callback,
                $message,
                $options
            ));
        });

        return $promise;
    }
732713
public function execute($options = array())
    {
        $message = $options['message'];
        $receivers = $message->getReceivers();

        $options['provide_link'] = true;
        foreach ($receivers as $receiver) {
            $this->sendNotification($receiver, $message->getBody(), array('message_type' => 'PM'));
        }
    }
1134075
public function publish(array $message, array $options = [])
    {
        $options      = $this->mergeOptions($options);
        $publishStart = microtime(true);

        if (!$this->queueExists()) {
            $this->create();
        }

        $result = $this->ironmq->postMessage(
            $this->getNameWithPrefix(),
            json_encode($message + ['_qpush_queue' => $this->name]),
            [
                'timeout'       => $options['message_timeout'],
                'delay'         => $options['message_delay'],
                'expires_in'    => $options['message_expiration']
            ]
        );

        $context = [
            'message_id'    => $result->id,
            'publish_time'  => microtime(true) - $publishStart
        ];
        $this->log(200, "Message has been published.", $context);

        return (int) $result->id;
    }
998829
public function message($message, $options = null) {
		$options = $this->_mergeOptions($options);

		$this->set($message, $options);

		$this->_assertSessionStackSize($options);
	}
174896
public function serialize(Message $message, array $options = [])
    {
        $userMessage = $this->getUserMessage($message);

        //mainly for tests or if someting went wrong
        if (!$userMessage) {
            $userMessage = new UserMessage();
        }

        $data = [
          'id' => $message->getUuid(),
          'object' => $message->getObject(),
          'content' => $message->getContent(),
          'to' => $message->getTo(),
          'meta' => [
            'date' => DateNormalizer::normalize($message->getDate()),
            'read' => $userMessage->isRead(),
            'removed' => $userMessage->isRemoved(),
            'sent' => $userMessage->isSent(),
            'umuuid' => $userMessage->getUuid(),
          ],
        ];

        $data['from'] = $message->getSender() ?
            $this->userSerializer->serialize($message->getSender(), [Options::SERIALIZE_MINIMAL]) :
            ['username' => $message->getSenderUsername()];

        if (in_array(Options::IS_RECURSIVE, $options)) {
            $data['children'] = array_map(function (Message $child) {
                return $this->serialize($child, $options);
            }, $message->getChildren()->toArray());
        }

        return $data;
    }
938657
public function process(QueueMessage $message, Structure $options)
    {
        $maxTries = (int) $options->get('tries', 5);

        $job = $message->getJob();
        /** @var JobInterface $job */
        $job = unserialize($job);

        try {
            // @before event
            $this->triggerEvent(
                'onWorkerBeforeJobRun',
                [
                    'worker' => $this,
                    'message' => $message,
                    'job' => $job,
                    'manager' => $this->manager,
                ]
            );

            // Fail if max attempts
            if ($maxTries !== 0 && $maxTries < $message->getAttempts()) {
                $this->manager->delete($message);

                throw new MaxAttemptsExceededException('Max attempts exceed for Message: ' . $message->getId());
            }

            // run
            $job->execute();

            // @after event
            $this->triggerEvent(
                'onWorkerAfterJobRun',
                [
                    'worker' => $this,
                    'message' => $message,
                    'job' => $job,
                    'manager' => $this->manager,
                ]
            );

            $this->manager->delete($message);
        } catch (\Throwable $t) {
            $this->handleJobException($job, $message, $options, $t);
        } finally {
            if (!$message->isDeleted()) {
                $this->manager->release($message, (int) $options->get('delay', 0));
            }
        }
    }
737918
public function createContext(array $options, MessageInterface $message = null, Exchange $exchange = null, \Exception $exception = null)
    {
        $context = [
            self::CONTEXT_OPTIONS => $options,
            self::CONTEXT_VARS => [],
            self::CONTEXT_RESULTS => [],
        ];

        if ($exchange) {
            $context[self::CONTEXT_EXCHANGE] = $exchange;
            if (!$message) {
                $message = $exchange->getIn();
            }
        }

        if ($message) {
            $context[self::CONTEXT_MSG] = $message;
            $context[self::CONTEXT_HEADERS] = $message->getHeaders();
            $context[self::CONTEXT_BODY] = $message->getBody();
        }

        if ($exception) {
            $context[self::CONTEXT_EXCEPTION] = $exception;
        }

        return $context;
    }