 /// `upstream |> startAsync(queue) ~> { ... }` is same as `dispatch_async(queue, { upstream ~> {...} })`,
 /// but it guarantees the `upstream` to start on target `queue` if not started yet.
 ///
-public func startAsync<T>(queue: DispatchQueue) -> (upstream: Stream<T>) -> Stream<T>
+public func startAsync<T>(_ queue: DispatchQueue) -> (upstream: Stream<T>) -> Stream<T>
 {
     return { (upstream: Stream<T>) in
         return Stream<T> { progress, fulfill, reject, configure in

    }
}

public func distinct<H: Hashable>(upstream: Stream<H>) -> Stream<H>
{
    return Stream<H> { progress, fulfill, reject, configure in
        


/// useful for injecting side-effects
/// a.k.a Rx.do, tap
public func peek<T>(_ peekClosure: (T) -> Void) -> (upstream: Stream<T>) -> Stream<T>
{
    return { (upstream: Stream<T>) in
        return Stream<T> { progress, fulfill, reject, configure in
