fix: check [DONE] SSE marker before deserialization in stream_mapped_raw_events

PR #537 refactored the streaming pipeline and removed reqwest-eventsource.
The old `stream()` function checked for the `[DONE]` SSE end-of-stream
marker *before* passing event data to `serde_json::from_str`. The
refactored version delegated to `stream_mapped_raw_events`, which checks
`[DONE]` *after* calling the `event_mapper` closure. Since the mapper in
`stream()` deserializes JSON, this causes `[DONE]` to fail with:

  failed deserialization of: [DONE]

Move the `[DONE]` check before the mapper call in both the WASM
(futures::stream::unfold) and non-WASM (tokio::mpsc) implementations.

Closes #542

Author: zhaojing.jz

async-openai/src/client.rs

@@ -791,12 +791,8 @@ where
     let event_stream = Box::pin(eventsource_stream::EventStream::new(byte_stream));
 
     Box::pin(futures::stream::unfold(
-        (event_stream, event_mapper, false),
-        |(mut event_stream, event_mapper, finished)| async move {
-            if finished {
-                return None;
-            }
-
+        (event_stream, event_mapper),
+        |(mut event_stream, event_mapper)| async move {
             loop {
                 let event = match event_stream.next().await {
                     Some(Ok(event)) => event,
@@ -805,20 +801,22 @@ where
                             Err(OpenAIError::StreamError(Box::new(
                                 StreamError::EventStream(error.to_string()),
                             ))),
-                            (event_stream, event_mapper, true),
+                            (event_stream, event_mapper),
                         ));
                     }
                     None => return None,
                 };
 
-                let done = event.data == "[DONE]";
+                if event.data == "[DONE]" {
+                    return None;
+                }
 
                 if event.event == "keepalive" {
                     continue;
                 }
 
                 let response = event_mapper(event);
-                return Some((response, (event_stream, event_mapper, done)));
+                return Some((response, (event_stream, event_mapper)));
             }
         },
     ))
@@ -856,7 +854,9 @@ where
                     break;
                 }
             };
-            let done = event.data == "[DONE]";
+            if event.data == "[DONE]" {
+                break;
+            }
 
             if event.event == "keepalive" {
                 continue;
@@ -867,10 +867,6 @@ where
             if tx.send(response).is_err() {
                 break;
             }
-
-            if done {
-                break;
-            }
         }
     });