| 254 | } |
| 255 | |
| 256 | int HttpMessage::OnBody(const char *at, const size_t length) { |
| 257 | if (_vmsgbuilder) { |
| 258 | if (_stage != HTTP_ON_BODY) { |
| 259 | // only add prefix at first entry. |
| 260 | *_vmsgbuilder << "\n<\n"; |
| 261 | } |
| 262 | if (_read_body_progressively && |
| 263 | // If the status_code is non-OK, the body is likely to be the error |
| 264 | // description which is very helpful for debugging. Otherwise |
| 265 | // the body is probably streaming data which is too long to print. |
| 266 | header().status_code() == HTTP_STATUS_OK) { |
| 267 | LOG(INFO) << '\n' << _vmsgbuilder->buf(); |
| 268 | _vmsgbuilder.reset(NULL); |
| 269 | } else { |
| 270 | if (_vbodylen < (size_t)FLAGS_http_verbose_max_body_length) { |
| 271 | int plen = std::min(length, (size_t)FLAGS_http_verbose_max_body_length |
| 272 | - _vbodylen); |
| 273 | std::string str = butil::ToPrintableString( |
| 274 | at, plen, std::numeric_limits<size_t>::max()); |
| 275 | _vmsgbuilder->write(str.data(), str.size()); |
| 276 | } |
| 277 | _vbodylen += length; |
| 278 | } |
| 279 | } |
| 280 | if (_stage != HTTP_ON_BODY) { |
| 281 | _stage = HTTP_ON_BODY; |
| 282 | } |
| 283 | if (!_read_body_progressively) { |
| 284 | // Normal read. |
| 285 | if (NULL != _current_source_iobuf) { |
| 286 | _current_source_iobuf->append_to( |
| 287 | &_body, length, _parsed_block_size + (at - _current_block_base)); |
| 288 | } else { |
| 289 | _body.append(at, length); |
| 290 | } |
| 291 | return 0; |
| 292 | } |
| 293 | // Progressive read. |
| 294 | std::unique_lock<butil::Mutex> mu(_body_mutex); |
| 295 | ProgressiveReader* r = _body_reader; |
| 296 | while (r == NULL) { |
| 297 | // When _body is full, the sleep-waiting may block parse handler |
| 298 | // of the protocol. A more efficient solution is to remove the |
| 299 | // socket from epoll and add it back when the _body is not full, |
| 300 | // which requires a set of complicated "pause" and "unpause" |
| 301 | // asynchronous API. We just leave the job to bthread right now |
| 302 | // to make everything work. |
| 303 | if ((int64_t)_body.size() <= FLAGS_socket_max_unwritten_bytes) { |
| 304 | _body.append(at, length); |
| 305 | return 0; |
| 306 | } |
| 307 | mu.unlock(); |
| 308 | bthread_usleep(10000/*10ms*/); |
| 309 | mu.lock(); |
| 310 | r = _body_reader; |
| 311 | } |
| 312 | // Safe to operate _body_reader outside lock because OnBody() is |
| 313 | // guaranteed to be called by only one thread. |
no test coverage detected