| 580 | }; |
| 581 | |
| 582 | int Stream::Consume(void *meta, bthread::TaskIterator<butil::IOBuf*>& iter) { |
| 583 | Stream* s = (Stream*)meta; |
| 584 | s->StopIdleTimer(); |
| 585 | if (iter.is_queue_stopped()) { |
| 586 | scoped_ptr<Stream> recycled_stream(s); |
| 587 | // Indicating the queue was closed. |
| 588 | if (s->_host_socket) { |
| 589 | DereferenceSocket(s->_host_socket); |
| 590 | s->_host_socket = NULL; |
| 591 | } |
| 592 | if (s->_options.handler != NULL) { |
| 593 | int error_code; |
| 594 | std::string error_text; |
| 595 | { |
| 596 | BAIDU_SCOPED_LOCK(s->_connect_mutex); |
| 597 | error_code = s->_error_code; |
| 598 | error_text = s->_error_text; |
| 599 | } |
| 600 | if (error_code != 0) { |
| 601 | // The stream is closed abnormally. |
| 602 | s->_options.handler->on_failed(s->id(), error_code, error_text); |
| 603 | } |
| 604 | s->_options.handler->on_closed(s->id()); |
| 605 | } |
| 606 | return 0; |
| 607 | } |
| 608 | DEFINE_SMALL_ARRAY(butil::IOBuf*, buf_list, s->_options.messages_in_batch, 256); |
| 609 | MessageBatcher mb(buf_list, s->_options.messages_in_batch, s); |
| 610 | bool has_timeout_task = false; |
| 611 | for (; iter; ++iter) { |
| 612 | butil::IOBuf* t= *iter; |
| 613 | if (t == TIMEOUT_TASK) { |
| 614 | has_timeout_task = true; |
| 615 | } else { |
| 616 | if (s->_parse_rpc_response) { |
| 617 | s->_parse_rpc_response = false; |
| 618 | s->HandleRpcResponse(t); |
| 619 | } else { |
| 620 | mb.push(t); |
| 621 | } |
| 622 | } |
| 623 | } |
| 624 | if (s->_options.handler != NULL) { |
| 625 | if (has_timeout_task && mb.total_length() == 0) { |
| 626 | s->_options.handler->on_idle_timeout(s->id()); |
| 627 | } |
| 628 | } |
| 629 | mb.flush(); |
| 630 | |
| 631 | auto total_length = mb.total_length(); |
| 632 | if (total_length > 0) { |
| 633 | // fast path for connected stream |
| 634 | if (s->_connected.load(butil::memory_order_acquire)){ |
| 635 | if (s->_remote_settings.need_feedback()) { |
| 636 | s->_local_consumed += total_length; |
| 637 | s->SendFeedback(s->_local_consumed); |
| 638 | } |
| 639 | } else { |