         }
     }
     
-    public func onSuccess(context: ExecutionContext = QueueContext.futuresDefault, cancelToken: CancelToken = CancelToken(), success: (T) -> Void){
+    public func onSuccess(context: ExecutionContext = QueueContext.futuresDefault, cancelToken: CancelToken = CancelToken(), success: @escaping (T) -> Void){
         onComplete(context: context, cancelToken: cancelToken) { result in
             switch result {
             case .success(let value):

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

        }
    }
    
    public func onFailure(context: ExecutionContext = QueueContext.futuresDefault, cancelToken: CancelToken = CancelToken(), failure: @escaping (Swift.Error) -> Void) {
        onComplete(context: context, cancelToken: cancelToken) { result in
            switch result {
            case .failure(let error):
