| 80 | } |
| 81 | |
| 82 | ParseResult ParseCouchbaseMessage(butil::IOBuf* source, Socket* socket, |
| 83 | bool /*read_eof*/, const void* /*arg*/) { |
| 84 | while (1) { |
| 85 | const uint8_t* p_cbmagic = (const uint8_t*)source->fetch1(); |
| 86 | if (NULL == p_cbmagic) { |
| 87 | return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA); |
| 88 | } |
| 89 | if (*p_cbmagic != (uint8_t)CB_MAGIC_RESPONSE) { |
| 90 | return MakeParseError(PARSE_ERROR_TRY_OTHERS); |
| 91 | } |
| 92 | char buf[24]; |
| 93 | const uint8_t* p = (const uint8_t*)source->fetch(buf, sizeof(buf)); |
| 94 | if (NULL == p) { |
| 95 | return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA); |
| 96 | } |
| 97 | const CouchbaseResponseHeader* header = (const CouchbaseResponseHeader*)p; |
| 98 | uint32_t total_body_length = butil::NetToHost32(header->total_body_length); |
| 99 | if (source->size() < sizeof(*header) + total_body_length) { |
| 100 | return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA); |
| 101 | } |
| 102 | |
| 103 | if (!IsSupportedCommand(header->command)) { |
| 104 | LOG(WARNING) << "Not support command=" << header->command; |
| 105 | source->pop_front(sizeof(*header) + total_body_length); |
| 106 | return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA); |
| 107 | } |
| 108 | |
| 109 | PipelinedInfo pi; |
| 110 | if (!socket->PopPipelinedInfo(&pi)) { |
| 111 | LOG(WARNING) << "No corresponding PipelinedInfo in socket, drop"; |
| 112 | source->pop_front(sizeof(*header) + total_body_length); |
| 113 | return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA); |
| 114 | } |
| 115 | MostCommonMessage* msg = |
| 116 | static_cast<MostCommonMessage*>(socket->parsing_context()); |
| 117 | if (msg == NULL) { |
| 118 | msg = MostCommonMessage::Get(); |
| 119 | socket->reset_parsing_context(msg); |
| 120 | } |
| 121 | |
| 122 | // endianness conversions. |
| 123 | const CouchbaseResponseHeader local_header = { |
| 124 | header->magic, |
| 125 | header->command, |
| 126 | butil::NetToHost16(header->key_length), |
| 127 | header->extras_length, |
| 128 | header->data_type, |
| 129 | butil::NetToHost16(header->status), |
| 130 | total_body_length, |
| 131 | butil::NetToHost32(header->opaque), |
| 132 | butil::NetToHost64(header->cas_value), |
| 133 | }; |
| 134 | msg->meta.append(&local_header, sizeof(local_header)); |
| 135 | source->pop_front(sizeof(*header)); |
| 136 | source->cutn(&msg->meta, total_body_length); |
| 137 | if (++msg->pi.count >= pi.count) { |
| 138 | CHECK_EQ(msg->pi.count, pi.count); |
| 139 | msg = static_cast<MostCommonMessage*>(socket->release_parsing_context()); |
no test coverage detected