如何解决Spring RSocket：未提取元数据的 @DestinationVariable不使用 WebSockets
我正在尝试基于 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()
}
}
解决方法
我认为您没有填写目标变量。 rsocket-demo 应用中的这个示例正在运行。
@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"}}'
我已经解决了这个问题，但如果能确认它是最佳实践解决方案会很棒...我希望这些信息对其他人有帮助...
因此，问题似乎是 @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"}
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。
