UPD: Added basic support for timeout to cli_proton_python.sender

Author: fgiorgetti

cli_proton_python/sender.py

@@ -33,6 +33,31 @@
 from cli_proton_python import coreclient, options, utils
 
 
+class Timeout(object):  # pylint: disable=too-few-public-methods
+    """ Scheduler object for timeout control """
+
+    def __init__(self, parent, event):
+        """
+        Timeout constructor
+        saving parent_ptr to get current data
+        :param parent: related sender object
+        :type parent: cli_proton_python.Send
+        :param event: reactor event
+        :type event: proton.Event
+        """
+        self.parent_ptr = parent
+        self.parent_event = event
+
+    def on_timer_task(self, _):
+        """ on_timer_task action handler
+        """
+        # Stops container
+        if self.parent_event.container:
+            self.parent_event.container.stop()
+
+        self.parent_ptr.tear_down(self.parent_event)
+
+
 class Send(coreclient.CoreClient):
     """ Proton reactive API python sender client
 
@@ -207,6 +232,21 @@ def prepare_message(self):
 
         return msg
 
+    def schedule_timeout(self, event):
+        """
+        Cancels an existing timeout (if one exists).
+        Next it schedules a new timeout if "--timeout" provided.
+        :param event: Reactor event
+        :type event: proton.Event
+        """
+        # Cancels an existing timeout
+        if self.timeout:
+            self.timeout.cancel()
+
+        # Schedules a new timeout if --timeout provided.
+        if self.opts.timeout:
+            self.timeout = event.reactor.schedule(self.opts.timeout, Timeout(self, event))
+
     def send_message(self):
         """ sends a message """
         # close the connection if nothing to send
@@ -221,6 +261,9 @@ def send_message(self):
 
         self.print_message(self.msg)
 
+        # set up a new timeout (if one provided)
+        self.schedule_timeout(self.event)
+
         if self.opts.log_stats == 'endpoints':
             utils.dump_event(self.event)
 
@@ -252,6 +295,9 @@ def on_start(self, event):
         else:
             event.container.create_sender(self.url, options=self.link_opts)
 
+        # Schedule a timeout (if needed)
+        self.schedule_timeout(event)
+
     def on_sendable(self, event):
         """
         called when sending can proceed
@@ -401,6 +447,9 @@ def transaction_process(self, event):
 
             self.print_message(msg)
 
+            # Schedule a timeout (if needed)
+            self.schedule_timeout(event)
+
             if self.opts.duration != 0 and self.opts.duration_mode == 'after-send':
                 utils.sleep4next(self.start_tm, self.msg_total_cnt, self.opts.duration,
                                  self.msg_processed_cnt + self.current_batch)
@@ -451,6 +500,9 @@ def on_start(self, event):
         self.sender = event.container.create_sender(conn, self.url.path, options=self.link_opts)
         event.container.declare_transaction(conn, handler=self)
 
+        # Schedule a timeout (if needed)
+        self.schedule_timeout(event)
+
     def on_transaction_declared(self, event):
         """
         called when the transaction is declared

tests/test_integration.py

@@ -529,6 +529,15 @@ def test_timeout_receiver(self):
         self.run_receiver(recv_opts)
         self.assertTrue(1.1 > time.time() - tstamp > 0.9)
 
+    def test_timeout_sender(self):
+        """ tests sender's time-out option """
+        send_opts = self.get_sender_opts()
+        send_opts.broker_url = '127.0.0.1:5673/examples'
+        send_opts.timeout = 1
+        tstamp = time.time()
+        self.run_sender(send_opts)
+        self.assertTrue(1.1 > time.time() - tstamp > 0.9)
+
     @unittest.skip("known issue #17")
     def test_duration_timeout_receiver(self):
         """ tests combining receiver's duration and timeout option """