AsyncAPI Request/Reply feature

[adamhicks]

AsyncAPI Request/Reply Feature - Design Discussion

I'd like to propose and discuss adding support for AsyncAPI 3.0 Request/Reply patterns to Microcks. This feature would enable mocking of bi-directional async message flows where a service listens for requests on one channel and sends correlated replies on another channel.

The Request/Reply pattern is a common messaging pattern in event-driven architectures and is now officially supported in AsyncAPI 3.0 through the reply object. This discussion outlines the analysis, design options, and implementation approach for bringing this capability to Microcks.

We want to use this feature at Dojo and I'm keen to extend Microcks rather than build an internal tool. If you and/or your team could also benefit from this feature then please use this discussion as an opportunity to share what you would like to see. Feel free to comment on any of my thoughts and assumptions.

---

Background & Use Cases

What is Request/Reply in AsyncAPI?

Request/Reply is an asynchronous communication pattern where:

1. A requestor sends a message to a channel
2. A replier (the mock service) listens on the request channel
3. Upon receiving a request, the replier sends a correlated response to either a general reply channel or a dynamic one specified in the request

Key characteristics:

• Reactive: Responses correlate one to one with requests
• Flexible routing: Reply destination can be static or dynamically specified in the request
• Contextual Content: Reply usually contains information about something in the request
• Not an event: Messages are not an indication of something that has happened

Example

Order Status Queries

A service that responds to requests about an order with the status of that order.

Client   > [orders/status/request] - {"orderId":"123456"}                     > Microcks
Microcks > [orders/status/reply]   - {"orderId":"123456","status":"PENDING"}  > Client

Specification

asyncapi: 3.0.0
info:
  title: Order status example
  version: 1.0.0
  description: Example document for an application that processes order status requests and replies with the order and its status
channels:
  orderStatusRequests:
    address: /orders/status/request
    messages:
      statusRequest:
        $ref: '#/components/messages/statusRequest'
  orderStatusResponses:
    address: /orders/status/reply
    messages:
      statusResponse:
        $ref: '#/components/messages/statusResponse'
operations:
  orderStatus:
    action: receive
    channel:
      $ref: '#/channels/orderStatusRequests'
    messages:
      - $ref: '#/channels/orderStatusRequests/messages/statusRequest'
    reply:
      channel:
        $ref: '#/channels/orderStatusResponses'
      messages:
        - $ref: '#/channels/orderStatusResponses/messages/statusResponse'
components:
  messages:
    statusRequest:
      payload:
        type: object
        properties:
          orderId:
            type: string
    statusResponse:
      payload:
        type: object
        properties:
          orderId:
            type: string
          status:
            type: string

Reply Channels

AsyncAPI 3.0 supports two patterns for specifying reply channels:

1. Static Reply Channel:

channels:
  statusRequests:
    address: /send-requests-here
  statusReplies:
    address: /get-replies-here
operations:
  requestStatus:
    action: send
    channel:
      $ref: '#/channels/statusRequests'
    reply:
      channel:
        $ref: '#/channels/statusReplies'

2. Dynamic Reply Channel:

channels:
  statusRequests:
    address: /send-requests-here
  statusReplies:
    address: null
operations:
  requestStatus:
    action: send
    channel:
      $ref: '#/channels/statusRequests'
    reply:
      channel:
        $ref: '#/channels/statusReplies'
      address:
        description: The response destination is dynamically set according to the `replyTo` field in the request header
        location: "$message.header#/replyTo"  # Dynamic in the request

Correlation ID

Correlation IDs are a way for requestors to link responses to requests that they have made.

AsyncAPI 3.0 supports correlation IDs in both headers and payload. The correlationId object uses runtime expressions to specify where the correlation ID is located.

Correlation ID:

components:
  correlationIds:
    common:
        description: Correlation ID for request/reply matching
        location: "$message.header#/correlationId"
        # location: "$message.payload#/correlationId" if the field is in the payload instead
  messages:
    requestMessage:
      headers:
        type: object
        properties:
          correlationId:
            type: string
            description: A UUID supplied by the sender to determine correlated responses
      payload:
        # ... payload definition
      correlationId:
        $ref: "#/components/correlationIds/common"

    replyMessage:
      headers:
        type: object
        properties:
          correlationId:
            type: string
            description: Copied from the request message header
      payload:
        # ... payload definition
      correlationId:
        $ref: "#/components/correlationIds/common"

Specification with examples

asyncapi: 3.0.0
id: 'urn:io.microcks.example.payment-status'
info:
  title: Payment Status API
  version: 1.0.0
  description: Example showing request/reply pattern

channels:
  statusRequests:
    address: /payment-requests
    messages:
      status.request:
        description: A request message for payment status
        correlationId:
          $ref: '#/components/correlationIds/requestId'
        payload:
          type: object
          properties:
            requestId:
              type: string
            orderId:
              type: string
        examples:
          - name: status-request
            summary: Status request example
            payload:
              requestId: '12345'
              orderId: 'A001'
  statusReplies:
    address: /payment-replies
    messages:
      status.reply:
        description: A reply message with payment status
        correlationId:
          $ref: '#/components/correlationIds/requestId'
        payload:
          type: object
          properties:
            requestId:
              type: string
            orderId:
              type: string
            status:
              type: string
              enum: [pending, completed, failed]
            amount:
              type: number
        examples:
          - name: completed-status-reply
            summary: Completed status reply example
            payload:
              requestId: '{{request.payload/requestId}}'
              orderId: '{{request.payload/orderId}}'
              status: completed
              amount: 250.00

operations:
  requestStatus:
    action: receive
    channel:
      $ref: '#/channels/statusRequests'
    messages:
      - $ref: '#/channels/statusRequests/messages/status.request'
    reply:
      channel:
        $ref: '#/channels/statusReplies'
      messages:
        - $ref: '#/channels/statusReplies/messages/status.reply'

components:
  correlationIds:
    requestId:
      description: Correlation ID based on requestId property
      location: $message.payload#/requestId

---

Implementation Plan

Requirements

To support Request/Reply, we need to:

• Parse and store reply metadata from AsyncAPI specs
• Add consumer capabilities to listen for requests
• Support dynamic reply addressing

To make Request/Reply more usable, we should:

• Construct reply messages using a template (supporting the correlation ID spec)

Domain Model

We'll need some extra models to hold reply info in, these will be populated during AsyncAPI spec parsing.

• ReplyInfo: Contains either reply channel or a dynamic definition of where to find it

• Operation: Needs to hold request/reply info

• RequestReplyEvent: As an alternative to UnidirectionalEvent

• CorrelationID: Contains where to find the correlation ID in the EventMessage

Async Minion

These models will be communicated to the Async Minion which needs to schedule broker specific handlers. We need:

• A singleton manager for request/reply handlers (similar to ProducerScheduler) which can

  • Create a handler for each request/reply operation
  • Start a handler when a request/reply operation is discovered
  • Stop handlers when services are deleted

• Handler classes for each type of broker

  • Handles a single request/reply operation
  • Runs in its own thread
  • Can be started and stopped
  • Consumes from a channel
  • Creates and publishes messages to reply channels

Integration Points

AsyncMockDefinitionUpdater:

• Detect operations with isRequestReply() == true
• Start RequestReplyHandler instead of adding to scheduled production
• Ensure proper cleanup on service updates

ProducerManager:

• Skip request/reply operations in scheduled production loop
• Expose method for one-time message publication (used by handlers)

---

Open Questions

should we use Quartz triggers to schedule responses rather than having a thread for each operation?

My proposal would create a new thread for every Request/Reply operation, if this efficient (enough)?

Can we take advantage of Quartz, given that we're already using it? (I don't know Quartz)

should we extend the reply spec (i.e. add x-microcks-operation) to add dispatcher rules?

I feel like we shouldn't try and do this yet, some of the features could be useful, such as serving different messages based on request payload but some of the options won't make sense for events. e.g. URI_[...], PROXY and PROXY_FALLBACK

what if the user specifies multiple operations that consume from the same channel?

Should we just let the independent handlers race? Should we throw an error and only let one work?

UI?

I don't fully know where event specs are visible on microcks, will we need to make some changes to surface the reply info?

Should we support request on one protocol, reply on another?

I don't think this is a popular use case, but I could be proven wrong.

---

Looking forward to your thoughts! 🚀

[dipak0000812]

Thanks for the detailed write-up — this aligns very well with how async testing is structured today in Microcks.

From reviewing the current async test flow (AsyncAPITestRunner → async-minion → AsyncAPITestManager → MessageConsumptionTask), a request/reply implementation could likely start with a minimal, additive scope:

Suggested first step (MVP):

Detect request/reply semantics from AsyncAPI v3 (operation.reply)

For such operations, have the minion:

Consume a request message

Extract a correlation ID using the AsyncAPI runtime expression

Produce a single reply message on the resolved reply channel

Reuse existing schema validation on the reply

Design notes:

Keeping correlation header-based initially feels like the safest, protocol-agnostic choice

Reusing existing timeout semantics avoids introducing new lifecycle rules

Treating request/reply handlers as opt-in (and excluding them from scheduled production) keeps backward compatibility

I agree this is best approached incrementally — starting with static reply channels and a simple correlation strategy should already unlock many real-world use cases, and richer patterns can follow once this foundation is validated.

Happy to help review or test as the design evolves.

[adamhicks]

Suggested first step (MVP):
...

Yep, agreed on those 👍

Keeping correlation header-based initially feels like the safest, protocol-agnostic choice

We use payload-based at Dojo, so I have an incentive to implement both 😄

Treating request/reply handlers as opt-in (and excluding them from scheduled production) keeps backward compatibility

💯

[lbroudoux]

Thanks, @adamhicks, for initializing this and for all the great materials.

Here are my comments and initial thoughts below in the order of reading:

• I think there are some glitches in the static and dynamic reply channels example - IIRC the operations.channel.reply element should be moved to a channels.reply one.
• I think we should be careful of the send vs receive operation semantics here => Is it correct to say that both can be used in request/reply use-case? Can what we're describing be applied to both?
• Domain model evolution:
  • ReplyInfo: Shouldn't it be tied to an Operation like we have the Bindings currently? Couldn't we consider that an Operation from an event-typed service that has reply info is of specific type request/reply?
  • RequestReplyEvent: I came to the need of something similar while reading again OpenAPI callbacks & webhooks + Sync-to-Async triggers features and design #1039 and thinking about how to represent messages in the case of callbacks and maybe webhooks ... Could we find some common ground here to avoid the proliferation of too many concepts? In the case of callbacks and webhooks, I was thinking much more about a new Exchange implementation that would mix elements from the original interaction (so requests/responses) and correlated async transmissions (so event-based messages). Could we find a common way to represent this?
  • CorrelationID: isn't it a similar concept to dispatcher and dispatching rules? We're doing correlation dispatching based with details on how to acquire the dispatching/correlation information?
• AsyncMinion: good analysis - we need some kind of RequestReplyHandlerManager here ;-)

Regarding the Open Questions:

My proposal would create a new thread for every Request/Reply operation, if this efficient (enough)?

Agree. Let's face it: we shouldn't have hundreds of request/reply handlers running in the minion.

I feel like we shouldn't try and do this yet, some of the features could be useful, such as serving different messages based on request payload but some of the options won't make sense for events. e.g. URI_[...], PROXY and PROXY_FALLBACK

Agree. Moreover, the asynchronous mocking doesn't use dispatchers at all as of today. We could introduce just one (as explained above in my comment on CorrelationID. In the past, I suggested introducing others, like ANY or STREAM, to better control the number and sequence of messages sent by the mock producer, but it didn't resonate with the community.

What if the user specifies multiple operations that consume from the same channel?

We should be able to handle this and take care of how it translates on the concrete binding (typically, it would mean different consumer groups on Kafka).

I don't fully know where event specs are visible on microcks, will we need to make some changes to surface the reply info?

Yes, we'll need to introduce some new elements in the UI.

Should we support request on one protocol, reply on another?

It probably depends on the bindings - IIRC some bindings are defined at the channel levels, some others are only defined at the operation or server levels. We must, in any case, start with a green path that uses the same binding for both request and reply.

Other questions I have:

• What concrete destinations should we use for both listening to incoming requests and producing replies? For mocking, we decided to have Microcks create dedicated destinations to avoid collisions between different versions of the same API or with the application channels. I think it's safe to keep the same approach for the destination the request/reply handler will consume from but what bout the destination for reply? It's probably safe (because of dynamic routing and correlation ID) to use the standard destination/channel provided in the spec here? Thoughts?

• What's the best choice as the first supported protocol for an MVP? I would say Kafka but what do you think?

[adamhicks]

matching request and reply examples must have the same name

I actually didn't realise this is how it worked for request/response, that's simple and easy to use, great 👍

Also, I thought the JSON_BODY would be the only practical way of getting the behaviour mentioned before, now that I know it isn't and the naming will solve most problems, I don't think we'll need to make many (if any) changes to the x-microcks-operation extension here for it to work fine.

[kinlane]

I stumbled on naming too. Once I got handle on a whole new world opened up. I am going to work on naming strategies and stories series to help folks plan better on this front. It saves a lot of headaches and banging your dead.

[lbroudoux]

I don't think we'll need to make many (if any) changes to the x-microcks-operation extension here for it to work fine.

Actually, there are 2 different things to consider for request/response matching:

• How do we gather together requests and responses in the artifacts to present something consistent through Microcks UI (and API of course)? This is to allow people to understand "how it is expected to work" in real life while browsing, discovering the API. This part is solved with the naming convention of the different request and response example fragments: we need the same name to understand it's the same exchange we're representing.
• How do we select at runtime, in the most deterministic way, a response when a request comes in? This is to make the mocks transparent to users and support dynamic behaviour. And this part is solved with a dispatcher and its dispatcher rules. In a perfect world, this aligns with naming conventions, but as we don't know the name of an incoming request, we need another mechanism to compute something that allows us to find the correct response.

So both things are actually needed to have useful mocks that align with what's communicated in the UI.

[adamhicks]

How do we select at runtime, in the most deterministic way, a response when a request comes in?

I'm in agreement (still learning about some of these concepts) that we will need to run events through the dispatcher to categorise the incoming message.

I was trying to express that I hope we wouldn't need to change much in computeDispatchCriteria to get it to work with events.

• QUERY_HEADER can be used on the message headers
• JSON_BODY can be used on the payload
• GROOVY just needs those objects in the right names
• URI_ we already use for the channel name

How do we gather together requests and responses in the artifacts to present something consistent through Microcks UI?
This part is solved with the naming convention of the different request and response example fragments: we need the same name to understand it's the same exchange we're representing.

Understood 👍

[lbroudoux]

Yeah. Got it. Actually, you're right, we'll end up keeping the same logic we have in computeDispatchCriteria(). Sorry if I overflow you with too many details or thinkings 😉

[lbroudoux]

Hey @adamhicks

I scratched my head yesterday about the domain model evolution we can push to have something consistent with the existing part and the one we also have to manage with #1039. I came up with the following design that starts to make sense IMHO.

Here's the classes diagram I can imagine at the moment, to support all our use cases:

classDiagram
   class ReplyInfo {
       +String address
       +String addressLocation
   }
   class Exchange 
   <<abstract>> Exchange
   Service *-- Operation : composition
   Operation *-- Exchange : composition
   class Operation {
       +ReplyInfo replyInfo
   } 
   Exchange <|-- RequestResponsePair
   Exchange <|-- UnidirectionalEvent
   Exchange <|-- RequestReplyEvents
   RequestResponsePair <|-- RequestResponsePairWithCB
   class RequestResponsePair {
     +Request request
     +Response response  
   }
   class UnidirectionalEvent {
     +EventMessage eventMessage
   }
   class RequestReplyEvents {
     +EventMessage request
     +EventMessage reply  
   }
   class RequestResponsePairWithCB {
     +[RequestResponsePair] callbacks 
   }
   class Message 
   <<abstract>> Message
   Message <|-- Request
   Message <|-- Response
   Message <|-- EventMessage
   class Message {
      +String callbackId 
   }
   class Request {
      +String responseId
   }
   class Response {
      +String dispatchCriteria 
   }
   class EventMessage {
      +String dispatchCriteria
      +String replyId
   }

Loading

You'll find the elements from your proposition we discussed: RequestReplyEvents as a new type of exchange, ReplyInfo as a new operation attribute (with a XOR of address and address location). I also think we need to add an extra replyId attribute to the EventMessage persistent entity to track the association between a request event and its matching reply and rebuild the exchange when required.

For OpenAPI callbacks, I imagine introducing the new RequestResponsePairWithCB exchange type to allow integrating callbacks (that are just request/response pairs after all). We'll also need to keep track of the callback identifier (is this a createEvent or an updateEvent callback?) - so I think of introducing a new callbackId at the Message level. I don't like that idea much, and I would prefer something more generic, but naming is hard!

Note

In the case of callbacks, we also need some kind of address location (typically something like {$request.body#/callbackUrl}) to be able to find the callback URL from the original incoming request. I think I could reuse the ReplyInfo operation for this. What do you think?

I am still a bit stuck on the callback-specific question of how to specify/know the delay between the original request processing and the callback submission. This can be a low random number (2-5 secs) in case of a single callback events but we need explicit information if there is more than 1 callback - in order to keep the temporal logic of events (createEvent must be sent back before updateEvent but Microcks can’t guess this based on event names). but that's another story...

For OpenAPI webhooks, I realize that there was, in fact, an additional Operation that also follows the request/response paradigm. The only difference is that the registration/unregistration endpoints are out of scope in the spec.

What do you think? Does it make sense?

[adamhicks]

I think I had misunderstood the way that Operations were stored. I had assumed that we were storing Exchanges as entire mongo documents, with nested messages inside, but now I understand that we store the messages separately and then need to put them into Exchange objects (in memory) when needed.

Because of this misunderstanding, I was also assuming that we could store the message id references inside the operation data and if we needed to find where a message was used, then we'd need to search Operations for it's ID.

As long as I'm understanding correctly now, then, yes I agree we need new references for replies and callbacks, and any other relationships in the future.

I think of introducing a new callbackId at the Message level. I don't like that idea much, and I would prefer something more generic, but naming is hard!

Could we add the callbackId to the Request model instead perhaps? I think something generic might be confusing, if we need a direct relationship to a specific part of an operation (the callback object).

I think the RequestResponsePairWithCB model will contain an array of Callback objects (instead of RequestResponsePairs) which will allow us to store the callbackId.

I think I could reuse the ReplyInfo operation for this. What do you think?

I'm a bit reluctant about reusing ReplyInfo for this. I anticipate needing a place to store channel and binding information for the Operation and I think ReplyInfo will be suited for that. I think it might be neater to store the callback location in my proposed new object Callback above.

What do you think? Does it make sense?

Yes, this makes sense overall. I feel confident that any remaining uncertainties could be worked out over PRs. What do you think?

[lbroudoux]

Could we add the callbackId to the Request model instead perhaps? I think something generic might be confusing, if we need a direct relationship to a specific part of an operation (the callback object).

Yes, I thought the same. I'm experimenting at the moment what would be the most convenient.

I think the RequestResponsePairWithCB model will contain an array of Callback objects (instead of RequestResponsePairs) which will allow us to store the callbackId.

As you said above, in fact Exchanges are not persisted as is. They're rebuild on purpose from request / response so I don't see any other place to persist this.

I think it might be neater to store the callback location in my proposed new object Callback above.

Agree. I think about introducing CallbackInfo as a Map<String, String> because in fact every callback may use a different request-based expression to get the callback url.

Yes, this makes sense overall. I feel confident that any remaining uncertainties could be worked out over PRs. What do you think?

Certainly! I think we're good to go! Actually, I've started the callback implementation this morning as an experiment 😉