From: Yechan Choi <yechan9.choi@samsung.com>
Date: Fri, 12 Aug 2022 11:59:59 +0000 (+0900)
Subject: [Edge] support edge
X-Git-Tag: accepted/tizen/unified/20220919.012623~8
X-Git-Url: http://review.tizen.org/git/?a=commitdiff_plain;h=516e36becea6fbdb1e4409a1021be3e068061047;p=platform%2Fupstream%2Fnnstreamer-edge.git

[Edge] support edge

- If no client_id is found in the data, publish it to all connected clients.
- Add NNS_EDGE_FLAG_PUBSUB.

Signed-off-by: Yechan Choi <yechan9.choi@samsung.com>
Signed-off-by: gichan <gichan2.jang@samsung.com>
---

diff --git a/include/nnstreamer-edge.h b/include/nnstreamer-edge.h
index fd6b2dd..193d447 100644
--- a/include/nnstreamer-edge.h
+++ b/include/nnstreamer-edge.h
@@ -63,13 +63,13 @@ typedef enum {
 } nns_edge_connect_type_e;
 
 typedef enum {
-  NNS_EDGE_FLAG_NONE = 0,
-  NNS_EDGE_FLAG_RECV = (1 << 0),
-  NNS_EDGE_FLAG_SEND = (1 << 1),
-  NNS_EDGE_FLAG_SERVER = (1 << 2),
+  NNS_EDGE_NODE_TYPE_QUERY_CLIENT = 0,
+  NNS_EDGE_NODE_TYPE_QUERY_SERVER,
+  NNS_EDGE_NODE_TYPE_PUB,
+  NNS_EDGE_NODE_TYPE_SUB,
 
-  NNS_EDGE_FLAG_ALL = (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER)
-} nns_edge_flag_e;
+  NNS_EDGE_NODE_TYPE_UNKNOWN,
+} nns_edge_node_type_e;
 
 /**
  * @brief Callback for the nnstreamer edge event.
@@ -172,7 +172,7 @@ typedef void (*nns_edge_data_destroy_cb) (void *data);
  * // Release edge handle.
  * nns_edge_release_handle (edge_h);
  */
-int nns_edge_create_handle (const char *id, nns_edge_connect_type_e connect_type, int flags, nns_edge_h *edge_h);
+int nns_edge_create_handle (const char *id, nns_edge_connect_type_e connect_type, nns_edge_node_type_e node_type, nns_edge_h *edge_h);
 
 /**
  * @brief Start the nnstreamer edge. After the start, the edge can accept a new connection or request a connection.
diff --git a/src/libnnstreamer-edge/nnstreamer-edge-internal.c b/src/libnnstreamer-edge/nnstreamer-edge-internal.c
index 1a7bc20..00cd80f 100644
--- a/src/libnnstreamer-edge/nnstreamer-edge-internal.c
+++ b/src/libnnstreamer-edge/nnstreamer-edge-internal.c
@@ -98,6 +98,12 @@ typedef struct
   nns_edge_conn_s *conn;
 } nns_edge_thread_data_s;
 
+/**
+ * @brief Create message handle thread.
+ */
+static int
+_nns_edge_create_message_thread (nns_edge_handle_s * eh, nns_edge_conn_s * conn,
+    int64_t client_id);
 /**
  * @brief Set socket option.
  * @todo handle connection type (TCP/UDP).
@@ -148,7 +154,7 @@ _send_raw_data (nns_edge_conn_s * conn, void *data, size_t size)
   ssize_t rret;
 
   while (sent < size) {
-    rret = send (conn->sockfd, (char *) data + sent, size - sent, 0);
+    rret = send (conn->sockfd, (char *) data + sent, size - sent, MSG_NOSIGNAL);
 
     if (rret <= 0) {
       nns_edge_loge ("Failed to send raw data.");
@@ -619,6 +625,34 @@ _nns_edge_remove_connection (nns_edge_handle_s * eh, int64_t client_id)
       _nns_edge_release_connection_data (cdata);
       return;
     }
+    prev = cdata;
+    cdata = cdata->next;
+  }
+}
+
+/**
+ * @brief Remove nnstreamer-edge connection data.
+ * @note This function should be called with handle lock.
+ */
+static void
+_nns_edge_remove_connection_by_conn (nns_edge_handle_s * eh,
+    nns_edge_conn_s * conn)
+{
+  nns_edge_conn_data_s *cdata, *prev;
+
+  cdata = (nns_edge_conn_data_s *) eh->connections;
+  prev = NULL;
+
+  while (cdata) {
+    if (cdata->sink_conn == conn || cdata->src_conn == conn) {
+      if (prev)
+        prev->next = cdata->next;
+      else
+        eh->connections = cdata->next;
+
+      _nns_edge_release_connection_data (cdata);
+      return;
+    }
 
     prev = cdata;
     cdata = cdata->next;
@@ -676,95 +710,6 @@ _nns_edge_connect_socket (nns_edge_conn_s * conn)
   return true;
 }
 
-/**
- * @brief Connect to the destination node. (host:sender(sink) - dest:receiver(listener, src))
- */
-static int
-_nns_edge_connect_to (nns_edge_handle_s * eh, int64_t client_id,
-    const char *host, int port)
-{
-  nns_edge_conn_s *conn = NULL;
-  nns_edge_conn_data_s *conn_data;
-  nns_edge_cmd_s cmd;
-  char *host_str;
-  bool done = false;
-  int ret;
-
-  conn = (nns_edge_conn_s *) calloc (1, sizeof (nns_edge_conn_s));
-  if (!conn) {
-    nns_edge_loge ("Failed to allocate client data.");
-    goto error;
-  }
-
-  conn->host = nns_edge_strdup (host);
-  conn->port = port;
-  conn->sockfd = -1;
-
-  if (!_nns_edge_connect_socket (conn)) {
-    goto error;
-  }
-
-  if (!(eh->flags & NNS_EDGE_FLAG_SERVER)) {
-    /* Receive capability and client ID from server. */
-    _nns_edge_cmd_init (&cmd, _NNS_EDGE_CMD_ERROR, client_id);
-    ret = _nns_edge_cmd_receive (conn, &cmd);
-    if (ret != NNS_EDGE_ERROR_NONE) {
-      nns_edge_loge ("Failed to receive capability.");
-      goto error;
-    }
-
-    if (cmd.info.cmd != _NNS_EDGE_CMD_CAPABILITY) {
-      nns_edge_loge ("Failed to get capability.");
-      _nns_edge_cmd_clear (&cmd);
-      goto error;
-    }
-
-    client_id = eh->client_id = cmd.info.client_id;
-
-    /* Check compatibility. */
-    ret = _nns_edge_invoke_event_cb (eh, NNS_EDGE_EVENT_CAPABILITY,
-        cmd.mem[0], cmd.info.mem_size[0], NULL);
-    _nns_edge_cmd_clear (&cmd);
-
-    if (ret != NNS_EDGE_ERROR_NONE) {
-      nns_edge_loge ("The event returns error, capability is not acceptable.");
-      _nns_edge_cmd_init (&cmd, _NNS_EDGE_CMD_ERROR, client_id);
-    } else {
-      /* Send host and port to destination. */
-      _nns_edge_cmd_init (&cmd, _NNS_EDGE_CMD_HOST_INFO, client_id);
-
-      host_str = nns_edge_get_host_string (eh->host, eh->port);
-      cmd.info.num = 1;
-      cmd.info.mem_size[0] = strlen (host_str) + 1;
-      cmd.mem[0] = host_str;
-    }
-
-    ret = _nns_edge_cmd_send (conn, &cmd);
-    _nns_edge_cmd_clear (&cmd);
-
-    if (ret != NNS_EDGE_ERROR_NONE) {
-      nns_edge_loge ("Failed to send host info.");
-      goto error;
-    }
-  }
-
-  conn_data = _nns_edge_add_connection (eh, client_id);
-  if (conn_data) {
-    /* Close old connection and set new one. */
-    _nns_edge_close_connection (conn_data->sink_conn);
-    conn_data->sink_conn = conn;
-    done = true;
-  }
-
-error:
-  if (!done) {
-    _nns_edge_close_connection (conn);
-    return NNS_EDGE_ERROR_CONNECTION_FAILURE;
-  }
-
-  return NNS_EDGE_ERROR_NONE;
-}
-
 /**
  * @brief Message thread, receive buffer from the client.
  */
@@ -931,10 +876,12 @@ _nns_edge_send_thread (void *thread_data)
 
           conn_data = (nns_edge_conn_data_s *) eh->connections;
           while (conn_data) {
-            /** @todo update code for each connect type */
             conn = conn_data->sink_conn;
-            _nns_edge_transfer_data (conn, data_h, conn_data->id);
-
+            ret = _nns_edge_transfer_data (conn, data_h, conn_data->id);
+            if (NNS_EDGE_ERROR_NONE != ret) {
+              nns_edge_loge ("Failed to transfer data. Close the connection.");
+              _nns_edge_remove_connection_by_conn (eh, conn);
+            }
             conn_data = conn_data->next;
           }
         } else {
@@ -959,7 +906,6 @@ _nns_edge_send_thread (void *thread_data)
     }
     nns_edge_data_destroy (data_h);
   }
-
   return NULL;
 }
 
@@ -987,6 +933,104 @@ _nns_edge_create_send_thread (nns_edge_handle_s * eh)
   return NNS_EDGE_ERROR_NONE;
 }
 
+/**
+ * @brief Connect to the destination node. (host:sender(sink) - dest:receiver(listener, src))
+ */
+static int
+_nns_edge_connect_to (nns_edge_handle_s * eh, int64_t client_id,
+    const char *host, int port)
+{
+  nns_edge_conn_s *conn = NULL;
+  nns_edge_conn_data_s *conn_data;
+  nns_edge_cmd_s cmd;
+  char *host_str;
+  bool done = false;
+  int ret;
+
+  conn = (nns_edge_conn_s *) calloc (1, sizeof (nns_edge_conn_s));
+  if (!conn) {
+    nns_edge_loge ("Failed to allocate client data.");
+    goto error;
+  }
+
+  conn->host = nns_edge_strdup (host);
+  conn->port = port;
+  conn->sockfd = -1;
+
+  if (!_nns_edge_connect_socket (conn)) {
+    goto error;
+  }
+
+  if ((NNS_EDGE_NODE_TYPE_QUERY_CLIENT == eh->node_type)
+      || (NNS_EDGE_NODE_TYPE_SUB == eh->node_type)) {
+    /* Receive capability and client ID from server. */
+    _nns_edge_cmd_init (&cmd, _NNS_EDGE_CMD_ERROR, client_id);
+    ret = _nns_edge_cmd_receive (conn, &cmd);
+    if (ret != NNS_EDGE_ERROR_NONE) {
+      nns_edge_loge ("Failed to receive capability.");
+      goto error;
+    }
+
+    if (cmd.info.cmd != _NNS_EDGE_CMD_CAPABILITY) {
+      nns_edge_loge ("Failed to get capability.");
+      _nns_edge_cmd_clear (&cmd);
+      goto error;
+    }
+
+    client_id = eh->client_id = cmd.info.client_id;
+
+    /* Check compatibility. */
+    ret = _nns_edge_invoke_event_cb (eh, NNS_EDGE_EVENT_CAPABILITY,
+        cmd.mem[0], cmd.info.mem_size[0], NULL);
+    _nns_edge_cmd_clear (&cmd);
+
+    if (ret != NNS_EDGE_ERROR_NONE) {
+      nns_edge_loge ("The event returns error, capability is not acceptable.");
+      _nns_edge_cmd_init (&cmd, _NNS_EDGE_CMD_ERROR, client_id);
+    } else {
+      /* Send host and port to destination. */
+      _nns_edge_cmd_init (&cmd, _NNS_EDGE_CMD_HOST_INFO, client_id);
+
+      host_str = nns_edge_get_host_string (eh->host, eh->port);
+      cmd.info.num = 1;
+      cmd.info.mem_size[0] = strlen (host_str) + 1;
+      cmd.mem[0] = host_str;
+    }
+
+    ret = _nns_edge_cmd_send (conn, &cmd);
+    _nns_edge_cmd_clear (&cmd);
+
+    if (ret != NNS_EDGE_ERROR_NONE) {
+      nns_edge_loge ("Failed to send host info.");
+      goto error;
+    }
+  }
+
+  conn_data = _nns_edge_add_connection (eh, client_id);
+  if (conn_data) {
+    /* Close old connection and set new one. */
+    _nns_edge_close_connection (conn_data->sink_conn);
+    conn_data->sink_conn = conn;
+    done = true;
+  }
+
+  if (NNS_EDGE_NODE_TYPE_SUB == eh->node_type) {
+    ret = _nns_edge_create_message_thread (eh, conn, client_id);
+    if (ret != NNS_EDGE_ERROR_NONE) {
+      nns_edge_loge ("Failed to create message handle thread.");
+      goto error;
+    }
+  }
+
+error:
+  if (!done) {
+    _nns_edge_close_connection (conn);
+    return NNS_EDGE_ERROR_CONNECTION_FAILURE;
+  }
+
+  return NNS_EDGE_ERROR_NONE;
+}
+
 /**
  * @brief Accept socket and create message thread in socket listener thread.
  */
@@ -1015,13 +1059,16 @@ _nns_edge_accept_socket (nns_edge_handle_s * eh)
 
   _set_socket_option (conn->sockfd);
 
-  if (eh->flags & NNS_EDGE_FLAG_SERVER)
+  if ((NNS_EDGE_NODE_TYPE_QUERY_SERVER == eh->node_type)
+      || (NNS_EDGE_NODE_TYPE_PUB == eh->node_type)) {
     client_id = nns_edge_generate_client_id ();
-  else
+  } else {
     client_id = eh->client_id;
+  }
 
   /* Send capability and info to check compatibility. */
-  if (eh->flags & NNS_EDGE_FLAG_SERVER) {
+  if ((NNS_EDGE_NODE_TYPE_QUERY_SERVER == eh->node_type)
+      || (NNS_EDGE_NODE_TYPE_PUB == eh->node_type)) {
     if (!STR_IS_VALID (eh->caps_str)) {
       nns_edge_loge ("Cannot accept socket, invalid server capability.");
       goto error;
@@ -1037,7 +1084,9 @@ _nns_edge_accept_socket (nns_edge_handle_s * eh)
       nns_edge_loge ("Failed to send capability.");
       goto error;
     }
+  }
 
+  if (NNS_EDGE_NODE_TYPE_QUERY_SERVER == eh->node_type) {
     /* Receive host info from destination. */
     ret = _nns_edge_cmd_receive (conn, &cmd);
     if (ret != NNS_EDGE_ERROR_NONE) {
@@ -1062,18 +1111,23 @@ _nns_edge_accept_socket (nns_edge_handle_s * eh)
     }
   }
 
-  ret = _nns_edge_create_message_thread (eh, conn, client_id);
-  if (ret != NNS_EDGE_ERROR_NONE) {
-    nns_edge_loge ("Failed to create message handle thread.");
-    goto error;
-  }
-
   conn_data = _nns_edge_add_connection (eh, client_id);
-  if (conn_data) {
-    /* Close old connection and set new one. */
+
+  if (eh->node_type == NNS_EDGE_NODE_TYPE_QUERY_CLIENT ||
+      eh->node_type == NNS_EDGE_NODE_TYPE_QUERY_SERVER) {
+    ret = _nns_edge_create_message_thread (eh, conn, client_id);
+    if (ret != NNS_EDGE_ERROR_NONE) {
+      nns_edge_loge ("Failed to create message handle thread.");
+      goto error;
+    }
     _nns_edge_close_connection (conn_data->src_conn);
     conn_data->src_conn = conn;
     done = true;
+  } else {
+    /* Close old connection and set new one. */
+    _nns_edge_close_connection (conn_data->sink_conn);
+    conn_data->sink_conn = conn;
+    done = true;
   }
 
 error:
@@ -1177,7 +1231,7 @@ error:
  */
 int
 nns_edge_create_handle (const char *id, nns_edge_connect_type_e connect_type,
-    int flags, nns_edge_h * edge_h)
+    nns_edge_node_type_e node_type, nns_edge_h * edge_h)
 {
   nns_edge_handle_s *eh;
 
@@ -1195,8 +1249,8 @@ nns_edge_create_handle (const char *id, nns_edge_connect_type_e connect_type,
    * @todo handle flag (receive | send)
    * e.g., send only case: listener is unnecessary.
    */
-  if (flags <= 0 || !(flags & NNS_EDGE_FLAG_ALL)) {
-    nns_edge_loge ("Invalid param, set exact edge flags.");
+  if (node_type < 0 || node_type >= NNS_EDGE_NODE_TYPE_UNKNOWN) {
+    nns_edge_loge ("Invalid param, set exact node type.");
     return NNS_EDGE_ERROR_INVALID_PARAMETER;
   }
 
@@ -1219,7 +1273,7 @@ nns_edge_create_handle (const char *id, nns_edge_connect_type_e connect_type,
   eh->port = 0;
   eh->dest_host = nns_edge_strdup ("localhost");
   eh->dest_port = 0;
-  eh->flags = flags;
+  eh->node_type = node_type;
   eh->broker_h = NULL;
   eh->connections = NULL;
   nns_edge_metadata_init (&eh->meta);
@@ -1263,7 +1317,8 @@ nns_edge_start (nns_edge_h edge_h)
     }
   }
 
-  if (eh->flags & NNS_EDGE_FLAG_SERVER) {
+  if ((NNS_EDGE_NODE_TYPE_QUERY_SERVER == eh->node_type)
+      || (NNS_EDGE_NODE_TYPE_PUB == eh->node_type)) {
     if (NNS_EDGE_CONNECT_TYPE_HYBRID == eh->connect_type) {
       char *topic, *msg;
 
@@ -1301,17 +1356,18 @@ nns_edge_start (nns_edge_h edge_h)
     }
   }
 
-  /* Start listener thread to accept socket. */
-  if (NNS_EDGE_CONNECT_TYPE_TCP == eh->connect_type ||
-      NNS_EDGE_CONNECT_TYPE_HYBRID == eh->connect_type) {
+  if ((NNS_EDGE_NODE_TYPE_QUERY_CLIENT == eh->node_type)
+      || (NNS_EDGE_NODE_TYPE_QUERY_SERVER == eh->node_type)
+      || (NNS_EDGE_NODE_TYPE_PUB == eh->node_type)) {
+    /* Start listener thread to accept socket. */
     if (!_nns_edge_create_socket_listener (eh)) {
       nns_edge_loge ("Failed to create socket listener.");
       ret = NNS_EDGE_ERROR_IO;
       goto done;
     }
-  }
 
-  ret = _nns_edge_create_send_thread (eh);
+    ret = _nns_edge_create_send_thread (eh);
+  }
 
 done:
   nns_edge_unlock (eh);
@@ -1331,7 +1387,6 @@ nns_edge_release_handle (nns_edge_h edge_h)
     nns_edge_loge ("Invalid param, given edge handle is null.");
     return NNS_EDGE_ERROR_INVALID_PARAMETER;
   }
-
   nns_edge_lock (eh);
 
   if (!NNS_EDGE_MAGIC_IS_VALID (eh)) {
@@ -1744,7 +1799,8 @@ nns_edge_get_info (nns_edge_h edge_h, const char *key, char **value)
   } else if (0 == strcasecmp (key, "DEST_PORT")) {
     *value = nns_edge_strdup_printf ("%d", eh->dest_port);
   } else if (0 == strcasecmp (key, "CLIENT_ID")) {
-    if (eh->flags & NNS_EDGE_FLAG_SERVER) {
+    if ((NNS_EDGE_NODE_TYPE_QUERY_SERVER == eh->node_type)
+        || (NNS_EDGE_NODE_TYPE_PUB == eh->node_type)) {
       nns_edge_loge ("Cannot get the client ID, it was started as a server.");
       ret = NNS_EDGE_ERROR_INVALID_PARAMETER;
     } else {
diff --git a/src/libnnstreamer-edge/nnstreamer-edge-internal.h b/src/libnnstreamer-edge/nnstreamer-edge-internal.h
index bd72251..f46c509 100644
--- a/src/libnnstreamer-edge/nnstreamer-edge-internal.h
+++ b/src/libnnstreamer-edge/nnstreamer-edge-internal.h
@@ -76,7 +76,7 @@ typedef struct {
   int port; /**< port number (0~65535, default 0 to get available port.) */
   char *dest_host; /**< destination IP address (broker or target device) */
   int dest_port; /**< destination port number (broker or target device) */
-  int flags;
+  nns_edge_node_type_e node_type;
   nns_edge_metadata_s meta;
 
   /* Edge event callback and user data */
diff --git a/tests/unittest_nnstreamer-edge-aitt.cc b/tests/unittest_nnstreamer-edge-aitt.cc
index 1520319..53d67df 100644
--- a/tests/unittest_nnstreamer-edge-aitt.cc
+++ b/tests/unittest_nnstreamer-edge-aitt.cc
@@ -154,7 +154,7 @@ TEST(edgeAitt, connectLocal)
   /* Prepare server (127.0.0.1:port) */
   val = nns_edge_strdup_printf ("%d", port);
   nns_edge_create_handle ("temp-sender", NNS_EDGE_CONNECT_TYPE_AITT,
-      (NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER), &server_h);
+      NNS_EDGE_NODE_TYPE_PUB, &server_h);
   nns_edge_set_info (server_h, "IP", "127.0.0.1");
   nns_edge_set_info (server_h, "PORT", val);
   nns_edge_set_info (server_h, "DEST_IP", "127.0.0.1");
@@ -165,13 +165,13 @@ TEST(edgeAitt, connectLocal)
 
   /* Prepare client */
   nns_edge_create_handle ("temp-receiver", NNS_EDGE_CONNECT_TYPE_AITT,
-      (NNS_EDGE_FLAG_RECV), &client1_h);
+      NNS_EDGE_NODE_TYPE_SUB, &client1_h);
   nns_edge_set_event_callback (client1_h, _test_edge_event_cb, _td_client1);
   nns_edge_set_info (client1_h, "TOPIC", "AITT_TEST_TOPIC");
   _td_client1->handle = client1_h;
 
   nns_edge_create_handle ("temp-client2", NNS_EDGE_CONNECT_TYPE_AITT,
-      (NNS_EDGE_FLAG_RECV), &client2_h);
+      NNS_EDGE_NODE_TYPE_SUB, &client2_h);
   nns_edge_set_event_callback (client2_h, _test_edge_event_cb, _td_client2);
   nns_edge_set_info (client2_h, "TOPIC", "AITT_TEST_TOPIC");
   _td_client2->handle = client2_h;
@@ -278,7 +278,7 @@ TEST(edgeAitt, connectInvalidParam2_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_AITT,
-      NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &edge_h);
+      NNS_EDGE_NODE_TYPE_PUB, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
   nns_edge_set_info (edge_h, "DEST_IP", "f.a.i.l");
   nns_edge_set_info (edge_h, "DEST_PORT", "1883");
@@ -303,7 +303,7 @@ TEST(edgeAitt, connectInvalidParam3_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_AITT,
-      NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &edge_h);
+      NNS_EDGE_NODE_TYPE_PUB, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
   nns_edge_set_info (edge_h, "DEST_IP", "127.0.0.1");
   nns_edge_set_info (edge_h, "DEST_PORT", "0");
@@ -343,7 +343,7 @@ TEST(edgeAitt, closeInvalidParam2_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_AITT,
-      NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &edge_h);
+      NNS_EDGE_NODE_TYPE_PUB, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_aitt_close (edge_h);
@@ -381,7 +381,7 @@ TEST(edgeAitt, publishInvalidParam2_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_AITT,
-      NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &edge_h);
+      NNS_EDGE_NODE_TYPE_PUB, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
   nns_edge_set_info (edge_h, "DEST_HOST", "127.0.0.1");
   nns_edge_set_info (edge_h, "DEST_PORT", "1883");
@@ -411,7 +411,7 @@ TEST(edgeAitt, publishInvalidParam3_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_AITT,
-      NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &edge_h);
+      NNS_EDGE_NODE_TYPE_PUB, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
   nns_edge_set_info (edge_h, "DEST_HOST", "127.0.0.1");
   nns_edge_set_info (edge_h, "DEST_PORT", "1883");
@@ -441,7 +441,7 @@ TEST(edgeAitt, publishInvalidParam4_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_AITT,
-      NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &edge_h);
+      NNS_EDGE_NODE_TYPE_PUB, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
   nns_edge_set_info (edge_h, "DEST_HOST", "127.0.0.1");
   nns_edge_set_info (edge_h, "DEST_PORT", "1883");
@@ -479,7 +479,7 @@ TEST(edgeAitt, subscribeInvalidParam2_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_AITT,
-      NNS_EDGE_FLAG_RECV, &edge_h);
+      NNS_EDGE_NODE_TYPE_SUB, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   nns_edge_set_info (edge_h, "DEST_HOST", "127.0.0.1");
diff --git a/tests/unittest_nnstreamer-edge.cc b/tests/unittest_nnstreamer-edge.cc
index 4e9a1a1..b27699d 100644
--- a/tests/unittest_nnstreamer-edge.cc
+++ b/tests/unittest_nnstreamer-edge.cc
@@ -152,7 +152,7 @@ TEST(edge, connectLocal)
   /* Prepare server (127.0.0.1:port) */
   val = nns_edge_strdup_printf ("%d", port);
   nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER), &server_h);
+      NNS_EDGE_NODE_TYPE_QUERY_SERVER, &server_h);
   nns_edge_set_event_callback (server_h, _test_edge_event_cb, _td_server);
   nns_edge_set_info (server_h, "IP", "127.0.0.1");
   nns_edge_set_info (server_h, "PORT", val);
@@ -162,13 +162,13 @@ TEST(edge, connectLocal)
 
   /* Prepare client */
   nns_edge_create_handle ("temp-client1", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &client1_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &client1_h);
   nns_edge_set_event_callback (client1_h, _test_edge_event_cb, _td_client1);
   nns_edge_set_info (client1_h, "CAPS", "test client1");
   _td_client1->handle = client1_h;
 
   nns_edge_create_handle ("temp-client2", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &client2_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &client2_h);
   nns_edge_set_event_callback (client2_h, _test_edge_event_cb, _td_client2);
   nns_edge_set_info (client2_h, "CAPS", "test client2");
   _td_client2->handle = client2_h;
@@ -271,7 +271,7 @@ TEST(edge, createHandleInvalidParam01_n)
   int ret;
 
   ret = nns_edge_create_handle (NULL, NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_NE (ret, NNS_EDGE_ERROR_NONE);
 }
 
@@ -284,7 +284,7 @@ TEST(edge, createHandleInvalidParam02_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_UNKNOWN,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_NE (ret, NNS_EDGE_ERROR_NONE);
 }
 
@@ -296,7 +296,7 @@ TEST(edge, createHandleInvalidParam03_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), NULL);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, NULL);
   EXPECT_NE (ret, NNS_EDGE_ERROR_NONE);
 }
 
@@ -309,7 +309,7 @@ TEST(edge, createHandleInvalidParam04_n)
   int ret;
 
   ret = nns_edge_create_handle ("", NNS_EDGE_CONNECT_TYPE_MQTT,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_NE (ret, NNS_EDGE_ERROR_NONE);
 }
 
@@ -322,7 +322,7 @@ TEST(edge, createHandleInvalidParam05_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_HYBRID,
-      NNS_EDGE_FLAG_NONE, &edge_h);
+      NNS_EDGE_NODE_TYPE_UNKNOWN, &edge_h);
   EXPECT_NE (ret, NNS_EDGE_ERROR_NONE);
 }
 
@@ -347,7 +347,7 @@ TEST(edge, startInvalidParam02_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   eh = (nns_edge_handle_s *) edge_h;
@@ -383,7 +383,7 @@ TEST(edge, releaseHandleInvalidParam02_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   eh = (nns_edge_handle_s *) edge_h;
@@ -411,7 +411,7 @@ TEST(edge, setEventCbSetNullCallback)
   ASSERT_TRUE (_td != NULL);
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_set_event_callback (edge_h, _test_edge_event_cb, _td);
@@ -454,7 +454,7 @@ TEST(edge, setEventCbInvalidParam02_n)
   ASSERT_TRUE (_td != NULL);
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   eh = (nns_edge_handle_s *) edge_h;
@@ -492,7 +492,7 @@ TEST(edge, connectInvalidParam02_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_set_event_callback (edge_h, _test_edge_event_cb, NULL);
@@ -519,7 +519,7 @@ TEST(edge, connectInvalidParam03_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_set_event_callback (edge_h, _test_edge_event_cb, NULL);
@@ -541,7 +541,7 @@ TEST(edge, connectInvalidParam04_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_set_event_callback (edge_h, _test_edge_event_cb, NULL);
@@ -563,7 +563,7 @@ TEST(edge, connectInvalidParam05_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_set_event_callback (edge_h, _test_edge_event_cb, NULL);
@@ -602,7 +602,7 @@ TEST(edge, disconnectInvalidParam02_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   eh = (nns_edge_handle_s *) edge_h;
@@ -649,7 +649,7 @@ TEST(edge, sendInvalidParam02_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_data_create (&data_h);
@@ -682,7 +682,7 @@ TEST(edge, sendInvalidParam03_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_send (edge_h, NULL);
@@ -713,7 +713,7 @@ TEST(edge, setInfoInvalidParam02_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   eh = (nns_edge_handle_s *) edge_h;
@@ -737,7 +737,7 @@ TEST(edge, setInfoInvalidParam03_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_set_info (edge_h, NULL, "temp-caps");
@@ -756,7 +756,7 @@ TEST(edge, setInfoInvalidParam04_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_set_info (edge_h, "", "temp-caps");
@@ -775,7 +775,7 @@ TEST(edge, setInfoInvalidParam05_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_set_info (edge_h, "caps", NULL);
@@ -794,7 +794,7 @@ TEST(edge, setInfoInvalidParam06_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_set_info (edge_h, "caps", "");
@@ -813,7 +813,7 @@ TEST(edge, setInfoInvalidParam07_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   /* Not allowed key */
@@ -835,7 +835,7 @@ TEST(edge, setInfoInvalidParam08_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   /* Invalid port number */
@@ -860,7 +860,7 @@ TEST(edge, getInfo)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_set_info (edge_h, "capability", "capa-for-test");
@@ -961,7 +961,7 @@ TEST(edge, getInfoInvalidParam02_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   eh = (nns_edge_handle_s *) edge_h;
@@ -986,7 +986,7 @@ TEST(edge, getInfoInvalidParam03_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_get_info (edge_h, NULL, &value);
@@ -1006,7 +1006,7 @@ TEST(edge, getInfoInvalidParam04_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_get_info (edge_h, "", &value);
@@ -1025,7 +1025,7 @@ TEST(edge, getInfoInvalidParam05_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_get_info (edge_h, "temp-key", NULL);
@@ -1044,7 +1044,7 @@ TEST(edge, getInfoInvalidParam06_n)
   int ret;
 
   ret = nns_edge_create_handle ("temp-id", NNS_EDGE_CONNECT_TYPE_TCP,
-      (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER), &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_SERVER, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   /* Cannot get the client ID if handle is server */
@@ -3538,7 +3538,7 @@ TEST(edgeMqtt, connectLocal)
 
   /* Prepare server (127.0.0.1:port) */
   nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_HYBRID,
-      NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &server_h);
+      NNS_EDGE_NODE_TYPE_QUERY_SERVER, &server_h);
   nns_edge_set_event_callback (server_h, _test_edge_hybrid_event_cb, _td_server);
   nns_edge_set_info (server_h, "HOST", "localhost");
   nns_edge_set_info (server_h, "PORT", "0");
@@ -3550,7 +3550,7 @@ TEST(edgeMqtt, connectLocal)
 
   /* Prepare client */
   nns_edge_create_handle ("temp-client", NNS_EDGE_CONNECT_TYPE_HYBRID,
-      NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND, &client_h);
+     NNS_EDGE_NODE_TYPE_QUERY_CLIENT, &client_h);
   nns_edge_set_event_callback (client_h, _test_edge_hybrid_event_cb, _td_client);
   nns_edge_set_info (client_h, "CAPS", "test client");
   nns_edge_set_info (client_h, "HOST", "localhost");
@@ -3647,7 +3647,7 @@ TEST(edgeMqtt, connectInvalidParam2_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_HYBRID,
-      NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_SERVER, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
   nns_edge_set_info (edge_h, "DEST_HOST", "tcp://localhost");
   nns_edge_set_info (edge_h, "DEST_PORT", "1883");
@@ -3671,7 +3671,7 @@ TEST(edgeMqtt, connectInvalidParam3_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_HYBRID,
-      NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_SERVER, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
   nns_edge_set_info (edge_h, "DEST_HOST", "tcp://localhost");
   nns_edge_set_info (edge_h, "DEST_PORT", "1883");
@@ -3695,7 +3695,7 @@ TEST(edgeMqtt, connectInvalidParam4_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_HYBRID,
-      NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_SERVER, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
   nns_edge_set_info (edge_h, "DEST_HOST", "tcp://none");
   nns_edge_set_info (edge_h, "DEST_PORT", "1883");
@@ -3733,7 +3733,7 @@ TEST(edgeMqtt, closeInvalidParam2_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_HYBRID,
-      NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_SERVER, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_mqtt_close (edge_h);
@@ -3771,7 +3771,7 @@ TEST(edgeMqtt, publishInvalidParam2_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_HYBRID,
-      NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_SERVER, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
   nns_edge_set_info (edge_h, "DEST_HOST", "tcp://localhost");
   nns_edge_set_info (edge_h, "DEST_PORT", "1883");
@@ -3800,7 +3800,7 @@ TEST(edgeMqtt, publishInvalidParam3_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_HYBRID,
-      NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_SERVER, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
   nns_edge_set_info (edge_h, "DEST_HOST", "tcp://localhost");
   nns_edge_set_info (edge_h, "DEST_PORT", "1883");
@@ -3829,7 +3829,7 @@ TEST(edgeMqtt, publishInvalidParam4_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_HYBRID,
-      NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_SERVER, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
   nns_edge_set_info (edge_h, "DEST_HOST", "tcp://localhost");
   nns_edge_set_info (edge_h, "DEST_PORT", "1883");
@@ -3867,7 +3867,7 @@ TEST(edgeMqtt, subscribeInvalidParam2_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_HYBRID,
-      NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_SERVER, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
 
   ret = nns_edge_mqtt_subscribe (edge_h);
@@ -3904,7 +3904,7 @@ TEST(edgeMqtt, getMessageInvalidParam2_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_HYBRID,
-      NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_SERVER, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
   nns_edge_set_info (edge_h, "DEST_HOST", "tcp://localhost");
   nns_edge_set_info (edge_h, "DEST_PORT", "1883");
@@ -3932,7 +3932,7 @@ TEST(edgeMqtt, getMessageWithinTimeout_n)
     return;
 
   ret = nns_edge_create_handle ("temp-server", NNS_EDGE_CONNECT_TYPE_HYBRID,
-      NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER, &edge_h);
+      NNS_EDGE_NODE_TYPE_QUERY_SERVER, &edge_h);
   EXPECT_EQ (ret, NNS_EDGE_ERROR_NONE);
   nns_edge_set_info (edge_h, "DEST_HOST", "tcp://localhost");
   nns_edge_set_info (edge_h, "DEST_PORT", "1883");