Port: port improvements

- avoid exception when sending message on closed port (closes #307)
- port.msg()/msg_send() now return False if message cannot be sent
- publicize task.remove_port() for coherency
- add test

Change-Id: I6431cb3869acf9fa39b611593f79ef5ff28c6a8e

Author: thiell

lib/ClusterShell/Task.py

@@ -655,8 +655,8 @@ def _add_port(self, port):
         self._engine.add(port)
 
     @tasksyncmethod()
-    def _remove_port(self, port):
-        """Remove a port from Engine (private method)."""
+    def remove_port(self, port):
+        """Close and remove a port from task previously created with port()."""
         self._engine.remove(port)
 
     def port(self, handler=None, autoclose=False):

lib/ClusterShell/Worker/EngineClient.py

@@ -478,12 +478,11 @@ def __repr__(self):
                                                     id(self), fd_in, fd_out)
 
     def _start(self):
+        """Start port."""
         return self
 
     def _close(self, abort, timeout):
-        """
-        Close port pipes.
-        """
+        """Close port."""
         if not self._msgq.empty():
             # purge msgq
             try:
@@ -513,7 +512,15 @@ def msg(self, send_msg, send_once=False):
         """
         Port message send method that will wait for acknowledgement
         unless the send_once parameter if set.
+
+        May be called from another thread. Will generate ev_msg() on
+        Port event handler (in Port task/thread).
+
+        Return False if the message cannot be sent (eg. port closed).
         """
+        if self._msgq is None: # called after port closed?
+            return False
+
         pmsg = EnginePort._Msg(send_msg, not send_once)
         self._msgq.put(pmsg, block=True, timeout=None)
         try:
@@ -525,6 +532,8 @@ def msg(self, send_msg, send_once=False):
 
     def msg_send(self, send_msg):
         """
-        Port message send-once method (no acknowledgement).
+        Port message send-once method (no acknowledgement). See msg().
+
+        Return False if the message cannot be sent (eg. port closed).
         """
-        self.msg(send_msg, send_once=True)
+        return self.msg(send_msg, send_once=True)

tests/TaskPortTest.py

@@ -8,6 +8,7 @@
 import pickle
 import sys
 import threading
+import time
 import unittest
 
 sys.path.insert(0, '../lib')
@@ -46,15 +47,39 @@ def ev_msg(self, port, msg):
         self.assert_(TaskPortTest.got_msg)
 
     def testPortRemove(self):
-        """test port remove [private as of 1.2]"""
-
-        task = Task()
+        """test remove_port()"""
 
         class PortHandler(EventHandler):
             def ev_msg(self, port, msg):
                 pass
 
+        task = Task() # new thread
         port = task.port(handler=PortHandler(), autoclose=True)
         task.resume()
-        task._remove_port(port)
+        task.remove_port(port)
         task_wait()
+
+    def testPortClosed(self):
+        """test port msg on closed port"""
+        # test sending message to "stillborn" port
+        self.port_msg_result = None
+
+        # thread will wait a bit and send a port message
+        def test_thread_start(port, test):
+            time.sleep(0.5)
+            test.port_msg_result = port.msg('foobar')
+
+        class TestHandler(EventHandler):
+            pass
+
+        task = task_self()
+        test_handler = TestHandler()
+        task.timer(0.2, handler=test_handler, autoclose=False)
+        port = task.port(handler=test_handler, autoclose=True)
+        thread = threading.Thread(None, test_thread_start, args=(port, self))
+        thread.setDaemon(True)
+        thread.start()
+        task.resume()
+        task.abort(kill=True) # will remove_port()
+        thread.join()
+        self.assertEqual(self.port_msg_result, False) # test vs. None and True