 }
 
 public extension EventsSource {
+  /// **internal use only**
+  func _onEvent(
+    executor: Executor = .primary,
+    _ block: @escaping (_ event: Event, _ originalExecutor: Executor) -> Void) {
+    let handler = self.makeHandler(executor: executor, block)
+    self._asyncNinja_retainHandlerUntilFinalization(handler)
+  }
+
+  /// **internal use only**
+  func _onEvent<C: ExecutionContext>(
+    context: C,
+    executor: Executor? = nil,
+    _ block: @escaping (_ strongContext: C, _ event: Event, _ originalExecutor: Executor) -> Void) {
+    _onEvent(executor: executor ?? context.executor) {
+      [weak context] (event, originalExecutor) in
+      if let context = context {
+        block(context, event, originalExecutor)
+      }
+    }
+  }
+
   /// Subscribes for buffered and new values (both update and completion) for the channel
   ///
   /// - Parameters:

    context: C,
    executor: Executor? = nil,
    _ block: @escaping (_ strongContext: C, _ event: Event) -> Void) {

    let handler = self.makeHandler(executor: executor ?? context.executor)
    {
      [weak context] (event, originalExecutor) in
      guard let context = context else { return }
      block(context, event)
    }

    if let handler = handler {
      context.releaseOnDeinit(handler)
    }
  }


  func onEvent(
    executor: Executor = .primary,
    _ block: @escaping (_ event: Event) -> Void) {
    _onEvent { (event, originalExecutor) in
      block(event)
    }
  }

  /// Subscribes for buffered and new values (both update and completion) for the channel
