Pub/sub

local socket = require'socket'

-- формат передачи - |тип [тема] данные|
local client = {}
client.__index = client

function client:new(ip, port)
    local o = {}                 -- прототип объекта клиента
    o.ip = ip or '127.0.0.1'     -- данные сервера
    o.port = port or '7777'      -- куда публикуем всякую дрянь
    o.callbacks = {}             -- колбеки для тем подписок
    o.socket = socket.tcp()
    print('C: connect ', o.ip, o.port, ':', o.socket:connect(o.ip, o.port), 5)
    o.socket:settimeout(0)       -- убираем блокировки (выдаст timeout коль ничего нет в сокете)
    return setmetatable(o, self) -- присобачиваем прототипу нас как класс (благодаря __index)
end

function client:close()
    self.socket:close()
    if self.callbacks.close then
        self.callbacks.close(1)
    end
end

function client:send(type, ...)
    local data = table.concat({type, ...}, ' '):gsub([[\+n]], [[\\n]])
    print('C: send: ', self.socket:send(data..'\r\n'))
end

function client:update()
    local data, err = self.socket:receive'*l' -- принимаем строчку от сервера
    if err == 'closed' then return  nil, 'closed' end

    --тип [топик] данные
    while data do
        print('C: RCV', data)
        data = data:gsub([[\\n]], [[\n]]) -- делаем обратную замену, как при отправке
        local type, body = data:match'(%S+)%s(.*)' -- нам нужно выделить тип сообщения и тело сообщения
        if type == 'pub' then -- если это какая-то публикация
            print('C: IS PUB!')
            local topic, message = body:match'(%S+)%s(.*)' -- выделяем топик и сообщение из тела
            if self.callbacks[topic] then
                self.callbacks[topic](message) -- дёргаем колбек
            end
        elseif type == 'disconnect' then
            self:close()
        end
        data = self.socket:receive'*l'
    end
end

-- При подписке, мы пихаем функцию, которая будет дёргаться когда придёт что-то по данному топику
function client:subscribe(topic, callback)
    self:send('sub', topic)
    self.callbacks[topic] = callback
end

function client:unsubscribe(topic)
    self:send('unsub', topic)
    self.callbacks[topic] = nil -- удаляем колбек на этот топик, коль отписались
end

function client:publish(topic, ...)
    self:send('pub', topic, ...)
end

local server = {}
server.__index = server

function server:new(port, ip)  -- ip тут совсем не обязателен, поэтому второй (как опция фильтра айпишников)
    local o = {}
    o.ip = ip or '*'             -- при создании слушающего сокета, ip работает как фильтр айпишников, по которым приходят сообщения
    o.port = port or '7777'      -- а порт - как порт который мы слушаем
    o.clients = {}               -- список приконнекченного народа и их подписок как элементы таблицы
    o.socket = socket.tcp()
    o.socket:settimeout(0)       -- убираем блокировки (выдаст timeout коль ничего нет в сокете)
    o.socket:setsockname(o.ip, o.port)

    print('S: listen ', o.ip, o.port, ':', o.socket:listen())
    return setmetatable(o, self) -- присобачиваем прототипу нас как класс (благодаря __index)
end

function server:send(sock, ...)
    local data = table.concat({...}, ' ') -- конкатенируем всю фигню в строку и балуемся с экранированием переносов строки.
    print('S: send: ', data)
    sock:send(data..'\r\n')
end

-- Пересылка сообщения по выбранному топику
function server:repost(topic, data)
    for sock, pubs in pairs(self.clients) do
        if pubs[topic] then     -- Если клиент подписан на наш топик - отправляем ему
            self:send(sock, data) -- Тут - полные данные, без парсинга. Что приняли то и отослали.
        end
    end
end

function server:update()
    local sock = self.socket:accept() -- к нам кто-то законнектился? Отлично!
    while sock do
        print('S: New sock!')
        sock:settimeout(0)
        self.clients[sock] = {} -- инициализируем список подписок
        sock = self.socket:accept()
    end

    -- Проходим по списку клиентов, принимаем от них данные, реагируем
    for sock, pubs in pairs(self.clients) do
        local data, err = sock:receive'*l'
        if err == 'closed' then  -- если клиент отвалился - удаляем его нафиг, он нам больше не нужен
            print('S: '..(sock:getpeername())..' CLOSED')
            self.clients[sock] = nil
            break
        end

        while data do
            local type, body = data:match'(%S+)%s(%S+)'
            print('S: type/body', type, body)
            if type == 'sub' then       -- подписка
                print('S: IS SUB!')
                pubs[body] = true         -- body как топик (тип подписки)
            elseif type == 'unsub' then -- отмена подписки
                print('S: IS UNSUB!')
                pubs[body] = nil
            elseif type == 'pub' then   -- публикация
                self:repost(body, data)   -- мы так и не изменяли data, поэтому перешлём в неизменном виде
            elseif type == 'disconnect' then
                print('S: IS DUSCONNECT!')
                self.clients[sock] = nil  -- удаляем нафиг, вместе с подписками
            end
            data = sock:receive'*l'
        end
    end
end

return {
    client = client,
    server = server
}