aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
authorGarrett D'Amore <garrett@damore.org>2017-08-08 21:19:09 -0700
committerGarrett D'Amore <garrett@damore.org>2017-08-09 02:38:55 -0700
commitd64f12553eb6ceb67ed6f6a5b2ceb6c061d375ba (patch)
treef6bdac79578176f0d00528d191f862009e761eac /src
parent5f0398de8edd1ed4ddbf6455c66273a6608aad9a (diff)
downloadnng-d64f12553eb6ceb67ed6f6a5b2ceb6c061d375ba.tar.gz
nng-d64f12553eb6ceb67ed6f6a5b2ceb6c061d375ba.tar.bz2
nng-d64f12553eb6ceb67ed6f6a5b2ceb6c061d375ba.zip
fixes #44 open protocol by "name" (symbol) instead number
fixes #38 Make protocols "pluggable", or at least optional This is a breaking change, as we've done away with the central registered list of protocols, and instead demand the user call nng_xxx_open() where xxx is a protocol name. (We did keep a table around in the compat framework though.) There is a nice way for protocols to plug in via an nni_proto_open(), where they can use a generic constructor that they use to build a protocol specific constructor (passing their ops vector in.)
Diffstat (limited to 'src')
-rw-r--r--src/core/device.c4
-rw-r--r--src/core/protocol.c83
-rw-r--r--src/core/protocol.h36
-rw-r--r--src/core/socket.c134
-rw-r--r--src/core/socket.h7
-rw-r--r--src/nng.c16
-rw-r--r--src/nng.h67
-rw-r--r--src/nng_compat.c35
-rw-r--r--src/protocol/bus/bus.c21
-rw-r--r--src/protocol/pair/pair.c29
-rw-r--r--src/protocol/pipeline/pull.c32
-rw-r--r--src/protocol/pipeline/push.c32
-rw-r--r--src/protocol/pubsub/pub.c21
-rw-r--r--src/protocol/pubsub/sub.c26
-rw-r--r--src/protocol/reqrep/rep.c12
-rw-r--r--src/protocol/reqrep/req.c19
-rw-r--r--src/protocol/survey/respond.c12
-rw-r--r--src/protocol/survey/survey.c14
18 files changed, 338 insertions, 262 deletions
diff --git a/src/core/device.c b/src/core/device.c
index cd0c8dfc..81aafc3d 100644
--- a/src/core/device.c
+++ b/src/core/device.c
@@ -80,8 +80,8 @@ nni_device(nni_sock *sock1, nni_sock *sock2)
rv = NNG_EINVAL;
goto out;
}
- if ((sock1->s_peer != sock2->s_protocol) ||
- (sock2->s_peer != sock1->s_protocol)) {
+ if ((sock1->s_peer_id.p_id != sock2->s_self_id.p_id) ||
+ (sock2->s_peer_id.p_id != sock1->s_self_id.p_id)) {
rv = NNG_EINVAL;
goto out;
}
diff --git a/src/core/protocol.c b/src/core/protocol.c
index a60454de..7faf9068 100644
--- a/src/core/protocol.c
+++ b/src/core/protocol.c
@@ -1,5 +1,6 @@
//
// Copyright 2017 Garrett D'Amore <garrett@damore.org>
+// Copyright 2017 Capitar IT Group BV <info@capitar.com>
//
// This software is supplied under the terms of the MIT License, a
// copy of which should be located in the distribution where this
@@ -13,82 +14,14 @@
// Protocol related stuff - generically.
-// The list of protocols is hardwired. This is reasonably unlikely to
-// change, as adding new protocols is not something intended to be done
-// outside of the core.
-extern nni_proto nni_bus_proto;
-extern nni_proto nni_pair_proto;
-extern nni_proto nni_rep_proto;
-extern nni_proto nni_req_proto;
-extern nni_proto nni_pub_proto;
-extern nni_proto nni_sub_proto;
-extern nni_proto nni_push_proto;
-extern nni_proto nni_pull_proto;
-extern nni_proto nni_surveyor_proto;
-extern nni_proto nni_respondent_proto;
-
-static nni_proto *protocols[] = {
- // clang-format off
- &nni_bus_proto,
- &nni_pair_proto,
- &nni_rep_proto,
- &nni_req_proto,
- &nni_pub_proto,
- &nni_sub_proto,
- &nni_push_proto,
- &nni_pull_proto,
- &nni_surveyor_proto,
- &nni_respondent_proto,
- NULL
- // clang-format on
-};
-
-nni_proto *
-nni_proto_find(uint16_t num)
-{
- int i;
- nni_proto *p;
-
- for (i = 0; (p = protocols[i]) != NULL; i++) {
- if (p->proto_self == num) {
- break;
- }
- }
- return (p);
-}
-
-const char *
-nni_proto_name(uint16_t num)
-{
- nni_proto *p;
-
- if ((p = nni_proto_find(num)) == NULL) {
- return (NULL);
- }
- return (p->proto_name);
-}
-
-uint16_t
-nni_proto_number(const char *name)
-{
- nni_proto *p;
- int i;
-
- for (i = 0; (p = protocols[i]) != NULL; i++) {
- if (strcmp(p->proto_name, name) == 0) {
- return (p->proto_self);
- }
- }
- return (NNG_PROTO_NONE);
-}
-
-uint16_t
-nni_proto_peer(uint16_t num)
+int
+nni_proto_open(nng_socket *sockidp, const nni_proto *proto)
{
- nni_proto *p;
+ int rv;
+ nni_sock *sock;
- if ((p = nni_proto_find(num)) == NULL) {
- return (NNG_PROTO_NONE);
+ if ((rv = nni_sock_open(&sock, proto)) == 0) {
+ *sockidp = nni_sock_id(sock); // Keep socket held open.
}
- return (p->proto_peer);
+ return (rv);
}
diff --git a/src/core/protocol.h b/src/core/protocol.h
index 3a133469..14954b49 100644
--- a/src/core/protocol.h
+++ b/src/core/protocol.h
@@ -1,5 +1,6 @@
//
// Copyright 2017 Garrett D'Amore <garrett@damore.org>
+// Copyright 2017 Capitar IT Group BV <info@capitar.com>
//
// This software is supplied under the terms of the MIT License, a
// copy of which should be located in the distribution where this
@@ -47,7 +48,7 @@ struct nni_proto_pipe_ops {
};
struct nni_proto_sock_ops {
- // sock_initf creates the protocol instance, which will be stored on
+ // sock_init creates the protocol instance, which will be stored on
// the socket. This is run without the sock lock held, and allocates
// storage or other resources for the socket.
int (*sock_init)(void **, nni_sock *);
@@ -82,15 +83,31 @@ struct nni_proto_sock_ops {
nni_msg *(*sock_sfilter)(void *, nni_msg *);
};
+typedef struct nni_proto_id {
+ uint16_t p_id;
+ const char *p_name;
+} nni_proto_id;
+
struct nni_proto {
- uint16_t proto_self; // our 16-bit D
- uint16_t proto_peer; // who we peer with (ID)
- const char * proto_name; // Our name
+ uint32_t proto_version; // Ops vector version
+ nni_proto_id proto_self; // Our identity
+ nni_proto_id proto_peer; // Peer identity
uint32_t proto_flags; // Protocol flags
const nni_proto_sock_ops *proto_sock_ops; // Per-socket opeations
const nni_proto_pipe_ops *proto_pipe_ops; // Per-pipe operations.
};
+// We quite intentionally use a signature where the upper word is nonzero,
+// which ensures that if we get garbage we will reject it. This is more
+// likely to mismatch than all zero bytes would. The actual version is
+// stored in the lower word; this is not semver -- the numbers are just
+// increasing - we doubt it will increase more than a handful of times
+// during the life of the project. If we add a new version, please keep
+// the old version around -- it may be possible to automatically convert
+// older versions in the future.
+#define NNI_PROTOCOL_V0 0x50520000 // "pr\0\0"
+#define NNI_PROTOCOL_VERSION NNI_PROTOCOL_V0
+
// These flags determine which operations make sense. We use them so that
// we can reject attempts to create notification fds for operations that make
// no sense.
@@ -98,11 +115,10 @@ struct nni_proto {
#define NNI_PROTO_FLAG_SND 2 // Protocol can send
#define NNI_PROTO_FLAG_SNDRCV 3 // Protocol can both send & recv
-// These functions are not used by protocols, but rather by the socket
-// core implementation. The lookups can be used by transports as well.
-extern nni_proto * nni_proto_find(uint16_t);
-extern const char *nni_proto_name(uint16_t);
-extern uint16_t nni_proto_number(const char *);
-extern uint16_t nni_proto_peer(uint16_t);
+// nni_proto_open is called by the protocol to create a socket instance
+// with its ops vector. The intent is that applications will only see
+// the single protocol-specific constructure, like nng_pair_v0_open(),
+// which should just be a thin wrapper around this.
+extern int nni_proto_open(nng_socket *, const nni_proto *);
#endif // CORE_PROTOCOL_H
diff --git a/src/core/socket.c b/src/core/socket.c
index f2dbb573..f01379a8 100644
--- a/src/core/socket.c
+++ b/src/core/socket.c
@@ -83,7 +83,7 @@ nni_sock_pipe_start(nni_pipe *pipe)
// We're closing, bail out.
return (NNG_ECLOSED);
}
- if (nni_pipe_peer(pipe) != sock->s_peer) {
+ if (nni_pipe_peer(pipe) != sock->s_peer_id.p_id) {
// Peer protocol mismatch.
return (NNG_EPROTO);
}
@@ -175,7 +175,7 @@ nni_sock_pipe_ready(nni_sock *sock, nni_pipe *pipe)
nni_mtx_unlock(&sock->s_mx);
return (NNG_ECLOSED);
}
- if (nni_pipe_peer(pipe) != sock->s_peer) {
+ if (nni_pipe_peer(pipe) != sock->s_peer_id.p_id) {
nni_mtx_unlock(&sock->s_mx);
return (NNG_EPROTO);
}
@@ -305,47 +305,6 @@ nni_sock_unnotify(nni_sock *sock, nni_notify *notify)
NNI_FREE_STRUCT(notify);
}
-static nni_msg *
-nni_sock_nullfilter(void *arg, nni_msg *mp)
-{
- NNI_ARG_UNUSED(arg);
- return (mp);
-}
-
-static int
-nni_sock_nullgetopt(void *arg, int num, void *data, size_t *szp)
-{
- NNI_ARG_UNUSED(arg);
- NNI_ARG_UNUSED(num);
- NNI_ARG_UNUSED(data);
- NNI_ARG_UNUSED(szp);
- return (NNG_ENOTSUP);
-}
-
-static int
-nni_sock_nullsetopt(void *arg, int num, const void *data, size_t sz)
-{
- NNI_ARG_UNUSED(arg);
- NNI_ARG_UNUSED(num);
- NNI_ARG_UNUSED(data);
- NNI_ARG_UNUSED(sz);
- return (NNG_ENOTSUP);
-}
-
-static void
-nni_sock_nullop(void *arg)
-{
- NNI_ARG_UNUSED(arg);
-}
-
-static int
-nni_sock_nullstartpipe(void *arg)
-{
- NNI_ARG_UNUSED(arg);
-
- return (0);
-}
-
static void *
nni_sock_ctor(uint32_t id)
{
@@ -456,23 +415,23 @@ nni_sock_sys_fini(void)
nni_mtx_fini(&nni_sock_lk);
}
-// nn_sock_open creates the underlying socket.
int
-nni_sock_open(nni_sock **sockp, uint16_t pnum)
+nni_sock_open(nni_sock **sockp, const nni_proto *proto)
{
nni_sock * sock;
- nni_proto * proto;
int rv;
nni_proto_sock_ops *sops;
nni_proto_pipe_ops *pops;
uint32_t sockid;
+ if (proto->proto_version != NNI_PROTOCOL_VERSION) {
+ // unsupported protocol version
+ return (NNG_ENOTSUP);
+ }
+
if ((rv = nni_init()) != 0) {
return (rv);
}
- if ((proto = nni_proto_find(pnum)) == NULL) {
- return (NNG_ENOTSUP);
- }
rv = nni_objhash_alloc(nni_socks, &sockid, (void **) &sock);
if (rv != 0) {
@@ -480,40 +439,20 @@ nni_sock_open(nni_sock **sockp, uint16_t pnum)
}
// We make a copy of the protocol operations.
- sock->s_protocol = proto->proto_self;
- sock->s_peer = proto->proto_peer;
+ sock->s_self_id = proto->proto_self;
+ sock->s_peer_id = proto->proto_peer;
sock->s_flags = proto->proto_flags;
sock->s_sock_ops = *proto->proto_sock_ops;
+ sock->s_pipe_ops = *proto->proto_pipe_ops;
sops = &sock->s_sock_ops;
- if (sops->sock_sfilter == NULL) {
- sops->sock_sfilter = nni_sock_nullfilter;
- }
- if (sops->sock_rfilter == NULL) {
- sops->sock_rfilter = nni_sock_nullfilter;
- }
- if (sops->sock_getopt == NULL) {
- sops->sock_getopt = nni_sock_nullgetopt;
- }
- if (sops->sock_setopt == NULL) {
- sops->sock_setopt = nni_sock_nullsetopt;
- }
- if (sops->sock_close == NULL) {
- sops->sock_close = nni_sock_nullop;
- }
- if (sops->sock_open == NULL) {
- sops->sock_open = nni_sock_nullop;
- }
- sock->s_pipe_ops = *proto->proto_pipe_ops;
- pops = &sock->s_pipe_ops;
- if (pops->pipe_start == NULL) {
- pops->pipe_start = nni_sock_nullstartpipe;
- }
- if (pops->pipe_stop == NULL) {
- pops->pipe_stop = nni_sock_nullop;
- }
+ NNI_ASSERT(sock->s_sock_ops.sock_open != NULL);
+ NNI_ASSERT(sock->s_sock_ops.sock_close != NULL);
+
+ NNI_ASSERT(sock->s_pipe_ops.pipe_start != NULL);
+ NNI_ASSERT(sock->s_pipe_ops.pipe_stop != NULL);
- if ((rv = sops->sock_init(&sock->s_data, sock)) != 0) {
+ if ((rv = sock->s_sock_ops.sock_init(&sock->s_data, sock)) != 0) {
nni_objhash_unref(nni_socks, sockid);
return (rv);
}
@@ -730,7 +669,9 @@ nni_sock_sendmsg(nni_sock *sock, nni_msg *msg, nni_time expire)
}
besteffort = sock->s_besteffort;
- msg = sock->s_sock_ops.sock_sfilter(sock->s_data, msg);
+ if (sock->s_sock_ops.sock_sfilter != NULL) {
+ msg = sock->s_sock_ops.sock_sfilter(sock->s_data, msg);
+ }
nni_mtx_unlock(&sock->s_mx);
if (msg == NULL) {
@@ -780,9 +721,11 @@ nni_sock_recvmsg(nni_sock *sock, nni_msg **msgp, nni_time expire)
if (rv != 0) {
return (rv);
}
- nni_mtx_lock(&sock->s_mx);
- msg = sock->s_sock_ops.sock_rfilter(sock->s_data, msg);
- nni_mtx_unlock(&sock->s_mx);
+ if (sock->s_sock_ops.sock_rfilter != NULL) {
+ nni_mtx_lock(&sock->s_mx);
+ msg = sock->s_sock_ops.sock_rfilter(sock->s_data, msg);
+ nni_mtx_unlock(&sock->s_mx);
+ }
if (msg != NULL) {
break;
}
@@ -797,13 +740,13 @@ nni_sock_recvmsg(nni_sock *sock, nni_msg **msgp, nni_time expire)
uint16_t
nni_sock_proto(nni_sock *sock)
{
- return (sock->s_protocol);
+ return (sock->s_self_id.p_id);
}
uint16_t
nni_sock_peer(nni_sock *sock)
{
- return (sock->s_peer);
+ return (sock->s_peer_id.p_id);
}
nni_duration
@@ -924,10 +867,13 @@ nni_sock_setopt(nni_sock *sock, int opt, const void *val, size_t size)
nni_mtx_unlock(&sock->s_mx);
return (NNG_ECLOSED);
}
- rv = sock->s_sock_ops.sock_setopt(sock->s_data, opt, val, size);
- if (rv != NNG_ENOTSUP) {
- nni_mtx_unlock(&sock->s_mx);
- return (rv);
+ if (sock->s_sock_ops.sock_setopt != NULL) {
+ rv =
+ sock->s_sock_ops.sock_setopt(sock->s_data, opt, val, size);
+ if (rv != NNG_ENOTSUP) {
+ nni_mtx_unlock(&sock->s_mx);
+ return (rv);
+ }
}
switch (opt) {
case NNG_OPT_LINGER:
@@ -970,11 +916,15 @@ nni_sock_getopt(nni_sock *sock, int opt, void *val, size_t *sizep)
nni_mtx_unlock(&sock->s_mx);
return (NNG_ECLOSED);
}
- rv = sock->s_sock_ops.sock_getopt(sock->s_data, opt, val, sizep);
- if (rv != NNG_ENOTSUP) {
- nni_mtx_unlock(&sock->s_mx);
- return (rv);
+ if (sock->s_sock_ops.sock_getopt != NULL) {
+ rv = sock->s_sock_ops.sock_getopt(
+ sock->s_data, opt, val, sizep);
+ if (rv != NNG_ENOTSUP) {
+ nni_mtx_unlock(&sock->s_mx);
+ return (rv);
+ }
}
+
switch (opt) {
case NNG_OPT_LINGER:
rv = nni_getopt_duration(&sock->s_linger, val, sizep);
diff --git a/src/core/socket.h b/src/core/socket.h
index 54c1b4da..6824b04b 100644
--- a/src/core/socket.h
+++ b/src/core/socket.h
@@ -25,8 +25,9 @@ struct nni_socket {
nni_list_node s_node;
- uint16_t s_protocol;
- uint16_t s_peer;
+ nni_proto_id s_self_id;
+ nni_proto_id s_peer_id;
+
uint32_t s_flags;
nni_proto_pipe_ops s_pipe_ops;
@@ -67,7 +68,7 @@ extern void nni_sock_sys_fini(void);
extern int nni_sock_find(nni_sock **, uint32_t);
extern void nni_sock_rele(nni_sock *);
-extern int nni_sock_open(nni_sock **, uint16_t);
+extern int nni_sock_open(nni_sock **, const nni_proto *);
extern void nni_sock_close(nni_sock *);
extern void nni_sock_closeall(void);
extern int nni_sock_shutdown(nni_sock *);
diff --git a/src/nng.c b/src/nng.c
index d5cc4f2b..815fb2b8 100644
--- a/src/nng.c
+++ b/src/nng.c
@@ -21,22 +21,6 @@
#include <string.h>
-int
-nng_open(nng_socket *sidp, uint16_t proto)
-{
- int rv;
- nni_sock *sock;
-
- if ((rv = nni_sock_open(&sock, proto)) != 0) {
- return (rv);
- }
- *sidp = nni_sock_id(sock);
-
- // Keep the socket "held" until it is explicitly closed.
-
- return (0);
-}
-
void
nng_fini(void)
{
diff --git a/src/nng.h b/src/nng.h
index aac7ec89..99c95b92 100644
--- a/src/nng.h
+++ b/src/nng.h
@@ -52,11 +52,6 @@ typedef struct nng_snapshot nng_snapshot;
typedef struct nng_stat nng_stat;
typedef uint32_t nng_endpoint; // XXX: REMOVE ME.
-// nng_open simply creates a socket of the given class. It returns an
-// error code on failure, or zero on success. The socket starts in cooked
-// mode.
-NNG_DECL int nng_open(nng_socket *, uint16_t proto);
-
// nng_fini is used to terminate the library, freeing certain global resources.
// Its a good idea to call this with atexit() or during application shutdown.
// For most cases, this call is optional, but failure to do so may cause
@@ -293,20 +288,58 @@ enum nng_flag_enum {
// a valid protocol numbered 0 (NNG_PROTO_NONE).
#define NNG_PROTO(major, minor) (((major) *16) + (minor))
enum nng_proto_enum {
- NNG_PROTO_NONE = NNG_PROTO(0, 0),
- NNG_PROTO_PAIR = NNG_PROTO(1, 0),
- NNG_PROTO_PUB = NNG_PROTO(2, 0),
- NNG_PROTO_SUB = NNG_PROTO(2, 1),
- NNG_PROTO_REQ = NNG_PROTO(3, 0),
- NNG_PROTO_REP = NNG_PROTO(3, 1),
- NNG_PROTO_PUSH = NNG_PROTO(5, 0),
- NNG_PROTO_PULL = NNG_PROTO(5, 1),
- NNG_PROTO_SURVEYOR = NNG_PROTO(6, 2),
- NNG_PROTO_RESPONDENT = NNG_PROTO(6, 3),
- NNG_PROTO_BUS = NNG_PROTO(7, 0),
- NNG_PROTO_STAR = NNG_PROTO(100, 0),
+ NNG_PROTO_NONE = NNG_PROTO(0, 0),
+ NNG_PROTO_PAIR_V0 = NNG_PROTO(1, 0),
+ NNG_PROTO_PUB_V0 = NNG_PROTO(2, 0),
+ NNG_PROTO_SUB_V0 = NNG_PROTO(2, 1),
+ NNG_PROTO_REQ_V0 = NNG_PROTO(3, 0),
+ NNG_PROTO_REP_V0 = NNG_PROTO(3, 1),
+ NNG_PROTO_PUSH_V0 = NNG_PROTO(5, 0),
+ NNG_PROTO_PULL_V0 = NNG_PROTO(5, 1),
+ NNG_PROTO_SURVEYOR_V0 = NNG_PROTO(6, 2),
+ NNG_PROTO_RESPONDENT_V0 = NNG_PROTO(6, 3),
+ NNG_PROTO_BUS_V0 = NNG_PROTO(7, 0),
+ NNG_PROTO_STAR_V0 = NNG_PROTO(100, 0),
+
+ // "Legacy" names. Please use explicit versioned names above.
+ NNG_PROTO_BUS = NNG_PROTO_BUS_V0,
+ NNG_PROTO_PAIR = NNG_PROTO_PAIR_V0,
+ NNG_PROTO_SUB = NNG_PROTO_SUB_V0,
+ NNG_PROTO_PUB = NNG_PROTO_PUB_V0,
+ NNG_PROTO_REQ = NNG_PROTO_REQ_V0,
+ NNG_PROTO_REP = NNG_PROTO_REP_V0,
+ NNG_PROTO_PUSH = NNG_PROTO_PUSH_V0,
+ NNG_PROTO_PULL = NNG_PROTO_PULL_V0,
+ NNG_PROTO_SURVEYOR = NNG_PROTO_SURVEYOR_V0,
+ NNG_PROTO_RESPONDENT = NNG_PROTO_RESPONDENT_V0,
};
+// Builtin protocol socket constructors.
+extern int nng_bus0_open(nng_socket *);
+extern int nng_pair0_open(nng_socket *);
+extern int nng_pub0_open(nng_socket *);
+extern int nng_sub0_open(nng_socket *);
+extern int nng_push0_open(nng_socket *);
+extern int nng_pull0_open(nng_socket *);
+extern int nng_req0_open(nng_socket *);
+extern int nng_rep0_open(nng_socket *);
+extern int nng_surveyor0_open(nng_socket *);
+extern int nng_respondent0_open(nng_socket *);
+
+// Default versions. These provide compile time defaults; note that
+// the actual protocols are baked into the binary; this should avoid
+// suprising. Choosing a new protocol should be done explicitly.
+#define nng_bus_open nng_bus0_open
+#define nng_pair_open nng_pair0_open
+#define nng_pub_open nng_pub0_open
+#define nng_sub_open nng_sub0_open
+#define nng_push_open nng_push0_open
+#define nng_pull_open nng_pull0_open
+#define nng_req_open nng_req0_open
+#define nng_rep_open nng_rep0_open
+#define nng_surveyor_open nng_surveyor0_open
+#define nng_respondent_open nng_respondent0_open
+
// Options. We encode option numbers as follows:
//
// <level> - 0: socket, 1: transport
diff --git a/src/nng_compat.c b/src/nng_compat.c
index cb8d29ab..34fc6553 100644
--- a/src/nng_compat.c
+++ b/src/nng_compat.c
@@ -87,17 +87,48 @@ nn_errno(void)
return (errno);
}
+static const struct {
+ uint16_t p_id;
+ int (*p_open)(nng_socket *);
+} nn_protocols[] = {
+ // clang-format off
+ { NNG_PROTO_BUS_V0, nng_bus_open },
+ { NNG_PROTO_PAIR_V0, nng_pair_open },
+ { NNG_PROTO_PUSH_V0, nng_push_open },
+ { NNG_PROTO_PULL_V0, nng_pull_open },
+ { NNG_PROTO_PUB_V0, nng_pub_open },
+ { NNG_PROTO_SUB_V0, nng_sub_open },
+ { NNG_PROTO_REQ_V0, nng_req_open },
+ { NNG_PROTO_REP_V0, nng_rep_open },
+ { NNG_PROTO_SURVEYOR_V0, nng_surveyor_open },
+ { NNG_PROTO_RESPONDENT_V0, nng_respondent_open },
+ { NNG_PROTO_NONE, NULL },
+ // clang-format on
+};
+
int
nn_socket(int domain, int protocol)
{
nng_socket sock;
int rv;
+ int i;
if ((domain != AF_SP) && (domain != AF_SP_RAW)) {
- errno = EAFNOSUPPORT;
+ nn_seterror(EAFNOSUPPORT);
return (-1);
}
- if ((rv = nng_open(&sock, protocol)) != 0) {
+
+ for (i = 0; nn_protocols[i].p_id != NNG_PROTO_NONE; i++) {
+ if (nn_protocols[i].p_id == protocol) {
+ break;
+ }
+ }
+ if (nn_protocols[i].p_open == NULL) {
+ nn_seterror(ENOTSUP);
+ return (-1);
+ }
+
+ if ((rv = nn_protocols[i].p_open(&sock)) != 0) {
nn_seterror(rv);
return (-1);
}
diff --git a/src/protocol/bus/bus.c b/src/protocol/bus/bus.c
index 3070b90d..c5a6ce06 100644
--- a/src/protocol/bus/bus.c
+++ b/src/protocol/bus/bus.c
@@ -104,6 +104,14 @@ nni_bus_sock_open(void *arg)
}
static void
+nni_bus_sock_close(void *arg)
+{
+ nni_bus_sock *psock = arg;
+
+ nni_aio_cancel(&psock->aio_getq, NNG_ECLOSED);
+}
+
+static void
nni_bus_pipe_fini(void *arg)
{
nni_bus_pipe *ppipe = arg;
@@ -383,6 +391,7 @@ static nni_proto_sock_ops nni_bus_sock_ops = {
.sock_init = nni_bus_sock_init,
.sock_fini = nni_bus_sock_fini,
.sock_open = nni_bus_sock_open,
+ .sock_close = nni_bus_sock_close,
.sock_setopt = nni_bus_sock_setopt,
.sock_getopt = nni_bus_sock_getopt,
};
@@ -390,10 +399,16 @@ static nni_proto_sock_ops nni_bus_sock_ops = {
// This is the global protocol structure -- our linkage to the core.
// This should be the only global non-static symbol in this file.
nni_proto nni_bus_proto = {
- .proto_self = NNG_PROTO_BUS,
- .proto_peer = NNG_PROTO_BUS,
- .proto_name = "bus",
+ .proto_version = NNI_PROTOCOL_VERSION,
+ .proto_self = { NNG_PROTO_BUS_V0, "bus" },
+ .proto_peer = { NNG_PROTO_BUS_V0, "bus" },
.proto_flags = NNI_PROTO_FLAG_SNDRCV,
.proto_sock_ops = &nni_bus_sock_ops,
.proto_pipe_ops = &nni_bus_pipe_ops,
};
+
+int
+nng_bus0_open(nng_socket *sidp)
+{
+ return (nni_proto_open(sidp, &nni_bus_proto));
+}
diff --git a/src/protocol/pair/pair.c b/src/protocol/pair/pair.c
index 55ce5aa9..59465293 100644
--- a/src/protocol/pair/pair.c
+++ b/src/protocol/pair/pair.c
@@ -231,6 +231,18 @@ nni_pair_send_cb(void *arg)
nni_msgq_aio_get(psock->uwq, &ppipe->aio_getq);
}
+static void
+nni_pair_sock_open(void *arg)
+{
+ NNI_ARG_UNUSED(arg);
+}
+
+static void
+nni_pair_sock_close(void *arg)
+{
+ NNI_ARG_UNUSED(arg);
+}
+
static int
nni_pair_sock_setopt(void *arg, int opt, const void *buf, size_t sz)
{
@@ -263,9 +275,6 @@ nni_pair_sock_getopt(void *arg, int opt, void *buf, size_t *szp)
return (rv);
}
-// This is the global protocol structure -- our linkage to the core.
-// This should be the only global non-static symbol in this file.
-
static nni_proto_pipe_ops nni_pair_pipe_ops = {
.pipe_init = nni_pair_pipe_init,
.pipe_fini = nni_pair_pipe_fini,
@@ -276,15 +285,23 @@ static nni_proto_pipe_ops nni_pair_pipe_ops = {
static nni_proto_sock_ops nni_pair_sock_ops = {
.sock_init = nni_pair_sock_init,
.sock_fini = nni_pair_sock_fini,
+ .sock_open = nni_pair_sock_open,
+ .sock_close = nni_pair_sock_close,
.sock_setopt = nni_pair_sock_setopt,
.sock_getopt = nni_pair_sock_getopt,
};
nni_proto nni_pair_proto = {
- .proto_self = NNG_PROTO_PAIR,
- .proto_peer = NNG_PROTO_PAIR,
- .proto_name = "pair",
+ .proto_version = NNI_PROTOCOL_VERSION,
+ .proto_self = { NNG_PROTO_PAIR_V0, "pair" },
+ .proto_peer = { NNG_PROTO_PAIR_V0, "pair" },
.proto_flags = NNI_PROTO_FLAG_SNDRCV,
.proto_sock_ops = &nni_pair_sock_ops,
.proto_pipe_ops = &nni_pair_pipe_ops,
};
+
+int
+nng_pair0_open(nng_socket *sidp)
+{
+ return (nni_proto_open(sidp, &nni_pair_proto));
+}
diff --git a/src/protocol/pipeline/pull.c b/src/protocol/pipeline/pull.c
index cde79824..e3c73342 100644
--- a/src/protocol/pipeline/pull.c
+++ b/src/protocol/pipeline/pull.c
@@ -56,9 +56,7 @@ nni_pull_sock_fini(void *arg)
{
nni_pull_sock *pull = arg;
- if (pull != NULL) {
- NNI_FREE_STRUCT(pull);
- }
+ NNI_FREE_STRUCT(pull);
}
static int
@@ -163,6 +161,18 @@ nni_pull_putq(nni_pull_pipe *pp, nni_msg *msg)
nni_msgq_aio_put(pull->urq, &pp->putq_aio);
}
+static void
+nni_pull_sock_open(void *arg)
+{
+ NNI_ARG_UNUSED(arg);
+}
+
+static void
+nni_pull_sock_close(void *arg)
+{
+ NNI_ARG_UNUSED(arg);
+}
+
static int
nni_pull_sock_setopt(void *arg, int opt, const void *buf, size_t sz)
{
@@ -195,8 +205,6 @@ nni_pull_sock_getopt(void *arg, int opt, void *buf, size_t *szp)
return (rv);
}
-// This is the global protocol structure -- our linkage to the core.
-// This should be the only global non-static symbol in this file.
static nni_proto_pipe_ops nni_pull_pipe_ops = {
.pipe_init = nni_pull_pipe_init,
.pipe_fini = nni_pull_pipe_fini,
@@ -207,15 +215,23 @@ static nni_proto_pipe_ops nni_pull_pipe_ops = {
static nni_proto_sock_ops nni_pull_sock_ops = {
.sock_init = nni_pull_sock_init,
.sock_fini = nni_pull_sock_fini,
+ .sock_open = nni_pull_sock_open,
+ .sock_close = nni_pull_sock_close,
.sock_setopt = nni_pull_sock_setopt,
.sock_getopt = nni_pull_sock_getopt,
};
nni_proto nni_pull_proto = {
- .proto_self = NNG_PROTO_PULL,
- .proto_peer = NNG_PROTO_PUSH,
- .proto_name = "pull",
+ .proto_version = NNI_PROTOCOL_VERSION,
+ .proto_self = { NNG_PROTO_PULL_V0, "pull" },
+ .proto_peer = { NNG_PROTO_PUSH_V0, "push" },
.proto_flags = NNI_PROTO_FLAG_RCV,
.proto_pipe_ops = &nni_pull_pipe_ops,
.proto_sock_ops = &nni_pull_sock_ops,
};
+
+int
+nng_pull0_open(nng_socket *sidp)
+{
+ return (nni_proto_open(sidp, &nni_pull_proto));
+}
diff --git a/src/protocol/pipeline/push.c b/src/protocol/pipeline/push.c
index b7d4322c..14b3b191 100644
--- a/src/protocol/pipeline/push.c
+++ b/src/protocol/pipeline/push.c
@@ -63,9 +63,19 @@ nni_push_sock_fini(void *arg)
{
nni_push_sock *push = arg;
- if (push != NULL) {
- NNI_FREE_STRUCT(push);
- }
+ NNI_FREE_STRUCT(push);
+}
+
+static void
+nni_push_sock_open(void *arg)
+{
+ NNI_ARG_UNUSED(arg);
+}
+
+static void
+nni_push_sock_close(void *arg)
+{
+ NNI_ARG_UNUSED(arg);
}
static void
@@ -221,8 +231,6 @@ nni_push_sock_getopt(void *arg, int opt, void *buf, size_t *szp)
return (rv);
}
-// This is the global protocol structure -- our linkage to the core.
-// This should be the only global non-static symbol in this file.
static nni_proto_pipe_ops nni_push_pipe_ops = {
.pipe_init = nni_push_pipe_init,
.pipe_fini = nni_push_pipe_fini,
@@ -233,15 +241,23 @@ static nni_proto_pipe_ops nni_push_pipe_ops = {
static nni_proto_sock_ops nni_push_sock_ops = {
.sock_init = nni_push_sock_init,
.sock_fini = nni_push_sock_fini,
+ .sock_open = nni_push_sock_open,
+ .sock_close = nni_push_sock_close,
.sock_setopt = nni_push_sock_setopt,
.sock_getopt = nni_push_sock_getopt,
};
nni_proto nni_push_proto = {
- .proto_self = NNG_PROTO_PUSH,
- .proto_peer = NNG_PROTO_PULL,
- .proto_name = "push",
+ .proto_version = NNI_PROTOCOL_VERSION,
+ .proto_self = { NNG_PROTO_PUSH_V0, "push" },
+ .proto_peer = { NNG_PROTO_PULL_V0, "pull" },
.proto_flags = NNI_PROTO_FLAG_SND,
.proto_pipe_ops = &nni_push_pipe_ops,
.proto_sock_ops = &nni_push_sock_ops,
};
+
+int
+nng_push0_open(nng_socket *sidp)
+{
+ return (nni_proto_open(sidp, &nni_push_proto));
+}
diff --git a/src/protocol/pubsub/pub.c b/src/protocol/pubsub/pub.c
index e32f179a..161a5d79 100644
--- a/src/protocol/pubsub/pub.c
+++ b/src/protocol/pubsub/pub.c
@@ -98,6 +98,14 @@ nni_pub_sock_open(void *arg)
}
static void
+nni_pub_sock_close(void *arg)
+{
+ nni_pub_sock *pub = arg;
+
+ nni_aio_cancel(&pub->aio_getq, NNG_ECLOSED);
+}
+
+static void
nni_pub_pipe_fini(void *arg)
{
nni_pub_pipe *pp = arg;
@@ -319,15 +327,22 @@ nni_proto_sock_ops nni_pub_sock_ops = {
.sock_init = nni_pub_sock_init,
.sock_fini = nni_pub_sock_fini,
.sock_open = nni_pub_sock_open,
+ .sock_close = nni_pub_sock_close,
.sock_setopt = nni_pub_sock_setopt,
.sock_getopt = nni_pub_sock_getopt,
};
nni_proto nni_pub_proto = {
- .proto_self = NNG_PROTO_PUB,
- .proto_peer = NNG_PROTO_SUB,
- .proto_name = "pub",
+ .proto_version = NNI_PROTOCOL_VERSION,
+ .proto_self = { NNG_PROTO_PUB_V0, "pub" },
+ .proto_peer = { NNG_PROTO_SUB_V0, "sub" },
.proto_flags = NNI_PROTO_FLAG_SND,
.proto_sock_ops = &nni_pub_sock_ops,
.proto_pipe_ops = &nni_pub_pipe_ops,
};
+
+int
+nng_pub0_open(nng_socket *sidp)
+{
+ return (nni_proto_open(sidp, &nni_pub_proto));
+}
diff --git a/src/protocol/pubsub/sub.c b/src/protocol/pubsub/sub.c
index 03d76e2d..53f01e0f 100644
--- a/src/protocol/pubsub/sub.c
+++ b/src/protocol/pubsub/sub.c
@@ -79,6 +79,18 @@ nni_sub_sock_fini(void *arg)
NNI_FREE_STRUCT(sub);
}
+static void
+nni_sub_sock_open(void *arg)
+{
+ NNI_ARG_UNUSED(arg);
+}
+
+static void
+nni_sub_sock_close(void *arg)
+{
+ NNI_ARG_UNUSED(arg);
+}
+
static int
nni_sub_pipe_init(void **spp, nni_pipe *pipe, void *ssock)
{
@@ -330,16 +342,24 @@ static nni_proto_pipe_ops nni_sub_pipe_ops = {
static nni_proto_sock_ops nni_sub_sock_ops = {
.sock_init = nni_sub_sock_init,
.sock_fini = nni_sub_sock_fini,
+ .sock_open = nni_sub_sock_open,
+ .sock_close = nni_sub_sock_close,
.sock_setopt = nni_sub_sock_setopt,
.sock_getopt = nni_sub_sock_getopt,
.sock_rfilter = nni_sub_sock_rfilter,
};
nni_proto nni_sub_proto = {
- .proto_self = NNG_PROTO_SUB,
- .proto_peer = NNG_PROTO_PUB,
- .proto_name = "sub",
+ .proto_version = NNI_PROTOCOL_VERSION,
+ .proto_self = { NNG_PROTO_SUB_V0, "sub" },
+ .proto_peer = { NNG_PROTO_PUB_V0, "pub" },
.proto_flags = NNI_PROTO_FLAG_RCV,
.proto_sock_ops = &nni_sub_sock_ops,
.proto_pipe_ops = &nni_sub_pipe_ops,
};
+
+int
+nng_sub0_open(nng_socket *sidp)
+{
+ return (nni_proto_open(sidp, &nni_sub_proto));
+}
diff --git a/src/protocol/reqrep/rep.c b/src/protocol/reqrep/rep.c
index c7546182..1adfd0f8 100644
--- a/src/protocol/reqrep/rep.c
+++ b/src/protocol/reqrep/rep.c
@@ -488,10 +488,16 @@ static nni_proto_sock_ops nni_rep_sock_ops = {
};
nni_proto nni_rep_proto = {
- .proto_self = NNG_PROTO_REP,
- .proto_peer = NNG_PROTO_REQ,
- .proto_name = "rep",
+ .proto_version = NNI_PROTOCOL_VERSION,
+ .proto_self = { NNG_PROTO_REP_V0, "rep" },
+ .proto_peer = { NNG_PROTO_REQ_V0, "req" },
.proto_flags = NNI_PROTO_FLAG_SNDRCV,
.proto_sock_ops = &nni_rep_sock_ops,
.proto_pipe_ops = &nni_rep_pipe_ops,
};
+
+int
+nng_rep0_open(nng_socket *sidp)
+{
+ return (nni_proto_open(sidp, &nni_rep_proto));
+}
diff --git a/src/protocol/reqrep/req.c b/src/protocol/reqrep/req.c
index 7ec53c90..d0dd3887 100644
--- a/src/protocol/reqrep/req.c
+++ b/src/protocol/reqrep/req.c
@@ -112,6 +112,12 @@ nni_req_sock_init(void **reqp, nni_sock *sock)
}
static void
+nni_req_sock_open(void *arg)
+{
+ NNI_ARG_UNUSED(arg);
+}
+
+static void
nni_req_sock_close(void *arg)
{
nni_req_sock *req = arg;
@@ -626,6 +632,7 @@ static nni_proto_pipe_ops nni_req_pipe_ops = {
static nni_proto_sock_ops nni_req_sock_ops = {
.sock_init = nni_req_sock_init,
.sock_fini = nni_req_sock_fini,
+ .sock_open = nni_req_sock_open,
.sock_close = nni_req_sock_close,
.sock_setopt = nni_req_sock_setopt,
.sock_getopt = nni_req_sock_getopt,
@@ -634,10 +641,16 @@ static nni_proto_sock_ops nni_req_sock_ops = {
};
nni_proto nni_req_proto = {
- .proto_self = NNG_PROTO_REQ,
- .proto_peer = NNG_PROTO_REP,
- .proto_name = "req",
+ .proto_version = NNI_PROTOCOL_VERSION,
+ .proto_self = { NNG_PROTO_REQ_V0, "req" },
+ .proto_peer = { NNG_PROTO_REP_V0, "rep" },
.proto_flags = NNI_PROTO_FLAG_SNDRCV,
.proto_sock_ops = &nni_req_sock_ops,
.proto_pipe_ops = &nni_req_pipe_ops,
};
+
+int
+nng_req0_open(nng_socket *sidp)
+{
+ return (nni_proto_open(sidp, &nni_req_proto));
+}
diff --git a/src/protocol/survey/respond.c b/src/protocol/survey/respond.c
index 089e730e..4db79b86 100644
--- a/src/protocol/survey/respond.c
+++ b/src/protocol/survey/respond.c
@@ -504,10 +504,16 @@ static nni_proto_sock_ops nni_resp_sock_ops = {
};
nni_proto nni_respondent_proto = {
- .proto_self = NNG_PROTO_RESPONDENT,
- .proto_peer = NNG_PROTO_SURVEYOR,
- .proto_name = "respondent",
+ .proto_version = NNI_PROTOCOL_VERSION,
+ .proto_self = { NNG_PROTO_RESPONDENT_V0, "respondent" },
+ .proto_peer = { NNG_PROTO_SURVEYOR_V0, "surveyor" },
.proto_flags = NNI_PROTO_FLAG_SNDRCV,
.proto_sock_ops = &nni_resp_sock_ops,
.proto_pipe_ops = &nni_resp_pipe_ops,
};
+
+int
+nng_respondent0_open(nng_socket *sidp)
+{
+ return (nni_proto_open(sidp, &nni_respondent_proto));
+}
diff --git a/src/protocol/survey/survey.c b/src/protocol/survey/survey.c
index 633e1491..91fe4ad3 100644
--- a/src/protocol/survey/survey.c
+++ b/src/protocol/survey/survey.c
@@ -470,13 +470,17 @@ static nni_proto_sock_ops nni_surv_sock_ops = {
.sock_sfilter = nni_surv_sock_sfilter,
};
-// This is the global protocol structure -- our linkage to the core.
-// This should be the only global non-static symbol in this file.
nni_proto nni_surveyor_proto = {
- .proto_self = NNG_PROTO_SURVEYOR,
- .proto_peer = NNG_PROTO_RESPONDENT,
- .proto_name = "surveyor",
+ .proto_version = NNI_PROTOCOL_VERSION,
+ .proto_self = { NNG_PROTO_SURVEYOR_V0, "surveyor" },
+ .proto_peer = { NNG_PROTO_RESPONDENT_V0, "respondent" },
.proto_flags = NNI_PROTO_FLAG_SNDRCV,
.proto_sock_ops = &nni_surv_sock_ops,
.proto_pipe_ops = &nni_surv_pipe_ops,
};
+
+int
+nng_surveyor0_open(nng_socket *sidp)
+{
+ return (nni_proto_open(sidp, &nni_surveyor_proto));
+}