         return stream
     }
     
-    public func recoverWith(context: ExecutionContext = QueueContext.futuresDefault, cancelToken: CancelToken = CancelToken(), recovery: (Swift.Error) throws  -> Future<T>) -> FutureStream<T> {
+    public func recoverWith(context: ExecutionContext = QueueContext.futuresDefault, cancelToken: CancelToken = CancelToken(), recovery: @escaping (Swift.Error) throws  -> Future<T>) -> FutureStream<T> {
         let stream = FutureStream<T>(capacity: capacity)
         onComplete(context: context, cancelToken: cancelToken) { result in
             switch result {

        return stream
    }
    
    public func forEach(context: ExecutionContext = QueueContext.futuresDefault, cancelToken: CancelToken = CancelToken(), apply: (T) -> Void) {
        onComplete(context: context, cancelToken: cancelToken) { result in
            result.forEach(apply)
        }
    }

    public func andThen(context: ExecutionContext = QueueContext.futuresDefault, cancelToken: CancelToken = CancelToken(), success: (T) -> Void) -> FutureStream<T> {
        let stream = FutureStream<T>(capacity: capacity)
        stream.onSuccess(context: context, cancelToken: cancelToken, success: success)
        onComplete(context: context, cancelToken: cancelToken) { result in

        }
    }

    public func mapError(_ mapping: (Swift.Error) -> Swift.Error) -> Try<T> {
        switch self {
        case .success(let value):
            return Try(value)
