@param Message $message
@param Envelope $envelope

@return bool
494001
protected function consumeCallback(Message $message, Envelope $envelope = null) : bool
    {
        if ($message->getMessage()) {
            $this->count++;
        }

        if ($message->getMessage()) {
            return $this->consume($message, $envelope);
        }

        if ($this->mustQuit()) {
            $message->quit(true);
        }

        return true;
    }
1284463
private function isRequestMethodOk(Envelope $envelope, Request $request): bool
    {
        if (in_array($request->getMethod(), $this->allowedMethods)) {
            return true;
        }

        $envelope
            ->set('code', Code::UNIMPLEMENTED)
            ->set('http_code', HttpCode::HTTP_METHOD_NOT_ALLOWED())
            ->set('error_name', 'MethodNotAllowed')
            ->set(
                'error_message',
                sprintf('You can only use HTTP [%s] on this service.', implode(',', $this->allowedMethods))
            );

        return false;
    }
749113
public function send(SmsMessage $message, $to)
    {

        $text = trim($message->content);
        if ($this->validate()) {
            return $this->push($to, $text);
        }

        return false;
    }
1080415
public function hasExceptionOfMessage($message)
    {
        foreach ($this->_exceptions as $e) {
            if ($message == $e->getMessage()) {
                return true;
            }
        }

        return false;
    }
291087
protected function buildEnvelope($method = 'email', $email = null, $message = null)
    {
        return new Envelope(
                array(
                    'sendMethod' => $method,
                    'email' => $email,
                    'invoiceEmail' => $message,
                )
        );
    }
926498
public function reply ($message) {
		// Make sure we actually have a text to reply to
		if (!$this->inbound_message) {
			return false;
		}

		return $this->sendText($this->from, $this->to, $message);
	}
75101
public function hasMessage(GmailMessageInterface $message): bool
    {
        if ($this->messages->contains($message)) {
            return true;
        }

        return false;
    }
913910
public function receive($envelope, $queue) {
        $nowTime = time();
        // 保持心跳
        $this->ping();

        $msgJson = $envelope->getBody();
        $msgJson = json_decode($msgJson, true);
        if ($msgJson['eventKey'] == 'Console') {
            $queue->ack($envelope->getDeliveryTag());
            switch ($msgJson['data']) {
                case 'reload':
                    echo sprintf("[date:%s]Event RabbitMQ: queue key '%s' reload.\n", date('Y-m-d H:i:s', $nowTime), $this->queueKey);
                    $this->rabbtMq->disconnection();
                    exit();
                    break;
                case 'ping':
                    echo sprintf("[date:%s]Event RabbitMQ: queue key '%s' ping.\n", date('Y-m-d H:i:s', $nowTime), $this->queueKey);
                    return;
            }
            return ;
        }

        // 消息转发到指定队列执行
        if ($this->queueConfig['forward']) {
            $msgJson['forward'] = $this->queueKey ;
            $msgJson['forwardLists'][] = $this->queueKey;// 转发列表，可能发生多次转发
            $forwardQueueConfig = RabbitMqConfig::getQueue($this->queueConfig['forward_queue_key']);
            // 消息转发
            $rabbitMq = new RabbitMq($this->exchangeName, $this->exchangeType);
            $rabbitMq->send($msgJson, $forwardQueueConfig['route_key']);
            $queue->ack($envelope->getDeliveryTag());
            return;
        }

        // 设置事件消费是的队列信息
        $this->eventUseQueueKey     = $this->queueKey;
        $this->eventUseQueueConfig  = $this->queueConfig;
        $this->eventUseQueueType    = $this->queueType;
        // 消息来自转发
        if (isset($msgJson['forward'])) {
            $this->eventUseQueueKey     = $msgJson['forwardLists'][0];
            $this->eventUseQueueConfig  = RabbitMqConfig::getQueue($this->eventUseQueueKey);
            $this->eventUseQueueType    = $this->eventUseQueueConfig['queue_type'];
        }

        // 消息版本不一致, 队列重启
        if ($this->version != $msgJson['version']) {
            echo sprintf("[date:%][message-version: %s][queue-version: %]The message version is different from the queue version \n", date('Y-m-d H:i:s', $nowTime));
            $queue->ack($envelope->getDeliveryTag());
            exit();
        }
        
        try {

            $callbackResult = $this->callback($msgJson);
            if ($callbackResult->getCode() === Code::SUCCESS) {
                $queue->ack($envelope->getDeliveryTag());
                $this->updateAdditional($msgJson, $callbackResult, null);
            } else {
                $queue->ack($envelope->getDeliveryTag());
                $this->updateAdditional($msgJson, $callbackResult, null);
            }

        } catch (\Exception $e) { // 非法消息,直接确认
            echo $e;
            $queue->ack($envelope->getDeliveryTag());
            $callbackResult = new CallbackResult(Code::EXCEPTION, [], '异常');
            $this->updateAdditional($msgJson, $callbackResult, $e);
            return ;
        }

    }
65011
public function sendMessageToEmail(Message $message, $user) {
        if (is_object($user)) {
            $toEmail = $user->getEmail();
        } else {
            $toEmail = $user;
        }

        if ($this->container->hasParameter('mailer_email')) {
            $mailer_email = $this->container->getParameter('mailer_email');
        } else if ($this->container->hasParameter('mailer_user')) {
            $mailer_email = $this->container->getParameter('mailer_user');
        } else {
            $mailer_email = "define@youremail.com";
        }
        $emailMessage = \Swift_Message::newInstance()
            ->setSubject($message->getMySenderEmailSubject($this->container, $user))
            ->setFrom($mailer_email)
            ->setTo($toEmail)
            ->setBody(
                $message->getMySenderEmailBody($this->container, $user),
                'text/html'
            )
        ;

        $senderAttach = $message->getMySenderEmailAttach($this->container, $user);
        if ($senderAttach != null) {
            if (is_array($senderAttach)) {
                foreach($senderAttach as $attach) {
                    $emailMessage->attach(\Swift_Attachment::fromPath($attach));
                }
            } else {
                $emailMessage->attach(\Swift_Attachment::fromPath($senderAttach));
            }
        }

        $this->container->get('mailer')->send($emailMessage);

        return true;
    }
1020278
private function handleEvent(Envelope $envelope, Request $request, Event $event): Envelope
    {
        try {
            $this->pbjx->publish($event);
        } catch (\Throwable $e) {
            return $this->handleException($envelope, $request, $event, $e);
        }

        $envelope
            ->set('code', Code::OK)
            ->set('http_code', HttpCode::HTTP_ACCEPTED())
            ->set('message_ref', $event->generateMessageRef());

        if ($request->attributes->getBoolean('pbjx_console')) {
            $envelope->set('message', $event);
        }

        return $envelope;
    }
1010684
public function setStatus($status, $envelope = null)
	{
		// Check if we're not in a final state yet
		if ($this->status > 0) {
			throw new \RuntimeException('Cannot change status from final state ' . $this->status . ' to state ' . $status . '.');
		}

		// Check if this is a valid state
		if ( !in_array($status, array_keys(self::$statusDescriptionMapping)) ) {
			throw new \InvalidArgumentException('Status ' . $status . ' is not a valid status.');
		}

		// Check if the retry envelope is not this envelope
		if ($this === $envelope) {
			throw new \InvalidArgumentException('Retry envelope cannot be set to this envelope.');
		}

		// Save it!
		$this->status = $status;
		$this->retryEnvelope = $envelope;
	}
585796
protected void processEnvelope(Envelope envelope) {

        ActorScope scope = envelope.getScope();

        // Log.d("ACTOR_TRACE", "processEnvelope: " + envelope.getMessage() + " -> " + scope.getPath());

        if (actorSystem.getTraceInterface() != null) {
            actorSystem.getTraceInterface().onEnvelopeDelivered(envelope);
        }

        long start = ActorTime.currentTime();
        boolean isDisconnected = false;

        if (scope.getActor() == null) {
            if (envelope.getMessage() == PoisonPill.INSTANCE) {
                // Not creating actor for PoisonPill
                return;
            }
            try {
                Actor actor = scope.getProps().create();
                CurrentActor.setCurrentActor(actor);
                actor.initActor(scope.getUuid(), scope.getPath(), new ActorContext(scope), scope.getMailbox());
                for (ActorExtension e : actor.getExtensions()) {
                    e.preStart();
                }
                actor.preStart();
                scope.onActorCreated(actor);
            } catch (Exception e) {
                e.printStackTrace();
                if (envelope.getSender() != null) {
                    envelope.getSender().send(new DeadLetter("Unable to create actor"));
                }
                return;
            }
        }

        try {
            if (envelope.getMessage() == StartActor.INSTANCE) {
                // Already created actor
                return;
            } else if (envelope.getMessage() == Ping.INSTANCE) {
                // No op
                return;
            } else if (envelope.getMessage() == PoisonPill.INSTANCE) {
                isDisconnected = true;
                for (ActorExtension e : scope.getActor().getExtensions()) {
                    e.postStop();
                }
                scope.getActor().postStop();
                scope.onActorDie();
                for (Envelope e : scope.getMailbox().allEnvelopes()) {
                    if (e.getSender() != null) {
                        e.getSender().send(new DeadLetter(e.getMessage()));
                    }
                }
                scope.getMailbox().clear();
                synchronized (LOCK) {
                    scopes.remove(scope.getPath());
                    endpoints.remove(scope.getPath());
                    dispatcher.getQueue().disconnectMailbox(scope.getMailbox());
                }
            } else {
                CurrentActor.setCurrentActor(scope.getActor());
                scope.setSender(envelope.getSender());
                for (ActorExtension e : scope.getActor().getExtensions()) {
                    if (e.onReceive(envelope.getMessage())) {
                        return;
                    }
                }
                scope.getActor().onReceive(envelope.getMessage());
            }
        } catch (Exception e) {
            if (actorSystem.getTraceInterface() != null) {
                actorSystem.getTraceInterface().onActorDie(scope.getActorRef(), e);
            }
            scope.onActorDie();
            isDisconnected = true;
            synchronized (LOCK) {
                scopes.remove(scope.getPath());
                endpoints.remove(scope.getPath());
                dispatcher.getQueue().disconnectMailbox(scope.getMailbox());
            }
        } finally {
            if (actorSystem.getTraceInterface() != null) {
                actorSystem.getTraceInterface().onEnvelopeProcessed(envelope, ActorTime.currentTime() - start);
            }
            CurrentActor.setCurrentActor(null);
            if (!isDisconnected) {
                dispatcher.getQueue().unlockMailbox(envelope.getMailbox());
            }
        }
    }
755247
private function isJsonOk(Envelope $envelope, Request $request): bool
    {
        if (JSON_ERROR_NONE === json_last_error()) {
            return true;
        }

        $envelope
            ->set('code', Code::INVALID_ARGUMENT)
            ->set('http_code', HttpCode::HTTP_UNSUPPORTED_MEDIA_TYPE())
            ->set('error_name', 'UnsupportedMediaType')
            ->set('error_message', 'Invalid json: ' . json_last_error_msg());

        return false;
    }
91180
public function send(Message $message) : void
    {
        header('Content-Type: text/xml', true);

        $xml = $this->getOutputToSend($message);
        if ($xml !== false) {
            Utils::getContainer()->debugMessage($xml, 'out');
            echo $xml;
        }

        // DOMDocument::saveXML() returned false. Something is seriously wrong here. Not much we can do.
        exit(0);
    }
845994
protected function fireEvent($event, $message)
    {
        if (!isset($this->notification)) {
            return true;
        }

        return $this->getNotification()->fire($event, $this, $message);
    }
1051226
public function deleteMessage(Message $message)
    {
        if ($this->db->delete('QueueMessages', ['handle' => $message->handle])) {
            return true;
        }

        return false;
    }
188244
protected function retryMessage(EnvelopeInterface $envelope, \Exception $exception = null)
    {
        $attempt = $this->getAttemptValue($envelope);
        if ($attempt >= $this->maxAttempts) {
            throw new ProcessExhaustedException(sprintf('Exhausted after failing %d attempt(s)', $attempt));
        }

        $this->logger->debug(sprintf('Requeueing message (%d attempts left)', $this->maxAttempts - $attempt));

        return $this->strategy->retry($envelope, ++$attempt, $exception);
    }