| 240 | |
| 241 | |
| 242 | void HttpProxy::stream( |
| 243 | const Owned<Request>& request, |
| 244 | const Future<string>& chunk) |
| 245 | { |
| 246 | CHECK_SOME(pipe); |
| 247 | CHECK_NOTNULL(request.get()); |
| 248 | |
| 249 | http::Pipe::Reader reader = pipe.get(); |
| 250 | |
| 251 | bool finished = false; // Whether we're done streaming. |
| 252 | |
| 253 | if (chunk.isReady()) { |
| 254 | std::ostringstream out; |
| 255 | |
| 256 | if (chunk->empty()) { |
| 257 | // Finished reading. |
| 258 | out << "0\r\n" << "\r\n"; |
| 259 | finished = true; |
| 260 | } else { |
| 261 | out << std::hex << chunk->size() << "\r\n"; |
| 262 | out << chunk.get(); |
| 263 | out << "\r\n"; |
| 264 | |
| 265 | // Keep reading. |
| 266 | reader.read() |
| 267 | .onAny(defer(self(), &Self::stream, request, lambda::_1)); |
| 268 | } |
| 269 | |
| 270 | // Always persist the connection when streaming is not finished. |
| 271 | socket_manager->send( |
| 272 | new DataEncoder(out.str()), |
| 273 | finished ? request->keepAlive : true, |
| 274 | socket); |
| 275 | } else if (chunk.isFailed()) { |
| 276 | VLOG(1) << "Failed to read from stream: " << chunk.failure(); |
| 277 | // TODO(bmahler): Have to close connection if headers were sent! |
| 278 | socket_manager->send(InternalServerError(), *request, socket); |
| 279 | finished = true; |
| 280 | } else { |
| 281 | VLOG(1) << "Failed to read from stream: discarded"; |
| 282 | // TODO(bmahler): Have to close connection if headers were sent! |
| 283 | socket_manager->send(InternalServerError(), *request, socket); |
| 284 | finished = true; |
| 285 | } |
| 286 | |
| 287 | if (finished) { |
| 288 | reader.close(); |
| 289 | pipe = None(); |
| 290 | next(); |
| 291 | } |
| 292 | } |
| 293 | |
| 294 | } // namespace process { |