Implement noun listening

- Replace queue with list

bc.lua

@@ -6,33 +6,43 @@ modem = require("modem")
 
 serializer = require("serialization")
 
-Queue = {first = 0, last = -1}
+List = {last = -1}
 
-function Queue:new()
+function List:new()
   local o = {}
   setmetatable(o, self)
   self.__index = self
   return o
 end
 
-function Queue:insert(value)
+function List:insert(value)
   local last = self.last + 1
   self.last = last
   self[last] = value
+  return last
 end
 
-function Queue:isempty()
+function List:iter()
-  return self.first > self.last
+  local i = -1
+  local last = self.last
+  local list = self
+  return function()
+    while true do
+      i = i + 1
+      if i <= last then
+        if list[i] ~= nil then
+          return list[i]
+        end
+      else
+        return nil
+      end
+    end
+  end
 end
 
-function Queue:remove()
+function List:remove(id)
-  local first = self.first
+  local value = self[id]
-  if self:isempty() then
+  self[id] = nil
-    error("Queue is empty")
-  end
-  local value = self[first]
-  self[first] = nil
-  self.first = first + 1
   return value
 end
 
@@ -43,6 +53,9 @@ message_type = {
   hello = 4,
   goodbye = 5,
   hello_response = 6,
+  request_listening = 7,
+  request_stop_listening = 8,
+  listener_update = 9,
 }
 
 CFG_PORT = 1234
@@ -53,9 +66,11 @@ function bc:init(my_address, local_nouns, local_verbs)
   local o = {}
   setmetatable(o, self)
   self.__index = self
-  o.queue = Queue:new()
   o.remote_verbs = {}
   o.remote_nouns = {}
+  o.local_listeners = {}
+  o.remote_listeners = {}
+  o.listening_remotes = {}
   -- Modem listener
   function modem_listener(localAddress, remoteAddress, port, dist, message)
     dbg = message
@@ -96,14 +111,20 @@ function bc:init(my_address, local_nouns, local_verbs)
         for i,noun in ipairs(message.nouns) do
           o.remote_nouns[noun] = remoteAddress
         end
+      elseif message.ty == message_type.request_listening then
+        o.listening_remotes[message.noun] = o.listening_remotes[message.noun] or {}
+        o.listening_remotes[message.noun][remoteAddress] = o.listening_remotes[message.noun][remoteAddress] or {}
+        o.listening_remotes[message.noun][remoteAddress][message.id] = {query=message.query, qparam=message.qparam}
+      elseif message.ty == message_type.listener_update then
+        o.remote_listeners[message.noun][message.id].callback(o, message.value)
       end
     end
   end
   -- Change this for deployment VVV
   o.modem = modem:init(my_address, modem_listener)
   -- Init local stuff
-  o.local_verbs = local_verbs
+  o.local_verbs = local_verbs or {}
-  o.local_nouns = local_nouns
+  o.local_nouns = local_nouns or {}
   -- Send own nouns and verbs and request all remotes to send theirs
   local mynouns = {}
   for noun in pairs(local_nouns) do
@@ -135,6 +156,41 @@ end
 
 function bc:set_noun(n, value)
   if self.local_nouns[n] ~= nil then
+    -- Call local listeners
+    if self.local_listeners[n] ~= nil then
+      for listener in self.local_listeners[n]:iter() do
+        if    (listener.query == "onchange" and self.local_nouns[n] ~= value)
+           or (listener.query == "onrising" and self.local_nouns[n] < value)
+           or (listener.query == "onfalling" and self.local_nouns[n] > value)
+           or (listener.query == "onvalue" and value == listener.qparam)
+           or (listener.query == "onabove" and value > listener.qparam)
+           or (listener.query == "onbelow" and value < listener.qparam)
+           then
+          self.local_nouns[n] = value -- Apply here, too, because else there could be glitches
+          listener.callback(self, value)
+        end
+      end
+    end
+    -- Call remote listeners
+    if self.listening_remotes[n] ~= nil then
+      for address,remote in pairs(self.listening_remotes[n]) do
+        for id, listener in pairs(remote) do
+          if   (listener.query == "onchange" and self.local_nouns[n] ~= value)
+            or (listener.query == "onrising" and self.local_nouns[n] < value)
+            or (listener.query == "onfalling" and self.local_nouns[n] > value)
+            or (listener.query == "onvalue" and value == listener.qparam)
+            or (listener.query == "onabove" and value > listener.qparam)
+            or (listener.query == "onbelow" and value < listener.qparam)
+             then
+            self.local_nouns[n] = value -- Apply here, too, because else there could be glitches
+            -- Send update
+            self.modem:send(address, CFG_PORT, serializer.serialize({ty=message_type.listener_update, noun=n, value=value, id=id}))
+          end
+        end
+      end
+
+    end
+    -- Apply change
     self.local_nouns[n] = value
   else
     error("Can't set remote nouns")
@@ -151,4 +207,35 @@ function bc:call_verb(v, param)
   end
 end
 
+function bc:listen_noun(n, query, qparam, callback)
+  if self.local_nouns[n] ~= nil then -- Local listening
+    self.local_listeners[n] = self.local_listeners[n] or List:new()
+    local id = self.local_listeners[n]:insert({query=query, qparam=qparam, callback=callback})
+    return id
+  elseif self.remote_nouns[n] ~= nil then -- Remote listening
+    self.remote_listeners[n] = self.remote_listeners[n] or List:new()
+    local id = self.remote_listeners[n]:insert({query=query, qparam=qparam, callback=callback})
+    -- Request remote listening
+    self.modem:send(self.remote_nouns[n], CFG_PORT, serializer.serialize({ty=message_type.request_listening, noun=n, query=query, qparam=qparam, id=id}))
+    return id
+  else
+    error("Noun not found, Node might be offline")
+  end
+end
+
+function bc:listen_cancel(n, id)
+  if self.local_nouns[n] ~= nil then -- Local listening
+    if self.local_listeners[n] ~= nil then
+      self.local_listeners[n]:remove(id)
+    end
+  elseif self.remote_nouns[n] ~= nil then -- Remote listening
+    if self.remote_listeners[n] ~= nil then
+      self.remote_listeners[n]:remove(id)
+      self.modem:send(self.remote_nouns[n], CFG_PORT, serializer.serialize({ty=message_type.request_stop_listening, noun=n, id=id}))
+    end
+  else
+    error("Noun not found, Node might be offline")
+  end
+end
+
 return bc

test_bc.lua

@@ -66,3 +66,87 @@ function test_multinode_get_noun()
   assert_equal(key, bc1:get_noun("light"), "Local get failed")
   assert_equal(key, bc2:get_noun("light"), "Remote get failed")
 end
+
+function test_multinode_listening()
+  bc1 = bc:init("a1", {["foo"]=123}, {})
+  bc2 = bc:init("a2", {}, {})
+  local i = false
+  -- Local listening
+  local id = bc1:listen_noun("foo", "onchange", nil, function(bc, foo)
+    i = true
+  end)
+  bc1:set_noun("foo", 111)
+  assert_true(i, "Local listening failed")
+  i = false
+  bc1:listen_cancel("foo", id) -- Test wether cancelling works
+  local j = false
+  bc2:listen_noun("foo", "onchange", nil, function(bc, foo)
+    j = true
+  end)
+  bc1:set_noun("foo", 1234)
+  assert_true(j, "Remote listening failed")
+  assert_false(i, "Cancelling listener failed")
+end
+
+function test_listen_modes()
+  bci = bc:init("a1", {["noun"]=10}, {})
+  local i = false
+  local id = 0
+  -- onchange
+  id = bci:listen_noun("noun", "onchange", nil, function(bc, noun)
+    i = true
+  end)
+  bci:set_noun("noun", 11)
+  assert_true(i, "Listening \"onchange\" failed")
+  bci:listen_cancel("noun", id)
+  i = false
+  -- onrising
+  id = bci:listen_noun("noun", "onrising", nil, function(bc, noun)
+    i = true
+  end)
+  bci:set_noun("noun", 10)
+  assert_false(i, "Listening \"onrising\" failed(A)")
+  bci:set_noun("noun", 11)
+  assert_true(i, "Listening \"onrising\" failed(B)")
+  bci:listen_cancel("noun", id)
+  i = false
+  -- onfalling
+  id = bci:listen_noun("noun", "onfalling", nil, function(bc, noun)
+    i = true
+  end)
+  bci:set_noun("noun", 12)
+  assert_false(i, "Listening \"onfalling\" failed(A)")
+  bci:set_noun("noun", 10)
+  assert_true(i, "Listening \"onfalling\" failed(B)")
+  bci:listen_cancel("noun", id)
+  i = false
+  -- onvalue
+  id = bci:listen_noun("noun", "onvalue", 99, function(bc, noun)
+    i = true
+  end)
+  bci:set_noun("noun", 100)
+  assert_false(i, "Listening \"onvalue\" failed(A)")
+  bci:set_noun("noun", 99)
+  assert_true(i, "Listening \"onvalue\" failed(B)")
+  bci:listen_cancel("noun", id)
+  i = false
+  -- onabove
+  id = bci:listen_noun("noun", "onabove", 100, function(bc, noun)
+    i = true
+  end)
+  bci:set_noun("noun", 10)
+  assert_false(i, "Listening \"onabove\" failed(A)")
+  bci:set_noun("noun", 110)
+  assert_true(i, "Listening \"onabove\" failed(B)")
+  bci:listen_cancel("noun", id)
+  i = false
+  -- onbelow
+  id = bci:listen_noun("noun", "onbelow", 10, function(bc, noun)
+    i = true
+  end)
+  bci:set_noun("noun", 11)
+  assert_false(i, "Listening \"onbelow\" failed(A)")
+  bci:set_noun("noun", 1)
+  assert_true(i, "Listening \"onbelow\" failed(B)")
+  bci:listen_cancel("noun", id)
+end