afb-client-demo: Allow pipe of messages
[src/app-framework-binder.git] / src / main-afb-client-demo.c
1 /*
2  * Copyright (C) 2015-2019 "IoT.bzh"
3  * Author "Fulup Ar Foll"
4  * Author José Bollo <jose.bollo@iot.bzh>
5  *
6  * Licensed under the Apache License, Version 2.0 (the "License");
7  * you may not use this file except in compliance with the License.
8  * You may obtain a copy of the License at
9  *
10  *   http://www.apache.org/licenses/LICENSE-2.0
11  *
12  * Unless required by applicable law or agreed to in writing, software
13  * distributed under the License is distributed on an "AS IS" BASIS,
14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15  * See the License for the specific language governing permissions and
16  * limitations under the License.
17  */
18
19 #define _GNU_SOURCE
20
21 #include <stdlib.h>
22 #include <stdio.h>
23 #include <stdint.h>
24 #include <string.h>
25 #include <unistd.h>
26 #include <fcntl.h>
27 #include <sys/stat.h>
28 #include <sys/types.h>
29 #include <errno.h>
30
31 #include <systemd/sd-event.h>
32 #include <json-c/json.h>
33 #if !defined(JSON_C_TO_STRING_NOSLASHESCAPE)
34 #define JSON_C_TO_STRING_NOSLASHESCAPE 0
35 #endif
36
37 #include "afb-wsj1.h"
38 #include "afb-ws-client.h"
39 #include "afb-proto-ws.h"
40
41 enum {
42         Exit_Success      = 0,
43         Exit_Error        = 1,
44         Exit_HangUp       = 2,
45         Exit_Input_Fail   = 3,
46         Exit_Bad_Arg      = 4,
47         Exit_Cant_Connect = 5
48 };
49
50
51 /* declaration of functions */
52 static void on_wsj1_hangup(void *closure, struct afb_wsj1 *wsj1);
53 static void on_wsj1_call(void *closure, const char *api, const char *verb, struct afb_wsj1_msg *msg);
54 static void on_wsj1_event(void *closure, const char *event, struct afb_wsj1_msg *msg);
55
56 static void on_pws_hangup(void *closure);
57 static void on_pws_reply(void *closure, void *request, struct json_object *result, const char *error, const char *info);
58 static void on_pws_event_create(void *closure, const char *event_name, int event_id);
59 static void on_pws_event_remove(void *closure, const char *event_name, int event_id);
60 static void on_pws_event_subscribe(void *closure, void *request, const char *event_name, int event_id);
61 static void on_pws_event_unsubscribe(void *closure, void *request, const char *event_name, int event_id);
62 static void on_pws_event_push(void *closure, const char *event_name, int event_id, struct json_object *data);
63 static void on_pws_event_broadcast(void *closure, const char *event_name, struct json_object *data);
64
65 static void idle();
66 static int process_stdin();
67 static int on_stdin(sd_event_source *src, int fd, uint32_t revents, void *closure);
68
69 static void wsj1_emit(const char *api, const char *verb, const char *object);
70 static void pws_call(const char *verb, const char *object);
71
72 /* the callback interface for wsj1 */
73 static struct afb_wsj1_itf wsj1_itf = {
74         .on_hangup = on_wsj1_hangup,
75         .on_call = on_wsj1_call,
76         .on_event = on_wsj1_event
77 };
78
79 /* the callback interface for pws */
80 static struct afb_proto_ws_client_itf pws_itf = {
81         .on_reply = on_pws_reply,
82         .on_event_create = on_pws_event_create,
83         .on_event_remove = on_pws_event_remove,
84         .on_event_subscribe = on_pws_event_subscribe,
85         .on_event_unsubscribe = on_pws_event_unsubscribe,
86         .on_event_push = on_pws_event_push,
87         .on_event_broadcast = on_pws_event_broadcast,
88 };
89
90 /* global variables */
91 static struct afb_wsj1 *wsj1;
92 static struct afb_proto_ws *pws;
93 static int breakcon;
94 static int exonrep;
95 static int callcount;
96 static int human;
97 static int raw;
98 static int keeprun;
99 static int direct;
100 static int echo;
101 static int synchro;
102 static int usein;
103 static sd_event *loop;
104 static sd_event_source *evsrc;
105 static char *sessionid = "afb-client-demo";
106 static int exitcode = 0;
107
108 /* print usage of the program */
109 static void usage(int status, char *arg0)
110 {
111         char *name = strrchr(arg0, '/');
112         name = name ? name + 1 : arg0;
113         fprintf(status ? stderr : stdout, "usage: %s [-H [-r]] [-b] [-e] uri [api verb [data]]\n", name);
114         fprintf(status ? stderr : stdout, "       %s -d [-H [-r]] [-b] [-e] uri [verb [data]]\n", name);
115         fprintf(status ? stderr : stdout, "\n"
116                 "allowed options\n"
117                 "  --break, -b         Break connection just after event/call has been emitted.\n"
118                 "  --direct, -d        Direct api\n"
119                 "  --echo, -e          Echo inputs\n"
120                 "  --help, -h          Display this help\n"
121                 "  --human, -H         Display human readable JSON\n"
122                 "  --raw, -r           Raw output (default)\n"
123                 "  --sync, -s          Synchronous: wait for answers (like -p 1)\n"
124                 "  --keep-running, -k  Keep running until disconnect, even if input closed\n"
125                 "  --pipe, -p COUNT    Allow to pipe COUNT requests\n"
126                 "Example:\n"
127                 " %s --human 'localhost:1234/api?token=HELLO&uuid=magic' hello ping\n"
128                 "\n", name
129         );
130
131         exit(status);
132 }
133
134 /* entry function */
135 int main(int ac, char **av, char **env)
136 {
137         int rc;
138         char *a0, *an;
139
140         /* get the program name */
141         a0 = av[0];
142
143         /* check options */
144         while (ac > 1 && (an = av[1])[0] == '-') {
145                 if (an[1] == '-') {
146                         /* long option */
147
148                         if (!strcmp(an, "--human")) /* request for human output */
149                                 human = 1;
150
151                         else if (!strcmp(an, "--raw")) /* request for raw output */
152                                 raw = 1;
153
154                         else if (!strcmp(an, "--direct")) /* request for direct api */
155                                 direct = 1;
156
157                         else if (!strcmp(an, "--break")) /* request to break connection */
158                                 breakcon = 1;
159
160                         else if (!strcmp(an, "--keep-running")) /* request to break connection */
161                                 keeprun = 1;
162
163                         else if (!strcmp(an, "--sync")) /* request to break connection */
164                                 synchro = 1;
165
166                         else if (!strcmp(an, "--echo")) /* request to echo inputs */
167                                 echo = 1;
168
169                         else if (!strcmp(an, "--pipe") && av[2] && atoi(av[2]) > 0) {
170                                 synchro = atoi(av[2]);
171                                 av++;
172                                 ac--;
173                         }
174
175                         /* emit usage and exit */
176                         else
177                                 usage(strcmp(an, "--help") ? Exit_Bad_Arg : Exit_Success, a0);
178                 } else {
179                         /* short option(s) */
180                         for (rc = 1 ; an[rc] ; rc++)
181                                 switch (an[rc]) {
182                                 case 'H': human = 1; break;
183                                 case 'r': raw = 1; break;
184                                 case 'd': direct = 1; break;
185                                 case 'b': breakcon = 1; break;
186                                 case 'k': keeprun = 1; break;
187                                 case 's': synchro = 1; break;
188                                 case 'e': echo = 1; break;
189                                 case 'p': if (av[2] && atoi(av[2]) > 0) { synchro = atoi(av[2]); av++; ac--; break; } /*@fallthrough@*/
190                                 default: usage(an[rc] != 'h' ? Exit_Bad_Arg : Exit_Success, a0);
191                                 }
192                 }
193                 av++;
194                 ac--;
195         }
196
197         /* check the argument count */
198         if (ac != 2 && ac != 4 && ac != 5)
199                 usage(1, a0);
200
201         /* set raw by default */
202         if (!human)
203                 raw = 1;
204
205         /* get the default event loop */
206         rc = sd_event_default(&loop);
207         if (rc < 0) {
208                 fprintf(stderr, "connection to default event loop failed: %s\n", strerror(-rc));
209                 return 1;
210         }
211
212         /* connect the websocket wsj1 to the uri given by the first argument */
213         if (direct) {
214                 pws = afb_ws_client_connect_api(loop, av[1], &pws_itf, NULL);
215                 if (pws == NULL) {
216                         fprintf(stderr, "connection to %s failed: %m\n", av[1]);
217                         return Exit_Cant_Connect;
218                 }
219                 afb_proto_ws_on_hangup(pws, on_pws_hangup);
220         } else {
221                 wsj1 = afb_ws_client_connect_wsj1(loop, av[1], &wsj1_itf, NULL);
222                 if (wsj1 == NULL) {
223                         fprintf(stderr, "connection to %s failed: %m\n", av[1]);
224                         return Exit_Cant_Connect;
225                 }
226         }
227
228         /* test the behaviour */
229         if (ac == 2) {
230                 /* get requests from stdin */
231                 usein = 1;
232                 fcntl(0, F_SETFL, O_NONBLOCK);
233                 if (sd_event_add_io(loop, &evsrc, 0, EPOLLIN, on_stdin, NULL) < 0)
234                         evsrc = NULL;
235         } else {
236                 /* the request is defined by the arguments */
237                 usein = 0;
238                 exonrep = !keeprun;
239                 if (direct)
240                         pws_call(av[2], av[3]);
241                 else
242                         wsj1_emit(av[2], av[3], av[4]);
243         }
244
245         /* loop until end */
246         idle();
247         return 0;
248 }
249
250 static void idle()
251 {
252         for(;;) {
253                 if (!usein) {
254                         if (!keeprun && !callcount)
255                                 exit(exitcode);
256                         sd_event_run(loop, 30000000);
257                 }
258                 else if (!synchro || callcount < synchro) {
259                         if (!process_stdin() && usein)
260                                 sd_event_run(loop, 100000);
261                 } else {
262                         sd_event_run(loop, 30000000);
263                 }
264         }
265 }
266
267 /* decrement the count of calls */
268 static void dec_callcount()
269 {
270         callcount--;
271         if (exonrep && !callcount)
272                 exit(exitcode);
273 }
274
275 /* called when wsj1 hangsup */
276 static void on_wsj1_hangup(void *closure, struct afb_wsj1 *wsj1)
277 {
278         printf("ON-HANGUP\n");
279         fflush(stdout);
280         exit(Exit_HangUp);
281 }
282
283 /* called when wsj1 receives a method invocation */
284 static void on_wsj1_call(void *closure, const char *api, const char *verb, struct afb_wsj1_msg *msg)
285 {
286         int rc;
287         if (raw)
288                 printf("%s\n", afb_wsj1_msg_object_s(msg));
289         if (human)
290                 printf("ON-CALL %s/%s:\n%s\n", api, verb,
291                                 json_object_to_json_string_ext(afb_wsj1_msg_object_j(msg),
292                                                         JSON_C_TO_STRING_PRETTY|JSON_C_TO_STRING_NOSLASHESCAPE));
293         fflush(stdout);
294         rc = afb_wsj1_reply_error_s(msg, "\"unimplemented\"", NULL);
295         if (rc < 0)
296                 fprintf(stderr, "replying failed: %m\n");
297 }
298
299 /* called when wsj1 receives an event */
300 static void on_wsj1_event(void *closure, const char *event, struct afb_wsj1_msg *msg)
301 {
302         if (raw)
303                 printf("%s\n", afb_wsj1_msg_object_s(msg));
304         if (human)
305                 printf("ON-EVENT %s:\n%s\n", event,
306                                 json_object_to_json_string_ext(afb_wsj1_msg_object_j(msg),
307                                                         JSON_C_TO_STRING_PRETTY|JSON_C_TO_STRING_NOSLASHESCAPE));
308         fflush(stdout);
309 }
310
311 /* called when wsj1 receives a reply */
312 static void on_wsj1_reply(void *closure, struct afb_wsj1_msg *msg)
313 {
314         int iserror = !afb_wsj1_msg_is_reply_ok(msg);
315         exitcode = iserror ? Exit_Error : Exit_Success;
316         if (raw)
317                 printf("%s\n", afb_wsj1_msg_object_s(msg));
318         if (human)
319                 printf("ON-REPLY %s: %s\n%s\n", (char*)closure,
320                                 iserror ? "ERROR" : "OK",
321                                 json_object_to_json_string_ext(afb_wsj1_msg_object_j(msg),
322                                                         JSON_C_TO_STRING_PRETTY|JSON_C_TO_STRING_NOSLASHESCAPE));
323         fflush(stdout);
324         free(closure);
325         dec_callcount();
326 }
327
328 /* makes a call */
329 static void wsj1_call(const char *api, const char *verb, const char *object)
330 {
331         static int num = 0;
332         char *key;
333         int rc;
334
335         /* allocates an id for the request */
336         rc = asprintf(&key, "%d:%s/%s", ++num, api, verb);
337
338         /* echo the command if asked */
339         if (echo)
340                 printf("SEND-CALL %s/%s %s\n", api, verb, object?:"null");
341
342         /* send the request */
343         callcount++;
344         rc = afb_wsj1_call_s(wsj1, api, verb, object, on_wsj1_reply, key);
345         if (rc < 0) {
346                 fprintf(stderr, "calling %s/%s(%s) failed: %m\n", api, verb, object);
347                 dec_callcount();
348         }
349 }
350
351 /* sends an event */
352 static void wsj1_event(const char *event, const char *object)
353 {
354         int rc;
355
356         /* echo the command if asked */
357         if (echo)
358                 printf("SEND-EVENT: %s %s\n", event, object?:"null");
359
360         rc = afb_wsj1_send_event_s(wsj1, event, object);
361         if (rc < 0)
362                 fprintf(stderr, "sending !%s(%s) failed: %m\n", event, object);
363 }
364
365 /* emits either a call (when api!='!') or an event */
366 static void wsj1_emit(const char *api, const char *verb, const char *object)
367 {
368         if (object == NULL || object[0] == 0)
369                 object = "null";
370
371         if (api[0] == '!' && api[1] == 0)
372                 wsj1_event(verb, object);
373         else
374                 wsj1_call(api, verb, object);
375         if (breakcon)
376                 exit(0);
377 }
378
379 /* process stdin */
380 static int process_stdin()
381 {
382         static size_t count = 0;
383         static char line[16384];
384         static char sep[] = " \t";
385         static char sepnl[] = " \t\n";
386
387         int result = 0;
388         ssize_t rc = 0;
389         size_t pos;
390
391         /* read the buffer */
392         while (sizeof line > count) {
393                 rc = read(0, line + count, sizeof line - count);
394                 if (rc >= 0 || errno != EINTR)
395                         break;
396         }
397         if (rc < 0) {
398                 if (errno == EAGAIN)
399                         return 0;
400                 fprintf(stderr, "read error: %m\n");
401                 exit(Exit_Input_Fail);
402         }
403         if (rc == 0) {
404                 usein = count != 0;
405                 if (!usein && !keeprun) {
406                         if (!callcount)
407                                 exit(exitcode);
408                         exonrep = 1;
409                 }
410         }
411         count += (size_t)rc;
412         if (synchro && callcount >= synchro)
413                 return 0;
414
415         /* normalise the buffer content */
416         /* TODO: handle backspace \x7f ? */
417         /* process the lines */
418         pos = 0;
419         for(;;) {
420                 size_t i, api[2], verb[2], rest[2];
421                 i = pos;
422                 while(i < count && strchr(sep, line[i])) i++;
423                 api[0] = i; while(i < count && !strchr(sepnl, line[i])) i++; api[1] = i;
424                 while(i < count && strchr(sep, line[i])) i++;
425                 if (direct) {
426                         verb[0] = api[0];
427                         verb[1] = api[1];
428                 } else {
429                         verb[0] = i; while(i < count && !strchr(sepnl, line[i])) i++; verb[1] = i;
430                         while(i < count && strchr(sep, line[i])) i++;
431                 }
432                 rest[0] = i; while(i < count && line[i] != '\n') i++; rest[1] = i;
433                 if (i == count) break;
434                 line[i++] = 0;
435                 pos = i;
436                 if (api[0] == api[1]) {
437                         /* empty line */
438                 } else if (line[api[0]] == '#') {
439                         /* comment */
440                 } else if (verb[0] == verb[1]) {
441                         fprintf(stderr, "verb missing, bad line: %s\n", line+pos);
442                 } else {
443                         line[api[1]] = line[verb[1]] = 0;
444                         if (direct)
445                                 pws_call(line + verb[0], line + rest[0]);
446                         else
447                                 wsj1_emit(line + api[0], line + verb[0], line + rest[0]);
448                         result = 1;
449                         break;
450                 }
451         }
452         count -= pos;
453         if (count == sizeof line) {
454                 fprintf(stderr, "overflow\n");
455                 exit(Exit_Input_Fail);
456         }
457         if (count)
458                 memmove(line, line + pos, count);
459
460         return result;
461 }
462
463 /* called when something happens on stdin */
464 static int on_stdin(sd_event_source *src, int fd, uint32_t revents, void *closure)
465 {
466         process_stdin();
467         if (!usein) {
468                 sd_event_source_unref(src);
469                 evsrc = NULL;
470         }
471         return 1;
472 }
473
474 static void on_pws_reply(void *closure, void *request, struct json_object *result, const char *error, const char *info)
475 {
476         int iserror = !!error;
477         exitcode = iserror ? Exit_Error : Exit_Success;
478         error = error ?: "success";
479         if (raw) {
480                 /* TODO: transitionnal: fake the structured response */
481                 struct json_object *x = json_object_new_object(), *y = json_object_new_object();
482                 json_object_object_add(x, "jtype", json_object_new_string("afb-reply"));
483                 json_object_object_add(x, "request", y);
484                 json_object_object_add(y, "status", json_object_new_string(error));
485                 if (info)
486                         json_object_object_add(y, "info", json_object_new_string(info));
487                 if (result)
488                         json_object_object_add(x, "response", json_object_get(result));
489
490                 printf("%s\n", json_object_to_json_string_ext(x, JSON_C_TO_STRING_NOSLASHESCAPE));
491                 json_object_put(x);
492         }
493         if (human)
494                 printf("ON-REPLY %s: %s %s\n%s\n", (char*)request, error, info ?: "", json_object_to_json_string_ext(result, JSON_C_TO_STRING_PRETTY|JSON_C_TO_STRING_NOSLASHESCAPE));
495         fflush(stdout);
496         free(request);
497         dec_callcount();
498 }
499
500 static void on_pws_event_create(void *closure, const char *event_name, int event_id)
501 {
502         printf("ON-EVENT-CREATE: [%d:%s]\n", event_id, event_name);
503         fflush(stdout);
504 }
505
506 static void on_pws_event_remove(void *closure, const char *event_name, int event_id)
507 {
508         printf("ON-EVENT-REMOVE: [%d:%s]\n", event_id, event_name);
509         fflush(stdout);
510 }
511
512 static void on_pws_event_subscribe(void *closure, void *request, const char *event_name, int event_id)
513 {
514         printf("ON-EVENT-SUBSCRIBE %s: [%d:%s]\n", (char*)request, event_id, event_name);
515         fflush(stdout);
516 }
517
518 static void on_pws_event_unsubscribe(void *closure, void *request, const char *event_name, int event_id)
519 {
520         printf("ON-EVENT-UNSUBSCRIBE %s: [%d:%s]\n", (char*)request, event_id, event_name);
521         fflush(stdout);
522 }
523
524 static void on_pws_event_push(void *closure, const char *event_name, int event_id, struct json_object *data)
525 {
526         if (raw)
527                 printf("ON-EVENT-PUSH: [%d:%s]\n%s\n", event_id, event_name, json_object_to_json_string_ext(data, JSON_C_TO_STRING_NOSLASHESCAPE));
528         if (human)
529                 printf("ON-EVENT-PUSH: [%d:%s]\n%s\n", event_id, event_name, json_object_to_json_string_ext(data, JSON_C_TO_STRING_PRETTY|JSON_C_TO_STRING_NOSLASHESCAPE));
530         fflush(stdout);
531 }
532
533 static void on_pws_event_broadcast(void *closure, const char *event_name, struct json_object *data)
534 {
535         if (raw)
536                 printf("ON-EVENT-BROADCAST: [%s]\n%s\n", event_name, json_object_to_json_string_ext(data, JSON_C_TO_STRING_NOSLASHESCAPE));
537         if (human)
538                 printf("ON-EVENT-BROADCAST: [%s]\n%s\n", event_name, json_object_to_json_string_ext(data, JSON_C_TO_STRING_PRETTY|JSON_C_TO_STRING_NOSLASHESCAPE));
539         fflush(stdout);
540 }
541
542 /* makes a call */
543 static void pws_call(const char *verb, const char *object)
544 {
545         static int num = 0;
546         char *key;
547         int rc;
548         struct json_object *o;
549         enum json_tokener_error jerr;
550
551         /* allocates an id for the request */
552         rc = asprintf(&key, "%d:%s", ++num, verb);
553
554         /* echo the command if asked */
555         if (echo)
556                 printf("SEND-CALL: %s %s\n", verb, object?:"null");
557
558         /* send the request */
559         callcount++;
560         if (object == NULL || object[0] == 0)
561                 o = NULL;
562         else {
563                 o = json_tokener_parse_verbose(object, &jerr);
564                 if (jerr != json_tokener_success)
565                         o = json_object_new_string(object);
566         }
567         rc = afb_proto_ws_client_call(pws, verb, o, sessionid, key, NULL);
568         json_object_put(o);
569         if (rc < 0) {
570                 fprintf(stderr, "calling %s(%s) failed: %m\n", verb, object?:"");
571                 dec_callcount();
572         }
573         if (breakcon)
574                 exit(0);
575 }
576
577 /* called when pws hangsup */
578 static void on_pws_hangup(void *closure)
579 {
580         printf("ON-HANGUP\n");
581         fflush(stdout);
582         exit(Exit_HangUp);
583 }