resumeListeningTo method

void resumeListeningTo({
  1. required String topic,
})

Resume all the subscriptions from this instance of Fkafka to the topic.

Implementation

void resumeListeningTo({
  required String topic,
}) {
  _controllers.putIfAbsent(
    topic,
    () => StreamController.broadcast(),
  );

  final subscribers = _subscribers[topic] ?? <FkafkaSubscriber>[];

  for (var i = 0; i < subscribers.length; i++) {
    final subscriber = subscribers[i];
    subscriber.subscription.cancel();

    final subscription = _controllers[topic]!.stream.listen(
      (event) {
        subscriber.onTopic(event.data);
      },
    );

    subscribers[i] = subscriber.copyWith(
      isActive: true,
      subscription: subscription,
    );
  }
}