消息处理

@param \AMQPEnvelope $envelope
@param \AMQPQueue $queue
548657
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: gateway reload.\n", date('Y-m-d H:i:s', $nowTime));
                    $this->rabbtMq->disconnection();
                    exit();
                    break;
                case 'ping':
                    echo sprintf("[date:%s]Event RabbitMQ: gateway ping.\n", date('Y-m-d H:i:s', $nowTime));
                    return;
                    break;
            }
            return ;
        }
        try {
            $eventConfig = RabbitMqConfig::getEvent($msgJson['eventKey']);
        } catch (ConfigException $e) { // 非法消息,直接确认
            echo sprintf("[date:%s]Event RabbitMQ: eventKey \"%s\" not found.\n", date('Y-m-d H:i:s', $nowTime), $msgJson['eventKey']);
            $queue->ack($envelope->getDeliveryTag());
            return ;
        }

        // 合法的消息持久化
        $eventData = [
            '_id'           => $msgJson['_id'],
            'status'        => 0,// 未处理状态
            'eventKey'      => $msgJson['eventKey'],
            'time'          => $nowTime,
            'ip'            => $msgJson['ip'],
            'version'       => $this->version,
            'data'          => $msgJson['data'],
        ];

        // 持久化到mongodb
        try {
            $collectionName = Helper::getCollectionName();
            MongoDB::getInstance()->insert($collectionName, $eventData);
        } catch (\Exception $e) {
            echo sprintf("[date:%s]Event RabbitMQ: %s.\n", date('Y-m-d H:i:s', $nowTime), $e->getMessage());
            $queue->ack($envelope->getDeliveryTag());
        }
        // 消息转发
        $queueConfig = RabbitMqConfig::getQueue($eventConfig['queue_key']);
        $rabbitMq = new RabbitMq($this->exchangeName, $this->exchangeType);
        $rabbitMq->send($eventData, $queueConfig['route_key']);
        $queue->ack($envelope->getDeliveryTag());
    }
905708
public function pushMessage($queueName, $message)
    {
        $amqpMessage = new AMQPMessage($message, $this->defaultMessageParams);
        $this->getChannel()->basic_publish($amqpMessage, $this->exchange, $queueName);
    }
1008628
public function setQueue(QueueContract $queue)
    {
        if ($this->mailer instanceof MailerContract) {
            $this->mailer->setQueue($queue);
        }

        $this->queue = $queue;

        return $this;
    }
433376
protected function doOnce(string $queueName, Envelope $envelope) : array
    {
        $result = false;
        $message = $envelope->unwrap();

        $this->getLogger()->debug('Handling message {msg}', ['msg' => self::nameOf($message)]);
        try {
            $result = $this->handleMessage($message);
        } catch (Exception\MustStop $e) {
            $this->driver->ack($queueName, $envelope);
            throw $e;
        } catch (Exception\ShouldReleaseMessage $e) {
            $this->driver->release($queueName, $envelope);
            $this->getLogger()->debug('Releasing message {msg} due to {cls} exception: {err}', [
                'msg' => self::nameOf($message),
                'cls' => get_class($e),
                'err' => $e->getMessage(),
            ]);
            return [$result, true];
        } catch (\Exception $e) {
            // any other exception is simply logged. We and marked as failed
            // below. We only log here because we can't make guarantees about
            // the implementation of the handler and whether or not it actually
            // throws exceptions on failure (see PcntlForkingHandler).
            $this->getLogger()->critical('Unexpected {cls} exception handling {name} message: {msg}', [
                'cls'   => get_class($e),
                'name'  => self::nameOf($message),
                'msg'   => $e->getMessage()
            ]);
            $result = false;
        }
        $this->getLogger()->debug('Handled message {msg}', ['msg' => self::nameOf($message)]);

        if ($result) {
            $willRetry = false;
            $this->driver->ack($queueName, $envelope);
            $this->getLogger()->debug('Acknowledged message {msg}', ['msg' => self::nameOf($message)]);
        } else {
            $willRetry = $this->failed($queueName, $envelope);
            $this->getLogger()->debug('Failed message {msg}', ['msg' => self::nameOf($message)]);
        }

        return [$result, $willRetry];
    }
1221000
protected function consume(Message $message, Envelope $envelope = null) : bool
    {
        if (!$envelope instanceof Job) {
            throw new InvalidException(sprintf("Invalid class '%s'", get_class($envelope)));
        }

        try {
            $job = $envelope->getCommand();

            $options = json_encode($job->getInput()->getOptions());
            $arguments = json_encode($job->getInput()->getArguments());

            if (!$this->output->isDebug()) {
                $arguments = strlen($arguments) > 50 ? substr($arguments, 0, 50) . '...' : $arguments;
                $options = strlen($options) > 50 ? substr($options, 0, 50) . '...' : $options;
            }

            $this->output->writeln(sprintf(
                "New job with options '%s' and arguments '%s'",
                $envelope->getClass(),
                $options,
                $arguments
            ), OutputInterface::VERBOSITY_VERY_VERBOSE);

            $job->setOutput($this->output);

            $return = $job->execute($job->input, $this->output) == 0;

            return $return;
        } catch (\Throwable $exception) {
            if (!$exception instanceof \Exception) {
                $exception = new FatalThrowableError($exception);
            }

            $this->output->writeError(sprintf(
                "%s on job '%s' with message '%s'",
                get_class($exception),
                $envelope->getClass(),
                $exception->getMessage()
            ));

            Application::writeException($exception, $this->output, ConsoleOutputInterface::VERBOSITY_VERBOSE);

            self::emit(new ErrorEvent($exception));

            if ($this->input->getOption('retry') && !$exception instanceof FailureException) {
                if ($this->input->getOption('retry') > count($envelope->getHistory())) {
                    if ($this->input->getOption('stop-on-error')) {
                        $this->exitCode = 1;
                    }

                    $this->queue()->publish($envelope->serialize());
                    throw new ManualRequeueException($this->input->getOption('stop-on-error') === 1);
                } else {
                    $errorMessage = sprintf(
                        "Max retries %s reach for %s on job '%s' with message '%s'",
                        $this->input->getOption('retry'),
                        get_class($exception),
                        $envelope->getClass(),
                        $exception->getMessage()
                    );

                    $this->output->writeError($errorMessage);
                    throw new FailureException($message, $errorMessage, $exception);
                }
            }

            if ($this->input->getOption('stop-on-error') && !$exception instanceof FailureException) {
                $this->exitCode = 1;
                throw $exception;
            }

            return true;
        }
    }
1111714
public function build(AMQPEnvelope $envelope)
    {
        $wrapper = $this->wrap($envelope);

        return new $this->delivery_skeleton($wrapper->getBody(), $wrapper->getEnvelope(), $wrapper->getProperties());
    }
1290613
protected static function _saveBuffer($queue){
        $worker_last_report_status  = self::$_last_busy_ts >= self::$_last_idle_ts ? 'busy' : 'idle';
        $worker_last_report_ts      = max( self::$_last_busy_ts ,self::$_last_idle_ts );
        
        $worker_pid = self::getPid();
        $worker_key = self::worker_status_monitoring_key_prefix.$worker_pid;
        $value = json_encode([
            'queue'                     => $queue,
            'comment'                   => 'worker_status_report',
            'worker_last_report_status' => $worker_last_report_status,      // worker 进程最后一次触发事件时的状态
            'worker_last_report_ts'     => $worker_last_report_ts,          // worker 进程最后一次触发事件的时间
            'save_ts'                   => time(),                          // 本次缓冲中的数据保存到 redis 的时间
            'last_save_ts'              => self::$_buffer_last_flush_ts,    // 上次缓冲中的数据保存到 redis 的时间
            'busy'                      => [
                'process_count_meanwhile' => self::$_busy_event_buffer_counter, // 可用于计算任务的消费速度
                'last_process_ts'         => self::$_last_busy_ts,
            ],
            'idle'                      => [
                'sleep_count_meanwhile' => self::$_idle_event_buffer_counter,
                'last_sleep_ts'         => self::$_last_idle_ts,
            ],
        ]);

        $flushHandler = self::getRedisForStatics();
        $flushHandler->rPush($worker_key ,$value);     // 添加到 List 的尾部
        unset($flushHandler);
    }
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;
    }
780217
function(envelope, $done) {
    var __pushJob = _.bind(pushNewJob, null, envelope)
      , flow = [__pushJob]

    if (envelope.id) {
      var __rmJob = _.bind(removeMappedJob, null, envelope.jobType, envelope.id)
      flow = [__rmJob].concat(flow)
    }

    async.series(flow, function(err, results) {
      var _job = results.length > 1 ? results[1] : results[0]
        , _rm = results.length > 1 ? results[0] : null

      if (_rm)
        LOG.info(['Replaced old job [', envelope.id, ']'].join(' '));

      $done(err, _job)
    })
  }
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;
    }
337250
protected function _consume(AMQPQueue $queue, callable $callable)
    {
        $tag = uniqid() . microtime(true);

        $callback = function (AMQPEnvelope $envelope, AMQPQueue $queue) use ($callable) {
            return $this->_callback($callable, $envelope, $queue);
        };

        $queue->consume($callback, AMQP_NOPARAM, $tag);
    }
901149
public function send(QueueInterface $queue, Message $message)
    {
        $headers = $message->getHeaders();
        $properties = $message->getProperties();

        $headers['content_type'] = $message->getContentType();

        $transportMessage = $this->createTransportMessage();
        $transportMessage->setBody($message->getBody());
        $transportMessage->setHeaders($headers);
        $transportMessage->setProperties($properties);

        $transportMessage->setMessageId($message->getMessageId());
        $transportMessage->setTimestamp($message->getTimestamp());

        if ($message->getDelay()) {
            $transportMessage->setDelay($message->getDelay());
        }

        if ($message->getPriority()) {
            $this->setMessagePriority($transportMessage, $message->getPriority());
        }

        if ($message->getExpire()) {
            throw new \InvalidArgumentException('Expire is not supported by the transport');
        }

        $this->session->createProducer()->send($queue, $transportMessage);
    }
258039
private function handleRequest(Envelope $envelope, Request $request, PbjxRequest $pbjxRequest): Envelope
    {
        try {
            $response = $this->pbjx->request($pbjxRequest);
        } catch (\Throwable $e) {
            return $this->handleException($envelope, $request, $pbjxRequest, $e);
        }

        return $envelope
            ->set('code', Code::OK)
            ->set('http_code', HttpCode::HTTP_OK())
            ->set('etag', $response->get('etag'))
            ->set('message_ref', $response->generateMessageRef())
            ->set('message', $response);
    }
239353
public void schedule(Envelope envelope) {
        if (envelope.getMailbox() != this) {
            throw new RuntimeException("envelope.mailbox != this mailbox");
        }

        queueCollection.post(queueId, envelope);
    }
211758
protected function failed(string $queueName, Envelope $env) : bool
    {
        $retry = $this->retries->canRetry($env);
        if ($retry) {
            $delay = $this->retries->retryDelay($env);
            $this->getDriver()->retry($queueName, $env->retry($delay));
        } else {
            $this->getDriver()->fail($queueName, $env);
        }

        return $retry;
    }
291647
@Override
	public void queueEnvelope(Envelope envelope) {
		if (envelope.hasBuffer()) {
			throw new IllegalStateException("Envelope for OutputChannel has Buffer attached.");
		}

		for (AbstractEvent event : envelope.deserializeEvents()) {
			if (event.getClass() == ChannelCloseEvent.class) {
				synchronized (this.closeLock) {
					this.receiverCloseRequested = true;
					this.closeLock.notifyAll();
				}
			} 
			else if (event instanceof AbstractTaskEvent) {
				if (LOG.isDebugEnabled()) {
					LOG.debug("OutputChannel received task event: " + event);
				}
				
				this.outputGate.deliverEvent((AbstractTaskEvent) event);
			}
			else {
				throw new RuntimeException("OutputChannel received an event that is neither close nor task event.");
			}
		}
	}
1244949
public void invalidate(Envelope envelope) {
		List<String> keys = index.getOverlappingKeys(envelope);
		if (CacheIndexService.ALL_KEYS == keys) {
			log.debug("clear all keys from cache");
			clear();
		} else {
			log.debug("invalidate keys {}", keys);
			for (String key : keys) {
				remove(key);
			}
		}
	}