src/afb-svc.c

   1 /*
   2  * Copyright (C) 2016, 2017 "IoT.bzh"
   3  * Author: José Bollo <jose.bollo@iot.bzh>
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *   http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  */
  17
  18 #define _GNU_SOURCE
  19
  20 #include <stdlib.h>
  21 #include <string.h>
  22 #include <errno.h>
  23
  24 #include <json-c/json.h>
  25
  26 #include <afb/afb-binding-v1.h>
  27 #include <afb/afb-binding-v2.h>
  28
  29 #include "afb-session.h"
  30 #include "afb-context.h"
  31 #include "afb-evt.h"
  32 #include "afb-msg-json.h"
  33 #include "afb-svc.h"
  34 #include "afb-xreq.h"
  35 #include "afb-cred.h"
  36 #include "afb-apiset.h"
  37 #include "afb-hook.h"
  38 #include "jobs.h"
  39 #include "verbose.h"
  40
  41
  42 #define HOOK(x,...) if((svc)->hookflags & afb_hook_flag_svc_##x) afb_hook_svc_##x(__VA_ARGS__)
  43
  44 /*
  45  * Structure for requests initiated by the service
  46  */
  47 struct svc_req
  48 {
  49         struct afb_xreq xreq;
  50
  51         struct afb_svc *svc;
  52
  53         /* the args */
  54         void (*callback)(void*, int, struct json_object*);
  55         void *closure;
  56
  57         /* sync */
  58         struct jobloop *jobloop;
  59         struct json_object *result;
  60         int status;
  61         int async;
  62 };
  63
  64 /* functions for services */
  65 static void svc_on_event(void *closure, const char *event, int eventid, struct json_object *object);
  66 static void svc_call(void *closure, const char *api, const char *verb, struct json_object *args,
  67                                 void (*callback)(void*, int, struct json_object*), void *cbclosure);
  68 static int svc_call_sync(void *closure, const char *api, const char *verb, struct json_object *args,
  69                                 struct json_object **result);
  70
  71 /* the interface for services */
  72 static const struct afb_service_itf service_itf = {
  73         .call = svc_call,
  74         .call_sync = svc_call_sync
  75 };
  76
  77 /* the interface for events */
  78 static const struct afb_evt_itf evt_itf = {
  79         .broadcast = svc_on_event,
  80         .push = svc_on_event
  81 };
  82
  83 /* functions for requests of services */
  84 static void svcreq_destroy(struct afb_xreq *xreq);
  85 static void svcreq_reply(struct afb_xreq *xreq, int status, json_object *obj);
  86
  87 /* interface for requests of services */
  88 const struct afb_xreq_query_itf afb_svc_xreq_itf = {
  89         .unref = svcreq_destroy,
  90         .reply = svcreq_reply
  91 };
  92
  93 /* the common session for services sharing their session */
  94 static struct afb_session *common_session;
  95
  96 static inline struct afb_service to_afb_service(struct afb_svc *svc)
  97 {
  98         return (struct afb_service){ .itf = &service_itf, .closure = svc };
  99 }
 100
 101 /*
 102  * Frees a service
 103  */
 104 void afb_svc_destroy(struct afb_svc *svc, struct afb_service *service)
 105 {
 106         if (service)
 107                 *service = (struct afb_service){ .itf = NULL, .closure = NULL };
 108         if (svc) {
 109                 if (svc->listener != NULL)
 110                         afb_evt_listener_unref(svc->listener);
 111                 if (svc->session)
 112                         afb_session_unref(svc->session);
 113                 afb_apiset_unref(svc->apiset);
 114                 free(svc);
 115         }
 116 }
 117
 118 /*
 119  * Creates a new service
 120  */
 121 struct afb_svc *afb_svc_create(
 122                         const char *api,
 123                         struct afb_apiset *apiset,
 124                         int share_session,
 125                         void (*on_event)(const char *event, struct json_object *object),
 126                         struct afb_service *service
 127 )
 128 {
 129         struct afb_svc *svc;
 130
 131         /* allocates the svc handler */
 132         svc = calloc(1, sizeof * svc);
 133         if (svc == NULL) {
 134                 errno = ENOMEM;
 135                 return NULL;
 136         }
 137
 138         /* instanciate the apiset */
 139         svc->api = api;
 140         svc->apiset = afb_apiset_addref(apiset);
 141
 142         /* instanciate the session */
 143         if (share_session) {
 144                 /* session shared with other svcs */
 145                 if (common_session == NULL) {
 146                         common_session = afb_session_create (NULL, 0);
 147                         if (common_session == NULL)
 148                                 goto error;
 149                 }
 150                 svc->session = afb_session_addref(common_session);
 151         } else {
 152                 /* session dedicated to the svc */
 153                 svc->session = afb_session_create (NULL, 0);
 154                 if (svc->session == NULL)
 155                         goto error;
 156         }
 157
 158         svc->hookflags = afb_hook_flags_svc(svc->api);
 159         if (service)
 160                 *service = to_afb_service(svc);
 161
 162         /* initialises the listener if needed */
 163         if (on_event) {
 164                 svc->on_event = on_event;
 165                 svc->listener = afb_evt_listener_create(&evt_itf, svc);
 166                 if (svc->listener == NULL)
 167                         goto error;
 168         }
 169
 170         return svc;
 171
 172 error:
 173         afb_svc_destroy(svc, service);
 174         return NULL;
 175 }
 176
 177 /*
 178  * Starts a new service (v1)
 179  */
 180 int afb_svc_start_v1(struct afb_svc *svc, int (*start)(struct afb_service))
 181 {
 182         int rc;
 183
 184         HOOK(start_before, svc);
 185         rc = start(to_afb_service(svc));
 186         HOOK(start_after, svc, rc);
 187         return rc;
 188 }
 189
 190 /*
 191  * Starts a new service (v2)
 192  */
 193 int afb_svc_start_v2(struct afb_svc *svc, int (*start)())
 194 {
 195         int rc;
 196
 197         HOOK(start_before, svc);
 198         rc = start();
 199         HOOK(start_after, svc, rc);
 200         return rc;
 201 }
 202
 203 /*
 204  * Request to updates the hooks
 205  */
 206 void afb_svc_update_hook(struct afb_svc *svc)
 207 {
 208         svc->hookflags = afb_hook_flags_svc(svc->api);
 209 }
 210
 211 /*
 212  * Propagates the event to the service
 213  */
 214 static void svc_on_event(void *closure, const char *event, int eventid, struct json_object *object)
 215 {
 216         struct afb_svc *svc = closure;
 217
 218         HOOK(on_event_before, svc, event, eventid, object);
 219         svc->on_event(event, object);
 220         HOOK(on_event_after, svc, event, eventid, object);
 221         json_object_put(object);
 222 }
 223
 224 /*
 225  * create an svc_req
 226  */
 227 static struct svc_req *svcreq_create(struct afb_svc *svc, const char *api, const char *verb, struct json_object *args)
 228 {
 229         struct svc_req *svcreq;
 230         size_t lenapi, lenverb;
 231         char *copy;
 232
 233         /* allocates the request */
 234         lenapi = 1 + strlen(api);
 235         lenverb = 1 + strlen(verb);
 236         svcreq = malloc(lenapi + lenverb + sizeof *svcreq);
 237         if (svcreq != NULL) {
 238                 /* initialises the request */
 239                 afb_xreq_init(&svcreq->xreq, &afb_svc_xreq_itf);
 240                 afb_context_init(&svcreq->xreq.context, svc->session, NULL);
 241                 svcreq->xreq.context.validated = 1;
 242                 copy = (char*)&svcreq[1];
 243                 memcpy(copy, api, lenapi);
 244                 svcreq->xreq.api = copy;
 245                 copy = &copy[lenapi];
 246                 memcpy(copy, verb, lenverb);
 247                 svcreq->xreq.verb = copy;
 248                 svcreq->xreq.listener = svc->listener;
 249                 svcreq->xreq.json = args;
 250                 svcreq->svc = svc;
 251         }
 252         return svcreq;
 253 }
 254
 255 /*
 256  * destroys the svc_req
 257  */
 258 static void svcreq_destroy(struct afb_xreq *xreq)
 259 {
 260         struct svc_req *svcreq = CONTAINER_OF_XREQ(struct svc_req, xreq);
 261
 262         afb_context_disconnect(&svcreq->xreq.context);
 263         json_object_put(svcreq->xreq.json);
 264         afb_cred_unref(svcreq->xreq.cred);
 265         free(svcreq);
 266 }
 267
 268 static void svcreq_sync_leave(struct svc_req *svcreq)
 269 {
 270         struct jobloop *jobloop = svcreq->jobloop;
 271
 272         if (jobloop) {
 273                 svcreq->jobloop = NULL;
 274                 jobs_leave(jobloop);
 275         }
 276 }
 277
 278 static void svcreq_reply(struct afb_xreq *xreq, int status, json_object *obj)
 279 {
 280         struct svc_req *svcreq = CONTAINER_OF_XREQ(struct svc_req, xreq);
 281         if (svcreq->async) {
 282                 struct afb_svc *svc = svcreq->svc;
 283                 if (svcreq->callback)
 284                         svcreq->callback(svcreq->closure, status, obj);
 285                 HOOK(call_result, svc, status, obj);
 286                 json_object_put(obj);
 287         } else {
 288                 svcreq->status = status;
 289                 svcreq->result = obj;
 290                 svcreq_sync_leave(svcreq);
 291         }
 292 }
 293
 294 static void svcreq_sync_enter(int signum, void *closure, struct jobloop *jobloop)
 295 {
 296         struct svc_req *svcreq = closure;
 297
 298         if (!signum) {
 299                 svcreq->jobloop = jobloop;
 300                 afb_xreq_process(&svcreq->xreq, svcreq->svc->apiset);
 301         } else {
 302                 svcreq->result = afb_msg_json_internal_error();
 303                 svcreq->status = -1;
 304                 svcreq_sync_leave(svcreq);
 305         }
 306 }
 307
 308 /*
 309  * Initiates a call for the service
 310  */
 311 static void svc_call(void *closure, const char *api, const char *verb, struct json_object *args, void (*callback)(void*, int, struct json_object*), void *cbclosure)
 312 {
 313         struct afb_svc *svc = closure;
 314         struct svc_req *svcreq;
 315         struct json_object *ierr;
 316
 317         HOOK(call, svc, api, verb, args);
 318
 319         /* allocates the request */
 320         svcreq = svcreq_create(svc, api, verb, args);
 321         if (svcreq == NULL) {
 322                 ERROR("out of memory");
 323                 json_object_put(args);
 324                 ierr = afb_msg_json_internal_error();
 325                 if (callback)
 326                         callback(cbclosure, -1, ierr);
 327                 HOOK(call_result, svc, -1, ierr);
 328                 json_object_put(ierr);
 329                 return;
 330         }
 331
 332         /* initialises the request */
 333         svcreq->jobloop = NULL;
 334         svcreq->callback = callback;
 335         svcreq->closure = cbclosure;
 336         svcreq->async = 1;
 337
 338         /* terminates and frees ressources if needed */
 339         afb_xreq_process(&svcreq->xreq, svc->apiset);
 340 }
 341
 342 static int svc_call_sync(void *closure, const char *api, const char *verb, struct json_object *args,
 343                                 struct json_object **result)
 344 {
 345         struct afb_svc *svc = closure;
 346         struct svc_req *svcreq;
 347         struct json_object *resu;
 348         int rc;
 349
 350         HOOK(callsync, svc, api, verb, args);
 351
 352         /* allocates the request */
 353         svcreq = svcreq_create(svc, api, verb, args);
 354         if (svcreq == NULL) {
 355                 ERROR("out of memory");
 356                 errno = ENOMEM;
 357                 json_object_put(args);
 358                 resu = afb_msg_json_internal_error();
 359                 rc = -1;
 360         } else {
 361                 /* initialises the request */
 362                 svcreq->jobloop = NULL;
 363                 svcreq->callback = NULL;
 364                 svcreq->result = NULL;
 365                 svcreq->status = 0;
 366                 svcreq->async = 0;
 367                 afb_xreq_addref(&svcreq->xreq);
 368                 rc = jobs_enter(NULL, 0, svcreq_sync_enter, svcreq);
 369                 if (rc >= 0)
 370                         rc = svcreq->status;
 371                 resu = (rc >= 0 || svcreq->result) ? svcreq->result : afb_msg_json_internal_error();
 372                 afb_xreq_unref(&svcreq->xreq);
 373         }
 374         HOOK(callsync_result, svc, rc, resu);
 375         if (result)
 376                 *result = resu;
 377         else
 378                 json_object_put(resu);
 379         return rc;
 380 }
 381