zhaohai666 commented on code in PR #1256: URL: https://github.com/apache/rocketmq-clients/pull/1256#discussion_r3688339165
########## 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: Fixed ########## 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: Fixed ########## 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: Fixed ########## 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: Fixed -- 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]
