Keep alive upd multicast

multicast-keep-alive/receiver.py

 import socket
 import struct
 import sys
+from threading import *
+import time
 
-samples_count = int(sys.argv[1])
+host = "192.168.2.1"
+port = 8000
+one_microsecond = 1 / 1000000.
 multicast_group = '224.3.29.71'
 server_address = ('', 10000)
+timeout = int(sys.argv[1])
+samples_count = int(sys.argv[2])
 
-# Create the socket
-sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+# send to edge server unicast keep-alive messages
+def send_msg():
+  while True:
+    msg = str(int(time.time() * 1000000))
+    s.send(msg.encode("utf-8"))
+    print "\t-> %s (unicast)" %msg
+    time.sleep(timeout * one_microsecond)
 
-# Bind to the server address
-sock.bind(server_address)
+s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+s.connect((host,port))
+send_thread = Thread(target = send_msg)
+send_thread.daemon = True
+send_thread.start()
+#s.close()
 
+# Subscribe to UDP datagrams published from edge server
+sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+sock.bind(server_address)
 # Tell the operating system to add the socket to the multicast group
 # on all interfaces.
 group = socket.inet_aton(multicast_group)
@@ -19,18 +37,15 @@ mreq = struct.pack('4sL', group, socket.INADDR_ANY)
 sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
 
 last_micro_second = 0
-# Receive/respond loop
 i = 0
 l = []
 while i < samples_count:
     data, address = sock.recvfrom(1024)
     current_micro_second = int(data)
     diff = current_micro_second - last_micro_second
-    #if diff > 200:
-    #    print >>sys.stderr, diff
     last_micro_second = current_micro_second
     if i > 0:
-      # omot first cycle as we care for diff between cycles
+      # omit first cycle as we care for diff between cycles
       l.append(str(diff))
     i += 1
 

multicast-keep-alive/sender.py

 import socket
 import struct
 import sys, time
+from threading import *
 
+one_microsecond = 1 / 1000000.0
+host = "192.168.2.1"
+port = 8000
 multicast_group = ('224.3.29.71', 10000)
+timeout = int(sys.argv[1])
 
-# Create the datagram socket
-sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+# open socket unicast port for connection from couplers
+serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+serversocket.bind((host, port))
+serversocket.listen(5)
 
-one_microsecond = 1 / 1000000.0
+def receive_msg():
+  conn, addr = serversocket.accept()
+  while True:
+    try:
+      msg = conn.recv(1024).decode("utf8")
+      print "\t<- %s (unicast)" %msg
+    except OSError as error:
+      return error
+
+receive_thread = Thread(target=receive_msg)
+receive_thread.daemon = True
+receive_thread.start()
+print  "Started unicast keep-alive server"
+
+print "Start sending UDP keep-alive datagrams"
+sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
 # Set the time-to-live for messages to 1 so they do not go past the
 # local network segment.
 ttl = struct.pack('b', 1)
 sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, ttl)
 last_micro_seconds = int(time.time() * 1000000)
 
-# XXX: make CLI arguments
-timeout = int(sys.argv[1])
-timeout_tolerance = int(sys.argv[2])
-warning_limit = timeout + timeout_tolerance
-
+# send over UDP keep-alive messages
 while True:
     micro_seconds = int(time.time() * 1000000)
     diff = micro_seconds - last_micro_seconds
     last_micro_seconds = micro_seconds
     message = str(micro_seconds)
-    if diff / 1000000 > warning_limit:
-        print >>sys.stderr, '%s' %diff
+    print "-> %s (udp)" %message
     sent = sock.sendto(message, multicast_group)
     time.sleep(timeout * one_microsecond) # wait interval