feat: support kafka consumer for pubsub scenario#6995
Conversation
| -- Its second return value is a string type error message, no need to return when | ||
| -- no error exists. | ||
| -- | ||
| -- @function core.pubsub.on |
| end | ||
|
|
||
| -- skip this loop for non-fatal errors | ||
| log.error("failed to receive websocket frame: "..err) |
There was a problem hiding this comment.
| log.error("failed to receive websocket frame: "..err) | |
| log.error("failed to receive websocket frame: ", err) |
| -- terminate the event loop when a fatal error occurs | ||
| if ws.fatal then | ||
| ws:send_close() | ||
| return "websocket server: "..err |
There was a problem hiding this comment.
Better to use break in the while loop and handle the error in the same place
| -- the pub-sub messages use binary, if the message is not | ||
| -- binary, skip this message | ||
| if raw_type ~= "binary" then | ||
| goto continue |
There was a problem hiding this comment.
Better to add warn log for unexpected input
| goto continue | ||
| end | ||
|
|
||
| local data = pb.decode("PubSubReq", raw_data) |
There was a problem hiding this comment.
Should we check decode error?
| "enable_tls": true, | ||
| "ssl_verify": true, | ||
| "enable_sasl": true, | ||
| "sasl_username": "user", |
There was a problem hiding this comment.
Please update the example when the code part is accepted.
| ngx.say(body) | ||
| } | ||
| } | ||
| --- error_code: 200 |
There was a problem hiding this comment.
We don't need to check the default error_code
| pb.option("int64_as_string") | ||
| local pubsub_protoc = protoc.new() | ||
| pubsub_protoc:addpath("apisix/include/apisix/model") | ||
| local ok, err = pcall(pubsub_protoc.loadfile, pubsub_protoc, "pubsub.proto") |
| --- config | ||
| location /t { | ||
| content_by_lua_block { | ||
| local protoc = require("protoc") |
There was a problem hiding this comment.
We can exact the common part into helper, and put it in https://github.com/apache/apisix/tree/master/t/lib?
| 1failed to fetch message, topic: not-exist, partition: 0, err: not found topic | ||
| 2offset: 0 | ||
| 3offset: 30 | ||
| 4offset: 14 msg: testmsg15 |
There was a problem hiding this comment.
Let's add a comment about where the msg is from
3 participants
Description
Support kafka consumers for publish-subscribe scenarios.
Implement proposal # (6874)
Checklist