Appwrite/appwrite
1
0
Fork 0
mirror of https://github.com/appwrite/appwrite.git synced 2026-05-26 13:51:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

Enhance realtime message handling to support initial connection payload and improve query subscription logic

Browse Source
This commit is contained in:
ArnabChatterjee20k
2026-04-07 16:55:10 +05:30
parent d5fe5c34af
commit ca62504b5a
3 changed files with 65 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/realtime.php
+15 -2
View File
@@ -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.');
src/Appwrite/Messaging/Adapter/Realtime.php
+14 -11
View File
@@ -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])) {
tests/e2e/Services/Realtime/RealtimeCustomClientQueryTestWithMessage.php
+36 -38
View File
@@ -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();
}