RongtongJin commented on code in PR #1256: URL: https://github.com/apache/rocketmq-clients/pull/1256#discussion_r3662348481
########## php/SendMessageHandler.php: ########## @@ -0,0 +1,692 @@ +<?php +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +namespace Apache\Rocketmq; + +use Apache\Rocketmq\V2\MessagingServiceClient; +use Apache\Rocketmq\V2\SendMessageRequest; +use Apache\Rocketmq\V2\Message; +use Apache\Rocketmq\V2\SystemProperties; +use Apache\Rocketmq\V2\Resource; +use Apache\Rocketmq\V2\Encoding; +use Google\Protobuf\Timestamp; + +/** + * SendMessageHandler — Handles message sending, batching, and retry logic. + * + * Extracted from Producer to separate send concerns: + * - send() / sendAsync(): single message sending + * - sendBatch() / sendBatchAsync(): batch message sending + * - Convenience builders: priority, delayed, FIFO messages + * - Retry with deadline, queue rotation, and backoff + * - Protobuf message enrichment (toProtobufMessage) + * + * Dependencies are injected; the handler has no lifecycle state of its own. + * The Producer is responsible for running-state checks before delegating here. + */ +class SendMessageHandler +{ + private readonly Logger $logger; + + /** + * @param MessagingServiceClient $client gRPC client for send calls + * @param ProducerSettings $settings Producer configuration (retry, timeouts, etc.) + * @param MessageValidator $validator Message validation and type detection + * @param PublishingRouteManager $routeManager Route lookup and broker isolation + * @param \Closure $interceptorExecutor fn(string $hookPoint, array $context): void + * @param \Closure $metadataBuilder fn(?int $timeoutMs): array + * @param \Closure $callOptionsResolver fn(?int $overrideTimeout): array + * @param \Closure $operationTimeoutFn fn(string $operation): int (microseconds) + */ + public function __construct( + private readonly MessagingServiceClient $client, + private readonly ProducerSettings $settings, + private readonly MessageValidator $validator, + private readonly PublishingRouteManager $routeManager, + private readonly \Closure $interceptorExecutor, + private readonly \Closure $metadataBuilder, + private readonly \Closure $callOptionsResolver, + private readonly \Closure $operationTimeoutFn, + ) { + $this->logger = Logger::getInstance('Producer'); + } + + // ==================== Send ==================== + + /** + * Send a single message with retry. + * + * @param Message $message The message to send + * @return array{messageId: string, transactionId: string, recallHandle: string, code: int, message: string} + * @throws \RuntimeException If no queue is available or all retries fail + */ + public function send(Message $message): array + { + $this->validator->validateMessage($message); + + $topic = $message->getTopic()->getName(); + $loadBalancer = $this->routeManager->getPublishingLoadBalancer($topic); + + $sysProps = $message->getSystemProperties(); + $hasMessageGroup = $sysProps !== null && $sysProps->hasMessageGroup(); + if ($hasMessageGroup) { + $messageQueue = $loadBalancer->takeMessageQueueByMessageGroup($sysProps->getMessageGroup()); + if (!$messageQueue) { + throw new \RuntimeException( + "No available message queue for message group: {$sysProps->getMessageGroup()}" + ); + } + $candidates = [$messageQueue]; + } else { + $candidates = $loadBalancer->takeMessageQueue( + $this->routeManager->getIsolatedBrokerNames(), + $this->settings->getMaxAttempts() + ); + if (empty($candidates)) { + throw new \RuntimeException("No available message queue for topic: {$topic}"); + } + } + + if ($this->validator->isValidateMessageType()) { + $msgType = $this->validator->detectMessageType($message, false); + $loadBalancer->validateMessageTypeAgainstQueue($candidates[0], $msgType, $topic); + } + + $request = $this->wrapSendMessageRequest([$message], $candidates[0]); + return $this->sendMessageWithRetry($request, $message, $candidates, $this->settings->getMaxAttempts()); + } + + /** + * Send a message asynchronously (Swoole coroutine or Generator fallback). + * + * @param Message $message The message to send + * @return array|\Generator + */ + public function sendAsync(Message $message): array|\Generator + { + if (SwooleCompat::isAvailable() && SwooleCompat::inCoroutine()) { + $channel = new \Swoole\Coroutine\Channel(1); + \Swoole\Coroutine::create(function () use ($message, $channel) { + try { + $result = $this->send($message); + $channel->push(['success' => true, 'result' => $result]); + } catch (\Throwable $e) { + $channel->push(['success' => false, 'exception' => $e]); + } + }); + $data = $channel->pop($this->settings->getRequestTimeout() / 1000.0); + if ($data === false) { + throw new \RuntimeException( + "Send async Request timeout {$this->settings->getRequestTimeout()}ms" + ); + } + if (isset($data['exception'])) { + throw $data['exception']; + } + return $data['result'] ?? null; + } + return $this->sendSyncFallback($message); + } + + // ==================== Batch Send ==================== + + /** + * Send a batch of messages. + * + * All messages must share the same topic. If any message has a messageGroup (FIFO), + * all must belong to the same group. Message types must be uniform. + * + * @param array<Message> $messages Messages to send + * @return array<array{messageId: string, transactionId: string, recallHandle: string, code: int, message: string}> + * @throws \InvalidArgumentException If batch is empty, topics differ, or types/groups conflict + * @throws \RuntimeException If no queue is available or all retries fail + */ + public function sendBatch(array $messages): array + { + if (empty($messages)) { + throw new \InvalidArgumentException("Batch messages cannot be empty"); + } + + $topic = $messages[0]->getTopic()->getName(); + $messageTypes = []; + $messageGroups = []; + $hasFifoMessage = false; + foreach ($messages as $msg) { + if ($msg->getTopic()->getName() !== $topic) { + throw new \InvalidArgumentException("All messages in a batch must have the same topic"); + } + $this->validator->validateMessage($msg); + if ($this->validator->isValidateMessageType()) { + $messageTypes[] = $this->validator->detectMessageType($msg, false); + } + $sysProps = $msg->getSystemProperties(); + if ($sysProps !== null && $sysProps->hasMessageGroup()) { + $hasFifoMessage = true; + $messageGroups[] = $sysProps->getMessageGroup(); + } + } + if ($this->validator->isValidateMessageType() && count(array_unique($messageTypes)) > 1) { + throw new \InvalidArgumentException('Messages to send different message types , please check'); + } + if ($hasFifoMessage && count(array_unique($messageGroups)) > 1) { + throw new \InvalidArgumentException("FIFO messages to send have different message groups, please check"); + } + + $loadBalancer = $this->routeManager->getPublishingLoadBalancer($topic); + $isolatedBroker = $this->routeManager->getIsolatedBrokerNames(); + + if ($hasFifoMessage) { + $messageGroup = $messageGroups[0]; + $mq = $loadBalancer->takeMessageQueueByMessageGroup($messageGroup); + $messageQueue = $mq !== null ? [$mq] : []; + } else { + $messageQueue = $loadBalancer->takeMessageQueue($isolatedBroker, $this->settings->getMaxAttempts()); + } + if (empty($messageQueue)) { + throw new \RuntimeException("No available message queue for topic: {$topic}"); + } + + $request = $this->wrapSendMessageRequest($messages, $messageQueue[0]); + return $this->sendBatchWithRetry($request, $messages, $messageQueue, $this->settings->getMaxAttempts()); + } + + /** + * Send a batch of messages asynchronously (Swoole coroutine or Generator fallback). + * + * @param array<Message> $messages Messages to send + * @return array|\Generator + */ + public function sendBatchAsync(array $messages): array|\Generator + { + if (SwooleCompat::isAvailable() && SwooleCompat::inCoroutine()) { + $channel = new \Swoole\Coroutine\Channel(1); + \Swoole\Coroutine::create(function () use ($messages, $channel) { + try { + $result = $this->sendBatch($messages); + $channel->push(['success' => true, 'result' => $result]); + } catch (\Throwable $e) { + $channel->push(['success' => false, 'exception' => $e]); + } + }); + $data = $channel->pop($this->settings->getRequestTimeout() / 1000.0); + if ($data === false) { + throw new \RuntimeException( + "Send batch async Request timeout {$this->settings->getRequestTimeout()}ms" + ); + } + if (isset($data['exception'])) { + throw $data['exception']; + } + return $data['result'] ?? null; + } + return $this->sendBatchSyncFallback($messages); + } + + // ==================== Convenience Builders ==================== + + /** + * Build a message with custom system properties (used by convenience send methods). + * + * @param string $topic Topic name + * @param string $body Message body + * @param string $tag Optional message tag + * @param callable $configurator fn(SystemProperties): void to set priority/group/delay + * @return Message + */ + public function buildConvenienceMessage(string $topic, string $body, string $tag, callable $configurator): Message + { + $topicResource = new Resource(); + $topicResource->setName($topic); + + $sysProps = new SystemProperties(); + if (!empty($tag)) { + $sysProps->setTag($tag); + } + $configurator($sysProps); + + $message = new Message(); + $message->setTopic($topicResource); + $message->setBody($body); + $message->setSystemProperties($sysProps); + + return $message; + } + + // ==================== Message Building ==================== + + /** + * Detect message type via MessageValidator. + * + * @param Message $msg + * @param bool $txEnabled Whether to consider TRANSACTION type + * @return int MessageType constant + */ + public function detectMessageType(Message $msg, bool $txEnabled = false): int + { + return $this->validator->detectMessageType($msg, $txEnabled); + } + + private function createTimestamp(): Timestamp + { + $now = microtime(true); + $timestamp = new Timestamp(); + $timestamp->setSeconds((int)$now); + $timestamp->setNanos((int)(($now - (int)$now) * 1000000000)); + return $timestamp; + } + + /** + * Convert a user-facing Message into a fully enriched protobuf Message for sending. + * + * Assigns messageId, bornTimestamp, bornHost, encoding, queueId, messageType, + * and copies over optional fields (tag, keys, messageGroup, deliveryTimestamp, + * liteTopic, priority, traceContext) from the input message. + */ + private function toProtobufMessage(Message $msg, object $messageQueue, bool $txEnabled = false): Message + { + $messageId = MessageIdCodec::getInstance()->nextMessageId()->toString(); + + $systemProperties = new SystemProperties(); + $systemProperties->setMessageId($messageId); + $systemProperties->setBornTimestamp($this->createTimestamp()); + $systemProperties->setBornHost(gethostname() ?: 'localhost'); + + // Preserve encoding from input message; default to IDENTITY + $inputSysProps = $msg->getSystemProperties(); + $encoding = Encoding::IDENTITY; + if ($inputSysProps !== null) { + $inputEncoding = $inputSysProps->getBodyEncoding(); + if ($inputEncoding !== Encoding::ENCODING_UNSPECIFIED) { + $encoding = $inputEncoding; + } + } + $systemProperties->setBodyEncoding($encoding); + $queueId = $messageQueue->getId(); + if ($queueId !== null) { + $systemProperties->setQueueId($queueId); + } + $systemProperties->setMessageType($this->detectMessageType($msg, $txEnabled)); + + if ($inputSysProps) { + if ($inputSysProps->hasTag()) { + $systemProperties->setTag($inputSysProps->getTag()); + } + if (!ProtobufUtil::isRepeatedFieldEmpty($inputSysProps->getKeys())) { + $systemProperties->setKeys($inputSysProps->getKeys()); + } + if ($inputSysProps->hasMessageGroup()) { + $systemProperties->setMessageGroup($inputSysProps->getMessageGroup()); + } + if ($inputSysProps->hasDeliveryTimestamp()) { + $systemProperties->setDeliveryTimestamp($inputSysProps->getDeliveryTimestamp()); + } + if ($inputSysProps->hasLiteTopic()) { + $systemProperties->setLiteTopic($inputSysProps->getLiteTopic()); + } + if ($inputSysProps->hasPriority()) { + $systemProperties->setPriority($inputSysProps->getPriority()); + } + if ($inputSysProps->hasTraceContext()) { + $systemProperties->setTraceContext($inputSysProps->getTraceContext()); + } + } + + $topicResource = new Resource(); + $topicResource->setName($msg->getTopic()->getName()); + + $protoMsg = new Message(); + $protoMsg->setTopic($topicResource); + $protoMsg->setBody($msg->getBody()); + $protoMsg->setSystemProperties($systemProperties); + + $userProps = $msg->getUserProperties(); + if (!ProtobufUtil::isMapFieldEmpty($userProps)) { + foreach ($userProps as $key => $value) { + $protoMsg->getUserProperties()[$key] = $value; + } + } + + return $protoMsg; + } + + /** + * Wrap messages into a SendMessageRequest (non-transaction). + */ + public function wrapSendMessageRequest(array $messages, object $messageQueue): SendMessageRequest + { + $enriched = []; + foreach ($messages as $msg) { + $enriched[] = $this->toProtobufMessage($msg, $messageQueue); + } + $request = new SendMessageRequest(); + $request->setMessages($enriched); + return $request; + } + + /** + * Wrap messages into a SendMessageRequest with transaction message type. + * + * Used by TransactionTrait for half-message sending. + */ + public function wrapTransactionMessageRequest(array $messages, object $messageQueue): SendMessageRequest + { + $enriched = []; + foreach ($messages as $msg) { + $enriched[] = $this->toProtobufMessage($msg, $messageQueue, true); + } + $request = new SendMessageRequest(); + $request->setMessages($enriched); + return $request; + } + + // ==================== Retry Logic ==================== + + /** + * Send a single message with retry, deadline, and queue rotation. + * + * On each failed attempt, the failed broker endpoint is isolated and the + * next candidate queue is tried. Retry delay follows the configured + * ExponentialBackoffRetryPolicy with jitter. + * + * @param SendMessageRequest $request The gRPC request + * @param Message $message The original user message (for interceptor context) + * @param array $candidates Candidate message queues for rotation + * @param int $maxAttempts Maximum number of attempts + * @return array{messageId: string, transactionId: string, recallHandle: string, code: int, message: string} + * @throws \RuntimeException If deadline exceeded or all attempts fail + */ + public function sendMessageWithRetry( + SendMessageRequest $request, + Message $message, + array $candidates, + int $maxAttempts + ): array { + $lastException = null; + $startTime = microtime(true); + $candidateCount = count($candidates); + $currentMessageQueue = $candidates[0]; + + $operationTimeout = ($this->operationTimeoutFn)('SEND_MESSAGE'); + $deadlineMicroseconds = $startTime + ($operationTimeout / 1000000); + + for ($attempt = 1; $attempt <= $maxAttempts; $attempt++) { + $now = microtime(true); + if ($now >= $deadlineMicroseconds) { + throw new \RuntimeException( + "Send message deadline exceeded after " . + round(($now - $startTime) * 1000, 2) . "ms" + ); + } + + if ($attempt > 1 && $candidateCount > 1) { + $queueIndex = IntMath::mod($attempt, $candidateCount); + $currentMessageQueue = $candidates[$queueIndex]; + $request = $this->wrapSendMessageRequest([$message], $currentMessageQueue); Review Comment: This retry path always rebuilds the request with `wrapSendMessageRequest()`, even when the original request was created by `wrapTransactionMessageRequest()`. After a transient failure, a half message can therefore be retried as a normal, immediately visible message, and the result may no longer contain a valid transaction ID. `TransactionTrait` also records the endpoint of `$messageQueue[0]` instead of the queue that actually succeeded. Please preserve the original message type during retries and return/use the successful queue endpoint for transaction tracking. ########## php/MessageView.php: ########## @@ -0,0 +1,433 @@ +<?php +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +namespace Apache\Rocketmq; + +use Apache\Rocketmq\V2\Message; +use Apache\Rocketmq\V2\Endpoints; +use Apache\Rocketmq\V2\Encoding; + +/** + * MessageView - Rich wrapper for consumed messages. + * + * Wraps the raw protobuf Message received from the broker with additional + * delivery metadata (receipt handle, endpoints, delivery attempt count, + * born timestamp, born host) that is needed for ack/nack/retry operations. + * + * On construction, verifies body integrity (CRC32) and decompresses GZIP if needed. + * Marks the message as corrupted if verification or decompression fails. + */ +class MessageView implements MessageViewInterface +{ + private Message $message; + private ?string $receiptHandle; + private ?Endpoints $endpoints; + private int $deliveryAttempt; + private string $bodyStr = ''; + private bool $corrupted = false; + private int $bornTimestamp = 0; + private string $bornHost = ''; + private int $decodeTimestamp = 0; + + /** + * Construct a MessageView, extracting metadata and processing body. + * + * @param Message $message The protobuf message from broker + * @param string|null $receiptHandle Receipt handle for ack/nack + * @param Endpoints|null $endpoints Broker endpoints + * @param int $deliveryAttempt Number of delivery attempts (starts at 1) + */ + public function __construct(Message $message, ?string $receiptHandle = null, ?Endpoints $endpoints = null, int $deliveryAttempt = 1) + { + $this->message = $message; + if ($receiptHandle === null) { + $sysProps = $message->getSystemProperties(); + $receiptHandle = $sysProps?->getReceiptHandle() ?: null; + } + $this->receiptHandle = $receiptHandle; + $this->endpoints = $endpoints; + $this->deliveryAttempt = max(1, $deliveryAttempt); + $this->decodeTimestamp = time(); + + // Extract born timestamp and host from system properties + $sysProps = $message->getSystemProperties(); + if ($sysProps) { + $bornTs = $sysProps->getBornTimestamp(); + if ($bornTs) { + $this->bornTimestamp = $bornTs->getSeconds() ?? 0; + } + $this->bornHost = $sysProps->getBornHost() ?? ''; + + // Verify body integrity and decompress + $this->bodyStr = $this->processBody($message, $sysProps); + } + + if ($this->bodyStr === null) { + $body = $message->getBody(); + $this->bodyStr = is_string($body) ? $body : (string)$body; + } + } + + /** + * Get the topic resource. + * @return object The topic resource + */ + public function getTopicResource(): object + { + return $this->message->getTopic(); + } + + /** + * Process body: verify integrity and decompress if needed. + * + * @param Message $message The protobuf message + * @param object|null $sysProps System properties from the message + * @return string|null The processed body string, or null if processing was skipped + */ + private function processBody(Message $message, $sysProps): ?string + { + $rawBody = $message->getBody(); + $body = is_string($rawBody) ? $rawBody : (string)$rawBody; + + if ($body === '') { + return ''; + } + + // Step 1: Decompress if encoding is GZIP + $encoding = Encoding::IDENTITY; + if ($sysProps !== null) { + $encoding = $sysProps->getBodyEncoding(); + } + + if ($encoding === Encoding::GZIP) { + $decompressed = @gzdecode($body); + if ($decompressed === false) { + $this->corrupted = true; + Logger::getInstance('MessageView')->warning("Failed to decompress GZIP body for messageId=" . $this->getMessageId()); + return null; + } + $body = $decompressed; + } + + // Step 2: Verify body integrity via digest + if ($sysProps !== null) { + $bodyDigest = $sysProps->getBodyDigest(); + if ($bodyDigest !== null && $bodyDigest !== '') { + // getBodyDigest returns a Digest object, extract type and checksum + $digestType = is_object($bodyDigest) ? $bodyDigest->getType() : null; + $digestChecksum = is_object($bodyDigest) ? $bodyDigest->getChecksum() : (string)$bodyDigest; + if ($digestChecksum !== '' && $digestChecksum !== null) { + $this->verifyBodyDigest($body, $digestChecksum, $digestType); + } + } + } + + return $body; + } + + /** + * Set the receipt handle for ack/nack operations. + * + * @param string|null $receiptHandle The receipt handle + * @return void + */ + public function setReceiptHandle(?string $receiptHandle): void + { + $this->receiptHandle = $receiptHandle; + } + + /** + * Verify body integrity using the given digest type and checksum. + * + * @param string $body The message body string to verify + * @param string $checksum The expected checksum + * @param int|null $digestType The digest type (CRC32, MD5, SHA1) + * @return void + */ + private function verifyBodyDigest(string $body, string $checksum, $digestType): void + { + $computed = ''; + if ($digestType === \Apache\Rocketmq\V2\DigestType::CRC32) { + $computed = sprintf('%u', crc32($body)); Review Comment: `sprintf('%u', crc32($body))` produces a decimal checksum, while RocketMQ CRC32 digests use uppercase hexadecimal; `Utilities::crc32CheckSum()` already implements the expected format. In addition, `processBody()` currently verifies the digest after GZIP decompression, although the digest covers the encoded body bytes. As written, valid messages can be marked corrupted and NACKed. Please verify the raw body before decompression and reuse the existing checksum helper. ########## php/SipHash24.php: ########## @@ -0,0 +1,294 @@ +<?php +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +namespace Apache\Rocketmq; + +/** + * SipHash-2-4 implementation compatible with Guava's Hashing.sipHash24(). + * + * Uses 64-bit arithmetic with platform-aware handling: + * - 64-bit PHP: Native integer operations + * - 32-bit PHP: Split into high/low 32-bit parts + * + * Default key (0, 0) matches Guava's Hashing.sipHash24() behavior. + * + * Note: On 32-bit platforms, hash values may exceed PHP_INT_MAX and be + * represented as floats. Use with caution in array keys or strict comparisons. + */ +class SipHash24 +{ + /** @var int */ + private int $k0; + /** @var int */ + private int $k1; + + // 64-bit mask constant (avoid float conversion) + private const MASK_64 = -1; // All bits set to 1 in two's complement + + /** + * Constructor - initializes SipHash-2-4 with given key. + * + * @param int $k0 Key part 0 (will be masked to 64 bits) + * @param int $k1 Key part 1 (will be masked to 64 bits) + */ + public function __construct(int $k0 = 0, int $k1 = 0) Review Comment: This is not compatible with `Hashing.sipHash24()` in Guava: Guava uses the fixed key bytes `00..0f`, not an all-zero key, and SipHash input words are little-endian while `readLong()` currently places the first input byte in the most-significant position. Consequently, the same message group can map to different queues in PHP and the Java/Node clients, breaking cross-language FIFO ordering. Please use the standard key/byte order and add known SipHash vectors plus cross-client queue-selection tests. ########## php/RpcClientManager.php: ########## @@ -0,0 +1,294 @@ +<?php +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +namespace Apache\Rocketmq; + + +use Apache\Rocketmq\V2\MessagingServiceClient; +use Grpc\ChannelCredentials; + +class RpcClientManager +{ + private static ?self $instance = null; + + private array $clients = []; + private array $mocks = []; + private array $clientLastUsedTime = []; + private int $idleTimeoutSeconds = 1800; // 30 minutes + private int $checkIntervalSeconds = 60; // 1 minute + private int $lastCheckTime = 0; + private Logger $logger; + + /** + * Initialize client manager with logger and timestamp. + */ + private function __construct() + { + $this->logger = Logger::getInstance('RpcClientManager'); + $this->lastCheckTime = time(); + } + + /** + * Get the singleton instance, creating it if necessary. + * + * @return self + */ + public static function getInstance(): self + { + if (self::$instance === null) { + self::$instance = new self(); + } + return self::$instance; + } + + /** + * Reset the singleton instance (primarily for testing). + * + * @return void + */ + public static function reset(): void + { + self::$instance = null; + } + + /** + * Get or create a MessagingServiceClient for the given endpoints. + * + * Clients are cached and reused based on endpoint + TLS configuration. + * Idle clients are automatically cleaned up every 60 seconds if unused for 30 minutes. + * + * @param string $endpoints Server endpoint in format "host:port" + * @param array $options Optional configuration: + * - 'tlsCredentials': TlsCredentials instance for TLS/mTLS + * - 'credentials': Pre-created ChannelCredentials + * @return MessagingServiceClient gRPC client instance + */ + public function getClient(string $endpoints, array $options = []): MessagingServiceClient + { + if (trim($endpoints) === '') { + throw new \InvalidArgumentException('endpoints must not be empty'); + } + + $credentials = $this->resolveCredentials($options); + $key = $this->makeKey($endpoints, $options); + + // Check mock registry first + if (isset($this->mocks[$key])) { + $this->clientLastUsedTime[$key] = time(); + return $this->mocks[$key]; + } + + if (!isset($this->clients[$key])) { + $this->logger->info("Creating new RPC client for: {$endpoints}"); + $opts = ['credentials' => $credentials]; + + // Merge channel args from TlsCredentials (e.g., SSL target name override for dev) + if (isset($options['tlsCredentials']) && $options['tlsCredentials'] instanceof TlsCredentials) { + // Extract host from endpoints for ssl_target_name_override + $targetHost = $endpoints; + if (str_contains($endpoints, ':')) { + $targetHost = explode(":", $endpoints)[0]; + } + $opts = array_merge($opts, $options['tlsCredentials']->getChannelArgs($targetHost)); + } + + $this->clients[$key] = new MessagingServiceClient($endpoints, $opts); + } + + $this->clientLastUsedTime[$key] = time(); + + // Periodically clean up idle connections + $now = time(); + if ($now - $this->lastCheckTime >= $this->checkIntervalSeconds) { + $this->cleanupIdleClients(); + $this->lastCheckTime = $now; + } + + return $this->clients[$key]; + } + + /** + * Register a mock MessagingServiceClient for the given endpoints. + * Subsequent calls to getClient() with matching endpoints will return this mock. + * + * @param string $endpoints Server endpoint in format "host:port" + * @param MessagingServiceClient $mock The mock client to return + * @return void + */ + public function registerMock(string $endpoints, MessagingServiceClient $mock): void + { + $key = $this->makeKey($endpoints, []); + $this->mocks[$key] = $mock; + $this->logger->info("Registered mock client for: {$key}"); + } + + /** + * Remove all registered mocks. + * + * @return void + */ + public function clearMocks(): void + { + $this->mocks = []; + } + + /** + * Release a specific client connection by endpoint prefix. + * + * All clients whose key starts with the given endpoint will be removed. + * + * @param string $endpoints Endpoint prefix to match (e.g., "localhost:8080") + * @return void + */ + public function releaseClient(string $endpoints): void + { + $keysToRemove = []; + foreach ($this->clients as $key => $client) { + if (strpos($key, $endpoints) === 0) { + $keysToRemove[] = $key; + } + } + + foreach ($keysToRemove as $key) { + unset($this->clients[$key]); + unset($this->clientLastUsedTime[$key]); + $this->logger->info("Released RPC client: {$key}"); + } + } + + /** + * Release all client connections and clear the cache. + * + * @return void + */ + public function releaseAll(): void + { + $count = count($this->clients); + $this->clients = []; + $this->clientLastUsedTime = []; + $this->logger->info("Released all {$count} RPC clients"); + } + + /** + * Get the number of active connections in the pool. + * + * @return int Number of cached client connections + */ + public function getConnectionCount(): int + { + return count($this->clients); + } + + /** + * Clean up idle client connections that haven't been used for more than idleTimeoutSeconds. + * + * This method is called automatically every checkIntervalSeconds (60s) when getClient() is invoked. + * + * @return void + */ + private function cleanupIdleClients(): void + { + $now = time(); + $keysToRemove = []; + + foreach ($this->clientLastUsedTime as $key => $lastUsed) { + if ($now - $lastUsed > $this->idleTimeoutSeconds) { + $keysToRemove[] = $key; + } + } + + foreach ($keysToRemove as $key) { + unset($this->clients[$key]); + unset($this->clientLastUsedTime[$key]); + $this->logger->info("Cleaned up idle RPC client: {$key}"); + } + } + + /** + * Generate a unique cache key based on endpoint and TLS configuration. + * + * Key format: "{endpoint}:{tlsFingerprint}" + * Examples: + * - "localhost:8080:insecure" + * - "localhost:8080:tls|ca:/path/to/ca.pem" + * - "localhost:8080:mtls:/path/to/client.pem|no-verify" + * + * @param string $endpoints Server endpoint + * @param array $options Client options containing TLS credentials + * @return string Unique cache key + */ + private function makeKey(string $endpoints, array $options): string + { + $tlsFingerprint = 'insecure'; Review Comment: The cache key does not include `sslEnabled`. When neither `tlsCredentials` nor pre-created `credentials` is provided, both the default TLS configuration and `sslEnabled=false` produce the same `<endpoint>:insecure` key. The channel created first is then reused for the other configuration, which can silently make a TLS client use a plaintext channel. Please include the resolved transport/TLS mode in the cache key. ########## php/PushConsumer.php: ########## @@ -0,0 +1,1343 @@ +<?php +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +namespace Apache\Rocketmq; + + +use Apache\Rocketmq\V2\MessagingServiceClient; +use Apache\Rocketmq\V2\QueryAssignmentRequest; +use Apache\Rocketmq\V2\QueryAssignmentResponse; +use Apache\Rocketmq\V2\QueryRouteRequest; +use Apache\Rocketmq\V2\Resource; +use Apache\Rocketmq\V2\FilterExpression; +use Apache\Rocketmq\V2\MessageQueue; +use Apache\Rocketmq\V2\Settings; +use Apache\Rocketmq\V2\ClientType; +use Apache\Rocketmq\V2\VerifyMessageCommand; +use Apache\Rocketmq\V2\UA; +use Apache\Rocketmq\V2\Language; +use Apache\Rocketmq\V2\TelemetryCommand; +use Apache\Rocketmq\V2\Subscription; +use Apache\Rocketmq\V2\SubscriptionEntry; +use Apache\Rocketmq\V2\Endpoints; +use Apache\Rocketmq\V2\Address; +use Apache\Rocketmq\V2\AddressScheme; +use Apache\Rocketmq\V2\HeartbeatRequest; +use Apache\Rocketmq\V2\NotifyClientTerminationRequest; +use Google\Protobuf\Duration; +use Grpc\ChannelCredentials; + +/** + * PushConsumer - Push-style consumer referencing Java PushConsumerImpl. + * + * Architecture (adapted for PHP single-threaded model): + * - Blocks in start() with a main polling loop + * - Periodically scans assignments via QueryAssignment gRPC + * - Creates/drops ProcessQueue per assigned MessageQueue + * - ProcessQueue caches messages and dispatches to ConsumeService + * - ConsumeService invokes user callback sequentially (no thread pool in PHP) + * + * Configuration options mirror Java PushConsumerBuilderImpl. + */ +class PushConsumer implements ConsumerInterface +{ + use ClientTrait { + buildMetadata as public; + } + + private readonly MessagingServiceClient $client; + protected readonly string $clientId; + protected readonly TelemetrySession $telemetrySession; + private array $subscriptionExpressions = []; + private array $cacheAssignments = []; + private array $processQueueTable = []; + private bool $heartbeatInProgress = false; + protected ?ConsumeService $consumeService = null; + protected bool $isRunning = false; + protected bool $shutdownRequested = false; + protected readonly Logger $logger; + + // Builder options + /** @var callable|null */ + protected mixed $messageListener = null; + private int $maxCacheMessageCount = 4096; + private int $maxCacheMessageSizeInBytes = 67108864; // 64MB + private int $awaitDuration = 5; // seconds + private int $scanIntervalSeconds = 5; + private bool $fifo = false; + private int $receiveBatchSize = 32; + protected bool $enableFifoConsumeAccelerator = false; + private bool $isLiteConsumer = false; + private readonly ?SessionCredentials $credentials; + private readonly string $namespace; + private int $lastHeartbeatTime = 0; + private ?int $shutdownDrainDeadline = null; + private ?ExponentialBackoffRetryPolicy $retryPolicy; + private readonly ?TlsCredentials $tlsCredentials; + private readonly bool $sslEnabled; + private array $interceptors = []; + + /** + * Constructor with builder-style options. + * + * @param string $endpoints gRPC server endpoint + * @param string $consumerGroup Consumer group name + * @param array $options Configuration options + * - clientId: string, custom client identifier (default: 'php-push-consumer-{pid}-{time}') + * - messageListener: callable|null, message consumption callback + * - subscriptionExpressions: array<string,string>, topic subscription map (topic => expression) + * - maxCacheMessageCount: int, max cached messages in memory (default: 4096) + * - maxCacheMessageSizeInBytes: int, max cached message total size (default: 67108864, 64MB) + * - awaitDuration: int, long polling timeout in seconds (default: 5) + * - scanIntervalSeconds: int, assignment scan interval in seconds (default: 5) + * - fifo: bool, enable FIFO message consumption mode (default: false) + * - receiveBatchSize: int, max messages per receive batch (default: 32) + * - enableFifoConsumeAccelerator: bool, enable FIFO consume accelerator (default: false) + * - isLiteConsumer: bool, enable lite consumer mode (default: false) + * - credentials: SessionCredentials|null, AK/SK authentication credentials + * - namespace: string, resource namespace prefix (default: '') + * - tlsCredentials: TlsCredentials|null, TLS/SSL configuration + * - sslEnabled: bool, enable SSL for gRPC channel (default: true) + */ + public function __construct( + protected readonly string $endpoints, + protected readonly string $consumerGroup, + array $options = [] + ) { + if (empty($consumerGroup)) { + throw new \InvalidArgumentException("PushConsumer consumerGroup cannot be empty"); + } + $this->clientId = $options['clientId'] ?? ('php-push-consumer-' . getmypid() . '-' . time()); + $this->messageListener = $options['messageListener'] ?? null; + $this->subscriptionExpressions = $options['subscriptionExpressions'] ?? []; + $this->maxCacheMessageCount = $options['maxCacheMessageCount'] ?? 4096; + $this->maxCacheMessageSizeInBytes = $options['maxCacheMessageSizeInBytes'] ?? 67108864; + $this->awaitDuration = $options['awaitDuration'] ?? 5; + $this->scanIntervalSeconds = $options['scanIntervalSeconds'] ?? 5; + $this->fifo = $options['fifo'] ?? false; + $this->receiveBatchSize = $options['receiveBatchSize'] ?? 32; + $this->enableFifoConsumeAccelerator = $options['enableFifoConsumeAccelerator'] ?? false; + $this->isLiteConsumer = $options['isLiteConsumer'] ?? false; + $this->namespace = $options['namespace'] ?? ''; + $this->tlsCredentials = $options['tlsCredentials'] ?? null; + $this->sslEnabled = $options['sslEnabled'] ?? true; + + // Set AK/SK credentials if provided + $this->credentials = (isset($options['credentials']) && $options['credentials'] instanceof SessionCredentials) + ? $options['credentials'] + : null; + + $this->logger = Logger::getInstance('PushConsumer'); + + // Use RpcClientManager for connection pooling + $this->client = RpcClientManager::getInstance()->getClient($endpoints, [ + 'tlsCredentials' => $this->tlsCredentials, + 'sslEnabled' => $options['sslEnabled'] ?? true, + ]); + + $this->telemetrySession = TelemetrySession::getInstance($this->client, $endpoints, $this->clientId, $this->credentials, $this->namespace); + $this->retryPolicy = new ExponentialBackoffRetryPolicy(5, 1000, 30000, 2.0); + } + + /** + * Check if this is a FIFO consumer. + * @return bool + */ + public function fifo(): bool + { + return $this->fifo; + } + + /** + * Subscribe to a topic. + * + * @param string $topic Topic name + * @param string $expression Filter expression (default "*") + * @return $this + */ + public function subscribe(string $topic, string $expression = '*'): self + { + if ($this->isRunning) { + // Dynamic runtime subscription: update subscription expressions + $this->subscriptionExpressions[$topic] = $expression; + $this->logger->info("Dynamically subscribed to topic: {$topic}"); + return $this; + } + $this->subscriptionExpressions[$topic] = $expression; + return $this; + } + + /** + * Get the retry policy. + * + * @return ExponentialBackoffRetryPolicy|null + */ + public function getRetryPolicy(): ?ExponentialBackoffRetryPolicy + { + return $this->retryPolicy; + } + + /** + * Unsubscribe from a topic. + * + * @param string $topic Topic name + * @return $this + */ + public function unsubscribe(string $topic): self + { + if ($this->isRunning) { + // Dynamic runtime unsubscription + unset($this->subscriptionExpressions[$topic]); + unset($this->cacheAssignments[$topic]); + // Drop related ProcessQueues + $processQueue = $this->processQueueTable; + foreach ($processQueue as $key => $pq) { + $mq = $pq->getMessageQueue(); + if ($mq->getTopic()->getName() === $topic) { + $pq->drop(); + unset($this->processQueueTable[$key]); + } + } + $this->logger->info("Dynamically unsubscribed from topic: {$topic}"); + return $this; + } + unset($this->subscriptionExpressions[$topic]); + unset($this->cacheAssignments[$topic]); + return $this; + } + + /** + * Set the message listener callback. + * + * @param callable $listener function($messageView): int + * @return $this + */ + public function setMessageListener(callable $listener): self + { + $this->checkNotRunning(); + $this->messageListener = $listener; + return $this; + } + + /** + * Start the PushConsumer. Blocks in the main polling loop. + * + * @throws \RuntimeException If messageListener or subscriptions are not set + */ + public function start(): void + { + if ($this->isRunning) { + return; + } + + if ($this->messageListener === null) { + throw new \RuntimeException("PushConsumer messageListener is not set"); + } + + if (empty($this->subscriptionExpressions)) { + throw new \RuntimeException("PushConsumer has no subscriptions"); + } + + $this->logger->info("PushConsumer starting, clientId={$this->clientId}"); + + try { + $this->establishTelemetrySession(); + + // Register settings change callback + $this->registerSettingsCallback(); + + $this->onStartBeforeLoop(); + + // Create consume service (Standard, FIFO, or LiteFIFO) + if ($this->isLiteConsumer) { + $this->consumeService = new LiteFifoConsumeService($this->logger, $this->messageListener, $this, $this->enableFifoConsumeAccelerator); + } elseif ($this->fifo) { + $this->consumeService = new FifoConsumeService($this->logger, $this->messageListener, $this, $this->enableFifoConsumeAccelerator); + } else { + $this->consumeService = new StandardConsumeService($this->logger, $this->messageListener, $this); + } + + $this->registerSignalHandlers(); + $this->isRunning = true; + + $this->logger->info("PushConsumer started successfully, clientId={$this->clientId}"); + + // Initial assignment scan + $this->scanAssignments(); + + // Main polling loop + $lastScanTime = time(); + + while ($this->isRunning && !$this->shutdownRequested) { + // Dispatch pending signals + if (function_exists('pcntl_signal_dispatch')) { + pcntl_signal_dispatch(); + } + + if ($this->telemetrySession) { + $this->telemetrySession->pollTelemetry(); + } + if ($this->shutdownRequested) { + break; + } + + $now = time(); + if ($now - $lastScanTime >= $this->scanIntervalSeconds) { + $this->scanAssignments(); + $this->onScanCycleComplete(); + $lastScanTime = $now; + } + + // Periodic heartbeat + $this->onHeartbeatTick(); + + // Fetch messages from each active ProcessQueue + $this->fetchMessageInterleavedHeartbeat(); + // Short sleep between iterations + SwooleCompat::sleep(100000); + + // Periodic garbage collection + gc_collect_cycles(); + } + + // Graceful shutdown drain phase + $this->drainInFlightMessages(); + + $this->shutdown(); + + } catch (\Exception $e) { + $this->logger->error("PushConsumer start failed: " . $e->getMessage()); + $this->onStop(); + throw $e; + } + } + + /** + * Start the PushConsumer with a timeout. Blocks for at most the given seconds. + * + * @param int $seconds Maximum duration in seconds + * @return void + * @throws \RuntimeException If messageListener or subscriptions are not set + */ + public function startWithTimeout(int $seconds): void + { + if ($this->isRunning()) { + return; + } + + if ($this->messageListener === null) { + throw new \RuntimeException("PushConsumer messageListener is not set"); + } + if (empty($this->subscriptionExpressions)) { + throw new \RuntimeException("PushConsumer has no subscriptions"); + } + $this->logger->info("PushConsumer starting with timeout {$seconds} seconds, clientId={$this->clientId}"); + try { + $this->establishTelemetrySession(); + $this->registerSettingsCallback(); + $this->onStartBeforeLoop(); + if ($this->isLiteConsumer) { + $this->consumeService = new LiteFifoConsumeService($this->logger, $this->messageListener, $this, $this->enableFifoConsumeAccelerator); + } elseif ($this->fifo) { + $this->consumeService = new FifoConsumeService($this->logger, $this->messageListener, $this, $this->enableFifoConsumeAccelerator); + } else { + $this->consumeService = new StandardConsumeService($this->logger, $this->messageListener, $this); + } + $this->registerSignalHandlers(); + $this->isRunning = true; + $this->logger->info("PushConsumer running with timeout {$seconds} seconds, clientId={$this->clientId}"); + $this->scanAssignments(); + $deadline = time() + $seconds; + $lastScanTime = time(); + while ($this->isRunning && !$this->shutdownRequested && time() < $deadline) { + if (function_exists('pcntl_signal_dispatch')) { + pcntl_signal_dispatch(); + } + if ($this->telemetrySession) { + $this->telemetrySession->pollTelemetry(); + } + if ($this->shutdownRequested) { + break; + } + $now = time(); + if ($now - $lastScanTime >= $this->scanIntervalSeconds) { + $this->scanAssignments(); + $this->onScanCycleComplete(); + $lastScanTime = $now; + } + + $this->onHeartbeatTick(); + + $this->fetchMessageInterleavedHeartbeat(); + SwooleCompat::sleep(100000); + gc_collect_cycles(); + } + + $this->logger->info("PushConsumer startWithTimeout completed after {$seconds}s, clientId={$this->clientId}"); + $this->onStop(); + } catch (\Exception $e) { + $this->logger->error("PushConsumer startWithTimeout failed: " . $e->getMessage()); + $this->onStop(); + throw $e; + } + } + + /** + * Get the consume service instance. + * + * @return ConsumeService|null + */ + public function getConsumeService(): ?ConsumeService + { + return $this->consumeService; + } + + /** + * Hook called before the main polling loop starts. Override in subclasses. + * + * @return void + */ + protected function onStartBeforeLoop(): void + { + + } + + /** + * Hook called when the consumer stops. Override in subclasses. + * + * @return void + */ + protected function onStop(): void + { + + } + + /** + * Create a heartbeat request with client type and group. + * + * @return HeartbeatRequest + */ + private function wrapHeartbeatRequest(): \Apache\Rocketmq\V2\HeartbeatRequest + { + $request = new HeartbeatRequest(); + $request->setClientType($this->getClientType()); + $request->setGroup($this->getGroupResource()); + return $request; + } + + /** + * Fetch messages from each active ProcessQueue, interleaved with heartbeat ticks. + * + * @return void + */ + private function fetchMessageInterleavedHeartbeat(): void + { + $processQueues = $this->processQueueTable; + foreach ($processQueues as $key => $pq) { + if ($pq->isDropped() || $pq->expired()) { + $pq->drop(); + unset($this->processQueueTable[$key]); + continue; + } + if (!$pq->isCacheFull()) { + $this->onHeartbeatTick(); + $pq->fetchMessages(); + } + } + } + + /** + * Drain in-flight messages before shutdown. Waits up to 30s for cached messages + * to be consumed, preventing message loss on abrupt termination. + */ + private function drainInFlightMessages(): void + { + $drainStart = microtime(true); + $drainTimeout = 30; // seconds + $drainIterations = 0; + + $this->logger->info("PushConsumer drain phase: waiting for in-flight messages to be consumed"); + + // Step 1: Mark all queues as dropped to stop fetching NEW messages + // This prevents new messages from being added to the cache + $processQueues = $this->processQueueTable; + foreach ($processQueues as $pq) { + $pq->drop(); + } + + // Step 2: Wait for cached messages to be consumed + // Note: ConsumeService checks isDropped() at the START of consume(), + // but since we call it AFTER drop(), it will skip consumption. + // Solution: Manually iterate and consume cached messages here, + // bypassing the isDropped() check in ConsumeService. + while (microtime(true) - $drainStart < $drainTimeout) { + $remainingCount = 0; + $activeQueues = 0; + + foreach ($this->processQueueTable as $pq) { + $messages = $pq->getCachedMessages(); + $cachedCount = count($messages); + $remainingCount += $cachedCount; + + if ($cachedCount > 0 && $this->consumeService !== null) { + $activeQueues++; + + // Manually consume each cached message, bypassing isDropped() check + // We copy the list first to avoid modification during iteration + $toConsume = array_values($messages); + foreach ($toConsume as $messageView) { + // Skip if already evicted + if (!in_array($messageView, $pq->getCachedMessages(), true)) { + continue; + } + + try { + // Call the message listener directly + $result = $this->consumeService->consumeMessage($messageView); + + // Handle result : SUCCESS, SUSPEND, or FAILURE + if ($result === \Apache\Rocketmq\ConsumeResult::SUCCESS) { + $this->consumeService->ackMessage($messageView); + $pq->evictMessage($messageView); + } elseif ($result instanceof \Apache\Rocketmq\ConsumeResultSuspend) { + // Respect the suspend time during drain + $suspendSec = (int)ceil($result->getSuspendTimeMs() / 1000); + $this->consumeService->nackMessage($messageView, 1, $suspendSec); + $pq->evictMessage($messageView); + } else { + $this->consumeService->nackMessage($messageView); + $pq->evictMessage($messageView); + } + + // Evict from cache + $pq->evictMessage($messageView); + } catch (\Exception $e) { + $this->logger->error("Drain phase consume error: " . $e->getMessage()); + // On error, nack and evict + try { + $this->consumeService->nackMessage($messageView); + } catch (\Exception $ackError) { + $this->logger->warning("Failed to nack message during drain: " . $ackError->getMessage()); + } + $pq->evictMessage($messageView); + } + } + } + } + + if ($remainingCount === 0) { + $this->logger->info("PushConsumer drain phase completed after {$drainIterations} iterations, all messages consumed"); + return; + } + + // Log progress every 50 iterations (~5 seconds) + if ($drainIterations % 50 === 0 && $drainIterations > 0) { + $elapsed = round(microtime(true) - $drainStart, 1); + $this->logger->info("PushConsumer drain progress: {$remainingCount} messages remaining in {$activeQueues} queues after {$elapsed}s"); + } + + SwooleCompat::sleep(100000); // 100ms + $drainIterations++; + } + + $remainingCount = 0; + $processQueues = $this->processQueueTable; + foreach ($processQueues as $pq) { + $remainingCount += count($pq->getCachedMessages()); + } + $this->logger->warning("PushConsumer drain phase timed out after {$drainTimeout}s, {$remainingCount} messages remaining in " . count($this->processQueueTable) . " queues"); + } + + /** + * Request graceful shutdown. + */ + public function shutdown(): void + { + if (!$this->isRunning) { + return; + } + + $this->logger->info("PushConsumer shutting down, clientId={$this->clientId}"); + + $this->isRunning = false; + + // Notify server of client termination + $this->notifyClientTermination(); + + // Drop all ProcessQueues + foreach ($this->processQueueTable as $pq) { + $pq->drop(); + } + $this->processQueueTable = []; + + // Close telemetry session + if ($this->telemetrySession) { + $this->telemetrySession->close(); + } + + $this->logger->info("PushConsumer shutdown complete, clientId={$this->clientId}"); + } + + /** + * Signal handler for graceful shutdown. + */ + public function requestShutdown(): void + { + $this->shutdownRequested = true; + } + + /** + * Register a message interceptor. + * + * @param MessageInterceptor $interceptor + * @return $this + */ + public function addInterceptor(MessageInterceptor $interceptor): self + { + $this->interceptors[] = $interceptor; + return $this; + } + + /** + * Execute interceptors at a given hook point. + * + * @param string $hookPoint The hook point identifier + * @param array $context Additional context for the interceptor + * @return void + */ + public function executeInterceptors(string $hookPoint, array $context = []): void + { + if (empty($this->interceptors)) { + return; + } + foreach ($this->interceptors as $interceptor) { + try { + $interceptor->intercept($hookPoint, $context); + } catch (\Exception $e) { + $this->logger->warning("Interceptor failed at {$hookPoint}: " . $e->getMessage()); + } + } + } + + /** + * Get the client type identifier. + * + * @return int The PUSH_CONSUMER client type + */ + protected function getClientType(): int + { + return ClientType::PUSH_CONSUMER; + } + + /** + * Register SIGTERM/SIGINT signal handlers. + */ + protected function registerSignalHandlers(): void + { + if (function_exists('pcntl_signal')) { + $self = $this; + pcntl_signal(SIGTERM, function() use ($self) { + $self->requestShutdown(); + }); + pcntl_signal(SIGINT, function() use ($self) { + $self->requestShutdown(); + }); + $this->logger->info("PushConsumer signal handlers registered"); + } + } + + /** + * Establish Telemetry Session with the server for this consumer group. + * + * @return void + * @throws \RuntimeException If session establishment fails + */ + protected function establishTelemetrySession(): void + { + $ua = new UA(); + $ua->setLanguage(Language::PHP); + $ua->setVersion(ClientConstants::CLIENT_VERSION); + + $subscriptionEntries = []; + foreach ($this->subscriptionExpressions as $topic => $expression) { + $filterExpression = new FilterExpression(); + $filterExpression->setExpression($expression); + + $topicResource = new Resource(); + $topicResource->setName($topic); + + $subscriptionEntry = new SubscriptionEntry(); + $subscriptionEntry->setTopic($topicResource); + $subscriptionEntry->setExpression($filterExpression); + + $subscriptionEntries[] = $subscriptionEntry; + } + + $subscription = new Subscription(); + $groupResource = new Resource(); + $groupResource->setName($this->consumerGroup); + $subscription->setGroup($groupResource); + $subscription->setSubscriptions($subscriptionEntries); + + $settings = new Settings(); + $settings->setClientType($this->getClientType()); + $settings->setUserAgent($ua); + $settings->setSubscription($subscription); + + $settings->setAccessPoint($this->parseEndpoints($this->endpoints)); + $timeoutDuration = new Duration(); + $timeoutDuration->setSeconds(3); + $timeoutDuration->setNanos(0); + $settings->setRequestTimeout($timeoutDuration); + + $command = new TelemetryCommand(); + $command->setSettings($settings); + + $success = $this->telemetrySession->createStreamAndSync($command); Review Comment: `createStreamAndSync()` only creates the telemetry stream and writes the Settings command; it does not wait for the server Settings response. Therefore the earlier `syncSettings()` timeout/error fix still does not protect PushConsumer startup, and consumption can begin without server-accepted settings or backoff values. Please use `syncSettings()` here, or otherwise wait for and validate the server response before marking startup successful. ########## php/PushConsumer.php: ########## @@ -0,0 +1,1343 @@ +<?php +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +namespace Apache\Rocketmq; + + +use Apache\Rocketmq\V2\MessagingServiceClient; +use Apache\Rocketmq\V2\QueryAssignmentRequest; +use Apache\Rocketmq\V2\QueryAssignmentResponse; +use Apache\Rocketmq\V2\QueryRouteRequest; +use Apache\Rocketmq\V2\Resource; +use Apache\Rocketmq\V2\FilterExpression; +use Apache\Rocketmq\V2\MessageQueue; +use Apache\Rocketmq\V2\Settings; +use Apache\Rocketmq\V2\ClientType; +use Apache\Rocketmq\V2\VerifyMessageCommand; +use Apache\Rocketmq\V2\UA; +use Apache\Rocketmq\V2\Language; +use Apache\Rocketmq\V2\TelemetryCommand; +use Apache\Rocketmq\V2\Subscription; +use Apache\Rocketmq\V2\SubscriptionEntry; +use Apache\Rocketmq\V2\Endpoints; +use Apache\Rocketmq\V2\Address; +use Apache\Rocketmq\V2\AddressScheme; +use Apache\Rocketmq\V2\HeartbeatRequest; +use Apache\Rocketmq\V2\NotifyClientTerminationRequest; +use Google\Protobuf\Duration; +use Grpc\ChannelCredentials; + +/** + * PushConsumer - Push-style consumer referencing Java PushConsumerImpl. + * + * Architecture (adapted for PHP single-threaded model): + * - Blocks in start() with a main polling loop + * - Periodically scans assignments via QueryAssignment gRPC + * - Creates/drops ProcessQueue per assigned MessageQueue + * - ProcessQueue caches messages and dispatches to ConsumeService + * - ConsumeService invokes user callback sequentially (no thread pool in PHP) + * + * Configuration options mirror Java PushConsumerBuilderImpl. + */ +class PushConsumer implements ConsumerInterface +{ + use ClientTrait { + buildMetadata as public; + } + + private readonly MessagingServiceClient $client; + protected readonly string $clientId; + protected readonly TelemetrySession $telemetrySession; + private array $subscriptionExpressions = []; + private array $cacheAssignments = []; + private array $processQueueTable = []; + private bool $heartbeatInProgress = false; + protected ?ConsumeService $consumeService = null; + protected bool $isRunning = false; + protected bool $shutdownRequested = false; + protected readonly Logger $logger; + + // Builder options + /** @var callable|null */ + protected mixed $messageListener = null; + private int $maxCacheMessageCount = 4096; + private int $maxCacheMessageSizeInBytes = 67108864; // 64MB + private int $awaitDuration = 5; // seconds + private int $scanIntervalSeconds = 5; + private bool $fifo = false; + private int $receiveBatchSize = 32; + protected bool $enableFifoConsumeAccelerator = false; + private bool $isLiteConsumer = false; + private readonly ?SessionCredentials $credentials; + private readonly string $namespace; + private int $lastHeartbeatTime = 0; + private ?int $shutdownDrainDeadline = null; + private ?ExponentialBackoffRetryPolicy $retryPolicy; + private readonly ?TlsCredentials $tlsCredentials; + private readonly bool $sslEnabled; + private array $interceptors = []; + + /** + * Constructor with builder-style options. + * + * @param string $endpoints gRPC server endpoint + * @param string $consumerGroup Consumer group name + * @param array $options Configuration options + * - clientId: string, custom client identifier (default: 'php-push-consumer-{pid}-{time}') + * - messageListener: callable|null, message consumption callback + * - subscriptionExpressions: array<string,string>, topic subscription map (topic => expression) + * - maxCacheMessageCount: int, max cached messages in memory (default: 4096) + * - maxCacheMessageSizeInBytes: int, max cached message total size (default: 67108864, 64MB) + * - awaitDuration: int, long polling timeout in seconds (default: 5) + * - scanIntervalSeconds: int, assignment scan interval in seconds (default: 5) + * - fifo: bool, enable FIFO message consumption mode (default: false) + * - receiveBatchSize: int, max messages per receive batch (default: 32) + * - enableFifoConsumeAccelerator: bool, enable FIFO consume accelerator (default: false) + * - isLiteConsumer: bool, enable lite consumer mode (default: false) + * - credentials: SessionCredentials|null, AK/SK authentication credentials + * - namespace: string, resource namespace prefix (default: '') + * - tlsCredentials: TlsCredentials|null, TLS/SSL configuration + * - sslEnabled: bool, enable SSL for gRPC channel (default: true) + */ + public function __construct( + protected readonly string $endpoints, + protected readonly string $consumerGroup, + array $options = [] + ) { + if (empty($consumerGroup)) { + throw new \InvalidArgumentException("PushConsumer consumerGroup cannot be empty"); + } + $this->clientId = $options['clientId'] ?? ('php-push-consumer-' . getmypid() . '-' . time()); + $this->messageListener = $options['messageListener'] ?? null; + $this->subscriptionExpressions = $options['subscriptionExpressions'] ?? []; + $this->maxCacheMessageCount = $options['maxCacheMessageCount'] ?? 4096; + $this->maxCacheMessageSizeInBytes = $options['maxCacheMessageSizeInBytes'] ?? 67108864; + $this->awaitDuration = $options['awaitDuration'] ?? 5; + $this->scanIntervalSeconds = $options['scanIntervalSeconds'] ?? 5; + $this->fifo = $options['fifo'] ?? false; + $this->receiveBatchSize = $options['receiveBatchSize'] ?? 32; + $this->enableFifoConsumeAccelerator = $options['enableFifoConsumeAccelerator'] ?? false; + $this->isLiteConsumer = $options['isLiteConsumer'] ?? false; + $this->namespace = $options['namespace'] ?? ''; + $this->tlsCredentials = $options['tlsCredentials'] ?? null; + $this->sslEnabled = $options['sslEnabled'] ?? true; + + // Set AK/SK credentials if provided + $this->credentials = (isset($options['credentials']) && $options['credentials'] instanceof SessionCredentials) + ? $options['credentials'] + : null; + + $this->logger = Logger::getInstance('PushConsumer'); + + // Use RpcClientManager for connection pooling + $this->client = RpcClientManager::getInstance()->getClient($endpoints, [ + 'tlsCredentials' => $this->tlsCredentials, + 'sslEnabled' => $options['sslEnabled'] ?? true, + ]); + + $this->telemetrySession = TelemetrySession::getInstance($this->client, $endpoints, $this->clientId, $this->credentials, $this->namespace); + $this->retryPolicy = new ExponentialBackoffRetryPolicy(5, 1000, 30000, 2.0); + } + + /** + * Check if this is a FIFO consumer. + * @return bool + */ + public function fifo(): bool + { + return $this->fifo; + } + + /** + * Subscribe to a topic. + * + * @param string $topic Topic name + * @param string $expression Filter expression (default "*") + * @return $this + */ + public function subscribe(string $topic, string $expression = '*'): self + { + if ($this->isRunning) { + // Dynamic runtime subscription: update subscription expressions + $this->subscriptionExpressions[$topic] = $expression; + $this->logger->info("Dynamically subscribed to topic: {$topic}"); + return $this; + } + $this->subscriptionExpressions[$topic] = $expression; + return $this; + } + + /** + * Get the retry policy. + * + * @return ExponentialBackoffRetryPolicy|null + */ + public function getRetryPolicy(): ?ExponentialBackoffRetryPolicy + { + return $this->retryPolicy; + } + + /** + * Unsubscribe from a topic. + * + * @param string $topic Topic name + * @return $this + */ + public function unsubscribe(string $topic): self + { + if ($this->isRunning) { + // Dynamic runtime unsubscription + unset($this->subscriptionExpressions[$topic]); + unset($this->cacheAssignments[$topic]); + // Drop related ProcessQueues + $processQueue = $this->processQueueTable; + foreach ($processQueue as $key => $pq) { + $mq = $pq->getMessageQueue(); + if ($mq->getTopic()->getName() === $topic) { + $pq->drop(); + unset($this->processQueueTable[$key]); + } + } + $this->logger->info("Dynamically unsubscribed from topic: {$topic}"); + return $this; + } + unset($this->subscriptionExpressions[$topic]); + unset($this->cacheAssignments[$topic]); + return $this; + } + + /** + * Set the message listener callback. + * + * @param callable $listener function($messageView): int + * @return $this + */ + public function setMessageListener(callable $listener): self + { + $this->checkNotRunning(); + $this->messageListener = $listener; + return $this; + } + + /** + * Start the PushConsumer. Blocks in the main polling loop. + * + * @throws \RuntimeException If messageListener or subscriptions are not set + */ + public function start(): void + { + if ($this->isRunning) { + return; + } + + if ($this->messageListener === null) { + throw new \RuntimeException("PushConsumer messageListener is not set"); + } + + if (empty($this->subscriptionExpressions)) { + throw new \RuntimeException("PushConsumer has no subscriptions"); + } + + $this->logger->info("PushConsumer starting, clientId={$this->clientId}"); + + try { + $this->establishTelemetrySession(); + + // Register settings change callback + $this->registerSettingsCallback(); + + $this->onStartBeforeLoop(); + + // Create consume service (Standard, FIFO, or LiteFIFO) + if ($this->isLiteConsumer) { + $this->consumeService = new LiteFifoConsumeService($this->logger, $this->messageListener, $this, $this->enableFifoConsumeAccelerator); + } elseif ($this->fifo) { + $this->consumeService = new FifoConsumeService($this->logger, $this->messageListener, $this, $this->enableFifoConsumeAccelerator); + } else { + $this->consumeService = new StandardConsumeService($this->logger, $this->messageListener, $this); + } + + $this->registerSignalHandlers(); + $this->isRunning = true; + + $this->logger->info("PushConsumer started successfully, clientId={$this->clientId}"); + + // Initial assignment scan + $this->scanAssignments(); + + // Main polling loop + $lastScanTime = time(); + + while ($this->isRunning && !$this->shutdownRequested) { + // Dispatch pending signals + if (function_exists('pcntl_signal_dispatch')) { + pcntl_signal_dispatch(); + } + + if ($this->telemetrySession) { + $this->telemetrySession->pollTelemetry(); + } + if ($this->shutdownRequested) { + break; + } + + $now = time(); + if ($now - $lastScanTime >= $this->scanIntervalSeconds) { + $this->scanAssignments(); + $this->onScanCycleComplete(); + $lastScanTime = $now; + } + + // Periodic heartbeat + $this->onHeartbeatTick(); + + // Fetch messages from each active ProcessQueue + $this->fetchMessageInterleavedHeartbeat(); + // Short sleep between iterations + SwooleCompat::sleep(100000); + + // Periodic garbage collection + gc_collect_cycles(); + } + + // Graceful shutdown drain phase + $this->drainInFlightMessages(); + + $this->shutdown(); + + } catch (\Exception $e) { + $this->logger->error("PushConsumer start failed: " . $e->getMessage()); + $this->onStop(); + throw $e; + } + } + + /** + * Start the PushConsumer with a timeout. Blocks for at most the given seconds. + * + * @param int $seconds Maximum duration in seconds + * @return void + * @throws \RuntimeException If messageListener or subscriptions are not set + */ + public function startWithTimeout(int $seconds): void + { + if ($this->isRunning()) { + return; + } + + if ($this->messageListener === null) { + throw new \RuntimeException("PushConsumer messageListener is not set"); + } + if (empty($this->subscriptionExpressions)) { + throw new \RuntimeException("PushConsumer has no subscriptions"); + } + $this->logger->info("PushConsumer starting with timeout {$seconds} seconds, clientId={$this->clientId}"); + try { + $this->establishTelemetrySession(); + $this->registerSettingsCallback(); + $this->onStartBeforeLoop(); + if ($this->isLiteConsumer) { + $this->consumeService = new LiteFifoConsumeService($this->logger, $this->messageListener, $this, $this->enableFifoConsumeAccelerator); + } elseif ($this->fifo) { + $this->consumeService = new FifoConsumeService($this->logger, $this->messageListener, $this, $this->enableFifoConsumeAccelerator); + } else { + $this->consumeService = new StandardConsumeService($this->logger, $this->messageListener, $this); + } + $this->registerSignalHandlers(); + $this->isRunning = true; + $this->logger->info("PushConsumer running with timeout {$seconds} seconds, clientId={$this->clientId}"); + $this->scanAssignments(); + $deadline = time() + $seconds; + $lastScanTime = time(); + while ($this->isRunning && !$this->shutdownRequested && time() < $deadline) { + if (function_exists('pcntl_signal_dispatch')) { + pcntl_signal_dispatch(); + } + if ($this->telemetrySession) { + $this->telemetrySession->pollTelemetry(); + } + if ($this->shutdownRequested) { + break; + } + $now = time(); + if ($now - $lastScanTime >= $this->scanIntervalSeconds) { + $this->scanAssignments(); + $this->onScanCycleComplete(); + $lastScanTime = $now; + } + + $this->onHeartbeatTick(); + + $this->fetchMessageInterleavedHeartbeat(); + SwooleCompat::sleep(100000); + gc_collect_cycles(); + } + + $this->logger->info("PushConsumer startWithTimeout completed after {$seconds}s, clientId={$this->clientId}"); + $this->onStop(); + } catch (\Exception $e) { + $this->logger->error("PushConsumer startWithTimeout failed: " . $e->getMessage()); + $this->onStop(); + throw $e; + } + } + + /** + * Get the consume service instance. + * + * @return ConsumeService|null + */ + public function getConsumeService(): ?ConsumeService + { + return $this->consumeService; + } + + /** + * Hook called before the main polling loop starts. Override in subclasses. + * + * @return void + */ + protected function onStartBeforeLoop(): void + { + + } + + /** + * Hook called when the consumer stops. Override in subclasses. + * + * @return void + */ + protected function onStop(): void + { + + } + + /** + * Create a heartbeat request with client type and group. + * + * @return HeartbeatRequest + */ + private function wrapHeartbeatRequest(): \Apache\Rocketmq\V2\HeartbeatRequest + { + $request = new HeartbeatRequest(); + $request->setClientType($this->getClientType()); + $request->setGroup($this->getGroupResource()); + return $request; + } + + /** + * Fetch messages from each active ProcessQueue, interleaved with heartbeat ticks. + * + * @return void + */ + private function fetchMessageInterleavedHeartbeat(): void + { + $processQueues = $this->processQueueTable; + foreach ($processQueues as $key => $pq) { + if ($pq->isDropped() || $pq->expired()) { + $pq->drop(); + unset($this->processQueueTable[$key]); + continue; + } + if (!$pq->isCacheFull()) { + $this->onHeartbeatTick(); + $pq->fetchMessages(); + } + } + } + + /** + * Drain in-flight messages before shutdown. Waits up to 30s for cached messages + * to be consumed, preventing message loss on abrupt termination. + */ + private function drainInFlightMessages(): void + { + $drainStart = microtime(true); + $drainTimeout = 30; // seconds + $drainIterations = 0; + + $this->logger->info("PushConsumer drain phase: waiting for in-flight messages to be consumed"); + + // Step 1: Mark all queues as dropped to stop fetching NEW messages + // This prevents new messages from being added to the cache + $processQueues = $this->processQueueTable; + foreach ($processQueues as $pq) { + $pq->drop(); + } + + // Step 2: Wait for cached messages to be consumed + // Note: ConsumeService checks isDropped() at the START of consume(), + // but since we call it AFTER drop(), it will skip consumption. + // Solution: Manually iterate and consume cached messages here, + // bypassing the isDropped() check in ConsumeService. + while (microtime(true) - $drainStart < $drainTimeout) { + $remainingCount = 0; + $activeQueues = 0; + + foreach ($this->processQueueTable as $pq) { + $messages = $pq->getCachedMessages(); + $cachedCount = count($messages); + $remainingCount += $cachedCount; + + if ($cachedCount > 0 && $this->consumeService !== null) { + $activeQueues++; + + // Manually consume each cached message, bypassing isDropped() check + // We copy the list first to avoid modification during iteration + $toConsume = array_values($messages); + foreach ($toConsume as $messageView) { + // Skip if already evicted + if (!in_array($messageView, $pq->getCachedMessages(), true)) { + continue; + } + + try { + // Call the message listener directly + $result = $this->consumeService->consumeMessage($messageView); + + // Handle result : SUCCESS, SUSPEND, or FAILURE + if ($result === \Apache\Rocketmq\ConsumeResult::SUCCESS) { + $this->consumeService->ackMessage($messageView); + $pq->evictMessage($messageView); + } elseif ($result instanceof \Apache\Rocketmq\ConsumeResultSuspend) { + // Respect the suspend time during drain + $suspendSec = (int)ceil($result->getSuspendTimeMs() / 1000); + $this->consumeService->nackMessage($messageView, 1, $suspendSec); + $pq->evictMessage($messageView); + } else { + $this->consumeService->nackMessage($messageView); + $pq->evictMessage($messageView); + } + + // Evict from cache + $pq->evictMessage($messageView); + } catch (\Exception $e) { + $this->logger->error("Drain phase consume error: " . $e->getMessage()); + // On error, nack and evict + try { + $this->consumeService->nackMessage($messageView); + } catch (\Exception $ackError) { + $this->logger->warning("Failed to nack message during drain: " . $ackError->getMessage()); + } + $pq->evictMessage($messageView); + } + } + } + } + + if ($remainingCount === 0) { + $this->logger->info("PushConsumer drain phase completed after {$drainIterations} iterations, all messages consumed"); + return; + } + + // Log progress every 50 iterations (~5 seconds) + if ($drainIterations % 50 === 0 && $drainIterations > 0) { + $elapsed = round(microtime(true) - $drainStart, 1); + $this->logger->info("PushConsumer drain progress: {$remainingCount} messages remaining in {$activeQueues} queues after {$elapsed}s"); + } + + SwooleCompat::sleep(100000); // 100ms + $drainIterations++; + } + + $remainingCount = 0; + $processQueues = $this->processQueueTable; + foreach ($processQueues as $pq) { + $remainingCount += count($pq->getCachedMessages()); + } + $this->logger->warning("PushConsumer drain phase timed out after {$drainTimeout}s, {$remainingCount} messages remaining in " . count($this->processQueueTable) . " queues"); + } + + /** + * Request graceful shutdown. + */ + public function shutdown(): void + { + if (!$this->isRunning) { + return; + } + + $this->logger->info("PushConsumer shutting down, clientId={$this->clientId}"); + + $this->isRunning = false; + + // Notify server of client termination + $this->notifyClientTermination(); + + // Drop all ProcessQueues + foreach ($this->processQueueTable as $pq) { + $pq->drop(); + } + $this->processQueueTable = []; + + // Close telemetry session + if ($this->telemetrySession) { + $this->telemetrySession->close(); + } + + $this->logger->info("PushConsumer shutdown complete, clientId={$this->clientId}"); + } + + /** + * Signal handler for graceful shutdown. + */ + public function requestShutdown(): void + { + $this->shutdownRequested = true; + } + + /** + * Register a message interceptor. + * + * @param MessageInterceptor $interceptor + * @return $this + */ + public function addInterceptor(MessageInterceptor $interceptor): self + { + $this->interceptors[] = $interceptor; + return $this; + } + + /** + * Execute interceptors at a given hook point. + * + * @param string $hookPoint The hook point identifier + * @param array $context Additional context for the interceptor + * @return void + */ + public function executeInterceptors(string $hookPoint, array $context = []): void + { + if (empty($this->interceptors)) { + return; + } + foreach ($this->interceptors as $interceptor) { + try { + $interceptor->intercept($hookPoint, $context); + } catch (\Exception $e) { + $this->logger->warning("Interceptor failed at {$hookPoint}: " . $e->getMessage()); + } + } + } + + /** + * Get the client type identifier. + * + * @return int The PUSH_CONSUMER client type + */ + protected function getClientType(): int + { + return ClientType::PUSH_CONSUMER; + } + + /** + * Register SIGTERM/SIGINT signal handlers. + */ + protected function registerSignalHandlers(): void + { + if (function_exists('pcntl_signal')) { + $self = $this; + pcntl_signal(SIGTERM, function() use ($self) { + $self->requestShutdown(); + }); + pcntl_signal(SIGINT, function() use ($self) { + $self->requestShutdown(); + }); + $this->logger->info("PushConsumer signal handlers registered"); + } + } + + /** + * Establish Telemetry Session with the server for this consumer group. + * + * @return void + * @throws \RuntimeException If session establishment fails + */ + protected function establishTelemetrySession(): void + { + $ua = new UA(); + $ua->setLanguage(Language::PHP); + $ua->setVersion(ClientConstants::CLIENT_VERSION); + + $subscriptionEntries = []; + foreach ($this->subscriptionExpressions as $topic => $expression) { + $filterExpression = new FilterExpression(); + $filterExpression->setExpression($expression); + + $topicResource = new Resource(); + $topicResource->setName($topic); + + $subscriptionEntry = new SubscriptionEntry(); + $subscriptionEntry->setTopic($topicResource); + $subscriptionEntry->setExpression($filterExpression); + + $subscriptionEntries[] = $subscriptionEntry; + } + + $subscription = new Subscription(); + $groupResource = new Resource(); + $groupResource->setName($this->consumerGroup); + $subscription->setGroup($groupResource); + $subscription->setSubscriptions($subscriptionEntries); + + $settings = new Settings(); + $settings->setClientType($this->getClientType()); + $settings->setUserAgent($ua); + $settings->setSubscription($subscription); + + $settings->setAccessPoint($this->parseEndpoints($this->endpoints)); + $timeoutDuration = new Duration(); + $timeoutDuration->setSeconds(3); + $timeoutDuration->setNanos(0); + $settings->setRequestTimeout($timeoutDuration); + + $command = new TelemetryCommand(); + $command->setSettings($settings); + + $success = $this->telemetrySession->createStreamAndSync($command); + if (!$success) { + throw new \RuntimeException("Failed to establish Telemetry Session"); + } + } + + /** + * Scan assignments for all subscribed topics. + */ + private function scanAssignments(): void + { + $this->logger->debug("PushConsumer scanning assignments"); + + $subscriptions = $this->subscriptionExpressions; + foreach ($subscriptions as $topic => $expression) { + try { + $assignments = $this->queryAssignment($topic); + $newAssignments = $assignments ? ProtobufUtil::repeatedFieldToArray($assignments->getAssignments()) : []; + + $oldAssignments = isset($this->cacheAssignments[$topic]) ? $this->cacheAssignments[$topic] : null; + $newIsEmpty = empty($newAssignments); + $oldIsEmpty = $oldAssignments === null || empty($oldAssignments); + if ($newIsEmpty && $oldIsEmpty) { + $this->logger->debug("PushConsumer acquired empty assignment from remote, would scan later, for topic $topic"); + continue; + } + $this->syncProcessQueues($topic, $newAssignments, $expression); + $this->cacheAssignments[$topic] = $newAssignments; + } catch (\Exception $e) { + $this->logger->warning("PushConsumer scanAssignments failed for topic={$topic}: " . $e->getMessage()); + } + } + } + + /** + * Sync ProcessQueues with the latest assignments, creating new queues and dropping stale ones. + * + * @param string $topic Topic name + * @param array $newAssignments Latest assignment list from the server + * @param string $expression Filter expression for the topic + * @return void + */ + private function syncProcessQueues(string $topic, array $newAssignments, string $expression): void + { + $latestMQKeys = []; + foreach ($newAssignments as $assignment) { + $mq = $assignment->getMessageQueue(); + $mqKey = $this->getMqKey($mq); + $latestMQKeys[$mqKey] = $mq; + } + if (empty($newAssignments)) { + $existingCount = count($this->processQueueTable); + if ($existingCount > 0) { + $this->logger->warning("Broker returned 0 assignments for topics={$topic}, keeping {$existingCount} existing ProcessQueues"); + } + return; Review Comment: An empty assignment set is a valid rebalance result, for example when all queues are reassigned to other consumers. Returning here keeps every existing `ProcessQueue` active, so this client continues fetching queues it no longer owns and may cause duplicate/competing consumption. Please allow the removal loop below to clear this topic's existing queues when the new assignment set is empty. ########## .github/workflows/php_build.yml: ########## @@ -17,9 +17,32 @@ jobs: uses: shivammathur/setup-php@v2 with: php-version: ${{ matrix.php-version }} + extensions: grpc, protobuf - name: Validate composer.json working-directory: ./php run: composer validate - name: Install Dependencies working-directory: ./php - run: composer install + run: composer install --no-interaction --prefer-dist + - name: Run PHPUnit Tests + if: runner.os != 'Windows' + working-directory: ./php + run: vendor/bin/phpunit --testsuite "RocketMQ PHP Test Suite" --no-coverage Review Comment: This named suite explicitly excludes `tests/integration`, and the workflow never invokes `RocketMQ PHP Integration Tests`. The current telemetry timeout integration test already disagrees with the updated implementation, but CI cannot detect that regression. Please add a separate integration-test step/job (or run both suites) rather than reporting only the unit suite as the PHP build result. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
