     fatalError() // abstract
   }
 
-  final public func onValue(executor: Executor = .primary, block: @escaping (Value) -> Void) {
+  // MARK: - private helpers
+  func makeDerivedChannel<T>(executor: Executor, onValue: @escaping (Producer<T>, Value) -> ()) -> Channel<T> {
+    let derivedChannel = Producer<T>()
+    weak var weakDerivedChannel: Producer<T>? = derivedChannel
+    weak var weakHandler: Handler? = nil
+
+    let handler = Handler(executor: executor) { [weak self] in
+      guard let derivedChannel = weakDerivedChannel else {
+        if let self_ = self, let handler = weakHandler {
+          self_.remove(handler: handler)
+        }
+        return
+      }
+
+      onValue(derivedChannel, $0)
+    }
+    weakHandler = handler
+    self.add(handler: handler)
+    return derivedChannel
+  }
+}
+
+public extension Channel {
+  func onValue(executor: Executor = .primary, block: @escaping (Value) -> Void) {
     let handler = Handler(executor: executor, block: block)
     self.add(handler: handler)
   }
 
-  final public func wait(waitingBlock: (DispatchSemaphore) -> DispatchTimeoutResult) -> T? {
+  func wait(waitingBlock: (DispatchSemaphore) -> DispatchTimeoutResult) -> T? {
     let sema = DispatchSemaphore(value: 0)
     var result: Value? = nil
 

    }
  }

  public func map<T>(executor: Executor = .primary, _ transform: @escaping  (Value) -> T) -> Channel<T> {
    return makeDerivedChannel { (producer, value) in
      executor.execute { producer.send(transform(value)) }
    }
  }

  public func changes() -> Channel<(T?, T)> {
    var previousValue: Value? = nil

    return makeDerivedChannel { (producer, value) in
      let change = (previousValue, value)
      producer.send(change)
      previousValue = value
    }
  }

  public func bufferedPairs() -> Channel<(T, T)> {
    return self.buffered(capacity: 2).map(executor: .immediate) { ($0[0], $0[1]) }
  }

  func makeDerivedChannel<T>(onValue: @escaping (Producer<T>, Value) -> ()) -> Channel<T> {
    let derivedChannel = Producer<T>()
    weak var weakDerivedChannel: Producer<T>? = derivedChannel
    weak var weakHandler: Handler? = nil

    let handler = Handler(executor: .immediate) { [weak self] in
      guard let derivedChannel = weakDerivedChannel else {
        if let self_ = self, let handler = weakHandler {
          self_.remove(handler: handler)
        }
        return
      }

      onValue(derivedChannel, $0)
    }
    weakHandler = handler
    self.add(handler: handler)
    return derivedChannel
  }

  public func buffered(capacity: Int) -> Channel<[T]> {
    var buffer = [T]()
    buffer.reserveCapacity(capacity)

    return makeDerivedChannel { (producer, value) in
      buffer.append(value)
      if capacity == buffer.count {
        producer.send(buffer)

    }
  }

  func enumerated() -> Channel<(Int, T)> {
    var index: Int64 = -1
    return self.map(executor: .immediate) {
      let localIndex = Int(OSAtomicIncrement64(&index))
      return (localIndex, $0)
    }
  }
}
