1+ import json
2+
3+ import logging
4+ import time
5+ from ..config .config import TopicConfig
6+ from ..resource_errors import ExceptionHandler
7+ from concurrent .futures import ThreadPoolExecutor
8+ from .abstract .topic_abstract import TopicAbstract
9+ from ..queue .models .queue_message import QueueMessage
10+ from azure .servicebus import ServiceBusClient , ServiceBusMessage
11+ from azure .servicebus import AutoLockRenewer
12+ import concurrent .futures as cf
13+ import threading
14+
15+
16+ logging .basicConfig (format = '%(asctime)s %(levelname)-8s %(message)s' , datefmt = '%Y-%m-%d %H:%M:%S' )
17+ logger = logging .getLogger ('AzureTopic' )
18+ logger .setLevel (logging .INFO )
19+
20+
21+ """
22+ AzureTopic class represents a topic in Azure Service Bus.
23+ Attributes:
24+ topic (str): The name of the topic.
25+ client (ServiceBusClient): The ServiceBusClient object used to interact with the Service Bus.
26+ max_concurrent_messages (int): The maximum number of concurrent messages to process.
27+ topic_name (str): The name of the topic.
28+ publisher (TopicSender): The TopicSender object used to send messages to the topic.
29+ executor (ThreadPoolExecutor): The ThreadPoolExecutor object used to execute callback functions.
30+ internal_count (int): The internal count of concurrent messages being processed.
31+ lock_renewal (AutoLockRenewer): The AutoLockRenewer object used to renew message locks.
32+ max_renewal_duration (int): The maximum duration in seconds to renew a message lock.
33+ wait_time_for_message (int): The maximum wait time in seconds to receive messages.
34+ Methods:
35+ publish(data: QueueMessage) -> None:
36+ Publishes a message to the topic.
37+ subscribe(subscription: str, callback) -> None:
38+ Subscribes to a subscription of the topic and processes incoming messages.
39+ internal_callback(message, callbackfn) -> ServiceBusMessage:
40+ Internal callback function that processes a message and invokes the callback function.
41+ settle_message(x: cf.Future) -> None:
42+ Sets the message as completed and updates the internal count.
43+ """
44+ class AzureTopic (TopicAbstract ):
45+ def __init__ (self , config : TopicConfig = None , topic_name = None , max_concurrent_messages :int = 1 ):
46+ self .topic = topic_name
47+ self .client = ServiceBusClient .from_connection_string (conn_str = config .connection_string , retry_total = 10 , retry_backoff_factor = 1 , retry_backoff_max = 30 )
48+ self .max_concurrent_messages = max_concurrent_messages
49+ self .topic_name = topic_name
50+ self .publisher = self .client .get_topic_sender (topic_name = topic_name )
51+ self .executor = ThreadPoolExecutor (max_workers = max_concurrent_messages )
52+ self .internal_count = 0
53+ self .lock_renewal = AutoLockRenewer (max_workers = max_concurrent_messages )
54+ self .max_renewal_duration = 86400 # Renew the message upto 1 day
55+ self .wait_time_for_message = 5
56+ self .thread_lock = threading .Lock ()
57+
58+
59+ def publish (self , data : QueueMessage ):
60+ """
61+ Publishes a message to the topic.
62+ Args:
63+ data (QueueMessage): The message to publish.
64+
65+ """
66+ message = QueueMessage .to_dict (data )
67+ self .publisher .send_messages (ServiceBusMessage (json .dumps (message )))
68+
69+ def subscribe (self , subscription : str , callback ):
70+
71+ """
72+ Subscribes to a subscription of the topic and processes incoming messages.
73+ Args:
74+ subscription (str): The name of the subscription to subscribe to.
75+ callback (function): The callback function to invoke for each message.
76+ """
77+ self .receiver = self .client .get_subscription_receiver (topic_name = self .topic_name , subscription_name = subscription )
78+ while True :
79+ try :
80+ to_receive = (self .max_concurrent_messages - self .internal_count )
81+ if to_receive > 0 :
82+ messages = self .receiver .receive_messages (max_message_count = to_receive , max_wait_time = self .wait_time_for_message )
83+ if not messages or len (messages ) == 0 :
84+ continue
85+ for message in messages :
86+ self .lock_renewal .register (self .receiver , message , max_lock_renewal_duration = self .max_renewal_duration )
87+ execution_task = self .executor .submit (self .internal_callback , message , callback )
88+ execution_task .add_done_callback (lambda x : self .settle_message (x ))
89+ else :
90+ time .sleep (self .wait_time_for_message )
91+ except Exception as e :
92+ logger .error (f'Error in receiving messages: { e } )
93+
94+
95+ def internal_callback (self , message , callbackfn ):
96+ """
97+ Internal callback function that processes a message and invokes the callback function.
98+ Args:
99+ message (ServiceBusMessage): The message to process.
100+ callbackfn (function): The callback function to invoke.
101+ Returns:
102+ ServiceBusMessage: The processed message.
103+ """
104+ try :
105+ with self .thread_lock :
106+ self .internal_count += 1 # thread safe.
107+ queue_message = QueueMessage .data_from (str (message ))
108+ callbackfn (queue_message )
109+ return [True ,message ]
110+ except Exception as e :
111+ logger .error (f'Error in processing message: { e } )
112+ return [False ,message ]
113+
114+
115+ def settle_message (self , x : cf .Future ):
116+ """
117+ Sets the message as completed and updates the internal count.
118+ Args:
119+ x (cf.Future): The future object representing the message processing.
120+ """
121+ # Lock the internal count
122+ with self .thread_lock :
123+ self .internal_count -= 1
124+ # Check if the future has an exception
125+ [is_success ,incoming_message ] = x .result ()
126+ if is_success :
127+ self .receiver .complete_message (incoming_message )
128+ else :
129+ print (f'Abandoning message: { incoming_message } )
130+ self .receiver .abandon_message (incoming_message ) # send back to the topic
131+ return
132+
133+
0 commit comments