path: root/src/message_port_local.c

/*
 * Copyright (c) 2017 Samsung Electronics Co., Ltd All Rights Reserved
 *
 * Licensed under the Apache License, Version 2.0 (the License);
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an AS IS BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#define _GNU_SOURCE

#include <bundle.h>
#include <bundle_internal.h>
#include <aul.h>

#include <sys/socket.h>
#include <stdlib.h>
#include <stdio.h>
#include <pthread.h>

#include <glib.h>
#include <gio/gio.h>
#include <gio/gunixfdlist.h>
#include <glib-unix.h>

#include "message_port_log.h"
#include "message_port.h"
#include "message_port_remote.h"
#include "message_port_common.h"

#define MAX_RETRY_CNT 10
#define SOCK_PAIR_SENDER 0
#define SOCK_PAIR_RECEIVER 1

static bool _initialized = false;
static GHashTable *__remote_app_info;
static GHashTable *__registered_callback_info_hash;
static const int MAX_MESSAGE_SIZE = 16 * 1024;

enum __certificate_info_type {
	UNKNOWN = 0,
	CERTIFICATE_MATCH,
	CERTIFICATE_NOT_MATCH,
};

typedef struct message_port_remote_port_info {
	char *remote_app_id;
	int certificate_info;
	GList *port_list;
} message_port_remote_app_info_s;

typedef struct port_list_info {
	message_port_remote_app_info_s *remote_app_info;
	char *port_name;
	char *encoded_bus_name;
	bool is_trusted;
	int send_sock_fd;
	bool exist;
	GIOChannel *gio_read;
	int g_src_id;
	GList *delayed_message_list;
	unsigned int delayed_message_size;
	int delay_src_id;
} port_list_info_s;

typedef struct registered_callback_info {
	char *remote_app_id;
	char *remote_port;
	bool is_trusted;
	int watcher_id;
	void *user_data;
	message_port_registration_event_cb registered_cb;
	message_port_registration_event_cb unregistered_cb;
} registered_callback_info_s;

enum transmission_sequence {
	SEQUENCE_START = 0,
	SEQUENCE_PORT_LEN,
	SEQUENCE_PORT_NAME,
	SEQUENCE_BIDIRECTION,
	SEQUENCE_TRUSTED,
	SEQUENCE_DTAT_LEN,
	SEQUENCE_DATA,
	SEQUENCE_END
};

typedef struct delay_message {
	unsigned int size;
	unsigned int sent_bytes;
	int sequence;
	int local_port_len;
	char *local_port_name;
	bool is_bidirection;
	bool local_trusted;
	int data_len;
	bundle_raw *data;
} delay_message_info_s;

extern pthread_mutex_t mutex;

/* LCOV_EXCL_START */
static void __free_delay_message_info(delay_message_info_s *message)
{
	if (message != NULL) {
		FREE_AND_NULL(message->local_port_name);
		FREE_AND_NULL(message->data);
		FREE_AND_NULL(message);
	}
}
/* LCOV_EXCL_STOP */

/* LCOV_EXCL_START */
static void __free_list_delay_message_info(gpointer data)
{
	delay_message_info_s *message = (delay_message_info_s *)data;

	if (message != NULL)
		__free_delay_message_info(message);
}
/* LCOV_EXCL_STOP */

/* LCOV_EXCL_START */
static void __clear_disconnect_socket(port_list_info_s *port_info)
{
	GError *error = NULL;

	if (port_info == NULL)
		return;

	if (port_info->gio_read != NULL) {
		g_io_channel_shutdown(port_info->gio_read, TRUE, &error);
		if (error) {
			_LOGE("g_io_channel_shutdown error : %s", error->message);
			g_error_free(error);
		}
		g_io_channel_unref(port_info->gio_read);
		port_info->gio_read = NULL;
	}

	if (port_info->g_src_id != 0) {
		g_source_remove(port_info->g_src_id);
		port_info->g_src_id = 0;
	}

	if (port_info->delay_src_id != 0) {
		g_source_remove(port_info->delay_src_id);
		port_info->delay_src_id = 0;
	}

	if (port_info->delayed_message_list != NULL) {
		g_list_free_full(port_info->delayed_message_list, __free_list_delay_message_info);
		/* can be reused */
		port_info->delayed_message_list = NULL;
	}

	port_info->delayed_message_size = 0;
	port_info->send_sock_fd = 0;
}
/* LCOV_EXCL_STOP */

/* LCOV_EXCL_START */
static void __free_port_info(gpointer data)
{
	port_list_info_s *port_info = (port_list_info_s *)data;
	message_port_remote_app_info_s *remote_app_info;

	if (port_info == NULL)
		return;

	remote_app_info = port_info->remote_app_info;

	_LOGI("__free_port_info : remote_app_id : %s port_name : %s",
			remote_app_info->remote_app_id,
			port_info->port_name);

	remote_app_info->port_list = g_list_remove(remote_app_info->port_list,
			port_info);

	__clear_disconnect_socket(port_info);

	if (port_info->encoded_bus_name)
		free(port_info->encoded_bus_name);
	if (port_info->port_name)
		free(port_info->port_name);

	free(port_info);

	if (g_list_length(remote_app_info->port_list) == 0) {
		g_hash_table_remove(__remote_app_info,
				remote_app_info->remote_app_id);
	}
}
/* LCOV_EXCL_STOP */

/* LCOV_EXCL_START */
static void __hash_destory_remote_value(gpointer data)
{
	message_port_remote_app_info_s *mri = (message_port_remote_app_info_s *)data;
	if (mri) {
		FREE_AND_NULL(mri->remote_app_id);
		if (mri->port_list)
			g_list_free_full(mri->port_list, __free_port_info);

		free(mri);
	}
}
/* LCOV_EXCL_STOP */

static void __registered_callback_info_free(gpointer data)
{
	registered_callback_info_s *callback_info = (registered_callback_info_s *)data;
	if (callback_info == NULL)
		return;

	if (callback_info->remote_app_id)
		free(callback_info->remote_app_id);

	if (callback_info->remote_port)
		free(callback_info->remote_port);

	free(callback_info);
}

static bool __initialize(void)
{
	if (!initialized_common) {
		if (!initialize_common())
			return false;
	}

	if (__remote_app_info == NULL) {
		__remote_app_info = g_hash_table_new_full(g_str_hash, g_str_equal, NULL, __hash_destory_remote_value);
		retvm_if(!__remote_app_info, false, "fail to create __remote_app_info");
	}

	if (__registered_callback_info_hash == NULL) {
		__registered_callback_info_hash = g_hash_table_new_full(g_direct_hash,  g_direct_equal, NULL, __registered_callback_info_free);
		retvm_if(!__registered_callback_info_hash, false, "fail to create __registered_callback_info_hash");
	}

	_initialized = true;

	return true;
}

static int __remote_port_compare_cb(gconstpointer a, gconstpointer b)
{
	port_list_info_s *key1 = (port_list_info_s *)a;
	port_list_info_s *key2 = (port_list_info_s *)b;

	if (key1->is_trusted == key2->is_trusted)
		return strcmp(key1->port_name, key2->port_name);

	return 1;
}

static port_list_info_s *__set_remote_port_info(const char *remote_app_id, const char *remote_port, bool is_trusted)
{
	int ret_val = MESSAGE_PORT_ERROR_NONE;
	port_list_info_s *port_info = (port_list_info_s *)calloc(1, sizeof(port_list_info_s));

	if (!port_info) {
		ret_val = MESSAGE_PORT_ERROR_OUT_OF_MEMORY;
		goto out;
	}
	port_info->port_name = strdup(remote_port);
	if (!port_info->port_name) {
		ret_val = MESSAGE_PORT_ERROR_OUT_OF_MEMORY;
		goto out;
	}
	port_info->is_trusted = is_trusted;
	port_info->encoded_bus_name = get_encoded_name(remote_app_id, remote_port, is_trusted);
	if (port_info->encoded_bus_name == NULL) {
		ret_val = MESSAGE_PORT_ERROR_OUT_OF_MEMORY;
		goto out;
	}
	port_info->send_sock_fd = 0;
out:
	if (ret_val != MESSAGE_PORT_ERROR_NONE) {
		if (port_info) {
			FREE_AND_NULL(port_info->port_name);
			FREE_AND_NULL(port_info->encoded_bus_name);
			free(port_info);
		}
		return NULL;
	}
	return port_info;
}

static message_port_remote_app_info_s *__set_remote_app_info(const char *remote_app_id, const char *remote_port, bool is_trusted)
{
	message_port_remote_app_info_s *remote_app_info = NULL;
	int ret_val = MESSAGE_PORT_ERROR_NONE;

	remote_app_info = (message_port_remote_app_info_s *)calloc(1, sizeof(message_port_remote_app_info_s));
	if (!remote_app_info) {
		ret_val = MESSAGE_PORT_ERROR_OUT_OF_MEMORY;
		goto out;
	}

	remote_app_info->remote_app_id = strdup(remote_app_id);
	if (remote_app_info->remote_app_id == NULL) {
		ret_val = MESSAGE_PORT_ERROR_OUT_OF_MEMORY;
		goto out;
	}

out:
	if (ret_val != MESSAGE_PORT_ERROR_NONE) {
		if (remote_app_info) {
			FREE_AND_NULL(remote_app_info->remote_app_id);
			FREE_AND_NULL(remote_app_info);
		}
		return NULL;
	}
	return remote_app_info;
}

static gboolean __socket_disconnect_handler(GIOChannel *gio,
		GIOCondition cond,
		gpointer data)
{
	_LOGI("__socket_disconnect_handler %d", cond);
	__free_port_info(data);

	return FALSE;
}

static int __get_remote_port_info(const char *remote_app_id, const char *remote_port, bool is_trusted,
		message_port_remote_app_info_s **mri, port_list_info_s **pli)
{
	message_port_remote_app_info_s *remote_app_info = NULL;
	port_list_info_s port_info;
	GList *cb_list = NULL;
	int ret_val = MESSAGE_PORT_ERROR_NONE;

	remote_app_info = (message_port_remote_app_info_s *)g_hash_table_lookup(__remote_app_info, remote_app_id);

	if (remote_app_info == NULL) {
		remote_app_info = __set_remote_app_info(remote_app_id, remote_port, is_trusted);

		if (remote_app_info == NULL) {
			ret_val = MESSAGE_PORT_ERROR_OUT_OF_MEMORY;
			goto out;
		}
		g_hash_table_insert(__remote_app_info, remote_app_info->remote_app_id, remote_app_info);
	}
	*mri = remote_app_info;

	port_info.port_name = strdup(remote_port);
	if (port_info.port_name == NULL) {
		ret_val = MESSAGE_PORT_ERROR_OUT_OF_MEMORY;
		goto out;
	}
	port_info.is_trusted = is_trusted;
	cb_list = g_list_find_custom(remote_app_info->port_list, &port_info,
					(GCompareFunc)__remote_port_compare_cb);
	if (port_info.port_name)
		free(port_info.port_name);
	if (cb_list == NULL) {
		port_list_info_s *tmp = __set_remote_port_info(remote_app_id, remote_port, is_trusted);
		if (tmp == NULL) {
			ret_val = MESSAGE_PORT_ERROR_OUT_OF_MEMORY;
			goto out;
		}
		remote_app_info->port_list = g_list_append(remote_app_info->port_list, tmp);
		tmp->remote_app_info = remote_app_info;
		*pli = tmp;
	} else {
		*pli = (port_list_info_s *)cb_list->data;
	}
out:

	return ret_val;
}

int check_remote_port(const char *remote_app_id, const char *remote_port, bool is_trusted, bool *exist)
{
	_LOGD("Check a remote port : [%s:%s]", remote_app_id, remote_port);

	GVariant *result = NULL;
	GError *err = NULL;
	int ret_val = MESSAGE_PORT_ERROR_NONE;
	char *bus_name = NULL;
	message_port_remote_app_info_s *remote_app_info = NULL;
	port_list_info_s *port_info = NULL;
	int local_reg_id = 0;
	message_port_local_port_info_s *mi = NULL;
	gboolean name_exist = false;

	if (!_initialized) {
		if (!__initialize())
			return MESSAGE_PORT_ERROR_IO_ERROR;
	}

	_LOGD("remote_app_id, app_id :[%s : %s] ", remote_app_id, app_id);

	ret_val = __get_remote_port_info(remote_app_id, remote_port, is_trusted, &remote_app_info, &port_info);
	if (ret_val != MESSAGE_PORT_ERROR_NONE)
		return ret_val;

	/* self check */
	if (strcmp(remote_app_id, app_id) == 0) {

		_LOGD("__is_local_port_registed ");
		if (!is_local_port_registed(remote_port, is_trusted, &local_reg_id, &mi))
			*exist = false;
		else
			*exist = true;

		_LOGD("__is_local_port_registed : %d ", *exist);
		return MESSAGE_PORT_ERROR_NONE;
	}

	port_info->exist = false;
	bus_name = port_info->encoded_bus_name;

	result = g_dbus_connection_call_sync(
			gdbus_conn,
			DBUS_SERVICE_DBUS,
			DBUS_PATH_DBUS,
			DBUS_INTERFACE_DBUS,
			"NameHasOwner",
			g_variant_new("(s)", bus_name),
			G_VARIANT_TYPE("(b)"),
			G_DBUS_CALL_FLAGS_NONE,
			-1,
			NULL,
			&err);

	if (err || (result == NULL)) {
		if (err) {
			_LOGE("No reply. error = %s", err->message);
			g_error_free(err);
		}
		ret_val = MESSAGE_PORT_ERROR_RESOURCE_UNAVAILABLE;
	} else {
		g_variant_get(result, "(b)", &name_exist);

		if (!name_exist) {
			_LOGI("Name not exist %s", bus_name);
			*exist = false;
			ret_val = MESSAGE_PORT_ERROR_NONE;
		} else {

			if (is_trusted) {
				if (remote_app_info->certificate_info != CERTIFICATE_MATCH) {
					if (!is_preloaded(app_id, remote_app_id)) {
						if (check_certificate(app_id, remote_app_id) != MESSAGE_PORT_ERROR_NONE) {
							ret_val = MESSAGE_PORT_ERROR_CERTIFICATE_NOT_MATCH;
							goto out;
						}
					}
					remote_app_info->certificate_info = CERTIFICATE_MATCH;
				}
			}
			port_info->exist = true;
			*exist = true;
			ret_val = MESSAGE_PORT_ERROR_NONE;
		}
	}
out:
	if (result)
		g_variant_unref(result);

	if (ret_val != MESSAGE_PORT_ERROR_NONE || !name_exist)
		__free_port_info((gpointer)port_info);

	return ret_val;
}

/* LCOV_EXCL_START */
static int __send_delayed_message(int sockfd, delay_message_info_s *message)
{
	unsigned int nb = 0;
	int sequence = message->sequence - 1;
	int ret = MESSAGE_PORT_ERROR_NONE;
	bool is_startline = true;
	int offset = 0;

	_LOGI("send_delayed_message : sockfd (%d) sequence(%d) sent byte(%d)",
		sockfd, message->sequence, message->sent_bytes);

	switch (message->sequence) {
	case SEQUENCE_START:
		sequence++;
		is_startline = false;

	case SEQUENCE_PORT_LEN:
		if (is_startline)
			offset = message->sent_bytes;

		ret = write_socket(sockfd, ((char *)&message->local_port_len) + offset,
				sizeof(message->local_port_len) - offset, &nb, &sequence);
		if (ret != MESSAGE_PORT_ERROR_NONE) {
			_LOGE("write local_port_len fail");
			goto out;
		}
		offset = 0;
		is_startline = false;

	case SEQUENCE_PORT_NAME:
		if (is_startline)
			offset = message->sent_bytes;

		if (message->local_port_len > 0)
			ret = write_socket(sockfd, message->local_port_name + offset,
				message->local_port_len - offset , &nb, &sequence);
		else
			sequence++;

		if (ret != MESSAGE_PORT_ERROR_NONE) {
			_LOGE("write local_port fail");
			goto out;
		}
		offset = 0;
		is_startline = false;

	case SEQUENCE_BIDIRECTION:
		if (is_startline)
			offset = message->sent_bytes;

		ret = write_socket(sockfd, ((char *)&message->is_bidirection) + offset,
				sizeof(message->is_bidirection) - offset, &nb, &sequence);
		if (ret != MESSAGE_PORT_ERROR_NONE) {
			_LOGE("write is_bidirection fail");
			goto out;
		}
		offset = 0;
		is_startline = false;

	case SEQUENCE_TRUSTED:
		if (is_startline)
			offset = message->sent_bytes;

		ret = write_socket(sockfd, ((char *)&message->local_trusted) + offset,
				sizeof(message->local_trusted) - offset, &nb, &sequence);
		if (ret != MESSAGE_PORT_ERROR_NONE) {
			_LOGE("write local_trusted fail");
			goto out;
		}
		offset = 0;
		is_startline = false;

	case SEQUENCE_DTAT_LEN:
		if (is_startline)
			offset = message->sent_bytes;

		ret = write_socket(sockfd, ((char *)&message->data_len) + offset,
				sizeof(message->data_len) - offset, &nb, &sequence);
		if (ret != MESSAGE_PORT_ERROR_NONE) {
			_LOGE("write data_len fail");
			goto out;
		}
		offset = 0;
		is_startline = false;

	case SEQUENCE_DATA:
		if (is_startline)
			offset = message->sent_bytes;

		ret = write_socket(sockfd, (char *)message->data + offset,
			message->data_len - offset, &nb, &sequence);

		if (ret != MESSAGE_PORT_ERROR_NONE) {
			_LOGE("write data fail");
			goto out;
		}
		offset = 0;
		is_startline = false;

	default:
		ret = MESSAGE_PORT_ERROR_NONE;

	}

out:
	if (ret == MESSAGE_PORT_ERROR_RESOURCE_UNAVAILABLE) {
		if (is_startline)
			message->sent_bytes += nb;
		else
			message->sent_bytes = nb;

		message->sequence = sequence;
		_LOGE("send_delayed_message fail : sockfd (%d) sequence(%d) sent byte(%d)",
			sockfd, message->sequence, message->sent_bytes);
	}

	return ret;

}
/* LCOV_EXCL_STOP */

/* LCOV_EXCL_START */
static gboolean __process_delayed_message(gint fd, GIOCondition cond, gpointer data)
{
	port_list_info_s *port_info = (port_list_info_s *)data;
	delay_message_info_s *message;
	int ret;

	if (port_info == NULL)
		return G_SOURCE_REMOVE;

	pthread_mutex_lock(&mutex);

	if (port_info->delayed_message_list == NULL) {
		port_info->delayed_message_size = 0;
		port_info->delay_src_id = 0;
		pthread_mutex_unlock(&mutex);
		return G_SOURCE_REMOVE;
	} else {
		message = g_list_nth_data(port_info->delayed_message_list, 0);
		ret = __send_delayed_message(port_info->send_sock_fd, message);

		if (ret == MESSAGE_PORT_ERROR_RESOURCE_UNAVAILABLE) {
			pthread_mutex_unlock(&mutex);
			return G_SOURCE_CONTINUE;
		} else if (ret == MESSAGE_PORT_ERROR_IO_ERROR) {
			__free_port_info((gpointer)port_info);
			pthread_mutex_unlock(&mutex);
			return G_SOURCE_REMOVE;
		}

		port_info->delayed_message_size -= message->size;

		port_info->delayed_message_list = g_list_remove(port_info->delayed_message_list, message);
		__free_delay_message_info(message);
	}

	pthread_mutex_unlock(&mutex);

	return G_SOURCE_CONTINUE;
}
/* LCOV_EXCL_STOP */

/* LCOV_EXCL_START */
static int __insert_delayed_message(port_list_info_s *port_info,
	int sequence,
	bundle_raw *kb_data,
	int data_len,
	unsigned int sent_bytes,
	const char *local_port,
	bool local_trusted,
	bool is_bidirection)
{
#define QUEUE_SIZE_MAX (1024 * 1024) /* 1MB per remote port (MAX) */

	unsigned int tmp_size;
	unsigned int message_size;
	int ret = MESSAGE_PORT_ERROR_NONE;

	if (port_info->delayed_message_size >= QUEUE_SIZE_MAX) {
		_LOGE("cache fail : delayed_message_size (%d), count(%d)",
			port_info->delayed_message_size, g_list_length(port_info->delayed_message_list));
		return MESSAGE_PORT_ERROR_RESOURCE_UNAVAILABLE;
	}

	delay_message_info_s *message = (delay_message_info_s *)calloc(1, sizeof(delay_message_info_s));
	retvm_if(!message, MESSAGE_PORT_ERROR_OUT_OF_MEMORY, "Malloc failed");

	message_size = sizeof(delay_message_info_s);

	message->sequence = sequence;
	tmp_size = strlen(local_port) + 1;
	message_size += tmp_size;
	message->local_port_len = tmp_size;
	message->local_port_name = strdup(local_port);
	if (message->local_port_name == NULL) {
		_LOGE("local_port_name strdup fail");
		ret = MESSAGE_PORT_ERROR_OUT_OF_MEMORY;
		goto out;
	}
	message->is_bidirection = is_bidirection;
	message->local_trusted = local_trusted;
	message_size += data_len;
	message->data_len = data_len;
	message->data = (bundle_raw *)strdup((const char *)kb_data);
	if (message->data == NULL) {
		_LOGE("data strdup fail");
		ret = MESSAGE_PORT_ERROR_OUT_OF_MEMORY;
		goto out;
	}


	message->sent_bytes = sent_bytes;
	message->size = message_size;
	port_info->delayed_message_size += message_size;

	port_info->delayed_message_list = g_list_append(port_info->delayed_message_list, message);

	if (port_info->delay_src_id == 0) {
			port_info->delay_src_id = g_unix_fd_add_full(G_PRIORITY_DEFAULT,
							port_info->send_sock_fd, G_IO_OUT, __process_delayed_message,
							port_info, NULL);
	}

	_LOGE("inserted : pm(%s) fd(%d) ms(%d) ds(%d) dlc(%d) sqn(%d) sb (%d)",
		port_info->port_name, port_info->send_sock_fd, message_size,
		port_info->delayed_message_size,
		g_list_length(port_info->delayed_message_list), sequence, sent_bytes);



out:
	if (ret != MESSAGE_PORT_ERROR_NONE)
		__free_delay_message_info(message);

	return ret;
}
/* LCOV_EXCL_STOP */

static int __message_port_send_async(port_list_info_s *port_info, bundle *kb, const char *local_port,
		bool local_trusted, bool is_bidirection)
{
	int ret = 0;
	int data_len;
	unsigned int nb = 0;
	bundle_raw *kb_data = NULL;
	int sequence = SEQUENCE_START;

	bundle_encode(kb, &kb_data, &data_len);
	if (kb_data == NULL) {
		_LOGE("bundle encode fail");
		ret = MESSAGE_PORT_ERROR_INVALID_PARAMETER;
		goto out;
	}

	if (data_len > MAX_MESSAGE_SIZE) {
		_LOGE("bigger than max size\n");
		ret = MESSAGE_PORT_ERROR_MAX_EXCEEDED;
		goto out;
	}

	if (g_list_length(port_info->delayed_message_list) > 0) {
		ret = MESSAGE_PORT_ERROR_RESOURCE_UNAVAILABLE;
		_LOGE("There are messages in the delayed_message_list (count %d)",
			g_list_length(port_info->delayed_message_list));
		goto out;
	}

	ret = write_string_to_socket(port_info->send_sock_fd, local_port,
			strlen(local_port) + 1, &nb, &sequence);
	if (ret != MESSAGE_PORT_ERROR_NONE) {
		_LOGE("write local_port fail");
		goto out;
	}

	ret = write_socket(port_info->send_sock_fd, (char *)&is_bidirection,
			sizeof(is_bidirection), &nb, &sequence);
	if (ret != MESSAGE_PORT_ERROR_NONE) {
		_LOGE("write is_bidirection fail");
		goto out;
	}

	ret = write_socket(port_info->send_sock_fd, (char *)&local_trusted,
			sizeof(local_trusted), &nb, &sequence);
	if (ret != MESSAGE_PORT_ERROR_NONE) {
		_LOGE("write local_trusted fail");
		goto out;
	}

	ret = write_string_to_socket(port_info->send_sock_fd, (void *)kb_data,
			data_len, &nb, &sequence);
	if (ret != MESSAGE_PORT_ERROR_NONE) {
		_LOGE("write kb_data fail");
		goto out;
	}

out:
	if (ret == MESSAGE_PORT_ERROR_RESOURCE_UNAVAILABLE) {
		ret = __insert_delayed_message(port_info, sequence, kb_data, data_len, nb,
			local_port, local_trusted, is_bidirection);
		if (ret != MESSAGE_PORT_ERROR_NONE)
			ret = MESSAGE_PORT_ERROR_IO_ERROR;
	}

	if (kb_data)
		free(kb_data);

	return ret;
}

int send_message(const char *remote_appid, const char *remote_port,
		const char *local_port, bool trusted_message, bool local_trusted, bool bi_dir, bundle *message)
{

	int ret = MESSAGE_PORT_ERROR_NONE;
	GUnixFDList *fd_list = NULL;

	int len = 0;
	bundle_raw *raw = NULL;
	char *bus_name = NULL;
	char *interface_name = NULL;

	message_port_remote_app_info_s *remote_app_info = NULL;
	port_list_info_s *port_info = NULL;
	GDBusMessage *msg = NULL;
	GError *err = NULL;
	GVariant *body = NULL;
	int sock_pair[2] = {0,};
	char buf[1024];

	if (!_initialized) {
		if (!__initialize())
			return MESSAGE_PORT_ERROR_IO_ERROR;
	}

	ret = __get_remote_port_info(remote_appid, remote_port, trusted_message, &remote_app_info, &port_info);
	if (ret != MESSAGE_PORT_ERROR_NONE)
		return ret;

	if (port_info->exist == false) {
		bool exist = false;
		_LOGD("port exist check !!");
		ret =  check_remote_port(remote_appid, remote_port, trusted_message, &exist);
		if (ret != MESSAGE_PORT_ERROR_NONE)
			return ret;
		else if (!exist)
			return MESSAGE_PORT_ERROR_PORT_NOT_FOUND;
	}

	if (port_info->send_sock_fd > 0) {
		ret = __message_port_send_async(port_info, message,
				(local_port) ? local_port : "", local_trusted, bi_dir);
	} else {

		bus_name = port_info->encoded_bus_name;
		interface_name = bus_name;

		if (bundle_encode(message, &raw, &len) != BUNDLE_ERROR_NONE) {
			ret = MESSAGE_PORT_ERROR_INVALID_PARAMETER;
			goto out;
		}

		if (MAX_MESSAGE_SIZE < len) {
			_LOGE("The size of message (%d) has exceeded the maximum limit.", len);
			ret = MESSAGE_PORT_ERROR_MAX_EXCEEDED;
			goto out;
		}

		body = g_variant_new("(ssbbssbus)", app_id, (local_port) ? local_port : "", local_trusted, bi_dir,
				remote_appid, remote_port, trusted_message, len, raw);
		if (strcmp(remote_appid, app_id) != 0) { /* self send */

			/*  if message-port fail to get socket pair, communicate using GDBus */
			if (aul_request_message_port_socket_pair(sock_pair) != AUL_R_OK) {
				_LOGE("error create socket pair");
			} else {

				_LOGI("sock pair : %d, %d",
						sock_pair[SOCK_PAIR_SENDER], sock_pair[SOCK_PAIR_RECEIVER]);
				fd_list = g_unix_fd_list_new();
				g_unix_fd_list_append(fd_list, sock_pair[SOCK_PAIR_RECEIVER], &err);
				if (err != NULL) {
					_LOGE("g_unix_fd_list_append [%s]", err->message);
					ret = MESSAGE_PORT_ERROR_IO_ERROR;
					g_error_free(err);
					goto out;
				}

				port_info->send_sock_fd = sock_pair[SOCK_PAIR_SENDER];
				close(sock_pair[SOCK_PAIR_RECEIVER]);
				sock_pair[SOCK_PAIR_RECEIVER] = 0;

				port_info->gio_read = g_io_channel_unix_new(port_info->send_sock_fd);
				if (!port_info->gio_read) {
					_LOGE("Error is %s\n", strerror_r(errno, buf, sizeof(buf)));
					ret = MESSAGE_PORT_ERROR_IO_ERROR;
					goto out;
				}

				port_info->g_src_id = g_io_add_watch(
						port_info->gio_read,
						G_IO_IN | G_IO_HUP,
						__socket_disconnect_handler,
						(gpointer)port_info);
				if (port_info->g_src_id == 0) {
					_LOGE("fail to add watch on socket");
					ret = MESSAGE_PORT_ERROR_IO_ERROR;
					goto out;
				}

			}
		}

		msg = g_dbus_message_new_method_call(bus_name, MESSAGEPORT_OBJECT_PATH, interface_name, "send_message");
		if (!msg) {
			_LOGE("Can't allocate new method call");
			ret = MESSAGE_PORT_ERROR_OUT_OF_MEMORY;
			goto out;
		}

		g_dbus_message_set_unix_fd_list(msg, fd_list);
		g_dbus_message_set_body(msg, body);
		g_dbus_connection_send_message(gdbus_conn, msg, G_DBUS_SEND_MESSAGE_FLAGS_NONE, NULL, &err);
		if (err != NULL) {
			_LOGE("No reply. error = %s", err->message);
			g_error_free(err);
			ret = MESSAGE_PORT_ERROR_IO_ERROR;
			goto out;
		}
	}

out:
	if (msg)
		g_object_unref(msg);
	if (raw)
		bundle_free_encoded_rawdata(&raw);
	if (fd_list)
		g_object_unref(fd_list);

	if (ret != MESSAGE_PORT_ERROR_NONE) {
		__free_port_info((gpointer)port_info);
		if (sock_pair[SOCK_PAIR_SENDER])
			close(sock_pair[SOCK_PAIR_SENDER]);
		if (sock_pair[SOCK_PAIR_RECEIVER])
			close(sock_pair[SOCK_PAIR_RECEIVER]);
	}

	return ret;
}

int send_bidirectional_message(int id, const char *remote_app_id, const char *remote_port,  bool trusted_message, bundle *message)
{
	message_port_local_port_info_s *local_info;
	int ret = get_local_port_info(id, &local_info);
	if (ret != MESSAGE_PORT_ERROR_NONE)
		return ret;

	_LOGD("bidirectional_message %s", local_info->port_name);
	return send_message(remote_app_id, remote_port,
			local_info->port_name, trusted_message, local_info->is_trusted, true, message);
}

static void __name_registered(GDBusConnection *connection,
		const gchar *name,
		const gchar *name_owner,
		gpointer user_data)
{

	registered_callback_info_s *info = (registered_callback_info_s *)user_data;
	if (info == NULL) {
		LOGE("NULL registered_callback_info");
		return;
	}

	_LOGI("watcher_id : %d, appeared name : %s , name_owner : %s\n", info->watcher_id, name, name_owner);
	if (info->registered_cb)
		info->registered_cb(info->remote_app_id, info->remote_port, info->is_trusted, info->user_data);
}

static void __name_unregistered(GDBusConnection *connection,
		const gchar *name,
		gpointer user_data)
{

	registered_callback_info_s *info = (registered_callback_info_s *)user_data;
	if (info == NULL) {
		LOGE("NULL registered_callback_info");
		return;
	}

	_LOGI("watcher_id : %d, vanished name : %s\n", info->watcher_id, name);
	if (info->unregistered_cb)
		info->unregistered_cb(info->remote_app_id, info->remote_port, info->is_trusted, info->user_data);
}

int watch_remote_port(int *watcher_id, const char *remote_app_id, const char *remote_port, bool trusted_remote_port, message_port_registration_event_cb registered_cb, message_port_registration_event_cb unregistered_cb, void *user_data)
{
	int ret_val = MESSAGE_PORT_ERROR_NONE;
	message_port_remote_app_info_s *remote_app_info = NULL;
	port_list_info_s *port_info = NULL;

	if (!_initialized) {
		if (!__initialize())
			return MESSAGE_PORT_ERROR_IO_ERROR;
	}
	_LOGI("remote_app_id, app_id :[%s : %s] ", remote_app_id, app_id);

	ret_val = __get_remote_port_info(remote_app_id, remote_port, trusted_remote_port, &remote_app_info, &port_info);
	if (ret_val != MESSAGE_PORT_ERROR_NONE) {
		_LOGE("Failed to get remote_port_info %d", ret_val);
		return ret_val;
	}

	registered_callback_info_s *registered_cb_info = (registered_callback_info_s *)calloc(1, sizeof(registered_callback_info_s));
	retvm_if(!registered_cb_info, MESSAGE_PORT_ERROR_OUT_OF_MEMORY, "Malloc failed");

	registered_cb_info->registered_cb = registered_cb;
	registered_cb_info->unregistered_cb = unregistered_cb;
	registered_cb_info->user_data = user_data;
	registered_cb_info->remote_app_id = strdup(remote_app_info->remote_app_id);
	if (registered_cb_info->remote_app_id == NULL) {
		free(registered_cb_info);
		_LOGE("Failed to alloc memory");
		return MESSAGE_PORT_ERROR_OUT_OF_MEMORY;
	}
	registered_cb_info->remote_port = strdup(port_info->port_name);
	if (registered_cb_info->remote_port == NULL) {
		free(registered_cb_info->remote_app_id);
		free(registered_cb_info);
		_LOGE("Failed to alloc memory");
		return MESSAGE_PORT_ERROR_OUT_OF_MEMORY;
	}

	registered_cb_info->watcher_id = g_bus_watch_name_on_connection(
			gdbus_conn,
			port_info->encoded_bus_name,
			G_BUS_NAME_WATCHER_FLAGS_NONE,
			__name_registered,
			__name_unregistered,
			registered_cb_info,
			NULL);
	if (registered_cb_info->watcher_id == 0) {
		free(registered_cb_info->remote_app_id);
		free(registered_cb_info->remote_port);
		free(registered_cb_info);
		_LOGE("Failed to watch name");
		return MESSAGE_PORT_ERROR_IO_ERROR;
	}

	g_hash_table_insert(__registered_callback_info_hash,
			GINT_TO_POINTER(registered_cb_info->watcher_id), registered_cb_info);

	*watcher_id = registered_cb_info->watcher_id;
	return MESSAGE_PORT_ERROR_NONE;
}

int remove_registration_event_cb(int watcher_id)
{
	registered_callback_info_s *registered_cb_info = NULL;
	gboolean remove_result = FALSE;

	if (watcher_id < 1)
		return MESSAGE_PORT_ERROR_INVALID_PARAMETER;

	registered_cb_info = g_hash_table_lookup(__registered_callback_info_hash, GINT_TO_POINTER(watcher_id));
	if (registered_cb_info == NULL)
		return MESSAGE_PORT_ERROR_INVALID_PARAMETER;

	remove_result = g_hash_table_remove(__registered_callback_info_hash, GINT_TO_POINTER(watcher_id));
	if (!remove_result)
		return MESSAGE_PORT_ERROR_IO_ERROR;

	g_bus_unwatch_name(watcher_id);

	return MESSAGE_PORT_ERROR_NONE;
}