chat server using openresty

[Abdul Rahman Madeni]

Hi

Any idea how to configure openresty as a chat server? Will it good enough to handle duplex messaging?

[Yichun Zhang (agentzh)]

Hello!

On Fri, May 30, 2014 at 8:02 PM, Abdul Rahman Madeni wrote:
> Any idea how to configure openresty as a chat server? Will it good enough to
> handle duplex messaging?
>

I'm going to add full duplex cosocket support to ngx_lua/openresty soon :)

Regards,
-agentzh

[develephant.net]

Hi agentzh,

First, thank you for adding the duplex socket support.  Now that it is operational, should this code work now, if implemented correctly?  Or is there a better way to achieve this?

local server = require "resty.websocket.server"
local redis  = require "resty.redis"

local function subscribe(ws)
  local sub = redis:new()
  sub:connect("127.0.0.1", 6379)
  sub:subscribe("chat.messages")
  while true do
    local bytes, err = sub:read_reply()
    if bytes then
        ws:send_text(bytes[3])
    end
  end
end

local ws, err = server:new{ timeout = 30000, max_payload_len = 65535 }

ngx.thread.spawn(subscribe, ws)

local pub = redis:new()
pub:connect("127.0.0.1", 6379)

while true do
  local bytes, typ, err = ws:recv_frame()
  if ws.fatal then return
  elseif not bytes then
      ws:send_ping()
  elseif typ == "close" then break
  elseif typ == "text"  then
      pub:publish("chat.messages", bytes)
  end
end

ws:send_close()

Thanks in advance.

-dev

[Yichun Zhang (agentzh)]

Hello!

On Sat, Aug 2, 2014 at 10:16 AM, develephant.net wrote:
> First, thank you for adding the duplex socket support. Now that it is
> operational, should this code work now, if implemented correctly? Or is
> there a better way to achieve this?
>

The basic idea is correct. But your code contains various important
bugs due to the lack of proper error handling.

1. You should handle errors for all those IO calls like send_text()
and read_reply().

2. Also, if one of your light threads run into an unrecoverable error,
you should abort the whole request (with all the light threads) with
ngx.exit(444), for example.

3. And, be very careful about your "while true" loops. Ensure these
loops will never enter dead tight loops in some code paths (for
example, your loop in the "subscribe" thread will enter infinite tight
hot loop when read_reply() returns an error. Be very careful about the
termination conditions for these loops.

4. It's just wrong and inefficient to send ping in case your
recv_frame() call is timed out. You're not handling "pong"
specifically there anyway. Also, sending "ping" in your main thread
may conflict with the send_text() call in your user thread
(full-duplex cosockets only allow a reader and a writer operating on
the same socket, not two writers).

5. Better enable the connection pool for your redis connections (via
the set_keepalive call) in case of no errors.

Regards,
-agentzh

[develephant.net]

Hi,

Thank you for your response.  I found that code from an older post, so I was just curious if it would work now.  I'll clean it up with the suggestions you mentioned, and see if I can get it running.

Thanks again.

[osman.t...@gmail.com]

Hi develephant, can you share your final results? I'm banging my head trying to make Openresty websockets work to build a chat system!

[radus...@ymail.com]

Here is my current working chat system mock: 

            local redis = require "resty.redis"

            local cjson = require "cjson"
            local red = redis:new()
            red:set_timeout(100) -- 0.1 sec
            local ok, err = red:connect("127.0.0.1", 6379)
            if not ok then
                ngx.say("failed to connect: ", err)
                return
            end

            local server = require "resty.websocket.server"

            local wb, err = server:new{
                timeout = 5000,
                max_payload_len = 65535
            }
            if not wb then
                ngx.log(ngx.ERR, "failed to new websocket: ", err)
                return ngx.exit(444)
            end

            local function subscribe (ws)

                local sub = redis:new()
                sub:connect("127.0.0.1", 6379)

                local res, err = sub:subscribe("chat:messages")
                if not res then
                    ngx.say("1: failed to subscribe: ", err)
                    return
                end

                while true do
                    local bytes, err = sub:read_reply()
                    if bytes then
                        ws:send_text(bytes[3])

                    else
                        ngx.log(ngx.ERR, "no data")
                    end
                end
            end

            ngx.thread.spawn(subscribe, wb)

            local pub = redis:new()
            pub:connect("127.0.0.1", 6379)
        
            while true do 

                local data, typ, err = wb:recv_frame()
                if wb.fatal then
                    return ngx.exit(444)
                end
                if not data then
                    local bytes, err = wb:send_ping()
                    if not bytes then
                        return ngx.exit(444)
                    end

                elseif typ == "close" then 
                    break

                elseif typ == "ping" then
                    local bytes, err = wb:send_pong()
                    if not bytes then
                        return ngx.exit(444)
                    end
                elseif typ == "pong" then
                    ngx.log(ngx.INFO, "client ponged")

                elseif typ == "text" then

                    pub:publish("chat:messages", data)
                end 
            end

            wb:send_close()

Perhaps Agentzh can review it, it's far from perfect :)