Spring RSocket: @DestinationVariable for metadata is not being extracted (not using WebSockets)

I am trying to build some microservices based upon Spring and RSocket. In particular, I need to stream files between two services, so the 'client' sends a Flux as the payload to the 'server'. This works great, but I am presenting a simple echo service, here, to resolve a metadata issue.

The problem that I cannot resolve is how to pass a JSON object as metadata from the client to the server. I am trying to use the @DestinationVariable in the service of the server, but it is always set to "{metadata}", which is the placeholder in the route to the endpoint of the service. I am sure that I am missing something simple, but I cannot see it. I hope that you might help me to figure this out...

I've posted this simple app's source code below the interaction information, below.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% RSOCKET SERVER TESTING WITH RSC CLI
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
rsc --debug --request --data "howdy" --metadataMimeType=application/json --metadata='{"message":"howdy partner!"}' --route blob.echo.{metadata} tcp://run.local:9888

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% RSOCKET SERVER TESTING WITH RSC CLI -- REQUEST-RESPONSE INTERACTION LOG
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
2021-04-10 19:26:32.769 DEBUG 55560 --- [actor-tcp-nio-2] io.rsocket.FrameLogger                   : sending -> 
Frame => Stream ID: 0 Type: SETUP Flags: 0b0 Length: 75
Data:

2021-04-10 19:26:32.772 DEBUG 55560 --- [actor-tcp-nio-2] io.rsocket.FrameLogger                   : sending -> 
Frame => Stream ID: 1 Type: REQUEST_RESPONSE Flags: 0b100000000 Length: 71
Metadata:
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| fe 00 00 15 14 62 6c 6f 62 2e 65 63 68 6f 2e 7b |.....blob.echo.{|
|00000010| 6d 65 74 61 64 61 74 61 7d 85 00 00 1c 7b 22 6d |metadata}....{"m|
|00000020| 65 73 73 61 67 65 22 3a 22 68 6f 77 64 79 20 70 |essage":"howdy p|
|00000030| 61 72 74 6e 65 72 21 22 7d                      |artner!"}       |
+--------+-------------------------------------------------+----------------+
Data:
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 68 6f 77 64 79                                  |howdy           |
+--------+-------------------------------------------------+----------------+
2021-04-10 19:26:32.917 DEBUG 55560 --- [actor-tcp-nio-2] io.rsocket.FrameLogger                   : receiving -> 
Frame => Stream ID: 1 Type: NEXT_COMPLETE Flags: 0b1100000 Length: 25
Data:
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 7b 22 6d 65 73 73 61 67 65 22 3a 22 68 6f 77 64 |{"message":"howd|
|00000010| 79 22 7d                                        |y"}             |
+--------+-------------------------------------------------+----------------+
{"message":"howdy"}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% RSOCKET SERVER SERVICE LOG
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
ECHO: Received Metadata: {metadata}       <============== This should be 'howdy partner!'
ECHO: Received Payload: howdy

################################################################################
## APPLICATION.PROPERTIES
################################################################################
# Create a server as an independent, embedded RSocket server.
# Besides the dependency requirements, the only required
# configuration is to define a port for that server.
spring.rsocket.server.port=9888
spring.rsocket.server.transport=tcp

# General RSocket settings
spring.main.lazy-initialization=true

# Other Server settings
server.port=8888

# Other Spring settings
spring.application.name=blob-service


@SpringBootApplication
class BlobApplication

fun main(args: Array<String>) {
    runApplication<BlobApplication>(*args)
}

////////////////////////////////////////////////////////////////////////////////
// RSOCKET SERVER CONFIGURATION
////////////////////////////////////////////////////////////////////////////////
@Configuration
class RSocketServerConfig {

    @Bean
    fun rsocketMessageHandler() = RSocketMessageHandler().apply {
        rSocketStrategies = rsocketStrategies()
    }
    @Bean
    fun rsocketStrategies() = RSocketStrategies.builder()
        .encoders { it.add(Jackson2CborEncoder()) }
        .decoders { it.add(Jackson2CborDecoder()) }
        .encoders { it.add(Jackson2JsonEncoder()) }
        .decoders { it.add(Jackson2JsonDecoder()) }
        .routeMatcher(PathPatternRouteMatcher())
        .build()
}

////////////////////////////////////////////////////////////////////////////////
// RSOCKET SERVER CONTROLLER
////////////////////////////////////////////////////////////////////////////////
@Controller()
class BlobRSocketController(val service: BlobService) {

    @MessageMapping("blob.echo.{metadata}")
    fun upload(
        @DestinationVariable("metadata") metadata: EchoRequest,
        payload: String
    ): Mono<EchoResponse> {

        return service.echo(metadata, payload)
    }
}

////////////////////////////////////////////////////////////////////////////////
// RSOCKET SERVER SERVICE INTERFACE
////////////////////////////////////////////////////////////////////////////////
interface StorageService {
    fun echo(metadata: EchoRequest, payload: String): Mono<EchoResponse>
}

////////////////////////////////////////////////////////////////////////////////
// RSOCKET SERVER SERVICE
////////////////////////////////////////////////////////////////////////////////
@Service
class BlobService : StorageService {

    override fun echo(metadata: EchoRequest, payload: String): Mono<EchoResponse> {
        println("ECHO: Received Metadata: ${metadata.message}")
        println("ECHO: Received Payload: $payload")
        return EchoResponse(payload).toMono()
    }
}

Upvotes: 1

Answers (2)

Yuri Schimke

Reputation: 13448

I think you are not filling in the destination variable. This example from the rsocket-demo app is working.

https://github.com/rsocket/rsocket-demo/blob/master/src/main/kotlin/io/rsocket/demo/chat/ChatController.kt

@MessageMapping(value = ["chat/{roomName}"])
  suspend fun room(
    @DestinationVariable roomName: String,
    events: Flow<Event>
  ): Flow<String> {

And client connects replacing {roomName} with some literal string

$ rsocket-cli wss://demo.rsocket.io/rsocket --route chat/hello --channel -i '{"join": {"name": "Yuri"}}'

Upvotes: 1

A Bit of Help

Reputation: 1766

I have resolved this issue, but it would be great to get confirmation that it is the best-practice solution... I hope that this info will be helpful to others...

So, the issue seemed to be that the use of @DestinationVariable is only possible with WebSockets transport. My application is using TCP, so the metadata had to be handled a little differently. The made the following changes to the previously posted code, and RSocket is working properly.

////////////////////////////////////////////////////////////////////////////////
// RSOCKET SERVER CONFIGURATION
////////////////////////////////////////////////////////////////////////////////
@Configuration
class RSocketServerConfig {

    @Bean
    fun rsocketMessageHandler() = RSocketMessageHandler().apply {
        rSocketStrategies = rsocketStrategies()
    }
    @Bean
    fun rsocketStrategies() = RSocketStrategies.builder()
        .encoders { it.add(Jackson2CborEncoder()) }
        .decoders { it.add(Jackson2CborDecoder()) }
        .encoders { it.add(Jackson2JsonEncoder()) }
        .decoders { it.add(Jackson2JsonDecoder()) }
        // ADDED THE FOLLOWING EXTRACTOR TO TELL THE SERVICE HOW TO EXTRACT MY METADATA JSON OBJECT THAT IS NAMED 'metadata'.
        .metadataExtractorRegistry { it.metadataToExtract(MimeTypeUtils.APPLICATION_JSON, EchoRequest::class.java, "metadata") }
        .routeMatcher(PathPatternRouteMatcher())
        .build()
}

////////////////////////////////////////////////////////////////////////////////
// RSOCKET SERVER CONTROLLER
////////////////////////////////////////////////////////////////////////////////
@Controller()
class BlobRSocketController(val service: BlobService) {

    @MessageMapping("blob.echo.{metadata}")
    fun upload(
        @Header("metadata") metadata: EchoRequest,
        // REMOVED THE FOLLOWING LINE, WHICH SEEMS TO ONLY BE USED FOR WEBSOCKET TRANSPORT.
        //@DestinationVariable("metadata") metadata: EchoRequest,
        payload: String
    ): Mono<EchoResponse> {

        return service.echo(metadata, payload)
    }
}

rsc --debug --request --data "howdy" --metadataMimeType=application/json --metadata='{"message":"howdy partner!"}' --route blob.echo.{metadata} tcp://run.local:9888

2021-04-10 21:00:34.507 DEBUG 58657 --- [actor-tcp-nio-2] io.rsocket.FrameLogger                   : sending -> 
Frame => Stream ID: 0 Type: SETUP Flags: 0b0 Length: 75
Data:

2021-04-10 21:00:34.510 DEBUG 58657 --- [actor-tcp-nio-2] io.rsocket.FrameLogger                   : sending -> 
Frame => Stream ID: 1 Type: REQUEST_RESPONSE Flags: 0b100000000 Length: 71
Metadata:
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| fe 00 00 15 14 62 6c 6f 62 2e 65 63 68 6f 2e 7b |.....blob.echo.{|
|00000010| 6d 65 74 61 64 61 74 61 7d 85 00 00 1c 7b 22 6d |metadata}....{"m|
|00000020| 65 73 73 61 67 65 22 3a 22 68 6f 77 64 79 20 70 |essage":"howdy p|
|00000030| 61 72 74 6e 65 72 21 22 7d                      |artner!"}       |
+--------+-------------------------------------------------+----------------+
Data:
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 68 6f 77 64 79                                  |howdy           |
+--------+-------------------------------------------------+----------------+
2021-04-10 21:00:46.807 DEBUG 58657 --- [actor-tcp-nio-2] io.rsocket.FrameLogger                   : receiving -> 
Frame => Stream ID: 1 Type: NEXT_COMPLETE Flags: 0b1100000 Length: 25
Data:
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 7b 22 6d 65 73 73 61 67 65 22 3a 22 68 6f 77 64 |{"message":"howd|
|00000010| 79 22 7d                                        |y"}             |
+--------+-------------------------------------------------+----------------+
{"message":"howdy"}

Upvotes: 0

Spring RSocket: @DestinationVariable for metadata is not being extracted (not using WebSockets)

Answers (2)

Related Questions