local fiber = require 'fiber'
local log = require 'log'
local mqtt = require 'mqtt'
local ioloop  = require 'mqtt.ioloop'
local connector = require 'app.tnt_mqtt_connector'
local M = {}
local client
ioloop = ioloop.get(true, { sleep_function = fiber.sleep })
function M.subscribe(topic, qos)
    client:subscribe {
        topic = topic,
        qos = qos or 0,
        no_local = true
    }
end
function M.unsubscribe(topic)
    client:unsubscribe { topic = topic }
end
function M.publish(topic, payload, qos)
    client:publish({
        topic: topic,
        payload: payload,
        qos: qos or 0
    })
end
function M.start(opts)
    client = mqtt.client {
        id = opts.client_id,
        uri = opts.host .. ':' .. tostring(opts.port),
        clean = opts.clean or true,
        connector = connector,
        version = mqtt.v50,
        keep_alive = opts.keep_alive or 60,
        reconnect = opts.reconnect or true,
        username = opts.user,
        password = opts.pass
    }
    client:on {
        connect = function(connack)
            if connack.rc ~= 0 then
                return log.error(
                    'mqtt error: error = %s, ack = %s',
                    connack:reason_string(),
                    connack
                )
            end
            log.info('mqtt: connected')
        end,
        message = handle_msg,
        error = handle_error
    }
    ioloop:add(client)
    fiber.create(function()
        while true do
            ioloop:iteration()
            fiber.yield()
        end
    end)
end
return M

local fiber = require 'fiber'
local clock = require 'clock'
local log = require 'log'
local LIMIT = 32000
local lock = fiber.cond()
local counter = 0
local t1 = clock.realtime()
for i = 1, LIMIT do
   fiber.create(function()
       lock:wait()
       counter = counter + 1
   end)
end
local t2 = clock.realtime()
log.info('fibers created, took %s seconds', t2 - t1)
t1 = clock.realtime()
fiber.sleep(10)
t2 = clock.realtime()
log.info('real sleep time: %s seconds', t2 - t1)
t1 = clock.realtime()
lock:broadcast()
fiber.yield()
t2 = clock.realtime()
log.info('fibers unlocked, counter is %s; took %s seconds', counter, t2 - t1)

fibers created, took 0.32585644721985 seconds
real sleep time: 9.6747360229492 seconds
fibers unlocked, counter is 32000; took 0.026816844940186 seconds

SystemError fiber mprotect failed: Cannot allocate memory
SystemError fiber mprotect failed: Cannot allocate memory
fatal error, exiting the event loop

local states = {
   OFFLINE = 0,
   ONLINE = 1,
   OPENING = 2,
   READY = 3,
   WAITING = 4
}
-- create space for user devices
local devices = box.schema.space.create('devices', {
    engine = 'memtx',
    if_not_exists = true
})
devices:format {
   {name = 'id', type = 'string'},
   {name = 'state', type = 'number'},
   {name = 'ts', type = 'number', is_nullable = true}
}
-- create index for timeouts and states
devices:create_index('pulse', {
    parts = {'state', 'ts'},
    type = 'tree',
    unique = false,
    if_not_exists = true
})
local function auto_yield()
   if not ticks then
       ticks = 512
   end
   local count = 0
   return function(...)
       count = count + 1
       if count % ticks == 0 then
           fiber.yield()
       end
       return ...
   end
end
local function get_expired(state, timeout)
   local now = math.floor(clock.realtime())
   return box.space.devices.index.pulse
       :pairs({states[state], now - timeout},
              {iterator = box.index.LT})
       :map(auto_yield())
       :take_while(function(device)
           return device.state == states[state]
        end)
       :filter(function(device)
           return now - device.ts > timeout
        end)
end