| 1101 | #endif |
| 1102 | |
| 1103 | ParseResult ParseH2Message(butil::IOBuf *source, Socket *socket, |
| 1104 | bool read_eof, const void *arg) { |
| 1105 | #if defined(BRPC_PROFILE_H2) |
| 1106 | bvar::ScopedTimer<bvar::Adder<int64_t> > tm(g_parse_time); |
| 1107 | #endif |
| 1108 | H2Context* ctx = static_cast<H2Context*>(socket->parsing_context()); |
| 1109 | if (ctx == NULL) { |
| 1110 | if (read_eof || source->empty()) { |
| 1111 | return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA); |
| 1112 | } |
| 1113 | const Server* server = static_cast<const Server*>(arg); |
| 1114 | ctx = new H2Context(socket, server); |
| 1115 | if (ctx->Init() != 0) { |
| 1116 | delete ctx; |
| 1117 | LOG(ERROR) << "Fail to init H2Context"; |
| 1118 | return MakeParseError(PARSE_ERROR_NO_RESOURCE); |
| 1119 | } |
| 1120 | socket->initialize_parsing_context(&ctx); |
| 1121 | } |
| 1122 | butil::IOBufBytesIterator it(*source); |
| 1123 | size_t last_bytes_left = it.bytes_left(); |
| 1124 | CHECK_EQ(last_bytes_left, source->size()); |
| 1125 | while (true) { |
| 1126 | ParseResult res = ctx->Consume(it, socket); |
| 1127 | if (res.is_ok()) { |
| 1128 | last_bytes_left = it.bytes_left(); |
| 1129 | if (res.message() == NULL) { |
| 1130 | // no message to process, continue parsing. |
| 1131 | continue; |
| 1132 | } |
| 1133 | } |
| 1134 | source->pop_front(source->size() - last_bytes_left); |
| 1135 | ctx->ClearAbandonedStreams(); |
| 1136 | return res; |
| 1137 | } |
| 1138 | } |
| 1139 | |
| 1140 | void H2Context::AddAbandonedStream(uint32_t stream_id) { |
| 1141 | std::unique_lock<butil::Mutex> mu(_abandoned_streams_mutex); |