Task refactoring (#47)

artemredkin · web-flow · commit 34e3403ccaf8 · 2019-06-24T16:30:03.000+03:00
* refactor Task to encapsulate promise
diff --git a/Sources/NIOHTTPClient/HTTPHandler.swift b/Sources/NIOHTTPClient/HTTPHandler.swift
@@ -236,31 +236,28 @@ internal extension URL {
     }
 }
 
-public extension HTTPClient {
-    final class Task<Response> {
+extension HTTPClient {
+    public final class Task<Response> {
         public let eventLoop: EventLoop
-        let future: EventLoopFuture<Response>
+        let promise: EventLoopPromise<Response>
 
         private var channel: Channel?
         private var cancelled: Bool
         private let lock: Lock
 
-        init(eventLoop: EventLoop, future: EventLoopFuture<Response>) {
+        public init(eventLoop: EventLoop) {
             self.eventLoop = eventLoop
-            self.future = future
+            self.promise = eventLoop.makePromise()
             self.cancelled = false
             self.lock = Lock()
         }
 
-        func setChannel(_ channel: Channel) -> Channel {
-            return self.lock.withLock {
-                self.channel = channel
-                return channel
-            }
+        public var futureResult: EventLoopFuture<Response> {
+            return self.promise.futureResult
         }
 
         public func wait() throws -> Response {
-            return try self.future.wait()
+            return try self.promise.futureResult.wait()
         }
 
         public func cancel() {
@@ -272,8 +269,19 @@ public extension HTTPClient {
             }
         }
 
-        public func cascade(promise: EventLoopPromise<Response>) {
-            self.future.cascade(to: promise)
+        func setChannel(_ channel: Channel) -> Channel {
+            return self.lock.withLock {
+                self.channel = channel
+                return channel
+            }
+        }
+
+        func succeed(_ value: Response) {
+            self.promise.succeed(value)
+        }
+
+        func fail(_ error: Error) {
+            self.promise.fail(error)
         }
     }
 }
@@ -296,17 +304,15 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler
 
     let task: HTTPClient.Task<T.Response>
     let delegate: T
-    let promise: EventLoopPromise<T.Response>
     let redirectHandler: RedirectHandler<T.Response>?
 
     var state: State = .idle
     var pendingRead = false
     var mayRead = true
 
-    init(task: HTTPClient.Task<T.Response>, delegate: T, promise: EventLoopPromise<T.Response>, redirectHandler: RedirectHandler<T.Response>?) {
+    init(task: HTTPClient.Task<T.Response>, delegate: T, redirectHandler: RedirectHandler<T.Response>?) {
         self.task = task
         self.delegate = delegate
-        self.promise = promise
         self.redirectHandler = redirectHandler
     }
 
@@ -347,13 +353,13 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler
                 self.delegate.didSendRequest(task: self.task)
 
                 let channel = context.channel
-                self.promise.futureResult.whenComplete { _ in
+                self.task.futureResult.whenComplete { _ in
                     channel.close(promise: nil)
                 }
             case .failure(let error):
                 self.state = .end
                 self.delegate.didReceiveError(task: self.task, error)
-                self.promise.fail(error)
+                self.task.fail(error)
                 context.close(promise: nil)
             }
         }
@@ -410,14 +416,14 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler
             switch self.state {
             case .redirected(let head, let redirectURL):
                 self.state = .end
-                self.redirectHandler?.redirect(status: head.status, to: redirectURL, promise: self.promise)
+                self.redirectHandler?.redirect(status: head.status, to: redirectURL, promise: self.task.promise)
                 context.close(promise: nil)
             default:
                 self.state = .end
                 do {
-                    self.promise.succeed(try self.delegate.didFinishRequest(task: self.task))
+                    self.task.succeed(try self.delegate.didFinishRequest(task: self.task))
                 } catch {
-                    self.promise.fail(error)
+                    self.task.fail(error)
                 }
             }
         }
@@ -433,7 +439,7 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler
         case .failure(let error):
             self.state = .end
             self.delegate.didReceiveError(task: self.task, error)
-            self.promise.fail(error)
+            self.task.fail(error)
         }
     }
 
@@ -442,12 +448,12 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler
             self.state = .end
             let error = HTTPClientError.readTimeout
             self.delegate.didReceiveError(task: self.task, error)
-            self.promise.fail(error)
+            self.task.fail(error)
         } else if (event as? TaskCancelEvent) != nil {
             self.state = .end
             let error = HTTPClientError.cancelled
             self.delegate.didReceiveError(task: self.task, error)
-            self.promise.fail(error)
+            self.task.fail(error)
         } else {
             context.fireUserInboundEventTriggered(event)
         }
@@ -461,7 +467,7 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler
             self.state = .end
             let error = HTTPClientError.remoteConnectionClosed
             self.delegate.didReceiveError(task: self.task, error)
-            self.promise.fail(error)
+            self.task.fail(error)
         }
     }
 
@@ -476,12 +482,12 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler
             default:
                 self.state = .end
                 self.delegate.didReceiveError(task: self.task, error)
-                self.promise.fail(error)
+                self.task.fail(error)
             }
         default:
             self.state = .end
             self.delegate.didReceiveError(task: self.task, error)
-            self.promise.fail(error)
+            self.task.fail(error)
         }
     }
 }
@@ -556,6 +562,6 @@ internal struct RedirectHandler<T> {
             request.headers.remove(name: "Proxy-Authorization")
         }
 
-        return self.execute(request).cascade(promise: promise)
+        return self.execute(request).futureResult.cascade(to: promise)
     }
 }
diff --git a/Sources/NIOHTTPClient/SwiftNIOHTTP.swift b/Sources/NIOHTTPClient/SwiftNIOHTTP.swift
@@ -105,13 +105,12 @@ public class HTTPClient {
 
     public func execute(request: Request, timeout: Timeout? = nil) -> EventLoopFuture<Response> {
         let accumulator = ResponseAccumulator(request: request)
-        return self.execute(request: request, delegate: accumulator, timeout: timeout).future
+        return self.execute(request: request, delegate: accumulator, timeout: timeout).futureResult
     }
 
     public func execute<T: HTTPClientResponseDelegate>(request: Request, delegate: T, timeout: Timeout? = nil) -> Task<T.Response> {
         let timeout = timeout ?? configuration.timeout
         let eventLoop = self.eventLoopGroup.next()
-        let promise: EventLoopPromise<T.Response> = eventLoop.makePromise()
 
         let redirectHandler: RedirectHandler<T.Response>?
         if self.configuration.followRedirects {
@@ -122,7 +121,7 @@ public class HTTPClient {
             redirectHandler = nil
         }
 
-        let task = Task(eventLoop: eventLoop, future: promise.futureResult)
+        let task = Task<T.Response>(eventLoop: eventLoop)
 
         var bootstrap = ClientBootstrap(group: self.eventLoopGroup)
             .channelOption(ChannelOptions.socket(SocketOptionLevel(IPPROTO_TCP), TCP_NODELAY), value: 1)
@@ -143,7 +142,7 @@ public class HTTPClient {
                         return channel.eventLoop.makeSucceededFuture(())
                     }
                 }.flatMap {
-                    let taskHandler = TaskHandler(task: task, delegate: delegate, promise: promise, redirectHandler: redirectHandler)
+                    let taskHandler = TaskHandler(task: task, delegate: delegate, redirectHandler: redirectHandler)
                     return channel.pipeline.addHandler(taskHandler)
                 }
             }
@@ -161,7 +160,7 @@ public class HTTPClient {
                 channel.writeAndFlush(request)
             }
             .whenFailure { error in
-                promise.fail(error)
+                task.fail(error)
             }
 
         return task
diff --git a/Tests/NIOHTTPClientTests/SwiftNIOHTTPTests.swift b/Tests/NIOHTTPClientTests/SwiftNIOHTTPTests.swift
@@ -16,7 +16,7 @@ import Foundation
 import NIO
 import NIOConcurrencyHelpers
 import NIOFoundationCompat
-@testable import NIOHTTP1
+import NIOHTTP1
 @testable import NIOHTTPClient
 import NIOSSL
 import XCTest
@@ -40,11 +40,10 @@ class SwiftHTTPTests: XCTestCase {
     func testHTTPPartsHandler() throws {
         let channel = EmbeddedChannel()
         let recorder = RecordingHandler<HTTPClientResponsePart, HTTPClientRequestPart>()
-        let promise: EventLoopPromise<Void> = channel.eventLoop.makePromise()
-        let task = Task(eventLoop: channel.eventLoop, future: promise.futureResult)
+        let task = Task<Void>(eventLoop: channel.eventLoop)
 
         try channel.pipeline.addHandler(recorder).wait()
-        try channel.pipeline.addHandler(TaskHandler(task: task, delegate: TestHTTPDelegate(), promise: promise, redirectHandler: nil)).wait()
+        try channel.pipeline.addHandler(TaskHandler(task: task, delegate: TestHTTPDelegate(), redirectHandler: nil)).wait()
 
         var request = try Request(url: "http://localhost/get")
         request.headers.add(name: "X-Test-Header", value: "X-Test-Value")
@@ -68,9 +67,8 @@ class SwiftHTTPTests: XCTestCase {
     func testHTTPPartsHandlerMultiBody() throws {
         let channel = EmbeddedChannel()
         let delegate = TestHTTPDelegate()
-        let promise: EventLoopPromise<Void> = channel.eventLoop.makePromise()
-        let task = Task(eventLoop: channel.eventLoop, future: promise.futureResult)
-        let handler = TaskHandler(task: task, delegate: delegate, promise: promise, redirectHandler: nil)
+        let task = Task<Void>(eventLoop: channel.eventLoop)
+        let handler = TaskHandler(task: task, delegate: delegate, redirectHandler: nil)
 
         try channel.pipeline.addHandler(handler).wait()
 
@@ -357,7 +355,7 @@ class SwiftHTTPTests: XCTestCase {
                 let delegate = CopyingDelegate { part in
                     writer.write(.byteBuffer(part))
                 }
-                return httpClient.execute(request: request, delegate: delegate).future
+                return httpClient.execute(request: request, delegate: delegate).futureResult
             } catch {
                 return httpClient.eventLoopGroup.next().makeFailedFuture(error)
             }
@@ -393,7 +391,7 @@ class SwiftHTTPTests: XCTestCase {
                 let delegate = CopyingDelegate { _ in
                     httpClient.eventLoopGroup.next().makeFailedFuture(HTTPClientError.invalidProxyResponse)
                 }
-                return httpClient.execute(request: request, delegate: delegate).future
+                return httpClient.execute(request: request, delegate: delegate).futureResult
             } catch {
                 return httpClient.eventLoopGroup.next().makeFailedFuture(error)
             }
@@ -442,7 +440,7 @@ class SwiftHTTPTests: XCTestCase {
 
         let request = try Request(url: "http://localhost:\(httpBin.port)/custom")
         let delegate = BackpressureTestDelegate(promise: httpClient.eventLoopGroup.next().makePromise())
-        let future = httpClient.execute(request: request, delegate: delegate).future
+        let future = httpClient.execute(request: request, delegate: delegate).futureResult
 
         let channel = try promise.futureResult.wait()
 

Original file line number	Diff line number	Diff line change
`@@ -236,31 +236,28 @@ internal extension URL {`
`236`	`236`	`}`
`237`	`237`	`}`
`238`	`238`
`239`		`-public extension HTTPClient {`
`240`		`- final class Task<Response> {`
	`239`	`+extension HTTPClient {`
	`240`	`+ public final class Task<Response> {`
`241`	`241`	`public let eventLoop: EventLoop`
`242`		`- let future: EventLoopFuture<Response>`
	`242`	`+ let promise: EventLoopPromise<Response>`
`243`	`243`
`244`	`244`	`private var channel: Channel?`
`245`	`245`	`private var cancelled: Bool`
`246`	`246`	`private let lock: Lock`
`247`	`247`
`248`		`- init(eventLoop: EventLoop, future: EventLoopFuture<Response>) {`
	`248`	`+ public init(eventLoop: EventLoop) {`
`249`	`249`	`self.eventLoop = eventLoop`
`250`		`- self.future = future`
	`250`	`+ self.promise = eventLoop.makePromise()`
`251`	`251`	`self.cancelled = false`
`252`	`252`	`self.lock = Lock()`
`253`	`253`	`}`
`254`	`254`
`255`		`- func setChannel(_ channel: Channel) -> Channel {`
`256`		`- return self.lock.withLock {`
`257`		`- self.channel = channel`
`258`		`- return channel`
`259`		`- }`
	`255`	`+ public var futureResult: EventLoopFuture<Response> {`
	`256`	`+ return self.promise.futureResult`
`260`	`257`	`}`
`261`	`258`
`262`	`259`	`public func wait() throws -> Response {`
`263`		`- return try self.future.wait()`
	`260`	`+ return try self.promise.futureResult.wait()`
`264`	`261`	`}`
`265`	`262`
`266`	`263`	`public func cancel() {`
`@@ -272,8 +269,19 @@ public extension HTTPClient {`
`272`	`269`	`}`
`273`	`270`	`}`
`274`	`271`
`275`		`- public func cascade(promise: EventLoopPromise<Response>) {`
`276`		`- self.future.cascade(to: promise)`
	`272`	`+ func setChannel(_ channel: Channel) -> Channel {`
	`273`	`+ return self.lock.withLock {`
	`274`	`+ self.channel = channel`
	`275`	`+ return channel`
	`276`	`+ }`
	`277`	`+ }`
	`278`	`+`
	`279`	`+ func succeed(_ value: Response) {`
	`280`	`+ self.promise.succeed(value)`
	`281`	`+ }`
	`282`	`+`
	`283`	`+ func fail(_ error: Error) {`
	`284`	`+ self.promise.fail(error)`
`277`	`285`	`}`
`278`	`286`	`}`
`279`	`287`	`}`
`@@ -296,17 +304,15 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler`
`296`	`304`
`297`	`305`	`let task: HTTPClient.Task<T.Response>`
`298`	`306`	`let delegate: T`
`299`		`- let promise: EventLoopPromise<T.Response>`
`300`	`307`	`let redirectHandler: RedirectHandler<T.Response>?`
`301`	`308`
`302`	`309`	`var state: State = .idle`
`303`	`310`	`var pendingRead = false`
`304`	`311`	`var mayRead = true`
`305`	`312`
`306`		`- init(task: HTTPClient.Task<T.Response>, delegate: T, promise: EventLoopPromise<T.Response>, redirectHandler: RedirectHandler<T.Response>?) {`
	`313`	`+ init(task: HTTPClient.Task<T.Response>, delegate: T, redirectHandler: RedirectHandler<T.Response>?) {`
`307`	`314`	`self.task = task`
`308`	`315`	`self.delegate = delegate`
`309`		`- self.promise = promise`
`310`	`316`	`self.redirectHandler = redirectHandler`
`311`	`317`	`}`
`312`	`318`
`@@ -347,13 +353,13 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler`
`347`	`353`	`self.delegate.didSendRequest(task: self.task)`
`348`	`354`
`349`	`355`	`let channel = context.channel`
`350`		`- self.promise.futureResult.whenComplete { _ in`
	`356`	`+ self.task.futureResult.whenComplete { _ in`
`351`	`357`	`channel.close(promise: nil)`
`352`	`358`	`}`
`353`	`359`	`case .failure(let error):`
`354`	`360`	`self.state = .end`
`355`	`361`	`self.delegate.didReceiveError(task: self.task, error)`
`356`		`- self.promise.fail(error)`
	`362`	`+ self.task.fail(error)`
`357`	`363`	`context.close(promise: nil)`
`358`	`364`	`}`
`359`	`365`	`}`
`@@ -410,14 +416,14 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler`
`410`	`416`	`switch self.state {`
`411`	`417`	`case .redirected(let head, let redirectURL):`
`412`	`418`	`self.state = .end`
`413`		`- self.redirectHandler?.redirect(status: head.status, to: redirectURL, promise: self.promise)`
	`419`	`+ self.redirectHandler?.redirect(status: head.status, to: redirectURL, promise: self.task.promise)`
`414`	`420`	`context.close(promise: nil)`
`415`	`421`	`default:`
`416`	`422`	`self.state = .end`
`417`	`423`	`do {`
`418`		`- self.promise.succeed(try self.delegate.didFinishRequest(task: self.task))`
	`424`	`+ self.task.succeed(try self.delegate.didFinishRequest(task: self.task))`
`419`	`425`	`} catch {`
`420`		`- self.promise.fail(error)`
	`426`	`+ self.task.fail(error)`
`421`	`427`	`}`
`422`	`428`	`}`
`423`	`429`	`}`
`@@ -433,7 +439,7 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler`
`433`	`439`	`case .failure(let error):`
`434`	`440`	`self.state = .end`
`435`	`441`	`self.delegate.didReceiveError(task: self.task, error)`
`436`		`- self.promise.fail(error)`
	`442`	`+ self.task.fail(error)`
`437`	`443`	`}`
`438`	`444`	`}`
`439`	`445`
`@@ -442,12 +448,12 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler`
`442`	`448`	`self.state = .end`
`443`	`449`	`let error = HTTPClientError.readTimeout`
`444`	`450`	`self.delegate.didReceiveError(task: self.task, error)`
`445`		`- self.promise.fail(error)`
	`451`	`+ self.task.fail(error)`
`446`	`452`	`} else if (event as? TaskCancelEvent) != nil {`
`447`	`453`	`self.state = .end`
`448`	`454`	`let error = HTTPClientError.cancelled`
`449`	`455`	`self.delegate.didReceiveError(task: self.task, error)`
`450`		`- self.promise.fail(error)`
	`456`	`+ self.task.fail(error)`
`451`	`457`	`} else {`
`452`	`458`	`context.fireUserInboundEventTriggered(event)`
`453`	`459`	`}`
`@@ -461,7 +467,7 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler`
`461`	`467`	`self.state = .end`
`462`	`468`	`let error = HTTPClientError.remoteConnectionClosed`
`463`	`469`	`self.delegate.didReceiveError(task: self.task, error)`
`464`		`- self.promise.fail(error)`
	`470`	`+ self.task.fail(error)`
`465`	`471`	`}`
`466`	`472`	`}`
`467`	`473`
`@@ -476,12 +482,12 @@ internal class TaskHandler<T: HTTPClientResponseDelegate>: ChannelInboundHandler`
`476`	`482`	`default:`
`477`	`483`	`self.state = .end`
`478`	`484`	`self.delegate.didReceiveError(task: self.task, error)`
`479`		`- self.promise.fail(error)`
	`485`	`+ self.task.fail(error)`
`480`	`486`	`}`
`481`	`487`	`default:`
`482`	`488`	`self.state = .end`
`483`	`489`	`self.delegate.didReceiveError(task: self.task, error)`
`484`		`- self.promise.fail(error)`
	`490`	`+ self.task.fail(error)`
`485`	`491`	`}`
`486`	`492`	`}`
`487`	`493`	`}`
`@@ -556,6 +562,6 @@ internal struct RedirectHandler<T> {`
`556`	`562`	`request.headers.remove(name: "Proxy-Authorization")`
`557`	`563`	`}`
`558`	`564`
`559`		`- return self.execute(request).cascade(promise: promise)`
	`565`	`+ return self.execute(request).futureResult.cascade(to: promise)`
`560`	`566`	`}`
`561`	`567`	`}`
Original file line number	Diff line number	Diff line change
`@@ -105,13 +105,12 @@ public class HTTPClient {`
`105`	`105`
`106`	`106`	`public func execute(request: Request, timeout: Timeout? = nil) -> EventLoopFuture<Response> {`
`107`	`107`	`let accumulator = ResponseAccumulator(request: request)`
`108`		`- return self.execute(request: request, delegate: accumulator, timeout: timeout).future`
	`108`	`+ return self.execute(request: request, delegate: accumulator, timeout: timeout).futureResult`
`109`	`109`	`}`
`110`	`110`
`111`	`111`	`public func execute<T: HTTPClientResponseDelegate>(request: Request, delegate: T, timeout: Timeout? = nil) -> Task<T.Response> {`
`112`	`112`	`let timeout = timeout ?? configuration.timeout`
`113`	`113`	`let eventLoop = self.eventLoopGroup.next()`
`114`		`- let promise: EventLoopPromise<T.Response> = eventLoop.makePromise()`
`115`	`114`
`116`	`115`	`let redirectHandler: RedirectHandler<T.Response>?`
`117`	`116`	`if self.configuration.followRedirects {`
`@@ -122,7 +121,7 @@ public class HTTPClient {`
`122`	`121`	`redirectHandler = nil`
`123`	`122`	`}`
`124`	`123`
`125`		`- let task = Task(eventLoop: eventLoop, future: promise.futureResult)`
	`124`	`+ let task = Task<T.Response>(eventLoop: eventLoop)`
`126`	`125`
`127`	`126`	`var bootstrap = ClientBootstrap(group: self.eventLoopGroup)`
`128`	`127`	`.channelOption(ChannelOptions.socket(SocketOptionLevel(IPPROTO_TCP), TCP_NODELAY), value: 1)`
`@@ -143,7 +142,7 @@ public class HTTPClient {`
`143`	`142`	`return channel.eventLoop.makeSucceededFuture(())`
`144`	`143`	`}`
`145`	`144`	`}.flatMap {`
`146`		`- let taskHandler = TaskHandler(task: task, delegate: delegate, promise: promise, redirectHandler: redirectHandler)`
	`145`	`+ let taskHandler = TaskHandler(task: task, delegate: delegate, redirectHandler: redirectHandler)`
`147`	`146`	`return channel.pipeline.addHandler(taskHandler)`
`148`	`147`	`}`
`149`	`148`	`}`
`@@ -161,7 +160,7 @@ public class HTTPClient {`
`161`	`160`	`channel.writeAndFlush(request)`
`162`	`161`	`}`
`163`	`162`	`.whenFailure { error in`
`164`		`- promise.fail(error)`
	`163`	`+ task.fail(error)`
`165`	`164`	`}`
`166`	`165`
`167`	`166`	`return task`