| 65 | } |
| 66 | |
| 67 | void SendMongoResponse::Run() { |
| 68 | std::unique_ptr<SendMongoResponse> delete_self(this); |
| 69 | ConcurrencyRemover concurrency_remover(status, &cntl, received_us); |
| 70 | Socket* socket = ControllerPrivateAccessor(&cntl).get_sending_socket(); |
| 71 | |
| 72 | if (cntl.IsCloseConnection()) { |
| 73 | socket->SetFailed(); |
| 74 | return; |
| 75 | } |
| 76 | |
| 77 | const MongoServiceAdaptor* adaptor = |
| 78 | server->options().mongo_service_adaptor; |
| 79 | butil::IOBuf res_buf; |
| 80 | if (cntl.Failed()) { |
| 81 | adaptor->SerializeError(res.header().response_to(), &res_buf); |
| 82 | } else if (res.has_message()) { |
| 83 | mongo_head_t header = { |
| 84 | res.header().message_length(), |
| 85 | res.header().request_id(), |
| 86 | res.header().response_to(), |
| 87 | res.header().op_code() |
| 88 | }; |
| 89 | res_buf.append(static_cast<const void*>(&header), sizeof(mongo_head_t)); |
| 90 | int32_t response_flags = res.response_flags(); |
| 91 | int64_t cursor_id = res.cursor_id(); |
| 92 | int32_t starting_from = res.starting_from(); |
| 93 | int32_t number_returned = res.number_returned(); |
| 94 | res_buf.append(&response_flags, sizeof(response_flags)); |
| 95 | res_buf.append(&cursor_id, sizeof(cursor_id)); |
| 96 | res_buf.append(&starting_from, sizeof(starting_from)); |
| 97 | res_buf.append(&number_returned, sizeof(number_returned)); |
| 98 | res_buf.append(res.message()); |
| 99 | } |
| 100 | |
| 101 | if (!res_buf.empty()) { |
| 102 | // Have the risk of unlimited pending responses, in which case, tell |
| 103 | // users to set max_concurrency. |
| 104 | Socket::WriteOptions wopt; |
| 105 | wopt.ignore_eovercrowded = true; |
| 106 | if (socket->Write(&res_buf, &wopt) != 0) { |
| 107 | PLOG(WARNING) << "Fail to write into " << *socket; |
| 108 | return; |
| 109 | } |
| 110 | } |
| 111 | } |
| 112 | |
| 113 | ParseResult ParseMongoMessage(butil::IOBuf* source, |
| 114 | Socket* socket, bool /*read_eof*/, const void *arg) { |
no test coverage detected