Server Sent Events under Http-Kit

[Lars Ole Avery Simonsen]

Hi guys

I am trying to implement Server Sent Event support in a small hobby project based on the http-kit server framework and compojure.

I am still quite new to clojure in general, so it may very well be that I am missing something obvious, but this SSE detail has me stumped.

What I have tried is to take inspiration from the Eventual library for the SSE implementation (which is aimed at jetty). I have changed the Eventual implementation so that it uses the http-kit async-channel implementation for data transmission. What I am seeing is that everything seems to run without errors except for the fact that no data is being received at the EventSource in the browser.

If I allow the sse channel to close immediately after opening (which is pretty useless), the initial message is received at the browser side, but if I keep the connection open, nothing gets received.

Having had a look at the http-kit internals, my suspicion is that the socket doesn't get flushed and so all my attempts at communication are sitting in some buffer somewhere.

All SSE implementations I have looked at for other platforms have explicit calls to flush the socket buffers on the long lived connection after each message transmission.

Does anyone have any suggestions as to a better way to achieve SSE support in a http-kit based server? Or maybe ideas as to what I might be doing wrong?

Thanks in advance.

LOAS

[Colin Yates]

Have you looked at sente?

[Zach Tellman]

I'm not sure about http-kit's streaming response implementation, but in Aleph [1] if you use a stream/channel as a response body each message from the stream will be immediately sent as an HTTP chunk.

Best,

Zach

[1] https://github.com/ztellman/aleph

[Malcolm Sparks]

Here is a solution for SSE over http-kit channels

(defn server-event-source [ch]
	;; Adding a mult here is dangerous because it bleeds the underlying
	;; channel dry. We should provide the mult via modular.async
	(let [m (async/mult ch)]
	(fn [req]
	(let [ch (async/chan 16)]
	(async/tap m ch)
	(with-channel req net-ch
	(on-close net-ch (fn [_]
	(async/untap m ch)
	(async/close! ch)))
	(send! net-ch {:headers headers} false)
	(go-loop []
	(when-let [data (<! ch)]
	(println "Got data! " data)
	(send! net-ch (->message data) false)
	(recur))))))))

https://github.com/juxt/modular/blob/master/modules/http-kit-events/src/modular/http_kit/events.clj

[Zach Tellman]

Does `send!` block if it needs to exert backpressure?  If so, the `go-loop` seems dangerous.  If not, how does backpressure work?

--
You received this message because you are subscribed to the Google
Groups "Clojure" group.
To post to this group, send email to clo...@googlegroups.com
Note that posts from new members are moderated - please be patient with your first post.
To unsubscribe from this group, send email to
clojure+u...@googlegroups.com
For more options, visit this group at
http://groups.google.com/group/clojure?hl=en
---
You received this message because you are subscribed to a topic in the Google Groups "Clojure" group.
To unsubscribe from this topic, visit https://groups.google.com/d/topic/clojure/DbBzM3AG9wM/unsubscribe.
To unsubscribe from this group and all its topics, send an email to clojure+u...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Malcolm Sparks]

http-kit's send! doesn't block normally - 

From http://www.http-kit.org/server.html :-

send!: Sends data to client and returns true if the data was successfully written to the output queue, or false if the channel is closed. Normally, checking the returned value is not needed. This function returns immediately (does not block).

http-kit will attempt to write to the output queue, otherwise will place the message in a pending queue backed by a java.util.LinkedList<ByteBuffer>, so there's no back-pressure (except for that exerted by an OutOfMemory exception)

AFAICT there's no way of finding whether http-kit's output channel is empty or not, that would certainly be a feature I'd like to see. 

Does aleph support this?

[Zach Tellman]

In Aleph, you simply return the channel as the body of the response [1], and messages are consumed only as they're able to be sent over the network.  This means that backpressure "just works" without any need to poll whether downstream buffers are full.  Obviously you can poll using the put-with-timeout mechanism in core.async et al if you want, though.

[1] https://gist.github.com/ztellman/613629a2a32908d096d5

[Lars Ole Avery Simonsen]

It turns out my main problem was in fact a dumb coding error I had in a function wrapping the http-kit send! function, so that I wasn't actually sending anything to the socket under certain circumstances - apart from that the code was seemingly sound.

I appreciate the replies, however. Aleph looks cool - will need to check that out.