[azopenai] EventReader should return an error if the stream is closed

sdk/ai/azopenai/CHANGELOG.md

@@ -11,6 +11,10 @@
 
 ### Bugs Fixed
 
+- EventReader, used by GetChatCompletionsStream and GetCompletionsStream for streaming results, would not return an 
+  error if the underlying Body reader was closed or EOF'd before the actual DONE: token arrived. This could result in an
+  infinite loop for callers. (PR#)
+
 ### Other Changes
 
 ## 0.1.1 (2023-07-26)

sdk/ai/azopenai/custom_models_test.go

@@ -55,7 +55,7 @@ func TestParseResponseError(t *testing.T) {
 
 	contentFilterResults := contentFilterErr.ContentFilterResults
 
-	// thsi comment was considered violent, so it was filtered.
+	// this comment was considered violent, so it was filtered.
 	require.Equal(t, &ContentFilterResultsViolence{
 		Filtered: to.Ptr(true),
 		Severity: to.Ptr(ContentFilterSeverityMedium)}, contentFilterResults.Violence)

sdk/ai/azopenai/event_reader.go

@@ -47,12 +47,19 @@ func (er *EventReader[T]) Read() (T, error) {
 				err := json.Unmarshal([]byte(tokens[1]), &data)
 				return data, err
 			default: // Any other event type is an unexpected
-				return data, errors.New("Unexpected event type: " + tokens[0])
+				return data, errors.New("unexpected event type: " + tokens[0])
 			}
 			// Unreachable
 		}
 	}
-	return *new(T), er.scanner.Err()
+
+	scannerErr := er.scanner.Err()
+
+	if scannerErr == nil {
+		return *new(T), errors.New("incomplete stream")
+	}
+
+	return *new(T), scannerErr
 }
 
 // Close closes the EventReader and any applicable inner stream state.

sdk/ai/azopenai/event_reader_test.go

@@ -24,7 +24,7 @@ func TestEventReader_InvalidType(t *testing.T) {
 
 	firstEvent, err := eventReader.Read()
 	require.Empty(t, firstEvent)
-	require.EqualError(t, err, "Unexpected event type: invaliddata")
+	require.EqualError(t, err, "unexpected event type: invaliddata")
 }
 
 type badReader struct{}
@@ -35,8 +35,23 @@ func (br badReader) Read(p []byte) (n int, err error) {
 
 func TestEventReader_BadReader(t *testing.T) {
 	eventReader := newEventReader[ChatCompletions](io.NopCloser(badReader{}))
+	defer eventReader.Close()
 
 	firstEvent, err := eventReader.Read()
 	require.Empty(t, firstEvent)
 	require.ErrorIs(t, io.ErrClosedPipe, err)
 }
+
+func TestEventReader_StreamIsClosedBeforeDone(t *testing.T) {
+	buff := strings.NewReader("data: {}")
+
+	eventReader := newEventReader[ChatCompletions](io.NopCloser(buff))
+
+	evt, err := eventReader.Read()
+	require.Empty(t, evt)
+	require.NoError(t, err)
+
+	evt, err = eventReader.Read()
+	require.Empty(t, evt)
+	require.EqualError(t, err, "incomplete stream")
+}