| 769 | } |
| 770 | |
| 771 | Connection::ProcessOutboundTransfersResult Connection::ProcessOutboundTransfers() { |
| 772 | while (!outbound_transfers_.empty()) { |
| 773 | OutboundTransfer* transfer = &(outbound_transfers_.front()); |
| 774 | |
| 775 | if (!transfer->TransferStarted()) { |
| 776 | if (transfer->is_for_outbound_call()) { |
| 777 | CallAwaitingResponse* car = FindOrDie(awaiting_response_, transfer->call_id()); |
| 778 | if (!car->call) { |
| 779 | // If the call has already timed out or has already been cancelled, the 'call' |
| 780 | // field would be set to NULL. In that case, don't bother sending it. |
| 781 | outbound_transfers_.pop_front(); |
| 782 | transfer->Abort(Status::Aborted("already timed out or cancelled")); |
| 783 | delete transfer; |
| 784 | continue; |
| 785 | } |
| 786 | |
| 787 | // If this is the start of the transfer, then check if the server has the |
| 788 | // required RPC flags. We have to wait until just before the transfer in |
| 789 | // order to ensure that the negotiation has taken place, so that the flags |
| 790 | // are available. |
| 791 | const set<RpcFeatureFlag>& required_features = car->call->required_rpc_features(); |
| 792 | if (!includes(remote_features_.begin(), remote_features_.end(), |
| 793 | required_features.begin(), required_features.end())) { |
| 794 | outbound_transfers_.pop_front(); |
| 795 | Status s = Status::NotSupported("server does not support the required RPC features"); |
| 796 | transfer->Abort(s); |
| 797 | Phase phase = negotiation_complete_ ? Phase::REMOTE_CALL : Phase::CONNECTION_NEGOTIATION; |
| 798 | car->call->SetFailed(std::move(s), phase); |
| 799 | // Test cancellation when 'call_' is in 'FINISHED_ERROR' state. |
| 800 | MaybeInjectCancellation(car->call); |
| 801 | car->call.reset(); |
| 802 | delete transfer; |
| 803 | continue; |
| 804 | } |
| 805 | |
| 806 | car->call->SetSending(); |
| 807 | |
| 808 | // Test cancellation when 'call_' is in 'SENDING' state. |
| 809 | MaybeInjectCancellation(car->call); |
| 810 | } |
| 811 | } |
| 812 | |
| 813 | last_activity_time_ = reactor_thread_->cur_time(); |
| 814 | Status status = transfer->SendBuffer(socket_.get()); |
| 815 | if (PREDICT_FALSE(!status.ok())) { |
| 816 | LOG(WARNING) << ToString() << " send error: " << status.ToString(); |
| 817 | reactor_thread_->DestroyConnection(this, status); |
| 818 | return kConnectionDestroyed; |
| 819 | } |
| 820 | |
| 821 | if (!transfer->TransferFinished()) { |
| 822 | DVLOG(3) << ToString() << ": writeHandler: xfer not finished."; |
| 823 | return kMoreToSend; |
| 824 | } |
| 825 | |
| 826 | outbound_transfers_.pop_front(); |
| 827 | delete transfer; |
| 828 | } |
nothing calls this directly
no test coverage detected