src/plugins/client-api/native/native-server.c

   1 /*
   2  * Copyright (c) 2012, 2013, Intel Corporation
   3  *
   4  * Redistribution and use in source and binary forms, with or without
   5  * modification, are permitted provided that the following conditions are
   6  * met:
   7  *
   8  *   * Redistributions of source code must retain the above copyright notice,
   9  *     this list of conditions and the following disclaimer.
  10  *   * Redistributions in binary form must reproduce the above copyright
  11  *     notice, this list of conditions and the following disclaimer in the
  12  *     documentation and/or other materials provided with the distribution.
  13  *   * Neither the name of Intel Corporation nor the names of its contributors
  14  *     may be used to endorse or promote products derived from this software
  15  *     without specific prior written permission.
  16  *
  17  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  18  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  19  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  20  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  21  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  22  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  23  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  24  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  25  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  26  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  27  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  28  */
  29
  30 #include <stdlib.h>
  31 #include <errno.h>
  32
  33 #include <murphy/common/debug.h>
  34 #include <murphy/common/log.h>
  35 #include <murphy/common/mm.h>
  36 #include <murphy/common/list.h>
  37 #include <murphy/common/transport.h>
  38 #include <murphy/common/native-types.h>
  39
  40 #include "srs/daemon/plugin.h"
  41 #include "srs/daemon/client.h"
  42
  43 #include "native-messages.h"
  44 #include "native-config.h"
  45
  46 #define PLUGIN_NAME    "native-client"
  47 #define PLUGIN_DESCR   "Native client plugin for SRS."
  48 #define PLUGIN_AUTHORS "Krisztian Litkey <kli@iki.fi>"
  49 #define PLUGIN_VERSION "0.0.1"
  50
  51
  52 /*
  53  * server runtime context
  54  */
  55
  56 typedef struct {
  57     srs_plugin_t    *self;               /* our plugin instance */
  58     const char      *address;            /* our transport address */
  59     int              sock;               /* or existing transport socket */
  60     mrp_transport_t *lt;                 /* transport we listen on */
  61     mrp_list_hook_t  clients;            /* connected clients */
  62     int              next_id;            /* next client id */
  63 } server_t;
  64
  65
  66 /*
  67  * a connected client
  68  */
  69
  70 typedef struct {
  71     srs_client_t    *c;                  /* associated SRS client */
  72     server_t        *s;                  /* server runtime context */
  73     mrp_transport_t *t;                  /* transport towards this client */
  74     mrp_list_hook_t  hook;               /* to list of native clients */
  75     int              id;                 /* client id */
  76 } client_t;
  77
  78
  79 static int focus_notify(srs_client_t *c, srs_voice_focus_t focus);
  80 static int command_notify(srs_client_t *c, int idx, int ntoken, char **tokens,
  81                           uint32_t *start, uint32_t *end,
  82                           srs_audiobuf_t *audio);
  83 static int voice_notify(srs_client_t *c, srs_voice_event_t *event);
  84
  85 static int reply_status(client_t *c, uint32_t reqno, int status,
  86                         const char *msg);
  87 #define reply_register   reply_status
  88 #define reply_unregister reply_status
  89 #define reply_focus      reply_status
  90 static int reply_render(client_t *c, uint32_t reqno, uint32_t id);
  91 static int reply_voiceqry(client_t *c, uint32_t reqno,
  92                           srs_voice_actor_t *actors, int nactor);
  93
  94 static client_t *create_client(server_t *s, mrp_transport_t *lt)
  95 {
  96     mrp_transport_t *t;
  97     client_t        *c;
  98
  99     c = mrp_allocz(sizeof(*c));
 100
 101     if (c != NULL) {
 102         mrp_list_init(&c->hook);
 103
 104         c->s  = s;
 105         c->id = s->next_id++;
 106         c->t  = mrp_transport_accept(lt, c, MRP_TRANSPORT_REUSEADDR);
 107
 108         if (c->t != NULL) {
 109             mrp_list_append(&s->clients, &c->hook);
 110
 111             return c;
 112         }
 113
 114         mrp_free(c);
 115     }
 116     else {
 117         t = mrp_transport_accept(lt, NULL, MRP_TRANSPORT_REUSEADDR);
 118         mrp_transport_destroy(t);
 119     }
 120
 121     return NULL;
 122 }
 123
 124
 125 static void destroy_client(client_t *c)
 126 {
 127     mrp_list_delete(&c->hook);
 128
 129     mrp_transport_destroy(c->t);
 130     client_destroy(c->c);
 131
 132     mrp_free(c);
 133 }
 134
 135
 136 static void register_client(client_t *c, srs_req_register_t *req)
 137 {
 138     static srs_client_ops_t ops = {
 139         .notify_focus   = focus_notify,
 140         .notify_command = command_notify,
 141         .notify_render  = voice_notify,
 142     };
 143
 144     srs_context_t  *srs    = c->s->self->srs;
 145     char           *name   = req->name;
 146     char           *appcls = req->appclass;
 147     char          **cmds   = req->commands;
 148     int             ncmd   = req->ncommand;
 149     char            id[64];
 150
 151     snprintf(id, sizeof(id), "native-client-%d", c->id);
 152
 153     mrp_debug("received register request from native client #%d", c->id);
 154
 155     c->c = client_create(srs, SRS_CLIENT_TYPE_EXTERNAL, name, appcls,
 156                          cmds, ncmd, id, &ops, c);
 157
 158     if (c->c != NULL)
 159         reply_register(c, req->reqno, SRS_STATUS_OK, "OK");
 160     else {
 161         reply_register(c, req->reqno, SRS_STATUS_FAILED, "failed");
 162         destroy_client(c);
 163     }
 164 }
 165
 166
 167 static void unregister_client(client_t *c, srs_req_unregister_t *req)
 168 {
 169     mrp_debug("received unregister request from native client #%d", c->id);
 170
 171     reply_unregister(c, req->reqno, SRS_STATUS_OK, "OK");
 172     destroy_client(c);
 173 }
 174
 175
 176 static void request_focus(client_t *c, srs_req_focus_t *req)
 177 {
 178     mrp_debug("received focus request from native client #%d", c->id);
 179
 180     if (client_request_focus(c->c, req->focus))
 181         reply_focus(c, req->reqno, SRS_STATUS_OK, "OK");
 182     else
 183         reply_focus(c, req->reqno, SRS_STATUS_FAILED, "failed");
 184 }
 185
 186
 187 static void request_voice(client_t *c, srs_req_voice_t *req)
 188 {
 189     const char *msg     = req->msg;
 190     const char *voice   = req->voice;
 191     double      rate    = req->rate;
 192     double      pitch   = req->pitch;
 193     int         timeout = req->timeout;
 194     int         events  = req->events;
 195     uint32_t    reqid;
 196
 197     mrp_debug("received voice render request from native client #%d", c->id);
 198
 199     reqid = client_render_voice(c->c, msg, voice, rate, pitch, timeout, events);
 200
 201     if (reqid != SRS_VOICE_INVALID)
 202         reply_render(c, req->reqno, reqid);
 203     else
 204         reply_status(c, req->reqno, SRS_STATUS_FAILED, "failed");
 205 }
 206
 207
 208 static void cancel_voice(client_t *c, srs_ccl_voice_t *req)
 209 {
 210     mrp_debug("received voice cancel request from native client #%d", c->id);
 211
 212     client_cancel_voice(c->c, req->id);
 213     reply_status(c, req->reqno, SRS_STATUS_OK, "OK");
 214 }
 215
 216
 217 static void query_voices(client_t *c, srs_req_voiceqry_t *req)
 218 {
 219     srs_voice_actor_t  *actors;
 220     int                 nactor;
 221
 222     mrp_debug("received voice query request from native client #%d", c->id);
 223
 224     nactor = client_query_voices(c->c, req->lang, &actors);
 225     reply_voiceqry(c, req->reqno, actors, nactor);
 226     client_free_queried_voices(actors);
 227 }
 228
 229
 230 static int reply_status(client_t *c, uint32_t reqno, int status,
 231                         const char *msg)
 232 {
 233     srs_rpl_status_t rpl;
 234
 235     mrp_debug("replying <%d, %s> to request #%d from native client #%d",
 236               status, msg, reqno, c->id);
 237
 238     rpl.type   = SRS_REPLY_STATUS;
 239     rpl.reqno  = reqno;
 240     rpl.status = status;
 241     rpl.msg    = (char *)msg;
 242
 243     return send_message(c->t, (srs_msg_t *)&rpl);
 244 }
 245
 246
 247 static int reply_render(client_t *c, uint32_t reqno, uint32_t id)
 248 {
 249     srs_rpl_voice_t rpl;
 250
 251     mrp_debug("replying <#%u> to request #%d from native client #%d", id,
 252               reqno, c->id);
 253
 254     rpl.type  = SRS_REPLY_RENDERVOICE;
 255     rpl.reqno = reqno;
 256     rpl.id    = id;
 257
 258     return send_message(c->t, (srs_msg_t *)&rpl);
 259 }
 260
 261
 262 static int focus_notify(srs_client_t *client, srs_voice_focus_t focus)
 263 {
 264     client_t        *c = (client_t *)client->user_data;
 265     srs_evt_focus_t  evt;
 266
 267     mrp_debug("relaying focus event to native client #%d", c->id);
 268
 269     evt.type  = SRS_EVENT_FOCUS;
 270     evt.focus = focus;
 271
 272     return send_message(c->t, (srs_msg_t *)&evt);
 273 }
 274
 275
 276 static int command_notify(srs_client_t *client, int idx,
 277                           int ntoken, char **tokens, uint32_t *start,
 278                           uint32_t *end, srs_audiobuf_t *audio)
 279 {
 280     client_t          *c = (client_t *)client->user_data;
 281     srs_evt_command_t  evt;
 282
 283     MRP_UNUSED(start);
 284     MRP_UNUSED(end);
 285     MRP_UNUSED(audio);
 286
 287     mrp_debug("relaying command event to native client #%d", c->id);
 288
 289     evt.type   = SRS_EVENT_COMMAND;
 290     evt.idx    = idx;
 291     evt.tokens = tokens;
 292     evt.ntoken = ntoken;
 293
 294     return send_message(c->t, (srs_msg_t *)&evt);
 295 }
 296
 297
 298 static int voice_notify(srs_client_t *client, srs_voice_event_t *event)
 299 {
 300     client_t        *c = (client_t *)client->user_data;
 301     srs_evt_voice_t  evt;
 302
 303     mrp_debug("relaying voice event to native client #%d", c->id);
 304
 305     evt.type  = SRS_EVENT_VOICE;
 306     evt.event = event->type;
 307     evt.id    = event->id;
 308
 309     if (event->type == SRS_VOICE_EVENT_PROGRESS) {
 310         evt.pcnt = event->data.progress.pcnt;
 311         evt.msec = event->data.progress.msec;
 312     }
 313     else {
 314         evt.pcnt = 0;
 315         evt.msec = 0;
 316     }
 317
 318     return send_message(c->t, (srs_msg_t *)&evt);
 319 }
 320
 321
 322 static int reply_voiceqry(client_t *c, uint32_t reqno,
 323                           srs_voice_actor_t *actors, int nactor)
 324 {
 325     srs_rpl_voiceqry_t rpl;
 326
 327     mrp_debug("replying to request #%u from native client #%d", reqno, c->id);
 328
 329     if (actors < 0)
 330         actors = 0;
 331
 332     rpl.type   = SRS_REPLY_QUERYVOICES;
 333     rpl.reqno  = reqno;
 334     rpl.actors = actors;
 335     rpl.nactor = nactor;
 336
 337     return send_message(c->t, (srs_msg_t *)&rpl);
 338 }
 339
 340
 341 static inline void dump_message(void *data, uint32_t type_id)
 342 {
 343     char buf[1024];
 344
 345     if (mrp_print_native(buf, sizeof(buf), data, type_id) > 0)
 346         mrp_debug("got message of type 0x%x: %s", type_id, buf);
 347 }
 348
 349
 350 static void connection_evt(mrp_transport_t *lt, void *user_data)
 351 {
 352     server_t *s = (server_t *)user_data;
 353     client_t *c;
 354
 355     c = create_client(s, lt);
 356
 357     if (c != NULL)
 358         mrp_log_info("Accepted new native client connection.");
 359     else
 360         mrp_log_error("Failed to accept new native client connection.");
 361 }
 362
 363
 364 static void closed_evt(mrp_transport_t *t, int error, void *user_data)
 365 {
 366     client_t *c = (client_t *)user_data;
 367
 368     MRP_UNUSED(t);
 369
 370     if (error != 0)
 371         mrp_log_error("Native client connection closed with error %d (%s).",
 372                       error, strerror(error));
 373     else
 374         mrp_log_info("Native client connection closed.");
 375
 376     destroy_client(c);
 377 }
 378
 379
 380 static void recv_evt(mrp_transport_t *t, void *data, uint32_t type_id,
 381                      void *user_data)
 382 {
 383     client_t  *c   = (client_t *)user_data;
 384     srs_msg_t *req = (srs_msg_t *)data;
 385
 386     MRP_UNUSED(t);
 387
 388     dump_message(data, type_id);
 389
 390     switch (req->type) {
 391     case SRS_REQUEST_REGISTER:
 392         register_client(c, &req->reg_req);
 393         break;
 394
 395     case SRS_REQUEST_UNREGISTER:
 396         unregister_client(c, &req->bye_req);
 397         break;
 398
 399     case SRS_REQUEST_FOCUS:
 400         request_focus(c, &req->focus_req);
 401         break;
 402
 403     case SRS_REQUEST_RENDERVOICE:
 404         request_voice(c, &req->voice_req);
 405         break;
 406
 407     case SRS_REQUEST_CANCELVOICE:
 408         cancel_voice(c, &req->voice_ccl);
 409         break;
 410
 411     case SRS_REQUEST_QUERYVOICES:
 412         query_voices(c, &req->voice_qry);
 413         break;
 414
 415     default:
 416         break;
 417     }
 418
 419 }
 420
 421
 422 static int transport_setup(server_t *s)
 423 {
 424     static mrp_transport_evt_t evt = {
 425         { .recvnative     = recv_evt },
 426         { .recvnativefrom = NULL     },
 427         .connection       = connection_evt,
 428         .closed           = closed_evt,
 429     };
 430
 431     srs_context_t  *srs  = s->self->srs;
 432     mrp_sockaddr_t  addr;
 433     socklen_t       alen;
 434     const char     *type, *opt, *val;
 435     int             flags, state, sock;
 436     void           *typemap;
 437
 438     alen = mrp_transport_resolve(NULL, s->address, &addr, sizeof(addr), &type);
 439
 440     if (alen < 0) {
 441         mrp_log_error("Failed to resolve transport address '%s'.",
 442                       s->address);
 443         goto fail;
 444     }
 445
 446     flags = MRP_TRANSPORT_REUSEADDR | MRP_TRANSPORT_NONBLOCK |  \
 447         MRP_TRANSPORT_MODE_NATIVE;
 448
 449     if (s->sock < 0)
 450         s->lt = mrp_transport_create(srs->ml, type, &evt, s, flags);
 451     else {
 452         state = MRP_TRANSPORT_LISTENED;
 453         sock  = s->sock;
 454         s->lt = mrp_transport_create_from(srs->ml, type, &sock, &evt,
 455                                           s, flags, state);
 456     }
 457
 458     if (s->lt == NULL) {
 459         mrp_log_error("Failed to create transport for native clients.");
 460         return FALSE;
 461     }
 462
 463     if ((typemap = register_message_types()) == NULL) {
 464         mrp_log_error("Failed to register native messages.");
 465         goto fail;
 466     }
 467
 468     if (!mrp_transport_setopt(s->lt, "type-map", typemap)) {
 469         mrp_log_error("Failed to set transport type map.");
 470         goto fail;
 471     }
 472
 473     if (s->sock < 0) {
 474         if (mrp_transport_bind(s->lt, &addr, alen) &&
 475             mrp_transport_listen(s->lt, 0)) {
 476             mrp_log_info("Listening on transport '%s'...", s->address);
 477
 478             return TRUE;
 479         }
 480         else
 481             mrp_log_error("Failed to bind/listen transport.");
 482     }
 483     else {
 484         mrp_log_info("Using passed in socket fd %d...", s->sock);
 485
 486         return TRUE;
 487     }
 488
 489  fail:
 490     if (s->lt) {
 491         mrp_transport_destroy(s->lt);
 492         s->lt = NULL;
 493     }
 494
 495     return FALSE;
 496 }
 497
 498
 499 static void transport_cleanup(server_t *s)
 500 {
 501     mrp_transport_destroy(s->lt);
 502     s->lt = NULL;
 503 }
 504
 505
 506 static int create_native(srs_plugin_t *plugin)
 507 {
 508     server_t *s;
 509
 510     mrp_debug("creating native client interface plugin");
 511
 512     if ((s = mrp_allocz(sizeof(*s))) == NULL)
 513         return FALSE;
 514
 515     mrp_list_init(&s->clients);
 516     s->self = plugin;
 517
 518     plugin->plugin_data = s;
 519
 520     return TRUE;
 521
 522  fail:
 523     mrp_free(s);
 524
 525     return FALSE;
 526 }
 527
 528
 529 static int config_native(srs_plugin_t *plugin, srs_cfg_t *cfg)
 530 {
 531     server_t *s = (server_t *)plugin->plugin_data;
 532
 533     mrp_debug("configure native client interface plugin");
 534
 535     s->address = srs_config_get_string(cfg, CONFIG_ADDRESS, DEFAULT_ADDRESS);
 536     s->sock    = srs_config_get_int32(cfg, CONFIG_SOCKET , DEFAULT_SOCKET);
 537
 538     if (s->sock < 0)
 539         mrp_log_info("Using native client transport: '%s'.", s->address);
 540     else
 541         mrp_log_info("Using native client socket: %d.", s->sock);
 542
 543     return TRUE;
 544 }
 545
 546
 547 static int start_native(srs_plugin_t *plugin)
 548 {
 549     server_t *s = (server_t *)plugin->plugin_data;
 550
 551     MRP_UNUSED(plugin);
 552
 553     return transport_setup(s);
 554 }
 555
 556
 557 static void stop_native(srs_plugin_t *plugin)
 558 {
 559     MRP_UNUSED(plugin);
 560 }
 561
 562
 563 static void destroy_native(srs_plugin_t *plugin)
 564 {
 565     server_t *s = (server_t *)plugin->plugin_data;
 566
 567     transport_cleanup(s);
 568     mrp_free(s);
 569 }
 570
 571
 572 SRS_DECLARE_PLUGIN(PLUGIN_NAME, PLUGIN_DESCR, PLUGIN_AUTHORS, PLUGIN_VERSION,
 573                    create_native, config_native, start_native, stop_native,
 574                    destroy_native)