daemon/src/ttd-app-interface.c

   1 /*
   2  * Copyright (c) 2018 Samsung Electronics Co., Ltd.
   3  *
   4  * Licensed under the Flora License, Version 1.1 (the License);
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  * http://floralicense.org/license/
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an AS IS BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #include <glib.h>
  18 #include <gio/gio.h>
  19 #include <gio/gunixsocketaddress.h>
  20 #include "ttd-log.h"
  21 #include "ttd-app-interface.h"
  22 #include "ttd-http.h"
  23 #include "common-app-inf.h"
  24 #include "ttd-app-data.h"
  25
  26 #define POST_DATA_URL "http://apitest.showiot.xyz/api/data"
  27 #define __POST_THREAD_RUN 1
  28 #define __POST_THREAD_STOP 0
  29 #define __QUEUE_TIME_OUT 500
  30
  31 #define APP_INF_INTROSPECTION \
  32 "<node>" \
  33 "  <interface name='"TTD_APP_INF_BUS_INF"'>" \
  34 "    <method name='"TTD_APP_INF_METHOD_REG"'>" \
  35 "      <arg type='s' name='project' direction='in'/>" \
  36 "      <arg type='s' name='token' direction='in'/>" \
  37 "      <arg type='s' name='appID' direction='in'/>" \
  38 "      <arg type='u' name='appPID' direction='in'/>" \
  39 "      <arg type='i' name='response' direction='out'/>" \
  40 "      <arg type='s' name='response_message' direction='out'/>" \
  41 "    </method>" \
  42 "    <method name='"TTD_APP_INF_METHOD_UNREG"'>" \
  43 "      <arg type='s' name='project' direction='in'/>" \
  44 "      <arg type='s' name='token' direction='in'/>" \
  45 "      <arg type='s' name='appID' direction='in'/>" \
  46 "      <arg type='u' name='appPID' direction='in'/>" \
  47 "      <arg type='i' name='response' direction='out'/>" \
  48 "      <arg type='s' name='response_message' direction='out'/>" \
  49 "    </method>" \
  50 "  </interface>" \
  51 "</node>"
  52
  53 typedef struct __thread_data_s {
  54         char *name;
  55         char *project;
  56         int disconnected;
  57         GMainContext *context;
  58         GMainLoop *loop;
  59         GThread *thread;
  60         GSocket *socket;
  61         GSocketAddress *addr;
  62         GAsyncQueue *data_queue;
  63         void *main_data;
  64 } thread_data;
  65
  66 struct _ttd_app_inf_h {
  67         guint gdbus_id;
  68         GDBusNodeInfo *introspection_data;
  69         GHashTable *app_sock_hash;
  70         GThread *post_thread;
  71         GAsyncQueue *data_queue;
  72 };
  73
  74 static const gpointer queue_quit_marker = (gpointer) &ttd_app_interface_init;
  75
  76 static char *
  77 __get_addr_name(const gchar *token, const gchar *appID, guint appPID)
  78 {
  79         return common_make_socket_addr_name(token, appID, appPID);
  80 }
  81
  82 static gboolean
  83 __on_disconnect(GIOChannel *ch, GIOCondition cond, gpointer user_data)
  84 {
  85         thread_data *data = user_data;
  86
  87         _E("connection of thread[%s] is broken", data->name);
  88
  89         data->disconnected = 1;
  90         g_main_loop_quit(data->loop);
  91
  92         return FALSE;
  93 }
  94
  95 static gboolean
  96 __receive_msg(GIOChannel *ch, GIOCondition cond, gpointer user_data)
  97 {
  98         thread_data *data = user_data;
  99         GSocket *socket = data->socket;
 100         GString *str = NULL;
 101         char buf[1024] = {'\0', };
 102         char *msg = NULL;
 103         gssize size = 0;
 104         gboolean connected = FALSE;
 105
 106         connected = g_socket_is_connected(socket);
 107         if (!connected) {
 108                 _E("socket is disconnected");
 109                 g_main_loop_quit(data->loop);
 110         }
 111
 112         while (0 < (size = g_socket_receive(socket, buf, 1024, NULL, NULL))) {
 113                 _D("size : %d", size);
 114                 if (!str)
 115                         str = g_string_new(NULL);
 116                 g_string_append_len(str, buf, size);
 117         }
 118
 119         if (str)
 120                 msg = g_string_free(str, FALSE);
 121
 122         if (!msg)
 123                 _E("Empty message");
 124         else {
 125                 char **strv = NULL;
 126                 int i;
 127
 128                 strv = g_strsplit(msg, "|", -1);
 129                 for (i = 0; strv[i]; i++) {
 130                         if (strv[i][0] == '\0') {
 131                                 g_free(strv[i]);
 132                                 continue;
 133                         } else {
 134                                 ttd_app_data *app_data = NULL;
 135                                 _D("strv[%d] : %s", i, strv[i]);
 136                                 app_data = ttd_app_data_new(data->project, strv[i]);
 137                                 if (app_data)
 138                                         g_async_queue_push(data->data_queue, app_data);
 139                                 else
 140                                         _E("failed to creat app_data for %s", strv[i]);
 141                                 g_free(strv[i]);
 142                         }
 143                 }
 144         }
 145         g_free(msg);
 146
 147         return TRUE;
 148 }
 149
 150 static guint
 151 io_add_watch_with_context(GMainContext *context,
 152         GIOChannel *ch,
 153         GIOCondition condition,
 154         GIOFunc func,
 155         gpointer user_data)
 156 {
 157         GSource *source = NULL;
 158         guint id = 0;
 159
 160         source = g_io_create_watch(ch, condition);
 161         g_source_set_priority(source, G_PRIORITY_DEFAULT);
 162         g_source_set_callback(source, (GSourceFunc)func, user_data, NULL);
 163         id = g_source_attach(source, context);
 164         g_source_unref(source);
 165
 166         return id;
 167 }
 168
 169 static gboolean __thread_terminated(gpointer user_data)
 170 {
 171         thread_data *data = user_data;
 172         ttd_app_inf_h handle = (ttd_app_inf_h) data->main_data;
 173
 174         retv_if(!data, FALSE);
 175         retv_if(!handle, FALSE);
 176         retv_if(!data->name, FALSE);
 177
 178         _D("thread for [%s] is terminated", data->name);
 179
 180         g_hash_table_remove(handle->app_sock_hash, data->name);
 181
 182         return FALSE;
 183 }
 184
 185 static gpointer new_socket_thread(gpointer user_data)
 186 {
 187         thread_data *data = user_data;
 188         GError *error = NULL;
 189         GIOChannel *ch = NULL;
 190         int fd = 0;
 191
 192         _D("thread for [%s] created",
 193                 g_unix_socket_address_get_path(G_UNIX_SOCKET_ADDRESS(data->addr)));
 194
 195         g_main_context_push_thread_default(data->context);
 196
 197         if (!g_socket_connect(data->socket, data->addr, NULL, &error)) {
 198                 _E("failed to g_socket_connect - %s", error->message);
 199                 g_error_free(error);
 200                 goto THREAD_EXIT;
 201         }
 202
 203         fd = g_socket_get_fd(data->socket);
 204         ch = g_io_channel_unix_new(fd);
 205         io_add_watch_with_context(data->context, ch, G_IO_IN, __receive_msg, data);
 206         io_add_watch_with_context(data->context, ch,
 207                 (GIOCondition) (G_IO_ERR | G_IO_HUP), __on_disconnect, data);
 208
 209         g_main_loop_run(data->loop);
 210
 211         g_io_channel_shutdown(ch, FALSE, &error);
 212         g_io_channel_unref(ch);
 213
 214 THREAD_EXIT:
 215         g_main_context_pop_thread_default(data->context);
 216
 217         g_socket_shutdown(data->socket, FALSE, TRUE, NULL);
 218         g_socket_close(data->socket, NULL);
 219
 220         if (data->disconnected)
 221                 g_main_context_invoke(NULL, __thread_terminated, data);
 222
 223         return NULL;
 224 }
 225
 226 static void thread_data_set_project(thread_data *data, const char *project)
 227 {
 228         ret_if(!data);
 229         ret_if(!project);
 230
 231         data->project = g_strdup(project);
 232 }
 233
 234 static void thread_data_set_name(thread_data *data, const char *name)
 235 {
 236         ret_if(!data);
 237         ret_if(!name);
 238
 239         data->name = g_strdup(name);
 240 }
 241
 242 static void thread_data_set_queue(thread_data *data, GAsyncQueue *queue)
 243 {
 244         ret_if(!data);
 245         ret_if(!queue);
 246
 247         data->data_queue = queue;
 248 }
 249
 250 static void thread_data_set_main_data(thread_data *data, void *main_data)
 251 {
 252         ret_if(!data);
 253         ret_if(!data);
 254
 255         data->main_data = main_data;
 256 }
 257
 258 static void thread_quit(thread_data *data)
 259 {
 260         gboolean running = FALSE;
 261
 262         running = g_main_loop_is_running(data->loop);
 263         if (running) {
 264                 g_main_loop_quit(data->loop);
 265                 _D("quit thread for [%s]",
 266                         g_unix_socket_address_get_path(G_UNIX_SOCKET_ADDRESS(data->addr)));
 267         } else {
 268                 _D("thread for [%s] is already quitted",
 269                         g_unix_socket_address_get_path(G_UNIX_SOCKET_ADDRESS(data->addr)));
 270         }
 271 }
 272
 273 static void thread_data_free(thread_data *data)
 274 {
 275         g_main_loop_unref(data->loop);
 276         g_main_context_unref(data->context);
 277         g_object_unref(data->socket);
 278         _D("remove data for thread related to [%s]",
 279                         g_unix_socket_address_get_path(G_UNIX_SOCKET_ADDRESS(data->addr)));
 280         g_object_unref(data->addr);
 281         g_free(data->name);
 282         g_free(data);
 283 }
 284
 285 static void thread_quit_and_free(thread_data *data)
 286 {
 287         thread_quit(data);
 288         g_thread_join(data->thread);
 289         thread_data_free(data);
 290 }
 291
 292 static thread_data *
 293 create_thread_for_socket(GSocket *sock, GSocketAddress *addr)
 294 {
 295         thread_data *data = NULL;
 296         GError *error = NULL;
 297
 298         data = g_try_new0(thread_data, 1);
 299         if (!data) {
 300                 _E("failed to get memory for thread_data");
 301                 return NULL;
 302         }
 303         data->disconnected = 0;
 304         data->socket = g_object_ref(sock);
 305         data->addr = g_object_ref(addr);
 306         data->context = g_main_context_new();
 307         data->loop = g_main_loop_new(data->context, FALSE);
 308         data->thread = g_thread_try_new(NULL, new_socket_thread, data, &error);
 309         if (!data->thread) {
 310                 _E("failed to create thread - %s", error->message);
 311                 g_error_free(error);
 312                 thread_data_free(data);
 313                 data = NULL;
 314         }
 315
 316         return data;
 317 }
 318
 319 static int
 320 __get_new_connection_thread(ttd_app_inf_h handle,
 321         const gchar *token, const gchar *appID, guint appPID, const char *project)
 322 {
 323         char *addr_name = NULL;
 324         GSocket *socket = NULL;
 325         GSocketAddress *addr = NULL;
 326         GError *error = NULL;
 327         thread_data *thread_d = NULL;
 328
 329         retv_if(!handle, -1);
 330         retv_if(!token, -1);
 331         retv_if(!appID, -1);
 332         retv_if(!appPID, -1);
 333         retv_if(!project, -1);
 334
 335         addr_name = __get_addr_name(token, appID, appPID);
 336         retv_if(!addr_name, -1);
 337
 338         socket = g_socket_new(G_SOCKET_FAMILY_UNIX,
 339                                 G_SOCKET_TYPE_STREAM,
 340                                 G_SOCKET_PROTOCOL_DEFAULT,
 341                                 &error);
 342         if (!socket) {
 343                 _E("failed to create socket - %s", error->message);
 344                 g_error_free(error);
 345                 goto FREE_N_RETURN_ERROR;
 346         }
 347         g_socket_set_blocking(socket, FALSE);
 348
 349         addr = g_unix_socket_address_new_with_type(
 350                         addr_name, -1, G_UNIX_SOCKET_ADDRESS_ABSTRACT);
 351         if (!addr) {
 352                 _E("failed to create socket address");
 353                 goto FREE_N_RETURN_ERROR;
 354         }
 355
 356         thread_d = create_thread_for_socket(socket, addr);
 357         if (!thread_d) {
 358                 _E("failed to create thread");
 359                 goto FREE_N_RETURN_ERROR;
 360         }
 361         thread_data_set_queue(thread_d, handle->data_queue);
 362         thread_data_set_main_data(thread_d, handle);
 363         thread_data_set_name(thread_d, addr_name);
 364         thread_data_set_project(thread_d, project);
 365         g_hash_table_insert(handle->app_sock_hash, addr_name, thread_d);
 366
 367         g_object_unref(socket);
 368         g_object_unref(addr);
 369
 370         return 0;
 371
 372 FREE_N_RETURN_ERROR:
 373         g_free(addr_name);
 374
 375         if (socket)
 376                 g_object_unref(socket);
 377
 378         if (addr)
 379                 g_object_unref(addr);
 380
 381         if (error)
 382                 g_error_free(error);
 383
 384         return -1;
 385 }
 386
 387 static void __data_queue_item_free(gpointer data)
 388 {
 389         if (!data)
 390                 return;
 391
 392         if (data == queue_quit_marker)
 393                 return;
 394
 395         ttd_app_data_free(data);
 396 }
 397
 398 static void __handle_method_register(GVariant *parameters,
 399         GDBusMethodInvocation *invocation, ttd_app_inf_h handle)
 400 {
 401         const gchar *project = NULL;
 402         const gchar *token = NULL;
 403         const gchar *appID = NULL;
 404         guint appPID = 0;
 405         gint response = 0;
 406         const gchar *response_msg = NULL;
 407
 408         g_variant_get(parameters, "(&s&s&su)", &project, &token, &appID, &appPID);
 409         _D("received register request from [%s-%u] - project[%s], token[%s]",
 410                 appID, appPID, project, token);
 411
 412         if (!appID) {
 413                 response = -1;
 414                 response_msg = "failed to get appID";
 415                 goto METHOD_RETURN;
 416         }
 417
 418         if (!appPID) {
 419                 response = -1;
 420                 response_msg = "failed to get appPID";
 421                 goto METHOD_RETURN;
 422         }
 423
 424         if (!token) {
 425                 response = -1;
 426                 response_msg = "failed to get token";
 427                 goto METHOD_RETURN;
 428         }
 429
 430         if (!project) {
 431                 response = -1;
 432                 response_msg = "failed to get project";
 433                 goto METHOD_RETURN;
 434         }
 435
 436         response =
 437                 __get_new_connection_thread(handle, token, appID, appPID, project);
 438         if (!response)
 439                 response_msg = "OK";
 440         else
 441                 response_msg = "failed to connect";
 442
 443 METHOD_RETURN:
 444         g_dbus_method_invocation_return_value(invocation,
 445                 g_variant_new("(is)", response, response_msg));
 446
 447 }
 448
 449 static void __handle_method_unregister(GVariant *parameters,
 450         GDBusMethodInvocation *invocation, ttd_app_inf_h handle)
 451 {
 452         const gchar *project = NULL;
 453         const gchar *token = NULL;
 454         const gchar *appID = NULL;
 455         guint appPID = 0;
 456         gint response = 0;
 457         const gchar *response_msg = NULL;
 458         char *key = NULL;
 459
 460         g_variant_get(parameters, "(&s&s&su)", &project, &token, &appID, &appPID);
 461         _D("received unregister request from [%s-%u] - project[%s] - token[%s]",
 462                 appID, appPID, project, token);
 463
 464         if (!appID) {
 465                 response = -1;
 466                 response_msg = "failed to get appID";
 467                 goto METHOD_RETURN;
 468         }
 469
 470         if (!appPID) {
 471                 response = -1;
 472                 response_msg = "failed to get appPID";
 473                 goto METHOD_RETURN;
 474         }
 475
 476         if (!token) {
 477                 response = -1;
 478                 response_msg = "failed to get token";
 479                 goto METHOD_RETURN;
 480         }
 481
 482         if (!project) {
 483                 response = -1;
 484                 response_msg = "failed to get project";
 485                 goto METHOD_RETURN;
 486         }
 487
 488         key = __get_addr_name(token, appID, appPID);
 489         g_hash_table_remove(handle->app_sock_hash, key);
 490         g_free(key);
 491
 492         response_msg = "OK";
 493
 494 METHOD_RETURN:
 495         g_dbus_method_invocation_return_value(invocation,
 496                 g_variant_new("(is)", response, response_msg));
 497 }
 498
 499 static void
 500 _ttd_handle_method_call(GDBusConnection *connection,
 501         const gchar *sender,
 502         const gchar *object_path,
 503         const gchar *interface_name,
 504         const gchar *method_name,
 505         GVariant *parameters,
 506         GDBusMethodInvocation *invocation,
 507         gpointer user_data)
 508 {
 509         ttd_app_inf_h handle = user_data;
 510         ret_if(!handle);
 511         ret_if(!parameters);
 512         ret_if(!invocation);
 513
 514         _D("method[%s] called by [%s]", method_name, sender);
 515
 516         if (g_strcmp0(method_name, TTD_APP_INF_METHOD_REG) == 0) {
 517                 __handle_method_register(parameters, invocation, handle);
 518         } else if (g_strcmp0(method_name, TTD_APP_INF_METHOD_UNREG) == 0) {
 519                 __handle_method_unregister(parameters, invocation, handle);
 520         } else {
 521                 _E("Unkwon Method - %s", method_name);
 522                 g_dbus_method_invocation_return_error(invocation, G_DBUS_ERROR,
 523                         G_DBUS_ERROR_UNKNOWN_METHOD, "Unknown Method");
 524         }
 525 }
 526
 527 static const GDBusInterfaceVTable interface_vtable = {
 528         _ttd_handle_method_call, NULL, NULL
 529 };
 530
 531 static void
 532 _ttd_app_inf_on_name_acquired(GDBusConnection *connection,
 533         const gchar *name, gpointer user_data)
 534 {
 535         _D("Aquried the name[%s]", name);
 536 }
 537
 538 static void
 539 _ttd_app_inf_on_name_lost(GDBusConnection *connection,
 540         const gchar *name, gpointer user_data)
 541 {
 542
 543         /* TODO : how to handle this situation? */
 544
 545         if (connection)
 546                 _E("Lost the name[%s]", name);
 547         else
 548                 _E("failed to connect to the bus [%s]", name);
 549 }
 550
 551 static void
 552 _ttd_app_inf_on_bus_acquired(GDBusConnection *connection,
 553         const gchar *name, gpointer user_data)
 554 {
 555         ttd_app_inf_h handle = user_data;
 556         guint r_id = 0; /* TODO : store it or not? */
 557
 558         _D("Acquired a message bus connection - [%s]", name);
 559
 560         r_id = g_dbus_connection_register_object(connection,
 561                         TTD_APP_INF_OBJECT_PATH,
 562                         handle->introspection_data->interfaces[0],
 563                         &interface_vtable,
 564                         handle, /* user_data */
 565                         NULL,   /* user_data_free_func */
 566                         NULL);  /* GError** */
 567
 568         if (r_id == 0)
 569                 _E("failed to g_dbus_connection_register_object()");
 570 }
 571
 572 static gpointer post_thread(gpointer user_data)
 573 {
 574         ttd_app_inf_h handle = user_data;
 575
 576         retv_if(!handle, NULL);
 577
 578         while (TRUE) {
 579                 ttd_app_data *app_data = NULL;
 580                 int ret = 0;
 581                 long res_code = 0;
 582
 583                 app_data = g_async_queue_pop(handle->data_queue);
 584                 if (!app_data)
 585                         continue;
 586
 587                 if (app_data == queue_quit_marker) {
 588                         _D("receiving queue_quit_marker");
 589                         break;
 590                 }
 591
 592                 ret = ttd_http_appdata_post(ttd_app_data_get_project_name(app_data),
 593                         ttd_app_data_get_data(app_data), &res_code);
 594                 if (!ret) {
 595                         _E("failed to post data, retry it - code[%ld]", res_code);
 596                         g_async_queue_push_front(handle->data_queue, app_data);
 597                         g_usleep((gulong)(G_USEC_PER_SEC/2));
 598                         /* 0.5 sec sleep before retry it */
 599                         continue;
 600                 }
 601
 602                 _D("msg posted : %s", ttd_app_data_get_data(app_data));
 603                 ttd_app_data_free(app_data);
 604         }
 605         return NULL;
 606 }
 607
 608 int ttd_app_interface_fini(ttd_app_inf_h handle)
 609 {
 610         retv_if(!handle, -1);
 611
 612         if (handle->gdbus_id > 0)
 613                 g_bus_unown_name(handle->gdbus_id);
 614
 615         if (handle->introspection_data)
 616                 g_dbus_node_info_unref(handle->introspection_data);
 617
 618         if (handle->app_sock_hash) {
 619                 g_hash_table_remove_all(handle->app_sock_hash);
 620                 g_hash_table_unref(handle->app_sock_hash);
 621         }
 622
 623         if (handle->data_queue) {
 624                 g_async_queue_push_front(handle->data_queue, queue_quit_marker);
 625                 g_async_queue_ref(handle->data_queue);
 626         }
 627
 628         if (handle->post_thread)
 629                 g_thread_join(handle->post_thread);
 630
 631         g_free(handle);
 632
 633         return 0;
 634 }
 635
 636 int ttd_app_interface_init(ttd_app_inf_h *handle)
 637 {
 638         ttd_app_inf_h _handle = NULL;
 639         GError *error = NULL;
 640         guint id = 0;
 641         GDBusNodeInfo *introspection_data = NULL;
 642
 643         retv_if(!handle, -1);
 644
 645         _handle = g_try_malloc0(sizeof(struct _ttd_app_inf_h));
 646         retv_if(!_handle, -1);
 647
 648         introspection_data =
 649                 g_dbus_node_info_new_for_xml(APP_INF_INTROSPECTION, NULL);
 650
 651         id = g_bus_own_name(G_BUS_TYPE_SYSTEM,
 652                         TTD_APP_INF_BUS_NAME,
 653                         (GBusNameOwnerFlags)(G_BUS_NAME_OWNER_FLAGS_ALLOW_REPLACEMENT
 654                         | G_BUS_NAME_OWNER_FLAGS_REPLACE),
 655                         _ttd_app_inf_on_bus_acquired,
 656                         _ttd_app_inf_on_name_acquired,
 657                         _ttd_app_inf_on_name_lost,
 658                         _handle,
 659                         NULL);
 660         if (id == 0) {
 661                 _E("failed to g_bus_own_name()");
 662                 g_dbus_node_info_unref(introspection_data);
 663                 g_free(_handle);
 664                 return -1;
 665         }
 666
 667         _handle->gdbus_id = id;
 668         _handle->introspection_data = introspection_data;
 669         _handle->app_sock_hash =
 670                 g_hash_table_new_full(g_str_hash, g_str_equal,
 671                         g_free, (GDestroyNotify)thread_quit_and_free);
 672
 673         _handle->data_queue = g_async_queue_new_full(__data_queue_item_free);
 674         _handle->post_thread =
 675                 g_thread_try_new(NULL, (GThreadFunc)post_thread, _handle, &error);
 676         if (!_handle->post_thread) {
 677                 _E("failed to create post thread - %s", error->message);
 678                 g_error_free(error);
 679                 ttd_app_interface_fini(_handle);
 680                 return -1;
 681         }
 682
 683         *handle = _handle;
 684
 685         return 0;
 686 }