RSocket Js 客户端未获取从 Spring 引导 RSocket 服务器生成的元素
RSocket Js client not getting elements generated from a Spring Boot RSocket Server
Github 上可用的示例项目:https://github.com/codependent/rsocket-rating-service
A Spring 启动 RSocket 服务器消息映射需要一个 requestResponse 请求,返回一个简单的 POJO:
@MessageMapping("request-rating")
fun getRatingWebSocket(ratingRequest: RatingRequest): Mono<Rating> {
return Mono.just(Rating(ratingRequest.songId, (0..10).random())).log()
.doOnNext {
logger.info("Next {}", it)
}
.doOnCancel {
logger.info("Cancel")
}
.doOnSuccess {
logger.info("Success {}", it)
}
.doOnError { throwable ->
logger.error("Error {}", throwable)
}
.doOnTerminate { logger.info("Terminate") }
}
在客户端,我有以下 JS 代码连接到 RSocket 服务器并请求一个值:
const {
RSocketClient,
JsonSerializer,
IdentitySerializer,
} = require('rsocket-core');
const RSocketWebSocketClient = require('rsocket-websocket-client').default;
const route = 'request-rating';
let client = undefined;
let rSocket = undefined;
function main() {
if (client !== undefined) {
client.close();
}
client = new RSocketClient({
serializers: {
data: JsonSerializer,
metadata: IdentitySerializer
},
setup: {
// 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',
},
transport: new RSocketWebSocketClient({
url: 'ws://localhost:8080/rating-ws'
}),
});
// Open the connection
client.connect().subscribe({
onComplete: socket => {
// socket provides the rsocket interactions fire/forget, request/response,
// request/stream, etc as well as methods to close the socket.
rSocket = socket;
},
onError: error => {
console.log("Connection has been refused due to ", error);
},
onSubscribe: cancel => {
/* call cancel() to abort */
}
});
document.getElementById('sendButton').addEventListener('click', requestRating);
}
function requestRating() {
rSocket.requestResponse({
data: {
'songId': document.getElementById("songId").value
},
metadata: String.fromCharCode(route.length) + route
}).subscribe({
onComplete: () => {
console.log('Complete')
},
onError: error => {
console.log("Connection has been closed due to " + error);
},
onNext: payload => {
console.log(payload.data);
},
onSubscribe: subscription => {
//subscription.request(1)
console.log("Subscribed")
}
});
}
document.addEventListener('DOMContentLoaded', main);
index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Rating Service</title>
<link href="/webjars/bootstrap/css/bootstrap.min.css" rel="stylesheet">
<script src="/webjars/jquery/jquery.min.js"></script>
<script src="/webjars/stomp-websocket/stomp.min.js"></script>
<script src="bundle.js"></script>
</head>
<body>
<div>Request your rating</div>
<div>
<label> ClientId:
<input id="songId" type="text" name="songId"/>
</label>
</div>
<div><input id="sendButton" type="button" name="send" value="Send"/></div>
</body>
</html>
访问 http://localhost:8080/index.html 后输入一些文本并推送。
请求到达记录正确生成 onNext 值的服务器:
[ctor-http-nio-6] reactor.Mono.Just.1 : | onSubscribe([Synchronous Fuseable] Operators.ScalarSubscription)
[ctor-http-nio-6] reactor.Mono.Just.1 : | request(1)
[ctor-http-nio-6] reactor.Mono.Just.1 : | onNext(Rating(songId=asdfas, value=0))
[ctor-http-nio-6] c.c.r.r.c.RatingServiceRestController : Next Rating(songId=asdfas, value=0)
[ctor-http-nio-6] c.c.r.r.c.RatingServiceRestController : Success Rating(songId=asdfas, value=0)
[ctor-http-nio-6] c.c.r.r.c.RatingServiceRestController : Terminate
[ctor-http-nio-6] reactor.Mono.Just.1 : | onComplete()
但是在客户端
onNext: payload => {
console.log(payload.data);
},
从未被调用,浏览器日志仅显示:
Subscribed
Complete
为什么它没有从服务器获取生成的值?
在RSocket JS中,不同于Reactive Streams,onComplete信号是通过requestResponse交互方式获取数据的。
这似乎不符合协议规范(https://rsocket.io/about/protocol):
Both (C)omplete and (N)ext set meaning PAYLOAD contains data and signals stream completion.
For example: An Observable stream receiving onNext(payload) followed by onComplete().
Just (C)omplete set meaning PAYLOAD contains no data and only signals stream completion.
For example: An Observable stream receiving onComplete().
Just (N)ext set meaning PAYLOAD contains data stream is NOT completed.
For example: An Observable stream receiving onNext(payload).
为了解决这个问题,我不得不将回调函数修改为:
onComplete: completeData => {
console.log('Complete '+ completeData.data)
},
Github 上可用的示例项目:https://github.com/codependent/rsocket-rating-service
A Spring 启动 RSocket 服务器消息映射需要一个 requestResponse 请求,返回一个简单的 POJO:
@MessageMapping("request-rating")
fun getRatingWebSocket(ratingRequest: RatingRequest): Mono<Rating> {
return Mono.just(Rating(ratingRequest.songId, (0..10).random())).log()
.doOnNext {
logger.info("Next {}", it)
}
.doOnCancel {
logger.info("Cancel")
}
.doOnSuccess {
logger.info("Success {}", it)
}
.doOnError { throwable ->
logger.error("Error {}", throwable)
}
.doOnTerminate { logger.info("Terminate") }
}
在客户端,我有以下 JS 代码连接到 RSocket 服务器并请求一个值:
const {
RSocketClient,
JsonSerializer,
IdentitySerializer,
} = require('rsocket-core');
const RSocketWebSocketClient = require('rsocket-websocket-client').default;
const route = 'request-rating';
let client = undefined;
let rSocket = undefined;
function main() {
if (client !== undefined) {
client.close();
}
client = new RSocketClient({
serializers: {
data: JsonSerializer,
metadata: IdentitySerializer
},
setup: {
// 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',
},
transport: new RSocketWebSocketClient({
url: 'ws://localhost:8080/rating-ws'
}),
});
// Open the connection
client.connect().subscribe({
onComplete: socket => {
// socket provides the rsocket interactions fire/forget, request/response,
// request/stream, etc as well as methods to close the socket.
rSocket = socket;
},
onError: error => {
console.log("Connection has been refused due to ", error);
},
onSubscribe: cancel => {
/* call cancel() to abort */
}
});
document.getElementById('sendButton').addEventListener('click', requestRating);
}
function requestRating() {
rSocket.requestResponse({
data: {
'songId': document.getElementById("songId").value
},
metadata: String.fromCharCode(route.length) + route
}).subscribe({
onComplete: () => {
console.log('Complete')
},
onError: error => {
console.log("Connection has been closed due to " + error);
},
onNext: payload => {
console.log(payload.data);
},
onSubscribe: subscription => {
//subscription.request(1)
console.log("Subscribed")
}
});
}
document.addEventListener('DOMContentLoaded', main);
index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Rating Service</title>
<link href="/webjars/bootstrap/css/bootstrap.min.css" rel="stylesheet">
<script src="/webjars/jquery/jquery.min.js"></script>
<script src="/webjars/stomp-websocket/stomp.min.js"></script>
<script src="bundle.js"></script>
</head>
<body>
<div>Request your rating</div>
<div>
<label> ClientId:
<input id="songId" type="text" name="songId"/>
</label>
</div>
<div><input id="sendButton" type="button" name="send" value="Send"/></div>
</body>
</html>
访问 http://localhost:8080/index.html 后输入一些文本并推送。
请求到达记录正确生成 onNext 值的服务器:
[ctor-http-nio-6] reactor.Mono.Just.1 : | onSubscribe([Synchronous Fuseable] Operators.ScalarSubscription)
[ctor-http-nio-6] reactor.Mono.Just.1 : | request(1)
[ctor-http-nio-6] reactor.Mono.Just.1 : | onNext(Rating(songId=asdfas, value=0))
[ctor-http-nio-6] c.c.r.r.c.RatingServiceRestController : Next Rating(songId=asdfas, value=0)
[ctor-http-nio-6] c.c.r.r.c.RatingServiceRestController : Success Rating(songId=asdfas, value=0)
[ctor-http-nio-6] c.c.r.r.c.RatingServiceRestController : Terminate
[ctor-http-nio-6] reactor.Mono.Just.1 : | onComplete()
但是在客户端
onNext: payload => {
console.log(payload.data);
},
从未被调用,浏览器日志仅显示:
Subscribed
Complete
为什么它没有从服务器获取生成的值?
在RSocket JS中,不同于Reactive Streams,onComplete信号是通过requestResponse交互方式获取数据的。
这似乎不符合协议规范(https://rsocket.io/about/protocol):
Both (C)omplete and (N)ext set meaning PAYLOAD contains data and signals stream completion. For example: An Observable stream receiving onNext(payload) followed by onComplete(). Just (C)omplete set meaning PAYLOAD contains no data and only signals stream completion. For example: An Observable stream receiving onComplete(). Just (N)ext set meaning PAYLOAD contains data stream is NOT completed. For example: An Observable stream receiving onNext(payload).
为了解决这个问题,我不得不将回调函数修改为:
onComplete: completeData => {
console.log('Complete '+ completeData.data)
},