Merge branch 'test4issue373' into 373

* test4issue373:
  Extend test_issue373 to test also with QoS 1 and 2
  Add helper script for test_issue373 to cyclically disconnect the VM running the broker
  Add log output for failed message queueing in test due to disconnect
  Do not reduce the tcp send buffer size for production code

# Conflicts:
#	test/test_issue373.c
Signed-off-by: Juergen Kosel <juergen.kosel@softing.com>

src/Socket.c

@@ -713,7 +713,7 @@ int Socket_new(char* addr, int port, int* sock)
 			if (setsockopt(*sock, SOL_SOCKET, SO_NOSIGPIPE, (void*)&opt, sizeof(opt)) != 0)
 				Log(LOG_ERROR, -1, "Could not set SO_NOSIGPIPE for socket %d", *sock);
 #endif
-#if 1
+#if defined(TESTING)
                         {
                             int optsend = 2 * 1440;
                             if (setsockopt(*sock, SOL_SOCKET, SO_SNDBUF, (void*)&optsend, sizeof(optsend)) != 0)

test/python/controlVMnetworkstate.py

+#!/usr/bin/python
+import os
+import sys
+import time
+import subprocess
+import random
+bindir='/usr/bin'
+sys.path.append(bindir)
+
+def input_sel(prompt,max_,selectionoption):
+    # let the user choose the VM and verify selection
+    while True:
+        try:
+            print ('Please select from the list of running VMs\n\n'+'\n'.join(selectionoption))
+            userin = int(raw_input(prompt))
+        except ValueError:
+            print('\nThat was not a number\n\n')
+            continue
+        if userin > max_:
+            print('\nInput must be less than or equal to {0}.\n\n'.format(max_))
+        elif userin < 1:
+            print('\nInput must be greater than or equal to 1\n\n')
+        else:
+            return userin
+
+def statustext(result):
+    if result == 0:
+        status = 'OK'
+    else:
+        status = 'Failed'
+    return status
+
+def controlvmnetworkstate():
+    try:
+        offtime = 600
+        ontime = 14
+        vmdict={}
+        vmlist=[]
+        executable = os.path.join(bindir, 'VBoxManage')
+
+        #retrieve a list of all running VMs
+        runningvms= subprocess.check_output('%s list runningvms' %executable,shell=True).splitlines()
+        if len(runningvms) != 0:
+            for n in range(0, len(runningvms)):
+                vmlist.append('%s: %s' %(n+1,runningvms[n].rsplit(' ',1)[0].strip('"')))
+                vmdict[n+1]=runningvms[n].rsplit(' ',1)[-1]
+            usersel=input_sel('\nEnter the number of the VM: ',len(runningvms),vmlist)
+
+        else:
+            print('Can not retrieve list of running VMs')
+            sys.exit()
+
+        vmuuid=vmdict[usersel]
+        while True:
+            offtime = random.randint(60, 90)
+            ontime = random.randint(10, 90)
+            timenow = time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime())
+            on = subprocess.call('%s controlvm %s setlinkstate1 on' %(executable,vmuuid),
+                                 shell=True)
+            status=statustext(on)
+            print ('%s: Plug Network cable into VM %s for %ds: %s' % (timenow, runningvms[usersel-1].rsplit(' ',1)[0].strip('"'),ontime, str(status)))
+            time.sleep(ontime)
+            timenow = time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime())
+            off = subprocess.call('%s controlvm %s setlinkstate1 off' %(executable,vmuuid),
+                                  shell=True)
+            status = statustext(off)
+            print ('%s: Unplug Network cable from VM %s for %ds: %s' % (timenow, runningvms[usersel-1].rsplit(' ',1)[0].strip('"'),offtime, str(status)))
+            time.sleep(offtime)
+    except KeyboardInterrupt:
+        sys.exit('\nUser Interrupt')
+    except Exception as e:
+        print("Error in %s in function %s: %s" % (__name__, sys._getframe().f_code.co_name, e.message))
+if __name__ == "__main__":
+    sys.exit(controlvmnetworkstate())

test/test_issue373.c

@@ -54,12 +54,16 @@ struct Options
 	char* proxy_connection;      /**< connection to proxy */
 	int verbose;
 	int test_no;
+	unsigned int QoS;
+	unsigned int iterrations;
 } options =
 {
 	"iot.eclipse.org:1883",
 	"localhost:1883",
 	0,
 	0,
+	0,
+	5
 };
 
 void getopts(int argc, char** argv)
@@ -227,7 +231,7 @@ int test373_messageArrived(void* context, char* topicName, int topicLen, MQTTAsy
 
 static char test373Payload[] = "No one is interested in this payload";
 
-int test373SendPublishMessage(MQTTAsync handle,int id)
+int test373SendPublishMessage(MQTTAsync handle,int id, const unsigned int QoS)
 {
 	int rc = 0;
 	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
@@ -240,7 +244,7 @@ int test373SendPublishMessage(MQTTAsync handle,int id)
 
 	pubmsg.payload = test373Payload;
 	pubmsg.payloadlen = sizeof(test373Payload);
-	pubmsg.qos = 0;
+	pubmsg.qos = QoS;
 	rc = MQTTAsync_sendMessage( handle, topic,&pubmsg,&opts);
 	if (rc == MQTTASYNC_SUCCESS)
 	{
@@ -264,7 +268,9 @@ int test_373(struct Options options)
 	char clientid[30 + sizeof(unique)];
 	heap_info* mqtt_mem = 0;
 
+	MyLog(LOGA_INFO, "Running test373 with QoS=%u, iterrations=%u\n",options.QoS,options.iterrations);
 	sprintf(clientid, "paho-test373-%s", unique);
+	connectCnt = 0;
 	rc = MQTTAsync_create(&mqttasyncContext, options.proxy_connection, clientid,
 			      MQTTCLIENT_PERSISTENCE_NONE,
 			      NULL);
@@ -290,7 +296,7 @@ int test_373(struct Options options)
 		goto exit;
 	}
 	MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR);
-	while (connectCnt < 40)
+	while (connectCnt < options.iterrations)
 	{
 		if (!connected)
 		{
@@ -323,7 +329,7 @@ int test_373(struct Options options)
 			int topicId;
 			for(topicId=0; topicId < 100; topicId++)
 			{
-				rc = test373SendPublishMessage(mqttasyncContext,topicId);
+				rc = test373SendPublishMessage(mqttasyncContext,topicId,options.QoS);
 				if (rc != MQTTASYNC_SUCCESS) break;
 			}
 			MySleep(100);
@@ -337,6 +343,7 @@ int test_373(struct Options options)
 	MyLog(LOGA_INFO, "MQTT mem current %ld, max %ld",mqtt_mem->current_size,mqtt_mem->max_size);
 #endif
 	MQTTAsync_disconnect(mqttasyncContext, NULL);
+	connected = 0;
 	MyLog(LOGA_INFO, "PublishCnt %d, FailedCnt %d, Pending %d maxPending %d",
 	      goodPublishCnt,failedPublishCnt,pendingMessageCnt,pendingMessageCntMax);
 #if !defined(_WINDOWS)
@@ -363,6 +370,7 @@ int main(int argc, char** argv)
 	int* numtests = &tests;
 	int rc = 0;
 	int (*tests[])() = { NULL, test_373};
+	unsigned int QoS;
 
 	sprintf(unique, "%u", rand());
 	MyLog(LOGA_INFO, "Random prefix/suffix is %s", unique);
@@ -374,9 +382,24 @@ int main(int argc, char** argv)
 	{ /* run all the tests */
 		for (options.test_no = 1; options.test_no < ARRAY_SIZE(tests); ++options.test_no)
 		{
-			failures = 0;
-			MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR);
-			rc += tests[options.test_no](options); /* return number of failures.  0 = test succeeded */
+			/* test with QoS 0, 1 and 2 and just 5 iterrations */
+			for (QoS = 0; QoS < 3; QoS++)
+			{
+				failures = 0;
+				options.QoS = QoS;
+				options.iterrations = 5;
+				MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR);
+				rc += tests[options.test_no](options); /* return number of failures.  0 = test succeeded */
+			}
+			if (rc == 0)
+			{
+				/* Test with much more iterrations for QoS = 0 */
+				failures = 0;
+				options.QoS = 0;
+				options.iterrations = 100;
+				MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR);
+				rc += tests[options.test_no](options); /* return number of failures.  0 = test succeeded */
+			}
 		}
 	}
 	else