[Streaming] Fix issues when not using try-with-resource

foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/OpenAiStreamingHandler.java

@@ -5,18 +5,12 @@
 import static com.sap.ai.sdk.foundationmodels.openai.OpenAiResponseHandler.parseErrorAndThrow;
 
 import com.sap.ai.sdk.foundationmodels.openai.model.StreamedDelta;
-import io.vavr.control.Try;
-import java.io.BufferedReader;
 import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
-import java.nio.charset.StandardCharsets;
 import java.util.stream.Stream;
 import javax.annotation.Nonnull;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hc.core5.http.ClassicHttpResponse;
-import org.apache.hc.core5.http.HttpEntity;
 
 @Slf4j
 @RequiredArgsConstructor
@@ -42,20 +36,8 @@ Stream<D> handleResponse(@Nonnull final ClassicHttpResponse response)
   @SuppressWarnings("PMD.CloseResource")
   private Stream<D> parseResponse(@Nonnull final ClassicHttpResponse response)
       throws OpenAiClientException {
-    final HttpEntity responseEntity = response.getEntity();
-    if (responseEntity == null) {
-      throw new OpenAiClientException("Response from OpenAI model was empty.");
-    }
-    final InputStream inputStream;
-    try {
-      inputStream = responseEntity.getContent();
-    } catch (IOException e) {
-      throw new OpenAiClientException("Failed to read response content.", e);
-    }
-    final var br = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8));
 
-    // https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#event_stream_format
-    return br.lines()
+    return StreamConverter.streamLines(response.getEntity())
         // half of the lines are empty newlines, the last line is "data: [DONE]"
         .filter(line -> !line.isEmpty() && !"data: [DONE]".equals(line.trim()))
         .peek(
@@ -74,10 +56,6 @@ private Stream<D> parseResponse(@Nonnull final ClassicHttpResponse response)
                 log.error("Failed to parse the following response from OpenAI model: {}", line);
                 throw new OpenAiClientException("Failed to parse delta message: " + line, e);
               }
-            })
-        .onClose(
-            () ->
-                Try.run(inputStream::close)
-                    .onFailure(e -> log.error("Could not close HTTP input stream", e)));
+            });
   }
 }

foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/StreamConverter.java

@@ -0,0 +1,108 @@
+package com.sap.ai.sdk.foundationmodels.openai;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Spliterator.NONNULL;
+import static java.util.Spliterator.ORDERED;
+
+import io.vavr.control.Try;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.UncheckedIOException;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.Spliterators;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hc.core5.http.HttpEntity;
+
+@Slf4j
+class StreamConverter {
+
+  /** see {@link BufferedReader#DEFAULT_CHAR_BUFFER_SIZE} */
+  static final int BUFFER_SIZE = 8192;
+
+  @SuppressWarnings("PMD.CloseResource")
+  @Nonnull
+  static Stream<String> streamLines(@Nullable final HttpEntity entity)
+      throws OpenAiClientException {
+    if (entity == null) {
+      throw new OpenAiClientException("OpenAI response was empty.");
+    }
+
+    final InputStream inputStream;
+    try {
+      inputStream = entity.getContent();
+    } catch (IOException e) {
+      throw new OpenAiClientException("Failed to read response content.", e);
+    }
+
+    final var reader = new BufferedReader(new InputStreamReader(inputStream, UTF_8), BUFFER_SIZE);
+    final CloseHandler closeHandler =
+        () ->
+            Try.run(reader::close)
+                .onFailure(e -> log.error("Could not close HTTP input stream", e));
+
+    final var iterator = new SequentialIterator<>(reader::readLine, closeHandler);
+    final var spliterator = Spliterators.spliteratorUnknownSize(iterator, ORDERED | NONNULL);
+    return StreamSupport.stream(spliterator, /* NOT PARALLEL */ false).onClose(closeHandler::close);
+  }
+
+  @FunctionalInterface
+  private interface ReadHandler<T> {
+    /**
+     * Read next entry for Stream.
+     *
+     * @return The next entry, or {@code null} when no further entry can be read.
+     * @throws IOException if no entry can be read anymore, unexpected.
+     */
+    @Nullable
+    T readEntry() throws IOException;
+  }
+
+  @FunctionalInterface
+  private interface CloseHandler {
+    /** Close handler to be called when Stream terminated. */
+    void close();
+  }
+
+  @RequiredArgsConstructor
+  private static class SequentialIterator<T> implements Iterator<T> {
+    private final ReadHandler<T> readHandler;
+    private final CloseHandler stopHandler;
+    private boolean done = false;
+    private T next = null;
+
+    @Override
+    public boolean hasNext() {
+      if (done) {
+        return false;
+      }
+      try {
+        next = readHandler.readEntry();
+        if (next == null) {
+          done = true;
+          stopHandler.close();
+        }
+      } catch (final IOException e) {
+        done = true;
+        stopHandler.close();
+        throw new UncheckedIOException("Iterator stopped unexpectedly.", e);
+      }
+      return !done;
+    }
+
+    @Override
+    public T next() {
+      if (next == null && !hasNext()) {
+        throw new NoSuchElementException();
+      }
+      return next;
+    }
+  }
+}

foundation-models/openai/src/test/java/com/sap/ai/sdk/foundationmodels/openai/StreamConverterTest.java

@@ -0,0 +1,47 @@
+package com.sap.ai.sdk.foundationmodels.openai;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicInteger;
+import lombok.SneakyThrows;
+import org.apache.hc.core5.http.ContentType;
+import org.apache.hc.core5.http.io.entity.InputStreamEntity;
+import org.junit.jupiter.api.Test;
+
+public class StreamConverterTest {
+  @SneakyThrows
+  @Test
+  void testStreamLines() {
+    final var TEMPLATE = "THIS\nIS\nA\nTEST\n";
+    final var input = TEMPLATE.repeat(StreamConverter.BUFFER_SIZE);
+    final var inputStream = spy(new ByteArrayInputStream(input.getBytes(StandardCharsets.UTF_8)));
+    final var entity = new InputStreamEntity(inputStream, ContentType.TEXT_PLAIN);
+
+    final var sut = StreamConverter.streamLines(entity);
+    verify(inputStream, never()).read();
+    verify(inputStream, never()).read(any());
+    verify(inputStream, never()).read(any(), anyInt(), anyInt());
+
+    final var streamCounter = new AtomicInteger(0);
+    sut.peek(s -> streamCounter.incrementAndGet())
+        .forEach(
+            s ->
+                assertThat(s)
+                    .containsAnyOf("THIS", "IS", "A", "TEST")
+                    .doesNotContainAnyWhitespaces());
+
+    assertThat(streamCounter).hasValue(StreamConverter.BUFFER_SIZE * 4);
+    verify(inputStream, times(TEMPLATE.length() + 1))
+        .read(any(), anyInt(), eq(StreamConverter.BUFFER_SIZE));
+    verify(inputStream, times(1)).close();
+  }
+}

sample-code/spring-app/src/main/java/com/sap/ai/sdk/app/controllers/OpenAiController.java

@@ -64,15 +64,10 @@ public static ResponseEntity<ResponseBodyEmitter> streamChatCompletionDeltas() {
     final Runnable consumeStream =
         () -> {
           final var totalOutput = new OpenAiChatCompletionOutput();
-          // try-with-resources ensures the stream is closed
-          try (stream) {
-            stream
-                .peek(totalOutput::addDelta)
-                .forEach(delta -> send(emitter, delta.getDeltaContent()));
-          } finally {
-            send(emitter, "\n\n-----Total Output-----\n\n" + objectToJson(totalOutput));
-            emitter.complete();
-          }
+          stream
+              .peek(totalOutput::addDelta)
+              .forEach(delta -> send(emitter, delta.getDeltaContent()));
+          send(emitter, "\n\n-----Total Output-----\n\n" + objectToJson(totalOutput));
         };
 
     ThreadContextExecutors.getExecutor().execute(consumeStream);