         }
     }
     
-    func map<B>(f: A -> B) -> Packet<B> {
+    public 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 nullify<B>() -> Packet<B> { return map { _ in undefined() } }
 
 }
 

    public func zipWithContext() -> Stream<(A, ExecutionContext)> {
        return Streams.pipe([.AllowSync]) { self.zipWith($0.calleeContext) }
    }
    
    public func groupBy<K: Hashable>(f: A -> K) -> Stream<Channel<(K, A)>> {
        return Streams.source() { outer in
            var base: Channel<A>?
            var map = [K: Dispatcher<(K, A)>]()
            outer.setCloseHandler {
                base?.close()
                base = nil
                for e in map.values { e.close() }
            }
            self.open(outer.calleeContext) {
                base = $0

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