Implement inputStream.read() based SSE processing

Co-authored-by: samvaity <16845631+samvaity@users.noreply.github.com>

Author: Copilot

sdk/clientcore/core/src/main/java/io/clientcore/core/utils/ServerSentEventUtils.java

@@ -11,10 +11,8 @@
 import io.clientcore.core.implementation.utils.ServerSentEventHelper;
 import io.clientcore.core.models.ServerSentResult;
 
-import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStream;
-import java.io.InputStreamReader;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -58,8 +56,8 @@ public static boolean isTextEventStreamContentType(String contentType) {
      */
     public static ServerSentResult processTextEventStream(InputStream inputStream, ServerSentEventListener listener)
         throws IOException {
-        try (BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8))) {
-            return processBuffer(reader, listener);
+        try (InputStream is = inputStream) {
+            return processBuffer(is, listener);
         }
     }
 
@@ -112,18 +110,106 @@ private static boolean isEndOfBlock(StringBuilder sb) {
     /**
      * Processes the SSE buffer and dispatches the event.
      *
-     * @param reader The {@link BufferedReader} to read data from.
+     * @param inputStream The {@link InputStream} to read data from.
      * @param listener The {@link ServerSentEventListener} object attached to the {@link HttpRequest}.
      */
-    private static ServerSentResult processBuffer(BufferedReader reader, ServerSentEventListener listener) {
+    private static ServerSentResult processBuffer(InputStream inputStream, ServerSentEventListener listener) {
         ServerSentEvent event = null;
         List<String> allCollectedData = new ArrayList<>(); // List to store all collected data blocks
 
         try {
             StringBuilder collectedData = new StringBuilder();
-            String line;
 
-            while ((line = reader.readLine()) != null) {
+            // Use a byte array to accumulate UTF-8 bytes and convert to string line by line
+            List<Byte> lineBytes = new ArrayList<>();
+
+            int b;
+            while ((b = inputStream.read()) != -1) {
+                byte currentByte = (byte) b;
+
+                if (currentByte == '\n') {
+                    // End of line found, convert accumulated bytes to string
+                    String line = convertBytesToString(lineBytes);
+                    collectedData.append(line).append("\n");
+                    lineBytes.clear();
+
+                    if (isEndOfBlock(collectedData)) {
+                        String temp = collectedData.toString();
+                        allCollectedData.add(temp.trim().replace("data: ", "")); // Add the collected data block to the list
+                        event = processLines(collectedData.toString().split("\n"));
+
+                        if (!Objects.equals(event.getEvent(), DEFAULT_EVENT) || event.getData() != null) {
+                            listener.onEvent(event);
+                        }
+
+                        collectedData = new StringBuilder(); // clear the collected data
+                    }
+                } else if (currentByte == '\r') {
+                    // Handle \r\n or standalone \r - peek ahead to see if \n follows
+                    int next = inputStream.read();
+                    if (next == -1) {
+                        // End of stream, treat \r as line ending
+                        String line = convertBytesToString(lineBytes);
+                        collectedData.append(line).append("\n");
+                        lineBytes.clear();
+
+                        if (isEndOfBlock(collectedData)) {
+                            String temp = collectedData.toString();
+                            allCollectedData.add(temp.trim().replace("data: ", "")); // Add the collected data block to the list
+                            event = processLines(collectedData.toString().split("\n"));
+
+                            if (!Objects.equals(event.getEvent(), DEFAULT_EVENT) || event.getData() != null) {
+                                listener.onEvent(event);
+                            }
+
+                            collectedData = new StringBuilder(); // clear the collected data
+                        }
+                        break;
+                    } else if (next == '\n') {
+                        // \r\n found, end of line
+                        String line = convertBytesToString(lineBytes);
+                        collectedData.append(line).append("\n");
+                        lineBytes.clear();
+
+                        if (isEndOfBlock(collectedData)) {
+                            String temp = collectedData.toString();
+                            allCollectedData.add(temp.trim().replace("data: ", "")); // Add the collected data block to the list
+                            event = processLines(collectedData.toString().split("\n"));
+
+                            if (!Objects.equals(event.getEvent(), DEFAULT_EVENT) || event.getData() != null) {
+                                listener.onEvent(event);
+                            }
+
+                            collectedData = new StringBuilder(); // clear the collected data
+                        }
+                    } else {
+                        // Standalone \r, treat as line ending and add the next byte to line bytes
+                        String line = convertBytesToString(lineBytes);
+                        collectedData.append(line).append("\n");
+                        lineBytes.clear();
+                        lineBytes.add((byte) next);
+
+                        if (isEndOfBlock(collectedData)) {
+                            String temp = collectedData.toString();
+                            allCollectedData.add(temp.trim().replace("data: ", "")); // Add the collected data block to the list
+                            event = processLines(collectedData.toString().split("\n"));
+
+                            if (!Objects.equals(event.getEvent(), DEFAULT_EVENT) || event.getData() != null) {
+                                listener.onEvent(event);
+                            }
+
+                            collectedData = new StringBuilder(); // clear the collected data
+                        }
+                    }
+                } else {
+                    // Regular byte, add to current line
+                    lineBytes.add(currentByte);
+                }
+            }
+
+            // Handle any remaining data that didn't end with a newline
+            if (!lineBytes.isEmpty()) {
+                String line = convertBytesToString(lineBytes);
                 collectedData.append(line).append("\n");
 
                 if (isEndOfBlock(collectedData)) {
@@ -134,8 +220,6 @@ private static ServerSentResult processBuffer(BufferedReader reader, ServerSentE
                     if (!Objects.equals(event.getEvent(), DEFAULT_EVENT) || event.getData() != null) {
                         listener.onEvent(event);
                     }
-
-                    collectedData = new StringBuilder(); // clear the collected data
                 }
             }
 
@@ -145,7 +229,24 @@ private static ServerSentResult processBuffer(BufferedReader reader, ServerSentE
                 event != null ? ServerSentEventHelper.getRetryAfter(event) : null, null);
         }
 
-        return new ServerSentResult(null, event.getId(), ServerSentEventHelper.getRetryAfter(event), allCollectedData);
+        return new ServerSentResult(null, event != null ? event.getId() : null,
+            event != null ? ServerSentEventHelper.getRetryAfter(event) : null, allCollectedData);
+    }
+
+    /**
+     * Converts a list of bytes to a UTF-8 string.
+     */
+    private static String convertBytesToString(List<Byte> bytes) {
+        if (bytes.isEmpty()) {
+            return "";
+        }
+
+        byte[] byteArray = new byte[bytes.size()];
+        for (int i = 0; i < bytes.size(); i++) {
+            byteArray[i] = bytes.get(i);
+        }
+
+        return new String(byteArray, StandardCharsets.UTF_8);
     }
 
     private static ServerSentEvent processLines(String[] lines) {