add async monitoring http KIKIMR-14742

ref:c51d608f0ae78f08597b88f837491da33a953ef6

1 files changed, 55 insertions, 6 deletions

diff --git a/library/cpp/actors/http/http.cpp b/library/cpp/actors/http/http.cpp
index 7125f9d8b04..3d07c870ced 100644
--- a/library/cpp/actors/http/http.cpp
+++ b/library/cpp/actors/http/http.cpp
@@ -63,6 +63,14 @@ void THttpRequest::Clear() {
 }
 
 template <>
+bool THttpParser<THttpRequest, TSocketBuffer>::HaveBody() const {
+    if (!Body.empty()) {
+        return true;
+    }
+    return (!ContentType.empty() || !ContentLength.empty() || !TransferEncoding.empty());
+}
+
+template <>
 void THttpParser<THttpRequest, TSocketBuffer>::Advance(size_t len) {
     TStringBuf data(Pos(), len);
     while (!data.empty()) {
@@ -98,7 +106,6 @@ void THttpParser<THttpRequest, TSocketBuffer>::Advance(size_t len) {
             case EParseStage::Header: {
                 if (ProcessData(Header, data, "\r\n", MaxHeaderSize)) {
                     if (Header.empty()) {
-                        Headers = TStringBuf(Headers.data(), data.begin() - Headers.begin());
                         if (HaveBody()) {
                             Stage = EParseStage::Body;
                         } else {
@@ -107,6 +114,7 @@ void THttpParser<THttpRequest, TSocketBuffer>::Advance(size_t len) {
                     } else {
                         ProcessHeader(Header);
                     }
+                    Headers = TStringBuf(Headers.data(), data.data() - Headers.data());
                 }
                 break;
             }
@@ -116,8 +124,13 @@ void THttpParser<THttpRequest, TSocketBuffer>::Advance(size_t len) {
                         Body = Content;
                         Stage = EParseStage::Done;
                     }
-                } else if (TransferEncoding == "chunked") {
+                } else if (TEqNoCase()(TransferEncoding, "chunked")) {
                     Stage = EParseStage::ChunkLength;
+                } else if (TotalSize.has_value()) {
+                    if (ProcessData(Content, data, GetBodySizeFromTotalSize())) {
+                        Body = Content;
+                        Stage = EParseStage::Done;
+                    }
                 } else {
                     // Invalid body encoding
                     Stage = EParseStage::Error;
@@ -189,6 +202,15 @@ THttpParser<THttpRequest, TSocketBuffer>::EParseStage THttpParser<THttpRequest,
 }
 
 template <>
+bool THttpParser<THttpResponse, TSocketBuffer>::HaveBody() const {
+    if (!Body.empty()) {
+        return true;
+    }
+    return (!Status.starts_with("1") && Status != "204" && Status != "304")
+        && (!ContentType.empty() || !ContentLength.empty() || !TransferEncoding.empty());
+}
+
+template <>
 THttpParser<THttpResponse, TSocketBuffer>::EParseStage THttpParser<THttpResponse, TSocketBuffer>::GetInitialStage() {
     return EParseStage::Protocol;
 }
@@ -237,6 +259,8 @@ void THttpParser<THttpResponse, TSocketBuffer>::Advance(size_t len) {
                     if (Header.empty()) {
                         if (HaveBody() && (ContentLength.empty() || ContentLength != "0")) {
                             Stage = EParseStage::Body;
+                        } else if (TotalSize.has_value() && !data.empty()) {
+                            Stage = EParseStage::Body;
                         } else {
                             Stage = EParseStage::Done;
                         }
@@ -252,8 +276,13 @@ void THttpParser<THttpResponse, TSocketBuffer>::Advance(size_t len) {
                     if (ProcessData(Body, data, FromString(ContentLength))) {
                         Stage = EParseStage::Done;
                     }
-                } else if (TransferEncoding == "chunked") {
+                } else if (TEqNoCase()(TransferEncoding, "chunked")) {
                     Stage = EParseStage::ChunkLength;
+                } else if (TotalSize.has_value()) {
+                    if (ProcessData(Content, data, GetBodySizeFromTotalSize())) {
+                        Body = Content;
+                        Stage = EParseStage::Done;
+                    }
                 } else {
                     // Invalid body encoding
                     Stage = EParseStage::Error;
@@ -333,9 +362,19 @@ void THttpParser<THttpResponse, TSocketBuffer>::ConnectionClosed() {
 }
 
 THttpOutgoingResponsePtr THttpIncomingRequest::CreateResponseString(TStringBuf data) {
+    THttpParser<THttpResponse, TSocketBuffer> parser(data);
+    THeadersBuilder headers(parser.Headers);
+    if (!WorkerName.empty()) {
+        headers.Set("X-Worker-Name", WorkerName);
+    }
     THttpOutgoingResponsePtr response = new THttpOutgoingResponse(this);
-    response->Append(data);
-    response->Reparse();
+    response->InitResponse(parser.Protocol, parser.Version, parser.Status, parser.Message);
+    response->Set(headers);
+    if (parser.HaveBody()) {
+        response->SetBody(parser.Body);
+    } else {
+        response->Set<&THttpResponse::ContentLength>("0");
+    }
     return response;
 }
 
@@ -601,11 +640,17 @@ void TCookiesBuilder::Set(TStringBuf name, TStringBuf data) {
 }
 
 THeaders::THeaders(TStringBuf headers) {
+    Parse(headers);
+}
+
+size_t THeaders::Parse(TStringBuf headers) {
+    auto start = headers.begin();
     for (TStringBuf param = headers.NextTok("\r\n"); !param.empty(); param = headers.NextTok("\r\n")) {
         TStringBuf name = param.NextTok(":");
         param.SkipPrefix(" ");
         Headers[name] = param;
     }
+    return headers.begin() - start;
 }
 
 TStringBuf THeaders::operator [](TStringBuf name) const {
@@ -636,7 +681,11 @@ TString THeaders::Render() const {
 }
 
 THeadersBuilder::THeadersBuilder()
-    :THeaders(TStringBuf())
+    : THeaders(TStringBuf())
+{}
+
+THeadersBuilder::THeadersBuilder(TStringBuf headers)
+    : THeaders(headers)
 {}
 
 THeadersBuilder::THeadersBuilder(const THeadersBuilder& builder) {