| 497 | } |
| 498 | |
| 499 | int Stream::OnReceived(const StreamFrameMeta& fm, butil::IOBuf *buf, Socket* sock) { |
| 500 | if (_host_socket == NULL) { |
| 501 | if (SetHostSocket(sock) != 0) { |
| 502 | return -1; |
| 503 | } |
| 504 | } |
| 505 | switch (fm.frame_type()) { |
| 506 | case FRAME_TYPE_FEEDBACK: |
| 507 | SetRemoteConsumed(fm.feedback().consumed_size()); |
| 508 | CHECK(buf->empty()); |
| 509 | break; |
| 510 | case FRAME_TYPE_DATA: |
| 511 | if (_pending_buf != NULL) { |
| 512 | _pending_buf->append(*buf); |
| 513 | buf->clear(); |
| 514 | } else { |
| 515 | _pending_buf = new butil::IOBuf; |
| 516 | _pending_buf->swap(*buf); |
| 517 | } |
| 518 | if (!fm.has_continuation()) { |
| 519 | butil::IOBuf *tmp = _pending_buf; |
| 520 | _pending_buf = NULL; |
| 521 | int rc = bthread::execution_queue_execute(_consumer_queue, tmp); |
| 522 | if (rc != 0) { |
| 523 | CHECK(false) << "Fail to push into channel"; |
| 524 | delete tmp; |
| 525 | Close(rc, "Fail to push into channel"); |
| 526 | } |
| 527 | } |
| 528 | break; |
| 529 | case FRAME_TYPE_RST: |
| 530 | RPC_VLOG << "stream=" << id() << " received rst frame"; |
| 531 | Close(ECONNRESET, "Received RST frame"); |
| 532 | break; |
| 533 | case FRAME_TYPE_CLOSE: |
| 534 | RPC_VLOG << "stream=" << id() << " received close frame"; |
| 535 | // TODO:: See the comments in Consume |
| 536 | Close(0, "Received CLOSE frame"); |
| 537 | break; |
| 538 | case FRAME_TYPE_UNKNOWN: |
| 539 | RPC_VLOG << "Received unknown frame"; |
| 540 | return -1; |
| 541 | } |
| 542 | return 0; |
| 543 | } |
| 544 | |
| 545 | class MessageBatcher { |
| 546 | public: |
no test coverage detected