AHC holds onto its body stream writer even after the request is complete

[weissi]

AHC seems to hold onto its body streaming delegate closure even after the request is complete. That means it's pretty much impossible to use it for bidi streaming without leaking it.

test case

    func testStuff() throws {
        let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
        defer {
            XCTAssertNoThrow(try group.syncShutdownGracefully())
        }

        let server = try ServerBootstrap(group: group)
            .childChannelInitializer { channel in
                channel.eventLoop.scheduleTask(in: .milliseconds(500)) {
                    channel.writeAndFlush(ByteBuffer(string: "HTTP/1.1 200 ok\r\ncontent-length: 2\r\n\r\nok")).map {
                        channel.close(promise: nil)
                    }.whenComplete { result in
                        print("server write complete", result)
                    }
                }
                return channel.eventLoop.makeSucceededFuture(())
            }
            .bind(host: "127.1", port: 0).wait()
        defer {
            XCTAssertNoThrow(try server.close().wait())
        }
        var request = try HTTPClient.Request(url: "http://\(server.localAddress!.ipAddress!):\(server.localAddress!.port!)/",
                                             method: .POST)
        let writerPromise = self.group.any().makePromise(of: HTTPClient.Body.StreamWriter.self)
        let donePromise = self.group.any().makePromise(of: Void.self)
        class LeakDetector {
            init() {
                print("INIT")
            }

            deinit {
                print("DEINIT")
            }
        }
        var leakDetector: LeakDetector? = LeakDetector()
        request.body = .stream { [leakDetector] writer in
            _ = leakDetector
            writerPromise.succeed(writer)
            return donePromise.futureResult
        }
        weak var weakLeak: LeakDetector? = leakDetector
        leakDetector = nil

        let resultFuture = self.httpClient.execute(request: request)
        request.body = nil // this is pointless but whatever
        writerPromise.futureResult.whenSuccess { writer in
            writer.write(.byteBuffer(ByteBuffer(string: "hello"))).map {
                print("written")
            }.cascade(to: donePromise)
        }
        print("HTTP sent")

        let result = try resultFuture.wait()

        XCTAssertEqual(.ok, result.status)
        XCTAssertEqual("ok", result.body.map { String(buffer: $0) } ?? "none")
        print("HTTP done", result)
        while weakLeak != nil {
            print("not nil")
            sleep(1)
        }
    }

[weissi]

Presumably the easiest fix is to stop holding onto Request.body in the RequestBag type once we've started sending the body.

The retain cycle is something like RequestBag -> Request -> (if it captures a promise to be fulfilled with BodyStreamWiter) BodyStreamWriter -> RequestBag