Introduce broadcast API for event sharing

packages/sdk/src/client/attachment.ts

@@ -107,4 +107,11 @@ export class Attachment<T, P extends Indexable> {
     clearTimeout(this.watchLoopTimerID);
     this.watchLoopTimerID = undefined;
   }
+
+  /**
+   * `unsetClient` unsets the client of the document.
+   */
+  public unsetClient(): void {
+    this.doc.setClient();
+  }
 }

packages/sdk/src/client/client.ts

@@ -42,6 +42,7 @@ import {
 import { OpSource } from '@yorkie-js-sdk/src/document/operation/operation';
 import { createAuthInterceptor } from '@yorkie-js-sdk/src/client/auth_interceptor';
 import { createMetricInterceptor } from '@yorkie-js-sdk/src/client/metric_interceptor';
+import { validateSerializable } from '../util/validator';
 
 /**
  * `SyncMode` defines synchronization modes for the PushPullChanges API.
@@ -303,6 +304,7 @@ export class Client {
     }
     doc.setActor(this.id!);
     doc.update((_, p) => p.set(options.initialPresence || {}));
+    doc.setClient(this);
 
     const syncMode = options.syncMode ?? SyncMode.Realtime;
     return this.enqueueTask(async () => {
@@ -584,6 +586,59 @@ export class Client {
     return this.conditions[condition];
   }
 
+  /**
+   * `broadcast` broadcasts the given payload to the given topic.
+   */
+  public broadcast(
+    docKey: DocumentKey,
+    topic: string,
+    payload: any,
+  ): Promise<void> {
+    if (!this.isActive()) {
+      throw new YorkieError(
+        Code.ErrClientNotActivated,
+        `${this.key} is not active`,
+      );
+    }
+    const attachment = this.attachmentMap.get(docKey);
+    if (!attachment) {
+      throw new YorkieError(
+        Code.ErrDocumentNotAttached,
+        `${docKey} is not attached`,
+      );
+    }
+
+    if (!validateSerializable(payload)) {
+      throw new YorkieError(
+        Code.ErrInvalidArgument,
+        'payload is not serializable',
+      );
+    }
+
+    return this.enqueueTask(async () => {
+      return this.rpcClient
+        .broadcast(
+          {
+            clientId: this.id!,
+            documentId: attachment.docID,
+            topic,
+            payload: new TextEncoder().encode(JSON.stringify(payload)),
+          },
+          { headers: { 'x-shard-key': `${this.apiKey}/${docKey}` } },
+        )
+        .then(() => {
+          logger.info(
+            `[BC] c:"${this.getKey()}" broadcasts d:"${docKey}" t:"${topic}"`,
+          );
+        })
+        .catch((err) => {
+          logger.error(`[BC] c:"${this.getKey()}" err :`, err);
+          this.handleConnectError(err);
+          throw err;
+        });
+    });
+  }
+
   /**
    * `runSyncLoop` runs the sync loop. The sync loop pushes local changes to
    * the server and pulls remote changes from the server.
@@ -747,6 +802,7 @@ export class Client {
       return;
     }
 
+    attachment.unsetClient();
     attachment.cancelWatchStream();
     this.attachmentMap.delete(docKey);
   }

packages/sdk/src/document/document.ts

@@ -78,6 +78,7 @@ import {
 import { History, HistoryOperation } from '@yorkie-js-sdk/src/document/history';
 import { setupDevtools } from '@yorkie-js-sdk/src/devtools';
 import * as Devtools from '@yorkie-js-sdk/src/devtools/types';
+import { Client } from '@yorkie-js-sdk/src/client/client';
 
 /**
  * `DocumentOptions` are the options to create a new document.
@@ -173,6 +174,11 @@ export enum DocEventType {
    * `PresenceChanged` means that the presences of the client has updated.
    */
   PresenceChanged = 'presence-changed',
+
+  /**
+   * `Broadcast` means that the message is broadcasted to clients who subscribe to the event.
+   */
+  Broadcast = 'broadcast',
 }
 
 /**
@@ -191,7 +197,8 @@ export type DocEvent<P extends Indexable = Indexable, T = OperationInfo> =
   | InitializedEvent<P>
   | WatchedEvent<P>
   | UnwatchedEvent<P>
-  | PresenceChangedEvent<P>;
+  | PresenceChangedEvent<P>
+  | BroadcastEvent;
 
 /**
  * `TransactionEvent` represents document events that occur within
@@ -371,6 +378,11 @@ export interface PresenceChangedEvent<P extends Indexable>
   value: { clientID: ActorID; presence: P };
 }
 
+export interface BroadcastEvent extends BaseDocEvent {
+  type: DocEventType.Broadcast;
+  value: { topic: string; payload: any };
+}
+
 type DocEventCallbackMap<P extends Indexable> = {
   default: NextFn<
     | SnapshotEvent
@@ -388,6 +400,7 @@ type DocEventCallbackMap<P extends Indexable> = {
   connection: NextFn<ConnectionChangedEvent>;
   status: NextFn<StatusChangedEvent>;
   sync: NextFn<SyncStatusChangedEvent>;
+  broadcast: (topic: string, payload: any) => void;
   all: NextFn<TransactionEvent<P>>;
 };
 export type DocEventTopic = keyof DocEventCallbackMap<never>;
@@ -589,6 +602,8 @@ export class Document<T, P extends Indexable = Indexable> {
    */
   private isUpdating: boolean;
 
+  private client?: Client;
+
   constructor(key: string, opts?: DocumentOptions) {
     this.opts = opts || {};
 
@@ -818,6 +833,15 @@ export class Document<T, P extends Indexable = Indexable> {
     error?: ErrorFn,
     complete?: CompleteFn,
   ): Unsubscribe;
+  /**
+   * `subscribe` registers a callback to subscribe to events on the document.
+   * The callback will be called when the document is changed.
+   */
+  public subscribe(
+    type: 'broadcast',
+    next: DocEventCallbackMap<P>['broadcast'],
+    error?: ErrorFn,
+  ): Unsubscribe;
   /**
    * `subscribe` registers a callback to subscribe to events on the document.
    */
@@ -966,6 +990,17 @@ export class Document<T, P extends Indexable = Indexable> {
           arg4,
         );
       }
+      if (arg1 === 'broadcast') {
+        const callback = arg2 as DocEventCallbackMap<P>['broadcast'];
+        return this.eventStream.subscribe((event) => {
+          for (const docEvent of event) {
+            if (docEvent.type !== DocEventType.Broadcast) {
+              continue;
+            }
+            callback(docEvent.value.topic, docEvent.value.payload);
+          }
+        }, arg3);
+      }
       if (arg1 === 'all') {
         const callback = arg2 as DocEventCallbackMap<P>['all'];
         return this.eventStream.subscribe(callback, arg3, arg4);
@@ -1024,6 +1059,7 @@ export class Document<T, P extends Indexable = Indexable> {
         complete,
       );
     }
+
     throw new YorkieError(Code.ErrInvalidArgument, `"${arg1}" is not a valid`);
   }
 
@@ -1250,6 +1286,15 @@ export class Document<T, P extends Indexable = Indexable> {
     return this.root.getGarbageLen();
   }
 
+  /**
+   * `setClient` sets the client of this document.
+   *
+   * @internal
+   */
+  public setClient(client?: Client): void {
+    this.client = client;
+  }
+
   /**
    * `getGarbageLenFromClone` returns the length of elements should be purged from clone.
    */
@@ -1468,7 +1513,8 @@ export class Document<T, P extends Indexable = Indexable> {
 
     if (resp.body.case === 'event') {
       const { type, publisher } = resp.body.value;
-      const event: Array<WatchedEvent<P> | UnwatchedEvent<P>> = [];
+      const event: Array<WatchedEvent<P> | UnwatchedEvent<P> | BroadcastEvent> =
+        [];
       if (type === PbDocEventType.DOCUMENT_WATCHED) {
         this.addOnlineClient(publisher);
         // NOTE(chacha912): We added to onlineClients, but we won't trigger watched event
@@ -1495,6 +1541,16 @@ export class Document<T, P extends Indexable = Indexable> {
             value: { clientID: publisher, presence },
           });
         }
+      } else if (type === PbDocEventType.DOCUMENT_BROADCAST) {
+        if (resp.body.value.body) {
+          const { topic, payload } = resp.body.value.body;
+          const decoder = new TextDecoder();
+
+          event.push({
+            type: DocEventType.Broadcast,
+            value: { topic, payload: JSON.parse(decoder.decode(payload)) },
+          });
+        }
       }
 
       if (event.length > 0) {
@@ -1970,4 +2026,18 @@ export class Document<T, P extends Indexable = Indexable> {
   public getRedoStackForTest(): Array<Array<HistoryOperation<P>>> {
     return this.internalHistory.getRedoStackForTest();
   }
+
+  /**
+   * `broadcast` the payload to the given topic.
+   */
+  public broadcast(topic: string, payload: any): Promise<void> {
+    if (this.client) {
+      return this.client.broadcast(this.getKey(), topic, payload);
+    }
+
+    throw new YorkieError(
+      Code.ErrClientNotFound,
+      'Document is not attached to a client',
+    );
+  }
 }

packages/sdk/src/util/validator.ts

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2024 The Yorkie Authors. All rights reserved.
+ *
+ * Licensed 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.
+ */
+
+/**
+ * `validateSerializable` returns whether the given value is serializable or not.
+ */
+export const validateSerializable = (value: any): boolean => {
+  try {
+    const serialized = JSON.stringify(value);
+
+    if (serialized === undefined) {
+      return false;
+    }
+  } catch (error) {
+    return false;
+  }
+  return true;
+};