【修改内容】调整目录结构

【陈伟灿】

common/api/com_api.c

+//=========kk=============
+#include "com_api.h"
+
+#define APP2MID "ipc:///tmp/app2mid.ipc"
+#define PLAT2MID "ipc:///tmp/plat2mid.ipc"
+
+typedef struct {
+    int n;          //nanomsg socket
+    int s;          //nanomsg recieve fd
+}nanomsg_info_t;
+  
+typedef struct {
+    nanomsg_info_t ba;
+}Bloop_ctrl_t;
+Bloop_ctrl_t Bloop_ctrl;
+static struct ev_io g_watcher;
+
+ipc_cb* g_cb = NULL;
+
+static void watcher_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("watcher_cb:%s recived\r\n\r\n", (char *)dat);
+	if (g_cb != NULL){
+		g_cb((void *)dat);
+	}
+    nn_freemsg(dat);
+}
+
+
+struct ev_loop* __loop_init(Bloop_ctrl_t *loop_ctrl)
+{
+    
+    struct ev_loop *loop = ev_loop_new(EVBACKEND_EPOLL);
+    if (NULL == loop) {
+        printf("create loop failed\r\n");
+        return NULL;
+    }
+    ev_io_init (&g_watcher, watcher_cb, loop_ctrl->ba.s, EV_READ);
+    ev_io_start (loop, &g_watcher);
+    return loop;
+}
+
+
+int __nanomsg_init(Bloop_ctrl_t *Bloop_ctrl, ipc_type type)
+{
+    Bloop_ctrl->ba.n = nn_socket(AF_SP, NN_PAIR);
+	printf("__nanomsg_initBloop_ctrl->ba.n=%d \r\n",Bloop_ctrl->ba.n);
+    if (Bloop_ctrl->ba.n < 0) {
+        return -1;
+    }
+
+	switch (type) {
+        case IPC_APP2MID:{
+            if (nn_connect(Bloop_ctrl->ba.n, APP2MID) < 0) {
+		        return -1;
+		    }
+        }
+        break;
+		case IPC_PLAT2MID: {
+            if (nn_connect(Bloop_ctrl->ba.n, PLAT2MID) < 0) {
+		        return -1;
+		    }
+        }
+        break;
+		case IPC_MID2APP: {
+            if (nn_bind(Bloop_ctrl->ba.n, APP2MID) < 0) {
+		        return -1;
+		    }
+        }
+        break;
+		case IPC_MID2PLAT: {
+            if (nn_bind(Bloop_ctrl->ba.n, PLAT2MID) < 0) {
+		        return -1;
+		    }
+        }
+        break;
+        default: {
+        }
+        break;
+    } 
+   
+    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;
+}
+
+
+
+
+/*=================================
+*  TODO: only one methd pair
+*
+*
+==================================*/
+void loop_thread(void *arg){
+	struct ev_loop* loop = __loop_init(&Bloop_ctrl);
+    if (NULL == loop) {
+        printf("loop init failed\r\n");
+        return ;
+    }
+	
+    ev_set_userdata(loop, &Bloop_ctrl);
+    ev_run (loop, 0);
+	
+}
+void kk_ipc_init(ipc_type type, ipc_cb cb)
+{
+	pthread_t pb;
+	if (g_cb == NULL){
+
+		printf("ipc has been inited!\r\n");
+		return;
+	}
+	
+    if (__nanomsg_init(&Bloop_ctrl, type) < 0) {
+        printf("nanomsg init failed\r\n");
+        return ;
+    }
+
+	if (0 != pthread_create(&pb, NULL, loop_thread, NULL)) {
+        printf("create pthread B failed\r\n");
+        return -1;
+    }
+	g_cb  = cb;
+    return ;
+	
+}
+
+int kk_ipc_dinit()
+{
+#if 0
+	if (Bloop_ctrl.ba.n > -1){
+		nn_shutdown(Bloop_ctrl.ba.n, 0);
+	}
+	ev_io_stop(g_watcher);
+    ev_break(EV_A_ EVBREAK_ALL);
+	g_cb = NULL;
+#endif	
+
+}
+
+int kk_ipc_send(void* data)
+{
+	//printf("kk_ipc_send 11111111\r\n");
+	if (data != NULL){
+		printf("kk_ipc_send data= %s\r\n", data);
+		char* buf = nn_allocmsg(strlen(data)+1, 0);
+		memcpy(buf, data,strlen(data)+1);
+		nn_send(Bloop_ctrl.ba.n, &buf, NN_MSG, NN_DONTWAIT);
+		//printf("kk_ipc_send: RECEIVED \"%s\"\n", data); 
+	}
+
+}

common/api/com_api.h

+/*
+ * Copyright (C) 2020-2020 ikonke
+ */
+
+#ifndef _KK_COM_API_H_
+#define _KK_COM_API_H_
+
+#if defined(__cplusplus)
+extern "C" {
+#endif
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+#include "ev.h"
+#include "nn.h"
+#include "pair.h"
+
+
+
+//=====kk======================
+typedef enum {
+    IPC_APP2MID = 0,
+    IPC_MID2APP,
+    IPC_MID2PLAT,
+    IPC_PLAT2MID
+} ipc_type;
+
+typedef void ipc_cb(void* data);
+void kk_ipc_init(ipc_type type, ipc_cb cb);
+int kk_ipc_dinit();
+int kk_ipc_send(void* data);
+
+#if defined(__cplusplus)
+}
+#endif
+#endif
+
+

common/api/iot.mk

+LIBA_TARGET 		:= libapi_com.a
+#
+
+CFLAGS 			+= -I$(TOP_DIR)/common/nanomsg/include
+CFLAGS 			+= -I$(TOP_DIR)/common/ev/include
+LDFLAGS         += -L$(TOP_DIR)/common/nanomsg -static -lnanomsg
+LDFLAGS         += -L$(TOP_DIR)/common/ev -static -lev  
+LDFLAGS         += -ldl
\ No newline at end of file

common/ev/iot.mk

+LIBA_TARGET 		:= libapi_ev.a
\ No newline at end of file

common/json/cJSON.h

+/*
+  Copyright (c) 2009 Dave Gamble
+ 
+  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 cJSON__h
+#define cJSON__h
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+/* cJSON Types: */
+#define cJSON_False 0
+#define cJSON_True 1
+#define cJSON_NULL 2
+#define cJSON_Number 3
+#define cJSON_String 4
+#define cJSON_Array 5
+#define cJSON_Object 6
+	
+#define cJSON_IsReference 256
+
+/* The cJSON structure: */
+typedef struct cJSON {
+	struct cJSON *next,*prev;	/* next/prev allow you to walk array/object chains. Alternatively, use GetArraySize/GetArrayItem/GetObjectItem */
+	struct cJSON *child;		/* An array or object item will have a child pointer pointing to a chain of the items in the array/object. */
+
+	int type;					/* The type of the item, as above. */
+
+	char *valuestring;			/* The item's string, if type==cJSON_String */
+	int valueint;				/* The item's number, if type==cJSON_Number */
+	double valuedouble;			/* The item's number, if type==cJSON_Number */
+
+	char *string;				/* The item's name string, if this item is the child of, or is in the list of subitems of an object. */
+} cJSON;
+
+typedef struct cJSON_Hooks {
+      void *(*malloc_fn)(size_t sz);
+      void (*free_fn)(void *ptr);
+} cJSON_Hooks;
+
+/* Supply malloc, realloc and free functions to cJSON */
+extern void cJSON_InitHooks(cJSON_Hooks* hooks);
+
+
+/* Supply a block of JSON, and this returns a cJSON object you can interrogate. Call cJSON_Delete when finished. */
+extern cJSON *cJSON_Parse(const char *value);
+/* Supply a block of JSON, and this returns a cJSON object you can interrogate. Call cJSON_Delete when finished.
+ * end_ptr will point to 1 past the end of the JSON object */
+extern cJSON *cJSON_Parse_Stream(const char *value, char **end_ptr);
+/* Render a cJSON entity to text for transfer/storage. Free the char* when finished. */
+extern char  *cJSON_Print(cJSON *item);
+extern char  *cJSON_Print_pure(cJSON *item);
+
+/* Render a cJSON entity to text for transfer/storage without any formatting. Free the char* when finished. */
+extern char  *cJSON_PrintUnformatted(cJSON *item);
+/* Delete a cJSON entity and all subentities. */
+extern void   cJSON_Delete(cJSON *c);
+
+/* Returns the number of items in an array (or object). */
+extern int	  cJSON_GetArraySize(cJSON *array);
+/* Retrieve item number "item" from array "array". Returns NULL if unsuccessful. */
+extern cJSON *cJSON_GetArrayItem(cJSON *array,int item);
+/* Get item "string" from object. Case insensitive. */
+extern cJSON *cJSON_GetObjectItem(cJSON *object,const char *string);
+
+/* These calls create a cJSON item of the appropriate type. */
+extern cJSON *cJSON_CreateNull();
+extern cJSON *cJSON_CreateTrue();
+extern cJSON *cJSON_CreateFalse();
+extern cJSON *cJSON_CreateBool(int b);
+extern cJSON *cJSON_CreateNumber(double num);
+extern cJSON *cJSON_CreateString(const char *string);
+extern cJSON *cJSON_CreateArray();
+extern cJSON *cJSON_CreateObject();
+
+/* These utilities create an Array of count items. */
+extern cJSON *cJSON_CreateIntArray(int *numbers,int count);
+extern cJSON *cJSON_CreateFloatArray(float *numbers,int count);
+extern cJSON *cJSON_CreateDoubleArray(double *numbers,int count);
+extern cJSON *cJSON_CreateStringArray(const char **strings,int count);
+
+/* Append item to the specified array/object. */
+extern void cJSON_AddItemToArray(cJSON *array, cJSON *item);
+extern void	cJSON_AddItemToObject(cJSON *object,const char *string,cJSON *item);
+/* Append reference to item to the specified array/object. Use this when you want to add an existing cJSON to a new cJSON, but don't want to corrupt your existing cJSON. */
+extern void cJSON_AddItemReferenceToArray(cJSON *array, cJSON *item);
+extern void	cJSON_AddItemReferenceToObject(cJSON *object,const char *string,cJSON *item);
+
+/* Remove/Detatch items from Arrays/Objects. */
+extern cJSON *cJSON_DetachItemFromArray(cJSON *array,int which);
+extern void   cJSON_DeleteItemFromArray(cJSON *array,int which);
+extern cJSON *cJSON_DetachItemFromObject(cJSON *object,const char *string);
+extern void   cJSON_DeleteItemFromObject(cJSON *object,const char *string);
+	
+/* Update array items. */
+extern void cJSON_ReplaceItemInArray(cJSON *array,int which,cJSON *newitem);
+extern void cJSON_ReplaceItemInObject(cJSON *object,const char *string,cJSON *newitem);
+
+#define cJSON_AddNullToObject(object,name)	cJSON_AddItemToObject(object, name, cJSON_CreateNull())
+#define cJSON_AddTrueToObject(object,name)	cJSON_AddItemToObject(object, name, cJSON_CreateTrue())
+#define cJSON_AddFalseToObject(object,name)		cJSON_AddItemToObject(object, name, cJSON_CreateFalse())
+#define cJSON_AddNumberToObject(object,name,n)	cJSON_AddItemToObject(object, name, cJSON_CreateNumber(n))
+#define cJSON_AddStringToObject(object,name,s)	cJSON_AddItemToObject(object, name, cJSON_CreateString(s))
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

common/json/iot.mk

+LIBA_TARGET 		:= libiot_cjson.a
\ No newline at end of file

common/mqtt/MQTTAsync.c

@@ -2885,7 +2885,7 @@ static int MQTTAsync_cleanSession(Clients* client)
 static int MQTTAsync_deliverMessage(MQTTAsyncs* m, char* topicName, size_t topicLen, MQTTAsync_message* mm)
 {
 	int rc;
-
+	printf("[%s][%d]\n",__FUNCTION__,__LINE__);
 	Log(TRACE_MIN, -1, "Calling messageArrived for client %s, queue depth %d",
 					m->c->clientID, m->c->messageQueue->count);
 	rc = (*(m->ma))(m->maContext, topicName, (int)topicLen, mm);
@@ -2902,6 +2902,7 @@ void Protocol_processPublication(Publish* publish, Clients* client, int allocate
 	MQTTAsync_message* mm = NULL;
 	MQTTAsync_message initialized = MQTTAsync_message_initializer;
 	int rc = 0;
+	printf("[%s][%d]\n",__FUNCTION__,__LINE__);
 
 	FUNC_ENTRY;
 	if ((mm = malloc(sizeof(MQTTAsync_message))) == NULL)

common/mqtt/MQTTProtocolClient.c

@@ -304,6 +304,7 @@ int MQTTProtocol_handlePublishes(void* pack, int sock)
 	Clients* client = NULL;
 	char* clientid = NULL;
 	int rc = TCPSOCKET_COMPLETE;
+	printf("[%s][%d]\n",__FUNCTION__,__LINE__);
 
 	FUNC_ENTRY;
 	client = (Clients*)(ListFindItem(bstate->clients, &sock, clientSocketCompare)->content);
@@ -311,8 +312,10 @@ int MQTTProtocol_handlePublishes(void* pack, int sock)
 	Log(LOG_PROTOCOL, 11, NULL, sock, clientid, publish->msgId, publish->header.bits.qos,
 					publish->header.bits.retain, publish->payloadlen, min(20, publish->payloadlen), publish->payload);
 
-	if (publish->header.bits.qos == 0)
+	if (publish->header.bits.qos == 0){
+			printf("[%s][%d]\n",__FUNCTION__,__LINE__);
 		Protocol_processPublication(publish, client, 1);
+		}
 	else if (!Socket_noPendingWrites(sock))
 		rc = SOCKET_ERROR; /* queue acks? */
 	else if (publish->header.bits.qos == 1)
@@ -321,6 +324,7 @@ int MQTTProtocol_handlePublishes(void* pack, int sock)
 	  rc = MQTTPacket_send_puback(publish->MQTTVersion, publish->msgId, &client->net, client->clientID);
 	  /* if we get a socket error from sending the puback, should we ignore the publication? */
 	  Protocol_processPublication(publish, client, 1);
+	  	printf("[%s][%d]\n",__FUNCTION__,__LINE__);
 	}
 	else if (publish->header.bits.qos == 2)
 	{
@@ -370,6 +374,7 @@ int MQTTProtocol_handlePublishes(void* pack, int sock)
 			publish1.payloadlen = m->publish->payloadlen;
 			publish1.MQTTVersion = m->MQTTVersion;
 			publish1.properties = m->properties;
+			printf("[%s][%d]\n",__FUNCTION__,__LINE__);
 
 			Protocol_processPublication(&publish1, client, 1);
 			ListRemove(&(state.publications), m->publish);
@@ -567,6 +572,7 @@ int MQTTProtocol_handlePubrels(void* pack, int sock)
 				publish.properties = m->properties;
 			else
 				Protocol_processPublication(&publish, client, 0); /* only for 3.1.1 and lower */
+				printf("[%s][%d]\n",__FUNCTION__,__LINE__);
 			#if !defined(NO_PERSISTENCE)
 				rc += MQTTPersistence_remove(client,
 						(m->MQTTVersion >= MQTTVERSION_5) ? PERSISTENCE_V5_PUBLISH_RECEIVED : PERSISTENCE_PUBLISH_RECEIVED,

common/nanomsg/include/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
+

common/nanomsg/include/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
+

common/nanomsg/include/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
+

common/nanomsg/include/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
+

common/nanomsg/include/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
+

common/nanomsg/include/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
+

common/nanomsg/include/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
+

common/nanomsg/include/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
+

common/nanomsg/include/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
+

common/nanomsg/include/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
+

common/nanomsg/include/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

common/nanomsg/include/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

common/nanomsg/iot.mk

-LIBA_TARGET 	:= libnanomsg.a
+LIBA_TARGET 		:= libapi_nanomsg.a
\ No newline at end of file

example/mqtt/iot.mk

-LIBA_TARGET 		:= libiot_mqtt_fnc.a
-
-#SRCS_example 		:= MQTTAsync_publish.c
-
-$(call Append_Conditional, TARGET, MQTTAsync_publish)
-
-#CFLAGS 			+= -I../../work/ssl/include -lrt
-#LDFLAGS 		+= -L../../work/ssl/lib
-#DEPENDS         += common/mqtt
-LDFLAGS         += -liot_mqtt -ldl
-

example/mqtt/mqtt_api.c

-#include <stdio.h>
-#include <stdlib.h>
-#include <string.h>
-#include "MQTTAsync.h"
-
-static const char* OPT_SEND = "MQTTAsync_sendMessage";
-static const char* OPT_SUB = "MQTTAsync_subscribe";
-static const char* OPT_UNSUB = "MQTTAsync_unsubscribe";
-
-static void onOptSuccess(void* context, MQTTAsync_successData* response)
-{
-	if(strcmp((char *)context,OPT_SEND)==0)
-	{
-		printf("MQTTAsync_sendMessage success,return token:%d,msg length:%d \n",
-		     response->token,response->alt.pub.message.payloadlen);
-
-	}
-	else if(strcmp((char *)context,OPT_SUB)==0)
-	{
-		printf("MQTTAsync_subscribe success,return token:%d \n",response->token);
-	}
-	else if(strcmp((char *)context,OPT_UNSUB)==0)
-	{
-		printf("MQTTAsync_unsubscribe success,return token:%d \n",response->token);
-	}
-}
-static void onOptFail(void* context,  MQTTAsync_failureData* response)
-{
-	if(strcmp((char *)context,OPT_SEND)==0)
-	{
-		printf("MQTTAsync_sendMessage fail,token:%d,code:%d,msg:%s \n",
-		     response->token,response->code,response->message);
-	}
-	else if(strcmp((char *)context,OPT_SUB)==0)
-	{
-		printf("MQTTAsync_subscribe fail,return token:%d \n",response->token);
-	}
-	else if(strcmp((char *)context,OPT_UNSUB)==0)
-	{
-		printf("MQTTAsync_unsubscribe fail,return token:%d \n",response->token);
-	}
-}
-
-int KK_MQTT_SubTopic(MQTTAsync handle,char *topicName,int qos,int waitTimeout)
-{
-	printf("to subtopic:%s \n",topicName);
-
-	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
-	int rc;
-
-	opts.onSuccess = onOptSuccess;
-	opts.onFailure = onOptFail;
-	opts.context = (void*)OPT_SUB;
-
-	if ((rc = MQTTAsync_subscribe(handle,topicName, qos, &opts)) != MQTTASYNC_SUCCESS)
-	{
-		printf("Failed to start subscribe, return code:%d.\n", rc);
-		return -1;
-	}
-	return 0;
-
-
-}
-
-int KK_MQTT_SendMsg(MQTTAsync handle,char *topicName,const char *payload,int qos)
-{
-
-	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
-	MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
-	int rc;
-
-	opts.onSuccess = onOptSuccess;
-	opts.onFailure = onOptFail;
-	opts.context =  (void*)OPT_SEND;
-
-	pubmsg.payload = (void*)payload;
-	pubmsg.payloadlen = strlen(payload);
-	pubmsg.qos = qos;
-	pubmsg.retained = 0;
-
-	printf("mqtt send payload len:%d,qos:%d.\n",pubmsg.payloadlen,qos);
-
-	if ((rc = MQTTAsync_sendMessage(handle, topicName, &pubmsg, &opts)) != MQTTASYNC_SUCCESS){
-		printf("Failed to start sendMessage, return code:%d.\n", rc);
-		return -1;
-	}
-	return rc;
-}
-int KK_MQTT_RecvMsg(MQTTAsync handle,const char *topicName,const char *payload)
-{
-	if(topicName == NULL)
-	{
-		return -1;
-	}
-	return 0;
-}
-
-int KK_MQTT_UnsubTopic(MQTTAsync handle,const char *topicName)
-{
-	printf("to unsubtopic:%s \n",topicName);
-
-	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
-	int rc;
-
-	opts.onSuccess = onOptSuccess;
-	opts.onFailure = onOptFail;
-	opts.context = (void*)OPT_UNSUB;
-
-	if ((rc = MQTTAsync_unsubscribe(handle,topicName,&opts)) != MQTTASYNC_SUCCESS){
-		printf("Failed to start unubscribe, return code:%d.\n", rc);
-		return -1;
-	}
-	return rc;
-}
-
-
-

makefile

@@ -9,13 +9,15 @@ include $(RULE_DIR)/funcs.mk
 COMP_LIB            := libiot_sdk.a
 COMP_LIB_COMPONENTS := \
 	src/test \
-    
+
+SUBDIRS                 += src/kcloud
 SUBDIRS                 += common/mqtt
-#SUBDIRS                 += common/nanomsg
-SUBDIRS                 += example/mqtt
+SUBDIRS                 += common/json
+SUBDIRS                 += common/api
+#SUBDIRS                 += example/mqtt
 SUBDIRS                 += example/nanomsg
 SUBDIRS                 += example/ev
-SUBDIRS                 += example/ev_nanomsg
+#SUBDIRS                 += example/ev_nanomsg
 
 $(call Append_Conditional, SUBDIRS)
 

example/mqtt/mqtt_config.h → src/kcloud/kcloud_config.h

@@ -5,6 +5,7 @@
 #define ADDRESS     "tcp://106.13.117.117:1883"
 #define CLIENTID    "1234"
 #define TOPIC       "cwctest"
+
 #define PAYLOAD     "Hello cwc World!"
 #define QOS         2
 #define TIMEOUT     10000L

src/kcloud/kcloud_data_handle.c

+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include "pair.h"
+#include "nn.h"
+#include "ev.h"
+#include "mqtt_api.h"
+typedef struct 
+{
+    int n;          
+    int s;          
+}kk_nanomsg_t;
+typedef struct 
+{
+    kk_nanomsg_t handle;
+}kk_nanomsg_handle_t;
+static kk_nanomsg_handle_t s_msgHdl;
+#define NANOMSG_ADDR  "ipc://b2a_loop"
+
+int KK_Sendto_DevData(const char *data, int len)
+{
+	kk_nanomsg_handle_t *msgHdl = &s_msgHdl;
+	if(len <= 0||data == NULL)
+	{
+		return -1;
+	}
+
+    int length = len + 1;
+    unsigned char *udata = nn_allocmsg(length, 0);
+    if (NULL != udata) 
+	{
+        memcpy(udata, data, length);
+        nn_send(msgHdl->handle.n, &udata, NN_MSG, NN_DONTWAIT);
+		INFO_PRINT("KK_data_send:%s \r\n\r\n", udata);
+    }	
+	
+}
+
+int KK_Sendto_CloudData(const char *data, int len)
+{
+	if(len <= 0||data == NULL)
+	{
+		return -1;
+	}
+	return KK_MQTT_SendMsg(TOPIC,data,len);
+	
+}
+static void kk_data_cb (struct ev_loop *loop ,struct ev_io *w, int revents)
+{
+	unsigned char *dat = NULL;
+    void *user_data = ev_userdata(loop);
+    kk_nanomsg_handle_t *msgHdl = (kk_nanomsg_handle_t *)user_data;
+
+    unsigned int bytes = nn_recv(msgHdl->handle.n, &dat, NN_MSG, NN_DONTWAIT);
+    if (bytes <= 0) 
+	{
+        return;
+    }
+	KK_Sendto_CloudData(dat,bytes);
+    INFO_PRINT("kk_data_cb:%s \r\n", (char *)dat);
+    nn_freemsg(dat);
+}
+
+static int kk_nanomsg_init(kk_nanomsg_handle_t *nanoMsgHdl)
+{
+	if(nanoMsgHdl == NULL)
+	{
+		return -1;
+	}
+ 	nanoMsgHdl->handle.n = nn_socket(AF_SP, NN_PAIR);	
+    if (nanoMsgHdl->handle.n < 0) 
+	{
+        return -1;
+    }	
+    if (nn_connect(nanoMsgHdl->handle.n,NANOMSG_ADDR) < 0) 
+	{
+        return -1;
+    }
+    size_t size = sizeof(size_t);
+    if (nn_getsockopt(nanoMsgHdl->handle.n, NN_SOL_SOCKET, NN_RCVFD, (char *)&nanoMsgHdl->handle.s, &size) < 0) 
+	{
+        return -1;
+    }	
+	return 0;
+}
+
+struct ev_loop* kk_loop_init(kk_nanomsg_handle_t *nanoMsgHdl)
+{
+    static struct ev_io watcher_b;
+    struct ev_loop *loop = ev_loop_new(EVBACKEND_EPOLL);
+    if (NULL == loop) 
+	{
+        ERROR_PRINT("create loop failed\r\n");
+        return NULL;
+    }
+    ev_io_init (&watcher_b, kk_data_cb, nanoMsgHdl->handle.s, EV_READ);
+    ev_io_start (loop, &watcher_b);
+    return loop;
+}
+void KK_Data_Handle(void)
+{
+	int rc = 0;
+
+	rc = kk_nanomsg_init(&s_msgHdl);
+    struct ev_loop* kloop = kk_loop_init(&s_msgHdl);
+    if (NULL == kloop) 
+	{
+        ERROR_PRINT("Bloop init failed\r\n");
+        return ;
+    }
+    ev_set_userdata(kloop, &s_msgHdl);
+    ev_run (kloop, 0);
+    return ;
+}
\ No newline at end of file

src/kcloud/kcloud_log.h

+#ifndef _LOGDEF_H_
+#define _LOGDEF_H_
+
+enum
+{
+	INFO,
+	WARNING,
+	ERROR,
+	ALL
+};
+#define DEBUG_LEVEL ERROR
+
+#define INFO_PRINT(info,...) \
+	do{ \
+		if(DEBUG_LEVEL>=INFO){	\
+			printf("Info %s,%s,%d:"info"",__FILE__,__FUNCTION__,__LINE__,##__VA_ARGS__);}\
+	}while(0)
+
+#define WARNING_PRINT(warning,...) \
+	do{ \
+		if(DEBUG_LEVEL>=WARNING){	\
+			printf("warning %s,%s,%d:"warning"",__FILE__,__FUNCTION__,__LINE__,##__VA_ARGS__);}\
+	}while(0)
+#define ERROR_PRINT(error,...) \
+	do{ \
+		if(DEBUG_LEVEL>=ERROR){	\
+			printf("error %s,%s,%d:"error"",__FILE__,__FUNCTION__,__LINE__,##__VA_ARGS__);}\
+	}while(0)	
+
+#endif
+
+

src/kcloud/kcloud_main.c

+/*******************************************************************************
+ * Copyright (c) 2012, 2020 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v2.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *   https://www.eclipse.org/legal/epl-2.0/
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial contribution
+ *******************************************************************************/
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <pthread.h>
+#include "MQTTAsync.h"
+#include "mqtt_api.h"
+
+#define THREAD_NUM  10
+static int mqtt_start(void)
+{
+	int count = 0;
+	MQTTAsync mqttClient;
+	int rc = 0;
+	mqttClient = KK_MQTT_Connect();
+	if(mqttClient == NULL)
+	{
+		WARNING_PRINT("KK_MQTT_Connect FAIL!!!\n");
+	}
+	else
+	{
+		INFO_PRINT("Waiting for publication of %s\n"
+		           "on topic %s for client with ClientID: %s\n",
+		           PAYLOAD, TOPIC, CLIENTID);
+	}
+	while(1)
+	{
+		usleep(100000L);
+		count++;
+		if(count>50)
+		{
+			count = 0;
+			INFO_PRINT("i am send now\n");
+			KK_MQTT_SendMsg(TOPIC,"hello my world",2);
+		}
+
+	}
+	INFO_PRINT("MQTTAsync_destroy\n");
+	MQTTAsync_destroy(&mqttClient);
+	return rc;
+
+}
+
+int main(int argc, char* argv[])
+{
+	int thread_count = 0;
+	pthread_t thread_num[THREAD_NUM];
+	memset(thread_num,0,sizeof(thread_num));
+
+	int rc = 0;
+	if(pthread_create(&thread_num[thread_count++],NULL,(void*)&KK_Data_Handle,NULL)!=0)
+	{
+		printf("pthread_create %d error!!!",(int)thread_num[thread_count-1]);
+	}
+	rc = mqtt_start();
+
+	return rc;
+}
+

src/kcloud/mqtt_api.h

+#ifndef _MQTT_API_H_
+#define _MQTT_API_H_
+#include "MQTTAsync.h"
+#include "kcloud_log.h"
+#include "kcloud_config.h"
+
+extern void KK_Data_Handle(void);
+
+extern int KK_MQTT_SubTopic(MQTTAsync handle,char *topicName,int qos,int waitTimeout);
+
+extern int KK_MQTT_SendMsg(char *topicName,const char *payload,int qos);
+
+extern int KK_MQTT_RecvMsg(MQTTAsync handle,const char *topicName,const char *payload);
+
+extern int KK_MQTT_UnsubTopic(MQTTAsync handle,const char *topicName);
+
+extern MQTTAsync KK_MQTT_Connect(void);
+
+#endif
+

src/test/iot.mk

 LIBA_TARGET 		:= libiot_test.a
+

src/test/test.c

-/*******************************************************************************
- * Copyright (c) 2012, 2020 IBM Corp.
- *
- * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v2.0
- * and Eclipse Distribution License v1.0 which accompany this distribution. 
- *
- * The Eclipse Public License is available at 
- *   https://www.eclipse.org/legal/epl-2.0/
- * and the Eclipse Distribution License is available at 
- *   http://www.eclipse.org/org/documents/edl-v10.php.
- *
- * Contributors:
- *    Ian Craggs - initial contribution
- *******************************************************************************/
-
-#include <stdio.h>
-int test()
+#include<stdio.h>
+int test(void)
 {
-	return 0;
+	return 1;
 }