1

我正在尝试基于 Spring 和 RSocket 构建一些微服务。特别是,我需要在两个服务之间传输文件,因此“客户端”将 Flux 作为有效负载发送到“服务器”。这很好用,但我在这里展示了一个简单的回显服务来解决元数据问题。

我无法解决的问题是如何将 JSON 对象作为元数据从客户端传递到服务器。我正在尝试在服务器的服务中使用@DestinationVariable,但它始终设置为“{metadata}”,这是到服务端点的路由中的占位符。我确信我错过了一些简单的东西,但我看不到它。我希望你能帮助我解决这个问题......

我在下面的交互信息下方发布了这个简单应用程序的源代码。

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

2 回答 2

1

我认为您没有填写目标变量。rsocket-demo 应用程序中的这个示例正在运行。

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

客户端连接用一些文字字符串替换 {roomName}

$ rsocket-cli wss://demo.rsocket.io/rsocket --route chat/hello --channel -i '{"join": {"name": "Yuri"}}'
于 2021-04-11T07:20:03.953 回答
0

我已经解决了这个问题,但是很高兴确认它是最佳实践解决方案......我希望这些信息对其他人有帮助......

因此,问题似乎是 @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"}

于 2021-04-11T04:08:40.687 回答