Pub/Sub with websockets about lua-resty-redis HOT 8 CLOSED

@samsonradu This use case will be trivial when the cosocket API supports full duplexing, which I'm going to implement soon. Basically, just create 2 "light theads" via ngx.thread.spawn(), one reads data from redis and write to the downstream websocket and the other reads data from the downstream websocket and pushes to redis. Such two-way communications are the most natural design pattern here :)

from lua-resty-redis.

Pretty new to the concepts here, can you show me how to achieve the one reads data from redis and write to the downstream websocket. Does the following still block in reading from the redis socket?

 local yield = coroutine.yield
 local res, err = red:subscribe("streampic")
 local buff = {}
 function reader ()
     local self = coroutine.running() 
     --read the pubsub and queue it 
     res, err = red:read_reply()
     if res then
         List.pushright(buff, res)
         yield(self)
     end 
 end 
 
 while true do
     local data, typ, err = wb:recv_frame()
     -- Websocket related code block (ping/close etc.)
     ngx.thread.spawn(reader)
     local item = List.popleft(buff)
     wb:send_binary(item)
 end
 wb:send_close()

from lua-resty-redis.

@samsonradu You should not call ngx.thread.spawn in your loop (you end up creating a lot of "light threads"). You only need 2 "light threads" here, one for each of the 2 transmission directions (from downstream to upstream and from upstream to downstream). Because you can reuse the current main light thread, you only need to create another one (child) "light thread" via ngx.thread.spawn. See the documentation for more details: https://github.com/openresty/lua-nginx-module#ngxthreadspawn

from lua-resty-redis.

@samsonradu Also, you do not need to use the coroutine.* API here at all, which does no good. You really need "light threads" and only "light threads" here instead.

from lua-resty-redis.

I only need a one-way data stream from a Redis PubSub to the connected websockets. A Node JS implementation would look like this:

 pubsubClient.on('message', function (channel, message){
   for (var id in subscribers){
       subscribers[id].write(message);
   }   
 });
 pubsubClient.subscribe("streampic");
 
 var subscribers = {};
 var wss = sockjs.createServer();
 wss.on('connection', function(socket) {
     socket.on('data', function(message) {
         subscribers[socket.id] = socket;
     }); 
     socket.on('close', function() {
         delete subscribers[socket.id];
     });
 });

While trying to achieve this in Lua seems so "stateless" :) Can I for example have a list of connected websockets? I also do not understand how to loop over the Redis PubSub and know when new data comes in:

local last = 0
function subscriber()
    res, err = red:read_reply()
    if res then
        ngx.log(ngx.ERR, "receiving ", res[3])
        last = res[3] 
    end
end 
-- not sure whether this will keep running forever or exits immediately
ngx.thread.spawn(subscriber)
while true do 
    -- websocket code
     wb:send_text(last)
end
wb:send_close()

The above gives me a timeout every 5 seconds.

 2014/06/16 00:54:05 [error] 3665#0: *1 lua tcp socket read timed out

from lua-resty-redis.

@samsonradu No, you cannot manipulate request A's cosocket objects (or "thread" objects) directly from within the context of request B.

But you can pass data via the ngx.shared.DICT API (see https://github.com/openresty/lua-nginx-module#ngxshareddict) or your own Lua module level data (see https://github.com/openresty/lua-nginx-module#data-sharing-within-an-nginx-worker ). You can poll the shared data periodically with ngx.sleep() in your request's Lua handler and having a background recurring timer handler querying the redis for all your concurrent downstream requests. See the lua-resty-upstream-healthcheck library for such an example.

Another option is to establish a Redis connection upon every nginx request.

from lua-resty-redis.

@samsonradu The simplest way to do this is something like below (untested, be careful about typos):

local ok, err = red:subscribe("streampic")
if not ok then
    wb:send_close()
    return
end
while true do
    local res, err = red:read_reply()
    if not res then
        -- error handling
        break
    end
    local item = res[3]
    local ok, err = wb:send_binary(item)
    if not ok then
        -- better error handling
        break
    end
end
local ok, err = wb:send_close()
if not ok then
    -- error handling
end

Please note that always do proper error handling yourself, especially in your infinite Lua loop, otherwise you can easily enter hot dead looping when error happens.

Also, do not read from the websocket when you're not expecting (immediate) messages from the downstream in the main processing loop. If you really need to do the other direction, as I've said, better create a "light thread" via ngx.thread.spawn for it. But do not mix up these two data flow directions in a single "light thread".

from lua-resty-redis.

Thanks, things are getting clearer as I go. Closing the issue now :)

from lua-resty-redis.