forked from Azure/azure-sdk-for-js
-
Notifications
You must be signed in to change notification settings - Fork 16
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[OpenAI] Re-implement SSEs (Azure#26704)
### Packages impacted by this PR @azure/openai ### Issues associated with this PR Azure#26376 ### Describe the problem that is addressed by this PR The old implementation of SSEs didn't do the right thing with regard to parsing events spanning multiple stream chunks. It converted every chunk to a string first which is not valid. ### What are the possible designs available to address the problem? If there are more than one possible design, why was the one in this PR chosen? This implementation follows closely the spec in https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation and is largely based on [`@microsoft/fetch-event-source`](https://www.npmjs.com/package/@microsoft/fetch-event-source)'s. I think we should consider moving the implementation to core to be used to interpret responses with content type value of `text/event-stream` and I can file an issue after merging this PR. ### Are there test cases added in this PR? _(If not, why?)_ Yes! ### Provide a list of related PRs _(if any)_ N/A ### Command used to generate this PR:**_(Applicable only to SDK release request PRs)_ ### Checklists - [x] Added impacted package name to the issue description - [ ] Does this PR needs any fixes in the SDK Generator?** _(If so, create an Issue in the [Autorest/typescript](https://github.com/Azure/autorest.typescript) repository and link it here)_ - [x] Added a changelog (if necessary) --------- Co-authored-by: Jeff Fisher <[email protected]>
- Loading branch information
1 parent
adf8bd0
commit a84e331
Showing
30 changed files
with
1,008 additions
and
183 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
35 changes: 35 additions & 0 deletions
35
sdk/openai/openai/sources/customizations/api/getSSEs.browser.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,35 @@ | ||
| // Copyright (c) Microsoft Corporation. | ||
| // Licensed under the MIT license. | ||
|
|
||
| import { StreamableMethod } from "@azure-rest/core-client"; | ||
| import { EventMessage, toSSE } from "./sse.js"; | ||
|
|
||
| async function* toAsyncIterable<T>(stream: ReadableStream<T>): AsyncIterable<T> { | ||
| const reader = stream.getReader(); | ||
| try { | ||
| while (true) { | ||
| const { value, done } = await reader.read(); | ||
| if (done) { | ||
| return; | ||
| } | ||
| yield value; | ||
| } | ||
| } finally { | ||
| reader.releaseLock(); | ||
| } | ||
| } | ||
|
|
||
| async function getStream<TResponse>( | ||
| response: StreamableMethod<TResponse> | ||
| ): Promise<AsyncIterable<Uint8Array>> { | ||
| const stream = (await response.asBrowserStream()).body; | ||
| if (!stream) throw new Error("No stream found in response. Did you enable the stream option?"); | ||
| return toAsyncIterable(stream); | ||
| } | ||
|
|
||
| export async function getSSEs( | ||
| response: StreamableMethod<unknown> | ||
| ): Promise<AsyncIterable<EventMessage>> { | ||
| const iter = await getStream(response); | ||
| return toSSE(iter); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,20 @@ | ||
| // Copyright (c) Microsoft Corporation. | ||
| // Licensed under the MIT license. | ||
|
|
||
| import { StreamableMethod } from "@azure-rest/core-client"; | ||
| import { EventMessage, toSSE } from "./sse.js"; | ||
|
|
||
| async function getStream<TResponse>( | ||
| response: StreamableMethod<TResponse> | ||
| ): Promise<AsyncIterable<Uint8Array>> { | ||
| const stream = (await response.asNodeStream()).body; | ||
| if (!stream) throw new Error("No stream found in response. Did you enable the stream option?"); | ||
| return stream as AsyncIterable<Uint8Array>; | ||
| } | ||
|
|
||
| export async function getSSEs( | ||
| response: StreamableMethod<unknown> | ||
| ): Promise<AsyncIterable<EventMessage>> { | ||
| const chunkIterator = await getStream(response); | ||
| return toSSE(chunkIterator); | ||
| } |
22 changes: 0 additions & 22 deletions
22
sdk/openai/openai/sources/customizations/api/getStream.browser.ts
This file was deleted.
Oops, something went wrong.
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| // Copyright (c) Microsoft Corporation. | ||
| // Licensed under the MIT license. | ||
|
|
||
| import { StreamableMethod } from "@azure-rest/core-client"; | ||
| import { getSSEs } from "./getSSEs.js"; | ||
| import { wrapError } from "./util.js"; | ||
|
|
||
| export async function* getOaiSSEs<TEvent>( | ||
| response: StreamableMethod<unknown>, | ||
| toEvent: (obj: Record<string, any>) => TEvent | ||
| ): AsyncIterable<TEvent> { | ||
| const stream = await getSSEs(response); | ||
| let isDone = false; | ||
| for await (const event of stream) { | ||
| if (isDone) { | ||
| // handle a case where the service sends excess stream | ||
| // data after the [DONE] event | ||
| continue; | ||
| } else if (event.data === "[DONE]") { | ||
| isDone = true; | ||
| } else { | ||
| yield toEvent( | ||
| wrapError( | ||
| () => JSON.parse(event.data), | ||
| "Error parsing an event. See 'cause' for more details" | ||
| ) | ||
| ); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.