我正在尝试基于 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()
}
}