Add test changes

test/test6.c

@@ -58,6 +58,7 @@ struct
 	char* password;
 	int verbose;
 	int persistence;
+	int payload_len;
 } opts =
 {
 	"tcp://localhost:1884",
@@ -68,12 +69,13 @@ struct
 	"Eclipse/Paho/restart_test/control",
 	"C_broken_client",
 	1,    /* slot_no */
-	2,    /* QoS */
+	0,    /* QoS */
 	0,    /* retained */
 	NULL,
 	NULL,
 	0,
 	1,
+	1000
 };
 
 void getopts(int argc, char** argv)
@@ -234,6 +236,40 @@ START_TIME_TYPE start_clock(void)
 }
 #endif
 
+int tests = 0;
+int failures = 0;
+FILE* xml;
+START_TIME_TYPE global_start_time;
+char output[3000];
+char* cur_output = output;
+
+#define assert(a, b, c, d) myassert(__FILE__, __LINE__, a, b, c, d)
+#define assert1(a, b, c, d, e) myassert(__FILE__, __LINE__, a, b, c, d, e)
+
+void myassert(char* filename, int lineno, char* description, int value,
+		char* format, ...)
+{
+	++tests;
+	if (!value)
+	{
+		va_list args;
+
+		++failures;
+		MyLog(LOGA_INFO, "Assertion failed, file %s, line %d, description: %s", filename,
+				lineno, description);
+
+		va_start(args, format);
+		vprintf(format, args);
+		va_end(args);
+
+		cur_output += sprintf(cur_output, "<failure type=\"%s\">file %s, line %d </failure>\n",
+                        description, filename, lineno);
+	}
+	else
+		MyLog(LOGA_DEBUG, "Assertion succeeded, file %s, line %d, description: %s",
+				filename, lineno, description);
+}
+
 
 #if defined(WIN32)
 long elapsed(START_TIME_TYPE start_time)
@@ -616,6 +652,7 @@ void one_iteration(void)
 	START_TIME_TYPE start_time;
 	int last_expected_count = expectedCount;
 	int test_interval = 30;
+	char* payload = malloc(opts.payload_len);
 
 	if (control_wait("start_measuring") == 0)
 		goto exit;
@@ -630,11 +667,9 @@ void one_iteration(void)
 	global_start_time = start_clock();
 	for (i = 1; i <= test_count; ++i)
 	{
-		char payload[128];
-
 		sprintf(payload, "message number %d", i);
 
-		rc = MQTTAsync_send(client, opts.topic, (int)(strlen(payload)+1), payload,
+		rc = MQTTAsync_send(client, opts.topic, opts.payload_len, payload,
 				        opts.qos, opts.retained, NULL);
 		while (rc != MQTTASYNC_SUCCESS)
 		{
@@ -687,6 +722,7 @@ void one_iteration(void)
 		sprintf(payload, "message number %d", seqno);
 		rc = MQTTAsync_send(client, opts.topic, (int)(strlen(payload)+1), payload,
 				opts.qos, opts.retained, &ropts);
+		assert("Good rc from send", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc);
 		while (rc != MQTTASYNC_SUCCESS)
 		{
 			MyLog(LOGA_DEBUG, "Rc %d from publish with payload %s, retrying", rc, payload);
@@ -697,6 +733,7 @@ void one_iteration(void)
 			mysleep(1);
 			rc = MQTTAsync_send(client, opts.topic, (int)(strlen(payload)+1), payload,
 				opts.qos, opts.retained, &ropts);
+			assert("Good rc from send", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc);
 		}
 		//MyLog(LOGA_DEBUG, "Successful publish with payload %s", payload);
 		while (seqno - messagesSent > 2000)
@@ -807,6 +844,7 @@ int sendAndReceive(void)
 
 	/* connect cleansession, and then disconnect, to clean up */
 	conn_opts.keepAliveInterval = 10;
+	conn_opts.maxInflight = test_count;
 	conn_opts.username = opts.username;
 	conn_opts.password = opts.password;
 	conn_opts.cleansession = 1;

test/test9.c

@@ -1860,7 +1860,7 @@ void test7cConnected(void* context, char* cause)
 
 void test7cOnConnectFailure(void* context, MQTTAsync_failureData* response)
 {
-	MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context);
+	MyLog(LOGA_DEBUG, "In c connect onFailure callback, context %p", context);
 
 	test7OnFailureCalled++;
 	test7Finished = 1;
@@ -1990,7 +1990,11 @@ int test7(struct Options options)
 	/* wait until d is ready: connected and subscribed */
 	count = 0;
 	while (!test7dReady && ++count < 10000)
+	{
+		if (test7Finished)
+		  goto exit;
 		MySleep(100);
+	}
 	assert("Count should be less than 10000", count < 10000, "count was %d", count); /* wrong */
 
 	rc = MQTTAsync_setConnected(c, c, test7cConnected);
@@ -2007,6 +2011,9 @@ int test7(struct Options options)
 	opts.onFailure = test7cOnConnectFailure;
 	opts.context = c;
 	opts.cleansession = 0;
+	/*opts.automaticReconnect = 1;
+	opts.minRetryInterval = 3;
+	opts.maxRetryInterval = 6;*/
 
 	MyLog(LOGA_DEBUG, "Connecting client c");
 	rc = MQTTAsync_connect(c, &opts);
@@ -2029,21 +2036,26 @@ int test7(struct Options options)
 	MyLog(LOGA_DEBUG, "Now we can send some messages to be buffered by TCP");
 
 	test7c_connected = 0;
-	char buf[20000];
+	char buf[5000000];
 	/* send some messages.  Then reconnect (check connected callback), and check that those messages are received */
 	for (i = 0; i < 50000; ++i)
 	{
 	  MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
-	  MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
+	  MQTTAsync_responseOptions pubopts = MQTTAsync_responseOptions_initializer;
 		pubmsg.qos = 0; /*i % 3;*/
 	  sprintf(buf, "QoS %d message", pubmsg.qos);
 	  pubmsg.payload = buf;
-	  pubmsg.payloadlen = 20000; //(int)(strlen(pubmsg.payload) + 1);
+	  pubmsg.payloadlen = 5000000; //(int)(strlen(pubmsg.payload) + 1);
 	  pubmsg.retained = 0;
-	  rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &opts);
+	  rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &pubopts);
 	  assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc);
 		if (rc != 0)
+		{
+		  //MyLog(LOGA_DEBUG, "Connecting client c");
+		  //rc = MQTTAsync_connect(c, &opts);
+			//MySleep(1000);
 		  break;
+		}
 	}
 
 #if 0
@@ -2081,13 +2093,13 @@ int test7(struct Options options)
  	assert("Number of getPendingTokens should be 0", i == 0, "i was %d ", i);
 #endif
 
+exit:
 	rc = MQTTAsync_disconnect(c, NULL);
  	assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc);
 
 	rc = MQTTAsync_disconnect(d, NULL);
  	assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc);
 
-exit:
 	MySleep(200);
 	MQTTAsync_destroy(&c);
 	MQTTAsync_destroy(&d);