Make sure writeComplete deals with socket error #373

src/MQTTAsync.c

@@ -361,7 +361,7 @@ static void MQTTProtocol_checkPendingWrites(void);
 static void MQTTAsync_freeServerURIs(MQTTAsyncs* m);
 static void MQTTAsync_freeCommand1(MQTTAsync_queuedCommand *command);
 static void MQTTAsync_freeCommand(MQTTAsync_queuedCommand *command);
-static void MQTTAsync_writeComplete(int socket);
+static void MQTTAsync_writeComplete(int socket, int rc);
 static int MQTTAsync_processCommand(void);
 static void MQTTAsync_checkTimeouts(void);
 static thread_return_type WINAPI MQTTAsync_sendThread(void* n);
@@ -1050,7 +1050,7 @@ static void MQTTAsync_freeCommand(MQTTAsync_queuedCommand *command)
 }
 
 
-static void MQTTAsync_writeComplete(int socket)
+static void MQTTAsync_writeComplete(int socket, int rc)
 {
 	ListElement* found = NULL;
 
@@ -1082,19 +1082,32 @@ static void MQTTAsync_writeComplete(int socket)
 
 			if (cur_response) /* we found a response */
 			{
-				if (command->type == PUBLISH && command->onSuccess)
+				if (command->type == PUBLISH)
 				{
-				  MQTTAsync_successData data;
-
-				  data.token = command->token;
-				  data.alt.pub.destinationName = command->details.pub.destinationName;
-				  data.alt.pub.message.payload = command->details.pub.payload;
-				  data.alt.pub.message.payloadlen = command->details.pub.payloadlen;
-				  data.alt.pub.message.qos = command->details.pub.qos;
-				  data.alt.pub.message.retained = command->details.pub.retained;
-				  Log(TRACE_MIN, -1, "Calling publish success for client %s", m->c->clientID);
-				  (*(command->onSuccess))(command->context, &data);
-			  }
+					if (rc == 1 && command->onSuccess)
+					{
+						MQTTAsync_successData data;
+
+						data.token = command->token;
+						data.alt.pub.destinationName = command->details.pub.destinationName;
+						data.alt.pub.message.payload = command->details.pub.payload;
+						data.alt.pub.message.payloadlen = command->details.pub.payloadlen;
+						data.alt.pub.message.qos = command->details.pub.qos;
+						data.alt.pub.message.retained = command->details.pub.retained;
+						Log(TRACE_MIN, -1, "Calling publish success for client %s", m->c->clientID);
+						(*(command->onSuccess))(command->context, &data);
+					}
+					else if (rc == -1 && command->onFailure)
+					{
+						MQTTAsync_failureData data;
+
+						data.token = command->token;
+						data.code = rc;
+						data.message = NULL;
+						Log(TRACE_MIN, -1, "Calling publish failure for client %s", m->c->clientID);
+						(*(command->onFailure))(command->context, &data);
+					}
+				}
 				if (com)
 				{
 				  ListDetach(m->responses, com);

src/MQTTClient.c

@@ -295,7 +295,7 @@ static MQTTPacket* MQTTClient_cycle(int* sock, unsigned long timeout, int* rc);
 static MQTTPacket* MQTTClient_waitfor(MQTTClient handle, int packet_type, int* rc, long timeout);
 /*static int pubCompare(void* a, void* b); */
 static void MQTTProtocol_checkPendingWrites(void);
-static void MQTTClient_writeComplete(int socket);
+static void MQTTClient_writeComplete(int socket, int rc);
 
 
 int MQTTClient_create(MQTTClient* handle, const char* serverURI, const char* clientId,
@@ -2082,7 +2082,7 @@ static void MQTTProtocol_checkPendingWrites(void)
 }
 
 
-static void MQTTClient_writeComplete(int socket)
+static void MQTTClient_writeComplete(int socket, int rc)
 {
 	ListElement* found = NULL;
 

src/Socket.c

@@ -428,10 +428,10 @@ int Socket_writev(int socket, iobuf* iovecs, int count, unsigned long* bytes)
 			rc = TCPSOCKET_INTERRUPTED;
 	}
 #else
-//#define TESTING
+#define TESTING
 #if defined(TESTING)
   static int i = 0;
-	if (++i % 100 == 1)
+	if (++i >= 10 && i < 21)
 	{
 		if (1)
 		{
@@ -444,7 +444,7 @@ int Socket_writev(int socket, iobuf* iovecs, int count, unsigned long* bytes)
 		  rc = SOCKET_ERROR;
 		}
 		/* should *bytes always be 0? */
-		if (1)
+		if (i == 20)
 		{
 		  printf("Shutdown socket\n");
 		  shutdown(socket, SHUT_WR);
@@ -756,7 +756,7 @@ void Socket_setWriteCompleteCallback(Socket_writeComplete* mywritecomplete)
 /**
  *  Continue an outstanding write for a particular socket
  *  @param socket that socket
- *  @return completion code
+ *  @return completion code: 0=incomplete, 1=complete, -1=socket error
  */
 int Socket_continueWrite(int socket)
 {
@@ -848,7 +848,7 @@ int Socket_continueWrites(fd_set* pwset)
 		int socket = *(int*)(curpending->content);
 		int rc = 0;
 
-		if (FD_ISSET(socket, pwset) && (rc = Socket_continueWrite(socket)))
+		if (FD_ISSET(socket, pwset) && ((rc = Socket_continueWrite(socket)) != 0))
 		{
 			if (!SocketBuffer_writeComplete(socket))
 				Log(LOG_SEVERE, -1, "Failed to remove pending write from socket buffer list");
@@ -860,8 +860,8 @@ int Socket_continueWrites(fd_set* pwset)
 			}
 			curpending = s.write_pending->current;
 
-			if (rc == 1 && writecomplete)
-				(*writecomplete)(socket);
+			if (writecomplete)
+				(*writecomplete)(socket, rc);
 		}
 		else
 			ListNextElement(s.write_pending, &curpending);

src/Socket.h

@@ -137,7 +137,7 @@ char* Socket_getpeer(int sock);
 void Socket_addPendingWrite(int socket);
 void Socket_clearPendingWrite(int socket);
 
-typedef void Socket_writeComplete(int socket);
+typedef void Socket_writeComplete(int socket, int rc);
 void Socket_setWriteCompleteCallback(Socket_writeComplete*);
 
 #endif /* SOCKET_H */