Author： huang.zhenling

Root Cause: 提交nanomsg demo和libev demo
How2Fix:

common/nanomsg/iot.mk

+LIBA_TARGET 	:= libnanomsg.a

example/ev/iot.mk

+LIBA_TARGET 		:= libev_fnc.a
+
+SRCS_example 		:= server.c
+
+$(call Append_Conditional, LIB_SRCS_EXCLUDE, server.c)
+$(call Append_Conditional, SRCS_ev-demo, server.c)
+$(call Append_Conditional, TARGET, ev-demo)
+
+#CFLAGS 			+= -I../../work/ssl/include -lrt
+#LDFLAGS 		+= -L../../work/ssl/lib
+#DEPENDS         += common/mqtt
+LDFLAGS         += -L$(TOP_DIR)/example/ev -static -lev -ldl
+
+
+

example/ev/server.c

+#include <stdio.h>
+#include "ev.h"
+
+ev_io stdin_watcher;
+ev_timer timeout_watcher;
+ 
+static void  stdin_cb(EV_P_ ev_io *w, int revents)
+{
+    puts("stdin ready");
+    ev_io_stop(EV_A_ w);
+    ev_break(EV_A_ EVBREAK_ALL);
+}
+
+static void timeout_cb(EV_P_ ev_timer *w, int revnts)
+{
+    puts("time out");
+    ev_break(EV_A_ EVBREAK_ONE);
+}
+ 
+
+int  main(void)
+{
+    struct ev_loop *loop = EV_DEFAULT;
+ 
+    ev_io_init(&stdin_watcher, stdin_cb, 0, EV_READ);
+    ev_io_start(loop, &stdin_watcher);
+ 
+    ev_timer_init(&timeout_watcher, timeout_cb, 10, 0);
+    ev_timer_start(loop, &timeout_watcher);
+ 
+    ev_run(loop, 0);
+ 
+    return 0;
+}

example/ev/说明.txt

+版本：libev-4.33
+发布包：http://dist.schmorp.de/libev/
+API：http://pod.tst.eu/http://cvs.schmorp.de/libev/ev.pod
+Csdn学习：https://blog.csdn.net/weixin_39510813/article/details/90711184
+Demo：https://www.cnblogs.com/wunaozai/p/3950249.html
+

example/ev_nanomsg/ev_nanomsg.c

+#include <stdio.h>
+#include <stdint.h>
+#include <unistd.h>
+#include <string.h>
+#include <time.h>
+#include "ev.h"
+#include <pthread.h>
+#include <nanomsg/pair.h>
+#include <nanomsg/nn.h>
+ 
+ 
+/***************************************************************************************************
+动态库：libev nanomsg
+概述：A B C 三个线程通过nanomsg通信，A线程作为主线程，控制中枢，B C请求均通过A.
+demo示范：
+    A为指令处理模块
+    B为指令接收模块
+    C为指令执行模块
+    B -> A   开灯
+    A -> C   开灯
+    C :      执行开灯
+    C -> A   OK
+    A -> B   OK
+总结：
+    这只是简单的测试使用例子，你可以通过在这个框架的基础上做更多的功能，对于多线程编程这将是一个不
+错的选择.
+***************************************************************************************************/
+ 
+typedef struct {
+    int n;          //nanomsg socket
+    int s;          //nanomsg recieve fd
+}nanomsg_info_t;
+ 
+typedef struct {
+    nanomsg_info_t ab;
+    nanomsg_info_t ac;
+}Aloop_ctrl_t;
+ 
+typedef struct {
+    nanomsg_info_t ba;
+}Bloop_ctrl_t;
+ 
+typedef struct {
+    nanomsg_info_t ca;
+}Cloop_ctrl_t;
+ 
+/*获取系统时间打印*/
+uint32_t print_timenow()
+{
+    time_t now;
+    struct tm *tm_now;
+    time(&now);
+    tm_now = localtime(&now);
+    uint32_t times = tm_now->tm_hour * 3600 + tm_now->tm_min * 60 + tm_now->tm_sec;
+    printf("[%02d:%02d:%02d]\r\n", tm_now->tm_hour, tm_now->tm_min, tm_now->tm_sec);
+    return times;
+}
+ 
+/*****************************************子线程C相关**********************************************/
+ 
+static void watcher_c_cb (struct ev_loop *loop ,struct ev_io *w, int revents)
+{
+    void *user_data = ev_userdata(loop);
+    Cloop_ctrl_t *Cloop_ctrl = (Cloop_ctrl_t *)user_data;
+    uint8_t *dat = NULL;
+    uint32_t bytes = nn_recv(Cloop_ctrl->ca.n, &dat, NN_MSG, NN_DONTWAIT);
+    if (bytes <= 0) {
+        return;
+    }
+    printf("C:%s (A->C)\r\n", (char *)dat);
+    nn_freemsg(dat);
+    //接收成功，发送OK
+    char *str = "OK";
+    uint8_t *udata = nn_allocmsg(3, 0);
+    if (NULL != udata) {
+        memcpy(udata, str, 3);
+        nn_send(Cloop_ctrl->ca.n, &udata, NN_MSG, NN_DONTWAIT);
+    }
+    
+}
+ 
+int C_nanomsg_init(Cloop_ctrl_t *Cloop_ctrl)
+{
+    Cloop_ctrl->ca.n = nn_socket(AF_SP, NN_PAIR);
+    if (Cloop_ctrl->ca.n < 0) {
+        return -1;
+    }
+    if (nn_connect(Cloop_ctrl->ca.n, "inproc://c2a_loop") < 0) {
+        return -1;
+    }
+    size_t size = sizeof(size_t);
+    if (nn_getsockopt(Cloop_ctrl->ca.n, NN_SOL_SOCKET, NN_RCVFD, (char *)&Cloop_ctrl->ca.s, &size) < 0) {
+        return -1;
+    }
+    return 0;
+}
+ 
+struct ev_loop* C_loop_init(Cloop_ctrl_t *Cloop_ctrl)
+{
+    static struct ev_io watcher_c;
+    struct ev_loop *loop = ev_loop_new(EVBACKEND_EPOLL);
+    if (NULL == loop) {
+        printf("create C loop failed\r\n");
+        return NULL;
+    }
+    ev_io_init (&watcher_c, watcher_c_cb, Cloop_ctrl->ca.s, EV_READ);
+    ev_io_start (loop, &watcher_c);
+    return loop;
+}
+ 
+void *C_thread(void *arg)
+{
+    Cloop_ctrl_t Cloop_ctrl;
+    if (C_nanomsg_init(&Cloop_ctrl) < 0) {
+        printf("nanomsg init failed\r\n");
+        return ;
+    }
+    struct ev_loop* Cloop = C_loop_init(&Cloop_ctrl);
+    if (NULL == Cloop) {
+        printf("Cloop init failed\r\n");
+        return ;
+    }
+    ev_set_userdata(Cloop, &Cloop_ctrl);
+    ev_run (Cloop, 0);
+    return ;
+}
+ 
+/*****************************************子线程B相关**********************************************/
+ 
+static void watcher_b_cb (struct ev_loop *loop ,struct ev_io *w, int revents)
+{
+    void *user_data = ev_userdata(loop);
+    Bloop_ctrl_t *Bloop_ctrl = (Bloop_ctrl_t *)user_data;
+    uint8_t *dat = NULL;
+    uint32_t bytes = nn_recv(Bloop_ctrl->ba.n, &dat, NN_MSG, NN_DONTWAIT);
+    if (bytes <= 0) {
+        return;
+    }
+    printf("B:%s (A->B)\r\n\r\n", (char *)dat);
+    nn_freemsg(dat);
+}
+ 
+static void watcher_timer_cb (struct ev_loop *loop ,struct ev_timer *w, int revents)
+{
+    static int i = 1;
+    char send_data[128] = {0};
+    void *user_data = ev_userdata(loop);
+    Bloop_ctrl_t *Bloop_ctrl = (Bloop_ctrl_t *)user_data;
+    sprintf(send_data, "Please turn on LED[%d]", i);
+    i ++;
+    int length = strlen(send_data) + 1;
+    uint8_t *udata = nn_allocmsg(length, 0);
+    if (NULL != udata) {
+        memcpy(udata, send_data, length);
+        nn_send(Bloop_ctrl->ba.n, &udata, NN_MSG, NN_DONTWAIT);
+    }
+    //如果定时器不重设，就会默认1秒进入一次回调
+    w->repeat = 10;
+    ev_timer_again(loop, w);
+}
+ 
+ 
+int B_nanomsg_init(Bloop_ctrl_t *Bloop_ctrl)
+{
+    Bloop_ctrl->ba.n = nn_socket(AF_SP, NN_PAIR);
+    if (Bloop_ctrl->ba.n < 0) {
+        return -1;
+    }
+    if (nn_connect(Bloop_ctrl->ba.n, "inproc://b2a_loop") < 0) {
+        return -1;
+    }
+    size_t size = sizeof(size_t);
+    if (nn_getsockopt(Bloop_ctrl->ba.n, NN_SOL_SOCKET, NN_RCVFD, (char *)&Bloop_ctrl->ba.s, &size) < 0) {
+        return -1;
+    }
+    return 0;
+}
+ 
+struct ev_loop* B_loop_init(Bloop_ctrl_t *Bloop_ctrl)
+{
+    static struct ev_io watcher_b;
+    static struct ev_timer watcher_timer;
+    struct ev_loop *loop = ev_loop_new(EVBACKEND_EPOLL);
+    if (NULL == loop) {
+        printf("create loop failed\r\n");
+        return NULL;
+    }
+    ev_io_init (&watcher_b, watcher_b_cb, Bloop_ctrl->ba.s, EV_READ);
+    ev_timer_init(&watcher_timer, watcher_timer_cb, 5, 1);
+    ev_io_start (loop, &watcher_b);
+    ev_timer_start (loop, &watcher_timer);
+    return loop;
+}
+ 
+void *B_thread(void *arg)
+{
+    Bloop_ctrl_t Bloop_ctrl;
+    if (B_nanomsg_init(&Bloop_ctrl) < 0) {
+        printf("nanomsg init failed\r\n");
+        return ;
+    }
+    struct ev_loop* Bloop = B_loop_init(&Bloop_ctrl);
+    if (NULL == Bloop) {
+        printf("Bloop init failed\r\n");
+        return ;
+    }
+    ev_set_userdata(Bloop, &Bloop_ctrl);
+    ev_run (Bloop, 0);
+    return ;
+}
+ 
+/*****************************************主线程A相关**********************************************/
+ 
+static void watcher_ab_cb (struct ev_loop *loop ,struct ev_io *w, int revents)
+{
+    void *user_data = ev_userdata(loop);
+    Aloop_ctrl_t *Aloop_ctrl = (Aloop_ctrl_t *)user_data;
+    uint8_t *dat = NULL;
+    uint32_t bytes = nn_recv(Aloop_ctrl->ab.n, &dat, NN_MSG, NN_DONTWAIT);
+    if (bytes <= 0) {
+        return;
+    }
+    //转发到C
+    printf("A:%s (B->A)\r\n", (char *)dat);
+    nn_send(Aloop_ctrl->ac.n, &dat, NN_MSG, NN_DONTWAIT);
+}
+ 
+static void watcher_ac_cb (struct ev_loop *loop ,struct ev_io *w, int revents)
+{
+    void *user_data = ev_userdata(loop);
+    Aloop_ctrl_t *Aloop_ctrl = (Aloop_ctrl_t *)user_data;
+    uint8_t *dat = NULL;
+    uint32_t bytes = nn_recv(Aloop_ctrl->ac.n, &dat, NN_MSG, NN_DONTWAIT);
+    if (bytes <= 0) {
+        return;
+    }
+    //转发到B
+    printf("A:%s (C->A)\r\n", (char *)dat);
+    nn_send(Aloop_ctrl->ab.n, &dat, NN_MSG, NN_DONTWAIT);
+}
+ 
+/*主事件nanomsg初始化*/
+int A_nanomsg_init(Aloop_ctrl_t *Aloop_ctrl)
+{
+    //ab通信的nanomsg初始化
+    Aloop_ctrl->ab.n = nn_socket(AF_SP, NN_PAIR);
+    if (Aloop_ctrl->ab.n < 0) {
+        return -1;
+    }
+    if (nn_bind(Aloop_ctrl->ab.n, "inproc://b2a_loop") < 0) {
+        return -1;
+    }
+    //获取此端口的接收数据fd描述符
+    size_t size = sizeof(size_t);
+    if (nn_getsockopt(Aloop_ctrl->ab.n, NN_SOL_SOCKET, NN_RCVFD, (char *)&Aloop_ctrl->ab.s, &size) < 0) {
+        return -1;
+    }
+    //ac通信的nanomsg初始化
+    Aloop_ctrl->ac.n = nn_socket(AF_SP, NN_PAIR);
+    if (Aloop_ctrl->ac.n < 0) {
+        return -1;
+    }
+    if (nn_bind(Aloop_ctrl->ac.n, "inproc://c2a_loop") < 0) {
+        return -1;
+    }
+    //获取此端口的接收数据fd描述符
+    if (nn_getsockopt(Aloop_ctrl->ac.n, NN_SOL_SOCKET, NN_RCVFD, (char *)&Aloop_ctrl->ac.s, &size) < 0) {
+        return -1;
+    }
+    return 0;
+}
+ 
+/*主事件循环初始化*/
+struct ev_loop* A_loop_init(Aloop_ctrl_t *Aloop_ctrl)
+{
+    static struct ev_io watcher_ab, watcher_ac;
+    struct ev_loop *loop = ev_loop_new(EVBACKEND_EPOLL);
+    if (NULL == loop) {
+        printf("create loop failed\r\n");
+        return NULL;
+    }
+    //传参
+    ev_set_userdata(loop, Aloop_ctrl);
+    //初始化
+    ev_io_init (&watcher_ab, watcher_ab_cb, Aloop_ctrl->ab.s, EV_READ);
+    ev_io_init (&watcher_ac, watcher_ac_cb, Aloop_ctrl->ac.s, EV_READ);
+    ev_io_start (loop, &watcher_ab);
+    ev_io_start (loop, &watcher_ac);
+    return loop;
+}
+ 
+/**************************************************************************************************/
+ 
+int main()
+{
+    pthread_t pb,pc;
+    Aloop_ctrl_t Aloop_ctrl;
+    if (A_nanomsg_init(&Aloop_ctrl) < 0) {
+        printf("nanomsg init failed\r\n");
+        return -1;
+    }
+    struct ev_loop* Aloop = A_loop_init(&Aloop_ctrl);
+    if (NULL == Aloop) {
+        printf("Aloop init failed\r\n");
+        return -1;
+    }
+    //创建线程B
+    if (0 != pthread_create(&pb, NULL, B_thread, NULL)) {
+        printf("create pthread B failed\r\n");
+        return -1;
+    }
+    //创建线程C
+    if (0 != pthread_create(&pc, NULL, C_thread, NULL)) {
+        printf("create pthread C failed\r\n");
+        return -1;
+    }
+    //运行
+    ev_run(Aloop, 0);
+    return 0;
+}
\ No newline at end of file

example/ev_nanomsg/iot.mk

+LIBA_TARGET 		:= libev_nano_fnc.a
+
+SRCS_example 		:= server.c
+
+$(call Append_Conditional, LIB_SRCS_EXCLUDE, ev_nanomsg.c)
+$(call Append_Conditional, SRCS_ev-nano-demo, ev_nanomsg.c)
+$(call Append_Conditional, TARGET, ev-nano-demo)
+
+#CFLAGS 			+= -I../../work/ssl/include -lrt
+#LDFLAGS 		+= -L../../work/ssl/lib
+#DEPENDS         += common/mqtt
+LDFLAGS         += -L$(TOP_DIR)/example/ev_nanomsg -static -lev -lnanomsg -lanl -ldl
+
+
+

example/ev_nanomsg/nanomsg/bus.h

+/*
+    Copyright (c) 2013 Martin Sustrik  All rights reserved.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef BUS_H_INCLUDED
+#define BUS_H_INCLUDED
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_PROTO_BUS 7
+
+#define NN_BUS (NN_PROTO_BUS * 16 + 0)
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/ev_nanomsg/nanomsg/inproc.h

+/*
+    Copyright (c) 2012 Martin Sustrik  All rights reserved.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef INPROC_H_INCLUDED
+#define INPROC_H_INCLUDED
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_INPROC -1
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/ev_nanomsg/nanomsg/ipc.h

+/*
+    Copyright (c) 2012 Martin Sustrik  All rights reserved.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef IPC_H_INCLUDED
+#define IPC_H_INCLUDED
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_IPC -2
+
+/* The object set here must be valid as long as you are using the socket */
+#define NN_IPC_SEC_ATTR 1
+#define NN_IPC_OUTBUFSZ 2
+#define NN_IPC_INBUFSZ 3
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/ev_nanomsg/nanomsg/pair.h

+/*
+    Copyright (c) 2012 Martin Sustrik  All rights reserved.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef PAIR_H_INCLUDED
+#define PAIR_H_INCLUDED
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_PROTO_PAIR 1
+
+#define NN_PAIR (NN_PROTO_PAIR * 16 + 0)
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/ev_nanomsg/nanomsg/pipeline.h

+/*
+    Copyright (c) 2012 Martin Sustrik  All rights reserved.
+    Copyright (c) 2013 GoPivotal, Inc.  All rights reserved.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef PIPELINE_H_INCLUDED
+#define PIPELINE_H_INCLUDED
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_PROTO_PIPELINE 5
+
+#define NN_PUSH (NN_PROTO_PIPELINE * 16 + 0)
+#define NN_PULL (NN_PROTO_PIPELINE * 16 + 1)
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/ev_nanomsg/nanomsg/protocol.h

+/*
+    Copyright (c) 2012-2013 Martin Sustrik  All rights reserved.
+    Copyright 2016 Garrett D'Amore <garrett@damore.org>
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef NN_PROTOCOL_INCLUDED
+#define NN_PROTOCOL_INCLUDED
+
+#include "utils/msg.h"
+#include "utils/list.h"
+
+#include <stddef.h>
+#include <stdint.h>
+
+struct nn_ctx;
+
+/******************************************************************************/
+/*  Pipe class.                                                               */
+/******************************************************************************/
+
+/*  Any combination of following flags can be returned from successful call
+    to nn_pipe_send or nn_pipe_recv. */
+
+/*  This flag means that the pipe can't be used for receiving (when returned
+    from nn_pipe_recv()) or sending (when returned from nn_pipe_send()).
+    Protocol implementation should not send/recv messages from the pipe until
+    the pipe is revived by in()/out() function. */
+#define NN_PIPE_RELEASE 1
+
+/*  Specifies that received message is already split into header and body.
+    This flag is used only by inproc transport to avoid merging and re-splitting
+    the messages passed with a single process. */
+#define NN_PIPE_PARSED 2
+
+/*  Events generated by the pipe. */
+#define NN_PIPE_IN 33987
+#define NN_PIPE_OUT 33988
+
+struct nn_pipe;
+
+/*  Associates opaque pointer to protocol-specific data with the pipe. */
+void nn_pipe_setdata (struct nn_pipe *self, void *data);
+
+/*  Retrieves the opaque pointer associated with the pipe. */
+void *nn_pipe_getdata (struct nn_pipe *self);
+
+/*  Send the message to the pipe. If successful, pipe takes ownership of the
+    messages. */
+int nn_pipe_send (struct nn_pipe *self, struct nn_msg *msg);
+
+/*  Receive a message from a pipe. 'msg' should not be initialised prior to
+    the call. It will be initialised when the call succeeds. */
+int nn_pipe_recv (struct nn_pipe *self, struct nn_msg *msg);
+
+/*  Get option for pipe. Mostly useful for endpoint-specific options  */
+void nn_pipe_getopt (struct nn_pipe *self, int level, int option,
+    void *optval, size_t *optvallen);
+
+
+/******************************************************************************/
+/*  Base class for all socket types.                                          */
+/******************************************************************************/
+
+struct nn_sockbase;
+
+/*  Any combination of these events can be returned from 'events' virtual
+    function. */
+#define NN_SOCKBASE_EVENT_IN 1
+#define NN_SOCKBASE_EVENT_OUT 2
+
+/*  To be implemented by individual socket types. */
+struct nn_sockbase_vfptr {
+
+    /*  Ask socket to stop. */
+    void (*stop) (struct nn_sockbase *self);
+
+    /*  Deallocate the socket. */
+    void (*destroy) (struct nn_sockbase *self);
+
+    /*  Management of pipes. 'add' registers a new pipe. The pipe cannot be used
+        to send to or to be received from at the moment. 'rm' unregisters the
+        pipe. The pipe should not be used after this call as it may already be
+        deallocated. 'in' informs the socket that pipe is readable. 'out'
+        informs it that it is writable. */
+    int (*add) (struct nn_sockbase *self, struct nn_pipe *pipe);
+    void (*rm) (struct nn_sockbase *self, struct nn_pipe *pipe);
+    void (*in) (struct nn_sockbase *self, struct nn_pipe *pipe);
+    void (*out) (struct nn_sockbase *self, struct nn_pipe *pipe);
+
+    /*  Return any combination of event flags defined above, thus specifying
+        whether the socket should be readable, writable, both or none. */
+    int (*events) (struct nn_sockbase *self);
+
+    /*  Send a message to the socket. Returns -EAGAIN if it cannot be done at
+        the moment or zero in case of success. */
+    int (*send) (struct nn_sockbase *self, struct nn_msg *msg);
+
+    /*  Receive a message from the socket. Returns -EAGAIN if it cannot be done
+        at the moment or zero in case of success. */
+    int (*recv) (struct nn_sockbase *self, struct nn_msg *msg);
+
+    /*  Set a protocol specific option. */
+    int (*setopt) (struct nn_sockbase *self, int level, int option,
+        const void *optval, size_t optvallen);
+
+    /*  Retrieve a protocol specific option. */
+    int (*getopt) (struct nn_sockbase *self, int level, int option,
+        void *optval, size_t *optvallen);
+};
+
+struct nn_sockbase {
+    const struct nn_sockbase_vfptr *vfptr;
+    struct nn_sock *sock;
+};
+
+/*  Initialise the socket base class. 'hint' is the opaque value passed to the
+    nn_transport's 'create' function. */
+void nn_sockbase_init (struct nn_sockbase *self,
+    const struct nn_sockbase_vfptr *vfptr, void *hint);
+
+/*  Terminate the socket base class. */
+void nn_sockbase_term (struct nn_sockbase *self);
+
+/*  Call this function when stopping is done. */
+void nn_sockbase_stopped (struct nn_sockbase *self);
+
+/*  Returns the AIO context associated with the socket. This function is
+    useful when socket type implementation needs to create async objects,
+    such as timers. */
+struct nn_ctx *nn_sockbase_getctx (struct nn_sockbase *self);
+
+/*  Retrieve a NN_SOL_SOCKET-level option. */
+int nn_sockbase_getopt (struct nn_sockbase *self, int option,
+    void *optval, size_t *optvallen);
+
+/*  Add some statistics for socket  */
+void nn_sockbase_stat_increment (struct nn_sockbase *self, int name,
+    int increment);
+
+/******************************************************************************/
+/*  The socktype class.                                                       */
+/******************************************************************************/
+
+/*  This structure defines a class factory for individual socket types. */
+
+/*  Specifies that the socket type can be never used to receive messages. */
+#define NN_SOCKTYPE_FLAG_NORECV 1
+
+/*  Specifies that the socket type can be never used to send messages. */
+#define NN_SOCKTYPE_FLAG_NOSEND 2
+
+struct nn_socktype {
+
+    /*  Domain and protocol IDs as specified in nn_socket() function. */
+    int domain;
+    int protocol;
+
+    /*  Any combination of the flags defined above. */
+    int flags;
+
+    /*  Function to create specific socket type. 'sockbase' is the output
+        parameter to return reference to newly created socket. This function
+        is called under global lock, so it is not possible that two sockets are
+        being created in parallel. */
+    int (*create) (void *hint, struct nn_sockbase **sockbase);
+
+    /*  Returns 1 if the supplied socket type is a valid peer for this socket,
+        0 otherwise. Note that the validation is done only within a single
+        SP protocol. Peers speaking other SP protocols are discarded by the
+        core and socket is not even asked to validate them. */
+    int (*ispeer) (int socktype);
+};
+
+#endif
+

example/ev_nanomsg/nanomsg/pubsub.h

+/*
+    Copyright (c) 2012 Martin Sustrik  All rights reserved.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef PUBSUB_H_INCLUDED
+#define PUBSUB_H_INCLUDED
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_PROTO_PUBSUB 2
+
+#define NN_PUB (NN_PROTO_PUBSUB * 16 + 0)
+#define NN_SUB (NN_PROTO_PUBSUB * 16 + 1)
+
+#define NN_SUB_SUBSCRIBE 1
+#define NN_SUB_UNSUBSCRIBE 2
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/ev_nanomsg/nanomsg/reqrep.h

+/*
+    Copyright (c) 2012 Martin Sustrik  All rights reserved.
+    Copyright 2016 Garrett D'Amore <garrett@damore.org>
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef REQREP_H_INCLUDED
+#define REQREP_H_INCLUDED
+
+#include "nn.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_PROTO_REQREP 3
+
+#define NN_REQ (NN_PROTO_REQREP * 16 + 0)
+#define NN_REP (NN_PROTO_REQREP * 16 + 1)
+
+#define NN_REQ_RESEND_IVL 1
+
+typedef union nn_req_handle {
+    int i;
+    void *ptr;
+} nn_req_handle;
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/ev_nanomsg/nanomsg/survey.h

+/*
+    Copyright (c) 2012 Martin Sustrik  All rights reserved.
+    Copyright 2015 Garrett D'Amore <garrett@damore.org>
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef SURVEY_H_INCLUDED
+#define SURVEY_H_INCLUDED
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_PROTO_SURVEY 6
+
+/*  NB: Version 0 used 16 + 0/1.  That version lacked backtraces, and so
+    is wire-incompatible with this version. */
+
+#define NN_SURVEYOR (NN_PROTO_SURVEY * 16 + 2)
+#define NN_RESPONDENT (NN_PROTO_SURVEY * 16 + 3)
+
+#define NN_SURVEYOR_DEADLINE 1
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/ev_nanomsg/nanomsg/tcp.h

+/*
+    Copyright (c) 2012 Martin Sustrik  All rights reserved.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef TCP_H_INCLUDED
+#define TCP_H_INCLUDED
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_TCP -3
+
+#define NN_TCP_NODELAY 1
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/ev_nanomsg/nanomsg/transport.h

+/*
+    Copyright (c) 2012-2014 Martin Sustrik  All rights reserved.
+    Copyright (c) 2013 GoPivotal, Inc.  All rights reserved.
+    Copyright 2016 Garrett D'Amore <garrett@damore.org>
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef NN_TRANSPORT_INCLUDED
+#define NN_TRANSPORT_INCLUDED
+
+#include "nn.h"
+
+#include "aio/fsm.h"
+
+#include "utils/list.h"
+#include "utils/msg.h"
+
+#include <stddef.h>
+
+/*  This is the API between the nanomsg core and individual transports. */
+
+struct nn_sock;
+
+/******************************************************************************/
+/*  Container for transport-specific socket options.                          */
+/******************************************************************************/
+
+struct nn_optset;
+
+struct nn_optset_vfptr {
+    void (*destroy) (struct nn_optset *self);
+    int (*setopt) (struct nn_optset *self, int option, const void *optval,
+        size_t optvallen);
+    int (*getopt) (struct nn_optset *self, int option, void *optval,
+        size_t *optvallen);
+};
+
+struct nn_optset {
+    const struct nn_optset_vfptr *vfptr;
+};
+
+/******************************************************************************/
+/*  The base class for endpoints.                                             */
+/******************************************************************************/
+
+/*  The best way to think about endpoints is that endpoint is an object created
+    by each nn_bind() or nn_connect() call. Each endpoint is associated with
+    exactly one address string (e.g. "tcp://127.0.0.1:5555"). */
+
+struct nn_ep;
+
+struct nn_ep_ops {
+
+    /*  Ask the endpoint to stop itself. The endpoint is allowed to linger
+        to send the pending outbound data. When done, it reports the fact by
+        invoking nn_ep_stopped() function. */
+    void (*stop) (void *);
+
+    /*  Deallocate the endpoint object. It will already have been stopped. */
+    void (*destroy) (void *);
+};
+
+/*  Set up an ep for use by a transport.  The final opaque argument is passed
+    as the first argument to the ops entry points. */
+void nn_ep_tran_setup (struct nn_ep *, const struct nn_ep_ops *, void *);
+
+/*  Notify the user that stopping is done. */
+void nn_ep_stopped (struct nn_ep *);
+
+/*  Returns the AIO context associated with the endpoint. */
+struct nn_ctx *nn_ep_getctx (struct nn_ep *);
+
+/*  Returns the address string associated with this endpoint. */
+const char *nn_ep_getaddr (struct nn_ep *self);
+
+/*  Retrieve value of a socket option. */
+void nn_ep_getopt (struct nn_ep *, int level, int option,
+    void *optval, size_t *optvallen);
+
+/*  Returns 1 if the specified socket type is a valid peer for this socket,
+    or 0 otherwise. */
+int nn_ep_ispeer (struct nn_ep *, int socktype);
+
+/*  Returns 1 if the ep's are valid peers for each other, 0 otherwise. */
+int nn_ep_ispeer_ep (struct nn_ep *, struct nn_ep *);
+
+/*  Notifies a monitoring system the error on this endpoint  */
+void nn_ep_set_error(struct nn_ep*, int errnum);
+
+/*  Notifies a monitoring system that error is gone  */
+void nn_ep_clear_error(struct nn_ep *);
+
+/*  Increments statistics counters in the socket structure  */
+void nn_ep_stat_increment(struct nn_ep *, int name, int increment);
+
+
+/******************************************************************************/
+/*  The base class for pipes.                                                 */
+/******************************************************************************/
+
+/*  Pipe represents one "connection", i.e. perfectly ordered uni- or
+    bi-directional stream of messages. One endpoint can create multiple pipes
+    (for example, bound TCP socket is an endpoint, individual accepted TCP
+    connections are represented by pipes. */
+
+struct nn_pipebase;
+
+/*  This value is returned by pipe's send and recv functions to signalise that
+    more sends/recvs are not possible at the moment. From that moment on,
+    the core will stop invoking the function. To re-establish the message
+    flow nn_pipebase_received (respectively nn_pipebase_sent) should
+    be called. */
+#define NN_PIPEBASE_RELEASE 1
+
+/*  Specifies that received message is already split into header and body.
+    This flag is used only by inproc transport to avoid merging and re-splitting
+    the messages passed with a single process. */
+#define NN_PIPEBASE_PARSED 2
+
+struct nn_pipebase_vfptr {
+
+    /*  Send a message to the network. The function can return either error
+        (negative number) or any combination of the flags defined above. */
+    int (*send) (struct nn_pipebase *self, struct nn_msg *msg);
+
+    /*  Receive a message from the network. The function can return either error
+        (negative number) or any combination of the flags defined above. */
+    int (*recv) (struct nn_pipebase *self, struct nn_msg *msg);
+};
+
+/*  Endpoint specific options. Same restrictions as for nn_pipebase apply  */
+struct nn_ep_options
+{
+    int sndprio;
+    int rcvprio;
+    int ipv4only;
+};
+
+/*  The member of this structure are used internally by the core. Never use
+    or modify them directly from the transport. */
+struct nn_pipebase {
+    struct nn_fsm fsm;
+    const struct nn_pipebase_vfptr *vfptr;
+    uint8_t state;
+    uint8_t instate;
+    uint8_t outstate;
+    struct nn_sock *sock;
+    void *data;
+    struct nn_fsm_event in;
+    struct nn_fsm_event out;
+    struct nn_ep_options options;
+};
+
+/*  Initialise the pipe.  */
+void nn_pipebase_init (struct nn_pipebase *self,
+    const struct nn_pipebase_vfptr *vfptr, struct nn_ep *ep);
+
+/*  Terminate the pipe. */
+void nn_pipebase_term (struct nn_pipebase *self);
+
+/*  Call this function once the connection is established. */
+int nn_pipebase_start (struct nn_pipebase *self);
+
+/*  Call this function once the connection is broken. */
+void nn_pipebase_stop (struct nn_pipebase *self);
+
+/*  Call this function when new message was fully received. */
+void nn_pipebase_received (struct nn_pipebase *self);
+
+/*  Call this function when current outgoing message was fully sent. */
+void nn_pipebase_sent (struct nn_pipebase *self);
+
+/*  Retrieve value of a socket option. */
+void nn_pipebase_getopt (struct nn_pipebase *self, int level, int option,
+    void *optval, size_t *optvallen);
+
+/*  Returns 1 is the specified socket type is a valid peer for this socket,
+    or 0 otherwise. */
+int nn_pipebase_ispeer (struct nn_pipebase *self, int socktype);
+
+/******************************************************************************/
+/*  The transport class.                                                      */
+/******************************************************************************/
+
+struct nn_transport {
+
+    /*  Name of the transport as it appears in the connection strings ("tcp",
+        "ipc", "inproc" etc. */
+    const char *name;
+
+    /*  ID of the transport. */
+    int id;
+
+    /*  Following methods are guarded by a global critical section. Two of these
+        function will never be invoked in parallel. The first is called when
+        the library is initialised, the second one when it is terminated, i.e.
+        when there are no more open sockets. Either of them can be set to NULL
+        if no specific initialisation/termination is needed. */
+    void (*init) (void);
+    void (*term) (void);
+
+    /*  Each of these functions creates an endpoint and sets up the newly
+        established endpoint in 'ep' parameter using nn_ep_tran_setup ().
+        These functions are guarded by a socket-wide critical section.
+        Two of these function will never be invoked in parallel on the same
+        socket. */
+    int (*bind) (struct nn_ep *);
+    int (*connect) (struct nn_ep *);
+
+    /*  Create an object to hold transport-specific socket options.
+        Set this member to NULL in case there are no transport-specific
+        socket options available. */
+    struct nn_optset *(*optset) (void);
+};
+
+#endif

example/ev_nanomsg/nanomsg/ws.h

+/*
+    Copyright (c) 2012 250bpm s.r.o.  All rights reserved.
+    Copyright (c) 2014 Wirebird Labs LLC.  All rights reserved.
+    Copyright 2015 Garrett D'Amore <garrett@damore.org>
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef WS_H_INCLUDED
+#define WS_H_INCLUDED
+
+#include "nn.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_WS -4
+
+/*  NN_WS level socket/cmsg options.  Note that only NN_WSMG_TYPE_TEXT and
+    NN_WS_MSG_TYPE_BINARY messages are supported fully by this implementation.
+    Attempting to set other message types is undefined.  */
+#define NN_WS_MSG_TYPE 1
+
+/*  WebSocket opcode constants as per RFC 6455 5.2  */
+#define NN_WS_MSG_TYPE_TEXT 0x01
+#define NN_WS_MSG_TYPE_BINARY 0x02
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

example/nanomsg/bus.c

+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#include <nanomsg/nn.h>
+#include <nanomsg/bus.h>
+
+void
+fatal(const char *func)
+{
+        fprintf(stderr, "%s: %s\n", func, nn_strerror(nn_errno()));
+        exit(1);
+}
+
+int
+node(const int argc, const char **argv)
+{
+        int sock;
+
+        if ((sock = nn_socket (AF_SP, NN_BUS)) < 0) {
+                fatal("nn_socket");
+        }
+          if (nn_bind(sock, argv[2]) < 0) {
+                fatal("nn_bind");
+        }
+
+        sleep(1); // wait for peers to bind
+        if (argc >= 3) {
+                for (int x = 3; x < argc; x++) {
+                        if (nn_connect(sock, argv[x]) < 0) {
+                                fatal("nn_connect");
+                        }
+                }
+        }
+        sleep(1); // wait for connections
+        int to = 100;
+        if (nn_setsockopt(sock, NN_SOL_SOCKET, NN_RCVTIMEO, &to,
+            sizeof (to)) < 0) {
+                fatal("nn_setsockopt");
+        }
+
+        // SEND
+        int sz_n = strlen(argv[1]) + 1; // '\0' too
+        printf("%s: SENDING '%s' ONTO BUS\n", argv[1], argv[1]);
+        if (nn_send(sock, argv[1], sz_n, 0) < 0) {
+                fatal("nn_send");
+        }
+
+        // RECV
+        for (;;) {
+                char *buf = NULL;
+                int recv = nn_recv(sock, &buf, NN_MSG, 0);
+                if (recv >= 0) {
+                        printf("%s: RECEIVED '%s' FROM BUS\n", argv[1], buf); 
+                        nn_freemsg(buf);
+                }
+        }
+        return (nn_shutdown(sock, 0));
+}
+
+int
+main(int argc, const char **argv)
+{
+        if (argc >= 3) {
+                return (node(argc, argv));
+        }
+        fprintf(stderr, "Usage: bus <NODE_NAME> <URL> <URL> ...\n");
+        return 1;
+}
+

example/nanomsg/iot.mk

+LIBA_TARGET 		:= libnanomsg_fnc.a
+
+SRCS_example 		:= pair.c 
+
+$(call Append_Conditional, LIB_SRCS_EXCLUDE, bus.c)
+$(call Append_Conditional, SRCS_bus, bus.c)
+$(call Append_Conditional, TARGET, bus)
+
+$(call Append_Conditional, LIB_SRCS_EXCLUDE, pair.c)
+$(call Append_Conditional, SRCS_pair, pair.c)
+$(call Append_Conditional, TARGET, pair)
+
+$(call Append_Conditional, LIB_SRCS_EXCLUDE, pipeline.c)
+$(call Append_Conditional, SRCS_pipeline, pipeline.c)
+$(call Append_Conditional, TARGET, pipeline)
+
+$(call Append_Conditional, LIB_SRCS_EXCLUDE, pubsub.c)
+$(call Append_Conditional, SRCS_pubsub, pubsub.c)
+$(call Append_Conditional, TARGET, pubsub)
+
+$(call Append_Conditional, LIB_SRCS_EXCLUDE, reqrep.c)
+$(call Append_Conditional, SRCS_reqrep, reqrep.c)
+$(call Append_Conditional, TARGET, reqrep)
+
+$(call Append_Conditional, LIB_SRCS_EXCLUDE, survey.c)
+$(call Append_Conditional, SRCS_survey, survey.c)
+$(call Append_Conditional, TARGET, survey)
+
+
+#CFLAGS 			+= -I../../work/ssl/include -lrt
+#LDFLAGS 		+= -L../../work/ssl/lib
+#DEPENDS         += common/mqtt
+LDFLAGS         += -L$(TOP_DIR)/example/nanomsg -static -lnanomsg -lanl
+
+
+

example/nanomsg/nanomsg/bus.h

+/*
+    Copyright (c) 2013 Martin Sustrik  All rights reserved.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef BUS_H_INCLUDED
+#define BUS_H_INCLUDED
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_PROTO_BUS 7
+
+#define NN_BUS (NN_PROTO_BUS * 16 + 0)
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/nanomsg/nanomsg/inproc.h

+/*
+    Copyright (c) 2012 Martin Sustrik  All rights reserved.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef INPROC_H_INCLUDED
+#define INPROC_H_INCLUDED
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_INPROC -1
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/nanomsg/nanomsg/ipc.h

+/*
+    Copyright (c) 2012 Martin Sustrik  All rights reserved.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef IPC_H_INCLUDED
+#define IPC_H_INCLUDED
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_IPC -2
+
+/* The object set here must be valid as long as you are using the socket */
+#define NN_IPC_SEC_ATTR 1
+#define NN_IPC_OUTBUFSZ 2
+#define NN_IPC_INBUFSZ 3
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/nanomsg/nanomsg/pair.h

+/*
+    Copyright (c) 2012 Martin Sustrik  All rights reserved.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef PAIR_H_INCLUDED
+#define PAIR_H_INCLUDED
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_PROTO_PAIR 1
+
+#define NN_PAIR (NN_PROTO_PAIR * 16 + 0)
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/nanomsg/nanomsg/pipeline.h

+/*
+    Copyright (c) 2012 Martin Sustrik  All rights reserved.
+    Copyright (c) 2013 GoPivotal, Inc.  All rights reserved.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef PIPELINE_H_INCLUDED
+#define PIPELINE_H_INCLUDED
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_PROTO_PIPELINE 5
+
+#define NN_PUSH (NN_PROTO_PIPELINE * 16 + 0)
+#define NN_PULL (NN_PROTO_PIPELINE * 16 + 1)
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/nanomsg/nanomsg/protocol.h

+/*
+    Copyright (c) 2012-2013 Martin Sustrik  All rights reserved.
+    Copyright 2016 Garrett D'Amore <garrett@damore.org>
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef NN_PROTOCOL_INCLUDED
+#define NN_PROTOCOL_INCLUDED
+
+#include "utils/msg.h"
+#include "utils/list.h"
+
+#include <stddef.h>
+#include <stdint.h>
+
+struct nn_ctx;
+
+/******************************************************************************/
+/*  Pipe class.                                                               */
+/******************************************************************************/
+
+/*  Any combination of following flags can be returned from successful call
+    to nn_pipe_send or nn_pipe_recv. */
+
+/*  This flag means that the pipe can't be used for receiving (when returned
+    from nn_pipe_recv()) or sending (when returned from nn_pipe_send()).
+    Protocol implementation should not send/recv messages from the pipe until
+    the pipe is revived by in()/out() function. */
+#define NN_PIPE_RELEASE 1
+
+/*  Specifies that received message is already split into header and body.
+    This flag is used only by inproc transport to avoid merging and re-splitting
+    the messages passed with a single process. */
+#define NN_PIPE_PARSED 2
+
+/*  Events generated by the pipe. */
+#define NN_PIPE_IN 33987
+#define NN_PIPE_OUT 33988
+
+struct nn_pipe;
+
+/*  Associates opaque pointer to protocol-specific data with the pipe. */
+void nn_pipe_setdata (struct nn_pipe *self, void *data);
+
+/*  Retrieves the opaque pointer associated with the pipe. */
+void *nn_pipe_getdata (struct nn_pipe *self);
+
+/*  Send the message to the pipe. If successful, pipe takes ownership of the
+    messages. */
+int nn_pipe_send (struct nn_pipe *self, struct nn_msg *msg);
+
+/*  Receive a message from a pipe. 'msg' should not be initialised prior to
+    the call. It will be initialised when the call succeeds. */
+int nn_pipe_recv (struct nn_pipe *self, struct nn_msg *msg);
+
+/*  Get option for pipe. Mostly useful for endpoint-specific options  */
+void nn_pipe_getopt (struct nn_pipe *self, int level, int option,
+    void *optval, size_t *optvallen);
+
+
+/******************************************************************************/
+/*  Base class for all socket types.                                          */
+/******************************************************************************/
+
+struct nn_sockbase;
+
+/*  Any combination of these events can be returned from 'events' virtual
+    function. */
+#define NN_SOCKBASE_EVENT_IN 1
+#define NN_SOCKBASE_EVENT_OUT 2
+
+/*  To be implemented by individual socket types. */
+struct nn_sockbase_vfptr {
+
+    /*  Ask socket to stop. */
+    void (*stop) (struct nn_sockbase *self);
+
+    /*  Deallocate the socket. */
+    void (*destroy) (struct nn_sockbase *self);
+
+    /*  Management of pipes. 'add' registers a new pipe. The pipe cannot be used
+        to send to or to be received from at the moment. 'rm' unregisters the
+        pipe. The pipe should not be used after this call as it may already be
+        deallocated. 'in' informs the socket that pipe is readable. 'out'
+        informs it that it is writable. */
+    int (*add) (struct nn_sockbase *self, struct nn_pipe *pipe);
+    void (*rm) (struct nn_sockbase *self, struct nn_pipe *pipe);
+    void (*in) (struct nn_sockbase *self, struct nn_pipe *pipe);
+    void (*out) (struct nn_sockbase *self, struct nn_pipe *pipe);
+
+    /*  Return any combination of event flags defined above, thus specifying
+        whether the socket should be readable, writable, both or none. */
+    int (*events) (struct nn_sockbase *self);
+
+    /*  Send a message to the socket. Returns -EAGAIN if it cannot be done at
+        the moment or zero in case of success. */
+    int (*send) (struct nn_sockbase *self, struct nn_msg *msg);
+
+    /*  Receive a message from the socket. Returns -EAGAIN if it cannot be done
+        at the moment or zero in case of success. */
+    int (*recv) (struct nn_sockbase *self, struct nn_msg *msg);
+
+    /*  Set a protocol specific option. */
+    int (*setopt) (struct nn_sockbase *self, int level, int option,
+        const void *optval, size_t optvallen);
+
+    /*  Retrieve a protocol specific option. */
+    int (*getopt) (struct nn_sockbase *self, int level, int option,
+        void *optval, size_t *optvallen);
+};
+
+struct nn_sockbase {
+    const struct nn_sockbase_vfptr *vfptr;
+    struct nn_sock *sock;
+};
+
+/*  Initialise the socket base class. 'hint' is the opaque value passed to the
+    nn_transport's 'create' function. */
+void nn_sockbase_init (struct nn_sockbase *self,
+    const struct nn_sockbase_vfptr *vfptr, void *hint);
+
+/*  Terminate the socket base class. */
+void nn_sockbase_term (struct nn_sockbase *self);
+
+/*  Call this function when stopping is done. */
+void nn_sockbase_stopped (struct nn_sockbase *self);
+
+/*  Returns the AIO context associated with the socket. This function is
+    useful when socket type implementation needs to create async objects,
+    such as timers. */
+struct nn_ctx *nn_sockbase_getctx (struct nn_sockbase *self);
+
+/*  Retrieve a NN_SOL_SOCKET-level option. */
+int nn_sockbase_getopt (struct nn_sockbase *self, int option,
+    void *optval, size_t *optvallen);
+
+/*  Add some statistics for socket  */
+void nn_sockbase_stat_increment (struct nn_sockbase *self, int name,
+    int increment);
+
+/******************************************************************************/
+/*  The socktype class.                                                       */
+/******************************************************************************/
+
+/*  This structure defines a class factory for individual socket types. */
+
+/*  Specifies that the socket type can be never used to receive messages. */
+#define NN_SOCKTYPE_FLAG_NORECV 1
+
+/*  Specifies that the socket type can be never used to send messages. */
+#define NN_SOCKTYPE_FLAG_NOSEND 2
+
+struct nn_socktype {
+
+    /*  Domain and protocol IDs as specified in nn_socket() function. */
+    int domain;
+    int protocol;
+
+    /*  Any combination of the flags defined above. */
+    int flags;
+
+    /*  Function to create specific socket type. 'sockbase' is the output
+        parameter to return reference to newly created socket. This function
+        is called under global lock, so it is not possible that two sockets are
+        being created in parallel. */
+    int (*create) (void *hint, struct nn_sockbase **sockbase);
+
+    /*  Returns 1 if the supplied socket type is a valid peer for this socket,
+        0 otherwise. Note that the validation is done only within a single
+        SP protocol. Peers speaking other SP protocols are discarded by the
+        core and socket is not even asked to validate them. */
+    int (*ispeer) (int socktype);
+};
+
+#endif
+

example/nanomsg/nanomsg/pubsub.h

+/*
+    Copyright (c) 2012 Martin Sustrik  All rights reserved.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef PUBSUB_H_INCLUDED
+#define PUBSUB_H_INCLUDED
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_PROTO_PUBSUB 2
+
+#define NN_PUB (NN_PROTO_PUBSUB * 16 + 0)
+#define NN_SUB (NN_PROTO_PUBSUB * 16 + 1)
+
+#define NN_SUB_SUBSCRIBE 1
+#define NN_SUB_UNSUBSCRIBE 2
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/nanomsg/nanomsg/reqrep.h

+/*
+    Copyright (c) 2012 Martin Sustrik  All rights reserved.
+    Copyright 2016 Garrett D'Amore <garrett@damore.org>
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef REQREP_H_INCLUDED
+#define REQREP_H_INCLUDED
+
+#include "nn.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_PROTO_REQREP 3
+
+#define NN_REQ (NN_PROTO_REQREP * 16 + 0)
+#define NN_REP (NN_PROTO_REQREP * 16 + 1)
+
+#define NN_REQ_RESEND_IVL 1
+
+typedef union nn_req_handle {
+    int i;
+    void *ptr;
+} nn_req_handle;
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/nanomsg/nanomsg/survey.h

+/*
+    Copyright (c) 2012 Martin Sustrik  All rights reserved.
+    Copyright 2015 Garrett D'Amore <garrett@damore.org>
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef SURVEY_H_INCLUDED
+#define SURVEY_H_INCLUDED
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_PROTO_SURVEY 6
+
+/*  NB: Version 0 used 16 + 0/1.  That version lacked backtraces, and so
+    is wire-incompatible with this version. */
+
+#define NN_SURVEYOR (NN_PROTO_SURVEY * 16 + 2)
+#define NN_RESPONDENT (NN_PROTO_SURVEY * 16 + 3)
+
+#define NN_SURVEYOR_DEADLINE 1
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/nanomsg/nanomsg/tcp.h

+/*
+    Copyright (c) 2012 Martin Sustrik  All rights reserved.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef TCP_H_INCLUDED
+#define TCP_H_INCLUDED
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_TCP -3
+
+#define NN_TCP_NODELAY 1
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

example/nanomsg/nanomsg/transport.h

+/*
+    Copyright (c) 2012-2014 Martin Sustrik  All rights reserved.
+    Copyright (c) 2013 GoPivotal, Inc.  All rights reserved.
+    Copyright 2016 Garrett D'Amore <garrett@damore.org>
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef NN_TRANSPORT_INCLUDED
+#define NN_TRANSPORT_INCLUDED
+
+#include "nn.h"
+
+#include "aio/fsm.h"
+
+#include "utils/list.h"
+#include "utils/msg.h"
+
+#include <stddef.h>
+
+/*  This is the API between the nanomsg core and individual transports. */
+
+struct nn_sock;
+
+/******************************************************************************/
+/*  Container for transport-specific socket options.                          */
+/******************************************************************************/
+
+struct nn_optset;
+
+struct nn_optset_vfptr {
+    void (*destroy) (struct nn_optset *self);
+    int (*setopt) (struct nn_optset *self, int option, const void *optval,
+        size_t optvallen);
+    int (*getopt) (struct nn_optset *self, int option, void *optval,
+        size_t *optvallen);
+};
+
+struct nn_optset {
+    const struct nn_optset_vfptr *vfptr;
+};
+
+/******************************************************************************/
+/*  The base class for endpoints.                                             */
+/******************************************************************************/
+
+/*  The best way to think about endpoints is that endpoint is an object created
+    by each nn_bind() or nn_connect() call. Each endpoint is associated with
+    exactly one address string (e.g. "tcp://127.0.0.1:5555"). */
+
+struct nn_ep;
+
+struct nn_ep_ops {
+
+    /*  Ask the endpoint to stop itself. The endpoint is allowed to linger
+        to send the pending outbound data. When done, it reports the fact by
+        invoking nn_ep_stopped() function. */
+    void (*stop) (void *);
+
+    /*  Deallocate the endpoint object. It will already have been stopped. */
+    void (*destroy) (void *);
+};
+
+/*  Set up an ep for use by a transport.  The final opaque argument is passed
+    as the first argument to the ops entry points. */
+void nn_ep_tran_setup (struct nn_ep *, const struct nn_ep_ops *, void *);
+
+/*  Notify the user that stopping is done. */
+void nn_ep_stopped (struct nn_ep *);
+
+/*  Returns the AIO context associated with the endpoint. */
+struct nn_ctx *nn_ep_getctx (struct nn_ep *);
+
+/*  Returns the address string associated with this endpoint. */
+const char *nn_ep_getaddr (struct nn_ep *self);
+
+/*  Retrieve value of a socket option. */
+void nn_ep_getopt (struct nn_ep *, int level, int option,
+    void *optval, size_t *optvallen);
+
+/*  Returns 1 if the specified socket type is a valid peer for this socket,
+    or 0 otherwise. */
+int nn_ep_ispeer (struct nn_ep *, int socktype);
+
+/*  Returns 1 if the ep's are valid peers for each other, 0 otherwise. */
+int nn_ep_ispeer_ep (struct nn_ep *, struct nn_ep *);
+
+/*  Notifies a monitoring system the error on this endpoint  */
+void nn_ep_set_error(struct nn_ep*, int errnum);
+
+/*  Notifies a monitoring system that error is gone  */
+void nn_ep_clear_error(struct nn_ep *);
+
+/*  Increments statistics counters in the socket structure  */
+void nn_ep_stat_increment(struct nn_ep *, int name, int increment);
+
+
+/******************************************************************************/
+/*  The base class for pipes.                                                 */
+/******************************************************************************/
+
+/*  Pipe represents one "connection", i.e. perfectly ordered uni- or
+    bi-directional stream of messages. One endpoint can create multiple pipes
+    (for example, bound TCP socket is an endpoint, individual accepted TCP
+    connections are represented by pipes. */
+
+struct nn_pipebase;
+
+/*  This value is returned by pipe's send and recv functions to signalise that
+    more sends/recvs are not possible at the moment. From that moment on,
+    the core will stop invoking the function. To re-establish the message
+    flow nn_pipebase_received (respectively nn_pipebase_sent) should
+    be called. */
+#define NN_PIPEBASE_RELEASE 1
+
+/*  Specifies that received message is already split into header and body.
+    This flag is used only by inproc transport to avoid merging and re-splitting
+    the messages passed with a single process. */
+#define NN_PIPEBASE_PARSED 2
+
+struct nn_pipebase_vfptr {
+
+    /*  Send a message to the network. The function can return either error
+        (negative number) or any combination of the flags defined above. */
+    int (*send) (struct nn_pipebase *self, struct nn_msg *msg);
+
+    /*  Receive a message from the network. The function can return either error
+        (negative number) or any combination of the flags defined above. */
+    int (*recv) (struct nn_pipebase *self, struct nn_msg *msg);
+};
+
+/*  Endpoint specific options. Same restrictions as for nn_pipebase apply  */
+struct nn_ep_options
+{
+    int sndprio;
+    int rcvprio;
+    int ipv4only;
+};
+
+/*  The member of this structure are used internally by the core. Never use
+    or modify them directly from the transport. */
+struct nn_pipebase {
+    struct nn_fsm fsm;
+    const struct nn_pipebase_vfptr *vfptr;
+    uint8_t state;
+    uint8_t instate;
+    uint8_t outstate;
+    struct nn_sock *sock;
+    void *data;
+    struct nn_fsm_event in;
+    struct nn_fsm_event out;
+    struct nn_ep_options options;
+};
+
+/*  Initialise the pipe.  */
+void nn_pipebase_init (struct nn_pipebase *self,
+    const struct nn_pipebase_vfptr *vfptr, struct nn_ep *ep);
+
+/*  Terminate the pipe. */
+void nn_pipebase_term (struct nn_pipebase *self);
+
+/*  Call this function once the connection is established. */
+int nn_pipebase_start (struct nn_pipebase *self);
+
+/*  Call this function once the connection is broken. */
+void nn_pipebase_stop (struct nn_pipebase *self);
+
+/*  Call this function when new message was fully received. */
+void nn_pipebase_received (struct nn_pipebase *self);
+
+/*  Call this function when current outgoing message was fully sent. */
+void nn_pipebase_sent (struct nn_pipebase *self);
+
+/*  Retrieve value of a socket option. */
+void nn_pipebase_getopt (struct nn_pipebase *self, int level, int option,
+    void *optval, size_t *optvallen);
+
+/*  Returns 1 is the specified socket type is a valid peer for this socket,
+    or 0 otherwise. */
+int nn_pipebase_ispeer (struct nn_pipebase *self, int socktype);
+
+/******************************************************************************/
+/*  The transport class.                                                      */
+/******************************************************************************/
+
+struct nn_transport {
+
+    /*  Name of the transport as it appears in the connection strings ("tcp",
+        "ipc", "inproc" etc. */
+    const char *name;
+
+    /*  ID of the transport. */
+    int id;
+
+    /*  Following methods are guarded by a global critical section. Two of these
+        function will never be invoked in parallel. The first is called when
+        the library is initialised, the second one when it is terminated, i.e.
+        when there are no more open sockets. Either of them can be set to NULL
+        if no specific initialisation/termination is needed. */
+    void (*init) (void);
+    void (*term) (void);
+
+    /*  Each of these functions creates an endpoint and sets up the newly
+        established endpoint in 'ep' parameter using nn_ep_tran_setup ().
+        These functions are guarded by a socket-wide critical section.
+        Two of these function will never be invoked in parallel on the same
+        socket. */
+    int (*bind) (struct nn_ep *);
+    int (*connect) (struct nn_ep *);
+
+    /*  Create an object to hold transport-specific socket options.
+        Set this member to NULL in case there are no transport-specific
+        socket options available. */
+    struct nn_optset *(*optset) (void);
+};
+
+#endif

example/nanomsg/nanomsg/ws.h

+/*
+    Copyright (c) 2012 250bpm s.r.o.  All rights reserved.
+    Copyright (c) 2014 Wirebird Labs LLC.  All rights reserved.
+    Copyright 2015 Garrett D'Amore <garrett@damore.org>
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"),
+    to deal in the Software without restriction, including without limitation
+    the rights to use, copy, modify, merge, publish, distribute, sublicense,
+    and/or sell copies of the Software, and to permit persons to whom
+    the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included
+    in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+    IN THE SOFTWARE.
+*/
+
+#ifndef WS_H_INCLUDED
+#define WS_H_INCLUDED
+
+#include "nn.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NN_WS -4
+
+/*  NN_WS level socket/cmsg options.  Note that only NN_WSMG_TYPE_TEXT and
+    NN_WS_MSG_TYPE_BINARY messages are supported fully by this implementation.
+    Attempting to set other message types is undefined.  */
+#define NN_WS_MSG_TYPE 1
+
+/*  WebSocket opcode constants as per RFC 6455 5.2  */
+#define NN_WS_MSG_TYPE_TEXT 0x01
+#define NN_WS_MSG_TYPE_BINARY 0x02
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

example/nanomsg/pair.c

+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#include <nanomsg/nn.h>
+#include <nanomsg/pair.h>
+
+#define NODE0 "node0"
+#define NODE1 "node1"
+
+void
+fatal(const char *func)
+{
+        fprintf(stderr, "%s: %s\n", func, nn_strerror(nn_errno()));
+        exit(1);
+}
+
+int
+send_name(int sock, const char *name)
+{
+        printf("%s: SENDING \"%s\"\n", name, name);
+        int sz_n = strlen(name) + 1; // '\0' too
+        return (nn_send(sock, name, sz_n, 0));
+}
+
+int
+recv_name(int sock, const char *name)
+{
+        char *buf = NULL;
+        int result = nn_recv(sock, &buf, NN_MSG, 0);
+        if (result > 0) {
+                printf("%s: RECEIVED \"%s\"\n", name, buf); 
+                nn_freemsg(buf);
+        }
+        return (result);
+}
+
+int
+send_recv(int sock, const char *name)
+{
+        int to = 100;
+        if (nn_setsockopt(sock, NN_SOL_SOCKET, NN_RCVTIMEO, &to,
+            sizeof (to)) < 0) {
+                fatal("nn_setsockopt");
+        }
+
+        for (;;) {
+                recv_name(sock, name);
+                sleep(1);
+                send_name(sock, name);
+        }
+}
+
+int
+node0(const char *url)
+{
+        int sock;
+        if ((sock = nn_socket(AF_SP, NN_PAIR)) < 0) {
+                fatal("nn_socket");
+        }
+         if (nn_bind(sock, url) < 0) {
+                fatal("nn_bind");
+        }
+        return (send_recv(sock, NODE0));
+}
+
+int
+node1(const char *url)
+{
+        int sock;
+        if ((sock = nn_socket(AF_SP, NN_PAIR)) < 0) {
+                fatal("nn_socket");
+        }
+        if (nn_connect(sock, url) < 0) {
+                fatal("nn_connect");
+        }
+        return (send_recv(sock, NODE1));
+}
+
+int
+main(const int argc, const char **argv)
+{
+        if ((argc > 1) && (strcmp(NODE0, argv[1]) == 0))
+                return (node0(argv[2]));
+
+        if ((argc > 1) && (strcmp(NODE1, argv[1]) == 0))
+                return (node1(argv[2]));
+
+        fprintf(stderr, "Usage: pair %s|%s <URL> <ARG> ...\n", NODE0, NODE1);
+        return 1;
+}
+

example/nanomsg/pipeline.c

+#include <stdlib.h>
+#include <stdio.h>
+#include <string.h>
+#include <unistd.h>
+#include <nanomsg/nn.h>
+#include <nanomsg/pipeline.h>
+
+#define NODE0 "node0"
+#define NODE1 "node1"
+
+void
+fatal(const char *func)
+{
+        fprintf(stderr, "%s: %s\n", func, nn_strerror(nn_errno()));
+        exit(1);
+}
+
+int
+node0(const char *url)
+{
+        int sock;
+        int rv;
+
+        if ((sock = nn_socket(AF_SP, NN_PULL)) < 0) {
+                fatal("nn_socket");
+        }
+        if ((rv = nn_bind(sock, url)) < 0) {
+                fatal("nn_bind");
+        }
+        for (;;) {
+                char *buf = NULL;
+                int bytes;
+                if ((bytes = nn_recv(sock, &buf, NN_MSG, 0)) < 0) {
+                        fatal("nn_recv");
+                }
+                printf("NODE0: RECEIVED \"%s\"\n", buf); 
+                nn_freemsg(buf);
+        }
+}
+
+int
+node1(const char *url, const char *msg)
+{
+        int sz_msg = strlen(msg) + 1; // '\0' too
+        int sock;
+        int rv;
+        int bytes;
+
+        if ((sock = nn_socket(AF_SP, NN_PUSH)) < 0) {
+                fatal("nn_socket");
+        }
+        if ((rv = nn_connect(sock, url)) < 0) {
+                fatal("nn_connect");
+        }
+        printf("NODE1: SENDING \"%s\"\n", msg);
+        if ((bytes = nn_send(sock, msg, sz_msg, 0)) < 0) {
+                fatal("nn_send");
+        }
+        sleep(1); // wait for messages to flush before shutting down
+        return (nn_shutdown(sock, 0));
+}
+
+int
+main(const int argc, const char **argv)
+{
+        if ((argc > 1) && (strcmp(NODE0, argv[1]) == 0))
+                return (node0(argv[2]));
+
+        if ((argc > 2) && (strcmp(NODE1, argv[1]) == 0))
+                return (node1(argv[2], argv[3]));
+
+        fprintf(stderr, "Usage: pipeline %s|%s <URL> <ARG> ...'\n",
+                NODE0, NODE1);
+        return (1);
+}
+

makefile

@@ -13,6 +13,9 @@ COMP_LIB_COMPONENTS := \
 SUBDIRS                 += common/mqtt
 #SUBDIRS                 += common/nanomsg
 SUBDIRS                 += example/mqtt
+SUBDIRS                 += example/nanomsg
+SUBDIRS                 += example/ev
+SUBDIRS                 += example/ev_nanomsg
 
 $(call Append_Conditional, SUBDIRS)