Make sure sse async channel gets closed.

[phronmophobic]

The events async channel does not ever get closed. Closing a channel is the best way to signal that a channel doesn't have any more info. Currently, the channel signals that it's done by putting a :done value on the channel. I think it would be preferable to just close the channel rather than send an unqualified :done as a sentinel, but I did not make that change since it would be backwards incompatible.

Closing the channel when the stream is finished also makes the channel compatible with existing builtin functions. For example:

(def response
  (openai/create-chat-completion {:model "gpt-4o"
                                  :messages [{:role "system" :content "You are a helpful assistant."}
                                             {:role "user" :content "hi"}]
                                  :stream true}
                                 {:api-key openai-key}))


(def messages (async/<!! (async/into [] response))) 

The above works after applying the attached change.

[wkok]

Thanks for your contribution!

I have done a few tests and mostly looks good, but I have noticed a (potential) regression with this example

The first time the go form runs is fine, but any subsequent executions of the go form (wiithout putting something new on the channel) then we seem to enter an infinite loop, which is different behavior than on main - to be honest I've not given it more thought yet, but as you pointed out we should maintain backward compatibility as far as possible.

I am away for 2 weeks and won't be able to look further but maybe you could give it some thought, I am happy to merge when I return

[phronmophobic]

The first time the go form runs is fine, but any subsequent executions of the go form (without putting something new on the channel) then we seem to enter an infinite loop, which is different behavior than on main - to be honest I've not given it more thought yet, but as you pointed out we should maintain backward compatibility as far as possible.

Yes, my change probably would break that example (I haven't tested, but logically the channel will always receive nil once closed).

On the one hand, if lots of folks used this example as a template, then the attached pull request would be breaking change. On the other hand, I do think that stopping on :done rather than a channel closing is non-idiomatic and bad practice.

I would write the example like:

(a/go
  (loop []
    ;; use when-let instead of let
    (when-let [event (a/<! events)]
      ;; when (not= :done event) not needed if `events` closes itself

      ;; Do something with the event token
      (prn event)

      (recur))))

Maybe there's a way to make the change backwards compatible? A backwards compatible option might be to add a :close? option that will close the channel on completion similar to https://clojuredocs.org/clojure.core.async/pipe.

Thanks for the library!

[wkok]

A :close? parameter (with default false) sounds like a sensible approach given the current usage in the wild.

Also, if it's not too much trouble, a basic unit test would be great 🙂

[phronmophobic]

I've rebased onto master and made the following updates:

• Added :stream/close? parameter to optionally enable closing the core.async channel in a backwards compatible way for sse. We talked about using :close?, but I thought :stream/close? might be slightly nicer. I'm happy to use whatever name you prefer
• Added tests

Thanks again @wkok ! Let me know if you'd like any other changes.

[wkok]

Looks good to me 👍

Thanks for your contribution!