Generic TCP Server for external requests

[Matt]

I am not a big fan of the example : http://openrb.com/example-lm2-as-tcp-ser...-requests/

It uses the following principle :
- the resident script is the TCP server :
  - manages TCP clients
  - listen on UDP for internal LM requests
- the resident script do receive client request and process them
- other LM scripts/events use UDP to communicate to the TCP server and transmit data to be sent

For me, there is some limitations :
- only 5 messages per second are read by TCP server on the UDP side
- data can be lost with UDP

As I need to send a lot of data, I did use the internal storage instead of UDP.
I am sharing this as a generic TCP server that can be customized for every need.

So the idea is the following :
- the resident script is the TCP server :
  - manage TCP clients
  - read the internal storage for internal LM requests
- the resident script do receive client request and process them
- other LM scripts/events use internal storage to communicate to the TCP server and transmit data to be sent


First, the User Library :

Code:

-- TCP Server user library

--general debug
tcp_debug = false
--networking debug
tcp_debug_tcp = false
-- additionnal debug
tcp_debug_verbose = false

--store a message to be send into the queue
function tcp_queue(message)
 if (tcp_debug_verbose) then
   alert("tcp_queue : "..message)
 end

 local tcp_Q = storage.get('tcpQ')
 table.insert(tcp_Q,message)
    storage.set('tcpQ', tcp_Q)
end

Now the resident script with sleep interval 0 :

Code:

require('copas')

--[[
Resident script for TCP server
Manage client connexions
Receive requests from clients
Also send events through tcpQ storage

debug variables are set in the user library
]]--

--communication port to listen on
local tcp_port = 6789

if (tcp_debug) then
 alert ("TCP Server start")
end


-- send a message to the peers
function tcp_sendmessage(message)
 if (tcp_debug_tcp) then
   alert ("tcp_sendmessage : ".. message)
    end
 for id, sock in pairs(tcp_clients) do
   sock:send(message .. '\r\n')
 end
end

   
-- incoming data handler, manage each message sent by clients to LM
function tcp_datahandler(sock, data)
 local ip, port
 ip, port = sock:getpeername()
   
 if tcp_debug_tcp then
   alert('tcp_datahandler receive data from %s:%d - %s', ip, port, data)
 end

-- add your code to process data

end


-- connection handler, manage new peers
function tcp_connhandler(sock)

 -- enable keep-alive to check for disconnect events
 sock:setoption('keepalive', true)

 local ip, port, data, err, id

 -- get ip and port from socket
 ip, port = sock:getpeername()

 -- client id
 id = string.format('%s:%d', ip, port)

 alert('tcp_connhandler connection from %s', id)

 -- save socket reference
 tcp_clients[ id ] = sock

 -- main reader loop
 while true do
   -- wait for single line of data (until \n, \r is ignored)
   data, err = copas.receive(sock, '*l')

   -- error while receiving
   if err then
     alert('tcp_connhandler closed connection from %s:%d', ip, port)
     -- remove socket reference
     tcp_clients[ id ] = nil
     return
   end

   -- handle data frame
   tcp_datahandler(sock, data)
 end
end



-- init server handler
if not tcp_ready then
 
 if (tcp_debug) then
   alert ("TCP Server init start")
 end

    -- list of client sockets
 tcp_clients = {}

 --variable for the queue
 tcp_Q = {}
 --empty the storage queue
 storage.set('tcpQ', tcp_Q)

 -- bind to port
 tcp_tcpserver = socket.bind('*', tcp_port)

 -- error while binding, try again later
 if not tcp_tcpserver then
   os.sleep(5)
   error('tcp server realization init error: cannot bind')
 end

 -- set server connection handler
 copas.addserver(tcp_tcpserver, tcp_connhandler)

 tcp_ready = true
 
 if (tcp_debug) then
   alert ("TCP Server init end")
    end
end

-- loop to manage packets
while true do

 --retreive the queue
 tcp_Q = storage.get('tcpQ')
 storage.set('tcpQ', {})

 --sending the queue
 for index, value in ipairs(tcp_Q) do
   if tcp_debug then
        alert("TCP Server sending item "..tostring(index).." : "..tostring(value))
   end
   tcp_sendmessage(value)
 end

 --wait for tcp request during 0.1s
 copas.step(0.1)
end

if (tcp_debug) then
 alert ("TCP Server closed")
end

Event base script :

Code:

value = knxdatatype.decode(event.datahex, dt.bool)
tcp_queue('Lamp is' .. (value and 'ON' or 'OFF'))

Matthieu

[admin]

Our example can be tuned for more messages per second if needed, but it will increase CPU load slightly.
It's true that UDP packets can be lost over a network. In our example UDP communication happens locally over a virtual loopback interface, so it should be fine.

You example has a race conditions between storage.get and storage.set calls which can lead to some messages being lost.

Anyway, if you really need to process a large amount of data you should use some ready-made solutions like MQTT.

[Matt]

Thanks for the feedback, in my case (used with RTI, my user interface) the risk to send 10 to 20 message at the exact same time.
How will the UDP socket react if 10 packets come in one iteration ?
So far this method works fine.
If we could call a function of a residential script from an event script it would be perfect but I didn't find another way to do it.
Basically, it's more a service/daemon than a script.

[admin]

You can find an example of group monitoring script here:
http://forum.logicmachine.net/showthread.php?tid=31

Sockets are buffered on OS level so queue must get really big in order for packets to get dropped.

[Matt]

I am testing UDP again to compare.

Matthieu

---

I had issues with 10+ UDP requests per second with the following code :

Code:

while true do
 message = udpserver:receive()

 -- got message from udp, send to all active clients
 if message then
   for id, sock in pairs(clients) do
     sock:send(message .. '\r\n')
   end
 end

 copas.step(0.1)
end

With this code, it's working better :

Code:

while true do
 message = udpserver:receive()

 while message do
 -- got message from udp, send to all active clients
   for id, sock in pairs(clients) do
     sock:send(message .. '\r\n')
   end
   message = udpserver:receive()
 end

 copas.step(0.1)
end

It does process the UDP queue at each iteration instead of one by one.
What do you think about it ?
Matthieu

[admin]

You can try setting UDP socket timeout to 0, so that receive function returns either new message at once or nil + "timeout" notification. This should speed things up for you.