PubSub - Flutter

PubSub feature allows the participant to send and receive messages of the topics which he has subscribed.

note

From v4.x, publish(), subscribe() and unsubscribe() throw a PubSubException when the call does not take effect, so every call site needs a catch or a .catchError — an unhandled one becomes a runtime async error. See Handling failures below and section 1 of the Migration Guide.

Methods

publish()

This method is use for publishing message of specific topic.

Syntax

  Future<void> publish(
    String topic,
    String message,
    [PubSubPublishOptions options]
  )

Parameter Name	Type	Description
topic	String	This should be the topic for which you are publishing a message.
message	String	This is the actual message, which will be published to participants, who had subscribed to a particular topic.
options	PubSubPublishOptions	This is an object of PubSubPublishOptions, which provides an option, such as persist, which persists message history for upcoming participants.

Example

  // publish()
  ElevatedButton(
    onPressed: () async {
      try {
        // Publish a message
        await room.pubSub.publish(
          "CHAT", // Topic
          "Hello from Flutter!", // Message Content
          const PubSubPublishOptions(
            persist: true, // Stores the message in server for future participants
          ),
        );

        print("Message published");
      } on PubSubException catch (e) {
        print("Publish failed: ${e.message}");
      }
    },
    child: Text("Publish"),
  ),

subscribe()

This method is used to subscribe for particular topic. This method returns a list of messages which were sent earlier.

Syntax

  Future<PubSubMessages> subscribe(
    String topic,
    Function(PubSubMessage) onMessageReceived, {
    PubSubSubscribeOptions? options,
    void Function(List<PubSubMessage> messages, bool isLast)? onOldMessagesReceived,
    void Function(List<PubSubMessage> messages)? onBatchReceived,
    void Function(int droppedCount)? onMessageDrop,
  })

Parameter Name	Type	Description
topic	String	This should be the topic to be subscribed.
onMessageReceived	Function(PubSubMessage)	This is a handler function, which will be called when new message received.
options	PubSubSubscribeOptions	This bounds how much history is replayed on subscribe and how live messages are delivered on a busy topic. Optional.
onOldMessagesReceived	Function(List<PubSubMessage>, bool)	Called as stored history loads, oldest first, so a long conversation starts rendering before all of it has arrived. isLast is true on the final batch. Optional.
onBatchReceived	Function(List<PubSubMessage>)	On a busy topic live messages are delivered in groups. Called with each group. Optional.
onMessageDrop	Function(int)	Called with the number of live messages that could not be delivered. Optional.

The options are a PubSubSubscribeOptions, and every value on it is optional:

Option	Description
oldMessageLimit	How many stored messages are replayed on subscribe. null replays all, 0 replays none, N > 0 replays the last N.
realtimeOverflow	PubSubRealtimeOverflow.queue briefly holds live messages so a subscriber that falls behind can catch up; PubSubRealtimeOverflow.drop discards them. Use drop only where stale data has no value.
maxQueue	How far a subscriber may fall behind, counted in delivery batches, before the server gives up on it. Only allowed with queue.
newMessageLimit	Caps how many live messages this subscriber accepts at a time.

An out-of-range option value makes subscribe() throw an ArgumentError before anything is sent. See section 5 of the Migration Guide for the exact rules.

Example

  // subscribe()
  ElevatedButton(
    onPressed: () async {
      try {
        // Subscribe 'CHAT' topic and get persisted messages
        var messages = await room.pubSub.subscribe(
          "CHAT",
          messageHandler,
          options: const PubSubSubscribeOptions(oldMessageLimit: 100),
        );

        // Printing message list
        print("Messages: ${messages.messages.map((msg) => msg.message).join(" ")}");
      } on PubSubException catch (e) {
        // Nothing is left behind — messageHandler was not registered.
        print("Subscribe failed: ${e.message}");
      }
    },
    child: Text("Subscribe"),
  ),

  // Message Handler
  void messageHandler(msg){
    // Do something
    print("New message received: $msg");
  }

unsubscribe()

This method is used to unsubscribe the message topic.

Syntax

  Future<void> unsubscribe(
    String topic,
    Function(PubSubMessage) messageHandler,
  )

Parameter Name	Type	Description
topic	String	This should be the topic to be unsubscribed.
messageHandler	String	This is a handler function, which was passed in subscribe().

From v4.x this method completes only once the server has answered, bounded by the same 30 second timeout as publish(). Your handler is detached before anything that can throw, so a dispose() teardown always completes.

Example

  // unsubscribe
  ElevatedButton(
    onPressed: () async {
      try {
        // Unsubscribe 'CHAT' topic
        await room.pubSub.unsubscribe("CHAT", messageHandler);
      } on PubSubException catch (e) {
        print("Unsubscribe failed: ${e.message}");
      }
    },
    child: Text("UnSubscribe"),
  ),

Handling failures

From v4.x, all three methods throw a PubSubException when the call does not take effect. Dart cannot flag an unhandled one at compile time, so an unawaited onPressed: () => room.pubSub.publish(...) turns a failure into a runtime async error — give every call site a catch or a .catchError.

Situation	Exception	Code
publish() was rejected	PubSubPublishFailed	4087
subscribe() was rejected	PubSubSubscribeFailed	4088
unsubscribe() was rejected	PubSubUnsubscribeFailed	4089
Called before joining, or after leaving	PubSubMeetingNotJoined	3022
Called while reconnecting	PubSubMeetingReconnecting	3027

PubSubException is sealed, so a switch over these is exhaustive. The same failure is also delivered to Events.error with the same code, name and message — catch to handle one call, listen on the event for centralized logging.

Sample Code

import 'package:flutter/material.dart';
import 'package:videosdk/videosdk.dart';

// ChatScreen
class ChatScreen extends StatefulWidget {
  final Room room;
  const ChatScreen({required this.room});

  @override
  _ChatScreenState createState() => _ChatScreenState();
}

class _ChatScreenState extends State<ChatScreen> {
  @override
  void initState() {
    super.initState();

    widget.room.pubSub.subscribe("CHAT", messageHandler)
      .then((messages) =>
          print("Subscribed to chat: ${messages.messages.map((msg) => msg.message).join(" ")}"))
      // initState() cannot await, so failures are caught here.
      .catchError((Object e) {
        print("Subscribe failed: $e");
        return PubSubMessages(messages: const []);
      });
  }

  @override
  Widget build(BuildContext context) {
    final room = widget.room;
    return Scaffold(
      appBar: AppBar(title: const Text("PubSub Sample Code")),
      body: Center(
        child: ElevatedButton(
          child: const Text("Send Hello !"),
          // Publish Hello message
          onPressed: () async {
            try {
              await room.pubSub.publish(
                "CHAT",
                "Hello",
                const PubSubPublishOptions(persist: true),
              );
            } on PubSubException catch (e) {
              print("Publish failed: ${e.message}");
            }
          },
        ),
      ),
    );
  }

  // Handle incoming messages
  void messageHandler(PubSubMessage message) {
    print("New message: ${message.message}");
  }

  @override
  void dispose() {
    // Unsubscribe. dispose() cannot await, so the failure is logged instead.
    widget.room.pubSub
        .unsubscribe("CHAT", messageHandler)
        .catchError((Object e) => print("Unsubscribe failed: $e"));
    super.dispose();
  }
}