nnstreamer-edge is moved to other repository.
Remove related files and build scrit.
Signed-off-by: Jaeyun <jy1210.jung@samsung.com>
/usr/lib/nnstreamer/decoders/libnnstreamer_decoder_octet_stream.so
/usr/lib/nnstreamer/filters/libnnstreamer_filter_cpp.so
/usr/lib/*/libnnstreamer.so
-/usr/lib/*/libnnstreamer-edge-temp.so
/usr/lib/*/gstreamer-1.0/libnnstreamer.so
nnstreamer_deps += orc_dep
endif
+if nnstreamer_edge_support_is_available
+ nnstreamer_deps += nnstreamer_edge_support_deps
+endif
+
# Internal dependencies
nnstreamer_internal_deps = []
# Build libraries ("both_libraries" are supported from 0.46.)
nnstreamer_shared = shared_library('nnstreamer',
nnstreamer_sources,
- dependencies: [nnstreamer_deps, nns_edge_dep],
+ dependencies: nnstreamer_deps,
include_directories: nnstreamer_inc,
install: true,
install_dir: plugins_install_dir
nnstreamer_static = static_library('nnstreamer',
nnstreamer_sources,
- dependencies: [nnstreamer_deps, nns_edge_dep],
+ dependencies: nnstreamer_deps,
include_directories: nnstreamer_inc,
install: true,
install_dir: nnstreamer_libdir
'tensor_query_server.c',
]
-foreach s : tensor_query_sources
- nnstreamer_sources += join_paths(meson.current_source_dir(), s)
-endforeach
-
-# nnstreamer-edge
-# TODO:
-# 1. This will be migrated into nnstreamer-edge repogitory.
-# 2. glib dependency will be removed.
-nnstreamer_edge_sources = [
- join_paths(meson.current_source_dir(), 'nnstreamer_edge_common.c')
-]
-
-nnstreamer_edge_deps = [
- glib_dep, gio_dep, thread_dep
-]
-
-if aitt_support_is_available
- nnstreamer_edge_sources += join_paths(meson.current_source_dir(), 'nnstreamer_edge_aitt.c')
- nnstreamer_edge_deps += aitt_support_deps
-else
- nnstreamer_edge_sources += join_paths(meson.current_source_dir(), 'nnstreamer_edge_internal.c')
- if mqtt_support_is_available
- nnstreamer_edge_sources += join_paths(meson.current_source_dir(), 'nnstreamer_edge_mqtt.c')
- nnstreamer_edge_deps += [pahomqttc_dep, thread_dep]
- endif
+# tensor_query has dependency to nnstreamer-edge.
+if nnstreamer_edge_support_is_available
+ foreach s : tensor_query_sources
+ nnstreamer_sources += join_paths(meson.current_source_dir(), s)
+ endforeach
endif
-
-nns_edge_lib = shared_library('nnstreamer-edge-temp',
- nnstreamer_edge_sources,
- dependencies: nnstreamer_edge_deps,
- install: true,
- install_dir: nnstreamer_libdir
-)
-
-static_library('nnstreamer-edge-temp',
- nnstreamer_edge_sources,
- dependencies: nnstreamer_edge_deps,
- install: true,
- install_dir: nnstreamer_libdir
-)
-
-nns_edge_dep = declare_dependency(link_with: nns_edge_lib,
- dependencies: nnstreamer_edge_deps)
+++ /dev/null
-/* SPDX-License-Identifier: Apache-2.0 */
-/**
- * Copyright (C) 2022 Samsung Electronics Co., Ltd. All Rights Reserved.
- *
- * @file nnstreamer-edge-common.h
- * @date 6 April 2022
- * @brief Common util functions for nnstreamer edge.
- * @see https://github.com/nnstreamer/nnstreamer
- * @author Gichan Jang <gichan2.jang@samsung.com>
- * @bug No known bugs except for NYI items
- * @note This file is internal header for nnstreamer edge utils. DO NOT export this file.
- */
-
-#ifndef __NNSTREAMER_EDGE_COMMON_H__
-#define __NNSTREAMER_EDGE_COMMON_H__
-
-#include <glib.h> /** @todo remove glib */
-#include <fcntl.h>
-#include <netinet/tcp.h>
-#include <netinet/in.h>
-#include <pthread.h>
-#include <unistd.h>
-#include "nnstreamer-edge.h"
-
-typedef void *nns_edge_broker_h;
-
-#ifdef __cplusplus
-extern "C" {
-#endif /* __cplusplus */
-
-/**
- * @brief Utility to silence unused parameter warning for intentionally unused parameters (e.g., callback functions of a framework)
- */
-#ifndef UNUSED
-#define UNUSED(expr) do { (void)(expr); } while (0)
-#endif
-
-#define STR_IS_VALID(s) ((s) && (s)[0] != '\0')
-#define SAFE_FREE(p) do { if (p) { free (p); (p) = NULL; } } while (0)
-
-#define NNS_EDGE_MAGIC 0xfeedfeed
-#define NNS_EDGE_MAGIC_DEAD 0xdeaddead
-#define NNS_EDGE_MAGIC_IS_VALID(h) ((h) && (h)->magic == NNS_EDGE_MAGIC)
-
-#define nns_edge_lock_init(h) do { pthread_mutex_init (&(h)->lock, NULL); } while (0)
-#define nns_edge_lock_destroy(h) do { pthread_mutex_destroy (&(h)->lock); } while (0)
-#define nns_edge_lock(h) do { pthread_mutex_lock (&(h)->lock); } while (0)
-#define nns_edge_unlock(h) do { pthread_mutex_unlock (&(h)->lock); } while (0)
-
-/**
- * @brief Internal data structure for raw data.
- */
-typedef struct {
- void *data;
- size_t data_len;
- nns_edge_data_destroy_cb destroy_cb;
-} nns_edge_raw_data_s;
-
-/**
- * @brief Internal data structure for edge data.
- */
-typedef struct {
- unsigned int magic;
- unsigned int num;
- nns_edge_raw_data_s data[NNS_EDGE_DATA_LIMIT];
- GHashTable *info_table;
-} nns_edge_data_s;
-
-/**
- * @brief Internal data structure for edge event.
- */
-typedef struct {
- unsigned int magic;
- nns_edge_event_e event;
- nns_edge_raw_data_s data;
-} nns_edge_event_s;
-
-/**
- * @todo add log util for nnstreamer-edge.
- * 1. define tag (e.g., "nnstreamer-edge").
- * 2. consider macros to print function and line.
- * 3. new API to get last error.
- */
-#define nns_edge_logi g_info
-#define nns_edge_logw g_warning
-#define nns_edge_loge g_critical
-#define nns_edge_logd g_debug
-#define nns_edge_logf g_error
-
-/**
- * @brief Internal util function to get available port number.
- */
-int nns_edge_get_available_port (void);
-
-/**
- * @brief Free allocated memory.
- */
-void nns_edge_free (void *data);
-
-/**
- * @brief Allocate new memory and copy bytes.
- * @note Caller should release newly allocated memory using nns_edge_free().
- */
-void *nns_edge_memdup (const void *data, size_t size);
-
-/**
- * @brief Allocate new memory and copy string.
- * @note Caller should release newly allocated string using nns_edge_free().
- */
-char *nns_edge_strdup (const char *str);
-
-/**
- * @brief Allocate new memory and copy bytes of string.
- * @note Caller should release newly allocated string using nns_edge_free().
- */
-char *nns_edge_strndup (const char *str, size_t len);
-
-/**
- * @brief Allocate new memory and print formatted string.
- * @note Caller should release newly allocated string using nns_edge_free().
- */
-char *nns_edge_strdup_printf (const char *format, ...);
-
-/**
- * @brief Create nnstreamer edge event.
- * @note This is internal function for edge event.
- */
-int nns_edge_event_create (nns_edge_event_e event, nns_edge_event_h * event_h);
-
-/**
- * @brief Destroy nnstreamer edge event.
- * @note This is internal function for edge event.
- */
-int nns_edge_event_destroy (nns_edge_event_h event_h);
-
-/**
- * @brief Set event data.
- * @note This is internal function for edge event.
- */
-int nns_edge_event_set_data (nns_edge_event_h event_h, void *data, size_t data_len, nns_edge_data_destroy_cb destroy_cb);
-
-/**
- * @brief Get IP address
- * @note Caller should release newly allocated string using nns_edge_free().
- */
-gchar * nns_edge_get_ip_address (void);
-
-#ifdef __cplusplus
-}
-#endif /* __cplusplus */
-#endif /* __NNSTREAMER_EDGE_COMMON_H__ */
+++ /dev/null
-/* SPDX-License-Identifier: Apache-2.0 */
-/**
- * Copyright (C) 2022 Samsung Electronics Co., Ltd. All Rights Reserved.
- *
- * @file nnstreamer-edge-internal.h
- * @date 11 May 2022
- * @brief Internal functions to support communication among devices.
- * @see https://github.com/nnstreamer/nnstreamer
- * @author Gichan Jang <gichan2.jang@samsung.com>
- * @bug No known bugs except for NYI items
- * @note This file is internal header for nnstreamer edge. DO NOT export this file.
- */
-
-#ifndef __NNSTREAMER_EDGE_INTERNAL_H__
-#define __NNSTREAMER_EDGE_INTERNAL_H__
-
-#ifdef __cplusplus
-extern "C" {
-#endif /* __cplusplus */
-
-#include "nnstreamer-edge.h"
-#include "nnstreamer-edge-common.h"
-#include <gio/gio.h>
-
-/**
- * @brief Data structure for edge handle.
- */
-typedef struct {
- unsigned int magic;
- pthread_mutex_t lock;
- char *id;
- char *topic;
- nns_edge_connect_type_e connect_type;
- char *ip;
- int port;
- char *dest_ip; /**< destination port */
- int dest_port; /**< destination port */
- int flags;
-
- /* Edge event callback and user data */
- nns_edge_event_cb event_cb;
- void *user_data;
-
- int64_t client_id;
- char *caps_str;
-
- GHashTable *conn_table;
- GSocketListener *listener;
-
- /* MQTT */
- nns_edge_broker_h broker_h;
-} nns_edge_handle_s;
-
-#if defined(ENABLE_MQTT)
-/**
- * @brief Connect to MQTT.
- * @note This is internal function for MQTT broker. You should call this with edge-handle lock.
- */
-int nns_edge_mqtt_connect (nns_edge_h edge_h);
-
-/**
- * @brief Close the connection to MQTT.
- * @note This is internal function for MQTT broker. You should call this with edge-handle lock.
- */
-int nns_edge_mqtt_close (nns_edge_h edge_h);
-
-/**
- * @brief Publish raw data.
- * @note This is internal function for MQTT broker. You should call this with edge-handle lock.
- */
-int nns_edge_mqtt_publish (nns_edge_h edge_h, const void *data, const int length);
-
-/**
- * @brief Subscribe a topic.
- * @note This is internal function for MQTT broker. You should call this with edge-handle lock.
- */
-int nns_edge_mqtt_subscribe (nns_edge_h edge_h);
-
-/**
- * @brief Check mqtt connection
- */
-bool nns_edge_mqtt_is_connected (nns_edge_h edge_h);
-
-/**
- * @brief Get message from mqtt broker.
- */
-int nns_edge_mqtt_get_message (nns_edge_h edge_h, char **msg);
-
-#else
-/**
- * @todo consider to change code style later.
- * If MQTT is disabled, nnstreamer does not include nnstreamer_edge_mqtt.c, and changing code style will make error as it is not used function now.
- *
- * static int nns_edge_mqtt_publish (nns_edge_h edge_h, const void *data, const int length)
- * {
- * return NNS_EDGE_ERROR_NOT_SUPPORTED;
- * }
- */
-#define nns_edge_mqtt_connect(...) (NNS_EDGE_ERROR_NOT_SUPPORTED)
-#define nns_edge_mqtt_close(...) (NNS_EDGE_ERROR_NOT_SUPPORTED)
-#define nns_edge_mqtt_publish(...) (NNS_EDGE_ERROR_NOT_SUPPORTED)
-#define nns_edge_mqtt_subscribe(...) (NNS_EDGE_ERROR_NOT_SUPPORTED)
-#define nns_edge_mqtt_is_connected(...) (NNS_EDGE_ERROR_NOT_SUPPORTED)
-#define nns_edge_mqtt_get_message(...) (NNS_EDGE_ERROR_NOT_SUPPORTED)
-#endif
-
-#ifdef __cplusplus
-}
-#endif /* __cplusplus */
-#endif /* __NNSTREAMER_EDGE_INTERNAL_H__ */
+++ /dev/null
-/* SPDX-License-Identifier: LGPL-2.1-only */
-/**
- * Copyright (C) 2022 Gichan Jang <gichan2.jang@samsung.com>
- *
- * @file nnstreamer-edge.h
- * @date 25 Mar 2022
- * @brief Common library to support communication among devices.
- * @see https://github.com/nnstreamer/nnstreamer
- * @author Gichan Jang <gichan2.jang@samsung.com>
- * @bug No known bugs except for NYI items
- *
- * @note This file will be moved to nnstreamer-edge repo. (https://github.com/nnstreamer/nnstreamer-edge)
- *
- * @todo Update document and sample code.
- * 1. Add sample code when the 1st API set is complete - connection, pub/sub, request, ...
- * 2. Update license when migrating this into edge repo. (Apache-2.0)
- */
-
-#ifndef __NNSTREAMER_EDGE_H__
-#define __NNSTREAMER_EDGE_H__
-
-#include <errno.h>
-#include <limits.h>
-#include <stddef.h>
-#include <stdbool.h>
-#include <stdlib.h>
-#include <string.h>
-
-#ifdef __cplusplus
-extern "C" {
-#endif /* __cplusplus */
-
-typedef void *nns_edge_h;
-typedef void *nns_edge_event_h;
-typedef void *nns_edge_data_h;
-
-/**
- * @brief The maximum number of data instances that nnstreamer-edge data may have.
- */
-#define NNS_EDGE_DATA_LIMIT (16)
-
-/**
- * @brief Enumeration for the error codes of nnstreamer-edge.
- * @todo define detailed error code later (linux standard error, sync with tizen error code)
- */
-typedef enum {
- NNS_EDGE_ERROR_NONE = 0,
- NNS_EDGE_ERROR_INVALID_PARAMETER = -EINVAL,
- NNS_EDGE_ERROR_OUT_OF_MEMORY = -ENOMEM,
- NNS_EDGE_ERROR_IO = -EIO,
- NNS_EDGE_ERROR_CONNECTION_FAILURE = -ECONNREFUSED,
- NNS_EDGE_ERROR_UNKNOWN = (-1073741824LL),
- NNS_EDGE_ERROR_NOT_SUPPORTED = (NNS_EDGE_ERROR_UNKNOWN + 2),
-} nns_edge_error_e;
-
-typedef enum {
- NNS_EDGE_EVENT_UNKNOWN = 0,
- NNS_EDGE_EVENT_CAPABILITY,
- NNS_EDGE_EVENT_NEW_DATA_RECEIVED,
- NNS_EDGE_EVENT_CALLBACK_RELEASED,
-
- NNS_EDGE_EVENT_CUSTOM = 0x01000000
-} nns_edge_event_e;
-
-typedef enum {
- NNS_EDGE_CONNECT_TYPE_TCP = 0,
- NNS_EDGE_CONNECT_TYPE_UDP,
- NNS_EDGE_CONNECT_TYPE_MQTT,
- NNS_EDGE_CONNECT_TYPE_HYBRID,
-
- NNS_EDGE_CONNECT_TYPE_UNKNOWN
-} 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_FLAG_ALL = (NNS_EDGE_FLAG_RECV | NNS_EDGE_FLAG_SEND | NNS_EDGE_FLAG_SERVER)
-} nns_edge_flag_e;
-
-/**
- * @brief Callback for the nnstreamer edge event.
- * @note This callback will suspend data stream. Do not spend too much time in the callback.
- * @return User should return NNS_EDGE_ERROR_NONE if an event is successfully handled.
- */
-typedef int (*nns_edge_event_cb) (nns_edge_event_h event_h, void *user_data);
-
-/**
- * @brief Callback called when nnstreamer-edge data is released.
- */
-typedef void (*nns_edge_data_destroy_cb) (void *data);
-
-/**
- * @brief Create edge handle.
- */
-int nns_edge_create_handle (const char *id, nns_edge_connect_type_e connect_type, int flags, nns_edge_h *edge_h);
-
-/**
- * @brief Start the nnstreamer edge.
- */
-int nns_edge_start (nns_edge_h edge_h);
-
-/**
- * @brief Release the given handle.
- */
-int nns_edge_release_handle (nns_edge_h edge_h);
-
-/**
- * @brief Set the event callback.
- */
-int nns_edge_set_event_callback (nns_edge_h edge_h, nns_edge_event_cb cb, void *user_data);
-
-/**
- * @brief Connect to the destination node.
- */
-int nns_edge_connect (nns_edge_h edge_h, const char *dest_ip, int dest_port);
-
-/**
- * @brief Disconnect from the destination node.
- */
-int nns_edge_disconnect (nns_edge_h edge_h);
-
-/**
- * @brief Publish a message to desination (broker or connected node).
- */
-int nns_edge_publish (nns_edge_h edge_h, nns_edge_data_h data_h);
-
-/**
- * @brief Subscribe a message to a given topic.
- */
-int nns_edge_subscribe (nns_edge_h edge_h);
-
-/**
- * @brief Unsubscribe a message.
- */
-int nns_edge_unsubscribe (nns_edge_h edge_h);
-
-/**
- * @brief Set nnstreamer edge info.
- */
-int nns_edge_set_info (nns_edge_h edge_h, const char *key, const char *value);
-
-/**
- * @brief Get nnstreamer edge info.
- */
-int nns_edge_get_info (nns_edge_h edge_h, const char *key, char **value);
-
-/**
- * @brief Get the nnstreamer edge event type.
- */
-int nns_edge_event_get_type (nns_edge_event_h event_h, nns_edge_event_e *event);
-
-/**
- * @brief Parse edge event (NNS_EDGE_EVENT_NEW_DATA_RECEIVED) and get received data.
- * @note Caller should release returned edge data using nns_edge_data_destroy().
- */
-int nns_edge_event_parse_new_data (nns_edge_event_h event_h, nns_edge_data_h *data_h);
-
-/**
- * @brief Parse edge event (NNS_EDGE_EVENT_CAPABILITY) and get capability string.
- * @note Caller should release returned string using free().
- */
-int nns_edge_event_parse_capability (nns_edge_event_h event_h, char **capability);
-
-/**
- * @brief Create nnstreamer edge data.
- */
-int nns_edge_data_create (nns_edge_data_h *data_h);
-
-/**
- * @brief Destroy nnstreamer edge data.
- */
-int nns_edge_data_destroy (nns_edge_data_h data_h);
-
-/**
- * @brief Validate edge data handle.
- */
-int nns_edge_data_is_valid (nns_edge_data_h data_h);
-
-/**
- * @brief Copy edge data and return new handle.
- */
-int nns_edge_data_copy (nns_edge_data_h data_h, nns_edge_data_h *new_data_h);
-
-/**
- * @brief Add raw data into nnstreamer edge data.
- */
-int nns_edge_data_add (nns_edge_data_h data_h, void *data, size_t data_len, nns_edge_data_destroy_cb destroy_cb);
-
-/**
- * @brief Get the nnstreamer edge data.
- * @note DO NOT release returned data. You should copy the data to another buffer if the returned data is necessary.
- */
-int nns_edge_data_get (nns_edge_data_h data_h, unsigned int index, void **data, size_t *data_len);
-
-/**
- * @brief Get the number of nnstreamer edge data.
- */
-int nns_edge_data_get_count (nns_edge_data_h data_h, unsigned int *count);
-
-/**
- * @brief Set the information of edge data.
- */
-int nns_edge_data_set_info (nns_edge_data_h data_h, const char *key, const char *value);
-
-/**
- * @brief Get the information of edge data. Caller should release the returned value using free().
- */
-int nns_edge_data_get_info (nns_edge_data_h data_h, const char *key, char **value);
-
-#ifdef __cplusplus
-}
-#endif /* __cplusplus */
-#endif /* __NNSTREAMER_EDGE_H__ */
+++ /dev/null
-/* SPDX-License-Identifier: LGPL-2.1-only */
-/**
- * Copyright (C) 2022 Gichan Jang <gichan2.jang@samsung.com>
- *
- * @file nnstreamer_edge_aitt.c
- * @date 28 Mar 2022
- * @brief Common library to support communication among devices using aitt.
- * @see https://github.com/nnstreamer/nnstreamer
- * @author Gichan Jang <gichan2.jang@samsung.com>
- * @bug No known bugs except for NYI items
- */
-
-#include <aitt_c.h>
-#include "nnstreamer_edge_common.h"
-
-typedef void *nns_edge_aitt_h;
-typedef void *nns_edge_aitt_msg_h;
-typedef void *nns_edge_aitt_sub_h;
-
-/**
- * @brief Data structure for aitt handle.
- * @todo Update AITT-related handle later. This is internal struct to manage edge-handle with AITT.
- */
-typedef struct
-{
- nns_edge_connect_type_e protocol;
- struct
- {
- nns_edge_aitt_h aitt_h;
- nns_edge_aitt_msg_h msg_h;
- nns_edge_aitt_sub_h sub_h;
- };
-} nns_edge_handle_s;
+++ /dev/null
-/* SPDX-License-Identifier: Apache-2.0 */
-/**
- * Copyright (C) 2022 Samsung Electronics Co., Ltd. All Rights Reserved.
- *
- * @file nnstreamer-edge-common.c
- * @date 6 April 2022
- * @brief Common util functions for nnstreamer edge.
- * @see https://github.com/nnstreamer/nnstreamer
- * @author Gichan Jang <gichan2.jang@samsung.com>
- * @bug No known bugs except for NYI items
- */
-
-#define _GNU_SOURCE
-#include <stdio.h>
-#include <arpa/inet.h>
-#include <ifaddrs.h>
-#include "nnstreamer-edge-common.h"
-
-/**
- * @brief Internal util function to get available port number.
- */
-int
-nns_edge_get_available_port (void)
-{
- struct sockaddr_in sin;
- int port = 0, sock;
- socklen_t len = sizeof (struct sockaddr);
-
- sin.sin_family = AF_INET;
- sin.sin_addr.s_addr = INADDR_ANY;
- sin.sin_port = 0;
-
- sock = socket (AF_INET, SOCK_STREAM, 0);
- if (sock < 0) {
- nns_edge_loge ("Failed to get available port, socket creation failure.");
- return 0;
- }
-
- if (bind (sock, (struct sockaddr *) &sin, sizeof (struct sockaddr)) == 0) {
- if (getsockname (sock, (struct sockaddr *) &sin, &len) == 0) {
- port = ntohs (sin.sin_port);
- nns_edge_logi ("Available port number: %d", port);
- } else {
- nns_edge_logw ("Failed to read local socket info.");
- }
- }
- close (sock);
-
- return port;
-}
-
-/**
- * @brief Free allocated memory.
- */
-void
-nns_edge_free (void *data)
-{
- if (data)
- free (data);
-}
-
-/**
- * @brief Allocate new memory and copy bytes.
- * @note Caller should release newly allocated memory using nns_edge_free().
- */
-void *
-nns_edge_memdup (const void *data, size_t size)
-{
- void *mem = NULL;
-
- if (data && size > 0) {
- mem = malloc (size);
-
- if (mem) {
- memcpy (mem, data, size);
- } else {
- nns_edge_loge ("Failed to allocate memory (%zd).", size);
- }
- }
-
- return mem;
-}
-
-/**
- * @brief Allocate new memory and copy string.
- * @note Caller should release newly allocated string using nns_edge_free().
- */
-char *
-nns_edge_strdup (const char *str)
-{
- char *new_str = NULL;
- size_t len;
-
- if (str) {
- len = strlen (str);
-
- new_str = (char *) malloc (len + 1);
- if (new_str) {
- memcpy (new_str, str, len);
- new_str[len] = '\0';
- } else {
- nns_edge_loge ("Failed to allocate memory (%zd).", len + 1);
- }
- }
-
- return new_str;
-}
-
-/**
- * @brief Allocate new memory and print formatted string.
- * @note Caller should release newly allocated string using nns_edge_free().
- */
-char *
-nns_edge_strdup_printf (const char *format, ...)
-{
- char *new_str = NULL;
- va_list args;
- int len;
-
- va_start (args, format);
-#pragma GCC diagnostic push
-#pragma GCC diagnostic ignored "-Wformat-nonliteral"
- len = vasprintf (&new_str, format, args);
-#pragma GCC diagnostic pop
- if (len < 0)
- new_str = NULL;
- va_end (args);
-
- return new_str;
-}
-
-/**
- * @brief Create nnstreamer edge event.
- */
-int
-nns_edge_event_create (nns_edge_event_e event, nns_edge_event_h * event_h)
-{
- nns_edge_event_s *ee;
-
- if (!event_h) {
- nns_edge_loge ("Invalid param, event_h should not be null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (event <= NNS_EDGE_EVENT_UNKNOWN) {
- nns_edge_loge ("Invalid param, given event type is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- ee = (nns_edge_event_s *) malloc (sizeof (nns_edge_event_s));
- if (!ee) {
- nns_edge_loge ("Failed to allocate memory for edge event.");
- return NNS_EDGE_ERROR_OUT_OF_MEMORY;
- }
-
- memset (ee, 0, sizeof (nns_edge_event_s));
- ee->magic = NNS_EDGE_MAGIC;
- ee->event = event;
-
- *event_h = ee;
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Destroy nnstreamer edge event.
- */
-int
-nns_edge_event_destroy (nns_edge_event_h event_h)
-{
- nns_edge_event_s *ee;
-
- ee = (nns_edge_event_s *) event_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (ee)) {
- nns_edge_loge ("Invalid param, given edge event is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- ee->magic = NNS_EDGE_MAGIC_DEAD;
-
- if (ee->data.destroy_cb)
- ee->data.destroy_cb (ee->data.data);
-
- SAFE_FREE (ee);
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Set event data.
- */
-int
-nns_edge_event_set_data (nns_edge_event_h event_h, void *data, size_t data_len,
- nns_edge_data_destroy_cb destroy_cb)
-{
- nns_edge_event_s *ee;
-
- ee = (nns_edge_event_s *) event_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (ee)) {
- nns_edge_loge ("Invalid param, given edge event is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!data || data_len <= 0) {
- nns_edge_loge ("Invalid param, data should not be null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- /* Clear old data and set new one. */
- if (ee->data.destroy_cb)
- ee->data.destroy_cb (ee->data.data);
-
- ee->data.data = data;
- ee->data.data_len = data_len;
- ee->data.destroy_cb = destroy_cb;
-
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Get the nnstreamer edge event type.
- */
-int
-nns_edge_event_get_type (nns_edge_event_h event_h, nns_edge_event_e * event)
-{
- nns_edge_event_s *ee;
-
- ee = (nns_edge_event_s *) event_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (ee)) {
- nns_edge_loge ("Invalid param, given edge event is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!event) {
- nns_edge_loge ("Invalid param, event should not be null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- *event = ee->event;
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Parse edge event (NNS_EDGE_EVENT_NEW_DATA_RECEIVED) and get received data.
- * @note Caller should release returned edge data using nns_edge_data_destroy().
- */
-int
-nns_edge_event_parse_new_data (nns_edge_event_h event_h,
- nns_edge_data_h * data_h)
-{
- nns_edge_event_s *ee;
-
- ee = (nns_edge_event_s *) event_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (ee)) {
- nns_edge_loge ("Invalid param, given edge event is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!data_h) {
- nns_edge_loge ("Invalid param, data_h should not be null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (ee->event != NNS_EDGE_EVENT_NEW_DATA_RECEIVED) {
- nns_edge_loge ("The edge event has invalid event type.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- return nns_edge_data_copy ((nns_edge_data_h) ee->data.data, data_h);
-}
-
-/**
- * @brief Parse edge event (NNS_EDGE_EVENT_CAPABILITY) and get capability string.
- * @note Caller should release returned string using free().
- */
-int
-nns_edge_event_parse_capability (nns_edge_event_h event_h, char **capability)
-{
- nns_edge_event_s *ee;
-
- ee = (nns_edge_event_s *) event_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (ee)) {
- nns_edge_loge ("Invalid param, given edge event is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!capability) {
- nns_edge_loge ("Invalid param, capability should not be null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (ee->event != NNS_EDGE_EVENT_CAPABILITY) {
- nns_edge_loge ("The edge event has invalid event type.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- *capability = nns_edge_strdup (ee->data.data);
-
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Create nnstreamer edge data.
- */
-int
-nns_edge_data_create (nns_edge_data_h * data_h)
-{
- nns_edge_data_s *ed;
-
- if (!data_h) {
- nns_edge_loge ("Invalid param, data_h should not be null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- ed = (nns_edge_data_s *) malloc (sizeof (nns_edge_data_s));
- if (!ed) {
- nns_edge_loge ("Failed to allocate memory for edge data.");
- return NNS_EDGE_ERROR_OUT_OF_MEMORY;
- }
-
- memset (ed, 0, sizeof (nns_edge_data_s));
- ed->magic = NNS_EDGE_MAGIC;
- ed->info_table = g_hash_table_new_full (g_str_hash, g_str_equal,
- nns_edge_free, nns_edge_free);
-
- *data_h = ed;
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Destroy nnstreamer edge data.
- */
-int
-nns_edge_data_destroy (nns_edge_data_h data_h)
-{
- nns_edge_data_s *ed;
- unsigned int i;
-
- ed = (nns_edge_data_s *) data_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (ed)) {
- nns_edge_loge ("Invalid param, given edge data is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- ed->magic = NNS_EDGE_MAGIC_DEAD;
-
- for (i = 0; i < ed->num; i++) {
- if (ed->data[i].destroy_cb)
- ed->data[i].destroy_cb (ed->data[i].data);
- }
-
- g_hash_table_destroy (ed->info_table);
-
- SAFE_FREE (ed);
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Validate edge data handle.
- */
-int
-nns_edge_data_is_valid (nns_edge_data_h data_h)
-{
- nns_edge_data_s *ed;
-
- ed = (nns_edge_data_s *) data_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (ed)) {
- nns_edge_loge ("Invalid param, edge data handle is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Copy edge data and return new handle.
- */
-int
-nns_edge_data_copy (nns_edge_data_h data_h, nns_edge_data_h * new_data_h)
-{
- nns_edge_data_s *ed;
- nns_edge_data_s *copied;
- GHashTableIter iter;
- gpointer key, value;
- unsigned int i;
- int ret;
-
- ed = (nns_edge_data_s *) data_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (ed)) {
- nns_edge_loge ("Invalid param, edge data handle is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!new_data_h) {
- nns_edge_loge ("Invalid param, new_data_h should not be null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- ret = nns_edge_data_create (new_data_h);
- if (ret != NNS_EDGE_ERROR_NONE) {
- nns_edge_loge ("Failed to create new data handle.");
- return ret;
- }
-
- copied = (nns_edge_data_s *) (*new_data_h);
-
- copied->num = ed->num;
- for (i = 0; i < ed->num; i++) {
- copied->data[i].data = nns_edge_memdup (ed->data[i].data,
- ed->data[i].data_len);
- copied->data[i].data_len = ed->data[i].data_len;
- copied->data[i].destroy_cb = nns_edge_free;
- }
-
- g_hash_table_iter_init (&iter, ed->info_table);
- while (g_hash_table_iter_next (&iter, &key, &value)) {
- g_hash_table_insert (copied->info_table, nns_edge_strdup (key),
- nns_edge_strdup (value));
- }
-
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Add raw data into nnstreamer edge data.
- */
-int
-nns_edge_data_add (nns_edge_data_h data_h, void *data, size_t data_len,
- nns_edge_data_destroy_cb destroy_cb)
-{
- nns_edge_data_s *ed;
-
- ed = (nns_edge_data_s *) data_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (ed)) {
- nns_edge_loge ("Invalid param, given edge data is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (ed->num >= NNS_EDGE_DATA_LIMIT) {
- nns_edge_loge ("Cannot add data, the maximum number of edge data is %d.",
- NNS_EDGE_DATA_LIMIT);
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!data || data_len <= 0) {
- nns_edge_loge ("Invalid param, data should not be null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- ed->data[ed->num].data = data;
- ed->data[ed->num].data_len = data_len;
- ed->data[ed->num].destroy_cb = destroy_cb;
- ed->num++;
-
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Get the nnstreamer edge data.
- * @note DO NOT release returned data. You should copy the data to another buffer if the returned data is necessary.
- */
-int
-nns_edge_data_get (nns_edge_data_h data_h, unsigned int index, void **data,
- size_t *data_len)
-{
- nns_edge_data_s *ed;
-
- ed = (nns_edge_data_s *) data_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (ed)) {
- nns_edge_loge ("Invalid param, given edge data is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!data || !data_len) {
- nns_edge_loge ("Invalid param, data and len should not be null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (index >= ed->num) {
- nns_edge_loge
- ("Invalid param, the number of edge data is %u but requested %uth data.",
- ed->num, index);
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- *data = ed->data[index].data;
- *data_len = ed->data[index].data_len;
-
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Get the number of nnstreamer edge data.
- */
-int
-nns_edge_data_get_count (nns_edge_data_h data_h, unsigned int *count)
-{
- nns_edge_data_s *ed;
-
- ed = (nns_edge_data_s *) data_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (ed)) {
- nns_edge_loge ("Invalid param, given edge data is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!count) {
- nns_edge_loge ("Invalid param, count should not be null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- *count = ed->num;
-
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Set the information of edge data.
- */
-int
-nns_edge_data_set_info (nns_edge_data_h data_h, const char *key,
- const char *value)
-{
- nns_edge_data_s *ed;
-
- ed = (nns_edge_data_s *) data_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (ed)) {
- nns_edge_loge ("Invalid param, given edge data is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!STR_IS_VALID (key)) {
- nns_edge_loge ("Invalid param, given key is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!STR_IS_VALID (value)) {
- nns_edge_loge ("Invalid param, given value is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- g_hash_table_insert (ed->info_table, nns_edge_strdup (key),
- nns_edge_strdup (value));
-
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Get the information of edge data. Caller should release the returned value using free().
- */
-int
-nns_edge_data_get_info (nns_edge_data_h data_h, const char *key, char **value)
-{
- nns_edge_data_s *ed;
- char *val;
-
- ed = (nns_edge_data_s *) data_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (ed)) {
- nns_edge_loge ("Invalid param, given edge data is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!STR_IS_VALID (key)) {
- nns_edge_loge ("Invalid param, given key is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!value) {
- nns_edge_loge ("Invalid param, value should not be null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- val = g_hash_table_lookup (ed->info_table, key);
- if (!val) {
- nns_edge_loge ("Invalid param, cannot find info about '%s'.", key);
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- *value = nns_edge_strdup (val);
-
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Get IP address
- */
-gchar *
-nns_edge_get_ip_address (void)
-{
- struct ifaddrs *addrs, *run;
- gchar *ret = NULL;
-
- getifaddrs (&addrs);
- run = addrs;
- while (run) {
- if (run->ifa_addr && run->ifa_addr->sa_family == AF_INET) {
- struct sockaddr_in *pAddr = (struct sockaddr_in *) run->ifa_addr;
-
- if (NULL != strstr (run->ifa_name, "en") ||
- NULL != strstr (run->ifa_name, "et")) {
- g_free (ret);
- ret = g_strdup (inet_ntoa (pAddr->sin_addr));
- break;
- }
- }
- run = run->ifa_next;
- }
-
- freeifaddrs (addrs);
-
- if (NULL == ret)
- ret = g_strdup ("localhost");
-
- return ret;
-}
+++ /dev/null
-/* SPDX-License-Identifier: Apache-2.0 */
-/**
- * Copyright (C) 2022 Samsung Electronics Co., Ltd. All Rights Reserved.
- *
- * @file nnstreamer-edge-internal.c
- * @date 6 April 2022
- * @brief Common library to support communication among devices.
- * @see https://github.com/nnstreamer/nnstreamer
- * @author Gichan Jang <gichan2.jang@samsung.com>
- * @bug No known bugs except for NYI items
- */
-
-#include "nnstreamer-edge-common.h"
-#include "nnstreamer-edge-internal.h"
-
-#define N_BACKLOG 10
-#define DEFAULT_TIMEOUT_SEC 10
-#define _STR_NULL(str) ((str) ? (str) : "(NULL)")
-
-/**
- * @brief enum for nnstreamer edge query commands.
- */
-typedef enum
-{
- _NNS_EDGE_CMD_ERROR = 0,
- _NNS_EDGE_CMD_TRANSFER_DATA,
- _NNS_EDGE_CMD_HOST_INFO,
- _NNS_EDGE_CMD_CAPABILITY,
- _NNS_EDGE_CMD_END
-} nns_edge_cmd_e;
-
-/**
- * @brief Structure for edge command info. It should be fixed size.
- */
-typedef struct
-{
- unsigned int magic;
- nns_edge_cmd_e cmd;
- int64_t client_id;
-
- /* memory info */
- uint32_t num;
- size_t mem_size[NNS_EDGE_DATA_LIMIT];
-} nns_edge_cmd_info_s;
-
-/**
- * @brief Structure for edge command and buffers.
- */
-typedef struct
-{
- nns_edge_cmd_info_s info;
- void *mem[NNS_EDGE_DATA_LIMIT];
-} nns_edge_cmd_s;
-
-/**
- * @brief Data structure for edge connection.
- */
-typedef struct
-{
- char *ip;
- int port;
- int8_t running;
- pthread_t msg_thread;
- GSocket *socket;
-} nns_edge_conn_s;
-
-/**
- * @brief Data structure for connection data.
- */
-typedef struct
-{
- nns_edge_conn_s *src_conn;
- nns_edge_conn_s *sink_conn;
- int64_t id;
-} nns_edge_conn_data_s;
-
-/**
- * @brief Structures for thread data of message handling.
- */
-typedef struct
-{
- nns_edge_handle_s *eh;
- int64_t client_id;
- nns_edge_conn_s *conn;
-} nns_edge_thread_data_s;
-
-/**
- * @brief Send data to connected socket.
- */
-static bool
-_send_raw_data (GSocket * socket, void *data, size_t size)
-{
- size_t bytes_sent = 0;
- ssize_t rret;
- GError *err = NULL;
-
- while (bytes_sent < size) {
- rret = g_socket_send (socket, (char *) data + bytes_sent,
- size - bytes_sent, NULL, &err);
-
- if (rret == 0) {
- nns_edge_loge ("Connection closed.");
- return false;
- }
-
- if (rret < 0) {
- nns_edge_loge ("Error while sending data (%s).", err->message);
- g_clear_error (&err);
- return false;
- }
-
- bytes_sent += rret;
- }
-
- return true;
-}
-
-/**
- * @brief Receive data from connected socket.
- */
-static bool
-_receive_raw_data (GSocket * socket, void *data, size_t size)
-{
- size_t bytes_received = 0;
- ssize_t rret;
- GError *err = NULL;
-
- while (bytes_received < size) {
- rret = g_socket_receive (socket, (char *) data + bytes_received,
- size - bytes_received, NULL, &err);
-
- if (rret == 0) {
- nns_edge_loge ("Connection closed.");
- return false;
- }
-
- if (rret < 0) {
- nns_edge_loge ("Failed to read from socket (%s).", err->message);
- g_clear_error (&err);
- return false;
- }
-
- bytes_received += rret;
- }
-
- return true;
-}
-
-/**
- * @brief Parse string and get host IP:port.
- */
-static void
-_parse_host_str (const char *host, char **ip, int *port)
-{
- char *p = g_strrstr (host, ":");
-
- if (p) {
- *ip = g_strndup (host, (p - host));
- *port = (int) g_ascii_strtoll (p + 1, NULL, 10);
- }
-}
-
-/**
- * @brief Get host string (IP:port).
- */
-static void
-_get_host_str (const char *ip, const int port, char **host)
-{
- *host = nns_edge_strdup_printf ("%s:%d", ip, port);
-}
-
-/**
- * @brief Internal function to check connection.
- */
-static bool
-_nns_edge_check_connection (nns_edge_conn_s * conn)
-{
- GIOCondition condition;
-
- if (!conn || !conn->socket || g_socket_is_closed (conn->socket))
- return false;
-
- condition = g_socket_condition_check (conn->socket,
- G_IO_IN | G_IO_OUT | G_IO_PRI | G_IO_ERR | G_IO_HUP);
-
- if (!condition || (condition & (G_IO_ERR | G_IO_HUP))) {
- nns_edge_logw ("Socket is not available, possibly closed.");
- return false;
- }
-
- return true;
-}
-
-/**
- * @brief initialize edge command.
- */
-static void
-_nns_edge_cmd_init (nns_edge_cmd_s * cmd, nns_edge_cmd_e c, int64_t cid)
-{
- if (!cmd)
- return;
-
- memset (cmd, 0, sizeof (nns_edge_cmd_s));
- cmd->info.magic = NNS_EDGE_MAGIC;
- cmd->info.cmd = c;
- cmd->info.client_id = cid;
-}
-
-/**
- * @brief Clear allocated memory in edge command.
- */
-static void
-_nns_edge_cmd_clear (nns_edge_cmd_s * cmd)
-{
- unsigned int i;
-
- if (!cmd)
- return;
-
- cmd->info.magic = NNS_EDGE_MAGIC_DEAD;
-
- for (i = 0; i < cmd->info.num; i++) {
- SAFE_FREE (cmd->mem[i]);
- }
-}
-
-/**
- * @brief Validate edge command.
- */
-static bool
-_nns_edge_cmd_is_valid (nns_edge_cmd_s * cmd)
-{
- int command;
-
- if (!cmd)
- return false;
-
- command = (int) cmd->info.cmd;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (&cmd->info) ||
- (command < 0 || command >= _NNS_EDGE_CMD_END)) {
- return false;
- }
-
- return true;
-}
-
-/**
- * @brief Send edge command to connected device.
- */
-static int
-_nns_edge_cmd_send (nns_edge_conn_s * conn, nns_edge_cmd_s * cmd)
-{
- unsigned int n;
-
- if (!conn) {
- nns_edge_loge ("Failed to send command, edge connection is null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!_nns_edge_cmd_is_valid (cmd)) {
- nns_edge_loge ("Failed to send command, invalid command.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!_nns_edge_check_connection (conn)) {
- nns_edge_loge ("Failed to send command, socket has error.");
- return NNS_EDGE_ERROR_IO;
- }
-
- if (!_send_raw_data (conn->socket, &cmd->info, sizeof (nns_edge_cmd_info_s))) {
- nns_edge_loge ("Failed to send command to socket.");
- return NNS_EDGE_ERROR_IO;
- }
-
- for (n = 0; n < cmd->info.num; n++) {
- if (!_send_raw_data (conn->socket, cmd->mem[n], cmd->info.mem_size[n])) {
- nns_edge_loge ("Failed to send %uth memory to socket.", n);
- return NNS_EDGE_ERROR_IO;
- }
- }
-
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Receive edge command from connected device.
- */
-static int
-_nns_edge_cmd_receive (nns_edge_conn_s * conn, nns_edge_cmd_s * cmd)
-{
- unsigned int i, n;
- int ret = NNS_EDGE_ERROR_NONE;
-
- if (!conn || !cmd)
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
-
- if (!_nns_edge_check_connection (conn)) {
- nns_edge_loge ("Failed to receive command, socket has error.");
- return NNS_EDGE_ERROR_IO;
- }
-
- if (!_receive_raw_data (conn->socket, &cmd->info,
- sizeof (nns_edge_cmd_info_s))) {
- nns_edge_loge ("Failed to receive command from socket.");
- return NNS_EDGE_ERROR_IO;
- }
-
- if (!_nns_edge_cmd_is_valid (cmd)) {
- nns_edge_loge ("Failed to receive command, invalid command.");
- return NNS_EDGE_ERROR_IO;
- }
-
- nns_edge_logd ("Received command:%d (num:%u)", cmd->info.cmd, cmd->info.num);
- if (cmd->info.num >= NNS_EDGE_DATA_LIMIT) {
- nns_edge_loge ("Invalid request, the max memories for data transfer is %d.",
- NNS_EDGE_DATA_LIMIT);
- return NNS_EDGE_ERROR_IO;
- }
-
- for (n = 0; n < cmd->info.num; n++) {
- cmd->mem[n] = malloc (cmd->info.mem_size[n]);
- if (!cmd->mem[n]) {
- nns_edge_loge ("Failed to allocate memory to receive data from socket.");
- ret = NNS_EDGE_ERROR_OUT_OF_MEMORY;
- break;
- }
-
- if (!_receive_raw_data (conn->socket, cmd->mem[n], cmd->info.mem_size[n])) {
- nns_edge_loge ("Failed to receive %uth memory from socket.", n++);
- ret = NNS_EDGE_ERROR_IO;
- break;
- }
- }
-
- if (ret != NNS_EDGE_ERROR_NONE) {
- for (i = 0; i < n; i++) {
- SAFE_FREE (cmd->mem[i]);
- }
- }
-
- return ret;
-}
-
-/**
- * @brief Internal function to send edge data.
- */
-static int
-_nns_edge_transfer_data (nns_edge_conn_s * conn, nns_edge_data_h data_h,
- int64_t client_id)
-{
- nns_edge_cmd_s cmd;
- unsigned int i;
- int ret;
-
- _nns_edge_cmd_init (&cmd, _NNS_EDGE_CMD_TRANSFER_DATA, client_id);
-
- nns_edge_data_get_count (data_h, &cmd.info.num);
- for (i = 0; i < cmd.info.num; i++)
- nns_edge_data_get (data_h, i, &cmd.mem[i], &cmd.info.mem_size[i]);
-
- ret = _nns_edge_cmd_send (conn, &cmd);
-
- return ret;
-}
-
-/**
- * @brief Internal function to invoke event callback.
- * @note This function should be called with handle lock.
- */
-static int
-_nns_edge_invoke_event_cb (nns_edge_handle_s * eh, nns_edge_event_e event,
- void *data, size_t data_len, nns_edge_data_destroy_cb destroy_cb)
-{
- nns_edge_event_h event_h;
- int ret;
-
- if (!eh) {
- nns_edge_loge ("Invalid param, given edge handle is null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- /* If event callback is null, return ok. */
- if (!eh->event_cb) {
- nns_edge_logi ("The event callback is null, do nothing!");
- return NNS_EDGE_ERROR_NONE;
- }
-
- ret = nns_edge_event_create (event, &event_h);
- if (ret != NNS_EDGE_ERROR_NONE) {
- nns_edge_loge ("Failed to create new edge event.");
- return ret;
- }
-
- if (data) {
- ret = nns_edge_event_set_data (event_h, data, data_len, destroy_cb);
- if (ret != NNS_EDGE_ERROR_NONE) {
- nns_edge_loge ("Failed to handle edge event due to invalid event data.");
- goto error;
- }
- }
-
- ret = eh->event_cb (event_h, eh->user_data);
- if (ret != NNS_EDGE_ERROR_NONE) {
- nns_edge_loge ("The event callback returns error.");
- }
-
-error:
- nns_edge_event_destroy (event_h);
- return ret;
-}
-
-/**
- * @brief Close connection
- */
-static bool
-_nns_edge_close_connection (nns_edge_conn_s * conn)
-{
- if (!conn)
- return false;
-
- /* Stop and clear the message thread. */
- if (conn->msg_thread) {
- conn->running = 0;
- pthread_cancel (conn->msg_thread);
- pthread_join (conn->msg_thread, NULL);
- conn->msg_thread = 0;
- }
-
- if (conn->socket) {
- nns_edge_cmd_s cmd;
-
- /* Send error before closing the socket. */
- nns_edge_logd ("Send error cmd to close connection.");
- _nns_edge_cmd_init (&cmd, _NNS_EDGE_CMD_ERROR, 0);
- _nns_edge_cmd_send (conn, &cmd);
-
- /**
- * Close and release the socket.
- * Using GSocket, if its last reference is dropped, it will close socket automatically.
- */
- g_clear_object (&conn->socket);
- }
-
- SAFE_FREE (conn->ip);
- SAFE_FREE (conn);
- return true;
-}
-
-/**
- * @brief Get nnstreamer-edge connection data.
- * @note This function should be called with handle lock.
- */
-static nns_edge_conn_data_s *
-_nns_edge_get_connection (nns_edge_handle_s * eh, int64_t client_id)
-{
- if (!NNS_EDGE_MAGIC_IS_VALID (eh)) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- return NULL;
- }
-
- return g_hash_table_lookup (eh->conn_table, GUINT_TO_POINTER (client_id));
-}
-
-/**
- * @brief Get nnstreamer-edge connection data.
- * @note This function should be called with handle lock.
- */
-static nns_edge_conn_data_s *
-_nns_edge_add_connection (nns_edge_handle_s * eh, int64_t client_id)
-{
- nns_edge_conn_data_s *data = NULL;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (eh)) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- return NULL;
- }
-
- data = g_hash_table_lookup (eh->conn_table, GUINT_TO_POINTER (client_id));
-
- if (NULL == data) {
- data = (nns_edge_conn_data_s *) malloc (sizeof (nns_edge_conn_data_s));
- if (NULL == data) {
- nns_edge_loge ("Failed to allocate memory for connection data.");
- return NULL;
- }
-
- memset (data, 0, sizeof (nns_edge_conn_data_s));
- data->id = client_id;
-
- g_hash_table_insert (eh->conn_table, GUINT_TO_POINTER (client_id), data);
- }
-
- return data;
-}
-
-/**
- * @brief Remove nnstreamer-edge connection data. This will be called when removing connection data from hash table.
- */
-static void
-_nns_edge_remove_connection (gpointer data)
-{
- nns_edge_conn_data_s *cdata = (nns_edge_conn_data_s *) data;
-
- if (cdata) {
- _nns_edge_close_connection (cdata->src_conn);
- _nns_edge_close_connection (cdata->sink_conn);
- cdata->src_conn = cdata->sink_conn = NULL;
-
- SAFE_FREE (cdata);
- }
-}
-
-/**
- * @brief Get socket address
- */
-static bool
-_nns_edge_get_saddr (const char *ip, const int port, GSocketAddress ** saddr)
-{
- GError *err = NULL;
- GInetAddress *addr;
-
- /* look up name if we need to */
- addr = g_inet_address_new_from_string (ip);
- if (!addr) {
- GList *results;
- GResolver *resolver;
- resolver = g_resolver_get_default ();
- results = g_resolver_lookup_by_name (resolver, ip, NULL, &err);
- if (!results) {
- if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
- nns_edge_loge ("Failed to resolve ip, name resolver is cancelled.");
- } else {
- nns_edge_loge ("Failed to resolve ip '%s': %s", ip, err->message);
- }
- g_clear_error (&err);
- g_object_unref (resolver);
- return false;
- }
- /** @todo Try with the second address if the first fails */
- addr = G_INET_ADDRESS (g_object_ref (results->data));
- g_resolver_free_addresses (results);
- g_object_unref (resolver);
- }
-
- *saddr = g_inet_socket_address_new (addr, port);
- g_object_unref (addr);
-
- return true;
-}
-
-/**
- * @brief Connect to requested socket.
- */
-static bool
-_nns_edge_connect_socket (nns_edge_conn_s * conn)
-{
- GError *err = NULL;
- GSocketAddress *saddr = NULL;
- bool ret = false;
-
- if (!_nns_edge_get_saddr (conn->ip, conn->port, &saddr)) {
- nns_edge_loge ("Failed to get socket address");
- return ret;
- }
-
- /* create sending client socket */
- conn->socket =
- g_socket_new (g_socket_address_get_family (saddr), G_SOCKET_TYPE_STREAM,
- G_SOCKET_PROTOCOL_TCP, &err);
-
- if (!conn->socket) {
- nns_edge_loge ("Failed to create new socket");
- goto done;
- }
-
- /* setting TCP_NODELAY to true in order to avoid packet batching as known as Nagle's algorithm */
- if (!g_socket_set_option (conn->socket, IPPROTO_TCP, TCP_NODELAY, true, &err)) {
- nns_edge_loge ("Failed to set socket TCP_NODELAY option: %s", err->message);
- goto done;
- }
-
- if (!g_socket_connect (conn->socket, saddr, NULL, &err)) {
- if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
- nns_edge_logd ("Cancelled connecting");
- } else {
- nns_edge_loge ("Failed to connect to host, %s:%d", conn->ip, conn->port);
- }
- goto done;
- }
-
- /* now connected to the requested socket */
- ret = true;
-
-done:
- g_object_unref (saddr);
- g_clear_error (&err);
- return ret;
-}
-
-/**
- * @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 *dest_ip, int dest_port)
-{
- nns_edge_conn_s *conn = NULL;
- nns_edge_conn_data_s *conn_data;
- nns_edge_cmd_s cmd;
- char *host;
- bool done = false;
- int ret;
-
- conn = (nns_edge_conn_s *) malloc (sizeof (nns_edge_conn_s));
- if (!conn) {
- nns_edge_loge ("Failed to allocate client data.");
- goto error;
- }
-
- memset (conn, 0, sizeof (nns_edge_conn_s));
- conn->ip = nns_edge_strdup (dest_ip);
- conn->port = dest_port;
-
- 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 ip and port to destination. */
- _nns_edge_cmd_init (&cmd, _NNS_EDGE_CMD_HOST_INFO, client_id);
-
- _get_host_str (eh->ip, eh->port, &host);
- cmd.info.num = 1;
- cmd.info.mem_size[0] = strlen (host) + 1;
- cmd.mem[0] = host;
- }
-
- 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.
- */
-static void *
-_nns_edge_message_handler (void *thread_data)
-{
- nns_edge_thread_data_s *_tdata = (nns_edge_thread_data_s *) thread_data;
- nns_edge_handle_s *eh;
- nns_edge_conn_s *conn;
- nns_edge_cmd_s cmd;
- int64_t client_id;
- char *val;
- int ret;
-
- if (!_tdata) {
- nns_edge_loge ("Internal error, thread data is null.");
- return NULL;
- }
-
- eh = (nns_edge_handle_s *) _tdata->eh;
- conn = _tdata->conn;
- client_id = _tdata->client_id;
- SAFE_FREE (_tdata);
-
- conn->running = 1;
- while (conn->running) {
- nns_edge_data_h data_h;
- unsigned int i;
-
- /* Validate edge handle */
- if (!NNS_EDGE_MAGIC_IS_VALID (eh)) {
- nns_edge_loge ("The edge handle is invalid, it would be expired.");
- break;
- }
-
- /* Receive data from the client */
- _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 data from the connected node.");
- break;
- }
-
- if (cmd.info.cmd == _NNS_EDGE_CMD_ERROR) {
- nns_edge_loge ("Received error, stop msg thread.");
- _nns_edge_cmd_clear (&cmd);
- break;
- } else if (cmd.info.cmd != _NNS_EDGE_CMD_TRANSFER_DATA) {
- /** @todo handle other cmd later */
- _nns_edge_cmd_clear (&cmd);
- continue;
- }
-
- ret = nns_edge_data_create (&data_h);
- if (ret != NNS_EDGE_ERROR_NONE) {
- nns_edge_loge ("Failed to create data handle in msg thread.");
- _nns_edge_cmd_clear (&cmd);
- continue;
- }
-
- for (i = 0; i < cmd.info.num; i++)
- nns_edge_data_add (data_h, cmd.mem[i], cmd.info.mem_size[i], NULL);
-
- /* Set client ID in edge data */
- val = nns_edge_strdup_printf ("%ld", (long int) client_id);
- nns_edge_data_set_info (data_h, "client_id", val);
- SAFE_FREE (val);
-
- ret = _nns_edge_invoke_event_cb (eh, NNS_EDGE_EVENT_NEW_DATA_RECEIVED,
- data_h, sizeof (nns_edge_data_h), NULL);
- if (ret != NNS_EDGE_ERROR_NONE) {
- /* Try to get next request if server does not accept data from client. */
- nns_edge_logw ("The server does not accept data from client.");
- }
-
- nns_edge_data_destroy (data_h);
- _nns_edge_cmd_clear (&cmd);
- }
-
- conn->running = 0;
- return NULL;
-}
-
-/**
- * @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)
-{
- pthread_attr_t attr;
- int tid;
- nns_edge_thread_data_s *thread_data = NULL;
-
- thread_data =
- (nns_edge_thread_data_s *) malloc (sizeof (nns_edge_thread_data_s));
- if (!thread_data) {
- nns_edge_loge ("Failed to allocate edge thread data.");
- return NNS_EDGE_ERROR_OUT_OF_MEMORY;
- }
-
- /* Create message receving thread */
- pthread_attr_init (&attr);
- pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_JOINABLE);
-
- thread_data->eh = eh;
- thread_data->conn = conn;
- thread_data->client_id = client_id;
-
- tid = pthread_create (&conn->msg_thread, &attr, _nns_edge_message_handler,
- thread_data);
- pthread_attr_destroy (&attr);
-
- if (tid < 0) {
- nns_edge_loge ("Failed to create message handler thread.");
- conn->running = 0;
- conn->msg_thread = 0;
- SAFE_FREE (thread_data);
- return NNS_EDGE_ERROR_IO;
- }
-
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Callback for socket listener, accept socket and create message thread.
- */
-static void
-_nns_edge_accept_socket_async_cb (GObject * source, GAsyncResult * result,
- gpointer user_data)
-{
- GSocketListener *socket_listener = G_SOCKET_LISTENER (source);
- GSocket *socket = NULL;
- GError *err = NULL;
- nns_edge_handle_s *eh = (nns_edge_handle_s *) user_data;
- nns_edge_conn_s *conn = NULL;
- nns_edge_cmd_s cmd;
- bool done = false;
- char *connected_ip = NULL;
- int connected_port = 0;
- nns_edge_conn_data_s *conn_data = NULL;
- int64_t client_id;
- int ret;
-
- socket =
- g_socket_listener_accept_socket_finish (socket_listener, result, NULL,
- &err);
-
- if (!socket) {
- nns_edge_loge ("Failed to get socket: %s", err->message);
- g_clear_error (&err);
- return;
- }
- g_socket_set_timeout (socket, DEFAULT_TIMEOUT_SEC);
-
- /* create socket with connection */
- conn = (nns_edge_conn_s *) malloc (sizeof (nns_edge_conn_s));
- if (!conn) {
- nns_edge_loge ("Failed to allocate edge connection");
- goto error;
- }
-
- memset (conn, 0, sizeof (nns_edge_conn_s));
- conn->socket = socket;
-
- /* setting TCP_NODELAY to true in order to avoid packet batching as known as Nagle's algorithm */
- if (!g_socket_set_option (socket, IPPROTO_TCP, TCP_NODELAY, true, &err)) {
- nns_edge_loge ("Failed to set socket TCP_NODELAY option: %s", err->message);
- g_clear_error (&err);
- goto error;
- }
-
- if (eh->flags & NNS_EDGE_FLAG_SERVER)
- client_id = g_get_monotonic_time ();
- else
- client_id = eh->client_id;
-
- /* Send capability and info to check compatibility. */
- if (eh->flags & NNS_EDGE_FLAG_SERVER) {
- if (!STR_IS_VALID (eh->caps_str)) {
- nns_edge_loge ("Cannot accept socket, invalid capability.");
- goto error;
- }
-
- _nns_edge_cmd_init (&cmd, _NNS_EDGE_CMD_CAPABILITY, client_id);
- cmd.info.num = 1;
- cmd.info.mem_size[0] = strlen (eh->caps_str) + 1;
- cmd.mem[0] = eh->caps_str;
-
- ret = _nns_edge_cmd_send (conn, &cmd);
- if (ret != NNS_EDGE_ERROR_NONE) {
- nns_edge_loge ("Failed to send capability.");
- goto error;
- }
-
- /* Receive ip and port from destination. */
- ret = _nns_edge_cmd_receive (conn, &cmd);
- if (ret != NNS_EDGE_ERROR_NONE) {
- nns_edge_loge ("Failed to receive node info.");
- goto error;
- }
-
- if (cmd.info.cmd != _NNS_EDGE_CMD_HOST_INFO) {
- nns_edge_loge ("Failed to get host info.");
- _nns_edge_cmd_clear (&cmd);
- goto error;
- }
-
- _parse_host_str (cmd.mem[0], &connected_ip, &connected_port);
- _nns_edge_cmd_clear (&cmd);
-
- /* Connect to client listener. */
- ret = _nns_edge_connect_to (eh, client_id, connected_ip, connected_port);
- if (ret != NNS_EDGE_ERROR_NONE) {
- nns_edge_loge ("Failed to connect host %s:%d.",
- connected_ip, connected_port);
- goto error;
- }
- }
-
- 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. */
- _nns_edge_close_connection (conn_data->src_conn);
- conn_data->src_conn = conn;
- done = true;
- }
-
-error:
- if (!done) {
- _nns_edge_close_connection (conn);
- }
-
- if (eh->listener)
- g_socket_listener_accept_socket_async (eh->listener, NULL,
- (GAsyncReadyCallback) _nns_edge_accept_socket_async_cb, eh);
-
- SAFE_FREE (connected_ip);
-}
-
-/**
- * @brief Create edge handle.
- */
-int
-nns_edge_create_handle (const char *id, nns_edge_connect_type_e connect_type,
- int flags, nns_edge_h * edge_h)
-{
- nns_edge_handle_s *eh;
-
- if (!STR_IS_VALID (id)) {
- nns_edge_loge ("Invalid param, given ID is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (connect_type < 0 || connect_type >= NNS_EDGE_CONNECT_TYPE_UNKNOWN) {
- nns_edge_loge ("Invalid param, set valid connect type.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- /**
- * @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.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!edge_h) {
- nns_edge_loge ("Invalid param, edge_h should not be null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- eh = (nns_edge_handle_s *) malloc (sizeof (nns_edge_handle_s));
- if (!eh) {
- nns_edge_loge ("Failed to allocate memory for edge handle.");
- return NNS_EDGE_ERROR_OUT_OF_MEMORY;
- }
-
- memset (eh, 0, sizeof (nns_edge_handle_s));
- nns_edge_lock_init (eh);
- eh->magic = NNS_EDGE_MAGIC;
- eh->id = nns_edge_strdup (id);
- eh->connect_type = connect_type;
- eh->ip = nns_edge_strdup ("localhost");
- eh->port = 0;
- eh->dest_ip = nns_edge_strdup ("localhost");
- eh->dest_port = 0;
- eh->flags = flags;
- eh->broker_h = NULL;
-
- /* Connection data for each client ID. */
- eh->conn_table = g_hash_table_new_full (g_direct_hash, g_direct_equal, NULL,
- _nns_edge_remove_connection);
-
- *edge_h = eh;
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Start the nnstreamer edge.
- */
-int
-nns_edge_start (nns_edge_h edge_h)
-{
- GSocketAddress *saddr = NULL;
- GError *err = NULL;
- int ret = 0;
- nns_edge_handle_s *eh;
-
- eh = (nns_edge_handle_s *) edge_h;
- if (!eh) {
- 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)) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- nns_edge_unlock (eh);
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (eh->port <= 0) {
- eh->port = nns_edge_get_available_port ();
- if (eh->port <= 0) {
- nns_edge_loge ("Failed to start edge. Cannot get available port.");
- nns_edge_unlock (eh);
- return NNS_EDGE_ERROR_CONNECTION_FAILURE;
- }
- }
-
- if (eh->flags & NNS_EDGE_FLAG_SERVER) {
- if (NNS_EDGE_CONNECT_TYPE_HYBRID == eh->connect_type) {
- gchar *device, *topic, *msg;
-
- if (NNS_EDGE_ERROR_NONE != nns_edge_mqtt_connect (eh)) {
- nns_edge_loge
- ("Failed to start nnstreamer-edge. Connection failure to broker.");
- ret = NNS_EDGE_ERROR_CONNECTION_FAILURE;
- goto error;
- }
-
- /** @todo Set unique device name.
- * Device name should be unique. Consider using MAC address later.
- * Now, use ID received from the user.
- */
- device = g_strdup_printf ("device-%s", eh->id);
- topic = g_strdup_printf ("edge/inference/%s/%s/", device, eh->topic);
-
- g_free (device);
- g_free (eh->topic);
- eh->topic = topic;
- msg = nns_edge_strdup_printf ("%s:%d", eh->ip, eh->port);
-
- if (NNS_EDGE_ERROR_NONE != nns_edge_mqtt_publish (eh, msg,
- strlen (msg) + 1)) {
- nns_edge_loge ("Failed to publish the meesage: %s", msg);
- ret = NNS_EDGE_ERROR_IO;
- goto error;
- }
- nns_edge_free (msg);
- }
- }
-
- /* Initialize server src data. */
- eh->listener = g_socket_listener_new ();
- g_socket_listener_set_backlog (eh->listener, N_BACKLOG);
-
- if (!_nns_edge_get_saddr (eh->ip, eh->port, &saddr)) {
- nns_edge_loge ("Failed to get socket address");
- ret = NNS_EDGE_ERROR_CONNECTION_FAILURE;
- goto error;
- }
- if (!g_socket_listener_add_address (eh->listener, saddr,
- G_SOCKET_TYPE_STREAM, G_SOCKET_PROTOCOL_TCP, NULL, NULL, &err)) {
- nns_edge_loge ("Failed to add address: %s", err->message);
- g_clear_error (&err);
- ret = NNS_EDGE_ERROR_CONNECTION_FAILURE;
- goto error;
- }
-
- g_socket_listener_accept_socket_async (eh->listener, NULL,
- (GAsyncReadyCallback) _nns_edge_accept_socket_async_cb, eh);
-
-error:
- if (saddr)
- g_object_unref (saddr);
-
- nns_edge_unlock (eh);
- return ret;
-}
-
-/**
- * @brief Release the given handle.
- */
-int
-nns_edge_release_handle (nns_edge_h edge_h)
-{
- nns_edge_handle_s *eh;
-
- eh = (nns_edge_handle_s *) edge_h;
- if (!eh) {
- 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)) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- nns_edge_unlock (eh);
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (nns_edge_mqtt_is_connected (eh)) {
- if (NNS_EDGE_ERROR_NONE != nns_edge_mqtt_close (eh)) {
- nns_edge_logw ("Failed to close mqtt connection.");
- }
- }
-
- eh->magic = NNS_EDGE_MAGIC_DEAD;
- eh->event_cb = NULL;
- eh->user_data = NULL;
-
- if (eh->listener)
- g_clear_object (&eh->listener);
-
- g_hash_table_destroy (eh->conn_table);
- eh->conn_table = NULL;
-
- SAFE_FREE (eh->id);
- SAFE_FREE (eh->topic);
- SAFE_FREE (eh->caps_str);
- SAFE_FREE (eh->ip);
- SAFE_FREE (eh->dest_ip);
-
- nns_edge_unlock (eh);
- nns_edge_lock_destroy (eh);
- SAFE_FREE (eh);
-
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Set the event callback.
- */
-int
-nns_edge_set_event_callback (nns_edge_h edge_h, nns_edge_event_cb cb,
- void *user_data)
-{
- nns_edge_handle_s *eh;
- int ret;
-
- eh = (nns_edge_handle_s *) edge_h;
- if (!eh) {
- 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)) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- nns_edge_unlock (eh);
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- ret = _nns_edge_invoke_event_cb (eh, NNS_EDGE_EVENT_CALLBACK_RELEASED,
- NULL, 0, NULL);
- if (ret != NNS_EDGE_ERROR_NONE) {
- nns_edge_loge ("Failed to set new event callback.");
- nns_edge_unlock (eh);
- return ret;
- }
-
- eh->event_cb = cb;
- eh->user_data = user_data;
-
- nns_edge_unlock (eh);
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Connect to the destination node.
- */
-int
-nns_edge_connect (nns_edge_h edge_h, const char *dest_ip, int dest_port)
-{
- nns_edge_handle_s *eh;
- int ret;
- char *server_ip = NULL;
- int server_port;
-
- eh = (nns_edge_handle_s *) edge_h;
- if (!eh) {
- nns_edge_loge ("Invalid param, given edge handle is null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!STR_IS_VALID (dest_ip)) {
- nns_edge_loge ("Invalid param, given IP is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- nns_edge_lock (eh);
-
- if (!NNS_EDGE_MAGIC_IS_VALID (eh)) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- nns_edge_unlock (eh);
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!eh->event_cb) {
- nns_edge_loge ("NNStreamer-edge event callback is not registered.");
- nns_edge_unlock (eh);
- return NNS_EDGE_ERROR_CONNECTION_FAILURE;
- }
-
- SAFE_FREE (eh->dest_ip);
- eh->dest_ip = nns_edge_strdup (dest_ip);
- eh->dest_port = dest_port;
-
- if (NNS_EDGE_CONNECT_TYPE_HYBRID == eh->connect_type) {
- gchar *topic, *msg = NULL;
-
- if (!nns_edge_mqtt_is_connected (eh)) {
- if (NNS_EDGE_ERROR_NONE != nns_edge_mqtt_connect (eh)) {
- nns_edge_loge ("Connection failure to broker.");
- nns_edge_unlock (eh);
- return NNS_EDGE_ERROR_CONNECTION_FAILURE;
- }
- topic = g_strdup_printf ("edge/inference/+/%s/#", eh->topic);
- g_free (eh->topic);
- eh->topic = topic;
-
- if (NNS_EDGE_ERROR_NONE != nns_edge_mqtt_subscribe (eh)) {
- nns_edge_loge ("Failed to subscribe to topic: %s.", eh->topic);
- nns_edge_unlock (eh);
- return NNS_EDGE_ERROR_CONNECTION_FAILURE;
- }
- }
-
- ret = nns_edge_mqtt_get_message (eh, &msg);
- while (NNS_EDGE_ERROR_NONE == ret) {
- gchar **splits;
- splits = g_strsplit (msg, ":", -1);
- server_ip = g_strdup (splits[0]);
- server_port = g_ascii_strtoull (splits[1], NULL, 10);
- nns_edge_logd ("[DEBUG] Parsed server info: Server [%s:%d] ", server_ip,
- server_port);
-
- g_strfreev (splits);
- g_free (msg);
-
- ret = _nns_edge_connect_to (eh, eh->client_id, server_ip, server_port);
- if (NNS_EDGE_ERROR_NONE == ret) {
- break;
- }
- SAFE_FREE (server_ip);
- ret = nns_edge_mqtt_get_message (eh, &msg);
- }
- } else { /** case for NNS_EDGE_CONNECT_TYPE_TCP == eh->protocol */
- server_ip = nns_edge_strdup (dest_ip);
- server_port = dest_port;
- ret = _nns_edge_connect_to (eh, eh->client_id, server_ip, server_port);
- if (ret != NNS_EDGE_ERROR_NONE) {
- nns_edge_loge ("Failed to connect to %s:%d", server_ip, server_port);
- }
- }
-
- SAFE_FREE (server_ip);
- nns_edge_unlock (eh);
-
- return ret;
-}
-
-/**
- * @brief Disconnect from the destination node.
- */
-int
-nns_edge_disconnect (nns_edge_h edge_h)
-{
- nns_edge_handle_s *eh;
-
- eh = (nns_edge_handle_s *) edge_h;
- if (!eh) {
- 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)) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- nns_edge_unlock (eh);
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- g_hash_table_remove_all (eh->conn_table);
-
- nns_edge_unlock (eh);
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Publish a message to desination (broker or connected node).
- */
-int
-nns_edge_publish (nns_edge_h edge_h, nns_edge_data_h data_h)
-{
- nns_edge_handle_s *eh;
- nns_edge_conn_data_s *conn_data;
- int64_t client_id;
- char *val;
- int ret = NNS_EDGE_ERROR_NONE;
-
- eh = (nns_edge_handle_s *) edge_h;
- if (!eh) {
- nns_edge_loge ("Invalid param, given edge handle is null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (nns_edge_data_is_valid (data_h) != NNS_EDGE_ERROR_NONE) {
- nns_edge_loge ("Invalid param, given edge data is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- ret = nns_edge_data_get_info (data_h, "client_id", &val);
- if (ret != NNS_EDGE_ERROR_NONE) {
- nns_edge_loge ("Cannot find client ID in edge data.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- client_id = strtoll (val, NULL, 10);
- SAFE_FREE (val);
-
- nns_edge_lock (eh);
-
- /** @todo update code for each connect type */
- switch (eh->connect_type) {
- case NNS_EDGE_CONNECT_TYPE_TCP:
- case NNS_EDGE_CONNECT_TYPE_HYBRID:
- conn_data = _nns_edge_get_connection (eh, client_id);
- if (!conn_data) {
- nns_edge_loge
- ("Cannot find connection, invalid client ID or connection closed.");
- ret = NNS_EDGE_ERROR_INVALID_PARAMETER;
- break;
- }
-
- ret = _nns_edge_transfer_data (conn_data->sink_conn, data_h, client_id);
- if (ret != NNS_EDGE_ERROR_NONE)
- nns_edge_loge ("Failed to send edge data.");
- break;
- default:
- break;
- }
- nns_edge_unlock (eh);
- return ret;
-}
-
-/**
- * @brief Subscribe a message from broker.
- */
-int
-nns_edge_subscribe (nns_edge_h edge_h)
-{
- nns_edge_handle_s *eh;
- int ret = NNS_EDGE_ERROR_NONE;
-
- eh = (nns_edge_handle_s *) edge_h;
- if (!eh) {
- 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)) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- ret = NNS_EDGE_ERROR_INVALID_PARAMETER;
- goto done;
- }
-
- if (eh->connect_type != NNS_EDGE_CONNECT_TYPE_MQTT) {
- nns_edge_loge ("Invalid connect type, cannot subscribe a message.");
- ret = NNS_EDGE_ERROR_INVALID_PARAMETER;
- goto done;
- }
-
- if (!STR_IS_VALID (eh->topic)) {
- nns_edge_loge ("Invalid topic, cannot subscribe a message.");
- ret = NNS_EDGE_ERROR_INVALID_PARAMETER;
- goto done;
- }
- /** @todo update code (subscribe) */
-
-done:
- nns_edge_unlock (eh);
- return ret;
-}
-
-/**
- * @brief Unsubscribe a message.
- */
-int
-nns_edge_unsubscribe (nns_edge_h edge_h)
-{
- nns_edge_handle_s *eh;
- int ret = NNS_EDGE_ERROR_NONE;
-
- eh = (nns_edge_handle_s *) edge_h;
- if (!eh) {
- 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)) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- ret = NNS_EDGE_ERROR_INVALID_PARAMETER;
- goto done;
- }
-
- if (eh->connect_type != NNS_EDGE_CONNECT_TYPE_MQTT) {
- nns_edge_loge ("Invalid connect type, cannot subscribe a message.");
- ret = NNS_EDGE_ERROR_INVALID_PARAMETER;
- goto done;
- }
-
- if (!STR_IS_VALID (eh->topic)) {
- nns_edge_loge ("Invalid topic, cannot subscribe a message.");
- ret = NNS_EDGE_ERROR_INVALID_PARAMETER;
- goto done;
- }
-
- /** @todo update code (unsubscribe) */
-
-done:
- nns_edge_unlock (eh);
- return ret;
-}
-
-/**
- * @brief Set nnstreamer edge info.
- */
-int
-nns_edge_set_info (nns_edge_h edge_h, const char *key, const char *value)
-{
- nns_edge_handle_s *eh;
- int ret = NNS_EDGE_ERROR_NONE;
-
- eh = (nns_edge_handle_s *) edge_h;
- if (!eh) {
- nns_edge_loge ("Invalid param, given edge handle is null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!STR_IS_VALID (key)) {
- nns_edge_loge ("Invalid param, given key is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!STR_IS_VALID (value)) {
- nns_edge_loge ("Invalid param, given value is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- nns_edge_lock (eh);
-
- if (!NNS_EDGE_MAGIC_IS_VALID (eh)) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- nns_edge_unlock (eh);
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- /**
- * @todo User handles (replace or append) the capability of edge handle.
- * @todo Change key-value set as json or hash table.
- */
- if (0 == strcasecmp (key, "CAPS") || 0 == strcasecmp (key, "CAPABILITY")) {
- SAFE_FREE (eh->caps_str);
- eh->caps_str = nns_edge_strdup (value);
- } else if (0 == strcasecmp (key, "HOST")) {
- SAFE_FREE (eh->ip);
- eh->ip = nns_edge_strdup (value);
- } else if (0 == strcasecmp (key, "PORT")) {
- eh->port = strtoll (value, NULL, 10);
- } else if (0 == strcasecmp (key, "DEST_HOST")) {
- SAFE_FREE (eh->dest_ip);
- eh->dest_ip = nns_edge_strdup (value);
- } else if (0 == strcasecmp (key, "DEST_PORT")) {
- eh->dest_port = strtoll (value, NULL, 10);
- } else if (0 == strcasecmp (key, "TOPIC")) {
- SAFE_FREE (eh->topic);
- eh->topic = nns_edge_strdup (value);
- } else if (0 == strcasecmp (key, "ID") || 0 == strcasecmp (key, "CLIENT_ID")) {
- /* Not allowed key */
- nns_edge_loge ("Cannot update %s.", key);
- ret = NNS_EDGE_ERROR_INVALID_PARAMETER;
- } else {
- nns_edge_logw ("Failed to set edge info. Unknown key: %s", key);
- ret = NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- nns_edge_unlock (eh);
- return ret;
-}
-
-/**
- * @brief Get nnstreamer edge info.
- */
-int
-nns_edge_get_info (nns_edge_h edge_h, const char *key, char **value)
-{
- nns_edge_handle_s *eh;
- int ret = NNS_EDGE_ERROR_NONE;
-
- eh = (nns_edge_handle_s *) edge_h;
- if (!eh) {
- nns_edge_loge ("Invalid param, given edge handle is null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!STR_IS_VALID (key)) {
- nns_edge_loge ("Invalid param, given key is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!value) {
- nns_edge_loge ("Invalid param, value should not be null.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- nns_edge_lock (eh);
-
- if (!NNS_EDGE_MAGIC_IS_VALID (eh)) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- nns_edge_unlock (eh);
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- /**
- * @todo User handles (replace or append) the capability of edge handle.
- * @todo Change key-value set as json or hash table.
- */
- if (0 == strcasecmp (key, "CAPS") || 0 == strcasecmp (key, "CAPABILITY")) {
- *value = nns_edge_strdup (eh->caps_str);
- } else if (0 == strcasecmp (key, "HOST")) {
- *value = nns_edge_strdup (eh->ip);
- } else if (0 == strcasecmp (key, "PORT")) {
- *value = nns_edge_strdup_printf ("%d", eh->port);
- } else if (0 == strcasecmp (key, "DEST_IP")) {
- *value = nns_edge_strdup (eh->dest_ip);
- } else if (0 == strcasecmp (key, "DEST_PORT")) {
- *value = nns_edge_strdup_printf ("%d", eh->dest_port);
- } else if (0 == strcasecmp (key, "TOPIC")) {
- *value = nns_edge_strdup (eh->topic);
- } else if (0 == strcasecmp (key, "ID")) {
- *value = nns_edge_strdup (eh->id);
- } else if (0 == strcasecmp (key, "CLIENT_ID")) {
- if (eh->flags & NNS_EDGE_FLAG_SERVER) {
- nns_edge_loge ("Cannot get the client ID, it was started as a server.");
- ret = NNS_EDGE_ERROR_INVALID_PARAMETER;
- } else {
- *value = nns_edge_strdup_printf ("%ld", (long int) eh->client_id);
- }
- } else {
- nns_edge_logw ("Failed to get edge info. Unknown key: %s", key);
- ret = NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- nns_edge_unlock (eh);
- return ret;
-}
+++ /dev/null
-/* SPDX-License-Identifier: LGPL-2.1-only */
-/**
- * Copyright (C) 2022 Samsung Electronics Co., Ltd. All Rights Reserved.
- *
- * @file nnstreamer_edge_mqtt.c
- * @date 11 May 2022
- * @brief Internal functions to support MQTT protocol (Paho Asynchronous MQTT C Client Library).
- * @see https://github.com/nnstreamer/nnstreamer
- * @author Sangjung Woo <sangjung.woo@samsung.com>
- * @bug No known bugs except for NYI items
- */
-
-#if !defined(ENABLE_MQTT)
-#error "This file can be built with Paho MQTT library."
-#endif
-#define DEFAULT_SUB_TIMEOUT 1000000 /** 1 second */
-
-#include <unistd.h>
-#include <MQTTAsync.h>
-#include "nnstreamer-edge-common.h"
-#include "nnstreamer-edge-internal.h"
-
-/**
- * @brief Data structure for mqtt broker handle.
- */
-typedef struct
-{
- void *mqtt_h;
- GAsyncQueue *server_list;
- GMutex mqtt_mutex;
- GCond mqtt_gcond;
- gboolean mqtt_is_connected;
-} nns_edge_broker_s;
-
-/**
- * @brief Callback function to be called when the connection is lost.
- */
-static void
-mqtt_cb_connection_lost (void *context, char *cause)
-{
- nns_edge_handle_s *eh;
- nns_edge_broker_s *bh;
-
- eh = (nns_edge_handle_s *) context;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (eh) || !eh->broker_h) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- return;
- }
-
- bh = (nns_edge_broker_s *) eh->broker_h;
- nns_edge_logw ("MQTT connection is lost (ID:%s, Cause:%s).", eh->id, cause);
- g_mutex_lock (&bh->mqtt_mutex);
- bh->mqtt_is_connected = FALSE;
- g_cond_broadcast (&bh->mqtt_gcond);
- g_mutex_unlock (&bh->mqtt_mutex);
-
- if (eh->event_cb) {
- /** @todo send new event (MQTT disconnected) */
- }
-}
-
-/**
- * @brief Callback function to be called when the connection is completed.
- */
-static void
-mqtt_cb_connection_success (void *context, MQTTAsync_successData * response)
-{
- nns_edge_handle_s *eh;
- nns_edge_broker_s *bh;
-
- UNUSED (response);
- eh = (nns_edge_handle_s *) context;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (eh) || !eh->broker_h) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- return;
- }
-
- bh = (nns_edge_broker_s *) eh->broker_h;
-
- g_mutex_lock (&bh->mqtt_mutex);
- bh->mqtt_is_connected = TRUE;
- g_cond_broadcast (&bh->mqtt_gcond);
- g_mutex_unlock (&bh->mqtt_mutex);
-
- if (eh->event_cb) {
- /** @todo send new event (MQTT connected) */
- }
-}
-
-/**
- * @brief Callback function to be called when the connection is failed.
- */
-static void
-mqtt_cb_connection_failure (void *context, MQTTAsync_failureData * response)
-{
- nns_edge_handle_s *eh;
- nns_edge_broker_s *bh;
-
- UNUSED (response);
- eh = (nns_edge_handle_s *) context;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (eh) || !eh->broker_h) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- return;
- }
-
- bh = (nns_edge_broker_s *) eh->broker_h;
-
- nns_edge_logw ("MQTT connection is failed (ID:%s).", eh->id);
- g_mutex_lock (&bh->mqtt_mutex);
- bh->mqtt_is_connected = FALSE;
- g_cond_broadcast (&bh->mqtt_gcond);
- g_mutex_unlock (&bh->mqtt_mutex);
-
- if (eh->event_cb) {
- /** @todo send new event (MQTT connection failure) */
- }
-}
-
-/**
- * @brief Callback function to be called when the disconnection is completed.
- */
-static void
-mqtt_cb_disconnection_success (void *context, MQTTAsync_successData * response)
-{
- nns_edge_handle_s *eh;
- nns_edge_broker_s *bh;
-
- UNUSED (response);
- eh = (nns_edge_handle_s *) context;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (eh) || !eh->broker_h) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- return;
- }
-
- bh = (nns_edge_broker_s *) eh->broker_h;
-
- nns_edge_logi ("MQTT disconnection is completed (ID:%s).", eh->id);
- g_mutex_lock (&bh->mqtt_mutex);
- bh->mqtt_is_connected = FALSE;
- g_cond_broadcast (&bh->mqtt_gcond);
- g_mutex_unlock (&bh->mqtt_mutex);
-
- if (eh->event_cb) {
- /** @todo send new event (MQTT disconnected) */
- }
-}
-
-/**
- * @brief Callback function to be called when the disconnection is failed.
- */
-static void
-mqtt_cb_disconnection_failure (void *context, MQTTAsync_failureData * response)
-{
- nns_edge_handle_s *eh;
-
- UNUSED (response);
- eh = (nns_edge_handle_s *) context;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (eh)) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- return;
- }
-
- nns_edge_logw ("MQTT disconnection is failed (ID:%s).", eh->id);
- if (eh->event_cb) {
- /** @todo send new event (MQTT disconnection failure) */
- }
-}
-
-/**
- * @brief Callback function to be called when a message is arrived.
- * @return Return TRUE to prevent delivering the message again.
- */
-static int
-mqtt_cb_message_arrived (void *context, char *topic, int topic_len,
- MQTTAsync_message * message)
-{
- nns_edge_handle_s *eh;
- nns_edge_broker_s *bh;
- char *msg = NULL;
-
- UNUSED (topic);
- UNUSED (topic_len);
- UNUSED (message);
- eh = (nns_edge_handle_s *) context;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (eh) || !eh->broker_h) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- return TRUE;
- }
-
- if (0 >= message->payloadlen) {
- nns_edge_logw ("Invalid payload lenth: %d", message->payloadlen);
- return TRUE;
- }
-
- bh = (nns_edge_broker_s *) eh->broker_h;
-
- nns_edge_logd ("MQTT message is arrived (ID:%s, Topic:%s).",
- eh->id, eh->topic);
-
- msg = (char *) malloc (message->payloadlen);
- memcpy (msg, message->payload, message->payloadlen);
- g_async_queue_push (bh->server_list, msg);
-
- if (eh->event_cb) {
- /** @todo send new event (message arrived) */
- }
-
- return TRUE;
-}
-
-/**
- * @brief Connect to MQTT.
- * @note This is internal function for MQTT broker. You should call this with edge-handle lock.
- */
-int
-nns_edge_mqtt_connect (nns_edge_h edge_h)
-{
- nns_edge_handle_s *eh;
- nns_edge_broker_s *bh;
- MQTTAsync_connectOptions options = MQTTAsync_connectOptions_initializer;
- int ret = NNS_EDGE_ERROR_NONE;
- int64_t end_time;
- MQTTAsync handle;
- char *url;
- char *client_id;
-
- eh = (nns_edge_handle_s *) edge_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (eh)) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- bh = (nns_edge_broker_s *) malloc (sizeof (nns_edge_broker_s));
- if (!bh) {
- nns_edge_loge ("Failed to allocate memory for broker handle.");
- return NNS_EDGE_ERROR_OUT_OF_MEMORY;
- }
-
- url = nns_edge_strdup_printf ("%s:%d", eh->dest_ip, eh->dest_port);
- client_id = nns_edge_strdup_printf ("nns_edge_%s_%u", eh->id, getpid ());
-
- ret = MQTTAsync_create (&handle, url, client_id,
- MQTTCLIENT_PERSISTENCE_NONE, NULL);
- if (MQTTASYNC_SUCCESS != ret) {
- nns_edge_loge ("Failed to create MQTT handle.");
- ret = NNS_EDGE_ERROR_CONNECTION_FAILURE;
- goto error;
- }
-
- g_cond_init (&bh->mqtt_gcond);
- g_mutex_init (&bh->mqtt_mutex);
- bh->mqtt_is_connected = FALSE;
- bh->mqtt_h = handle;
- bh->server_list = g_async_queue_new ();
- eh->broker_h = bh;
-
- bh = (nns_edge_broker_s *) eh->broker_h;
- if (!bh->mqtt_h) {
- nns_edge_loge ("Invalid state, MQTT connection was not completed.");
- ret = NNS_EDGE_ERROR_IO;
- goto error;
- }
- handle = bh->mqtt_h;
-
- nns_edge_logi ("Trying to connect MQTT (ID:%s, URL:%s:%d).",
- eh->id, eh->dest_ip, eh->dest_port);
-
- MQTTAsync_setCallbacks (handle, edge_h,
- mqtt_cb_connection_lost, mqtt_cb_message_arrived, NULL);
-
- options.cleansession = 1;
- options.keepAliveInterval = 6;
- options.onSuccess = mqtt_cb_connection_success;
- options.onFailure = mqtt_cb_connection_failure;
- options.context = edge_h;
-
- if (MQTTAsync_connect (handle, &options) != MQTTASYNC_SUCCESS) {
- nns_edge_loge ("Failed to connect MQTT.");
- ret = NNS_EDGE_ERROR_CONNECTION_FAILURE;
- goto error;
- }
-
- /* Waiting for the connection */
- end_time = g_get_monotonic_time () + 5 * G_TIME_SPAN_SECOND;
- g_mutex_lock (&bh->mqtt_mutex);
- while (!bh->mqtt_is_connected) {
- if (!g_cond_wait_until (&bh->mqtt_gcond, &bh->mqtt_mutex, end_time)) {
- g_mutex_unlock (&bh->mqtt_mutex);
- nns_edge_loge ("Failed to connect to MQTT broker."
- "Please check broker is running status or broker host address.");
- goto error;
- }
- }
- g_mutex_unlock (&bh->mqtt_mutex);
- return NNS_EDGE_ERROR_NONE;
-
-error:
- nns_edge_mqtt_close (eh);
- return ret;
-}
-
-/**
- * @brief Close the connection to MQTT.
- * @note This is internal function for MQTT broker. You should call this with edge-handle lock.
- */
-int
-nns_edge_mqtt_close (nns_edge_h edge_h)
-{
- nns_edge_handle_s *eh;
- nns_edge_broker_s *bh;
- MQTTAsync handle;
- MQTTAsync_disconnectOptions options = MQTTAsync_disconnectOptions_initializer;
- char *msg;
-
- eh = (nns_edge_handle_s *) edge_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (eh) || !eh->broker_h) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- bh = (nns_edge_broker_s *) eh->broker_h;
-
- if (!bh->mqtt_h) {
- nns_edge_loge ("Invalid state, MQTT connection was not completed.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
- handle = bh->mqtt_h;
-
- nns_edge_logi ("Trying to disconnect MQTT (ID:%s, URL:%s:%d).",
- eh->id, eh->dest_ip, eh->dest_port);
-
- options.onSuccess = mqtt_cb_disconnection_success;
- options.onFailure = mqtt_cb_disconnection_failure;
- options.context = edge_h;
-
- /** Clear retained message */
- MQTTAsync_send (handle, eh->topic, 0, NULL, 1, 1, NULL);
-
- while (MQTTAsync_isConnected (handle)) {
- if (MQTTAsync_disconnect (handle, &options) != MQTTASYNC_SUCCESS) {
- nns_edge_loge ("Failed to disconnect MQTT.");
- return NNS_EDGE_ERROR_IO;
- }
- g_usleep (10000);
- }
- g_cond_clear (&bh->mqtt_gcond);
- g_mutex_clear (&bh->mqtt_mutex);
-
- MQTTAsync_destroy (&handle);
-
- while ((msg = g_async_queue_try_pop (bh->server_list))) {
- SAFE_FREE (msg);
- }
- g_async_queue_unref (bh->server_list);
- bh->server_list = NULL;
- SAFE_FREE (bh);
-
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Publish raw data.
- * @note This is internal function for MQTT broker. You should call this with edge-handle lock.
- */
-int
-nns_edge_mqtt_publish (nns_edge_h edge_h, const void *data, const int length)
-{
- nns_edge_handle_s *eh;
- nns_edge_broker_s *bh;
- MQTTAsync handle;
- int ret;
-
- eh = (nns_edge_handle_s *) edge_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (eh) || !eh->broker_h) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- if (!data || length <= 0) {
- nns_edge_loge ("Invalid param, given data is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- bh = (nns_edge_broker_s *) eh->broker_h;
- if (!bh->mqtt_h) {
- nns_edge_loge ("Invalid state, MQTT connection was not completed.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
- handle = bh->mqtt_h;
-
- if (!MQTTAsync_isConnected (handle)) {
- nns_edge_loge ("Failed to publish message, MQTT is not connected.");
- return NNS_EDGE_ERROR_IO;
- }
-
- /* Publish a message (default QoS 1 - at least once and retained true). */
- ret = MQTTAsync_send (handle, eh->topic, length, data, 1, 1, NULL);
- if (ret != MQTTASYNC_SUCCESS) {
- nns_edge_loge ("Failed to publish a message (ID:%s, Topic:%s).",
- eh->id, eh->topic);
- return NNS_EDGE_ERROR_IO;
- }
-
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Subscribe a topic.
- * @note This is internal function for MQTT broker. You should call this with edge-handle lock.
- */
-int
-nns_edge_mqtt_subscribe (nns_edge_h edge_h)
-{
- nns_edge_handle_s *eh;
- nns_edge_broker_s *bh;
- MQTTAsync handle;
- int ret;
-
- eh = (nns_edge_handle_s *) edge_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (eh) || !eh->broker_h) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- bh = (nns_edge_broker_s *) eh->broker_h;
- if (!bh->mqtt_h) {
- nns_edge_loge ("Invalid state, MQTT connection was not completed.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
- handle = bh->mqtt_h;
-
- if (!MQTTAsync_isConnected (handle)) {
- nns_edge_loge ("Invalid state, MQTT connection was not completed.");
- return NNS_EDGE_ERROR_IO;
- }
-
- /* Subscribe a topic (default QoS 1 - at least once). */
- ret = MQTTAsync_subscribe (handle, eh->topic, 1, NULL);
- if (ret != MQTTASYNC_SUCCESS) {
- nns_edge_loge ("Failed to subscribe a topic (ID:%s, Topic:%s).",
- eh->id, eh->topic);
- return NNS_EDGE_ERROR_IO;
- }
-
- return NNS_EDGE_ERROR_NONE;
-}
-
-/**
- * @brief Check mqtt connection
- */
-bool
-nns_edge_mqtt_is_connected (nns_edge_h edge_h)
-{
- nns_edge_handle_s *eh;
- nns_edge_broker_s *bh;
- MQTTAsync handle;
- eh = (nns_edge_handle_s *) edge_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (eh) || !eh->broker_h) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- return false;
- }
-
- bh = (nns_edge_broker_s *) eh->broker_h;
- if (!bh->mqtt_h) {
- nns_edge_loge ("Invalid state, MQTT connection was not completed.");
- return false;
- }
- handle = bh->mqtt_h;
-
- if (MQTTAsync_isConnected (handle)) {
- return true;
- }
-
- return false;
-}
-
-/**
- * @brief Get message from mqtt broker.
- */
-int
-nns_edge_mqtt_get_message (nns_edge_h edge_h, char **msg)
-{
- nns_edge_handle_s *eh;
- nns_edge_broker_s *bh;
-
- eh = (nns_edge_handle_s *) edge_h;
-
- if (!NNS_EDGE_MAGIC_IS_VALID (eh) || !eh->broker_h) {
- nns_edge_loge ("Invalid param, given edge handle is invalid.");
- return NNS_EDGE_ERROR_INVALID_PARAMETER;
- }
-
- bh = (nns_edge_broker_s *) eh->broker_h;
-
- *msg = g_async_queue_timeout_pop (bh->server_list, DEFAULT_SUB_TIMEOUT);
- if (!*msg) {
- nns_edge_loge ("Failed to get message from mqtt broker within timeout");
- return NNS_EDGE_ERROR_UNKNOWN;
- }
-
- return NNS_EDGE_ERROR_NONE;
-}
$(NNSTREAMER_GST_HOME)/elements/gsttensor_sparseutil.c \
$(NNSTREAMER_GST_HOME)/elements/gsttensor_split.c \
$(NNSTREAMER_GST_HOME)/elements/gsttensor_transform.c \
- $(NNSTREAMER_GST_HOME)/tensor_filter/tensor_filter.c \
+ $(NNSTREAMER_GST_HOME)/tensor_filter/tensor_filter.c
# tensor-query element with nnstreamer-edge
NNSTREAMER_QUERY_SRCS := \
$(NNSTREAMER_GST_HOME)/tensor_query/tensor_query_client.c \
$(NNSTREAMER_GST_HOME)/tensor_query/tensor_query_serversink.c \
$(NNSTREAMER_GST_HOME)/tensor_query/tensor_query_serversrc.c \
- $(NNSTREAMER_GST_HOME)/tensor_query/tensor_query_server.c \
- $(NNSTREAMER_GST_HOME)/tensor_query/nnstreamer_edge_common.c \
- $(NNSTREAMER_GST_HOME)/tensor_query/nnstreamer_edge_internal.c
+ $(NNSTREAMER_GST_HOME)/tensor_query/tensor_query_server.c
# source AMC (Android MediaCodec)
NNSTREAMER_SOURCE_AMC_SRCS := \
'target': 'npu-engine',
'project_args': { 'ENABLE_TRIX_ENGINE' : 1 }
},
- 'aitt-support': {
- 'target': 'aitt',
- 'project_args': { 'ENABLE_AITT' : 1 }
- },
'nnstreamer-edge-support': {
'target': 'nnstreamer-edge',
'project_args': { 'ENABLE_NNSTREAMER_EDGE': 1 }
option('lua-support', type: 'feature', value: 'auto')
option('mqtt-support', type: 'feature', value: 'auto')
option('tvm-support', type: 'feature', value: 'auto')
-option('query-hybrid-support', type: 'feature', value: 'auto')
option('trix-engine-support', type: 'feature', value: 'auto')
option('nnstreamer-edge-support', type: 'feature', value: 'auto')
-option('aitt-support', type: 'feature', value: 'auto')
option('mxnet-support', type: 'feature', value: 'auto')
# booleans & other options
%{_prefix}/lib/nnstreamer/decoders/libnnstreamer_decoder_octet_stream.so
%{_prefix}/lib/nnstreamer/filters/libnnstreamer_filter_cpp.so
%{gstlibdir}/libnnstreamer.so
-# TODO migrate nnstreamer-edge and remove below library
-%{_libdir}/libnnstreamer-edge-temp.so
%{_libdir}/libnnstreamer.so
%files single
test('unittest_join', unittest_join, env: testenv)
# Run unittest_query
- unittest_query = executable('unittest_query',
- join_paths('nnstreamer_query', 'unittest_query.cc'),
- dependencies: [nnstreamer_unittest_deps],
- install: get_option('install-test'),
- install_dir: unittest_install_dir
- )
+ if nnstreamer_edge_support_is_available
+ unittest_query = executable('unittest_query',
+ join_paths('nnstreamer_query', 'unittest_query.cc'),
+ dependencies: [nnstreamer_unittest_deps],
+ install: get_option('install-test'),
+ install_dir: unittest_install_dir
+ )
- test('unittest_query', unittest_query, env: testenv)
+ test('unittest_query', unittest_query, env: testenv)
+ endif
# Run unittest_edge
if nnstreamer_edge_support_is_available
test('unittest_edge', unittest_edge, env: testenv)
endif
-
+
# Run unittest_mqtt
if mqtt_support_is_available
unittest_mqtt_w_helper = executable('unittest_mqtt_w_helper',
projects / platform / upstream / nnstreamer.git / commitdiff