Merge pull request #2 from martosaur/am-expose-parser

Expose Parser and add OpenAI example in usage guide

Author: benjreinhart

README.md

@@ -54,6 +54,21 @@ IO.inspect(events)
 # [%{event: "message", data: "{\"complete\":true}"}]
 ```
 
+
+Alternatively, you can manually parse chunks using `ServerSentEvents.Parse.parse/2`:
+
+```elixir
+state = ServerSentEvents.Parser.new()
+{events, state} = ServerSentEvents.Parser.parse(state, "event: event\ndata: {\"complete\":")
+IO.inspect(events)  # []
+
+{events, state} = ServerSentEvents.Parser.parse(state, "true}\n\nevent: event\ndata: {")
+IO.inspect(events)  # [event: "event", %{data: "{\"complete\":true}"}]
+
+{events, state} = ServerSentEvents.Parser.parse(state, "\"key\":\"value\"}\n\n")
+IO.inspect(events)  # [event: "event", %{data: "{\"key\":\"value\"}"}]
+```
+
 Events are maps that always include `:data`, and may also include `:id`, `:event`, or `:retry`.
 The `:id`, `:event`, and `:data` values are binaries. The `:retry` value is a non-negative
 integer when present.

guides/usage.livemd

@@ -3,7 +3,7 @@
 ```elixir
 Mix.install([
   {:req, "~> 0.5.17"},
-  {:server_sent_events, "1.0.0"}
+  {:server_sent_events, "1.1.0"}
 ])
 ```
 
@@ -59,3 +59,59 @@ Enum.each(stream, fn event ->
   end
 end)
 ```
+
+## Req+OpenAI+SSE Parse
+
+To run this livebook, you must have a valid `OPENAI_API_KEY` environment variable enabled for this livebook.
+
+```elixir
+defmodule OpenAI do
+  def responses(request, callback \\ &Function.identity/1) do
+    Req.new(
+      base_url: "https://api.openai.com",
+      auth: {:bearer, System.fetch_env!("LB_OPENAI_API_KEY")},
+      method: :post,
+      url: "/v1/responses",
+      json: Map.put(request, :stream, true),
+      into: fn
+        {:data, data}, {req, %{status: 200} = res} ->
+          state = Req.Request.get_private(req, :sse_state, ServerSentEvents.Parser.new())
+          {events, state} = ServerSentEvents.Parser.parse(state, data)
+          req = Req.Request.put_private(req, :sse_state, state)
+  
+          res =
+            Enum.reduce(events, res, fn
+              %{data: encoded, event: "response.output_text.delta"}, res ->
+                %{"delta" => delta} = JSON.decode!(encoded)
+                callback.(delta)
+                res
+  
+              %{data: encoded, event: "response.completed"}, res ->
+                %{"response" => response} = JSON.decode!(encoded)
+                %{res | body: response}
+  
+              _, res ->
+                res
+            end)
+  
+          {:cont, {req, res}}
+
+        {:data, data}, {req, res} -> {:cont, {req, %{res | body: data}}}
+      end
+    )
+  end
+end
+```
+
+In this example we use a different Req's streaming style. Instead of setting `into: :self` and decoding a stream of incoming messages, we equip Req with 2-arity callback that is called on each chunk. In the callback, we use `ServerSentEvents.Parser` to parse incoming chunks while maintaining parser state in request struct. For each parsed `response.output_text.delta` event, we pass it to a client-provided callback, which can, for example, send delta to a LiveView process. Finally, once we receive `response.completed` event with a complete response body, we use this body as a final request result. This is useful if the caller wants to access final response body in addition to streamed deltas.
+
+```elixir
+%{
+  model: "gpt-5.4-nano",
+  reasoning: %{effort: "none"},
+  instructions: "You are a poet, fluent in multiple languages and emoji usage.",
+  input: "Write a poem across multiple languages and emojis!"
+}
+|> OpenAI.responses(&IO.write/1)
+|> Req.request!()
+```

lib/server_sent_events.ex

@@ -45,7 +45,7 @@ defmodule ServerSentEvents do
   """
   @spec decode_stream(Enumerable.t()) :: Enumerable.t(event())
   def decode_stream(stream) do
-    Stream.transform(stream, %Parser{phase: :start}, fn chunk, state ->
+    Stream.transform(stream, Parser.new(), fn chunk, state ->
       Parser.parse(state, chunk)
     end)
   end

lib/server_sent_events/parser.ex

@@ -1,12 +1,9 @@
 defmodule ServerSentEvents.Parser do
-  @moduledoc false
+  @moduledoc """
+  Low level SSE parser.
 
-  @type event :: %{
-          required(:data) => binary(),
-          optional(:event) => binary(),
-          optional(:id) => binary(),
-          optional(:retry) => non_neg_integer()
-        }
+  See [Usage Guide](guides/usage.livemd) for usage example.
+  """
 
   @type state :: %__MODULE__{
           phase: :start | :field | :key | :value_start | :value | :skip_line | :cr,
@@ -17,10 +14,54 @@ defmodule ServerSentEvents.Parser do
 
   defstruct [:phase, :key, :value, :event]
 
+  @doc """
+  Creates new Parser state that can be then passed to `parse/2`
+
+  ## Examples
+
+      iex> ServerSentEvents.Parser.new()
+      %ServerSentEvents.Parser{phase: :start, key: nil, value: nil, event: nil}
+  """
+  @spec new() :: state()
+  def new() do
+    %__MODULE__{phase: :start}
+  end
+
+  @doc """
+  Parses binary chunk and returns complete events and an active state
+
+  ## Examples
+
+      iex> ServerSentEvents.Parser.parse("id: 1\\nevent: message\\nretry: 5000\\ndata: hello\\n\\n")
+      {[%{data: "hello", id: "1", retry: 5000, event: "message"}],
+      %ServerSentEvents.Parser{phase: :field, key: nil, value: nil, event: nil}}
+  """
+  @spec parse(input :: binary()) :: {[ServerSentEvents.event()], state()}
   def parse(input) when is_binary(input) do
-    parse(%__MODULE__{phase: :start}, input)
+    new() |> parse(input)
   end
 
+  @doc """
+  Parses binary chunk using existing parser state and returns complete events
+  and an updated state
+
+  ## Examples
+
+      iex> state = ServerSentEvents.Parser.new()
+      %ServerSentEvents.Parser{phase: :start, key: nil, value: nil, event: nil}
+      iex> {_, state} = ServerSentEvents.Parser.parse(state, "id: 1\\nevent: message\\n")
+      {[],
+      %ServerSentEvents.Parser{
+        phase: :field,
+        key: nil,
+        value: nil,
+        event: %{id: "1", event: "message"}
+      }}
+      iex> {_, _state} = ServerSentEvents.Parser.parse(state, "retry: 5000\\ndata: hello\\n\\n")
+      {[%{data: "hello", id: "1", retry: 5000, event: "message"}],
+      %ServerSentEvents.Parser{phase: :field, key: nil, value: nil, event: nil}}
+  """
+  @spec parse(state(), input :: binary()) :: {[ServerSentEvents.event()], state()}
   def parse(%__MODULE__{phase: phase, key: key, value: value, event: event}, input)
       when is_binary(input) do
     parse(input, phase, key, value, event, [])

test/server_sent_events/parser_test.exs

@@ -1,6 +1,8 @@
 defmodule ServerSentEvents.ParserTest do
   use ExUnit.Case, async: true
 
+  doctest ServerSentEvents.Parser
+
   alias ServerSentEvents.Parser
 
   test "basic example" do