interesting behavior of manifold.stream/consume - it ignores buffers, so if your consumer function is slower than the producer the producer will block as if the stream was unbuffered, e.g.

(def stream (s/stream 5))

(future 
  (s/consume 
    (fn [x] 
      (println x) 
      (Thread/sleep 1000)) 
    stream))

(future
  (doseq [n (range 10)] 
    @(s/put! stream n)
    (println "put" n)))

gives

1
put 1
2
put 2
etc

if you swap the consume call with a loop that takes from the stream, the first 5 put!s succeed immediately, filling the buffer, as you would expect.

has anyone managed to get aleph websocket server to aggregate fragmented frames?

so my fix is one line but would probably be something you want to pass in as a config param

in the initialize-websocket-handler when you setup the pipeline right before adding the "websocket-handler" we need to add the "websocket-frame-aggregator"

(.addLast pipeline "websocket-frame-aggregator" (WebSocketFrameAggregator. (* 16 1024 1024)))

(that large number is what would be passed in as a config param I assume)

anyway this totally worked and I can now paste giant images over aleph sockets

@ztellman: forgot to tag you in the above