How to use aleph + manifold + gloss together?

[Hanh Huynh Huu]

I tried this sample code:

(ns bitcoin.core
  (:require [aleph.tcp :refer :all]
            [manifold.stream :as s]
            [gloss.core :refer :all]
            [gloss.io :as io]
            [byte-streams :as bs])
  )

(defcodec btc-message-header {:magic :int32-le
                              :command (string :utf-8 :length 12)
                              :length :int32-le
                              :checksum :int32-le })

(defn handler
  [ch _]
  (io/decode-stream ch btc-message-header)
  )

(start-server handler {:port 9340})

But it results in an exception

SEVERE: error in consume
java.lang.IllegalArgumentException: Don't know how to create ISeq from: io.netty.buffer.PooledUnsafeDirectByteBuf
	at clojure.lang.RT.seqFrom(RT.java:505)
	at clojure.lang.RT.seq(RT.java:486)
	at clojure.core$seq.invoke(core.clj:133)
	at clojure.core$empty_QMARK_.invoke(core.clj:5595)
	at gloss.data.bytes.core$create_buf_seq.invoke(core.clj:250)
	at gloss.core.formats$to_buf_seq.invoke(formats.clj:30)
	at gloss.io$decode_stream$f__7517.invoke(io.clj:148)
	at manifold.stream.Callback.put(stream.clj:541)
	at manifold.stream.graph$async_send.invoke(graph.clj:50)
	at manifold.stream.graph$async_connect$this__4017.invoke(graph.clj:172)
	at manifold.stream.graph$async_connect$this__4017$fn__4018$fn__4019.invoke(graph.clj:153)
	at clojure.core$trampoline.invoke(core.clj:5690)
	at manifold.stream.graph$async_connect$this__4017$fn__4018.invoke(graph.clj:153)
	at manifold.deferred.Listener$f__3159__auto____3360.invoke(deferred.clj:151)
	at manifold.deferred.Listener.onSuccess(deferred.clj:151)
	at manifold.deferred.Deferred$fn__3439.invoke(deferred.clj:216)
	at manifold.deferred.Deferred.success(deferred.clj:304)
	at manifold.deferred$success_BANG_.invoke(deferred.clj:177)
	at manifold.stream.core.Stream$fn__4131.invoke(core.clj:117)
	at manifold.stream.core.Stream.put(core.clj:116)
	at manifold.stream.core.Stream.put(core.clj:130)
	at aleph.netty$put_BANG_.invoke(netty.clj:199)
	at aleph.tcp$server_channel_handler$reify__6117.channelRead(tcp.clj:46)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:333)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:319)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:787)
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:130)
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
	at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:116)
	at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
	at java.lang.Thread.run(Thread.java:745)

What should I do instead?

Thanks,

--h

[Hanh Huynh Huu]

This works.

(s/consume (fn [b]
             (-> b bs/to-byte-buffer
                 (#(io/decode btc-message-header % false) )
                 println)) ch)

[ztellman]

Hi Hanh,

Gloss 0.2.3 didn't automatically coerce byte representations, but 0.2.4 does. Your initial code should work fine with that release.  Sorry for any confusion.

Zach

[Hanh Huynh Huu]

Hi Zach, thanks for your quick reply!