🌟 Roadmap 🌟

[Lancetnik]

Roadmap

We know that some users are waiting for new broker support (like SQS/MQTT/etc.), but please be patient a bit longer.

Due to the inner project structure becoming much harder to change with each new broker, we are now focusing on stabilizing the inner and outer API. Then we will be able to promote new broker support in an easier way (like we used to do before).

Road to 1.0.0

Polish Current Features

First of all, we should polish already existing brokers and core features:

• Complete Observability features
  • Add baggage feature support to tracing (assigned to @draincoder, Add OTel baggage support #1692)
  • Correct spans linking in batches case (assigned to @draincoder, fix: correct spans linking in batches case #1532)
  • Prometheus Middleware to collect metrics (Feature: Prometheus Middleware to collect metrics #1665)
  • Healthcheck mechanism for k8s-probes (Feature: add K8S probes checker API #1181)
  • Complete observability documentation
• Strong Pseudo-random values generation (like message IDs)
  • Add Nuid or another pseudo-random algorithm implementation (Feature: more stronger unique RPC subject names #1386)
  • Pass this object throughout the whole application to all places generating random values (0.6.0 #1779)
• Extended RMQ features (feat: extented RMQ features #1522)
  • Connections and channels pool support (Feature: use aio_pika.Pool to connect RMQ #975)
  • Allow using a new channel for specific requests
  • Concurent RPC requests (using channel pool) (Feature: Allow concurrent RPC calls on rabbit implementation #1318)
  • Publish Transactions support
• Extended NATS features
  • Suitable way to subscribe on multiple subjects in JetStream (Feature: support NATS filter subscription #1518)
  • Create KeyValue adapter to provide custom methods
  • Create ObjectStore adapter to provide custom methods
  • In-memory KeyValue and ObjectStore tests
• Extended Kafka features
  • Kafka manual commit support (Bug: Event ack'd after calling KafkaMessage.nack with auto_commit=False #1001)
  • Transactions support (Feature: add Kafka transactions support #1157)
  • Pattern subscription
• Extended Redis support
  • Add Redis Cluster support (Feature: Adds support for the Redis cluster #1170)
  • Concurrent messages processing (feature: concurrent Redis consuming #1507)

Inner Structure Refactoring

The next step is making Context non-global: we should pass it throughout the whole application from FastStream object -> Broker -> Subscriber -> Call. This step is also related to FastDepends 3.0 migration, and we can refactor dependency_overrides as well, and finally make Pydantic replaceable in serialization cases.

• Optional Pydantic (0.6.0 #1779)
  • Migrate to FastDepends 3.0 (0.6.0 #1779)
  • Provide an option to use custom Serializer Protocol implementation (0.6.0 #1779)
• Local dependency-overrides (0.6.0 #1779)
  • Migrate to FastDepends 3.0 (0.6.0 #1779)
  • Pass dependecy-overrides container from FastStream to the final consumers (0.6.0 #1779)
  • Complete Depends documentation by overrides section
• Local context (0.6.0 #1779)
  • Refactor all parsers to get regex in init instead of using context
  • Refactor ContextType to use local context (0.6.0 #1779)
  • Pass context to FastDepends as kwargs (0.6.0 #1779)
  • Pass context from FastStream to the final consumers (0.6.0 #1779)
  • Refactor logging to not use raw context (0.6.0 #1779)
• Finally, after all these steps, we can finalize the Middleware Public API (0.6.0 #1779)
  • Use Response DTO instead of options (0.6.0 #1779)
  • Add local context as an init option (0.6.0 #1779)

New in Testing

We plan to refactor the In-Memory TestClient to make it side-effect-less (currently, it has some problems with reusing).

• New TestClient logic (Bug: Adding a subscriber to a TestRabbitBroker persists between test runs #1036, Bug: starting the broker breaks in-memory behaviour #1479)
• Add tests for context (Support checking keys in tests #557)

New Brokers Support

Finally, we can take a break from refactoring and add new broker support:

• SQS support (SQS support #794)
• MQTT support (MQTT support #956)

1.0.0 Finalization

At the end of the year (hopefully), we plan to release the first major version - 1.0.0

• Public Multi-broker support (Using multiple brokers #526)
  • AsyncAPI schema merging
  • Cross-brokers In-Memory Tests
  • Correct logging resolution
• Extended subscriber API
  • broker.subsciber("test").get_one() - get concrete messages support (Add broker.subscriber.get_one() #1726)
  • async for msg in broker.subscriber("test") - messages iterator support

Community

Additionally, we are planning to start our Experts activity and provide users with a roadmap to study FastStream, a form to approve your expertise, and a public page with all approved experts to encourage people to help each other (and boost their resumes with the FastStream expert mark).

• Launch Experts initiative

AsyncAPI

Last but not least, we should migrate to the new major AsyncAPI version

• AsyncAPI 3.0 support (AsyncAPI 3.0 support #980, 0.6.0 #1779)
• AsyncAPI 2.6 compatibility (0.6.0 #1779)

1.0.0 +

We are not looking too far ahead, but we have few thoughts:

• RabbitMQ Streams binary protocol support
• Generate FastStream project from AsyncAPI specification
• Enable users to publish a test message from AsyncAPI Web right to the app (Feat: React children components support asyncapi/asyncapi-react#840)
  • With In-memory execution mode
  • With real broker publishing

Eventhough brokers are good, brokerless is much better, so ZMQ support looks promising to us

• ZMQ support (ZMQ support #1142)

And the last thing - we want to provide you with some persistence layer (like Faust table or smth) to allow windowing incoming messages stream.

• Incoming stream windowing

P.S. This plan is still pretty raw, and we can swap any of these points, but this is the general direction.

[r0kk]

Hi thank you for sharing the roadmap, I would like to point out that community can already use MQTT protocol since RabbitMQ broker supports MQTT protocol via plugin. This means that FastStream can be used for with MQTT protocol. Maybe tutorial or announcement could inform community about this option.

Thank you for all contributions.

[Lancetnik]

@r0kk thank you for your kind words! But I am not sure, that plugin is full-MQTT compatible. Is it suitable for all MQTT cases? Did you test FastStream with it?
Anyway, we plan to implement MQTT support right after 0.6.0 release (it should be in a few months), it's not too long to wait