Opened the gates to Multi-threading for the AP2 server instances.

Makes things smoother for multiple HomeKit connections.
Makes chasing bugs more fun/challenging.

Author: systemcrash

README.md

@@ -39,6 +39,17 @@ Next steps:
  - FairPlay v2 Support
 ---
 
+## Multiple Connections
+
+Since multithreading is now enabled, this allows multiple concurrent connections. There are no safeguards 
+built to prevent you playing multiple streams. Python multiprocessing makes this "DJ" mode a 
+possibility but makes stream management and session management (global state data) nigh impossible. So 
+threading is the right approach in the receiver. 
+
+HomeKit and other AP senders can now connect concurrently to the receiver and perform operations. This
+opens the path to Remote Control functionality.
+
+
 ## mDNS/ZeroConf
 
 If you encounter strange errors like NonUniqueNameException, or Address already in use, 

ap2-receiver.py

@@ -7,6 +7,8 @@
 import argparse
 import tempfile
 import multiprocessing
+import threading
+from threading import current_thread
 
 import pprint
 
@@ -111,6 +113,7 @@ def get_hex_bitmask(in_features):
 
 
 def update_status_flags(flag=None, on=False, push=True):
+    global MDNS_OBJ
     """ Use this to check for and send out updated status flags
     if flag is None, skip changing the flags (e.g. updates already queued)
     if on is true, add the flag in, otherwise remove it.
@@ -298,6 +301,7 @@ class AP2Handler(http.server.BaseHTTPRequestHandler):
     fairplay_keymsg = None
     ecdh_shared_key = None
     session = None
+    hap = None
 
     # Maps paths to methods a la HAP-python
     HANDLERS = {
@@ -353,7 +357,7 @@ def dispatch(self):
                 404,
                 f": Method {self.command} Path {path} endpoint not implemented"
             )
-            self.server.hap = None
+            self.hap = None
 
     def parse_request(self):
         self.raw_requestline = self.raw_requestline.replace(b"RTSP/1.0", b"HTTP/1.1")
@@ -507,10 +511,12 @@ def do_SETUP(self):
 
             self.server.streams.append(streamobj)
 
-            event_port, self.event_proc = EventGeneric.spawn(
-                self.server.server_address, name='events', shared_key=self.ecdh_shared_key, isDebug=DEBUG)
-            timing_port, self.timing_proc = EventGeneric.spawn(
-                self.server.server_address, name='ntp', shared_key=self.ecdh_shared_key, isDebug=DEBUG)
+            if not self.server.event_proc:
+                self.server.event_port, self.server.event_proc = EventGeneric.spawn(
+                    self.server.server_address, name='events', shared_key=self.ecdh_shared_key, isDebug=DEBUG)
+            if not self.server.timing_proc:
+                self.server.timing_port, self.server.timing_proc = EventGeneric.spawn(
+                    self.server.server_address, name='ntp', shared_key=self.ecdh_shared_key, isDebug=DEBUG)
             transport = self.headers["Transport"].split(';')
             res = []
             res.append("RTP/AVP/UDP")
@@ -551,10 +557,10 @@ def do_SETUP(self):
 
                 if "streams" not in plist:
                     self.logger.debug("Sending EVENT:")
-                    event_port, self.event_proc = EventGeneric.spawn(
+                    self.server.event_port, self.server.event_proc = EventGeneric.spawn(
                         self.server.server_address, name='events', shared_key=self.ecdh_shared_key, isDebug=DEBUG)
-                    device_setup["eventPort"] = event_port
-                    self.logger.debug(f"[+] eventPort={event_port}")
+                    device_setup["eventPort"] = self.server.event_port
+                    self.logger.debug(f"[+] eventPort={self.server.event_port}")
 
                     self.logger.debug(self.pp.pformat(device_setup))
                     res = writePlistToString(device_setup)
@@ -756,11 +762,10 @@ def do_TEARDOWN(self):
         # Send flag that we're no longer active
         update_status_flags(StatusFlags.getRecvSessActive(StatusFlags))
 
-        # Erase the hap() instance, otherwise reconnects fail
-        self.server.hap = None
-
-        if(self.timing_proc):
-            self.timing_proc.terminate()
+        try:
+            self.server.timing_proc.terminate()
+        except AttributeError:
+            pass
 
         if len(self.server.streams) == 0:
             session = None
@@ -938,12 +943,12 @@ def handle_pair_verify(self):
     def handle_pair_SV(self, op):
         body = self.rfile.read(int(self.headers["Content-Length"]))
 
-        if not self.server.hap:
-            self.server.hap = Hap(PI, DEBUG)
+        if not self.hap:
+            self.hap = Hap(PI, DEBUG)
         if op == 'verify':
-            res = self.server.hap.pair_verify(body)
+            res = self.hap.pair_verify(body)
         elif op == 'setup':
-            res = self.server.hap.pair_setup(body)
+            res = self.hap.pair_setup(body)
 
         self.send_response(200)
         self.send_header("Content-Length", len(res))
@@ -953,12 +958,12 @@ def handle_pair_SV(self, op):
         self.end_headers()
         self.wfile.write(res)
 
-        if self.server.hap.encrypted and self.server.hap.mfi_setup:
+        if self.hap.encrypted and self.hap.mfi_setup:
             self.logger.warning('MFi setup not yet possible. Disable feature bit 51.')
-        elif self.server.hap.encrypted:
-            hexdump(self.server.hap.accessory_shared_key) if DEBUG else ''
-            self.ecdh_shared_key = self.server.hap.accessory_shared_key
-            self.upgrade_to_encrypted(self.server.hap.accessory_shared_key)
+        elif self.hap.encrypted:
+            hexdump(self.hap.accessory_shared_key) if DEBUG else ''
+            self.ecdh_shared_key = self.hap.accessory_shared_key
+            self.upgrade_to_encrypted(self.hap.accessory_shared_key)
 
     def handle_pair_add(self):
         self.handle_pair_ARL('add')
@@ -976,11 +981,11 @@ def handle_pair_ARL(self, op):
         if content_len > 0:
             body = self.rfile.read(content_len)
             if op == 'add':
-                res = self.server.hap.pair_add(body)
+                res = self.hap.pair_add(body)
             elif op == 'remove':
-                res = self.server.hap.pair_remove(body)
+                res = self.hap.pair_remove(body)
             elif op == 'list':
-                res = self.server.hap.pair_list(body)
+                res = self.hap.pair_list(body)
             hexdump(res) if DEBUG else ''
             self.send_response(200)
             self.send_header("Content-Type", self.headers["Content-Type"])
@@ -1049,7 +1054,7 @@ def handle_configure(self):
                 HK_PW = pw
                 DEV_PROPS.setDevicePassword(pw)
 
-        accessory_id, accessory_ltpk = self.server.hap.configure()
+        accessory_id, accessory_ltpk = self.hap.configure()
         configure_info = {
             'Identifier': accessory_id.decode('utf-8'),
             hkac_s: hkac,
@@ -1166,7 +1171,7 @@ def unregister_mdns(zeroconf, info):
         zeroconf.close()
 
 
-class AP2Server(socketserver.TCPServer):
+class AP2Server(socketserver.ThreadingTCPServer):
     # Fixes 99% of scenarios on restart after we terminate uncleanly/crash
     # and port was not closed before crash (is still open).
     # AP2 client connects from random port.
@@ -1176,24 +1181,46 @@ class AP2Server(socketserver.TCPServer):
     def __init__(self, addr_port, handler):
         super().__init__(addr_port, handler)
         self.connections = {}
-        self.hap = None
+        """ Handle the HAP object here: it's not a fact that the HAP
+        connection is torn down, when the sessions and streams are. This means
+        HomeKit, RemoteControl and other niceties continue to work.
+        """
+        self.serv_addr, self.serv_port = addr_port
+        # self.hap = None  # thread local, not global.
+        self.event_proc = None
+        self.event_port = None
+        self.timing_proc = None
+        self.timing_port = None
         self.enc_layer = False
         self.streams = []
+        self.sessions = []
+        log_string = f'{self.__class__.__name__}: {self.serv_addr}:{self.serv_port}'
+        level = 'DEBUG' if DEBUG else 'INFO'
+        self.logger = get_screen_logger(log_string, level=level)
 
     # Override
     def get_request(self):
-        # Quick clean-up in case anything from before is still around.
-        self.hap = None
         client_socket, client_addr = super().get_request()
-        SCR_LOG.info(f"Opened connection from {client_addr[0]}:{client_addr[1]}")
+        self.logger.info(f"Opened connection from {client_addr[0]}:{client_addr[1]}")
         self.connections[client_addr] = client_socket
         return (client_socket, client_addr)
 
     def upgrade_to_encrypted(self, client_address, shared_key):
         client_socket = self.connections[client_address]
-        hap_socket = HAPSocket(client_socket, shared_key)
-        self.connections[client_address] = hap_socket
-        return hap_socket
+        self.hap_socket = HAPSocket(client_socket, shared_key)
+        self.logger.info(f"{current_thread().name}: Opened HAPSocket from {client_address[0]}:{client_address[1]}")
+        self.connections[client_address] = self.hap_socket
+        return self.hap_socket
+
+    # Override
+    def server_close(self):
+        if self.logger:
+            self.logger.debug('Removing AP2Server object.')
+        self.hap = None
+        self.hap_socket = None
+        self.streams.clear()
+        self.logger = None
+        self.shutdown()
 
 
 def list_network_interfaces():
@@ -1362,15 +1389,15 @@ def list_available_flags():
         PORT = 7000
         if IPV6 and not IPV4:
             with AP2Server((IPV6, PORT), AP2Handler) as httpd:
-                SCR_LOG.info(f"serving at port {PORT}")
                 IPADDR_BIN = IP6ADDR_BIN
                 IPADDR = IPV6
+                SCR_LOG.info(f"serving on {IPADDR}:{PORT}")
                 httpd.serve_forever()
         else:  # i.e. (IPV4 and not IPV6) or (IPV6 and IPV4)
             with AP2Server((IPV4, PORT), AP2Handler) as httpd:
-                SCR_LOG.info(F"serving at port {PORT}")
                 IPADDR_BIN = IP4ADDR_BIN
                 IPADDR = IPV4
+                SCR_LOG.info(f"serving on {IPADDR}:{PORT}")
                 httpd.serve_forever()
 
     except KeyboardInterrupt:

ap2/connections/event.py

@@ -1,6 +1,7 @@
 import socket
 import multiprocessing
 import os
+import time
 
 from ..utils import get_file_logger, get_free_port
 
@@ -35,10 +36,14 @@ def serve(self):
                 event_file = open(self.file, "wb")
             try:
                 while True:
-                    data = conn.recv(1)
+                    data = conn.recv(1, socket.MSG_WAITALL)
                     if data:
                         event_file.write(data)
                         pass
+                    else:
+                        # This while loop can run away.
+                        break
+                        raise KeyboardInterrupt
             except KeyboardInterrupt:
                 pass
             finally:

ap2/utils.py

@@ -80,6 +80,11 @@
             'level': 'DEBUG',
             'propagate': False,
         },
+        'Receiver*': {
+            'handlers': ['console'],
+            'level': 'DEBUG',
+            'propagate': False,
+        },
         'RTPBuffer': {
             'handlers': ['console'],
             'level': 'DEBUG',
@@ -123,7 +128,7 @@ def get_file_logger(name, level="INFO"):
 def get_screen_logger(name, level="INFO"):
     logger = logging.getLogger(name)
     logger.setLevel(level)
-    logger.propagate = False
+    # logger.propagate = False
     if level == 'DEBUG':
         print(f'[{name}] logging level: {level}')
     return logger