fixes #449 Want more flexible pipe events

This changes the signature of nng_pipe_notify(), and the associated
events.  The documentation is updated to reflect this.

We have also broken the lock up so that we don't hold the master
socket lock for some of these things, which may have beneficial
impact on performance.

4 files changed, 16 insertions, 0 deletions

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);