fix: Handle the streaming of JSON delimited by newlines

FUTURE_COPYBARA_INTEGRATE_REVIEW=#4861 from googleapis:release-please--branches--main 039f2cb
PiperOrigin-RevId: 719423860

Author: yeesian

vertexai/reasoning_engines/_reasoning_engines.py

@@ -840,9 +840,9 @@ def _method(self, **kwargs) -> Iterable[Any]:
             ),
         )
         for chunk in response:
-            parsed_json = _utils.to_parsed_json(chunk)
-            if parsed_json is not None:
-                yield parsed_json
+            for parsed_json in _utils.to_parsed_json(chunk):
+                if parsed_json is not None:
+                    yield parsed_json
 
     _method.__name__ = method_name
     _method.__doc__ = doc

vertexai/reasoning_engines/_utils.py

@@ -104,22 +104,30 @@ def to_parsed_json(body: httpbody_pb2.HttpBody) -> Any:
     data = getattr(body, "data", None)
 
     if content_type is None or data is None or "application/json" not in content_type:
-        return body
+        yield body
+        return
 
     try:
         utf8_data = data.decode("utf-8")
     except Exception as e:
         _LOGGER.warning(f"Failed to decode data: {data}. Exception: {e}")
-        return body
+        yield body
+        return
 
     if not utf8_data:
-        return None
-
-    try:
-        return json.loads(utf8_data)
-    except Exception as e:
-        _LOGGER.warning(f"Failed to parse JSON: {utf8_data}. Exception: {e}")
-        return body  # Return the raw body on error
+        yield None
+        return
+
+    # Handle the case of multiple dictionaries delimited by newlines.
+    for line in utf8_data.split("\n"):
+        if line:
+            try:
+                line = json.loads(line)
+            except Exception as e:
+                _LOGGER.warning(
+                    f"failed to parse json: {line}. Exception: {e}"
+                )
+            yield line
 
 
 def generate_schema(