仅将Websockets与Rsocket和Spring Webflux一起使用将消息发送给某些客户端 [英] Send message only to certain client using websockets with Rsocket and Spring Webflux
问题描述
我正在我的一个POC项目中尝试将Rsocket与websocket一起使用.在我的情况下，不需要用户登录.当我从其他服务收到消息时，我只想向某些客户端发送消息.基本上，我的流程是这样的.
I am trying to use Rsocket with websocket in one of my POC projects. In my case user login is not required. I would like to send a message only to certain clients when I receive a message from another service. Basically, my flow goes like this.
Service A Service B
|--------| websocket |------------------| Queue based comm |---------------|
| Web |----------------->| Rsocket server |--------------------->| Another |
| |<-----------------| using Websocket |<---------------------| service |
|--------| websocket |------------------| Queue based comm |---------------|
就我而言，我正在考虑为每个连接和每个请求使用唯一的ID.将两个标识符合并为相关ID，然后将消息发送到服务B ，当我从服务B 收到消息时，便会确定需要将其发送到哪个客户端.现在，我了解到我可能不需要2个服务来执行此操作，但是由于其他一些原因，我正在执行此操作.虽然我对如何实现其他部分有一个粗略的想法.我是Rsocket概念的新手.是否可以使用Spring Boot Webflux，Rsocket和websocket通过特定的ID向唯一的特定客户端发送消息?
In my case, I am thinking to use a unique id for each connection and each request. Merge both identifiers as correlation id and send the message to Service B and when I get the message from Service B figure which client it needs to go to and send it. Now I understand I may not need 2 services to do this but I am doing this for a few other reasons. Though I have a rough idea about how to implement other pieces. I am new to the Rsocket concept. Is it possible to send a message to the only certain client by a certain id using Spring Boot Webflux, Rsocket, and websocket?
推荐答案
基本上，我认为您在这里有两个选择.第一个是过滤来自Service B的通量，第二个是使用RSocketRequester和Map，如@NikolaB所述.
Basically, I think you have got two options here. The first one is to filter the Flux which comes from Service B, the second one is to use RSocketRequester and Map as @NikolaB described.
第一个选项:
data class News(val category: String, val news: String)
data class PrivateNews(val destination: String, val news: News)
class NewsProvider {
private val duration: Long = 250
private val externalNewsProcessor = DirectProcessor.create<News>().serialize()
private val sink = externalNewsProcessor.sink()
fun allNews(): Flux<News> {
return Flux
.merge(
carNews(), bikeNews(), cosmeticsNews(),
externalNewsProcessor)
.delayElements(Duration.ofMillis(duration))
}
fun externalNews(): Flux<News> {
return externalNewsProcessor;
}
fun addExternalNews(news: News) {
sink.next(news);
}
fun carNews(): Flux<News> {
return Flux
.just("new lambo!!", "amazing ferrari!", "great porsche", "very cool audi RS4 Avant", "Tesla i smarter than you")
.map { News("CAR", it) }
.delayElements(Duration.ofMillis(duration))
.log()
}
fun bikeNews(): Flux<News> {
return Flux
.just("specialized enduro still the biggest dream", "giant anthem fast as hell", "gravel long distance test")
.map { News("BIKE", it) }
.delayElements(Duration.ofMillis(duration))
.log()
}
fun cosmeticsNews(): Flux<News> {
return Flux
.just("nivea - no one wants to hear about that", "rexona anti-odor test")
.map { News("COSMETICS", it) }
.delayElements(Duration.ofMillis(duration))
.log()
}
}
@RestController
@RequestMapping("/sse")
@CrossOrigin("*")
class NewsRestController() {
private val log = LoggerFactory.getLogger(NewsRestController::class.java)
val newsProvider = NewsProvider()
@GetMapping(value = ["/news/{category}"], produces = [MediaType.TEXT_EVENT_STREAM_VALUE])
fun allNewsByCategory(@PathVariable category: String): Flux<News> {
log.info("hello, getting all news by category: {}!", category)
return newsProvider
.allNews()
.filter { it.category == category }
}
}
NewsProvider类是您的Service B的模拟，它应返回Flux<>.每当调用addExternalNews时，它将推送allNews方法返回的News.在NewsRestController类中，我们按类别过滤新闻.打开localhost:8080/sse/news/CAR上的浏览器，仅查看汽车新闻.
The NewsProvider class is a simulation of your Service B, which should return Flux<>. Whenever you call the addExternalNews it's going to push the News returned by the allNews method. In the NewsRestController class, we filter the news by category. Open the browser on localhost:8080/sse/news/CAR to see only car news.
如果您想改用RSocket，则可以使用如下方法:
If you want to use RSocket instead, you can use a method like this:
@MessageMapping("news.{category}")
fun allNewsByCategory(@DestinationVariable category: String): Flux<News> {
log.info("RSocket, getting all news by category: {}!", category)
return newsProvider
.allNews()
.filter { it.category == category }
}
第二个选项:
让RSocketRequester与@ConnectMapping一起存储在HashMap中(我使用vavr.io).
Let's store the RSocketRequester in the HashMap (I use vavr.io) with @ConnectMapping.
@Controller
class RSocketConnectionController {
private val log = LoggerFactory.getLogger(RSocketConnectionController::class.java)
private var requesterMap: Map<String, RSocketRequester> = HashMap.empty()
@Synchronized
private fun getRequesterMap(): Map<String, RSocketRequester> {
return requesterMap
}
@Synchronized
private fun addRequester(rSocketRequester: RSocketRequester, clientId: String) {
log.info("adding requester {}", clientId)
requesterMap = requesterMap.put(clientId, rSocketRequester)
}
@Synchronized
private fun removeRequester(clientId: String) {
log.info("removing requester {}", clientId)
requesterMap = requesterMap.remove(clientId)
}
@ConnectMapping("client-id")
fun onConnect(rSocketRequester: RSocketRequester, clientId: String) {
val clientIdFixed = clientId.replace("\"", "") //check serialezer why the add " to strings
// rSocketRequester.rsocket().dispose() //to reject connection
rSocketRequester
.rsocket()
.onClose()
.subscribe(null, null, {
log.info("{} just disconnected", clientIdFixed)
removeRequester(clientIdFixed)
})
addRequester(rSocketRequester, clientIdFixed)
}
@MessageMapping("private.news")
fun privateNews(news: PrivateNews, rSocketRequesterParam: RSocketRequester) {
getRequesterMap()
.filterKeys { key -> checkDestination(news, key) }
.values()
.forEach { requester -> sendMessage(requester, news) }
}
private fun sendMessage(requester: RSocketRequester, news: PrivateNews) {
requester
.route("news.${news.news.category}")
.data(news.news)
.send()
.subscribe()
}
private fun checkDestination(news: PrivateNews, key: String): Boolean {
val list = destinations(news)
return list.contains(key)
}
private fun destinations(news: PrivateNews): List<String> {
return news.destination
.split(",")
.map { it.trim() }
}
}
请注意，我们必须在rsocket-js客户端中添加两件事:SETUP帧中的有效负载以提供客户端ID并注册响应者，以处理RSocketRequester发送的消息.
Note that we have to add two things in the rsocket-js client: a payload in SETUP frame to provide client-id and register the Responder, to handle messages sent by RSocketRequester.
const client = new RSocketClient({
// send/receive JSON objects instead of strings/buffers
serializers: {
data: JsonSerializer,
metadata: IdentitySerializer
},
setup: {
//for connection mapping on server
payload: {
data: "provide-unique-client-id-here",
metadata: String.fromCharCode("client-id".length) + "client-id"
},
// ms btw sending keepalive to server
keepAlive: 60000,
// ms timeout if no keepalive response
lifetime: 180000,
// format of `data`
dataMimeType: "application/json",
// format of `metadata`
metadataMimeType: "message/x.rsocket.routing.v0"
},
responder: responder,
transport
});
For more information about that please see this question: How to handle message sent from server to client with RSocket?
这篇关于仅将Websockets与Rsocket和Spring Webflux一起使用将消息发送给某些客户端的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
