Hi, and thanks for reading my thread,
I am currently struggling to find a resolution for the following problem:
Our project setup uses the RSocket protocol for frontend <> bff communication. The frontend uses ‘rsocket-core’ and uses this to send queries/commands/streamrequests (querysubscription).
These RSocket messages hold 1) metadata and 2) payload data.
In our bff, we receive said requests in a generic controller with “socket.query” / “socket.subscription.query” / “socket.command” as paths (irrelevant I suppose for this issue).
An example of a working setup is:
@MessageMapping(QUERY_DESTINATION)
public Mono<SocketResponse<?>> query(
final Query query) {
return queryGateway.query(
query,
Object.class
)
.map(SocketResponse::ok);
}
Here, Query is a marker interface. Query/command objects example is as follows:
public record FindTotalsPerDayQuery(
DateRange appliedRange,
Set<String> someIds
) implements Query {}
Now, this setup is working – but we recently have the added requirement that the RSocket metadata should be available (for security/… purposes) on the (Query/Command)Message.
So for this, we’ve made some slight modification and this does seem to be working:
@MessageMapping(QUERY_DESTINATION)
public Mono<SocketResponse<?>> query(
final Query query,
@Headers Map<String, Object> headers) {
QueryMessage<Query, Object> genericQueryMessage = new GenericQueryMessage<>(query, ResponseTypes.instanceOf(Object.class))
.andMetaData(rSocketMetaDataMapper.mapToMetaData(headers).toMap());
return queryGateway.query(
genericQueryMessage,
Object.class
)
.map(SocketResponse::ok);
So what’s added here?
- Headers input variable – these are the RSocket headers, where we read the RSocket metadata from. Irrelevant for this problem.
- We wrap the input Query into a GenericQueryMessage
o We have to work with Object.class here – due to the generic setup that we are stuck with for now.
o Since it is now a GenericQueryMessage, we can add the metadata on the (Query/Command)Message.
The same setup is done for the command path as well as the subscription path.
This setup does work for queries (and probably commands, but not yet thoroughly (integration) tested).
The problem lies in the subscription path.
We went from:
@MessageMapping(SUBSCRIPTION_QUERY_DESTINATION)
public Flux<SocketResponse<?>> subscriptionQuery(
final Object query,
@Headers Map<String, Object> headers) {
return queryGateway.subscriptionQuery(
query,
ResponseTypes.instanceOf(Object.class)
)
.map(SocketResponse::ok);
}
to:
@MessageMapping(SUBSCRIPTION_QUERY_DESTINATION)
public Flux<SocketResponse<?>> subscriptionQuery(
final Object query,
@Headers Map<String, Object> headers) {
SubscriptionQueryMessage< Object, List< Object >, Object > subGenericQueryMessage = new GenericSubscriptionQueryMessage<>(
query,
ResponseTypes.multipleInstancesOf(Object.class),
ResponseTypes.instanceOf(Object.class)
)
.andMetaData(rSocketMetaDataMapper.mapToMetaData(headers).toMap());
return queryGateway.subscriptionQuery(
subGenericQueryMessage,
ResponseTypes.instanceOf(Object.class)
)
.map(SocketResponse::ok);
}
Things to note:
- Based the ResponseTypes (multipleInstancesOf, …) on this: Reactor Gateways - Axon Reference Guide
Whenever a subscription query is fired, I get a CANCELLED: AXONIQ-5000 error. By research, I see this means no handler can be found.
However when the “exact” same payload (query) is used in a query instead of a subscription, a handler is resolved.
On the axon server I get the following error:
Seeing the fact that the only change is the wrapper around the payload, I am quite confident something is wrong with the way I build the GenericSubscriptionQueryMessage.
However I fail to find a resolution or a path forward.
I am also open to alternatives on how to pass metadata along, but the resources I found indicate this as a way of working.
Thank you for your time,
Rhino
