diff options
| -rw-r--r-- | include/nng/nng.h | 3 | ||||
| -rw-r--r-- | src/core/dialer.c | 75 | ||||
| -rw-r--r-- | src/core/dialer.h | 15 | ||||
| -rw-r--r-- | src/core/listener.c | 1 | ||||
| -rw-r--r-- | src/core/url.c | 18 | ||||
| -rw-r--r-- | src/nng.c | 48 | ||||
| -rw-r--r-- | src/sp/transport.c | 2 | ||||
| -rw-r--r-- | src/sp/transport/tcp/tcp_test.c | 6 | ||||
| -rw-r--r-- | src/sp/transport/udp/udp_tran_test.c | 4 |
9 files changed, 155 insertions, 17 deletions
diff --git a/include/nng/nng.h b/include/nng/nng.h index 658968d5..1423759f 100644 --- a/include/nng/nng.h +++ b/include/nng/nng.h @@ -314,9 +314,11 @@ NNG_DECL int nng_listen(nng_socket, const char *, nng_listener *, int); // case, it will still be reconnected in the background -- only the initial // connection attempt is normally synchronous.) NNG_DECL int nng_dial(nng_socket, const char *, nng_dialer *, int); +NNG_DECL int nng_dial_url(nng_socket, const nng_url *url, nng_dialer *, int); // nng_dialer_create creates a new dialer, that is not yet started. NNG_DECL int nng_dialer_create(nng_dialer *, nng_socket, const char *); +NNG_DECL int nng_dialer_create_url(nng_dialer *, nng_socket, const nng_url *); // nng_listener_create creates a new listener, that is not yet started. NNG_DECL int nng_listener_create(nng_listener *, nng_socket, const char *); @@ -729,6 +731,7 @@ NNG_DECL nng_listener nng_pipe_listener(nng_pipe); #define NNG_OPT_SENDTIMEO "send-timeout" #define NNG_OPT_LOCADDR "local-address" #define NNG_OPT_REMADDR "remote-address" +#define NNG_OPT_URL "url" #define NNG_OPT_MAXTTL "ttl-max" #define NNG_OPT_RECVMAXSZ "recv-size-max" #define NNG_OPT_RECONNMINT "reconnect-time-min" diff --git a/src/core/dialer.c b/src/core/dialer.c index dfa198c5..2945fded 100644 --- a/src/core/dialer.c +++ b/src/core/dialer.c @@ -204,6 +204,80 @@ nni_dialer_bump_error(nni_dialer *d, int err) #endif } +static int +nni_dialer_init(nni_dialer *d, nni_sock *s, nni_sp_tran *tran) +{ + int rv; + + d->d_closed = false; + d->d_data = NULL; + d->d_ref = 1; + d->d_sock = s; + d->d_tran = tran; + nni_atomic_flag_reset(&d->d_started); + + // Make a copy of the endpoint operations. This allows us to + // modify them (to override NULLs for example), and avoids an extra + // dereference on hot paths. + d->d_ops = *tran->tran_dialer; + + NNI_LIST_NODE_INIT(&d->d_node); + NNI_LIST_INIT(&d->d_pipes, nni_pipe, p_ep_node); + + nni_mtx_init(&d->d_mtx); + + nni_aio_init(&d->d_con_aio, dialer_connect_cb, d); + nni_aio_init(&d->d_tmo_aio, dialer_timer_cb, d); + + nni_mtx_lock(&dialers_lk); + rv = nni_id_alloc32(&dialers, &d->d_id, d); + nni_mtx_unlock(&dialers_lk); + +#ifdef NNG_ENABLE_STATS + dialer_stats_init(d); +#endif + + if ((rv != 0) || + ((rv = d->d_ops.d_init(&d->d_data, &d->d_url, d)) != 0) || + ((rv = nni_sock_add_dialer(s, d)) != 0)) { + nni_mtx_lock(&dialers_lk); + nni_id_remove(&dialers, d->d_id); + nni_mtx_unlock(&dialers_lk); +#ifdef NNG_ENABLE_STATS + nni_stat_unregister(&d->st_root); +#endif + return (rv); + } + + return (0); +} + +int +nni_dialer_create_url(nni_dialer **dp, nni_sock *s, const nng_url *url) +{ + nni_sp_tran *tran; + nni_dialer *d; + int rv; + + if (((tran = nni_sp_tran_find(nng_url_scheme(url))) == NULL) || + (tran->tran_dialer == NULL)) { + return (NNG_ENOTSUP); + } + if ((d = NNI_ALLOC_STRUCT(d)) == NULL) { + return (NNG_ENOMEM); + } + if ((rv = nni_url_clone_inline(&d->d_url, url)) != 0) { + NNI_FREE_STRUCT(d); + return (rv); + } + if ((rv = nni_dialer_init(d, s, tran)) != 0) { + nni_dialer_destroy(d); + return (rv); + } + *dp = d; + return (0); +} + // nni_dialer_create creates a dialer on the socket. // The caller should have a hold on the socket, and on success // the dialer inherits the callers hold. (If the caller wants @@ -224,7 +298,6 @@ nni_dialer_create(nni_dialer **dp, nni_sock *s, const char *url_str) return (NNG_ENOMEM); } if ((rv = nni_url_parse_inline(&d->d_url, url_str)) != 0) { - nni_url_fini(&d->d_url); NNI_FREE_STRUCT(d); return (rv); } diff --git a/src/core/dialer.h b/src/core/dialer.h index 19b39cf2..f3fac7f1 100644 --- a/src/core/dialer.h +++ b/src/core/dialer.h @@ -12,13 +12,14 @@ #ifndef CORE_DIALER_H #define CORE_DIALER_H -extern int nni_dialer_find(nni_dialer **, uint32_t); -extern int nni_dialer_hold(nni_dialer *); -extern void nni_dialer_rele(nni_dialer *); -extern uint32_t nni_dialer_id(nni_dialer *); -extern int nni_dialer_create(nni_dialer **, nni_sock *, const char *); -extern void nni_dialer_close(nni_dialer *); -extern int nni_dialer_start(nni_dialer *, unsigned); +extern int nni_dialer_find(nni_dialer **, uint32_t); +extern int nni_dialer_hold(nni_dialer *); +extern void nni_dialer_rele(nni_dialer *); +extern uint32_t nni_dialer_id(nni_dialer *); +extern int nni_dialer_create(nni_dialer **, nni_sock *, const char *); +extern int nni_dialer_create_url(nni_dialer **, nni_sock *, const nng_url *); +extern void nni_dialer_close(nni_dialer *); +extern int nni_dialer_start(nni_dialer *, unsigned); extern nni_sock *nni_dialer_sock(nni_dialer *); extern int nni_dialer_setopt( diff --git a/src/core/listener.c b/src/core/listener.c index eedfbc42..8d884b5e 100644 --- a/src/core/listener.c +++ b/src/core/listener.c @@ -212,7 +212,6 @@ nni_listener_create(nni_listener **lp, nni_sock *s, const char *url_str) return (NNG_ENOMEM); } if ((rv = nni_url_parse_inline(&l->l_url, url_str)) != 0) { - nni_url_fini(&l->l_url); NNI_FREE_STRUCT(l); return (rv); } diff --git a/src/core/url.c b/src/core/url.c index 699734a7..530e299c 100644 --- a/src/core/url.c +++ b/src/core/url.c @@ -325,8 +325,8 @@ nni_url_default_port(const char *scheme) // Nanomsg URLs are always of the first form, we always require a // scheme with a leading //, such as http:// or tcp://. So our parser // is a bit more restricted, but sufficient for our needs. -int -nni_url_parse_inline(nng_url *url, const char *raw) +static int +nni_url_parse_inline_inner(nng_url *url, const char *raw) { size_t len; const char *s; @@ -508,6 +508,16 @@ nni_url_parse_inline(nng_url *url, const char *raw) } int +nni_url_parse_inline(nng_url *url, const char *raw) +{ + int rv = nni_url_parse_inline_inner(url, raw); + if (rv != 0) { + nni_url_fini(url); + } + return (rv); +} + +int nng_url_parse(nng_url **urlp, const char *raw) { nng_url *url; @@ -517,7 +527,7 @@ nng_url_parse(nng_url **urlp, const char *raw) return (NNG_ENOMEM); } if ((rv = nni_url_parse_inline(url, raw)) != 0) { - nng_url_free(url); + NNI_FREE_STRUCT(url); return (rv); } *urlp = url; @@ -529,6 +539,8 @@ nni_url_fini(nng_url *url) { if (url->u_bufsz != 0) { nni_free(url->u_buffer, url->u_bufsz); + url->u_buffer = NULL; + url->u_bufsz = 0; } } @@ -538,6 +538,33 @@ nng_dial(nng_socket sid, const char *addr, nng_dialer *dp, int flags) } int +nng_dial_url(nng_socket sid, const nng_url *url, nng_dialer *dp, int flags) +{ + nni_dialer *d; + int rv; + nni_sock *s; + + if ((rv = nni_sock_find(&s, sid.id)) != 0) { + return (rv); + } + if ((rv = nni_dialer_create_url(&d, s, url)) != 0) { + nni_sock_rele(s); + return (rv); + } + if ((rv = nni_dialer_start(d, flags)) != 0) { + nni_dialer_close(d); + return (rv); + } + if (dp != NULL) { + nng_dialer did; + did.id = nni_dialer_id(d); + *dp = did; + } + nni_dialer_rele(d); + return (0); +} + +int nng_listen(nng_socket sid, const char *addr, nng_listener *lp, int flags) { int rv; @@ -628,6 +655,27 @@ nng_dialer_create(nng_dialer *dp, nng_socket sid, const char *addr) } int +nng_dialer_create_url(nng_dialer *dp, nng_socket sid, const nng_url *url) +{ + nni_sock *s; + nni_dialer *d; + int rv; + nng_dialer did; + + if ((rv = nni_sock_find(&s, sid.id)) != 0) { + return (rv); + } + if ((rv = nni_dialer_create_url(&d, s, url)) != 0) { + nni_sock_rele(s); + return (rv); + } + did.id = nni_dialer_id(d); + *dp = did; + nni_dialer_rele(d); + return (0); +} + +int nng_dialer_start(nng_dialer did, int flags) { nni_dialer *d; diff --git a/src/sp/transport.c b/src/sp/transport.c index a2220e7b..919a0f37 100644 --- a/src/sp/transport.c +++ b/src/sp/transport.c @@ -41,7 +41,7 @@ nni_sp_tran_find(const char *url) NNI_LIST_FOREACH (&sp_tran_list, t) { size_t len = strlen(t->tran_scheme); if ((strncmp(url, t->tran_scheme, len) == 0) && - (strncmp(url + len, "://", 3) == 0)) { + (url[len] == ':' || url[len] == '\0')) { nni_rwlock_unlock(&sp_tran_lk); return (t); } diff --git a/src/sp/transport/tcp/tcp_test.c b/src/sp/transport/tcp/tcp_test.c index a95e48e7..82abcae1 100644 --- a/src/sp/transport/tcp/tcp_test.c +++ b/src/sp/transport/tcp/tcp_test.c @@ -172,10 +172,14 @@ test_tcp_keep_alive_option(void) bool v; int x; char *addr; + nng_url *url; NUTS_ADDR(addr, "tcp"); + // next cases are just to exercise nng_dialer_create_url + NUTS_PASS(nng_url_parse(&url, addr)); NUTS_OPEN(s); - NUTS_PASS(nng_dialer_create(&d, s, addr)); + NUTS_PASS(nng_dialer_create_url(&d, s, url)); + nng_url_free(url); NUTS_PASS(nng_dialer_get_bool(d, NNG_OPT_TCP_KEEPALIVE, &v)); NUTS_TRUE(v == false); NUTS_PASS(nng_dialer_set_bool(d, NNG_OPT_TCP_KEEPALIVE, true)); diff --git a/src/sp/transport/udp/udp_tran_test.c b/src/sp/transport/udp/udp_tran_test.c index 654df1e2..941b28e0 100644 --- a/src/sp/transport/udp/udp_tran_test.c +++ b/src/sp/transport/udp/udp_tran_test.c @@ -63,7 +63,6 @@ test_udp_port_zero_bind(void) nng_sockaddr sa; nng_listener l; nng_dialer d; - char addr[NNG_MAXADDRSTRLEN]; int port; const nng_url *u1; const nng_url *u2; @@ -77,12 +76,11 @@ test_udp_port_zero_bind(void) NUTS_MATCH(nng_url_scheme(u1), "udp"); NUTS_MATCH(nng_url_hostname(u1), "127.0.0.1"); NUTS_MATCH(nng_url_path(u1), ""); - nng_url_sprintf(addr, sizeof(addr), u1); NUTS_PASS(nng_listener_get_addr(l, NNG_OPT_LOCADDR, &sa)); NUTS_TRUE(sa.s_in.sa_family == NNG_AF_INET); NUTS_TRUE(sa.s_in.sa_port == nuts_be16(nng_url_port(u1))); NUTS_TRUE(sa.s_in.sa_addr == nuts_be32(0x7f000001)); - NUTS_PASS(nng_dial(s2, addr, &d, 0)); + NUTS_PASS(nng_dial_url(s2, u1, &d, 0)); NUTS_PASS(nng_dialer_get_url(d, &u2)); NUTS_MATCH(nng_url_scheme(u1), nng_url_scheme(u2)); NUTS_MATCH(nng_url_hostname(u1), nng_url_hostname(u2)); |