-
Notifications
You must be signed in to change notification settings - Fork 2.5k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
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. |
docs/en/latest/pubsub.md
Outdated
|
||
## 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.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
fixed
@@ -31,7 +31,7 @@ jobs: | |||
- linux_openresty_1_17 | |||
test_dir: | |||
- 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.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
plugin < pubsub < node, so we should put it here
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
fixed #7043 , According to #6995 (comment), I have alphabetically aligned it between node
and router
-- no error exists. | ||
-- | ||
-- @function core.pubsub.on | ||
-- @tparam string command to add callback |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The syntax is @tparam type name desc
, see
Line 110 in 6ddca10
-- @tparam string|table data The data to be encoded. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
fixed
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
continue adjust #7043
apisix/core/pubsub.lua
Outdated
-- handle client close connection | ||
if raw_type == "close" then | ||
ws:send_close() | ||
return |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Better to use break in the while loop and handle the common logic in the same place
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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,
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
fixed
apisix/core/pubsub.lua
Outdated
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.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Should we always close connection outside the loop?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
fixed
apisix/core/pubsub.lua
Outdated
|
||
local resp, err = handler(value) | ||
if not resp then | ||
send_error_resp(ws, sequence, err) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Would be better to handle err in the various send_xxx
operation
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
fixed #7043
t/pubsub/pubsub.t
Outdated
|
||
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.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Better not directly print the 201 status code, because when we rerun the test, code 200 will return instead.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
fixed
Other is good for me, CI is broken. |
apisix/core/pubsub.lua
Outdated
|
||
|
||
local function send_error_resp(ws, sequence, err_msg) | ||
ws:send_binary(pb.encode("PubSubResp", { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
fixed, according to the practice in grpc-transcode
, pb.encode
does error handling via pcall
, and logs are printed for errors in both cases
Description
Support for publish-subscribe scenarios implemented in the form of websocket + protobuf, this PR contains its basic framework implementation.
Split from #6995
Checklist