subscribe method

  1. @override
Future<void> subscribe(
  1. SubscribeRequest req,
  2. EventBusSubscribeServer stream
)
override

Implementation

@override
Future<void> subscribe(
    SubscribeRequest req, EventBusSubscribeServer stream) async {
  final sub = _Subscriber(req.filter, queueCapacity);
  _subscribers.add(sub);
  try {
    while (true) {
      final event = await sub.take(const Duration(milliseconds: 100));
      if (event != null) stream.send(event);
    }
  } finally {
    _subscribers.remove(sub);
  }
}