fixes #970 TCP connections through docker-proxy do not reconnect

6 files changed, 47 insertions, 15 deletions

diff --git a/docs/man/nng_stream_recv.3str.adoc b/docs/man/nng_stream_recv.3str.adoc
index 6b5e5b37..148be978 100644
--- a/docs/man/nng_stream_recv.3str.adoc
+++ b/docs/man/nng_stream_recv.3str.adoc
@@ -56,6 +56,7 @@ None.
 `NNG_ECANCELED`:: The operation was canceled.
 `NNG_ECLOSED`:: The connection was closed.
 `NNG_ECONNRESET`:: The peer closed the connection.
+`NNG_ECONNSHUT`:: Remote peer shutdown after sending data.
 `NNG_EINVAL`:: The _aio_ does not contain a valid scatter/gather vector.
 `NNG_ENOMEM`:: Insufficient free memory to perform the operation.
 `NNG_ETIMEDOUT`:: Timeout waiting for data from the connection.
diff --git a/include/nng/nng.h b/include/nng/nng.h
index fcc95668..7ec8c2c9 100644
--- a/include/nng/nng.h
+++ b/include/nng/nng.h
@@ -233,7 +233,8 @@ NNG_DECL int nng_socket_set_uint64(nng_socket, const char *, uint64_t);
 NNG_DECL int nng_socket_set_string(nng_socket, const char *, const char *);
 NNG_DECL int nng_socket_set_ptr(nng_socket, const char *, void *);
 NNG_DECL int nng_socket_set_ms(nng_socket, const char *, nng_duration);
-NNG_DECL int nng_socket_set_addr(nng_socket, const char *, const nng_sockaddr *);
+NNG_DECL int nng_socket_set_addr(
+    nng_socket, const char *, const nng_sockaddr *);
 
 NNG_DECL int nng_socket_get(nng_socket, const char *, void *, size_t *);
 NNG_DECL int nng_socket_get_bool(nng_socket, const char *, bool *);
@@ -349,7 +350,8 @@ NNG_DECL int nng_dialer_set_uint64(nng_dialer, const char *, uint64_t);
 NNG_DECL int nng_dialer_set_string(nng_dialer, const char *, const char *);
 NNG_DECL int nng_dialer_set_ptr(nng_dialer, const char *, void *);
 NNG_DECL int nng_dialer_set_ms(nng_dialer, const char *, nng_duration);
-NNG_DECL int nng_dialer_set_addr(nng_dialer, const char *, const nng_sockaddr *);
+NNG_DECL int nng_dialer_set_addr(
+    nng_dialer, const char *, const nng_sockaddr *);
 
 NNG_DECL int nng_dialer_get(nng_dialer, const char *, void *, size_t *);
 NNG_DECL int nng_dialer_get_bool(nng_dialer, const char *, bool *);
@@ -402,11 +404,11 @@ NNG_DECL int nng_listener_set_bool(nng_listener, const char *, bool);
 NNG_DECL int nng_listener_set_int(nng_listener, const char *, int);
 NNG_DECL int nng_listener_set_size(nng_listener, const char *, size_t);
 NNG_DECL int nng_listener_set_uint64(nng_listener, const char *, uint64_t);
-NNG_DECL int nng_listener_set_string(
-    nng_listener, const char *, const char *);
+NNG_DECL int nng_listener_set_string(nng_listener, const char *, const char *);
 NNG_DECL int nng_listener_set_ptr(nng_listener, const char *, void *);
 NNG_DECL int nng_listener_set_ms(nng_listener, const char *, nng_duration);
-NNG_DECL int nng_listener_set_addr(nng_listener, const char *, const nng_sockaddr *);
+NNG_DECL int nng_listener_set_addr(
+    nng_listener, const char *, const nng_sockaddr *);
 
 NNG_DECL int nng_listener_get(nng_listener, const char *, void *, size_t *);
 NNG_DECL int nng_listener_get_bool(nng_listener, const char *, bool *);
@@ -1080,6 +1082,7 @@ enum nng_errno_enum {
 	NNG_ENOARG       = 28,
 	NNG_EAMBIGUOUS   = 29,
 	NNG_EBADTYPE     = 30,
+	NNG_ECONNSHUT    = 31,
 	NNG_EINTERNAL    = 1000,
 	NNG_ESYSERR      = 0x10000000,
 	NNG_ETRANERR     = 0x20000000
diff --git a/src/platform/posix/posix_ipcconn.c b/src/platform/posix/posix_ipcconn.c
index 07ec6213..c9faded5 100644
--- a/src/platform/posix/posix_ipcconn.c
+++ b/src/platform/posix/posix_ipcconn.c
@@ -186,7 +186,7 @@ ipc_doread(ipc_conn *c)
 			// No bytes indicates a closed descriptor.
 			// This implicitly completes this (all!) aio.
 			nni_aio_list_remove(aio);
-			nni_aio_finish_error(aio, NNG_ECLOSED);
+			nni_aio_finish_error(aio, NNG_ECONNSHUT);
 			continue;
 		}
 
@@ -202,6 +202,22 @@ ipc_doread(ipc_conn *c)
 }
 
 static void
+ipc_error(void *arg, int err)
+{
+	ipc_conn *c = arg;
+	nni_aio *aio;
+
+	nni_mtx_lock(&c->mtx);
+	while (((aio = nni_list_first(&c->readq)) != NULL) ||
+	    ((aio = nni_list_first(&c->writeq)) != NULL)) {
+		nni_aio_list_remove(aio);
+		nni_aio_finish_error(aio, err);
+	}
+	nni_posix_pfd_close(c->pfd);
+	nni_mtx_unlock(&c->mtx);
+}
+
+static void
 ipc_close(void *arg)
 {
 	ipc_conn *c = arg;
@@ -225,7 +241,7 @@ ipc_cb(nni_posix_pfd *pfd, int events, void *arg)
 	ipc_conn *c = arg;
 
 	if (events & (POLLHUP | POLLERR | POLLNVAL)) {
-		ipc_close(c);
+		ipc_error(c, NNG_ECONNSHUT);
 		return;
 	}
 	nni_mtx_lock(&c->mtx);
diff --git a/src/platform/posix/posix_tcpconn.c b/src/platform/posix/posix_tcpconn.c
index 0d3c274d..6ca7013b 100644
--- a/src/platform/posix/posix_tcpconn.c
+++ b/src/platform/posix/posix_tcpconn.c
@@ -180,7 +180,7 @@ tcp_doread(nni_tcp_conn *c)
 			// No bytes indicates a closed descriptor.
 			// This implicitly completes this (all!) aio.
 			nni_aio_list_remove(aio);
-			nni_aio_finish_error(aio, NNG_ECLOSED);
+			nni_aio_finish_error(aio, NNG_ECONNSHUT);
 			continue;
 		}
 
@@ -196,6 +196,22 @@ tcp_doread(nni_tcp_conn *c)
 }
 
 static void
+tcp_error(void *arg, int err)
+{
+	nni_tcp_conn *c = arg;
+	nni_aio *aio;
+
+	nni_mtx_lock(&c->mtx);
+	while (((aio = nni_list_first(&c->readq)) != NULL) ||
+	    ((aio = nni_list_first(&c->writeq)) != NULL)) {
+		nni_aio_list_remove(aio);
+		nni_aio_finish_error(aio, err);
+	}
+	nni_posix_pfd_close(c->pfd);
+	nni_mtx_unlock(&c->mtx);
+}
+
+static void
 tcp_close(void *arg)
 {
 	nni_tcp_conn *c = arg;
@@ -242,7 +258,7 @@ tcp_cb(nni_posix_pfd *pfd, int events, void *arg)
 	nni_tcp_conn *c = arg;
 
 	if (events & (POLLHUP | POLLERR | POLLNVAL)) {
-		tcp_close(c);
+                tcp_error(c, NNG_ECONNSHUT);
 		return;
 	}
 	nni_mtx_lock(&c->mtx);
diff --git a/src/platform/windows/win_ipcconn.c b/src/platform/windows/win_ipcconn.c
index 4d267dd9..6aa30925 100644
--- a/src/platform/windows/win_ipcconn.c
+++ b/src/platform/windows/win_ipcconn.c
@@ -116,7 +116,7 @@ ipc_recv_cb(nni_win_io *io, int rv, size_t num)
 
 	if ((rv == 0) && (num == 0)) {
 		// A zero byte receive is a remote close from the peer.
-		rv = NNG_ECLOSED;
+		rv = NNG_ECONNSHUT;
 	}
 	nni_aio_finish_synch(aio, rv, num);
 }
@@ -240,10 +240,6 @@ ipc_send_cb(nni_win_io *io, int rv, size_t num)
 	}
 	nni_mtx_unlock(&c->mtx);
 
-	if ((rv == 0) && (num == 0)) {
-		// A zero byte receive is a remote close from the peer.
-		rv = NNG_ECLOSED;
-	}
 	nni_aio_finish_synch(aio, rv, num);
 }
 
diff --git a/src/platform/windows/win_tcpconn.c b/src/platform/windows/win_tcpconn.c
index 429c5a5d..bd464184 100644
--- a/src/platform/windows/win_tcpconn.c
+++ b/src/platform/windows/win_tcpconn.c
@@ -89,7 +89,7 @@ tcp_recv_cb(nni_win_io *io, int rv, size_t num)
 
 	if ((rv == 0) && (num == 0)) {
 		// A zero byte receive is a remote close from the peer.
-		rv = NNG_ECLOSED;
+		rv = NNG_ECONNSHUT;
 	}
 	nni_aio_finish_synch(aio, rv, num);
 }