Add offline buffering test, and reconnect function

Makefile

@@ -96,7 +96,7 @@ SYNC_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_C}}
 TEST_FILES_CS = test3
 SYNC_SSL_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_CS}}
 
-TEST_FILES_A = test4 test_mqtt4async
+TEST_FILES_A = test4 test9 test_mqtt4async
 ASYNC_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_A}}
 
 TEST_FILES_AS = test5

build.xml

@@ -82,6 +82,10 @@
   </target>
 	
   <target name="test" >
+    <exec executable="python3" dir="test" spawn="true"> 
+        <arg value="mqttsas.py" />
+        <arg value="${test.hostname}" />
+    </exec>
     <if>
       <os family="windows"/>
     <then>
@@ -89,7 +93,7 @@
     	<foreach target="runAtest" param="aTest" list="test1,test4"/>
 	  </then>
     <else>
-	    <foreach target="runAtest" param="aTest" list="test1,test2,test4"/>
+	    <foreach target="runAtest" param="aTest" list="test9,test1,test2,test4"/>
     </else>
     </if>
     <foreach target="runSSLtest" param="aTest" list="test3,test5"/>

src/MQTTAsync.c

@@ -858,7 +858,9 @@ int MQTTAsync_reconnect(MQTTAsync handle)
 	FUNC_ENTRY;
 	MQTTAsync_lock_mutex(mqttasync_mutex);
 
-	if (m->automaticReconnect && m->shouldBeConnected)
+	if (m->automaticReconnect) 
+	{
+    if (m->shouldBeConnected)
     {
 		  m->reconnectNow = 1;
 	  	if (m->retrying == 0)
@@ -868,6 +870,20 @@ int MQTTAsync_reconnect(MQTTAsync handle)
 	  	}
 	  	rc = MQTTASYNC_SUCCESS;
 		}
+	}
+	else
+	{
+    /* to reconnect, put the connect command to the head of the command queue */
+		MQTTAsync_queuedCommand* conn = malloc(sizeof(MQTTAsync_queuedCommand));
+		memset(conn, '\0', sizeof(MQTTAsync_queuedCommand));
+		conn->client = m;
+		conn->command = m->connect;
+		/* make sure that the version attempts are restarted */
+		if (m->c->MQTTVersion == MQTTVERSION_DEFAULT) 
+  		conn->command.details.conn.MQTTVersion = 0;
+		MQTTAsync_addCommand(conn, sizeof(m->connect));
+  	rc = MQTTASYNC_SUCCESS;
+	}
 
 	MQTTAsync_unlock_mutex(mqttasync_mutex);
 	FUNC_EXIT_RC(rc);
@@ -1367,11 +1383,14 @@ void MQTTAsync_checkTimeouts()
 		{
 			if (m->reconnectNow || MQTTAsync_elapsed(m->lastConnectionFailedTime) > (m->currentInterval * 1000))
 			{
-				/* put the connect command to the head of the command queue, using the next serverURI */
+				/* to reconnect put the connect command to the head of the command queue */
 				MQTTAsync_queuedCommand* conn = malloc(sizeof(MQTTAsync_queuedCommand));
 				memset(conn, '\0', sizeof(MQTTAsync_queuedCommand));
 				conn->client = m;
 				conn->command = m->connect;
+  			/* make sure that the version attempts are restarted */
+    		if (m->c->MQTTVersion == MQTTVERSION_DEFAULT) 
+      		conn->command.details.conn.MQTTVersion = 0;
 				Log(TRACE_MIN, -1, "Automatically attempting to reconnect");
 				MQTTAsync_addCommand(conn, sizeof(m->connect));
 				m->reconnectNow = 0;

src/MQTTAsync.h

@@ -465,7 +465,7 @@ DLLExport int MQTTAsync_setCallbacks(MQTTAsync handle, void* context, MQTTAsync_
 DLLExport int MQTTAsync_setConnected(MQTTAsync handle, void* context, MQTTAsync_connected* co);
 
 
-
+DLLExport int MQTTAsync_reconnect(MQTTAsync handle);
 		
 
 /**
@@ -755,8 +755,8 @@ typedef struct
 } MQTTAsync_connectOptions;
 
 
-#define MQTTAsync_connectOptions_initializer { {'M', 'Q', 'T', 'C'}, 4, 60, 1, 10, NULL, NULL, NULL, 30, 0, NULL, NULL, NULL, NULL, 0, NULL, 0, \
-	0, 1, 60}
+#define MQTTAsync_connectOptions_initializer { {'M', 'Q', 'T', 'C'}, 4, 60, 1, 10, NULL, NULL, NULL, 30, 0,\
+NULL, NULL, NULL, NULL, 0, NULL, 0, 0, 1, 60}
 
 /**
   * This function attempts to connect a previously-created client (see

test/mqttsas.py

+"""
+*******************************************************************
+  Copyright (c) 2013, 2016 IBM Corp.
+ 
+  All rights reserved. This program and the accompanying materials
+  are made available under the terms of the Eclipse Public License v1.0
+  and Eclipse Distribution License v1.0 which accompany this distribution. 
+ 
+  The Eclipse Public License is available at 
+     http://www.eclipse.org/legal/epl-v10.html
+  and the Eclipse Distribution License is available at 
+    http://www.eclipse.org/org/documents/edl-v10.php.
+ 
+  Contributors:
+     Ian Craggs - initial implementation and/or documentation
+*******************************************************************
+"""
+
+# Trace MQTT traffic
+import MQTTV311 as MQTTV3
+
+import socket, sys, select, traceback, datetime, os, socketserver
+
+logging = True
+myWindow = None
+
+
+def timestamp():
+  now = datetime.datetime.now()
+  return now.strftime('%Y%m%d %H%M%S')+str(float("."+str(now.microsecond)))[1:]
+
+
+class MyHandler(socketserver.StreamRequestHandler):
+
+  def handle(self):
+    if not hasattr(self, "ids"):
+      self.ids = {}
+    if not hasattr(self, "versions"):
+      self.versions = {}
+    inbuf = True
+    i = o = e = None
+    try:
+      clients = self.request
+      brokers = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+      brokers.connect((brokerhost, brokerport))
+      while inbuf != None:
+        (i, o, e) = select.select([clients, brokers], [], [])
+        for s in i:
+          if s == clients:
+            inbuf = MQTTV3.getPacket(clients) # get one packet
+            if inbuf == None:
+              break
+            try:
+              packet = MQTTV3.unpackPacket(inbuf)
+              if packet.fh.MessageType == MQTTV3.PUBLISH and \
+                  packet.topicName == "MQTTSAS topic" and \
+                  packet.data == b"TERMINATE":
+                print("Terminating client", self.ids[id(clients)])
+                brokers.close()
+                clients.close()
+                break
+              elif packet.fh.MessageType == MQTTV3.CONNECT:
+                self.ids[id(clients)] = packet.ClientIdentifier
+                self.versions[id(clients)] = 3
+              print(timestamp() , "C to S", self.ids[id(clients)], repr(packet))
+              print([hex(b) for b in inbuf])
+              print(inbuf)
+            except:
+              traceback.print_exc()
+            brokers.send(inbuf)       # pass it on
+          elif s == brokers:
+            inbuf = MQTTV3.getPacket(brokers) # get one packet
+            if inbuf == None:
+              break
+            try:
+              print(timestamp(), "S to C", self.ids[id(clients)], repr(MQTTV3.unpackPacket(inbuf)))
+            except:
+              traceback.print_exc()
+            clients.send(inbuf)
+      print(timestamp()+" client "+self.ids[id(clients)]+" connection closing")
+    except:
+      print(repr((i, o, e)), repr(inbuf))
+      traceback.print_exc()
+    if id(clients) in self.ids.keys():
+      del self.ids[id(clients)]
+    elif id(clients) in self.versions.keys():
+      del self.versions[id(clients)]
+
+class ThreadingTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
+  pass
+
+def run():
+  global brokerhost, brokerport
+  myhost = 'localhost'
+  if len(sys.argv) > 1:
+    brokerhost = sys.argv[1]
+  else:
+    brokerhost = 'localhost'
+
+  if len(sys.argv) > 2:
+    brokerport = int(sys.argv[2])
+  else:
+    brokerport = 1883
+
+  if brokerhost == myhost:
+    myport = brokerport + 1
+  else:
+    myport = 1883
+
+  print("Listening on port", str(myport)+", broker on port", brokerport)
+  s = ThreadingTCPServer(("", myport), MyHandler)
+  s.serve_forever()
+
+if __name__ == "__main__":
+  run()

test/test2.c

 /*******************************************************************************
- * Copyright (c) 2009, 2015 IBM Corp.
+ * Copyright (c) 2009, 2016 IBM Corp.
  *
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
@@ -12,6 +12,7 @@
  *
  * Contributors:
  *    Ian Craggs - initial API and implementation and/or initial documentation
+ *    Ian Craggs - fix thread id display
  *******************************************************************************/
 
 /**
@@ -360,7 +361,7 @@ thread_return_type WINAPI test1_sendAndReceive(void* n)
 	rc = MQTTClient_subscribe(c, test_topic, subsqos);
 	assert("Good rc from subscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
 
-	MyLog(LOGA_INFO, "Thread %d, %d messages at QoS %d", Thread_getid(), iterations, qos);
+	MyLog(LOGA_INFO, "Thread %u, %d messages at QoS %d", Thread_getid(), iterations, qos);
 	test1_pubmsg.payload = test1_pubmsg_check.payload;
 	test1_pubmsg.payloadlen = test1_pubmsg_check.payloadlen;
 	test1_pubmsg.retained = 0;