feat: add pubsub framework#7028
Conversation
UpdateThe pubsub module core and documentation was split to the current PR, and the review comments in #6995 have been modified. |
UpdateThe current pubsub module switches to use the pubsub module-level independent But what I'm not sure is if in the extreme case |
There won't be concurrent issue as pubsub only modify the global pb_state when the APISIX starts. |
|
|
||
| ## How to support other messaging systems | ||
|
|
||
| An extensible pubsub module is implemented in Apache APISIX, which is responsible for starting the WebSocket server, coding and decoding communication protocols, handling client commands, and through which new messaging system support can be simply added. |
There was a problem hiding this comment.
| An extensible pubsub module is implemented in Apache APISIX, which is responsible for starting the WebSocket server, coding and decoding communication protocols, handling client commands, and through which new messaging system support can be simply added. | |
| Apache APISIX implemented an extensible pubsub module responsible for starting the WebSocket server, coding and decoding communication protocols, handling client commands, and adding support for the new messaging system. |
| - t/plugin | ||
| - t/admin t/cli t/config-center-yaml t/control t/core t/debug t/discovery t/error_page t/misc | ||
| - t/admin t/cli t/config-center-yaml t/control t/core t/debug t/discovery t/error_page t/misc t/pubsub | ||
| - t/node t/router t/script t/stream-node t/utils t/wasm t/xds-library t/xrpc |
There was a problem hiding this comment.
plugin < pubsub < node, so we should put it here
There was a problem hiding this comment.
fixed #7043 , According to #6995 (comment), I have alphabetically aligned it between node and router
| -- handle client close connection | ||
| if raw_type == "close" then | ||
| ws:send_close() | ||
| return |
There was a problem hiding this comment.
Better to use break in the while loop and handle the common logic in the same place
There was a problem hiding this comment.
This is not an error, there is no return value (in fact the return value of wait has also been removed, which does not cause confusion in the location of the error handling code), and if the client initiates a close connection, the server will also close the connection and exit the loop directly, without logging and subsequent processing. The other cases have been logged and changed to post-processing.
There was a problem hiding this comment.
We can use break here and check fatal_err? It is smelly to use both break and return in the loop,
This will cause error-prone control flow,
| if err then | ||
| -- terminate the event loop when a fatal error occurs | ||
| if ws.fatal then | ||
| ws:send_close() |
There was a problem hiding this comment.
Should we always close connection outside the loop?
|
|
||
| local resp, err = handler(value) | ||
| if not resp then | ||
| send_error_resp(ws, sequence, err) |
There was a problem hiding this comment.
Would be better to handle err in the various send_xxx operation
|
|
||
| for _, data in ipairs(data) do | ||
| local code, body = t(data.url, ngx.HTTP_PUT, data.data) | ||
| ngx.say(code..body) |
There was a problem hiding this comment.
Better not directly print the 201 status code, because when we rerun the test, code 200 will return instead.
|
Other is good for me, CI is broken. |
|
|
||
|
|
||
| local function send_error_resp(ws, sequence, err_msg) | ||
| ws:send_binary(pb.encode("PubSubResp", { |
There was a problem hiding this comment.
we need to capture the return value, all right?
pb.encode and ws:send_binary, they may fail.
There was a problem hiding this comment.
fixed, according to the practice in grpc-transcode, pb.encode does error handling via pcall, and logs are printed for errors in both cases
4 participants
Description
Support for publish-subscribe scenarios implemented in the form of websocket + protobuf, this PR contains its basic framework implementation.
Split from #6995
Checklist