RSocket Rxjs Websocket Client

based on rsocket-js

setup

const socketClient = new RxRSocketClient("ws://localhost:8080");

or

const socketClient = new RxRSocketClient("ws://localhost:8080", 
    {
        // 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',
    });

connection

socketClient.connect();

connection status change listener

can use for reconnection logic or other things

socketClient.onConnectionStatusChange(status => {
   console.log(status);
});

requests examples

fireAndForget

socketClient.fireAndForget({message: "Hello"}, "fire.and.forget");

requestResponse

const stream$ = socketClient.requestResponse({message: "Hello"}, "request.response");
stream$.subscribe(value => {
    console.log(value);
});

requestStream

const stream$ = socketClient.requestStream({message: "Hello"}, "request.stream");
stream$.subscribe(value => {
    console.log(value);
});

requestChannel

const mouse$ = fromEvent(document, 'mousemove')
                .pipe(
                    map(e => ({x: e.x, y: e.y}))
                );

const stream$ = socketClient.requestChannel(mouse$, "request.channel");
stream$.subscribe(value => {
    console.log(value);
});

Java Spring Backend Example

pom.xml

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-rsocket</artifactId>
</dependency>

application.properties

spring.rsocket.server.port=8080
spring.rsocket.server.transport=websocket

Controller.class

@MessageMapping("fire.forget")
Mono<Void> fireForget(Payload data) {
    ...some logic
    return Mono.empty();
}

@MessageMapping("request.response")
Mono<Payload> requestResponse(Payload data) {
    return ...some mono;
}

@MessageMapping("request.stream")
Flux<Payload> requestStream(Payload data) {
    return ...some flux
}

@MessageMapping("request.channel")
Flux<Payload> requestStream(Flux<Payload> data) {
    return data.transform(...transformer);
}