COAP-18. Cancelling all threads and coap state machine when close is called.

coap/coap.py

@@ -38,6 +38,7 @@ def __init__(self,ipAddress='',udpPort=d.DEFAULT_UDP_PORT,testing=False):
         self.transmitters         = {}
         self.ackTimeout           = d.DFLT_ACK_TIMEOUT
         self.respTimeout          = d.DFLT_RESPONSE_TIMEOUT
+        self.coapTrans            = None
         if testing:
             self.socketUdp        = socketUdpDispatcher(
                 ipAddress         = self.ipAddress,
@@ -55,6 +56,8 @@ def __init__(self,ipAddress='',udpPort=d.DEFAULT_UDP_PORT,testing=False):
 
     def close(self):
         self.socketUdp.close()
+        if self.coapTrans:
+            self.coapTrans.close()
 
     #===== client
 
@@ -121,7 +124,7 @@ def _transmit(self,uri,confirmable,code,options=[],payload=[]):
             self._cleanupTransmitter()
             messageId        = self._getMessageID(destIp,destPort)
             token            = self._getToken(destIp,destPort)
-            newTransmitter   = coapTransmitter.coapTransmitter(
+            self.coapTrans   = coapTransmitter.coapTransmitter(
                 sendFunc     = self.socketUdp.sendUdp,
                 srcIp        = self.ipAddress,
                 srcPort      = self.udpPort,
@@ -138,9 +141,9 @@ def _transmit(self,uri,confirmable,code,options=[],payload=[]):
             )
             key              = (destIp,destPort,token,messageId)
             assert key not in self.transmitters.keys()
-            self.transmitters[key] = newTransmitter
+            self.transmitters[key] = self.coapTrans
 
-        return newTransmitter.transmit()
+        return self.coapTrans.transmit()
 
     def _getMessageID(self,destIp,destPort):
         '''

coap/coapTransmitter.py

@@ -211,6 +211,10 @@ def receiveMessage(self, timestamp, srcIp, srcPort, message):
         # signal reception
         self.rxMsgEvent.set()
 
+    def close(self):
+        self.coapError = e.coapException("Request cancelled")
+        self.fsmGoOn = False
+        self.rxMsgEvent.set()
     #======================= private ==========================================
 
     #===== fsm
@@ -339,6 +343,8 @@ def _action_WAITFORACK(self):
         while True:
             waitTimeLeft = startTime+ackMaxWait-time.time()
             if self.rxMsgEvent.wait(timeout=waitTimeLeft):
+                if not self.fsmGoOn:
+                    return
                 # I got message
                 with self.dataLock:
                     (timestamp,srcIp,srcPort,message) = self.LastRxPacket

coap/socketUdpReal.py

@@ -8,7 +8,7 @@ def emit(self, record):
 
 import socket
 import time
-
+import select
 import socketUdp
 import threading
 
@@ -59,16 +59,16 @@ def sendUdp(self,destIp,destPort,msg):
     def close(self):
         # declare that this thread has to stop
         self.goOn = False
-
-        # send some dummy value into the socket to trigger a read
-        self.socket_handler.sendto( 'stop', ('::1',self.udpPort) )
-
+        self.socket_handler.close()
     #======================== private =========================================
 
     def run(self):
         while self.goOn:
             try:
-                # blocking wait for something from UDP socket
+                # Blocking wait for something from UDP socket.
+                # A simple recvfrom() call does not returns when the socket is closed
+                # so we are using select() for detecting this condition.
+                i,o,e = select.select([self.socket_handler],[],[])
                 raw,conn = self.socket_handler.recvfrom(self.BUFSIZE)
             except socket.error as err:
                 log.critical("socket error: {0}".format(err))