11 files changed, 51 insertions, 0 deletions
diff --git a/src/protocol/bus0/bus.c b/src/protocol/bus0/bus.c
index 852de7c8..2427e836 100644
--- a/src/protocol/bus0/bus.c
+++ b/src/protocol/bus0/bus.c
@@ -193,6 +193,11 @@ bus0_pipe_start(void *arg)
 	bus0_pipe *p = arg;
 	bus0_sock *s = p->psock;
 
+	if (nni_pipe_peer(p->npipe) != NNI_PROTO_BUS_V0) {
+		// Peer protocol mismatch.
+		return (NNG_EPROTO);
+	}
+
 	nni_mtx_lock(&s->mtx);
 	nni_list_append(&s->pipes, p);
 	nni_mtx_unlock(&s->mtx);
diff --git a/src/protocol/pair0/pair.c b/src/protocol/pair0/pair.c
index 2fb42df5..87900793 100644
--- a/src/protocol/pair0/pair.c
+++ b/src/protocol/pair0/pair.c
@@ -129,6 +129,11 @@ pair0_pipe_start(void *arg)
 	pair0_pipe *p = arg;
 	pair0_sock *s = p->psock;
 
+	if (nni_pipe_peer(p->npipe) != NNI_PROTO_PAIR_V0) {
+		// Peer protocol mismatch.
+		return (NNG_EPROTO);
+	}
+
 	nni_mtx_lock(&s->mtx);
 	if (s->ppipe != NULL) {
 		nni_mtx_unlock(&s->mtx);
diff --git a/src/protocol/pair1/pair.c b/src/protocol/pair1/pair.c
index ab01e451..584c147d 100644
--- a/src/protocol/pair1/pair.c
+++ b/src/protocol/pair1/pair.c
@@ -173,6 +173,11 @@ pair1_pipe_start(void *arg)
 	uint32_t    id;
 	int         rv;
 
+	if (nni_pipe_peer(p->npipe) != NNI_PROTO_PAIR_V1) {
+		// Peer protocol mismatch.
+		return (NNG_EPROTO);
+	}
+
 	id = nni_pipe_id(p->npipe);
 	nni_mtx_lock(&s->mtx);
 	if ((rv = nni_idhash_insert(s->pipes, id, p)) != 0) {
diff --git a/src/protocol/pipeline0/pull.c b/src/protocol/pipeline0/pull.c
index 81f6c137..f7d5d9a2 100644
--- a/src/protocol/pipeline0/pull.c
+++ b/src/protocol/pipeline0/pull.c
@@ -112,6 +112,11 @@ pull0_pipe_start(void *arg)
 {
 	pull0_pipe *p = arg;
 
+	if (nni_pipe_peer(p->pipe) != NNI_PROTO_PUSH_V0) {
+		// Peer protocol mismatch.
+		return (NNG_EPROTO);
+	}
+
 	// Start the pending pull...
 	nni_pipe_recv(p->pipe, p->recv_aio);
 
diff --git a/src/protocol/pubsub0/sub.c b/src/protocol/pubsub0/sub.c
index c244b0ad..b4fa2e2a 100644
--- a/src/protocol/pubsub0/sub.c
+++ b/src/protocol/pubsub0/sub.c
@@ -143,6 +143,11 @@ sub0_pipe_start(void *arg)
 {
 	sub0_pipe *p = arg;
 
+	if (nni_pipe_peer(p->pipe) != NNI_PROTO_PUB_V0) {
+		// Peer protocol mismatch.
+		return (NNG_EPROTO);
+	}
+
 	nni_pipe_recv(p->pipe, p->aio_recv);
 	return (0);
 }
diff --git a/src/protocol/reqrep0/rep.c b/src/protocol/reqrep0/rep.c
index c483b777..24fc7335 100644
--- a/src/protocol/reqrep0/rep.c
+++ b/src/protocol/reqrep0/rep.c
@@ -346,6 +346,11 @@ rep0_pipe_start(void *arg)
 	rep0_sock *s = p->rep;
 	int        rv;
 
+	if (nni_pipe_peer(p->pipe) != NNI_PROTO_REQ_V0) {
+		// Peer protocol mismatch.
+		return (NNG_EPROTO);
+	}
+
 	if ((rv = nni_idhash_insert(s->pipes, nni_pipe_id(p->pipe), p)) != 0) {
 		return (rv);
 	}
diff --git a/src/protocol/reqrep0/xrep.c b/src/protocol/reqrep0/xrep.c
index 6dcfe6be..f6cd29eb 100644
--- a/src/protocol/reqrep0/xrep.c
+++ b/src/protocol/reqrep0/xrep.c
@@ -178,6 +178,11 @@ xrep0_pipe_start(void *arg)
 	xrep0_sock *s = p->rep;
 	int         rv;
 
+	if (nni_pipe_peer(p->pipe) != NNI_PROTO_REQ_V0) {
+		// Peer protocol mismatch.
+		return (NNG_EPROTO);
+	}
+
 	if ((rv = nni_idhash_insert(s->pipes, nni_pipe_id(p->pipe), p)) != 0) {
 		return (rv);
 	}
diff --git a/src/protocol/survey0/respond.c b/src/protocol/survey0/respond.c
index 7738a8b7..5f833ca6 100644
--- a/src/protocol/survey0/respond.c
+++ b/src/protocol/survey0/respond.c
@@ -341,6 +341,10 @@ resp0_pipe_start(void *arg)
 	resp0_sock *s = p->psock;
 	int         rv;
 
+	if (nni_pipe_peer(p->npipe) != NNI_PROTO_SURVEYOR_V0) {
+		return (NNG_EPROTO);
+	}
+
 	nni_mtx_lock(&s->mtx);
 	rv = nni_idhash_insert(s->pipes, p->id, p);
 	nni_mtx_unlock(&s->mtx);
diff --git a/src/protocol/survey0/survey.c b/src/protocol/survey0/survey.c
index 51bce0c8..24a35003 100644
--- a/src/protocol/survey0/survey.c
+++ b/src/protocol/survey0/survey.c
@@ -338,6 +338,10 @@ surv0_pipe_start(void *arg)
 	surv0_pipe *p = arg;
 	surv0_sock *s = p->sock;
 
+	if (nni_pipe_peer(p->npipe) != NNI_PROTO_RESPONDENT_V0) {
+		return (NNG_EPROTO);
+	}
+
 	nni_mtx_lock(&s->mtx);
 	nni_list_append(&s->pipes, p);
 	nni_mtx_unlock(&s->mtx);
diff --git a/src/protocol/survey0/xrespond.c b/src/protocol/survey0/xrespond.c
index bcbbcbc7..13a3d759 100644
--- a/src/protocol/survey0/xrespond.c
+++ b/src/protocol/survey0/xrespond.c
@@ -164,6 +164,10 @@ xresp0_pipe_start(void *arg)
 	xresp0_sock *s = p->psock;
 	int          rv;
 
+	if (nni_pipe_peer(p->npipe) != NNI_PROTO_SURVEYOR_V0) {
+		return (NNG_EPROTO);
+	}
+
 	p->id = nni_pipe_id(p->npipe);
 
 	nni_mtx_lock(&s->mtx);
diff --git a/src/protocol/survey0/xsurvey.c b/src/protocol/survey0/xsurvey.c
index 47ebef3c..db21e688 100644
--- a/src/protocol/survey0/xsurvey.c
+++ b/src/protocol/survey0/xsurvey.c
@@ -166,6 +166,10 @@ xsurv0_pipe_start(void *arg)
 	xsurv0_pipe *p = arg;
 	xsurv0_sock *s = p->psock;
 
+	if (nni_pipe_peer(p->npipe) != NNI_PROTO_RESPONDENT_V0) {
+		return (NNG_EPROTO);
+	}
+
 	nni_mtx_lock(&s->mtx);
 	nni_list_append(&s->pipes, p);
 	nni_mtx_unlock(&s->mtx);