Блог

Главная - Вопросы по программированию - Spring RSocket: @DestinationVariable для метаданных не извлекается (без использования веб-пакетов)

Spring RSocket: @DestinationVariable для метаданных не извлекается (без использования веб-пакетов)

#spring-webflux #messaging #spring-messaging #rsocket #rsocket-java

Вопрос:

Я пытаюсь создать несколько микросервисов на основе Spring и RSocket. В частности, мне нужно передавать файлы между двумя службами, поэтому «клиент» отправляет поток в качестве полезной нагрузки на «сервер». Это отлично работает, но я представляю здесь простой сервис echo, чтобы решить проблему с метаданными.

Проблема, которую я не могу решить, заключается в том, как передать объект JSON в качестве метаданных от клиента на сервер. Я пытаюсь использовать @DestinationVariable в службе сервера, но он всегда имеет значение «{метаданные}», которое является заполнителем в маршруте к конечной точке службы. Я уверен, что мне не хватает чего-то простого, но я этого не вижу. Я надеюсь, что вы могли бы помочь мне разобраться в этом…

Я разместил исходный код этого простого приложения под информацией о взаимодействии ниже.

 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% 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()
    }
}

Ответ №1:

Я думаю, что вы не заполняете переменную назначения. Этот пример из демонстрационного приложения rsocket работает.

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> {

И клиент подключается, заменяя {Имя комнаты} некоторой буквальной строкой

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

Ответ №2:

Я решил эту проблему, но было бы здорово получить подтверждение того, что это лучшее практическое решение… Я надеюсь, что эта информация будет полезна другим…

Таким образом, проблема, по-видимому, заключалась в том, что использование @DestinationVariable возможно только с транспортом WebSockets. Мое приложение использует TCP, поэтому метаданные пришлось обрабатывать немного по-другому. Были внесены следующие изменения в ранее опубликованный код, и RSocket работает должным образом.

 ////////////////////////////////////////////////////////////////////////////////
// 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"}
Метки: