Introduce broadcast API for event sharing

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.
@@ -584,6 +585,59 @@ export class Client {
     return this.conditions[condition];
   }
 
+  /**
+   * `broadcast` broadcasts the given payload to the given topic.
+   */
+  public broadcast<T, P extends Indexable>(
+    doc: Document<T, P>,
+    topic: string,
+    payload: any,
+  ): Promise<void> {
+    if (!this.isActive()) {
+      throw new YorkieError(
+        Code.ErrClientNotActivated,
+        `${this.key} is not active`,
+      );
+    }
+    const attachment = this.attachmentMap.get(doc.getKey());
+    if (!attachment) {
+      throw new YorkieError(
+        Code.ErrDocumentNotAttached,
+        `${doc.getKey()} 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}/${doc.getKey()}` } },
+        )
+        .then(() => {
+          logger.info(
+            `[BC] c:"${this.getKey()}" broadcasts d:"${doc.getKey()}" 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.

packages/sdk/src/document/document.ts

@@ -173,6 +173,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 +196,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 +377,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
@@ -589,6 +600,15 @@ export class Document<T, P extends Indexable = Indexable> {
    */
   private isUpdating: boolean;
 
+  /**
+   * `broadcastEventHandlers` is a map of broadcast event handlers.
+   * The key is the topic of the broadcast event, and the value is the handler.
+   */
+  private broadcastEventHandlers: Map<
+    string,
+    (topic: string, payload: any) => void
+  >;
+
   constructor(key: string, opts?: DocumentOptions) {
     this.opts = opts || {};
 
@@ -616,6 +636,8 @@ export class Document<T, P extends Indexable = Indexable> {
       redo: this.redo.bind(this),
     };
 
+    this.broadcastEventHandlers = new Map();
+
     setupDevtools(this);
   }
 
@@ -1027,6 +1049,35 @@ export class Document<T, P extends Indexable = Indexable> {
     throw new YorkieError(Code.ErrInvalidArgument, `"${arg1}" is not a valid`);
   }
 
+  /**
+   * subscribeBroadcastEvent registers a callback to subscribe to broadcast events
+   * on the document. The callback will be called when the document receives a
+   * broadcast event with the given topic.
+   */
+  public subscribeBroadcastEvent(
+    topic: string,
+    handler: (topic: string, payload: any) => void,
+    error?: ErrorFn,
+  ): Unsubscribe {
+    this.broadcastEventHandlers.set(topic, handler);
+
+    this.eventStream.subscribe((event) => {
+      for (const docEvent of event) {
+        if (docEvent.type !== DocEventType.Broadcast) {
+          continue;
+        }
+
+        if (docEvent.value.topic === topic) {
+          handler(topic, docEvent.value.payload);
+        }
+      }
+    }, error);
+
+    return () => {
+      this.broadcastEventHandlers.delete(topic);
+    };
+  }
+
   /**
    * `publish` triggers an event in this document, which can be received by
    * callback functions from document.subscribe().
@@ -1468,7 +1519,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 +1547,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) {
@@ -1584,6 +1646,14 @@ export class Document<T, P extends Indexable = Indexable> {
       const { clientID, presence } = event.value;
       this.presences.set(clientID, presence);
     }
+
+    if (event.type === DocEventType.Broadcast) {
+      const { topic, payload } = event.value;
+      const handler = this.broadcastEventHandlers.get(topic);
+      if (handler) {
+        handler(topic, payload);
+      }
+    }
   }
 
   /**

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;
+};

packages/sdk/test/integration/client_test.ts

@@ -863,4 +863,24 @@ describe.sequential('Client', function () {
       assert.equal(d1.toSortedJSON(), d2.toSortedJSON());
     }, task.name);
   });
+
+  it('Throw error when broadcasting unserializeable payload', async ({
+    task,
+  }) => {
+    const cli = new yorkie.Client(testRPCAddr);
+    await cli.activate();
+
+    const doc = new yorkie.Document<{ t: Text }>(toDocKey(`${task.name}`));
+    await cli.attach(doc);
+
+    // broadcast unserializable payload
+    const payload = () => {};
+    const broadcastTopic = 'test';
+
+    expect(async () =>
+      cli.broadcast(doc, broadcastTopic, payload),
+    ).rejects.toThrowErrorCode(Code.ErrInvalidArgument);
+
+    await cli.deactivate();
+  });
 });