| 36 | |
| 37 | |
| 38 | void HttpProxy::finalize() |
| 39 | { |
| 40 | // Need to make sure response producers know not to continue to |
| 41 | // create a response (streaming or otherwise). |
| 42 | if (pipe.isSome()) { |
| 43 | http::Pipe::Reader reader = pipe.get(); |
| 44 | reader.close(); |
| 45 | } |
| 46 | pipe = None(); |
| 47 | |
| 48 | while (!items.empty()) { |
| 49 | Item* item = items.front(); |
| 50 | |
| 51 | // Attempt to discard the future. |
| 52 | item->future.discard(); |
| 53 | |
| 54 | // But it might have already been ready. In general, we need to |
| 55 | // wait until this future is potentially ready in order to attempt |
| 56 | // to close a pipe if one exists. |
| 57 | item->future.onReady([](const Response& response) { |
| 58 | // Cleaning up a response (i.e., closing any open Pipes in the |
| 59 | // event Response::type is PIPE). |
| 60 | if (response.type == Response::PIPE) { |
| 61 | CHECK_SOME(response.reader); |
| 62 | http::Pipe::Reader reader = response.reader.get(); // Remove const. |
| 63 | reader.close(); |
| 64 | } |
| 65 | }); |
| 66 | |
| 67 | items.pop(); |
| 68 | delete item; |
| 69 | } |
| 70 | |
| 71 | // Just in case this process gets killed outside of `SocketManager::close`, |
| 72 | // remove the proxy from the socket. |
| 73 | socket_manager->unproxy(socket); |
| 74 | } |
| 75 | |
| 76 | |
| 77 | void HttpProxy::enqueue(const Response& response, const Request& request) |