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