     }
   }
 
-  public func map<T>(executor: Executor = .primary, _ transform: @escaping  (Value) -> T) -> Channel<T> {
-    return makeDerivedChannel { (producer, value) in
-      executor.execute { producer.send(transform(value)) }
+  func map<T>(executor: Executor = .primary, _ transform: @escaping (Value) -> T) -> Channel<T> {
+    return self.makeDerivedChannel(executor: executor) { (producer, value) in
+      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
+  func flatMap<T>(executor: Executor = .primary, transform: @escaping (Value) -> T?) -> Channel<T> {
+    return self.makeDerivedChannel(executor: executor) { (producer, value) in
+      if let transformedValue = transform(value) {
+        producer.send(transformedValue)
+      }
     }
   }
 
-  public func bufferedPairs() -> Channel<(T, T)> {
+  func flatMap<S: Sequence>(executor: Executor = .primary, transform: @escaping (Value) -> S) -> Channel<S.Iterator.Element> {
+    return self.makeDerivedChannel(executor: executor) { (producer, value) in
+      producer.send(transform(value))
+    }
+  }
+
+  func filter(executor: Executor = .primary, _ predicate: @escaping (Value) -> Bool) -> Channel<Value> {
+    return self.makeDerivedChannel(executor: executor) { (producer, value) in
+      if predicate(value) {
+        producer.send(value)
+      }
+    }
+  }
+
+  func changes() -> Channel<(T?, T)> {
+    var previousValue: Value? = nil
+
+    return self.makeDerivedChannel(executor: .immediate) { (producer, value) in
+      let change = (previousValue, value)
+      previousValue = value
+      producer.send(change)
+    }
+  }
+
+  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]> {
+  func buffered(capacity: Int) -> Channel<[T]> {
     var buffer = [T]()
     buffer.reserveCapacity(capacity)
 
-    return makeDerivedChannel { (producer, value) in
+    return self.makeDerivedChannel(executor: .immediate) { (producer, value) in
       buffer.append(value)
       if capacity == buffer.count {
         producer.send(buffer)

    fatalError() // abstract
  }

  final public 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? {
    let sema = DispatchSemaphore(value: 0)
    var result: Value? = nil


    }
  }

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