eventStream method

Stream<Event> eventStream({
  1. String? routingKeyFilter,
  2. String? krefFilter,
  3. String? cursor,
  4. String? consumerGroup,
  5. bool fromBeginning = false,
  6. Duration? timeout,
})
inherited

Subscribes to real-time events.

routingKeyFilter filters by routing key pattern (supports wildcards). krefFilter filters by kref pattern (supports wildcards). cursor resumes from a previous cursor position (Creator tier+). Pass the cursor from the last received event to continue after reconnection. consumerGroup enables load-balanced delivery across consumers in the same group (Enterprise tier only). fromBeginning starts from the earliest available events instead of live-only (Creator tier+, subject to retention). timeout optionally bounds the gRPC stream; when reached the stream terminates with a DEADLINE_EXCEEDED error.

Returns a stream of Event objects. The stream stays open until cancelled, the connection is lost, or timeout elapses.

Implementation

Stream<Event> eventStream({
  String? routingKeyFilter,
  String? krefFilter,
  String? cursor,
  String? consumerGroup,
  bool fromBeginning = false,
  Duration? timeout,
}) {
  final request = EventStreamRequest();
  if (routingKeyFilter != null && routingKeyFilter.isNotEmpty) {
    request.routingKeyFilter = routingKeyFilter;
  }
  if (krefFilter != null && krefFilter.isNotEmpty) {
    request.krefFilter = krefFilter;
  }
  if (cursor != null && cursor.isNotEmpty) {
    request.cursor = cursor;
  }
  if (consumerGroup != null && consumerGroup.isNotEmpty) {
    request.consumerGroup = consumerGroup;
  }
  if (fromBeginning) {
    request.fromBeginning = true;
  }
  // Streaming opts out of the default per-RPC deadline; only apply a
  // deadline when the caller explicitly passes one.
  final options = timeout == null
      ? streamCallOptions
      : streamCallOptions.mergedWith(CallOptions(timeout: timeout));
  return stub.eventStream(request, options: options);
}