Beta!

Author: dedok

.gitignore

@@ -5,7 +5,7 @@
 *.so.*
 *.dylib
 
-tests/out
+test/out
 build
 CMakeCache.txt
 CMakeFiles

examples/connect.lua

@@ -0,0 +1,18 @@
+#!/usr/bin/env tarantool
+
+-- Load Tarantool mqtt
+local mqtt = require('mqtt')
+
+-- Create instance
+connection = mqtt.new()
+
+-- Connect to the server
+connection:connect({host='127.0.0.1', port=1883})
+
+-- Set callback for recv new messages
+connection:on_message(function (message_id, topic, payload, gos, retain)
+  print('New message', message_id, topic, payload, gos, retain)
+end)
+
+-- Subscribe to a system topic
+connection:subscribe('$SYS/#')

examples/producer_consumer_queue.lua

@@ -0,0 +1,51 @@
+#!/usr/bin/env tarantool
+
+local mqtt = require('mqtt')
+local fiber = require('fiber')
+local yaml = require('yaml')
+
+box.cfg {wal_mode = 'none'}
+
+-- Queue out
+queue = box.schema.space.create('queue')
+queue:create_index('id')
+
+-- Create instance
+connection = mqtt.new()
+
+-- Connect to the server
+local ok, emsg = connection:connect({host='127.0.0.1', port=1883})
+if not ok then
+  error('connect ->', emsg)
+end
+
+-- Subscribe to a system topic
+local ok, emsg = connection:subscribe('tarantool/box/info')
+if not ok then
+  error('subscribe -> ',emsg)
+end
+
+--
+-- Work
+--
+
+-- Consumer
+connection:on_message(function (message_id, topic, payload, gos, retain)
+  queue:auto_increment{yaml.decode(payload)}
+  if queue:len() > 10 then
+    print('Queue [[')
+    for _, v in pairs(queue:select{}) do
+      print(_, yaml.encode(v[2]))
+      queue:delete(v[1])
+    end
+    print(']]')
+  end
+end)
+
+-- Producer
+fiber.create(function()
+  while true do
+    connection:publish('tarantool/box/info', yaml.encode{box.info()})
+    fiber.sleep(0.5)
+  end
+end)

mqtt/init.lua

@@ -59,11 +59,14 @@ local new = function(id, clean_session)
   return setmetatable({
     VERSION = mqtt:version(), -- Str fmt: X.X.X
     RECONNECT_INTERVAL = 0.5,
+    POLL_INTERVAL = 0.0,
 
     mqtt = mqtt,
     connected = false,
     auto_reconect = true,
     fiber = nil,
+
+    subscribers = {},
   }, mqtt_mt)
 end
 
@@ -87,7 +90,17 @@ mqtt_mt = {
     end,
     __try_reconnect = function(self)
       while not self.connected do
+
+        -- Reconnect
         self.connected, _ = self.mqtt:reconnect()
+
+        -- Restoring subscribing
+        if self.connected then
+          for topic, opts in pairs(self.subscribers) do
+            self.connected, _ = self.mqtt:subscribe(topic, opts.qos)
+          end
+        end
+
         fiber.sleep(self.RECONNECT_INTERVAL)
       end
     end,
@@ -102,10 +115,11 @@ mqtt_mt = {
           if self.auto_reconect then
             self:__try_reconnect()
           else
-            log.error("mqtt[STOPED]: lost connection, error: '%s'", emsg)
+            log.error(
+              "mqtt: the client is not currently connected, error %s", emsg)
           end
         end
-        fiber.sleep(0.3)
+        fiber.sleep(self.POLL_INTERVAL)
       end
     end,
 
@@ -164,27 +178,61 @@ mqtt_mt = {
 
       -- Start work
       self.fiber = fiber.create(
-          function()
-            self:__try_connect(opts)
-            self:__poll_forever()
-          end
-      )
+        function()
+          self:__try_connect(opts)
+          self:__poll_forever()
+        end)
 
       return true
     end,
 
+    --
+    -- Reconnect to a broker.
+    --
+    --  This function provides an easy way of reconnecting to a broker after a
+    --  connection has been lost. It uses the values that were provided in the
+    --  <connect> call. It must not be called before
+    --  <connect>.
+    --
+    --  NOTE
+    --    After reconnecting you must call<subscribe> for subscribing
+    --    to a topics.
+    --
+    reconnect = function()
+      return self.mqtt:reconnect()
+    end,
+
     --
     -- Subscribe to a topic.
     --
     -- Parameters:
     --  sub -  the subscription pattern.
     --  qos -  the requested Quality of Service for this subscription.
     --
-    --Returns:
-    -- true or false, integer mid
+    -- Returns:
+    --  true or false, integer mid or error message
+    --
+    subscribe = function(self, topic, qos)
+      local ok, mid_or_emsg = self.mqtt:subscribe(topic, qos)
+      if ok then
+        self.subscribers[topic] = {qos = qos}
+      end
+      return ok, mid_or_emsg
+    end,
+
+    --
+    -- Unsubscribe from a topic.
+    --
+    -- Parameters:
+    --  topic - the unsubscription pattern.
+    --
+    -- Returns:
+    --  true or false, integer mid or error message
     --
-    subscribe = function(self, channel, qos)
-      return self.mqtt:subscribe(channel, qos)
+    unsubscribe = function(self, topic)
+      local ok, emsg = self.mqtt:unsubscribe(topic)
+      self.subscribers[topic] = nil
+      return ok, emsg
     end,
 
     --

tests/pub_sub.lua test/pub_sub.luatests/pub_sub.lua renamed to test/pub_sub.lua

@@ -4,53 +4,73 @@ local mqtt = require('mqtt')
 local fiber = require('fiber')
 local yaml = require('yaml')
 local os = require('os')
+local log = require('log')
 
 box.cfg{wal_mode = 'none'}
 fan_in = box.schema.space.create('fan_in', {if_not_exists = true})
 fan_in:create_index('id')
 
 conn = mqtt.new()
 
+function die(msg)
+  error('[-] ' .. msg)
+  os.exit(1)
+end
+
 function W(ok, msg)
   if not ok then
-    error('[-] ' .. msg)
+    die(msg)
+  end
+  return ok, msg
+end
+
+function W2(ok, msg)
+  if not ok then
+    log.error(msg)
   end
   return ok, msg
 end
 
+local TIME = '' .. fiber.time()
+
 W(conn:on_message(
   function(mid, topic, payload, gos, retain)
+    W2(conn:unsubscribe('channel2/messages'))
     fan_in:auto_increment{topic, payload, gos, retain}
-    print("[+] Message - OK")
+    W2(conn:publish('channel/messages', TIME))
 end))
 
 W(conn:connect({host="127.0.0.1", port=1883}))
 W(conn:subscribe('channel/#'))
 W(conn:subscribe('channel2/#'))
 
 print('[+] Starting tnt loop ...')
+
+W(conn:publish('channel2/messages', TIME, 0, false))
+
 fiber.create(function()
-  local time = '' .. fiber.time()
   while true do
-    W(conn:publish('channel/messages', time, 0, false))
-    W(conn:publish('channel2/messages', time, 0, false))
-    if fan_in:len() == 10 then
+
+    if fan_in:len() > 10 then
       for k, v in pairs(fan_in:select{}) do
+
         if v[2] ~= 'channel/messages' and v[2] ~= 'channel2/messages' then
-          error('expected channel{2}/messages, got ' .. v[2])
+          die('expected channel{2}/messages, got ' .. v[2])
         end
-        if v[3] ~= time then
-          error('expected ' .. time .. ', got ' .. v[3])
+        if v[3] ~= TIME then
+          die('expected ' .. time .. ', got ' .. v[3])
         end
         if v[4] ~= 0 then
-          error('expected 0, got ' .. v[4])
+          die('expected 0, got ' .. v[4])
         end
         if v[5] ~= false then
-          error('expected false, got ' .. v[5])
+          die('expected false, got ' .. v[5])
         end
-        print('[+] Test -- OK')
-        os.exit(0)
+        --os.exit(0)
+        print('[+] Test -- OK', yaml.encode(v))
+        fan_in:delete{v[1]}
       end
+      os.exit(0)
     end
     fiber.sleep(0.0)
   end