Proposal : Support for NATS broker

[RenaudAbsolut]

I'm using this project for my own purposes, and I needed to implement support for the NATS broker.
I followed the structure of the project, adding the common modules in a subfolder asyncapi_python/nats, as well as the corresponding code generator in asyncapi_python_codegen/generator/nats.

This code works for my own use case.
The downside is that much of the code related to consumers, publishers, and code generation is shared between AMQP and NATS, which means my approach results in a lot of duplicated code.

I’d like to know if you would be interested in a pull request to integrate NATS support into this project?

[WiZeYAR]

This project was conceived in order to implement a schema-first version of faststream based on asyncapi spec.
The idea is to have more or less unified dev experience across brokers and programming languages (when python api gets stable).

So, your contribution is welcome, and I'm very open to the idea of adding NATS support :)

[WiZeYAR]

I guess, it is time to find ways to create some set of abstractions, as I've mainly been prototyping for a single broker only. We also need a way to ensure that swapping a broker wouldn't affect the application behavior.

[WiZeYAR]

I now wonder, if a change of a broker should lead to recompilation. Technically, no operations have changed, and thus the routing would be similar.

On the other hand, there might be protocols that support less use cases than others (like mqtt doesn't work with queues), so compiling mqtt as work queue would probably fail, right?

[RenaudAbsolut]

Hello, sorry for the late reply, I'm currently busy with other topics.

I'll try to clean up my NATS support to submit a pull request in the coming days.

To answer your question, my opinion is that since each broker has its own "bindings" section in the AsyncAPI schema, this inevitably leads to adaptations of EndpointParams and Operation for each broker.
This means that switching brokers necessarily requires regenerating the code for receivers and producers.
However, this doesn’t prevent the fact that concepts like queues, pub/sub, channels, RPC, etc., remain common to many brokers, which likely means that a lot of code could be shared between different implementations.

For example, here is a brief overview of the generated code for an consumer Operation pub/sub with message queuing that I can easily bind using the nats.py library.

    op: Any = api.Operation(
        name="receivePing",
        message_type=Ping,
        reply_type=None,
        subject_name="ping.pubsub",
        queue_name="workers",
        reply_name=None,
        op_type="queue-sub",
        )

For MQTT, there simply wouldn't be a concept of "queue" for operations and only "pub/sub" as accepted value for op_type

[WiZeYAR]

@RenaudAbsolut, I've made a small upgrade to abstractions in the past couple of weekends, adding the ground for further features, as well as ease of extensibility.

The code is now extendable in two directions:

• Wire -- a factory pattern class that turns an operation definition from the document into an endpoint. Every 'wire' is a factory that turns operation definition into a protocol specific producer or consumer of messages. Producers implement def send_batch(list[T_Send]) -> None, while consumers implement recv() -> AsyncGenerator[T_Recv] (the generic types are bound to protocols and are obliged to implement some methods, e.g. incoming message can be acked)

wire_a = InMemoryWire()
wire_b = AmqpWire(
    connection_url="amqp://guest:guest@localhost:5672/"
    service_name="test-integration",
)

The wire is added to the application:

app = Application(wire_a)

• Codec -- another factory that gets a root module of the generated code, and maps asyncapi message specs into generated classes. A codec performs encode/decode between the class (like BaseModel) and bytes. Codecs are automatically loaded via CodecRegistry, and are selected based on content type, provided by the message.

Currently, the only codec available is that maps BaseModel to json encoding, but I'm also planning to try other formats (like protobuf as a decent payload carrier)

Are you still interested in adding nats?