fs2-nats

A functional, streaming NATS client for Scala 3 — built on Cats Effect 3 and FS2.

Every subscription is a Stream. Every connection is a Resource. Nothing happens until you ask it to.

fs2-nats lets you talk to NATS the functional way. Subscriptions are native fs2.Streams, the client is a Resource that cleans up after itself, reconnection and backpressure are built in, and the whole surface — core pub/sub, request/reply, JetStream, Key-Value, Object Store, and micro services — stays inside Cats Effect from the first byte to the last. No callbacks, no hidden threads, no surprises.

Highlights

• Pure functional — built entirely on Cats Effect 3 and FS2, referentially transparent end to end
• Streaming first — every subscription is an fs2.Stream[F, NatsMessage]
• Request/Reply — shared-inbox request primitive with no-responders fast-fail
• JetStream — streams, consumers, persistent publish (PubAck), pull & push consume with full ack semantics
• Key-Value — buckets over JetStream with a Direct Get fast read path, optimistic concurrency, history, and watch
• Object Store — large binary objects chunked over JetStream with streaming put/get, SHA-256 verification, links, and watch
• Micro Services — typed request/reply endpoints (ADR-32) with compile-time subject patterns, typed errors, discovery, and per-endpoint stats
• Headers — full NATS 2.2+ headers support (HPUB/HMSG)
• Backpressure — configurable slow-consumer policies (block / drop / error)
• Reconnection — exponential backoff with full jitter, automatic subscription replay, and cluster failover
• Authentication — token, user/password, NKey (Ed25519), and decentralized JWT (.creds)
• TLS & mutual TLS — standard INFO-then-upgrade TLS with a caller-provided TLSContext

Compatibility

Core messaging, headers, and JetStream work against NATS Server 2.2+. The Key-Value and Object Store features use the JetStream Direct Get API, which requires NATS Server 2.9+.

Installation

sbt

libraryDependencies += "de.thatscalaguy" %% "fs2-nats" % "0.4.0"

Mill

ivy"de.thatscalaguy::fs2-nats:0.4.0"

scala-cli

//> using dep de.thatscalaguy::fs2-nats:0.4.0

Hello, NATS

Start a server with docker run -p 4222:4222 nats:latest, then:

import cats.effect.IO
import com.comcast.ip4s.{Host, Port}
import fs2.Chunk
import fs2.nats.client.{ClientConfig, NatsClient}

val config = ClientConfig(
  host = Host.fromString("localhost").get,
  port = Port.fromInt(4222).get
)

val program: IO[Unit] =
  NatsClient.connect[IO](config).use { client =>
    client.subscribe("hello.world").use { messages =>
      for
        _   <- client.publish("hello.world", Chunk.array("Hello, NATS!".getBytes))
        msg <- messages.take(1).compile.lastOrError
        _   <- IO.println(s"Received: ${msg.payloadAsString}")
      yield ()
    }
  }

Where to next

• Getting Started — connect, publish, subscribe, wildcards, queue groups, request/reply.
• JetStream — streams, persistent publish, and pull/push consumers.
• Key-Value Store — buckets, optimistic concurrency, history, and watch.
• Object Store — streaming large binary objects with SHA-256 verification.
• Micro Services — typed request/reply endpoints with discovery and stats.
• Authentication & TLS — token, user/password, NKey, .creds, and (mutual) TLS.

The code examples on every page are compiled against the published API as part of the build, so they stay in step with the library. Complete runnable programs for every module live in the examples/ directory on GitHub, and each feature page links to its example.