IllegalArgumentException and seemingly empty response body from :raw-stream example in docs

[skylize]

Clojure version 1.11.1
Clojure CLI version 1.11.1.1149
OpenJDK Runtime Environment build 18.0.1.1+2
OpenJDK 64-Bit Server VM build 18.0.1.1+2, mixed mode

Start a server as described earlier in docs, serving a delayed range of numbers at http://localhost:10000/numbers. Open a clean repl and paste in the following:

user=> (require '[aleph.http :as http]
                '[byte-streams :as bs]
                '[manifold.deferred :as d]
                '[manifold.stream :as s])
;; ... warning here about SLF4J defaulting to no-op logger ...
nil

user=> (->> @(http/get "http://localhost:10000/numbers"
                {:query-params {:count 10}})
            :body
            bs/to-line-seq
            (map #(Integer/parseInt %))
            doall)
(1 2 3 4 5 6 7 8 9 10)
;; looking good 🕶️ 

user=> (def raw-stream-connection-pool
         (http/connection-pool
          {:connection-options {:raw-stream? true}}))
#'user/raw-stream-connection-pool

user=> @(d/chain
         (http/get "http://localhost:10000/numbers"
                   {:query-params {:count 10}
                   :pool raw-stream-connection-pool})
         :body
         #(s/map bs/to-byte-array %)
         #(s/reduce conj [] %)
         bs/to-string)
;; BOOM! 💣 
Execution error (IllegalArgumentException) at byte-streams/convert (byte_streams.clj:212).
Don't know how to convert class clojure.lang.PersistentVector into class java.lang.String`

If I strip off the call to bs/to-string it no longer throws, but yields an empty vector.

user=> @(d/chain
         (http/get "http://localhost:10000/numbers"
                   {:query-params {:count 10}
                   :pool raw-stream-connection-pool})
         :body
         #(s/map bs/to-byte-array %)
         #(s/reduce conj [] %))
[]

[arnaudgeiser]

Just took a quick look.
Running your code ends up on this Exception being logged on my side:

[aleph-netty-client-event-pool-11] ERROR manifold.stream - error in stream handler
java.lang.IllegalArgumentException: Don't know how to convert class io.netty.buffer.PooledSlicedByteBuf into class [B
	at byte_streams$convert.invokeStatic(byte_streams.clj:212)
	at byte_streams$convert.invoke(byte_streams.clj:173)
	at byte_streams$to_byte_array.invokeStatic(byte_streams.clj:797)
	at byte_streams$to_byte_array.invoke(byte_streams.clj:789)
	at byte_streams$to_byte_array.invokeStatic(byte_streams.clj:792)
	at byte_streams$to_byte_array.invoke(byte_streams.clj:789)
	at manifold.stream$map$fn__20605.invoke(stream.clj:621)
	at manifold.stream.Callback.put(stream.clj:454)
	at manifold.stream.graph$async_send.invokeStatic(graph.clj:82)
	at manifold.stream.graph$async_send.invoke(graph.clj:78)
	at manifold.stream.graph$async_connect$this__10195.invoke(graph.clj:208)
	at manifold.stream.graph$async_connect$this__10195$fn__10196$fn__10197.invoke(graph.clj:189)
	at clojure.core$trampoline.invokeStatic(core.clj:6310)
	at clojure.core$trampoline.invoke(core.clj:6299)
	at manifold.stream.graph$async_connect$this__10195$fn__10196.invoke(graph.clj:189)
	at manifold.deferred.Listener.onSuccess(deferred.clj:220)
	at manifold.deferred.Deferred$fn__9541.invoke(deferred.clj:399)
	at manifold.deferred.Deferred.success(deferred.clj:399)
	at manifold.deferred$success_BANG_.invokeStatic(deferred.clj:244)
	at manifold.deferred$success_BANG_.invoke(deferred.clj:241)
	at manifold.deferred$eval9695$chain_SINGLEQUOTE____9716.invoke(deferred.clj:755)
	at manifold.deferred$eval9695$subscribe__9696$fn__9701.invoke(deferred.clj:716)
	at manifold.deferred.Listener.onSuccess(deferred.clj:220)
	at manifold.deferred.Deferred$fn__9549.invoke(deferred.clj:401)
	at manifold.deferred.Deferred.success(deferred.clj:401)
	at manifold.deferred$success_BANG_.invokeStatic(deferred.clj:246)
	at manifold.deferred$success_BANG_.invoke(deferred.clj:241)
	at manifold.stream.default.Stream$fn__10338.invoke(default.clj:173)
	at manifold.stream.default.Stream.put(default.clj:159)
	at manifold.stream.default.Stream.put(default.clj:195)
	at manifold.stream.BufferedStream.put(stream.clj:880)
	at aleph.netty$put_BANG_.invokeStatic(netty.clj:295)
	at aleph.netty$put_BANG_.invoke(netty.clj:294)
	at aleph.http.client$raw_client_handler$reify__26124.channelRead(client.clj:146)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:327)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:299)
	at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:722)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:658)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:584)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:496)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at manifold.executor$thread_factory$reify__8907$f__8908.invoke(executor.clj:47)
	at clojure.lang.AFn.run(AFn.java:22)
	at java.base/java.lang.Thread.run(Thread.java:833)
Execution error (IllegalArgumentException) at byte-streams/convert (byte_streams.clj:212).
Don't know how to convert class clojure.lang.PersistentVector into class java.lang.String

It's like the byte_streams conversion from ByteBuf into array-class is not working as expected.

[arnaudgeiser]

Got it probably, you have to use clj-commons/byte-streams instead of byte-streams which has been deprecated.
aleph defines the conversion on clj-commons/byte-streams now...

@KingMob : I think you might be interested by this, seems like a breaking change for people using byte-streams instead of clj-commons/byte-streams.

[arnaudgeiser]

We duplicated byte-streams into clj-commons/byte-streams but we should define the conversions on a single place instead.

https://github.com/clj-commons/byte-streams/blob/master/src/byte_streams.clj#L60-L62

[KingMob]

I'd rather not add extra work to keep a deprecated namespace in sync. What about potemkin/import-vars?

[skylize]

Thanks. Works with the new namespace.

In response I added a related issue upstream Deprecation of byte-streams namespace is undocumented

[KingMob]

So, the documentation is fixed by #60.

Longer term, I don't want to manually keep the old ns in sync, but we have other options. One is something like using (require '[clj-commons.byte-streams :refer :all]). The other is using Potemkin's import-vars. Thoughts?

Potemkin is more powerful, since it handles watching changes to the underlying vars, ferrying over metadata, etc. but require is simpler.

[arnaudgeiser]

AFAIK potemkin is not a byte-streams dependency currently.
I would keep it that that way and just do (require '[clj-commons.byte-streams :refer :all]).

[KingMob]

@arnaudgeiser Well, it's not a direct dep, but Zach has bits and pieces of Potemkin's code all over his repos. E.g., check utils.clj and you'll find early proto-versions of defprotocol+, deftype+, defrecord+, definterface+, and doit. I think he kept copying his util fns around and eventually built potemkin out of them.

If anything, I wonder if using potemkin might be preferable given that, since its versions of fns are smarter. For example, the byte-streams version of defprotocol+ never lets you redefine the protocol at all, which is awkward at the REPL; the potemkin version refuses to redefine it only if the bodies are the same, which is actually superior to the clojure core defprotocol (since it preserves equality of existing instances if the definition is the same).

Regardless, I'm fine with keeping it simple and doing :refer :all

[arnaudgeiser]

Actually, I don't have any strong opinions here.
Both of them are fine for me.

[KingMob]

@arnaudgeiser Silly me, refer :all doesn't work. It updates the local ns-map lookup table so you can find external vars, but it doesn't recreate those vars in the current ns, so everything else breaks. Potemkin is the only option, unless I want to make a zillion entries like (def foo clj-commons.byte-streams/foo)

[KingMob]

@arnaudgeiser After chatting with borkdude, I think we should just wait and see if this is even worth the hassle.