added new entrypoint for reader to websocket proxy #620
Conversation
hrsakai
added
the
type/feature
| // /ws/reader/persistent/my-property/my-cluster/my-ns/my-topic | ||
| checkArgument(parts.size() == 8, "Invalid topic name format"); | ||
| checkArgument(parts.get(1).equals("ws")); | ||
| checkArgument(parts.get(3).equals("persistent")); |
There was a problem hiding this comment.
What about non-persistent topics? I think we should support them as well in the WebSocket interface
There was a problem hiding this comment.
support non-persistent topics and delete ReaderHandler#checkRequestURI() because
AbstractWebSocketHandler class check the same.
|
retest this please |
|
retest this please |
1 similar comment
|
retest this please |
| service.getExecutor().execute(() -> receiveMessage()); | ||
| } | ||
| }).exceptionally(exception -> { | ||
| return null; |
There was a problem hiding this comment.
should we add a debug log here?
| } | ||
| updateDeliverMsgStat(msgSize); | ||
| int pending = pendingMessages.getAndDecrement(); | ||
| if (pending >= maxPendingMessages) { |
There was a problem hiding this comment.
I think this would be tricky.. here, in reader we are keep pushing messages to the client which may blow up client-memory. so, should we follow the same semantic as normal consumer where initially proxy will send queue-size messages to client and on every acknowledgement/readNext proxy send one more message.??
|
@rdhabalia |
| service.getExecutor().execute(() -> receiveMessage()); | ||
| } else { | ||
| // Resume delivery | ||
| receiveMessage(); |
There was a problem hiding this comment.
Should't we stop delivery if pending reached to maxPendingMessages else it will keep sending messages to the client ?
Should we do the similar like ConsumerHandler, after reaching maxPendingMessages, proxy-reader will send more message when it receives a new readNext request from client, same way consumer receives ack-request.? If that seems feasible solution then we can document the semantic into websocket-reader-doc as well.
There was a problem hiding this comment.
@rdhabalia
Please let me confirm.
I will try to change readerHandler like the following codes.
Does this match what you mean ?
public void writeSuccess() {
if (log.isDebugEnabled()) {
log.debug("[{}/{}] message is delivered successfully to {} ", reader.getTopic(),
subscription, getRemote().getInetSocketAddress().toString());
}
updateDeliverMsgStat(msgSize);
- pendingMessages.getAndDecrement();
}
.
.
.
int pending = pendingMessages.incrementAndGet();
if (pending < maxPendingMessages) {
// Start next read in a separate thread to avoid recursion
service.getExecutor().execute(() -> receiveMessage());
- } else {
- // Resume delivery
- receiveMessage();
}
.
.
.
@Override
public void onWebSocketText(String message) {
super.onWebSocketText(message);
+ int pending = pendingMessages.getAndDecrement();
+ if (pending >= maxPendingMessages) {
+ // Resume delivery
+ receiveMessage();
+ }
}
There was a problem hiding this comment.
yes, this would control message-delivery to the client.
There was a problem hiding this comment.
Thank you for your confirmation.
I will try to implement it.
|
@rdhabalia |
Motivation
no reader entrypoint on websocket proxy .
Modifications
added reader entrypoint to websocket proxy.
Result
we can use reader through websocket proxy