profunktor/fs2-rabbit

This system is a functional, streaming library for interacting with RabbitMQ, built on the fs2 and Cats Effect ecosystems. It provides type-safe abstractions for publishing and consuming messages, handling AMQP protocol details like headers and acknowledgments through composable algebraic interfaces. The library supports resilient stream execution with automatic recovery and offers modular integrations for JSON serialization via Circe.

19 changes.

The project underwent a major overhaul to migrate to Scala 3 and restructure into a multi-module build, replacing the initial blocking implementation with a granular, streaming-based architecture. This period focused on refactoring core abstractions into composable traits, introducing type-safe AMQP argument handling, and adding resilient stream execution with exponential backoff. Comprehensive test coverage was established for the new configuration models, JSON integration, and connection management logic.

7 changes.

The project introduced a comprehensive, type-safe AMQP model with dedicated encoding and decoding logic, replacing ad-hoc representations with structured case classes and value types. This effort was supported by extensive property-based testing using ScalaCheck to verify type class laws and ensure lossless serialization round-trips. Additionally, the codebase updated its Scala 3 compatibility layer and refined test environment configurations to support these changes.

The testkit module now includes a new package object providing implicit ScalaCheck Arbitrary and Cogen instances for core AMQP model types, including AmqpEnvelope, AmqpProperties, and various AmqpFieldValue subtypes (such as StringVal, IntVal, and TableVal). This enables users to easily generate random test data for these types in property-based tests without manually implementing generators.

testkit · high confidence

Introduced the \ResilientStream\ object in the \dev.profunktor.fs2rabbit.resiliency\ package to provide automatic error recovery for effectful fs2 streams. When a stream fails, it restarts after a delay that increases exponentially based on the retry count (defaulting to 5 seconds for the first retry, then scaling by the square of the attempt number). Fatal errors are logged and re-raised without restart, ensuring that non-recoverable issues do not cause infinite loops.

core/src/main/scala/dev/profunktor/fs2rabbit/resiliency · high confidence

Added Fs2JsonEncoder and Fs2JsonDecoder classes to the json-circe module, providing stream-based JSON transformations for AMQP messages using the Circe library. The decoder converts AMQP envelope payloads into typed case classes, while the encoder serializes case classes within AMQP messages into JSON strings, allowing users to integrate Circe-based JSON handling directly into their fs2 stream pipelines.

json-circe/src/main · high confidence

The repository has been initialized with essential configuration and documentation files. A \.scalafmt.conf\ file is added to enforce code formatting (version 3.11.5, Scala 2.13 source 3 dialect). A \.mergify.yml\ file configures automatic merging of Scala Steward pull requests. Documentation files including \README.md\, \CONTRIBUTING.md\, \CODE\_OF\_CONDUCT.md\, and \LICENSE\ are added. Development tooling includes a \docker-compose.yml\ for running RabbitMQ locally, and Nix flake files (\flake.nix\, \flake.lock\) for the development shell. The \.gitignore\ is updated to ignore Scala/IDE-specific artifacts (\.bsp\, \.metals\, \.bloop\, \target/\) and macOS files.

(repo-wide) · high confidence

The initial version of the Fs2Rabbit library, including its core connection/channel management, consumer/publisher logic, and utility helpers, has been removed from the codebase. The accompanying Demo application that illustrated these capabilities has also been deleted. This change eliminates the previous blocking-style connection and channel creation in favor of the streaming-based approach described in the commit messages, which is likely implemented in subsequent or separate changes.

src/main · high confidence

The library now enforces AMQP protocol compliance for message arguments by introducing a \SafeArg\ type and a \SafeArgument\ type class. This ensures that only supported types (such as String, Boolean, numeric types, Date, List, and Map) can be passed as arguments, preventing runtime errors from unsupported custom classes. The implementation includes implicit conversions to Java maps and specific instances for handling type conversions required by the AMQP protocol.

core/src/main/scala/dev/profunktor/fs2rabbit · high confidence

A new \javaConversion\ object has been introduced in the Scala 3 source directory to provide binary compatibility for Java interop. This object extends \AsJavaExtensions\ and \AsScalaExtensions\ from the standard library, replacing the previous reliance on the \collectionCompat\ library for \.asJava\ instances, ensuring seamless conversion between Scala and Java collections in the Scala 3 build.

core/src/main/scala-3 · high confidence

The core algebra in \core/src/main/scala/dev/profunktor/fs2rabbit/algebra\ has been restructured from a monolithic interface into distinct, focused traits—\Connection\, \Declaration\, \Binding\, \Consume\, \Publishing\, \AckConsuming\, \Acking\, \Deletion\, \Cancel\, \Publish\, and \InternalQueue\—each with its own implementation object (e.g., \ConnectionResource.make\, \Consume.make\) and \mapK\-capable ops wrappers. This change exposes lower-level AMQP operations (such as explicit \basicReject\, \basicNack\, and \addPublishingListener\) and internal queue management, allowing users to compose only the capabilities they need rather than depending on a single all-encompassing client interface.

core/src/main/scala/dev/profunktor/fs2rabbit/algebra · high confidence

The library now provides a comprehensive, type-safe model for AMQP entities in the \dev.profunktor.fs2rabbit.model\ package. This includes new case classes for \AmqpEnvelope\, \AmqpMessage\, \AmqpProperties\, and \Headers\, replacing previous ad-hoc representations. A key addition is the \Headers\ type, which uses a sealed \AmqpFieldValue\ hierarchy (supporting strings, numbers, timestamps, tables, etc.) and implicit \AmqpFieldEncoder\/\AmqpFieldDecoder\ instances to safely encode and decode header values. This change also introduces value types for exchanges, queues, and routing keys, along with codec support for converting between these models and the underlying RabbitMQ Java client types.

core/src/main/scala/dev/profunktor/fs2rabbit/model · high confidence

A new \javaConversion\ object has been introduced in the Scala 2.13-specific source directory to provide \AsJavaExtensions\ and \AsScalaExtensions\. This addition supports the \collectionCompat\ library usage for \.asJava\ instances, ensuring binary compatibility while maintaining the ability to convert between Scala and Java collection types.

core/src/main/scala-2.13 · high confidence

This change introduces new case classes and builder patterns in the config package to manage RabbitMQ connection settings and resource declarations. Fs2RabbitConfig now includes explicit fields for requeueOnReject, requestedHeartbeat, automaticTopologyRecovery, and clientProvidedConnectionName, allowing users to fine-tune connection behavior and recovery. Additionally, declaration.scala provides structured builders (DeclarationQueueConfig, DeclarationExchangeConfig) with dedicated methods for setting durability, exclusivity, auto-deletion, and queue types (Classic, Quorum, Stream), while deletion.scala offers configurable strategies for removing queues and exchanges based on usage and emptiness states.

core/src/main/scala/dev/profunktor/fs2rabbit/config · high confidence

The \core/src/main/scala/dev/profunktor/fs2rabbit/effects\ package introduces new typeclasses and utilities to decouple AMQP operations from specific effect types. \EnvelopeDecoder\ provides a unified way to extract properties, headers, and payloads from AMQP envelopes, including new typed header decoders (\headerAs\, \optHeaderAs\) and deprecating older string/int/long-specific methods. \BoolValue\ adds a typeclass for checking boolean configuration flags (e.g., durable, exclusive), while \Log\ and \StreamEval\ provide abstract interfaces for logging and stream evaluation backed by \cats.effect.Sync\. These changes allow users to compose AMQP interactions more flexibly within their own effect systems.

core/src/main/scala/dev/profunktor/fs2rabbit/effects · high confidence

The RabbitClient construction in the interpreter module has been replaced with a Builder pattern, allowing users to explicitly configure the underlying execution context and thread factory via \withExecutionContext\ and \withThreadFactory\ methods. This change introduces a new \LiveInternalQueue\ implementation for internal message buffering and adds \RabbitClientOps\ to support effect-type transformation via \imapK\ and error handling via \liftAttemptK\. The previous overloaded connection and client creation methods have been removed in favor of this single, configurable builder approach.

core/src/main/scala/dev/profunktor/fs2rabbit/interpreter · high confidence

The core program logic in the \program\ package has been reorganized into distinct modules: \AckConsumingProgram\, \AckingProgram\, \ConsumingProgram\, and \PublishingProgram\. This refactoring separates concerns for acknowledgment handling, consumption, and publishing, while introducing \imapK\ operations on the program interfaces to allow easier transformation between effect types (e.g., from \IO\ to \Task\). Users benefit from a cleaner internal structure that supports more flexible effect system integration without changing the external API surface.

core/src/main/scala/dev/profunktor/fs2rabbit/program · high confidence

The RabbitMQ test environment configuration has been updated to include specific settings for connection management and consumer behavior. The new advanced.config file sets the channel tick interval to 500ms and clears loopback users, while the rabbitmq.conf file defines a consumer timeout of 1000ms. These changes support the termination of streams when the server closes the connection.

rabbit-test-config · medium confidence

The example demos in \examples/src/main/scala\ have been updated to reflect the new client creation API. The \RabbitClient\ is now instantiated using a builder pattern (e.g., \RabbitClient.default(config).resource\), and the configuration object \Fs2RabbitConfig\ now uses \clientProvidedConnectionName\ instead of the previous \clientProvidedName\. Additionally, the \DropwizardMetricsDemo\ demonstrates the new optional \MetricsCollector\ integration, and the \RPCDemo\ and \AckerConsumerDemo\ showcase the updated channel and consumer creation flows.

examples/src/main/scala · high confidence

Added BaseSpec test trait; Added integration and resiliency tests for Fs2Rabbit; Added property-based tests for AMQP model types; Added property-based tests for model type class laws; Added test coverage for AMQP model types and codecs; Added tests for EnvelopeDecoder; Added tests for JSON encoding and decoding; Added unit tests for AMQP field values, properties, and declaration configurations.

The build configuration has been consolidated and updated to modern standards. The SBT version was upgraded from 0.13.8 to 1.13.0, and the \project/plugins.sbt\ file was rewritten to include a comprehensive set of plugins (such as sbt-ci-release, sbt-microsites, sbt-mdoc, and sbt-scoverage) with their latest versions. A new \project/Dependencies.scala\ file centralizes all library versions, setting Cats Effect to 3.7.1, FS2 to 3.14.0, Circe to 0.14.16, and the RabbitMQ client to 5.35.0, among others. Additionally, \project/PrjResolvers.scala\ was added to explicitly define Maven repository sources.

project · high confidence

The project has upgraded its primary Scala version to 3.3.8 while maintaining cross-compilation support for Scala 2.13.18. The build has been restructured into a multi-module architecture, separating the core library, JSON/Circe integration, test kit, examples, and microsite into distinct sub-projects. This change also updates the organization to 'dev.profunktor' and configures the build to use a centralized dependency management file (Dependencies.scala) for libraries such as fs2, cats-effect, and amqp-client.

(dependencies) · high confidence

Written by watchdog.canine.dev from the codebase's own history, inside the signed delivery this page is composed from.

• First survey — no prior run to compare against. CAI 58.

• Code Health 98
• Architecture 97
• Maturity 42
• Readiness 60
• Security 76

• 300 commits — 293 feature/other, 7 fixes

• (repo) — 87 commits
• project/Dependencies.scala — 75 commits
• (root) — 52 commits
• project/plugins.sbt — 39 commits
• project/build.properties — 23 commits
• core/src — 19 commits
• .github/workflows — 4 commits
• examples/src — 1 commit

• fix: Blocking state fix
• fix: Fix backoff logic in ResilientStream
• fix: Fix docs and adds tests for Headers (#975)
• fix: Fix scalafmt
• fix: Fix the release job
• fix: Merge pull request #906 from poohsen/fix-backoff-logic
• fix: Update scala to 3.6.3, cats to 2.12, docker-compose to 3.5, fix pipelines
• change: Add 'Reformat with scalafmt 3.8.6' to .git-blame-ignore-revs
• change: Add Budgetbakers as adopter (#976)
• change: Add Codacy as adopter (#974)
• change: Add Option decoder
• change: Add QueueType model
• change: Add decoder attempt instance
• change: Add docs (#1010)
• change: Add docs generation pipeline
• change: Add headers type (#956)
• change: Add scalafmt workflow (#1007)
• change: Add syntax to simplify usage (#988)
• change: Add tests
• change: Added automaticTopologyRecovery to Fs2RabbitConfig
• …and 280 more

Written by watchdog.canine.dev from the codebase's own history, inside the signed delivery this page is composed from.

profunktor/fs2-rabbit was measured the same way every project in this corpus was: the same rubric, at a pinned commit, with the result published in full. Point a surveyor at a repository you know and see whether you agree with it.