aboutsummaryrefslogtreecommitdiff
path: root/src/protocol/pipeline0
diff options
context:
space:
mode:
Diffstat (limited to 'src/protocol/pipeline0')
-rw-r--r--src/protocol/pipeline0/CMakeLists.txt23
-rw-r--r--src/protocol/pipeline0/pull.c250
-rw-r--r--src/protocol/pipeline0/pull.h28
-rw-r--r--src/protocol/pipeline0/push.c267
-rw-r--r--src/protocol/pipeline0/push.h28
5 files changed, 596 insertions, 0 deletions
diff --git a/src/protocol/pipeline0/CMakeLists.txt b/src/protocol/pipeline0/CMakeLists.txt
new file mode 100644
index 00000000..6153c5a7
--- /dev/null
+++ b/src/protocol/pipeline0/CMakeLists.txt
@@ -0,0 +1,23 @@
+#
+# 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
+# file was obtained (LICENSE.txt). A copy of the license may also be
+# found online at https://opensource.org/licenses/MIT.
+#
+
+# Pub/Sub protocol
+
+if (NNG_PROTO_PUSH0)
+ set(PUSH0_SOURCES protocol/pipeline0/push.c protocol/pipeline0/push.h)
+ install(FILES push.h DESTINATION include/nng/protocol/pipeline0)
+endif()
+
+if (NNG_PROTO_PULL0)
+ set(PULL0_SOURCES protocol/pipeline0/pull.c protocol/pipeline0/pull.h)
+ install(FILES pull.h DESTINATION include/nng/protocol/pipeline0)
+endif()
+
+set(NNG_SOURCES ${NNG_SOURCES} ${PUSH0_SOURCES} ${PULL0_SOURCES} PARENT_SCOPE) \ No newline at end of file
diff --git a/src/protocol/pipeline0/pull.c b/src/protocol/pipeline0/pull.c
new file mode 100644
index 00000000..8c16cb17
--- /dev/null
+++ b/src/protocol/pipeline0/pull.c
@@ -0,0 +1,250 @@
+//
+// 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
+// file was obtained (LICENSE.txt). A copy of the license may also be
+// found online at https://opensource.org/licenses/MIT.
+//
+
+#include <stdlib.h>
+#include <string.h>
+
+#include "core/nng_impl.h"
+
+// Pull protocol. The PULL protocol is the "read" side of a pipeline.
+
+#ifndef NNI_PROTO_PULL_V0
+#define NNI_PROTO_PULL_V0 NNI_PROTO(5, 1)
+#endif
+
+#ifndef NNI_PROTO_PUSH_V0
+#define NNI_PROTO_PUSH_V0 NNI_PROTO(5, 0)
+#endif
+
+typedef struct pull0_pipe pull0_pipe;
+typedef struct pull0_sock pull0_sock;
+
+static void pull0_putq_cb(void *);
+static void pull0_recv_cb(void *);
+static void pull0_putq(pull0_pipe *, nni_msg *);
+
+// pull0_sock is our per-socket protocol private structure.
+struct pull0_sock {
+ nni_msgq *urq;
+ int raw;
+};
+
+// pull0_pipe is our per-pipe protocol private structure.
+struct pull0_pipe {
+ nni_pipe * pipe;
+ pull0_sock *pull;
+ nni_aio * putq_aio;
+ nni_aio * recv_aio;
+};
+
+static int
+pull0_sock_init(void **sp, nni_sock *sock)
+{
+ pull0_sock *s;
+
+ if ((s = NNI_ALLOC_STRUCT(s)) == NULL) {
+ return (NNG_ENOMEM);
+ }
+ s->raw = 0;
+ s->urq = nni_sock_recvq(sock);
+
+ *sp = s;
+ return (0);
+}
+
+static void
+pull0_sock_fini(void *arg)
+{
+ pull0_sock *s = arg;
+
+ NNI_FREE_STRUCT(s);
+}
+
+static void
+pull0_pipe_fini(void *arg)
+{
+ pull0_pipe *p = arg;
+
+ nni_aio_fini(p->putq_aio);
+ nni_aio_fini(p->recv_aio);
+ NNI_FREE_STRUCT(p);
+}
+
+static int
+pull0_pipe_init(void **pp, nni_pipe *pipe, void *s)
+{
+ pull0_pipe *p;
+ int rv;
+
+ if ((p = NNI_ALLOC_STRUCT(p)) == NULL) {
+ return (NNG_ENOMEM);
+ }
+ if (((rv = nni_aio_init(&p->putq_aio, pull0_putq_cb, p)) != 0) ||
+ ((rv = nni_aio_init(&p->recv_aio, pull0_recv_cb, p)) != 0)) {
+ pull0_pipe_fini(p);
+ return (rv);
+ }
+
+ p->pipe = pipe;
+ p->pull = s;
+ *pp = p;
+ return (0);
+}
+
+static int
+pull0_pipe_start(void *arg)
+{
+ pull0_pipe *p = arg;
+
+ // Start the pending pull...
+ nni_pipe_recv(p->pipe, p->recv_aio);
+
+ return (0);
+}
+
+static void
+pull0_pipe_stop(void *arg)
+{
+ pull0_pipe *p = arg;
+
+ nni_aio_stop(p->putq_aio);
+ nni_aio_stop(p->recv_aio);
+}
+
+static void
+pull0_recv_cb(void *arg)
+{
+ pull0_pipe *p = arg;
+ nni_aio * aio = p->recv_aio;
+ nni_msg * msg;
+
+ if (nni_aio_result(aio) != 0) {
+ // Failed to get a message, probably the pipe is closed.
+ nni_pipe_stop(p->pipe);
+ return;
+ }
+
+ // Got a message... start the put to send it up to the application.
+ msg = nni_aio_get_msg(aio);
+ nni_msg_set_pipe(msg, nni_pipe_id(p->pipe));
+ nni_aio_set_msg(aio, NULL);
+ pull0_putq(p, msg);
+}
+
+static void
+pull0_putq_cb(void *arg)
+{
+ pull0_pipe *p = arg;
+ nni_aio * aio = p->putq_aio;
+
+ if (nni_aio_result(aio) != 0) {
+ // If we failed to put, probably NNG_ECLOSED, nothing else
+ // we can do. Just close the pipe.
+ nni_msg_free(nni_aio_get_msg(aio));
+ nni_aio_set_msg(aio, NULL);
+ nni_pipe_stop(p->pipe);
+ return;
+ }
+
+ nni_pipe_recv(p->pipe, p->recv_aio);
+}
+
+// pull0_putq schedules a put operation to the user socket (sendup).
+static void
+pull0_putq(pull0_pipe *p, nni_msg *msg)
+{
+ pull0_sock *s = p->pull;
+
+ nni_aio_set_msg(p->putq_aio, msg);
+
+ nni_msgq_aio_put(s->urq, p->putq_aio);
+}
+
+static void
+pull0_sock_open(void *arg)
+{
+ NNI_ARG_UNUSED(arg);
+}
+
+static void
+pull0_sock_close(void *arg)
+{
+ NNI_ARG_UNUSED(arg);
+}
+
+static int
+pull0_sock_setopt_raw(void *arg, const void *buf, size_t sz)
+{
+ pull0_sock *s = arg;
+ return (nni_setopt_int(&s->raw, buf, sz, 0, 1));
+}
+
+static int
+pull0_sock_getopt_raw(void *arg, void *buf, size_t *szp)
+{
+ pull0_sock *s = arg;
+ return (nni_getopt_int(s->raw, buf, szp));
+}
+
+static void
+pull0_sock_send(void *arg, nni_aio *aio)
+{
+ nni_aio_finish_error(aio, NNG_ENOTSUP);
+}
+
+static void
+pull0_sock_recv(void *arg, nni_aio *aio)
+{
+ pull0_sock *s = arg;
+
+ nni_msgq_aio_get(s->urq, aio);
+}
+
+static nni_proto_pipe_ops pull0_pipe_ops = {
+ .pipe_init = pull0_pipe_init,
+ .pipe_fini = pull0_pipe_fini,
+ .pipe_start = pull0_pipe_start,
+ .pipe_stop = pull0_pipe_stop,
+};
+
+static nni_proto_sock_option pull0_sock_options[] = {
+ {
+ .pso_name = NNG_OPT_RAW,
+ .pso_getopt = pull0_sock_getopt_raw,
+ .pso_setopt = pull0_sock_setopt_raw,
+ },
+ // terminate list
+ { NULL, NULL, NULL },
+};
+
+static nni_proto_sock_ops pull0_sock_ops = {
+ .sock_init = pull0_sock_init,
+ .sock_fini = pull0_sock_fini,
+ .sock_open = pull0_sock_open,
+ .sock_close = pull0_sock_close,
+ .sock_send = pull0_sock_send,
+ .sock_recv = pull0_sock_recv,
+ .sock_options = pull0_sock_options,
+};
+
+static nni_proto pull0_proto = {
+ .proto_version = NNI_PROTOCOL_VERSION,
+ .proto_self = { NNI_PROTO_PULL_V0, "pull" },
+ .proto_peer = { NNI_PROTO_PUSH_V0, "push" },
+ .proto_flags = NNI_PROTO_FLAG_RCV,
+ .proto_pipe_ops = &pull0_pipe_ops,
+ .proto_sock_ops = &pull0_sock_ops,
+};
+
+int
+nng_pull0_open(nng_socket *sidp)
+{
+ return (nni_proto_open(sidp, &pull0_proto));
+}
diff --git a/src/protocol/pipeline0/pull.h b/src/protocol/pipeline0/pull.h
new file mode 100644
index 00000000..75bded03
--- /dev/null
+++ b/src/protocol/pipeline0/pull.h
@@ -0,0 +1,28 @@
+//
+// 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
+// file was obtained (LICENSE.txt). A copy of the license may also be
+// found online at https://opensource.org/licenses/MIT.
+//
+
+#ifndef NNG_PROTOCOL_PIPELINE0_PULL_H
+#define NNG_PROTOCOL_PIPELINE0_PULL_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+NNG_DECL int nng_pull0_open(nng_socket *);
+
+#ifndef nng_pull_open
+#define nng_pull_open nng_pull0_open
+#endif
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif // NNG_PROTOCOL_PIPELINE0_PULL_H
diff --git a/src/protocol/pipeline0/push.c b/src/protocol/pipeline0/push.c
new file mode 100644
index 00000000..3dd83fe0
--- /dev/null
+++ b/src/protocol/pipeline0/push.c
@@ -0,0 +1,267 @@
+//
+// 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
+// file was obtained (LICENSE.txt). A copy of the license may also be
+// found online at https://opensource.org/licenses/MIT.
+//
+
+#include <stdlib.h>
+#include <string.h>
+
+#include "core/nng_impl.h"
+
+// Push protocol. The PUSH protocol is the "write" side of a pipeline.
+// Push distributes fairly, or tries to, by giving messages in round-robin
+// order.
+
+#ifndef NNI_PROTO_PULL_V0
+#define NNI_PROTO_PULL_V0 NNI_PROTO(5, 1)
+#endif
+
+#ifndef NNI_PROTO_PUSH_V0
+#define NNI_PROTO_PUSH_V0 NNI_PROTO(5, 0)
+#endif
+
+typedef struct push0_pipe push0_pipe;
+typedef struct push0_sock push0_sock;
+
+static void push0_send_cb(void *);
+static void push0_recv_cb(void *);
+static void push0_getq_cb(void *);
+
+// push0_sock is our per-socket protocol private structure.
+struct push0_sock {
+ nni_msgq *uwq;
+ int raw;
+};
+
+// push0_pipe is our per-pipe protocol private structure.
+struct push0_pipe {
+ nni_pipe * pipe;
+ push0_sock * push;
+ nni_list_node node;
+
+ nni_aio *aio_recv;
+ nni_aio *aio_send;
+ nni_aio *aio_getq;
+};
+
+static int
+push0_sock_init(void **sp, nni_sock *sock)
+{
+ push0_sock *s;
+
+ if ((s = NNI_ALLOC_STRUCT(s)) == NULL) {
+ return (NNG_ENOMEM);
+ }
+ s->raw = 0;
+ s->uwq = nni_sock_sendq(sock);
+ *sp = s;
+ return (0);
+}
+
+static void
+push0_sock_fini(void *arg)
+{
+ push0_sock *s = arg;
+
+ NNI_FREE_STRUCT(s);
+}
+
+static void
+push0_sock_open(void *arg)
+{
+ NNI_ARG_UNUSED(arg);
+}
+
+static void
+push0_sock_close(void *arg)
+{
+ NNI_ARG_UNUSED(arg);
+}
+
+static void
+push0_pipe_fini(void *arg)
+{
+ push0_pipe *p = arg;
+
+ nni_aio_fini(p->aio_recv);
+ nni_aio_fini(p->aio_send);
+ nni_aio_fini(p->aio_getq);
+ NNI_FREE_STRUCT(p);
+}
+
+static int
+push0_pipe_init(void **pp, nni_pipe *pipe, void *s)
+{
+ push0_pipe *p;
+ int rv;
+
+ if ((p = NNI_ALLOC_STRUCT(p)) == NULL) {
+ return (NNG_ENOMEM);
+ }
+ if (((rv = nni_aio_init(&p->aio_recv, push0_recv_cb, p)) != 0) ||
+ ((rv = nni_aio_init(&p->aio_send, push0_send_cb, p)) != 0) ||
+ ((rv = nni_aio_init(&p->aio_getq, push0_getq_cb, p)) != 0)) {
+ push0_pipe_fini(p);
+ return (rv);
+ }
+ NNI_LIST_NODE_INIT(&p->node);
+ p->pipe = pipe;
+ p->push = s;
+ *pp = p;
+ return (0);
+}
+
+static int
+push0_pipe_start(void *arg)
+{
+ push0_pipe *p = arg;
+ push0_sock *s = p->push;
+
+ if (nni_pipe_peer(p->pipe) != NNI_PROTO_PULL_V0) {
+ return (NNG_EPROTO);
+ }
+
+ // Schedule a receiver. This is mostly so that we can detect
+ // a closed transport pipe.
+ nni_pipe_recv(p->pipe, p->aio_recv);
+
+ // Schedule a sender.
+ nni_msgq_aio_get(s->uwq, p->aio_getq);
+
+ return (0);
+}
+
+static void
+push0_pipe_stop(void *arg)
+{
+ push0_pipe *p = arg;
+
+ nni_aio_stop(p->aio_recv);
+ nni_aio_stop(p->aio_send);
+ nni_aio_stop(p->aio_getq);
+}
+
+static void
+push0_recv_cb(void *arg)
+{
+ push0_pipe *p = arg;
+
+ // We normally expect to receive an error. If a pipe actually
+ // sends us data, we just discard it.
+ if (nni_aio_result(p->aio_recv) != 0) {
+ nni_pipe_stop(p->pipe);
+ return;
+ }
+ nni_msg_free(nni_aio_get_msg(p->aio_recv));
+ nni_aio_set_msg(p->aio_recv, NULL);
+ nni_pipe_recv(p->pipe, p->aio_recv);
+}
+
+static void
+push0_send_cb(void *arg)
+{
+ push0_pipe *p = arg;
+ push0_sock *s = p->push;
+
+ if (nni_aio_result(p->aio_send) != 0) {
+ nni_msg_free(nni_aio_get_msg(p->aio_send));
+ nni_aio_set_msg(p->aio_send, NULL);
+ nni_pipe_stop(p->pipe);
+ return;
+ }
+
+ nni_msgq_aio_get(s->uwq, p->aio_getq);
+}
+
+static void
+push0_getq_cb(void *arg)
+{
+ push0_pipe *p = arg;
+ nni_aio * aio = p->aio_getq;
+
+ if (nni_aio_result(aio) != 0) {
+ // If the socket is closing, nothing else we can do.
+ nni_pipe_stop(p->pipe);
+ return;
+ }
+
+ nni_aio_set_msg(p->aio_send, nni_aio_get_msg(aio));
+ nni_aio_set_msg(aio, NULL);
+
+ nni_pipe_send(p->pipe, p->aio_send);
+}
+
+static int
+push0_sock_setopt_raw(void *arg, const void *buf, size_t sz)
+{
+ push0_sock *s = arg;
+ return (nni_setopt_int(&s->raw, buf, sz, 0, 1));
+}
+
+static int
+push0_sock_getopt_raw(void *arg, void *buf, size_t *szp)
+{
+ push0_sock *s = arg;
+ return (nni_getopt_int(s->raw, buf, szp));
+}
+
+static void
+push0_sock_send(void *arg, nni_aio *aio)
+{
+ push0_sock *s = arg;
+
+ nni_msgq_aio_put(s->uwq, aio);
+}
+
+static void
+push0_sock_recv(void *arg, nni_aio *aio)
+{
+ nni_aio_finish_error(aio, NNG_ENOTSUP);
+}
+
+static nni_proto_pipe_ops push0_pipe_ops = {
+ .pipe_init = push0_pipe_init,
+ .pipe_fini = push0_pipe_fini,
+ .pipe_start = push0_pipe_start,
+ .pipe_stop = push0_pipe_stop,
+};
+
+static nni_proto_sock_option push0_sock_options[] = {
+ {
+ .pso_name = NNG_OPT_RAW,
+ .pso_getopt = push0_sock_getopt_raw,
+ .pso_setopt = push0_sock_setopt_raw,
+ },
+ // terminate list
+ { NULL, NULL, NULL },
+};
+
+static nni_proto_sock_ops push0_sock_ops = {
+ .sock_init = push0_sock_init,
+ .sock_fini = push0_sock_fini,
+ .sock_open = push0_sock_open,
+ .sock_close = push0_sock_close,
+ .sock_options = push0_sock_options,
+ .sock_send = push0_sock_send,
+ .sock_recv = push0_sock_recv,
+};
+
+static nni_proto push0_proto = {
+ .proto_version = NNI_PROTOCOL_VERSION,
+ .proto_self = { NNI_PROTO_PUSH_V0, "push" },
+ .proto_peer = { NNI_PROTO_PULL_V0, "pull" },
+ .proto_flags = NNI_PROTO_FLAG_SND,
+ .proto_pipe_ops = &push0_pipe_ops,
+ .proto_sock_ops = &push0_sock_ops,
+};
+
+int
+nng_push0_open(nng_socket *sidp)
+{
+ return (nni_proto_open(sidp, &push0_proto));
+}
diff --git a/src/protocol/pipeline0/push.h b/src/protocol/pipeline0/push.h
new file mode 100644
index 00000000..c7303b92
--- /dev/null
+++ b/src/protocol/pipeline0/push.h
@@ -0,0 +1,28 @@
+//
+// 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
+// file was obtained (LICENSE.txt). A copy of the license may also be
+// found online at https://opensource.org/licenses/MIT.
+//
+
+#ifndef NNG_PROTOCOL_PIPELINE0_PUSH_H
+#define NNG_PROTOCOL_PIPELINE0_PUSH_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+NNG_DECL int nng_push0_open(nng_socket *);
+
+#ifndef nng_push_open
+#define nng_push_open nng_push0_open
+#endif
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif // NNG_PROTOCOL_PIPELINE0_PUSH_H
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2026-01-20 01:45:12 +0000