orgnized codes and add dynamic recv

Author: shokakucarrier

charon/pkgs/radas_signature_handler.py

@@ -19,13 +19,14 @@
 import os
 import asyncio
 import sys
+import tempfile
 import uuid
 from typing import List, Any, Tuple, Callable, Dict
 from charon.config import get_config, RadasConfig
 from charon.pkgs.oras_client import OrasClient
-from proton import SSLDomain, SSLException, Message, Event
+from proton import SSLDomain, Message, Event
 from proton.handlers import MessagingHandler
-from proton.utils import BlockingConnection
+from proton.reactor import Container
 
 logger = logging.getLogger(__name__)
 
@@ -38,9 +39,12 @@ class UmbListener(MessagingHandler):
         this value transfers from the cmd flag, should register UmbListener when the client starts
     """
 
-    def __init__(self, sign_result_loc: str) -> None:
+    def __init__(self, sign_result_loc: str, payload: str) -> None:
         super().__init__()
         self.sign_result_loc = sign_result_loc
+        self.payload = payload
+        self.conn = None
+        self.timer_task = None  
 
     def on_start(self, event: Event) -> None:
         """
@@ -54,10 +58,33 @@ def on_start(self, event: Event) -> None:
         # explicit check to pass the type checker
         if rconf is None:
             sys.exit(1)
-        conn = event.container.connect(rconf.umb_target())
-        event.container.create_receiver(conn, rconf.result_queue())
+        
+        ssl_domain = SSLDomain(SSLDomain.MODE_CLIENT)
+        ssl_domain.set_credentials(
+            rconf.client_ca(),
+            rconf.client_key(),
+            rconf.client_key_password()
+        )
+        ssl_domain.set_trusted_ca_db(rconf.root_ca())
+        ssl_domain.set_peer_authentication(SSLDomain.VERIFY_PEER)
+
+        self.conn = event.container.connect(
+            url = rconf.umb_target(), 
+            ssl_domain=ssl_domain
+        )
+        self.sender = event.container.create_sender(self.conn)
+        self.receiver = event.container.create_receiver(self.conn, rconf.result_queue(), dynamic=True)
         logger.info("Listening on %s, queue: %s", rconf.umb_target(), rconf.result_queue())
 
+    def on_link_opened(self, event: Event) -> None:
+        """
+        On sender ready callback
+        """
+        if event.receiver == self.receiver:
+            if self.receiver.remote_source.address:
+                self.sender.send(Message(body=self.payload))
+                self.sender.close()
+
     def on_message(self, event: Event) -> None:
         """
         On message callback
@@ -113,8 +140,8 @@ def sign_in_radas(repo_url: str,
     logger.debug("params. repo_url: %s, requester: %s, sign_key: %s, result_path: %s,"
                  "radas_config: %s", repo_url, requester, sign_key, result_path, radas_config)
     request_id = str(uuid.uuid4())
-    exclude = list(ignore_patterns) if ignore_patterns else []
-    
+    exclude = ignore_patterns if ignore_patterns else []
+
     payload = {
         "request_id": request_id,
         "requested_by": requester,
@@ -124,33 +151,15 @@ def sign_in_radas(repo_url: str,
         "exclude": exclude
     }
 
-    try:
-        ssl_domain = SSLDomain(SSLDomain.MODE_CLIENT)
-        ssl_domain.set_credentials(
-            radas_config.client_ca(),
-            radas_config.client_key(),
-            radas_config.client_key_password()
-        )
-        ssl_domain.set_trusted_ca_db(radas_config.root_ca())
-        ssl_domain.set_peer_authentication(SSLDomain.VERIFY_PEER)
-
-        conn = BlockingConnection(radas_config.umb_target(), ssl_domain=ssl_domain)
-        try:
-            sender = conn.create_sender(radas_config.request_queue())
-            message = Message(body=json.dumps(payload))
-            sender.send(message)
-            logger.info("Successfully sent signing request ID: %s", request_id)
-        finally:
-            conn.close()
-            
-    except SSLException as e:
-        logger.error("SSL connection failed: %s", str(e))
-        sys.exit(1)
-    except Exception as e:
-        logger.error("Failed to send signing request: %s", str(e))
-        sys.exit(1)
+    temp_dir = tempfile.mktemp()
+    listener = UmbListener(temp_dir, json.dumps(payload))
 
-    # wait for AMQP message to be consumed then get response message from UMB
+    try:
+        Container(listener).run()
+        logger.info("Successfully sent signing request ID: %s", request_id)
+    finally:
+        if listener.conn and not listener.conn.closed:
+            listener.conn.close()
 
 
 def generate_radas_sign(top_level: str, sign_result_loc: str) -> Tuple[List[str], List[str]]: