Ensure that OHTTP test client calls response callback for chunked responses

This CL also make sure we always terminate the stdout body with a new line to facilitate terminal debugging, and also flushes body chunks to make sure they appear in logs as they come in.

PiperOrigin-RevId: 922169567

quiche/quic/masque/masque_ohttp_client.cc

diff --git a/quiche/quic/masque/masque_ohttp_client.cc b/quiche/quic/masque/masque_ohttp_client.cc
index 1efe7ad..0c3e4e9 100644
--- a/quiche/quic/masque/masque_ohttp_client.cc
+++ b/quiche/quic/masque/masque_ohttp_client.cc
@@ -1014,6 +1014,10 @@
     }
   }
   std::cout << response.body;
+  if (!response.body.empty() &&
+      response.body[response.body.size() - 1] != '\n') {
+    std::cout << std::endl;
+  }
   int16_t encapsulated_status_code =
       MasqueConnectionPool::GetStatusCode(response);
   if (per_request_config.expected_encapsulated_status_code().has_value()) {
@@ -1101,10 +1105,24 @@
     }
     return;
   }
-  if (end_stream && response_visitor_) {
-    response_visitor_->OnResponseDone(
-        request_id,
-        std::move(*pending_request.chunk_handler).ExtractResponse());
+  if (end_stream) {
+    Message response =
+        std::move(*pending_request.chunk_handler).ExtractResponse();
+    if (const auto& callback = pending_request.per_request_config
+                                   .encapsulated_response_body_callback();
+        callback) {
+      status = callback(response.body);
+      if (!status.ok()) {
+        Abort(status);
+        if (response_visitor_) {
+          response_visitor_->OnError(request_id, status);
+        }
+        return;
+      }
+    }
+    if (response_visitor_) {
+      response_visitor_->OnResponseDone(request_id, response);
+    }
   }
 }
 
@@ -1289,7 +1307,11 @@
     }
   }
 
-  std::cout << processed_chunk;
+  std::cout << processed_chunk << std::flush;
+  last_chunk_ended_without_newline_ =
+      !processed_chunk.empty() &&
+      processed_chunk[processed_chunk.size() - 1] != '\n';
+
   response_.body += processed_chunk;
 
   if (response_chunk_callback_) {
@@ -1305,6 +1327,9 @@
     }
     decompressor_->EndDecompression();
   }
+  if (last_chunk_ended_without_newline_) {
+    std::cout << std::endl;
+  }
   return absl::OkStatus();
 }
 absl::Status MasqueOhttpClient::ChunkHandler::OnTrailer(

quiche/quic/masque/masque_ohttp_client.h

diff --git a/quiche/quic/masque/masque_ohttp_client.h b/quiche/quic/masque/masque_ohttp_client.h
index 7b30b03..78497d3 100644
--- a/quiche/quic/masque/masque_ohttp_client.h
+++ b/quiche/quic/masque/masque_ohttp_client.h
@@ -352,6 +352,7 @@
     size_t body_chunk_count_ = 0;
     bool handle_gzip_response_ = false;
     bool is_gzipped_ = false;
+    bool last_chunk_ended_without_newline_ = false;
     std::unique_ptr<GzipDecompressor> decompressor_;
     const std::string info_;
   };