Implemented NATS client in julia. It is feature complete and matches performance of reference go implementation. NATS is a message broker similar to Kafka, Rabbit MQ and Redis.
It has all features from existing Core NATS clients written in other langs:
Fast protocol parser - can receive about 1.5M messages / s on my machine what is comparable with go reference implementation.
Handles NATS cluster topology changes dynamically
My use case is to distribute work items to multiple workers ran in Kubernetes cluester.
Next on top of this I am going to work on JetSteram.jl. It is a client for JetStream - native NATS persistance layer. It also have Key-Value storeage feature (like Redis).
This is wonderful, thank you for your work! I’ve been using the nats binary directly (from Yggdrasil) but this makes for some very verbose and inflexible code.
I’m hoping to return to NATS related work sometime in the next 6 months, and I’ll be happy to provide feedback and submit PRs where possible.
Might be cool to try a NATS-backed implementation of AbstractChannel.
It would be quite hard in core NATS, but should be doable in JetStream. I even wrote some code locally to prototype it. JetStream has “workqueue” retention policy and “sealed” property on stream. If this is not enough stream metadata can be used to indicate closed channel.
Hi! I’m the author of the other NATS.jl package! Congratulations for your efforts, your take seems to be very well polished.
I started my take some time ago to test the base NATs protocol but guess I made a mistake in that the client will connect to all the members of a cluster and will distribute messages across them and will listen to all of them, while probably only one connection is ok and the other urls are for failover.
My take was based on in-memory Julia Channels to try to ensure that the client could be used by a multithreaded application safely and that the operations were non-blocking. Is this something that is handled in some way in your package?
If you want some feedback, I’d get rid of all the “default connections” and in general global state outside of the main client itself, and I would keep passing the client explicitely.
I guess I could point to your package from my repo as I wasn’t planning on maintaining this effort.
I won’t complain if you want to register it in the general registry under NATS.jl
Thanks. I saw your package long after I started my development (even gave star to it) so it was hard to contribute to your code at this stage without destroying your API.
Interesting approach, but in practice I don’t think so it will speed up message deliveries or increase robustness. Critical part for performance seems to be protocol parser and buffering of outbound publications. I also had some chats with NATS server authors and really important is to not break “at most once” guarantee, what might be hard if you try to deliver message to multiple servers.
ADR-40 document describes how reconnects should be performed.
Yes! Underneath I use channels as well, connection should be threadsafe. I just found syntax with do more readable. Tests are ran with multithreading on CI, no serious issues so far, but maybe I omitted some use cases.
For multithreaded subscription message handling there is async_handlers (I think will rename to spawn) flag which allows to process each message in a separate task.
I found most of the time I use only single connection, so this default connection was for making code less verbose. I agree it might be over engineering and lead to confusion. Will think about deleting it or provide different syntax or API, something like with block or some other wrapper package. This needs some discussion, maybe will prepare some RFC document with possible solutions I see. Definitely passing connection object (which all other clients do) is not something I want to see in my code, cause this not feel to be Julia way.
Cool, I am pretty determined to invest more time into NATS client.
If more people request it surely will do, but it is too early to do it now I think, maybe in next month or two I will have final API ready (for instance default connection feature redesign). I need also to improve docs and run more tests in Kubernetes and JetStream environment.
I see, I think it’s just a matter of object.method(a, b) that in Julia becomes method(object, a, b), no big deal for me, in every other client from other languages (that I’m aware of), there is no implicit client connection, it’s all bound to objects, I think it would be best to keep it in that way.
I can see in some specific application, e.g. a web server or a worker, the convenience of having methods that just refer to the fact that you just want to publish a message or receive one, and the connection is implicitely the one of the application, but in a library, I think it’s best to pass around everything, especially stateful objects that hold external resources such as tcp connections.
Also, I don’t think you need to provide solutions to the method(main_object, arg1, arg2, …) with blocks or special syntax, hopefully someone else will propose a good solution, or we will just get over it and accept the method(object, a1, a2) syntax instead of the obj.method(…) one.
Also I was rethinking design of this default connection, and it seems what I really want is dynamically scoped variable. It looks like ScopedValues.jl will be available in 1.11 out of the box.
There is nice article linked in PR from Java design JEP 446: Scoped Values (Preview)
Use cases described there are exactly what I had in mind. I created draft PR with implementation, example usage is in PR description.
@Jakub_Wronowski I’m getting around to working with NATS (and hopefully contributing to the package, where possible), however I noticed there is no license in the repository. Would you be willing to license the repository under the MIT license?
This nats_server function is created automatically by build, right?
I think we should create NATSServer.jl with more features to it, might be useful for running tests.
using nats_server_jll
using NATS
nats_server() do cmd
nats_server_process = run(`$cmd`; wait = false)
nc = NATS.connect(; retry_on_init_fail = true)
subscribe(nc, "foo") do msg
@show NATS.payload(msg)
end
for i in 1:10
publish(nc, "foo", "test")
sleep(1)
end
NATS.drain(nc)
kill(nats_server_process)
end
Yes, exactly. Could be nice to have a NATSServer.jl package with additional functions related to nats_server flags like configuration, as well as wiring up its logging to Julia’s robust @log system
Hello @Jakub_Wronowski, I’m trying to use this package, but I realized it’s still unregistered and there’s no releases. I need to run this in a docker container, so in order to prove out consistency, I’d need to fix the version (because the main branch could change between docker builds). Are you able to create a release? Or is there another way to add this package that guarantees the code doesn’t change?