draft for streaming el fixes

Sources/AsyncHTTPClient/HTTPHandler.swift

@@ -765,12 +765,32 @@ extension TaskHandler: ChannelDuplexHandler {
             return context.eventLoop.makeSucceededFuture(())
         }
 
-        return body.stream(HTTPClient.Body.StreamWriter { part in
-            context.eventLoop.assertInEventLoop()
-            return context.writeAndFlush(self.wrapOutboundOut(.body(part))).map {
-                self.callOutToDelegateFireAndForget(value: part, self.delegate.didSendRequestPart)
+        func doIt() -> EventLoopFuture<Void> {
+            return body.stream(HTTPClient.Body.StreamWriter { part in
+                let promise = self.task.eventLoop.makePromise(of: Void.self)
+                // All writes have to be switched to the channel EL if channel and task ELs differ
+                if context.eventLoop.inEventLoop {
+                    context.writeAndFlush(self.wrapOutboundOut(.body(part)), promise: promise)
+                } else {
+                    context.eventLoop.execute {
+                        context.writeAndFlush(self.wrapOutboundOut(.body(part)), promise: promise)
+                    }
+                }
+
+                return promise.futureResult.map {
+                    self.callOutToDelegateFireAndForget(value: part, self.delegate.didSendRequestPart)
+                }
+            })
+        }
+
+        // Callout to the user to start body streaming should be on task EL
+        if self.task.eventLoop.inEventLoop {
+            return doIt()
+        } else {
+            return self.task.eventLoop.flatSubmit {
+                doIt()
             }
-        })
+        }
     }
 
     public func read(context: ChannelHandlerContext) {

Tests/AsyncHTTPClientTests/HTTPClientTests.swift

@@ -447,22 +447,40 @@ class HTTPClientTests: XCTestCase {
     }
 
     func testUploadStreaming() throws {
+        let group = getDefaultEventLoopGroup(numberOfThreads: 4)
+        defer {
+            XCTAssertNoThrow(try group.syncShutdownGracefully())
+        }
+
         let httpBin = HTTPBin()
-        let httpClient = HTTPClient(eventLoopGroupProvider: .shared(self.clientGroup))
+        let httpClient = HTTPClient(eventLoopGroupProvider: .shared(group))
         defer {
             XCTAssertNoThrow(try httpClient.syncShutdown())
             XCTAssertNoThrow(try httpBin.shutdown())
         }
 
+        let el1 = group.next()
+        let el2 = group.next()
+        XCTAssertFalse(el1 === el2)
+
         let body: HTTPClient.Body = .stream(length: 8) { writer in
+            XCTAssert(el1.inEventLoop)
             let buffer = ByteBuffer.of(string: "1234")
             return writer.write(.byteBuffer(buffer)).flatMap {
+                XCTAssert(el1.inEventLoop)
                 let buffer = ByteBuffer.of(string: "4321")
                 return writer.write(.byteBuffer(buffer))
             }
         }
 
-        let response = try httpClient.post(url: "http://localhost:\(httpBin.port)/post", body: body).wait()
+        do {
+            // Pre-populate pool with a connection on a different EL
+            let request = try HTTPClient.Request(url: "http://localhost:\(httpBin.port)/get", method: .GET)
+            XCTAssertNoThrow(try httpClient.execute(request: request, delegate: ResponseAccumulator(request: request), eventLoop: .delegateAndChannel(on: el2)).wait())
+        }
+
+        let request = try HTTPClient.Request(url: "http://localhost:\(httpBin.port)/post", method: .POST, body: body)
+        let response = try httpClient.execute(request: request, delegate: ResponseAccumulator(request: request), eventLoop: .delegate(on: el1)).wait()
         let bytes = response.body.flatMap { $0.getData(at: 0, length: $0.readableBytes) }
         let data = try JSONDecoder().decode(RequestInfo.self, from: bytes!)
 
@@ -1753,10 +1771,7 @@ class HTTPClientTests: XCTestCase {
         let bodyPromises = (0..<16).map { _ in group.next().makePromise(of: ByteBuffer.self) }
         let endPromise = group.next().makePromise(of: Void.self)
         let sentOffAllBodyPartsPromise = group.next().makePromise(of: Void.self)
-        // Because of https://github.com/swift-server/async-http-client/issues/200 we also need to pull off a terrible
-        // hack and get the internal EventLoop out :(. Once the bug is fixed, this promise should only get the
-        // StreamWriter.
-        let streamWriterPromise = group.next().makePromise(of: (EventLoop, HTTPClient.Body.StreamWriter).self)
+        let streamWriterPromise = group.next().makePromise(of: HTTPClient.Body.StreamWriter.self)
 
         func makeServer() -> Channel? {
             return try? ServerBootstrap(group: group)
@@ -1781,12 +1796,7 @@ class HTTPClientTests: XCTestCase {
                                            method: .POST,
                                            headers: ["transfer-encoding": "chunked"],
                                            body: .stream { streamWriter in
-                                               // Due to https://github.com/swift-server/async-http-client/issues/200
-                                               // we also need to pull off a terrible hack and get the internal
-                                               // EventLoop out :(. Once the bug is fixed, this promise should only get
-                                               // the StreamWriter.
-                                               let currentEL = MultiThreadedEventLoopGroup.currentEventLoop! // HACK!!
-                                               streamWriterPromise.succeed((currentEL, streamWriter))
+                                               streamWriterPromise.succeed(streamWriter)
                                                return sentOffAllBodyPartsPromise.futureResult
             })
         }
@@ -1811,9 +1821,7 @@ class HTTPClientTests: XCTestCase {
             buffer.clear()
             buffer.writeString(String(bodyChunkNumber, radix: 16))
             XCTAssertEqual(1, buffer.readableBytes)
-            XCTAssertNoThrow(try streamWriter.0.flatSubmit {
-                streamWriter.1.write(.byteBuffer(buffer))
-            }.wait())
+            XCTAssertNoThrow(try streamWriter.write(.byteBuffer(buffer)).wait())
             XCTAssertNoThrow(XCTAssertEqual(buffer, try bodyPromises[bodyChunkNumber].futureResult.wait()))
         }
         sentOffAllBodyPartsPromise.succeed(())