Introduce broadcast API for event sharing

What this PR does / why we need it?

This PR implements a broadcast API, which enables the sharing of a broader range of general events beyond the current document and presence events in Yorkie's Publish-Subscribe model.

Any background context you want to provide?

1. Broadcast Events:

Users can now broadcast custom events with a specified topic and payload. The payload can be of any type, as long as it is serializable.

// Broadcast an event with a topic and payload
const payload = 'hello';
doc.broadcast('TOPIC_NAME', payload);

2. Subscribe to Broadcast Events:

Users can subscribe to specific topics and handle the events via a callback function. The callback is triggered whenever an event with the corresponding topic is broadcast.

// Subscribe to a specific topic for broadcast events
doc.subscribe('broadcast', ({value: {topic, payload, clientID}}) => {
    // Handle the broadcast event for the specified topic
});

What are the relevant tickets?

Checklist

[x] Added relevant tests or not required
[x] Didn't break anything

Summary by CodeRabbit

New Features
- Introduced a broadcast method in the Client class to send messages to specified topics.
- Added new event types Broadcast and LocalBroadcast to enable clients to subscribe to broadcast messages.
- Implemented a method to validate the serializability of payloads for broadcasting.
- Enhanced the Document class with a subscribe method for clients to register handlers for broadcast events.
Bug Fixes
- Improved error handling for broadcasting unserializable payloads within the Client class.
Tests
- Added test cases to verify successful broadcasting of serializable payloads and error handling for invalid payloads.
- Increased test coverage for event subscriptions related to broadcast events.

Walkthrough

The recent changes introduce a broadcasting feature in the Client and Document classes, allowing clients to subscribe to broadcast events. New event types, Broadcast and LocalBroadcast, are added, along with corresponding interfaces and methods for managing subscriptions and broadcasting payloads with validation. Comprehensive test cases validate the new functionality, ensuring robust error handling and event management.

Changes

Files	Change Summary
`packages/sdk/src/client/client.ts`	Added `broadcast` method in `Client` class for payload broadcasting with validation and error handling.
`packages/sdk/src/document/document.ts`	Introduced `Broadcast` and `LocalBroadcast` event types, `BroadcastEvent` and `LocalBroadcastEvent` interfaces, and methods for subscribing to broadcast events. Updated event handling logic.
`packages/sdk/test/integration/client_test.ts`	Added test cases for successful broadcasting of serializable payloads and error handling for unserializable payloads, enhancing test coverage.

Poem

In the meadow where bytes play,
A broadcast hops and bounds today.
With messages shared, both near and far,
Events flutter like a shining star.
So let us dance, both rabbit and friend,
With joy in code, our fun won't end! 🐇✨

Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media?

- [X](https://twitter.com/intent/tweet?text=I%20just%20used%20%40coderabbitai%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20the%20proprietary%20code.%20Check%20it%20out%3A&url=https%3A//coderabbit.ai) - [Mastodon](https://mastodon.social/share?text=I%20just%20used%20%40coderabbitai%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20the%20proprietary%20code.%20Check%20it%20out%3A%20https%3A%2F%2Fcoderabbit.ai) - [Reddit](https://www.reddit.com/submit?title=Great%20tool%20for%20code%20review%20-%20CodeRabbit&text=I%20just%20used%20CodeRabbit%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20proprietary%20code.%20Check%20it%20out%3A%20https%3A//coderabbit.ai) - [LinkedIn](https://www.linkedin.com/sharing/share-offsite/?url=https%3A%2F%2Fcoderabbit.ai&mini=true&title=Great%20tool%20for%20code%20review%20-%20CodeRabbit&summary=I%20just%20used%20CodeRabbit%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20proprietary%20code)

Tips

### Chat There are 3 ways to chat with [CodeRabbit](https://coderabbit.ai): - Review comments: Directly reply to a review comment made by CodeRabbit. Example: - `I pushed a fix in commit .` - `Generate unit testing code for this file.` - `Open a follow-up GitHub issue for this discussion.` - Files and specific lines of code (under the "Files changed" tab): Tag `@coderabbitai` in a new review comment at the desired location with your query. Examples: - `@coderabbitai generate unit testing code for this file.` - `@coderabbitai modularize this function.` - PR comments: Tag `@coderabbitai` in a new PR comment to ask questions about the PR branch. For the best results, please provide a very specific query, as very limited context is provided in this mode. Examples: - `@coderabbitai generate interesting stats about this repository and render them as a table.` - `@coderabbitai show all the console.log statements in this repository.` - `@coderabbitai read src/utils.ts and generate unit testing code.` - `@coderabbitai read the files in the src/scheduler package and generate a class diagram using mermaid and a README in the markdown format.` - `@coderabbitai help me debug CodeRabbit configuration file.` Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. ### CodeRabbit Commands (Invoked using PR comments) - `@coderabbitai pause` to pause the reviews on a PR. - `@coderabbitai resume` to resume the paused reviews. - `@coderabbitai review` to trigger an incremental review. This is useful when automatic reviews are disabled for the repository. - `@coderabbitai full review` to do a full review from scratch and review all the files again. - `@coderabbitai summary` to regenerate the summary of the PR. - `@coderabbitai resolve` resolve all the CodeRabbit review comments. - `@coderabbitai configuration` to show the current CodeRabbit configuration for the repository. - `@coderabbitai help` to get help. ### Other keywords and placeholders - Add `@coderabbitai ignore` anywhere in the PR description to prevent this PR from being reviewed. - Add `@coderabbitai summary` to generate the high-level summary at a specific location in the PR description. - Add `@coderabbitai` anywhere in the PR title to generate the title automatically. ### CodeRabbit Configuration File (`.coderabbit.yaml`) - You can programmatically configure CodeRabbit by adding a `.coderabbit.yaml` file to the root of your repository. - Please see the [configuration documentation](https://docs.coderabbit.ai/guides/configure-coderabbit) for more information. - If your editor has YAML language server enabled, you can add the path at the top of this file to enable auto-completion and validation: `# yaml-language-server: $schema=https://coderabbit.ai/integrations/schema.v2.json` ### Documentation and Community - Visit our [Documentation](https://coderabbit.ai/docs) for detailed information on how to use CodeRabbit. - Join our [Discord Community](https://discord.com/invite/GsXnASn26c) to get help, request features, and share feedback. - Follow us on [X/Twitter](https://twitter.com/coderabbitai) for updates and announcements.

@sejongk

There are a few things I'd like to share:

1. How Handlers for Each Topic Are Managed:

Inside the Document class, I've added a broadcastEventHandlers which is a map to manage the handlers for each topic.

// document.ts 
export class Document<T, P extends Indexable = Indexable> {
  // skipped

  private broadcastEventHandlers: Map<
    string,
    (topic: string, payload: any) => void
  >;
}

When a user subscribes to a specific topic for broadcast events:

// User subscribes to a broadcast event with the topic "TOPIC_NAME"
doc.subscribeBroadcastEvent('TOPIC_NAME', (topic, payload) => {
    // Handle the broadcast event for the specified topic
});

The handler is registered in the broadcastEventHandlers map:

export class Document<T, P extends Indexable = Indexable> {

  public subscribeBroadcastEvent(
    topic: string,
    handler: (topic: string, payload: any) => void,
    error?: ErrorFn,
  ): Unsubscribe {

    // Register the handler in broadcastEventHandlers 
    this.broadcastEventHandlers.set(topic, handler);

2. How PbDocEventType is Converted to DocEventType:

Within the applyWatchStream method of the Document class, which applies the given watch stream response to the document, the conversion from PbDocEventType to DocEventType occurs as follows:

  public applyWatchStream(resp: WatchDocumentResponse) {
      // skipped
      } 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) {
        this.publish(event);
      }
    }
  }

3. Test Case for Sending Unserializable Payloads:

As discussed earlier, I've added a test case in client_test.ts to handle unserializable payloads. If there are any additional test cases needed, please let me know.

Lastly, I've verified that the broadcast API works as expected in several examples, but I believe further testing is necessary to ensure complete coverage.

@gwbaik9717 CI failed due to incorrect pnpm version in CI configuration. It is currently fixed in the main branch.

@sejongk

Here is an update.

1. Unified Broadcast Event Subscription

Users can now use the existing subscribe method to subscribe to broadcast events. Since subscribing to a broadcast event requires both a type and a topic, I've introduced a new SubscribePair type, which is structured as follows:

type SubscribePair = {
  type: string;
};

Building on this, the BroadcastSubscribePair extends SubscribePair:

type BroadcastSubscribePair = {
  type: 'broadcast';
  topic: string;
} & SubscribePair;

Here’s an example of how to subscribe to a broadcast event:

   const unsubscribe = d2.subscribe(
        { type: 'broadcast', topic: "TOPIC_NAME" },
        (topic, payload) => {},
   );

2. Call broadcast from Document not Client

Broadcasting is now handled directly by the Document not Client, aligning with the subscription mechanism.

To support this, I’ve added a client attribute within the Document class. This client attribute is an instance of the Client and is set when the client attaches to the document, and unset when it detaches.

export class Document<T, P extends Indexable = Indexable> {
  // skipped
  private client?: Client;

    /**
   * `setClient` sets the client of this document.
   *
   * @internal
   */
  public setClient(client?: Client): void {
    this.client = client;
  }

}

Here’s an example of how to broadcast a topic with payload.

// must be serializable
const payload = {a:1, b:"2"}
await doc.broadcast("TOPIC_NAME", payload);

@sejongk

Sorry for the late review. I'll check it today.

CC) @chacha912, @gwbaik9717

@hackerwins, I've reviewed, but could you please check it once more in case I missed anything? (Please check the comments I've left with cc regarding the subscribe interface and client property)

@hackerwins @chacha912 @sejongk

Here's an update.

In order to resolve the circular reference issue, we’ve removed the client reference from the Document class. Instead, we've decided to have Client subscribe to Document's broadcast events and handle them by calling Client's broadcast method.

However, this solution presents two challenges:

1. Unsubscribing Handlers

We had to ensure that handlers registered in subscribe are properly unsubscribed when document is detached from client. To address this, we’ve decided to manage unsubscribeBroadcastEvent within the Attachment class, since the lifecycle of the client’s attachment and the sub/unsub process are the same.

export class Attachment<T, P extends Indexable> {
  // skipped
  unsubscribeBroadcastEvent: Unsubscribe;
}

With this approach, when attaching a document to a client using the attach method of Client, we subscribe to the document’s event stream, and we unsubscribe when detaching:

  public attach<T, P extends Indexable>(
    doc: Document<T, P>,
    options: {
      initialPresence?: P;
      syncMode?: SyncMode;
    } = {},
  ): Promise<Document<T, P>> {
    // skipped
    const unsubscribeBroacastEvent = doc.subscribe(
      'broadcast',
      (topic, payload, onBroadcastError) => {
        try {
          this.broadcast(doc.getKey(), topic, payload);
        } catch (e: unknown) {
          if (e instanceof Error) {
            onBroadcastError?.(e);
          }
        }
      },
    );

    this.attachmentMap.set(
          doc.getKey(),
          new Attachment(
            this.reconnectStreamDelay,
            doc,
            res.documentId,
            syncMode,
            unsubscribeBroacastEvent,
         ),
    );

  private detachInternal(docKey: DocumentKey) {
    // skipped
    attachment.unsubscribeBroadcastEvent();

2. Error Handling for Broadcast Events

The second challenge involves handling errors during broadcast events, such as when broadcasting an unserializable payload. Initially, we could simply use a try-catch block to handle errors when directly accessing the client's broadcast method:

try {
   doc.broadcast(broadcastTopic, payload)
}catch(e){
 // do something
}

However, since we’re now dealing with an event stream, a different approach to error handling is required. To accommodate this, I've introduced a new interface for broadcasting events. This interface includes an error handling function that triggers when a broadcast event fails.

// Usage
const errorFn = () => {
    // Do something
}
doc.broadcast("topic", payload, errorFn)

Here's an update.

As mentioned earlier, we've decided to have the Client subscribe to the Document's broadcast events and handle them by invoking the Client's broadcast method. The challenge we encountered was distinguishing between local and remote broadcast events. Since the client is now subscribed to the document's broadcast events, both local and remote events are received, which creates the following issue:

Without differentiation, this setup can lead to an unexpected infinite loop. This happens because the client, upon receiving its own broadcast event, will rebroadcast it, creating a never-ending cycle.

This scenario can be reproduced with the following test code. The test simulates a situation where the publisher is broadcasting an event and, at the same time, subscribing to the broadcast event. Currently, since Yorkie does not support self-broadcast filtering, the subscribe handler gets called, leading to numerous recursive broadcasts, as evidenced by the test results:

 it('Should not trigger the handler for a broadcast event sent by the publisher to itself', async ({
    task,
  }) => {
    await withTwoClientsAndDocuments<{ t: Text }>(
      async (c1, d1, c2, d2) => {
        const eventCollector1 = new EventCollector<[string, any]>();
        const eventCollector2 = new EventCollector<[string, any]>();
        const broadcastTopic = 'test';
        const payload = { a: 1, b: '2' };

        // Publisher subscribes to the broadcast event
        const unsubscribe1 = d1.subscribe('broadcast', (topic, payload) => {
          if (topic === broadcastTopic) {
            eventCollector1.add([topic, payload]);
          }
        });

        const unsubscribe2 = d2.subscribe('broadcast', (topic, payload) => {
          if (topic === broadcastTopic) {
            eventCollector2.add([topic, payload]);
          }
        });

        d1.broadcast(broadcastTopic, payload);

        // Assuming that subscribers can receive the broadcast event within 1000ms.
        await new Promise((res) => setTimeout(res, 1000));

        unsubscribe1();
        unsubscribe2();

        assert.equal(eventCollector1.getLength(), 0);
        assert.equal(eventCollector2.getLength(), 1);
      },
      task.name,
      SyncMode.Realtime,
    );
  });

To resolve this issue, I've introduced a LocalBroadcast event type in addition to the existing Broadcast event. The LocalBroadcast event is only triggered when the local document attempts to broadcast. This addition allows us to clearly distinguish between local and remote broadcast events, preventing the infinite loop scenario by ensuring that a client does not handle its own broadcast events.

export enum DocEventType {
  /**
   * `Broadcast` means that the broadcast event is received from the remote client.
   */
  Broadcast = 'broadcast',

  /**
   * `LocalBroadcast` means that the broadcast event is sent from the local client.
   */
  LocalBroadcast = 'local-broadcast',
}

 /**
   * `broadcast` the payload to the given topic.
   */
  public broadcast(topic: string, payload: any, error?: ErrorFn) {
    const broadcastEvent: LocalBroadcastEvent = {
      type: DocEventType.LocalBroadcast,
      value: { topic, payload },
      error,
    };

    this.publish([broadcastEvent]);
  }

However, the above method still has a drawback. We may want to hide the LocalBroadcast event from users, as it is primarily intended for internal handling rather than user-facing logic.

doc.subscribe('local-broadcast', (event) => {
    // Users do not need to know about this event
},
);

yorkie-team / yorkie-js-sdk