87 lines
3.2 KiB
Python
87 lines
3.2 KiB
Python
#!/usr/bin/env python
|
|
|
|
import ctypes
|
|
import glob
|
|
import os
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
import zmq
|
|
import zmq.auth
|
|
|
|
from aman.com import AircraftReport_pb2
|
|
from aman.config import Server
|
|
|
|
class ReceiverThread(threading.Thread):
|
|
def __init__(self, socket):
|
|
threading.Thread.__init__(self)
|
|
self.socket = socket
|
|
|
|
def run(self):
|
|
while True:
|
|
try:
|
|
msg = self.socket.recv(zmq.NOBLOCK)
|
|
|
|
# parse the received message
|
|
report = AircraftReport_pb2.AircraftReport()
|
|
report.ParseFromString(msg)
|
|
except zmq.ZMQError as error:
|
|
if zmq.EAGAIN == error.errno:
|
|
time.sleep(0.5)
|
|
continue
|
|
else:
|
|
return
|
|
|
|
def threadId(self):
|
|
if hasattr(self, '_thread_id'):
|
|
return self._thread_id
|
|
for id, thread in threading._active.items():
|
|
if thread is self:
|
|
return id
|
|
|
|
def stopThread(self):
|
|
id = self.threadId()
|
|
res = ctypes.pythonapi.PyThreadState_SetAsyncExc(id, ctypes.py_object(SystemExit))
|
|
if 1 < res:
|
|
ctypes.pythonapi.PyThreadState_SetAsyncExc(id, 0)
|
|
|
|
# @brief Receives and sends messages to EuroScope plugins
|
|
class Euroscope:
|
|
# @brief Initializes the ZMQ socket
|
|
# @param[in] config The server configuration
|
|
def __init__(self, config : Server.Server):
|
|
self.context = zmq.Context()
|
|
|
|
# read the certificates
|
|
keyPairPath = glob.glob(os.path.join(config.ServerKeyPath, '*.key_secret'))
|
|
if 1 != len(keyPairPath):
|
|
sys.stderr.write('No public-private keypair found for the server certificate')
|
|
sys.exit(-1)
|
|
keyPair = zmq.auth.load_certificate(keyPairPath[0])
|
|
|
|
# initialize the receiver
|
|
self.receiverSocket = zmq.Socket(self.context, zmq.SUB)
|
|
self.receiverSocket.setsockopt(zmq.CURVE_PUBLICKEY, keyPair[0])
|
|
self.receiverSocket.setsockopt(zmq.CURVE_SECRETKEY, keyPair[1])
|
|
self.receiverSocket.setsockopt(zmq.CURVE_SERVER, True)
|
|
self.receiverSocket.bind('tcp://' + config.Address + ':' + str(config.PortReceiver))
|
|
self.receiverSocket.setsockopt(zmq.SUBSCRIBE, b'')
|
|
self.receiverThread = ReceiverThread(self.receiverSocket)
|
|
self.receiverThread.start()
|
|
print('Listening at tcp://' + config.Address + ':' + str(config.PortReceiver))
|
|
|
|
# initialize the notification
|
|
self.notificationSocket = zmq.Socket(self.context, zmq.PUB)
|
|
self.notificationSocket.setsockopt(zmq.CURVE_PUBLICKEY, keyPair[0])
|
|
self.notificationSocket.setsockopt(zmq.CURVE_SECRETKEY, keyPair[1])
|
|
self.notificationSocket.setsockopt(zmq.CURVE_SERVER, True)
|
|
self.notificationSocket.bind('tcp://' + config.Address + ':' + str(config.PortNotification))
|
|
print('Publishing at tcp://' + config.Address + ':' + str(config.PortNotification))
|
|
|
|
def __del__(self):
|
|
self.receiverThread.stopThread()
|
|
self.receiverThread.join()
|
|
self.receiverSocket.close()
|
|
self.notificationSocket.close()
|