From ca62504b5acfebf2ce97c7331824830a7bff2ddb Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 7 Apr 2026 16:55:10 +0530 Subject: [PATCH] Enhance realtime message handling to support initial connection payload and improve query subscription logic --- app/realtime.php | 17 ++++- src/Appwrite/Messaging/Adapter/Realtime.php | 25 ++++--- ...altimeCustomClientQueryTestWithMessage.php | 74 +++++++++---------- 3 files changed, 65 insertions(+), 51 deletions(-) diff --git a/app/realtime.php b/app/realtime.php index 796686be3e..3bceb54f26 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -701,7 +701,19 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, * Channels Check */ if (empty($channels)) { - throw new Exception(Exception::REALTIME_POLICY_VIOLATION, 'Missing channels'); + // in case of message based 'subscribe' channels will be empty at first and only projectId and roles will be available + $connectedPayloadJson = json_encode([ + 'type' => 'connected', + 'data' => [ + 'channels' => [], + 'subscriptions' => [], + 'user' => $user + ] + ]); + + $realtime->subscribe($project->getId(), $connection, '', $roles, [], []); + $server->send([$connection], $connectedPayloadJson); + return; } $names = array_keys($channels); @@ -995,8 +1007,9 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re if (!is_array($payload['channels']) || !array_is_list($payload['channels'])) { throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'channels is not a valid array.'); } + // registering the queries if not present and check in the same payload later on if (!array_key_exists('queries', $payload)) { - throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'queries is not present in payload.'); + $payload['queries'] = []; } if (!is_array($payload['queries']) || !array_is_list($payload['queries'])) { throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'queries is not a valid array.'); diff --git a/src/Appwrite/Messaging/Adapter/Realtime.php b/src/Appwrite/Messaging/Adapter/Realtime.php index 7a2b6fe19a..f9e97f0e45 100644 --- a/src/Appwrite/Messaging/Adapter/Realtime.php +++ b/src/Appwrite/Messaging/Adapter/Realtime.php @@ -74,18 +74,21 @@ class Realtime extends MessagingAdapter } $strings = []; - if (empty($queryGroup)) { - $strings[] = Query::select(['*'])->toString(); - } else { - foreach ($queryGroup as $query) { - $strings[] = $query->toString(); - } - } + $data = []; - $data = [ - 'strings' => $strings, - 'compiled' => RuntimeQuery::compile($queryGroup), - ]; + if (!empty($channels)) { + if (empty($queryGroup)) { + $strings[] = Query::select(['*'])->toString(); + } else { + foreach ($queryGroup as $query) { + $strings[] = $query->toString(); + } + } + $data = [ + 'strings' => $strings, + 'compiled' => RuntimeQuery::compile($queryGroup), + ]; + } foreach ($roles as $role) { if (!isset($this->subscriptions[$projectId][$role])) { diff --git a/tests/e2e/Services/Realtime/RealtimeCustomClientQueryTestWithMessage.php b/tests/e2e/Services/Realtime/RealtimeCustomClientQueryTestWithMessage.php index 0185eb873f..edce428e0f 100644 --- a/tests/e2e/Services/Realtime/RealtimeCustomClientQueryTestWithMessage.php +++ b/tests/e2e/Services/Realtime/RealtimeCustomClientQueryTestWithMessage.php @@ -27,7 +27,7 @@ class RealtimeCustomClientQueryTestWithMessage extends Scope /** * Same signature as `RealtimeBase::getWebsocket()`, but: * - never sends queries in the URL (avoids URL length limits) - * - once connected, updates the generated subscription using a bulk `type: "query"` message + * - once connected, sends channel/query data using a `type: "subscribe"` message */ private function getWebsocket( array $channels = [], @@ -42,7 +42,6 @@ class RealtimeCustomClientQueryTestWithMessage extends Scope $queryString = \http_build_query([ 'project' => $projectId, - 'channels' => $channels, ]); $client = new WebSocketClient( @@ -55,25 +54,30 @@ class RealtimeCustomClientQueryTestWithMessage extends Scope $connected = \json_decode($client->receive(), true); $this->assertEquals('connected', $connected['type'] ?? null); - if ($queries === null) { + if (empty($channels)) { return $client; } - $subscriptions = $connected['data']['subscriptions'] ?? []; - $this->assertNotEmpty($subscriptions); - $subscriptionId = $subscriptions[\array_key_first($subscriptions)]; - if ($queries === []) { $queries = [Query::select(['*'])->toString()]; } + $payload = [[ + 'channels' => $channels, + ]]; + + if ($queries !== null) { + $payload[0]['queries'] = $queries; + } + + $existingSubscriptions = $connected['data']['subscriptions'] ?? []; + if (!empty($existingSubscriptions)) { + $payload[0]['subscriptionId'] = $existingSubscriptions[\array_key_first($existingSubscriptions)]; + } + $client->send(\json_encode([ 'type' => 'subscribe', - 'data' => [[ - 'subscriptionId' => $subscriptionId, - 'channels' => $channels, - 'queries' => $queries, - ]], + 'data' => $payload, ])); $response = \json_decode($client->receive(), true); @@ -101,7 +105,6 @@ class RealtimeCustomClientQueryTestWithMessage extends Scope $projectId = $this->getProject()['$id']; $queryString = \http_build_query([ 'project' => $projectId, - 'channels' => $channels, ]); $client = new WebSocketClient( @@ -114,14 +117,9 @@ class RealtimeCustomClientQueryTestWithMessage extends Scope $connected = \json_decode($client->receive(), true); $this->assertEquals('connected', $connected['type'] ?? null); - $subscriptions = $connected['data']['subscriptions'] ?? []; - $this->assertNotEmpty($subscriptions); - $subscriptionId = $subscriptions[\array_key_first($subscriptions)]; - $client->send(\json_encode([ 'type' => 'subscribe', 'data' => [[ - 'subscriptionId' => $subscriptionId, 'channels' => $channels, 'queries' => $queryStrings, ]], @@ -182,7 +180,6 @@ class RealtimeCustomClientQueryTestWithMessage extends Scope $queryString = \http_build_query([ 'project' => $projectId, - 'channels' => ['documents'], ]); $client = new WebSocketClient( 'ws://appwrite.test/v1/realtime?' . $queryString, @@ -193,9 +190,12 @@ class RealtimeCustomClientQueryTestWithMessage extends Scope ); $connected = \json_decode($client->receive(), true); $this->assertEquals('connected', $connected['type'] ?? null); - $mapping = $connected['data']['subscriptions'] ?? []; - $this->assertNotEmpty($mapping); - $initialSubscriptionId = $mapping[\array_key_first($mapping)]; + $initialResponse = $this->sendSubscribeMessage($client, [[ + 'channels' => ['documents'], + 'queries' => [Query::select(['*'])->toString()], + ]]); + $initialSubscriptionId = $initialResponse['data']['subscriptions'][0]['subscriptionId'] ?? ''; + $this->assertNotEmpty($initialSubscriptionId); $q1 = [Query::equal('status', ['q1'])->toString()]; $r1 = $this->sendSubscribeMessage($client, [[ @@ -373,7 +373,7 @@ class RealtimeCustomClientQueryTestWithMessage extends Scope $client = $this->getWebsocketWithCustomQuery( [ - 'channels' => ['project'], + 'project' => $projectId, ], [ 'origin' => 'http://localhost', @@ -384,22 +384,18 @@ class RealtimeCustomClientQueryTestWithMessage extends Scope $response = \json_decode($client->receive(), true); $this->assertSame('connected', $response['type']); - $this->assertContains('project', $response['data']['channels']); - $this->assertArrayHasKey('subscriptions', $response['data']); - $this->assertIsArray($response['data']['subscriptions']); - $this->assertNotEmpty($response['data']['subscriptions']); + $subscribeResponse = $this->sendSubscribeMessage($client, [[ + 'channels' => ['project'], + 'queries' => [Query::select(['*'])->toString()], + ]]); + $this->assertCount(1, $subscribeResponse['data']['subscriptions']); + $this->assertSame(['project'], $subscribeResponse['data']['subscriptions'][0]['channels']); $client->close(); - $queryArray = [Query::select(['*'])->toString()]; $clientWithQuery = $this->getWebsocketWithCustomQuery( [ - 'channels' => ['project'], - 'project' => [ - 0 => [ - 0 => $queryArray[0], - ], - ], + 'project' => $projectId, ], [ 'origin' => 'http://localhost', @@ -410,10 +406,12 @@ class RealtimeCustomClientQueryTestWithMessage extends Scope $response = \json_decode($clientWithQuery->receive(), true); $this->assertSame('connected', $response['type']); - $this->assertContains('project', $response['data']['channels']); - $this->assertArrayHasKey('subscriptions', $response['data']); - $this->assertIsArray($response['data']['subscriptions']); - $this->assertNotEmpty($response['data']['subscriptions']); + $subscribeResponseWithQuery = $this->sendSubscribeMessage($clientWithQuery, [[ + 'channels' => ['project'], + 'queries' => [Query::select(['*'])->toString()], + ]]); + $this->assertCount(1, $subscribeResponseWithQuery['data']['subscriptions']); + $this->assertSame(['project'], $subscribeResponseWithQuery['data']['subscriptions'][0]['channels']); $clientWithQuery->close(); }