Send message only to certain client using websockets with Rsocket and Spring Webflux

Question

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

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?

Answer 1

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.

First option:

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

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.

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

Second option:

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

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?

Answer 2

I didn't yet personally use RSocket with WebSocket transport, but as stated in RSocket specification underlying transport protocol shouldn't even be important.

One RSocket component is at the same time server and a client. So when browsers connect to your RSocket "server" you can inject the RSocketRequester instance which you can then use to send messages to the "client".

You can then add these instances in your local cache (e.g. put them in some globally available ConcurrentHashMap with key of your choosing - something from which you'll know/be able to calculate to which clients should the message from Service B be propagated).

Then in the code where you receive message from Service B just fetch all RSocketRequester instances from the local cache which match your criteria and send them the message.