| 155 | Result cancel() { return operation.cancel(); } |
| 156 | |
| 157 | Result start(const HttpClientRequest& request, HttpClientResponse& response, |
| 158 | T_AsyncBuffersPool* requestBodyPool = nullptr) |
| 159 | { |
| 160 | SC_TRY_MSG(initialized, "HttpClientAsyncT: not initialized"); |
| 161 | if (request.body.isStreamed()) |
| 162 | { |
| 163 | SC_TRY_MSG(requestBodyPool != nullptr, "HttpClientAsyncT: streamed request body requires buffers pool"); |
| 164 | } |
| 165 | |
| 166 | requestBodyBuffersPool = requestBodyPool; |
| 167 | resetRequestBodyState(); |
| 168 | |
| 169 | responseBodyStream.setReadQueue(responseReadQueue); |
| 170 | SC_TRY(responseBodyStream.init(responseBuffersPool)); |
| 171 | SC_TRY(responseBodyStream.start()); |
| 172 | |
| 173 | if (request.body.isStreamed()) |
| 174 | { |
| 175 | SC_TRY(requestBodySink.init(*requestBodyBuffersPool, requestWriteQueue, *this)); |
| 176 | } |
| 177 | |
| 178 | if (not wakeUpStarted) |
| 179 | { |
| 180 | SC_TRY(wakeUp.start(*eventLoop)); |
| 181 | wakeUpStarted = true; |
| 182 | } |
| 183 | |
| 184 | HttpClientRequest asyncRequest = request; |
| 185 | if (asyncRequest.body.isStreamed()) |
| 186 | { |
| 187 | asyncRequest.body.provider = this; |
| 188 | } |
| 189 | |
| 190 | const Result res = operation.start(asyncRequest, response, this); |
| 191 | if (not res) |
| 192 | { |
| 193 | if (request.body.isStreamed()) |
| 194 | { |
| 195 | requestBodySink.destroy(); |
| 196 | } |
| 197 | responseBodyStream.destroy(); |
| 198 | return res; |
| 199 | } |
| 200 | return Result(true); |
| 201 | } |
| 202 | |
| 203 | [[nodiscard]] T_AsyncReadableStream& getResponseBodyStream() { return responseBodyStream; } |
| 204 | [[nodiscard]] T_AsyncWritableStream& getRequestBodySink() { return requestBodySink; } |