src/afb-xreq.c

   1 /*
   2  * Copyright (C) 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 #define NO_BINDING_VERBOSE_MACRO
  20
  21 #include <stdlib.h>
  22 #include <string.h>
  23 #include <errno.h>
  24
  25 #include <json-c/json.h>
  26 #include <afb/afb-binding.h>
  27
  28 #include "afb-context.h"
  29 #include "afb-xreq.h"
  30 #include "afb-evt.h"
  31 #include "afb-msg-json.h"
  32 #include "afb-subcall.h"
  33 #include "jobs.h"
  34 #include "verbose.h"
  35
  36
  37 static struct json_object *xreq_json_cb(void *closure);
  38 static struct afb_arg xreq_get_cb(void *closure, const char *name);
  39
  40 static void xreq_success_cb(void *closure, struct json_object *obj, const char *info);
  41 static void xreq_fail_cb(void *closure, const char *status, const char *info);
  42
  43 static const char *xreq_raw_cb(void *closure, size_t *size);
  44 static void xreq_send_cb(void *closure, const char *buffer, size_t size);
  45
  46 static void *xreq_context_get_cb(void *closure);
  47 static void xreq_context_set_cb(void *closure, void *value, void (*free_value)(void*));
  48
  49 static void xreq_addref_cb(void *closure);
  50 static void xreq_unref_cb(void *closure);
  51
  52 static void xreq_session_close_cb(void *closure);
  53 static int xreq_session_set_LOA_cb(void *closure, unsigned level);
  54
  55 static int xreq_subscribe_cb(void *closure, struct afb_event event);
  56 static int xreq_unsubscribe_cb(void *closure, struct afb_event event);
  57
  58 static void xreq_subcall_cb(
  59                 void *closure,
  60                 const char *api,
  61                 const char *verb,
  62                 struct json_object *args,
  63                 void (*callback)(void*, int, struct json_object*),
  64                 void *cb_closure);
  65
  66 const struct afb_req_itf xreq_itf = {
  67         .json = xreq_json_cb,
  68         .get = xreq_get_cb,
  69         .success = xreq_success_cb,
  70         .fail = xreq_fail_cb,
  71         .raw = xreq_raw_cb,
  72         .send = xreq_send_cb,
  73         .context_get = xreq_context_get_cb,
  74         .context_set = xreq_context_set_cb,
  75         .addref = xreq_addref_cb,
  76         .unref = xreq_unref_cb,
  77         .session_close = xreq_session_close_cb,
  78         .session_set_LOA = xreq_session_set_LOA_cb,
  79         .subscribe = xreq_subscribe_cb,
  80         .unsubscribe = xreq_unsubscribe_cb,
  81         .subcall = xreq_subcall_cb
  82 };
  83
  84
  85 static struct json_object *xreq_json_cb(void *closure)
  86 {
  87         struct afb_xreq *xreq = closure;
  88         return xreq->queryitf->json(xreq->query);
  89 }
  90
  91 static struct afb_arg xreq_get_cb(void *closure, const char *name)
  92 {
  93         struct afb_xreq *xreq = closure;
  94         if (xreq->queryitf->get)
  95                 return xreq->queryitf->get(xreq->query, name);
  96         else
  97                 return afb_msg_json_get_arg(xreq_json_cb(closure), name);
  98 }
  99
 100 static void xreq_success_cb(void *closure, struct json_object *obj, const char *info)
 101 {
 102         struct afb_xreq *xreq = closure;
 103         if (xreq->replied) {
 104                 ERROR("reply called more than one time!!");
 105                 json_object_put(obj);
 106         } else {
 107                 xreq->replied = 1;
 108                 if (xreq->queryitf->success)
 109                         xreq->queryitf->success(xreq->query, obj, info);
 110                 else
 111                         xreq->queryitf->reply(xreq->query, afb_msg_json_reply_ok(info, obj, &xreq->context, NULL));
 112         }
 113 }
 114
 115 static void xreq_fail_cb(void *closure, const char *status, const char *info)
 116 {
 117         struct afb_xreq *xreq = closure;
 118         if (xreq->replied) {
 119                 ERROR("reply called more than one time!!");
 120         } else {
 121                 xreq->replied = 1;
 122                 if (xreq->queryitf->fail)
 123                         xreq->queryitf->fail(xreq->query, status, info);
 124                 else
 125                         xreq->queryitf->reply(xreq->query, afb_msg_json_reply_error(status, info, &xreq->context, NULL));
 126         }
 127 }
 128
 129 static const char *xreq_raw_cb(void *closure, size_t *size)
 130 {
 131         struct afb_xreq *xreq = closure;
 132         const char *result = json_object_to_json_string(xreq->queryitf->json(xreq->query));
 133         if (size != NULL)
 134                 *size = strlen(result);
 135         return result;
 136 }
 137
 138 static void xreq_send_cb(void *closure, const char *buffer, size_t size)
 139 {
 140         struct json_object *obj = json_tokener_parse(buffer);
 141         if (!obj == !buffer)
 142                 xreq_success_cb(closure, obj, "fake send");
 143         else
 144                 xreq_fail_cb(closure, "fake-send-failed", "fake send");
 145 }
 146
 147 static void *xreq_context_get_cb(void *closure)
 148 {
 149         struct afb_xreq *xreq = closure;
 150         return afb_context_get(&xreq->context);
 151 }
 152
 153 static void xreq_context_set_cb(void *closure, void *value, void (*free_value)(void*))
 154 {
 155         struct afb_xreq *xreq = closure;
 156         afb_context_set(&xreq->context, value, free_value);
 157 }
 158
 159 static void xreq_addref_cb(void *closure)
 160 {
 161         struct afb_xreq *xreq = closure;
 162         afb_xreq_addref(xreq);
 163 }
 164
 165 void afb_xreq_addref(struct afb_xreq *xreq)
 166 {
 167         xreq->refcount++;
 168 }
 169
 170 static void xreq_unref_cb(void *closure)
 171 {
 172         struct afb_xreq *xreq = closure;
 173         afb_xreq_unref(xreq);
 174 }
 175
 176 void afb_xreq_unref(struct afb_xreq *xreq)
 177 {
 178         if (!--xreq->refcount) {
 179                 xreq->queryitf->unref(xreq->query);
 180         }
 181 }
 182
 183 static void xreq_session_close_cb(void *closure)
 184 {
 185         struct afb_xreq *xreq = closure;
 186         afb_context_close(&xreq->context);
 187 }
 188
 189 static int xreq_session_set_LOA_cb(void *closure, unsigned level)
 190 {
 191         struct afb_xreq *xreq = closure;
 192         return afb_context_change_loa(&xreq->context, level);
 193 }
 194
 195 static int xreq_subscribe_cb(void *closure, struct afb_event event)
 196 {
 197         struct afb_xreq *xreq = closure;
 198         if (xreq->listener)
 199                 return afb_evt_add_watch(xreq->listener, event);
 200         ERROR("no event listener, subscription impossible");
 201         errno = EINVAL;
 202         return -1;
 203 }
 204
 205 static int xreq_unsubscribe_cb(void *closure, struct afb_event event)
 206 {
 207         struct afb_xreq *xreq = closure;
 208         if (xreq->listener)
 209                 return afb_evt_remove_watch(xreq->listener, event);
 210         ERROR("no event listener, unsubscription impossible");
 211         errno = EINVAL;
 212         return -1;
 213 }
 214
 215 static void xreq_subcall_cb(void *closure, const char *api, const char *verb, struct json_object *args, void (*callback)(void*, int, struct json_object*), void *cb_closure)
 216 {
 217         struct afb_xreq *xreq = closure;
 218         afb_subcall(&xreq->context, api, verb, args, callback, cb_closure, (struct afb_req){ .itf = &xreq_itf, .closure = xreq });
 219 }
 220
 221 void afb_xreq_success_f(struct afb_xreq *xreq, struct json_object *obj, const char *info, ...)
 222 {
 223         char *message;
 224         va_list args;
 225         va_start(args, info);
 226         if (info == NULL || vasprintf(&message, info, args) < 0)
 227                 message = NULL;
 228         va_end(args);
 229         xreq_success_cb(xreq, obj, message);
 230         free(message);
 231 }
 232
 233 void afb_xreq_fail_f(struct afb_xreq *xreq, const char *status, const char *info, ...)
 234 {
 235         char *message;
 236         va_list args;
 237         va_start(args, info);
 238         if (info == NULL || vasprintf(&message, info, args) < 0)
 239                 message = NULL;
 240         va_end(args);
 241         xreq_fail_cb(xreq, status, message);
 242         free(message);
 243 }
 244
 245 static int xcheck(struct afb_xreq *xreq)
 246 {
 247         int stag = xreq->sessionflags;
 248
 249         if ((stag & (AFB_SESSION_CREATE|AFB_SESSION_CLOSE|AFB_SESSION_RENEW|AFB_SESSION_CHECK|AFB_SESSION_LOA_EQ)) != 0) {
 250                 if (!afb_context_check(&xreq->context)) {
 251                         afb_context_close(&xreq->context);
 252                         afb_xreq_fail_f(xreq, "failed", "invalid token's identity");
 253                         return 0;
 254                 }
 255         }
 256
 257         if ((stag & AFB_SESSION_CREATE) != 0) {
 258                 if (afb_context_check_loa(&xreq->context, 1)) {
 259                         afb_xreq_fail_f(xreq, "failed", "invalid creation state");
 260                         return 0;
 261                 }
 262                 afb_context_change_loa(&xreq->context, 1);
 263                 afb_context_refresh(&xreq->context);
 264         }
 265
 266         if ((stag & (AFB_SESSION_CREATE | AFB_SESSION_RENEW)) != 0)
 267                 afb_context_refresh(&xreq->context);
 268
 269         if ((stag & AFB_SESSION_CLOSE) != 0) {
 270                 afb_context_change_loa(&xreq->context, 0);
 271                 afb_context_close(&xreq->context);
 272         }
 273
 274         if ((stag & AFB_SESSION_LOA_GE) != 0) {
 275                 int loa = (stag >> AFB_SESSION_LOA_SHIFT) & AFB_SESSION_LOA_MASK;
 276                 if (!afb_context_check_loa(&xreq->context, loa)) {
 277                         afb_xreq_fail_f(xreq, "failed", "invalid LOA");
 278                         return 0;
 279                 }
 280         }
 281
 282         if ((stag & AFB_SESSION_LOA_LE) != 0) {
 283                 int loa = (stag >> AFB_SESSION_LOA_SHIFT) & AFB_SESSION_LOA_MASK;
 284                 if (afb_context_check_loa(&xreq->context, loa + 1)) {
 285                         afb_xreq_fail_f(xreq, "failed", "invalid LOA");
 286                         return 0;
 287                 }
 288         }
 289         return 1;
 290 }
 291
 292 static void xreq_run_cb(int signum, void *arg)
 293 {
 294         struct afb_xreq *xreq = arg;
 295
 296         if (signum == 0)
 297                 xreq->callback((struct afb_req){ .itf = &xreq_itf, .closure = xreq });
 298         else {
 299                 afb_xreq_fail_f(xreq, "aborted", "signal %s(%d) caught", strsignal(signum), signum);
 300
 301         }
 302         afb_xreq_unref(xreq);
 303 }
 304
 305 void afb_xreq_call(struct afb_xreq *xreq)
 306 {
 307         int rc;
 308         if (xcheck(xreq)) {
 309                 afb_xreq_addref(xreq);
 310                 rc = jobs_queue(xreq->group, xreq->timeout, xreq_run_cb, xreq);
 311                 if (rc < 0) {
 312                         /* TODO: allows or not to proccess it directly as when no threading? (see above) */
 313                         ERROR("can't process job with threads: %m");
 314                         afb_xreq_fail_f(xreq, "cancelled", "not able to pipe a job for the task");
 315                         xreq_unref_cb(xreq);
 316                 }
 317         }
 318 }
 319