【修改内容】1. midwareu与platform(或网关)交换通过nanomsg的pipe（plat或gw ->...

【修改内容】1. midwareu与platform(或网关)交换通过nanomsg的pipe（plat或gw -> midware(ccu)）和subpub(midware(ccu) -> plat或gw)的方式交互；2. platform或gw初始化nanomsg是需要传递唯一标识作为通道标志，建议用gw的mac地址；3. application与midware用nanomsg的pair协议
【提交人】huang.zhenling

application/kcloud/kcloud_main.c

@@ -68,7 +68,7 @@ int main(int argc, char* argv[])
 	/*set the callback to get the device date to cloud*/
 	HAL_SetProduct_Type(PRODUCT_TPYE);
 	HAL_SetProduct_Code(PRODUCT_CODE);	
-	kk_ipc_init(IPC_APP2MID,KK_Data_FromDev);
+	kk_ipc_init(IPC_APP2MID,KK_Data_FromDev,NULL,NULL);
 	rc = mqtt_start();
 
 	return rc;

common/api/com_api.h

@@ -15,7 +15,8 @@ extern "C" {
 #include "ev.h"
 #include "nn.h"
 #include "pair.h"
-
+#include "pubsub.h"
+#include "pipeline.h"
 
 
 //=====kk======================
@@ -62,10 +63,11 @@ typedef enum{
 #define MSG_INFO_STR              "info"
 #define MSG_INDENTIFIER_STR              "identifier"
 
-typedef void ipc_cb(void* data, int len);
-int kk_ipc_init(ipc_type type, ipc_cb cb);
+typedef void ipc_cb(void* data, int len, char* chalMark);
+int kk_ipc_init(ipc_type type, ipc_cb cb, char* chalMark, char* ip);
 int kk_ipc_dinit();
 int kk_ipc_send(ipc_type type, void* data, int len);
+int kk_ipc_send_ex(ipc_type type, void* data, int len, char* chalMark);
 
 #if defined(__cplusplus)
 }

common/api/iot.mk

 LIBSO_TARGET 		:= libapi_com.so
-
+LIBA_TARGET 		:= libapi_com.a
 
 CFLAGS 			+= -I$(TOP_DIR)/common/nanomsg/include
 CFLAGS 			+= -I$(TOP_DIR)/common/ev/include

common/zlog/iot.mk

 LIBSO_TARGET 		:= libzlog.so
+LIBA_TARGET 		:= libzlog.a
 

common/zlog/kk_log.c

 #include "kk_log.h"
 
-
 zlog_category_t *g_zlogC;
 int kk_zlog_init(char* module)
 {
@@ -8,7 +7,7 @@ int kk_zlog_init(char* module)
 	rc = zlog_init("zlog.conf");
 	
 	if (rc) {
-		printf("init failed\n");
+		printf(" kk_zlog_init init failed\n");
 		return -1;
 	
 	}
@@ -16,7 +15,7 @@ int kk_zlog_init(char* module)
 	g_zlogC = zlog_get_category(module);
 	
 	if (!g_zlogC) {
-		printf("get cat fail\n");
+		printf("kk_zlog_init get cat fail\n");
 		zlog_fini();
 		return -2;
 	}

midware/midware/dm/kk_dm_queue.h

@@ -11,7 +11,8 @@
 #include "klist.h"
 
 typedef struct {
-    iotx_dm_event_types_t type;
+    //iotx_dm_event_types_t type;
+    char* chalMark[30];
     char *data;
 } dm_queue_msg_t;
 

midware/midware/midware.c

@@ -74,7 +74,7 @@ void mid_cb(void* data, int len){
             cJSON_AddStringToObject(jsonplay, "mac", mac);
             void* out =  cJSON_Print(jsonplay);
             printf("dm_mgr_search_mac_by_topic out: %s  \r\n", out);
-            kk_ipc_send(IPC_MID2PLAT, out, strlen(out));
+            kk_ipc_send_ex(IPC_MID2PLAT, out, strlen(out), "1122334455667788");
             free(out);
             cJSON_Delete(jsonplay);
             cJSON_Delete(json);
@@ -84,13 +84,32 @@ void mid_cb(void* data, int len){
 	}
 }
 
-void mid2p_cb(void* data, int len){
+void mid2p_cb(void* data, int len, char* chalMark){
 	if (data != NULL){
 		//printf("mid2plat_cb: %s RECEIVED \r\n", data);
-		void* buf = malloc(len);
+		int res = 0;
+		void* buf = NULL;
+		dm_queue_msg_t *queue_msg = NULL;
+		queue_msg = malloc(sizeof(dm_queue_msg_t));
+		if (queue_msg == NULL){
+            ERROR_PRINT("mid2p_cb malloc queue_msg failed ");
+            return;
+		}
+		buf = malloc(len);
+		if (buf == NULL){
+            ERROR_PRINT("mid2p_cb malloc buf failed ");
+            return;
+		}
+		
         memcpy(buf, data, len);
-        int res = dm_queue_msg_insert2((void *)buf);
+        queue_msg.data = buf;
+        memset(queue_msg.chalMark, 0, sizeof(queue_msg.chalMark));
+        if(chalMark != NULL){
+            memcpy(queue_msg.chalMark, chalMark, sizeof(chalMark));
+        }
+        res = dm_queue_msg_insert2((void *)queue_msg);
         if (res != SUCCESS_RETURN) {
+            free(queue_msg);
             free(buf);
             return ;
         }
@@ -99,7 +118,7 @@ void mid2p_cb(void* data, int len){
 	}
 }
 
-void kk_platMsg_handle(void* data){
+void kk_platMsg_handle(void* data, char* chalMark){
     
     char *out;
     int res = 0;
@@ -149,13 +168,15 @@ void kk_platMsg_dispatch(void)
     while (CONFIG_DISPATCH_QUEUE_MAXLEN == 0 || count++ < CONFIG_DISPATCH_QUEUE_MAXLEN) {
 
         if (dm_queue_msg_next2(&data) == SUCCESS_RETURN) {
-            //dm_queue_msg_t *msg = (dm_queue_msg_t *)data;
+            dm_queue_msg_t *msg = (dm_queue_msg_t *)data;
             INFO_PRINT("kk_handle_platMsg_dispatch get call \n");
             if (kk_platMsg_handle) {
-                kk_platMsg_handle( data);
+                kk_platMsg_handle(msg->data,msg->chalMark);
             }
 
-            
+            if (msg->data != NULL){
+                free(msg->data);
+            }
             free(data);
             data = NULL;
         } else {
@@ -239,8 +260,8 @@ int main(const int argc, const char **argv)
 		kk_set_product_info();
         kk_tsl_api_init();
 
-        kk_ipc_init(IPC_MID2APP, mid_cb);
-        kk_ipc_init(IPC_MID2PLAT, mid2p_cb);
+        kk_ipc_init(IPC_MID2APP, mid_cb, NULL, NULL);
+        kk_ipc_init(IPC_MID2PLAT, mid2p_cb, NULL, "*");
 		
 
 

platform/zigbee/app/builder/Z3GatewayHost/com_api.h

@@ -15,7 +15,8 @@ extern "C" {
 #include "ev.h"
 #include "nn.h"
 #include "pair.h"
-
+#include "pubsub.h"
+#include "pipeline.h"
 
 
 //=====kk======================
@@ -26,15 +27,48 @@ typedef enum {
     IPC_PLAT2MID,
     IPC_UNDEF
 } ipc_type;
+	
+typedef enum{
+	/******MIDDWARE TO APP**************/
+	MSG_REGISTER = 0,
+	MSG_UNREGISTER,
+	MSG_TOPOADD,
+	MSG_TOPODELETE,
+	MSG_TOPOGET,
+	MSG_LISTFOUND,
+	MSG_LOGIN,
+	MSG_LOGOUT,
+	MSG_PROPERTYPOST,
+	MSG_EVENTPOST,
+	MSG_SERVICERESPONSE,
+	MSG_SETREPLY,
+	MSG_OTA_PROCESS,
+	MSG_OTA_INFORM,
+	/*******APP TO  MIDDWARE**************/
+ 	MSG_REGISTER_REPLY,
+ 	MSG_TOPOADD_REPLY,
+ 	MSG_OFFLINE_REPLY,
+ 	MSG_LOGIN_REPLY,
+ 	MSG_PROPERTYSET,
+ 	MSG_OTA_UPGRADE,
+ 	MSG_INVALID,
+ 	
+}kk_msg_type_t;
+
+#define MSG_TYPE_STR              "msgtype"
+#define MSG_PRODUCT_TYPE_STR      "product_type"
+#define MSG_DEVICE_NAME_STR       "device_name"
+#define MSG_PAYLOAD_STR           "payload"
+#define MSG_INFO_STR              "info"
+#define MSG_INDENTIFIER_STR              "identifier"
 
-typedef void ipc_cb(void* data, int len);
-int kk_ipc_init(ipc_type type, ipc_cb cb);
+typedef void ipc_cb(void* data, int len, char* chlMark);
+int kk_ipc_init(ipc_type type, ipc_cb cb, char* mac, char* ip);
 int kk_ipc_dinit();
 int kk_ipc_send(ipc_type type, void* data, int len);
+int kk_ipc_send_ex(ipc_type type, void* data, int len, char* filter);
 
 #if defined(__cplusplus)
 }
 #endif
 #endif
-
-

platform/zigbee/app/builder/Z3GatewayHost/rpc_api/src/rpc_interface_parse.c

@@ -21,6 +21,8 @@
 
 #include "com_api.h"
 #include "kk_test.h"
+//#include "kk_log.h"
+
 
 static struct jrpc_server my_server;
 
@@ -219,6 +221,7 @@ void _cb(void* data){
 
 int _init_param(struct jrpc_server *server) {
 	memset(server, 0, sizeof(struct jrpc_server));
+	//kk_zlog_init("paltform");
 
 	printf("getenv\r\n");
 	char * debug_level_env = getenv("HOME");
@@ -237,7 +240,7 @@ void ipcHandle(void)
 {
 	emberAfAppPrint( "Thread rpc Interface Parse create\n" );
     _init_param(&my_server);
-    kk_ipc_init(IPC_PLAT2MID, _cb);
+    kk_ipc_init(IPC_PLAT2MID, _cb, "1122334455667788","127.0.0.1");
 	emberAfAppPrint("sizeof(rpc_table)=%d,sizeof(rpc_table_s)=%d,%d\n",sizeof(rpc_table),sizeof(rpc_table_s),sizeof(rpc_table)/sizeof(rpc_table_s));
 	for(int i=0;i<sizeof(rpc_table)/sizeof(rpc_table_s);i++){
 		emberAfAppPrint("i=%d,%s\r\n",i,rpc_table[i].name);