| 1145 | } |
| 1146 | |
| 1147 | static int on_xrt_post_stream_chunk(const char* data, size_t dataLen, size_t, size_t, void* userData) { |
| 1148 | auto context = r_cast<XrtPostStreamContext*>(userData); |
| 1149 | if (!context || !data || dataLen == 0 || !context->partCallback || !*context->partCallback) { |
| 1150 | return context && context->request && context->request->cancelling.load(std::memory_order_relaxed); |
| 1151 | } |
| 1152 | std::string body(data, dataLen); |
| 1153 | context->completion->queuedChunks.fetch_add(1, std::memory_order_relaxed); |
| 1154 | context->completion->queuedBytes.fetch_add(dataLen, std::memory_order_relaxed); |
| 1155 | SharedApplication.invokeInLogic([request = context->request, partCallback = context->partCallback, stopped = context->stopped, completion = context->completion, body = std::move(body)]() { |
| 1156 | completion->handledChunks.fetch_add(1, std::memory_order_relaxed); |
| 1157 | completion->handledBytes.fetch_add(body.size(), std::memory_order_relaxed); |
| 1158 | if (*stopped) { |
| 1159 | try_finish_xrt_post_stream_in_logic(completion); |
| 1160 | return; |
| 1161 | } |
| 1162 | if ((*partCallback)(body)) { |
| 1163 | *stopped = true; |
| 1164 | request->cancelling.store(true, std::memory_order_relaxed); |
| 1165 | } |
| 1166 | try_finish_xrt_post_stream_in_logic(completion); |
| 1167 | }); |
| 1168 | return context->stopped->load(std::memory_order_relaxed) || |
| 1169 | (context->request && context->request->cancelling.load(std::memory_order_relaxed)); |
| 1170 | } |
| 1171 | |
| 1172 | struct XrtDownloadStreamContext { |
| 1173 | std::shared_ptr<HttpRequestState> request; |
nothing calls this directly
no test coverage detected