                         let k = f(+x)
                         var chan = map[k]
                         if (chan == nil) {
-                            chan = Dispatcher(outer.callerContext, outer.calleeContext)
+                            chan = ForeignSource()
                             map[k] = chan
                             outer.emitIfOpen(.Next(Box(chan!)))
                         }
                         chan!.emitValue((k, x.raw))
                     default:
-                        for o in map.values { o.emit(e.map { _ in undefined() } ) }
+                        for o in map.values { o.emit(e.nullify()) }
+                        outer.emit(e.nullify())
                     }
                 }
             }

        }
    }
    
    func map<B>(f: A -> B) -> Packet<B> {
        switch self {
        case let .Next(x): return .Next(x.map(f))
        case let .Fail(x): return .Fail(x)
        case let .Done( ): return .Done( )
        }
    }

}


    public func zipWithContext() -> Stream<(A, ExecutionContext)> {
        return Streams.pipe([.AllowSync]) { self.zipWith($0.calleeContext) }
    }

    // Carrying inside streams out of the outside stream causes to lose "stability of meaning" of stream.
    public func groupBy<K: Hashable>(f: A -> K) -> Stream<Stream<(K, A)>> {
        return Streams.source([.AllowSync]) { outer in
            var base: Channel<A>?
            var map = [K: ForeignSource<(K, A)>]()
            outer.setCloseHandler {
                map.removeAll(keepCapacity: false)
                base?.close()
                base = nil
            }
            self.open(outer.calleeContext) {
                base = $0
