/*
- Copyright 2015, 2016 IoT.bzh
+ Copyright (C) 2015-2018 IoT.bzh
author: José Bollo <jose.bollo@iot.bzh>
limitations under the License.
*/
+#define _GNU_SOURCE
#include <unistd.h>
#include <stdio.h>
#include <time.h>
#include <systemd/sd-event.h>
#include <json-c/json.h>
+#include <afb/afb-ws-client.h>
+#include <afb/afb-proto-ws.h>
+
#include "verbose.h"
#include "utils-jbus.h"
#include "utils-json.h"
-#include "afm.h"
-#include "afm-db.h"
-#include "afm-launch-mode.h"
-#include "afm-run.h"
+
+#define AFM_USER_DBUS_PATH "/org/AGL/afm/user"
/*
* name of the application
*/
static const char appname[] = "afm-user-daemon";
+/*
+ * string for printing version
+ */
+static const char versionstr[] =
+ "\n"
+ " %s version="AFM_VERSION"\n"
+ "\n"
+ " Copyright (C) 2015, 2016, 2017 \"IoT.bzh\"\n"
+ " AFB comes with ABSOLUTELY NO WARRANTY.\n"
+ " Licence Apache 2\n"
+ "\n";
+
/*
* string for printing usage
*/
static const char usagestr[] =
- "usage: %s [-q] [-v] [-m mode] [-r rootdir]... [-a appdir]...\n"
+ "usage: %s [option(s)] afm-main-uri\n"
"\n"
- " -a appdir adds an application directory\n"
- " -r rootdir adds a root directory of applications\n"
- " -m mode set default launch mode (local or remote)\n"
" -d run as a daemon\n"
" -u addr address of user D-Bus to use\n"
- " -s addr address of system D-Bus to use\n"
" -q quiet\n"
" -v verbose\n"
+ " -V version\n"
"\n";
/*
* Option definition for getopt_long
*/
-static const char options_s[] = "hdqvr:a:m:";
+static const char options_s[] = "hdqvVu:";
static struct option options_l[] = {
- { "root", required_argument, NULL, 'r' },
- { "application", required_argument, NULL, 'a' },
- { "mode", required_argument, NULL, 'm' },
{ "user-dbus", required_argument, NULL, 'u' },
- { "system-dbus", required_argument, NULL, 's' },
{ "daemon", no_argument, NULL, 'd' },
{ "quiet", no_argument, NULL, 'q' },
{ "verbose", no_argument, NULL, 'v' },
{ "help", no_argument, NULL, 'h' },
+ { "version", no_argument, NULL, 'V' },
{ NULL, 0, NULL, 0 }
};
/*
- * Connections to D-Bus
- * This is an array for using the function
- * jbus_read_write_dispatch_multiple
- * directly without transformations.
+ * The methods propagated
*/
-static struct jbus *jbuses[2];
-#define system_bus jbuses[0]
-#define user_bus jbuses[1]
+static const char *methods[] = {
+ "runnables",
+ "detail",
+ "start",
+ "once",
+ "terminate",
+ "pause",
+ "resume",
+ "stop",
+ "continue",
+ "runners",
+ "state",
+ "install",
+ "uninstall",
+ NULL
+};
/*
- * Handle to the database of applications
+ * Connections
*/
-static struct afm_db *afdb;
+static struct sd_event *evloop;
+static struct jbus *user_bus;
+static struct afb_proto_ws *pws;
+static char *sessionid;
+static const char *uri;
/*
- * Returned error strings
+ *
*/
-const char error_nothing[] = "[]";
-const char error_bad_request[] = "\"bad request\"";
-const char error_not_found[] = "\"not found\"";
-const char error_cant_start[] = "\"can't start\"";
-const char error_system[] = "\"system error\"";
+static void on_pws_hangup(void *closure);
+static void on_pws_reply(void *closure, void *request, struct json_object *obj, const char *error, const char *info);
+#if !defined(AFB_PROTO_WS_VERSION) || (AFB_PROTO_WS_VERSION < 3)
+static void on_pws_reply_success(void *closure, void *request, struct json_object *result, const char *info)
+ { on_pws_reply(closure, request, result, NULL, info); }
+static void on_pws_reply_fail(void *closure, void *request, const char *error, const char *info)
+ { on_pws_reply(closure, request, NULL, error, info); }
+#endif
+static void on_pws_event_broadcast(void *closure, const char *event_name, struct json_object *data);
+/* the callback interface for pws */
+static struct afb_proto_ws_client_itf pws_itf = {
+#if !defined(AFB_PROTO_WS_VERSION) || (AFB_PROTO_WS_VERSION < 3)
+ .on_reply_success = on_pws_reply_success,
+ .on_reply_fail = on_pws_reply_fail,
+#else
+ .on_reply = on_pws_reply,
+#endif
+ .on_event_broadcast = on_pws_event_broadcast,
+};
-/*
- * retrieves the 'runid' in 'obj' parameters received with the
- * request 'smsg' for the 'method'.
- *
- * Returns 1 in case of success.
- * Otherwise, if the 'runid' can't be retrived, an error stating
- * the bad request is replied for 'smsg' and 0 is returned.
- */
-static int onrunid(struct sd_bus_message *smsg, struct json_object *obj,
- const char *method, int *runid)
+static int try_connect_pws()
{
- if (!j_read_integer(obj, runid)
- && !j_read_integer_at(obj, "runid", runid)) {
- INFO("bad request method %s: %s", method,
- json_object_to_json_string(obj));
- jbus_reply_error_s(smsg, error_bad_request);
+ pws = afb_ws_client_connect_api(evloop, uri, &pws_itf, NULL);
+ if (pws == NULL) {
+ fprintf(stderr, "connection to %s failed: %m\n", uri);
return 0;
}
-
- INFO("method %s called for %d", method, *runid);
+ afb_proto_ws_on_hangup(pws, on_pws_hangup);
return 1;
}
-/*
- * Sends the reply 'resp' to the request 'smsg' if 'resp' is not NULL.
- * Otherwise, when 'resp' is NULL replies the error string 'errstr'.
- */
-static void reply(struct sd_bus_message *smsg, struct json_object *resp,
- const char *errstr)
-{
- if (resp)
- jbus_reply_j(smsg, resp);
- else
- jbus_reply_error_s(smsg, errstr);
-}
-
-/*
- * Sends the reply "true" to the request 'smsg' if 'status' is zero.
- * Otherwise, when 'status' is not zero replies the error string 'errstr'.
- */
-static void reply_status(struct sd_bus_message *smsg, int status, const char *errstr)
-{
- if (status)
- jbus_reply_error_s(smsg, errstr);
- else
- jbus_reply_s(smsg, "true");
-}
+static void attempt_connect_pws(int count);
-/*
- * On query "runnables" from 'smsg' with parameters of 'obj'.
- *
- * Nothing is expected in 'obj' that can be anything.
- */
-static void on_runnables(struct sd_bus_message *smsg, struct json_object *obj, void *unused)
+static int timehand(sd_event_source *s, uint64_t usec, void *userdata)
{
- struct json_object *resp;
- INFO("method runnables called");
- resp = afm_db_application_list(afdb);
- jbus_reply_j(smsg, resp);
- json_object_put(resp);
+ sd_event_source_unref(s);
+ attempt_connect_pws((int)(intptr_t)userdata);
+ return 0;
}
-/*
- * On query "detail" from 'smsg' with parameters of 'obj'.
- */
-static void on_detail(struct sd_bus_message *smsg, struct json_object *obj, void *unused)
+static void attempt_connect_pws(int count)
{
- const char *appid;
- struct json_object *resp;
-
- /* get the parameters */
- if (j_read_string(obj, &appid))
- ; /* appid as a string */
- else if (j_read_string_at(obj, "id", &appid))
- ; /* appid as obj.id string */
- else {
- INFO("method detail called but bad request!");
- jbus_reply_error_s(smsg, error_bad_request);
- return;
- }
-
- /* wants details for appid */
- INFO("method detail called for %s", appid);
- resp = afm_db_get_application_public(afdb, appid);
- reply(smsg, resp, error_not_found);
- json_object_put(resp);
-}
-
-
-/*
- * On query "start" from 'smsg' with parameters of 'obj'.
- */
-static void on_start(struct sd_bus_message *smsg, struct json_object *obj, void *unused)
-{
- const char *appid, *modestr;
- char *uri;
- struct json_object *appli, *resp;
- int runid;
- char runidstr[20];
- enum afm_launch_mode mode;
-
- /* get the parameters */
- mode = invalid_launch_mode;
- if (j_read_string(obj, &appid)) {
- mode = get_default_launch_mode();
- } else if (j_read_string_at(obj, "id", &appid)) {
- if (j_read_string_at(obj, "mode", &modestr)) {
- mode = launch_mode_of_name(modestr);
- } else {
- mode = get_default_launch_mode();
+ sd_event_source *s;
+ if (!try_connect_pws()) {
+ if (--count <= 0) {
+ ERROR("Definitely disconnected");
+ exit(1);
}
- }
- if (!is_valid_launch_mode(mode)) {
- jbus_reply_error_s(smsg, error_bad_request);
- return;
- }
-
- /* get the application */
- INFO("method start called for %s mode=%s", appid,
- name_of_launch_mode(mode));
- appli = afm_db_get_application(afdb, appid);
- if (appli == NULL) {
- jbus_reply_error_s(smsg, error_not_found);
- return;
- }
-
- /* launch the application */
- uri = NULL;
- runid = afm_run_start(appli, mode, &uri);
- if (runid <= 0) {
- jbus_reply_error_s(smsg, error_cant_start);
- free(uri);
- return;
- }
-
- if (uri == NULL) {
- /* returns only the runid */
- snprintf(runidstr, sizeof runidstr, "%d", runid);
- runidstr[sizeof runidstr - 1] = 0;
- jbus_reply_s(smsg, runidstr);
- return;
- }
-
- /* returns the runid and its uri */
- resp = json_object_new_object();
- if (resp != NULL && j_add_integer(resp, "runid", runid)
- && j_add_string(resp, "uri", uri))
- jbus_reply_j(smsg, resp);
- else {
- afm_run_terminate(runid);
- jbus_reply_error_s(smsg, error_system);
- }
- json_object_put(resp);
- free(uri);
-}
-
-/*
- * On query "pause" from 'smsg' with parameters of 'obj'.
- */
-static void on_pause(struct sd_bus_message *smsg, struct json_object *obj, void *unused)
-{
- int runid, status;
- if (onrunid(smsg, obj, "pause", &runid)) {
- status = afm_run_pause(runid);
- reply_status(smsg, status, error_not_found);
+ sd_event_add_time(evloop, &s, CLOCK_MONOTONIC, 5000000, 0, timehand, (void*)(intptr_t)count);
}
}
-/*
- * On query "resume" from 'smsg' with parameters of 'obj'.
- */
-static void on_resume(struct sd_bus_message *smsg, struct json_object *obj, void *unused)
-{
- int runid, status;
- if (onrunid(smsg, obj, "resume", &runid)) {
- status = afm_run_resume(runid);
- reply_status(smsg, status, error_not_found);
- }
-}
-
-/*
- * On query "stop" from 'smsg' with parameters of 'obj'.
- */
-static void on_stop(struct sd_bus_message *smsg, struct json_object *obj, void *unused)
-{
- NOTICE("call to obsolete 'stop'");
- on_pause(smsg, obj, unused);
-}
-
-/*
- * On query "continue" from 'smsg' with parameters of 'obj'.
- */
-static void on_continue(struct sd_bus_message *smsg, struct json_object *obj, void *unused)
-{
- NOTICE("call to obsolete 'continue'");
- on_resume(smsg, obj, unused);
-}
-
-/*
- * On query "terminate" from 'smsg' with parameters of 'obj'.
- */
-static void on_terminate(struct sd_bus_message *smsg, struct json_object *obj, void *unused)
+static void on_pws_reply(void *closure, void *request, struct json_object *obj, const char *error, const char *info)
{
- int runid, status;
- if (onrunid(smsg, obj, "terminate", &runid)) {
- status = afm_run_terminate(runid);
- reply_status(smsg, status, error_not_found);
- }
+ struct sd_bus_message *smsg = request;
+ if (error)
+ jbus_reply_error_s(smsg, error);
+ else
+ jbus_reply_j(smsg, obj);
}
-/*
- * On query "runners" from 'smsg' with parameters of 'obj'.
- */
-static void on_runners(struct sd_bus_message *smsg, struct json_object *obj, void *unused)
+static void on_pws_event_broadcast(void *closure, const char *event_name, struct json_object *data)
{
- struct json_object *resp;
- INFO("method runners called");
- resp = afm_run_list();
- jbus_reply_j(smsg, resp);
- json_object_put(resp);
+ jbus_send_signal_j(user_bus, "changed", data);
}
-/*
- * On query "state" from 'smsg' with parameters of 'obj'.
- */
-static void on_state(struct sd_bus_message *smsg, struct json_object *obj, void *unused)
+/* called when pws hangsup */
+static void on_pws_hangup(void *closure)
{
- int runid;
- struct json_object *resp;
- if (onrunid(smsg, obj, "state", &runid)) {
- resp = afm_run_state(runid);
- reply(smsg, resp, error_not_found);
- json_object_put(resp);
- }
+ struct afb_proto_ws *apw = pws;
+ pws = NULL;
+ afb_proto_ws_unref(apw);
+ attempt_connect_pws(10);
}
-/*
- * Calls the system daemon to achieve application management of
- * the 'method' gotten from 'smsg' with the parameter's string 'msg'.
- *
- * The principle is very simple: call the corresponding system method
- * and reply its response to the caller.
- *
- * The request and reply is synchronous and is blocking.
- * It is possible to implment it in an asynchrounous way but it
- * would brake the common behaviour. It would be a call like
- * jbus_call_ss(system_bus, method, msg, callback, smsg)
- */
-static void propagate(struct sd_bus_message *smsg, const char *msg, const char *method)
+/* propagate the call to the service */
+static void propagate(struct sd_bus_message *smsg, struct json_object *obj, void *closure)
{
- char *reply;
- INFO("method %s propagated with %s", method, msg);
- reply = jbus_call_ss_sync(system_bus, method, msg);
- if (reply) {
- jbus_reply_s(smsg, reply);
- free(reply);
- }
- else
- jbus_reply_error_s(smsg, error_system);
-}
-
-#if defined(EXPLICIT_CALL)
-/*
- * On query "install" from 'smsg' with parameters of 'msg'.
- */
-static void on_install(struct sd_bus_message *smsg, const char *msg, void *unused)
-{
- return propagate(smsg, msg, "install");
-}
+ int rc;
+ const char *verb = closure;
+ const char *onbehalf = NULL; /* TODO: on behalf of the client */
-/*
- * On query "uninstall" from 'smsg' with parameters of 'msg'.
- */
-static void on_uninstall(struct sd_bus_message *smsg, const char *msg, void *unused)
-{
- return propagate(smsg, msg, "uninstall");
-}
+ INFO("method %s propagated for %s", verb, json_object_to_json_string(obj));
+ if (!pws)
+ jbus_reply_error_s(smsg, "disconnected");
+ else {
+#if defined(AFB_PROTO_WS_VERSION) && (AFB_PROTO_WS_VERSION >= 3)
+ rc = afb_proto_ws_client_call(pws, verb, obj, sessionid, smsg, onbehalf);
+#else
+ rc = afb_proto_ws_client_call(pws, verb, obj, sessionid, smsg);
#endif
-
-/*
- * On system signaling that applications list changed
- */
-static void on_signal_changed(struct json_object *obj, void *unused)
-{
- /* update the database */
- afm_db_update_applications(afdb);
- /* re-propagate now */
- jbus_send_signal_j(user_bus, "changed", obj);
+ if (rc < 0)
+ ERROR("calling %s(%s) failed: %m\n", verb, json_object_to_json_string(obj));
+ }
}
/*
int rc;
if (address == NULL)
- return (isuser ? sd_bus_open_user : sd_bus_open_system)(ret);
+ return (isuser ? sd_bus_default_user : sd_bus_default_system)(ret);
rc = sd_bus_new(&b);
if (rc < 0)
int main(int ac, char **av)
{
int i, daemon = 0, rc;
- enum afm_launch_mode mode;
- struct sd_event *evloop;
- struct sd_bus *sysbus, *usrbus;
- const char *sys_bus_addr, *usr_bus_addr;
+ struct sd_bus *usrbus;
+ const char *usr_bus_addr;
+ const char **iter;
LOGAUTH(appname);
/* first interpretation of arguments */
- sys_bus_addr = NULL;
usr_bus_addr = NULL;
while ((i = getopt_long(ac, av, options_s, options_l, NULL)) >= 0) {
switch (i) {
case 'h':
printf(usagestr, appname);
return 0;
+ case 'V':
+ printf(versionstr, appname);
+ return 0;
case 'q':
if (verbosity)
verbosity--;
case 'd':
daemon = 1;
break;
- case 'r':
- break;
- case 'a':
- break;
- case 'm':
- mode = launch_mode_of_name(optarg);
- if (!is_valid_launch_mode(mode)) {
- ERROR("invalid mode '%s'", optarg);
- return 1;
- }
- set_default_launch_mode(mode);
- break;
case 'u':
usr_bus_addr = optarg;
break;
- case 's':
- sys_bus_addr = optarg;
- break;
case ':':
ERROR("missing argument value");
return 1;
}
}
- /* init random generator */
- srandom((unsigned int)time(NULL));
-
- /* init runners */
- if (afm_run_init()) {
- ERROR("afm_run_init failed");
- return 1;
- }
-
- /* init framework */
- afdb = afm_db_create();
- if (!afdb) {
- ERROR("afm_create failed");
+ /* check argument count */
+ if (optind >= ac) {
+ ERROR("Uri to the framework is missing");
return 1;
}
- if (afm_db_add_root(afdb, FWK_APP_DIR)) {
- ERROR("can't add root %s", FWK_APP_DIR);
+ if (optind + 1 != ac) {
+ ERROR("Extra parameters found");
return 1;
}
+ uri = av[optind];
- /* second interpretation of arguments */
- optind = 1;
- while ((i = getopt_long(ac, av, options_s, options_l, NULL)) >= 0) {
- switch (i) {
- case 'r':
- if (afm_db_add_root(afdb, optarg)) {
- ERROR("can't add root %s", optarg);
- return 1;
- }
- break;
- case 'a':
- if (afm_db_add_application(afdb, optarg)) {
- ERROR("can't add application %s", optarg);
- return 1;
- }
- break;
- }
- }
-
- /* update the database */
- if (afm_db_update_applications(afdb)) {
- ERROR("afm_update_applications failed");
- return 1;
- }
+ /* init sessionid */
+ asprintf(&sessionid, "%d-%s", (int)getuid(), appname);
/* daemonize if requested */
if (daemon && daemonize()) {
ERROR("can't create event loop");
return 1;
}
- rc = open_bus(&sysbus, 0, sys_bus_addr);
- if (rc < 0) {
- ERROR("can't create system bus");
- return 1;
- }
- rc = sd_bus_attach_event(sysbus, evloop, 0);
- if (rc < 0) {
- ERROR("can't attach system bus to event loop");
- return 1;
- }
rc = open_bus(&usrbus, 1, usr_bus_addr);
if (rc < 0) {
ERROR("can't create user bus");
return 1;
}
- /* connects to the system bus */
- system_bus = create_jbus(sysbus, AFM_SYSTEM_DBUS_PATH);
- if (!system_bus) {
- ERROR("create_jbus failed for system");
- return 1;
- }
-
- /* observe signals of system */
- if(jbus_on_signal_j(system_bus, "changed", on_signal_changed, NULL)) {
- ERROR("adding signal observer failed");
+ /* connect to framework */
+ if (!try_connect_pws()) {
+ ERROR("connection to %s failed: %m\n", uri);
return 1;
}
}
/* init services */
- if (jbus_add_service_j(user_bus, "runnables", on_runnables, NULL)
- || jbus_add_service_j(user_bus, "detail", on_detail, NULL)
- || jbus_add_service_j(user_bus, "start", on_start, NULL)
- || jbus_add_service_j(user_bus, "terminate", on_terminate, NULL)
- || jbus_add_service_j(user_bus, "pause", on_pause, NULL)
- || jbus_add_service_j(user_bus, "resume", on_resume, NULL)
- || jbus_add_service_j(user_bus, "stop", on_stop, NULL)
- || jbus_add_service_j(user_bus, "continue", on_continue, NULL)
- || jbus_add_service_j(user_bus, "runners", on_runners, NULL)
- || jbus_add_service_j(user_bus, "state", on_state, NULL)
-#if defined(EXPLICIT_CALL)
- || jbus_add_service_s(user_bus, "install", on_install, NULL)
- || jbus_add_service_s(user_bus, "uninstall", on_uninstall, NULL)
-#else
- || jbus_add_service_s(user_bus, "install", (void (*)(struct sd_bus_message *, const char *, void *))propagate, "install")
- || jbus_add_service_s(user_bus, "uninstall", (void (*)(struct sd_bus_message *, const char *, void *))propagate, "uninstall")
-#endif
- ) {
- ERROR("adding services failed");
- return 1;
+ for (iter = methods ; *iter ; iter ++) {
+ if (jbus_add_service_j(user_bus, *iter, propagate, (void*)*iter)) {
+ ERROR("adding services failed");
+ return 1;
+ }
}
/* start servicing */
return 0;
}
-
-
-
-
-
-
-
-
-
-
-
-
-
-
Code Review / src / app-framework-main.git / blobdiff