synchronize startup process

Author: Jackiebibili

apps/startup/apps.py

@@ -1,15 +1,40 @@
 from django.apps import AppConfig
-from ..ticketscraping.scraping import start
 from datetime import datetime
 from threading import Thread
+from multiprocessing import Process
+
+
+def run_prepare():
+    # import module inside the child process to prevent execution in the parent process
+    print(
+        f"ticket scraping service started at {datetime.now().strftime('%d/%m/%Y %H:%M:%S')}")
+
+    # start sender socket
+    from apps.ticketscraping.schedulers.async_tasks_scheduler import async_tasks_scheduler
+    conn_thread = Thread(target=async_tasks_scheduler.connect)
+    conn_thread.start()
+    # wait for async tasks handler to connect
+    conn_thread.join()
+
+    # start itself (scraping)
+    from apps.ticketscraping.scraping import start
+    start()
+
+
+def run():
+    # starter
+    p = Process(target=run_prepare, daemon=True)
+    p.start()
+    # start receiver socket
+    from apps.ticketscraping.connection.asyn_tasks_receiver import run
+    conn_process = Process(target=run)
+    conn_process.start()
+    conn_process.join()
 
 
 class MyAppConfig(AppConfig):
    name = "apps.startup"
    verbose_name = "start tmtracker"
 
    def ready(self):
-      print(
-          f"server started at {datetime.now().strftime('%d/%m/%Y %H:%M:%S')}")
-      # start scraping
-      Thread(target=start).start()
+        run()

apps/ticketscraping/connection/asyn_tasks_receiver.py

@@ -0,0 +1,35 @@
+# start sockets
+from threading import Thread
+from multiprocessing import Process
+from apps.ticketscraping.connection.receiver_process import ReceiverProcess
+from apps.ticketscraping.constants import SERVICE_LOCALHOST, ASYNC_TASKS_RECEIVER_PORT
+
+
+def run_prepare():
+   # start receiver socket
+   from apps.ticketscraping.connection.mail_receiver import run
+   conn_process = Process(target=run, daemon=True)
+   conn_process.start()
+
+   # start sender socket
+   from apps.ticketscraping.schedulers.mail_scheduler import mail_scheduler
+   conn_thread = Thread(target=mail_scheduler.connect)
+   conn_thread.start()
+   # wait for mailer to connect
+   conn_thread.join()
+
+   # start itself
+   from apps.ticketscraping.tasks.asynchronous import run_async_tasks
+   receiver = ReceiverProcess(run_async_tasks, SERVICE_LOCALHOST, ASYNC_TASKS_RECEIVER_PORT)
+   receiver.connect()
+   receiver.serve_forever()
+
+
+def run():
+   # starter
+   p = Process(target=run_prepare)
+   p.start()
+
+
+if __name__ == '__main__':
+   run()

apps/ticketscraping/connection/mail_receiver.py

@@ -0,0 +1,13 @@
+from apps.ticketscraping.connection.receiver_process import ReceiverProcess
+# from ..tasks.asynchronous import run_async_tasks
+from apps.ticketscraping.constants import SERVICE_LOCALHOST, MAIL_RECEIVER_PORT
+
+def run():
+   # start itself
+   receiver = ReceiverProcess(lambda x: print(
+       x), SERVICE_LOCALHOST, MAIL_RECEIVER_PORT)
+   receiver.connect()
+   receiver.serve_forever()
+
+if __name__ == '__main__':
+   run()

apps/ticketscraping/connection/receiver.py

@@ -0,0 +1,24 @@
+from multiprocessing.connection import Client
+from threading import Semaphore
+
+class Receiver:
+   def __init__(self, hostname: str, port: int):
+      self.lock = Semaphore(1)
+      self.hostname = hostname
+      self.port = port
+      self.conn = None
+
+   def connect(self):
+      self.conn = Client(address=(self.hostname, self.port,))
+
+   def recv(self):
+      if self.conn is None:
+         raise Exception('connection is not established')
+      self.lock.acquire()
+      res = self.conn.recv()
+      self.lock.release()
+      return res
+
+   def __del__(self):
+      if self.conn is not None: self.conn.close()
+

apps/ticketscraping/connection/receiver_process.py

@@ -0,0 +1,11 @@
+from .receiver import Receiver
+
+class ReceiverProcess(Receiver):
+   def __init__(self, action, hostname: str, port: int):
+      super().__init__(hostname, port)
+      self.action = action
+
+   def serve_forever(self):
+      while True:
+         res = self.recv()
+         self.action(*res)

apps/ticketscraping/connection/sender.py

@@ -0,0 +1,26 @@
+from multiprocessing.connection import Listener
+from threading import Semaphore
+
+class Sender:
+   def __init__(self, hostname: str, port: int):
+      self.lock = Semaphore(1)
+      self.hostname = hostname
+      self.port = port
+      self.conn = None
+
+   def connect(self):
+      listener = Listener(address=(self.hostname, self.port))
+      self.conn = listener.accept()
+      print("conn accepted ", self.port)
+
+   def send(self, *args):
+      if self.conn is None:
+         raise Exception('connection is not established')
+      self.lock.acquire()
+      self.conn.send(args)
+      self.lock.release()
+      return True
+
+   def __del__(self):
+      if self.conn is not None:
+         self.conn.close()

apps/ticketscraping/constants.py

@@ -1,6 +1,10 @@
-from functools import reduce
 from uuid import uuid4
 
+# services - async action handlers
+ASYNC_TASKS_RECEIVER_PORT = 8100
+MAIL_RECEIVER_PORT = 8200
+SERVICE_LOCALHOST = 'localhost'
+
 ANTIBOT_JS_CODE_URL = "https://epsf.ticketmaster.com/eps-d"
 TOKEN_INTERROGATION_URL = "https://epsf.ticketmaster.com/eps-d?d=www.ticketmaster.com"
 

apps/ticketscraping/schedulers/async_task_scheduler.py

@@ -0,0 +1,4 @@
+from ..connection.sender import Sender
+from ..constants import SERVICE_LOCALHOST, ASYNC_TASKS_RECEIVER_PORT
+
+async_tasks_scheduler = Sender(SERVICE_LOCALHOST, ASYNC_TASKS_RECEIVER_PORT)