Clojurians Log v2

Thanks a lot for this, looks neat! Im trying to change this code here: https://github.com/bob-cd/bob/blob/682423a48524a595310fddbc2b38c7093bd495cd/apiserver/src/apiserver/handlers.clj#L388 to use the new features like so:

(defn events
  [{{:keys [^Environment env name]} :stream :as req}]
  (let [{:keys [input-ch close-ch]} (sse/stream! req)
        consumer (.. env
                     consumerBuilder
                     (stream name)
                     (offset (OffsetSpecification/first))
                     (messageHandler
                      (reify MessageHandler
                        (handle [_ _ message]
                          (async/>!! input-ch {:data [(String/new (.getBodyAsBinary message))]}))))
                     build)]
    (async/<!! close-ch) ; To hold on til client disconnect
    (.close consumer)))

Feb 18, 2026 9:18:25 AM io.helidon.common.socket.SocketContext log                                                     WARNING: [0x3ffa0375 0x6395bc02] Request failed: HttpPrologue[protocol=HTTP, protocolVersion=1.1, method=GET, uriPath=/events, query=, fragment=], cannot send error response, as response already sent
io.helidon.http.RequestException: OutputStream already obtained                                                                at io.helidon.http.RequestException$Builder.build(RequestException.java:157)
        at io.helidon.webserver.http.ErrorHandlers.unhandledError(ErrorHandlers.java:205)
        at io.helidon.webserver.http.ErrorHandlers.lambda$handleError$1(ErrorHandlers.java:180)                                at java.base/java.util.Optional.ifPresentOrElse(Optional.java:198)
        at io.helidon.webserver.http.ErrorHandlers.handleError(ErrorHandlers.java:179)
        at io.helidon.webserver.http.ErrorHandlers.runWithErrorHandling(ErrorHandlers.java:113)
        at io.helidon.webserver.http.Filters.filter(Filters.java:83)
        at io.helidon.webserver.http.HttpRoutingImpl.route(HttpRoutingImpl.java:74)
        at io.helidon.webserver.http1.Http1Connection.route(Http1Connection.java:548)
        at io.helidon.webserver.http1.Http1Connection.handle(Http1Connection.java:216)
        at io.helidon.webserver.ConnectionHandler.run(ConnectionHandler.java:190)
        at io.helidon.common.task.InterruptableTask.call(InterruptableTask.java:47)
        at io.helidon.webserver.ThreadPerTaskExecutor$ThreadBoundFuture.run(ThreadPerTaskExecutor.java:240)
        at java.base/java.lang.VirtualThread.run(VirtualThread.java:456)
Caused by: java.lang.IllegalStateException: OutputStream already obtained
        at io.helidon.webserver.http1.Http1ServerResponse.outputStream(Http1ServerResponse.java:445)
        at io.helidon.webserver.http1.Http1ServerResponse.outputStream(Http1ServerResponse.java:234)
        at s_exp.hirundo.http.response$write_body_BANG_.invokeStatic(response.clj:12)
        at s_exp.hirundo.http.response$write_body_BANG_.invoke(response.clj:10)
        at s_exp.hirundo.http.response$set_response_BANG_.invokeStatic(response.clj:37)
        at s_exp.hirundo.http.response$set_response_BANG_.invoke(response.clj:32)
        at s_exp.hirundo.http.routing$set_ring1_handler_BANG_$reify__31804.handle(routing.clj:23)
        at io.helidon.webserver.http.HttpRoutingImpl$RoutingExecutor.doRoute(HttpRoutingImpl.java:174)
        at io.helidon.webserver.http.HttpRoutingImpl$RoutingExecutor.call(HttpRoutingImpl.java:132)
        at io.helidon.webserver.http.HttpRoutingImpl$RoutingExecutor.call(HttpRoutingImpl.java:110)
        at io.helidon.webserver.http.ErrorHandlers.runWithErrorHandling(ErrorHandlers.java:76)
        ... 8 more

you need to return nil from your handler

ah

I need to fix that one, but currently that's how it works

still see it with this:

(defn events
  [{{:keys [^Environment env name]} :stream :as req}]
  (let [{:keys [input-ch close-ch]} (sse/stream! req)
        consumer (.. env
                     consumerBuilder
                     (stream name)
                     (offset (OffsetSpecification/first))
                     (messageHandler
                      (reify MessageHandler
                        (handle [_ _ message]
                          (async/>!! input-ch {:data [(String/new (.getBodyAsBinary message))]}))))
                     build)]
    (async/<!! close-ch) ; To hold on til client disconnect
    (.close consumer)
    nil))

it all works fine from the client perspective though, just that stacktrace on the server logs

hmm it should just work. Could you provide a more minimal repro?

trying

is the way to wait for client disconnect correct? the read from the close-chan?

not sure if im doing that correctly, something i thought of, didnt see it in the examples. this is an endless stream til the client stops

yes, the input-ch also closes on disconnects, on-close will allow you to close a bit earlier potentially

will try to make a minimal repro soon

Thanks. I can look into it later also, I am at work now

same suffering here

I just released a new version: it makes the return value of stream! Closeable, that should be a bit safer (if your handler exited too early you could have errors previously)

you can do (with-open [s (stream! ...)] (do-stuff (:input-ch s)))

the with-open scope would block until processing ends (until response lifecycle is done, either client disconnection or input-ch closed, or error)

It's subject to changes, hence the alpha tag, but it works nicely.

works perfectly fine in all aspects for this simple reitit thing:

(ns main
  (:require
   [clojure.core.async :as async]
   [muuntaja.core :as m]
   [reitit.coercion.malli :as malli]
   [reitit.http :as http]
   [reitit.http.coercion :as coercion]
   [reitit.http.interceptors.exception :as exception]
   [reitit.http.interceptors.muuntaja :as muuntaja]
   [reitit.http.interceptors.parameters :as parameters]
   [reitit.interceptor.sieppari :as sieppari]
   [reitit.ring :as ring]
   [s-exp.hirundo :as srv]
   [s-exp.hirundo.sse :as sse])
  (:gen-class))

(defn events
  [req]
  (with-open [stream (sse/stream! req)]
    (loop [m 1]
      (async/>!! (:input-ch stream) {:data [m]})
      (Thread/sleep 1000)
      (recur (inc m)))
    (async/<!! (:close-ch stream))))

(def routes
  [["/events" {:get events}]])

(def server
  (http/ring-handler
   (http/router
    routes
    {:data {:coercion malli/coercion
            :muuntaja m/instance
            :interceptors [(parameters/parameters-interceptor)
                           (muuntaja/format-negotiate-interceptor)
                           (muuntaja/format-response-interceptor)
                           (exception/exception-interceptor)
                           (muuntaja/format-request-interceptor)
                           (coercion/coerce-exceptions-interceptor)
                           (coercion/coerce-response-interceptor)
                           (coercion/coerce-request-interceptor)]}})
   (ring/routes
    (ring/create-default-handler
     {:not-found (constantly {:status 404
                              :headers {"Content-Type" "application/json"}
                              :body "{\"message\": \"Took a wrong turn?\"}"})}))
   {:executor sieppari/executor}))

(defn -main
  []
  (srv/start! {:http-handler server
               :host "0.0.0.0"
               :port 7777}))

(comment
  (set! *warn-on-reflection* true)

  (def s
    (srv/start! (var server)
                {:host "0.0.0.0"
                 :port 7777}))

  (srv/stop! s))

was thinking if reitit was involved somehow but no

strange, maybe one of the interceptors you use in that app tries to use the outputstream before the sse stream can start

i think its the same set of interceptors, lemme double check

or maybe it sets some header or something

something with the rabbitmq thing, when i swap in the simple events generator as above, not issues. will report back later if ifind something

not seeing it in the stacktrace is also weird