#!/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, configPath : str, config : Server.Server): self.context = zmq.Context() # find the key directories serverKeyPath = os.path.join(os.path.join(configPath, 'keys'), 'server') if False == os.path.isdir(serverKeyPath): sys.stderr.write('No directory for the server key found') sys.exit(-1) print('Path to the server key: ' + serverKeyPath) clientKeyPath = os.path.join(os.path.join(configPath, 'keys'), 'clients') if False == os.path.isdir(clientKeyPath): sys.stderr.write('No directory for the client keys found') sys.exit(-1) print('Path to the client keys: ' + clientKeyPath) # read the certificates keyPairPath = glob.glob(os.path.join(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()