IF(HAVE_MAGIC_H)
IF(HAVE_LIBMAGIC_SO)
SET(HAVE_LIBMAGIC "1")
+ SET(LIBMAGIC_LDFLAGS -lmagic)
ENDIF(HAVE_LIBMAGIC_SO)
ENDIF(HAVE_MAGIC_H)
-IF(NOT HAVE_LIBMAGIC)
- MESSAGE(FATAL_ERROR "\"magic.h\" or \"libmagic.so\" missing.
- Please install the \"file-devel\" or \"libmagic-dev\" package !")
-ENDIF(NOT HAVE_LIBMAGIC)
-ADD_DEFINITIONS(-DUSE_MAGIC_MIME_TYPE)
-
-PKG_CHECK_MODULES(libsystemd REQUIRED libsystemd>=222)
-PKG_CHECK_MODULES(libmicrohttpd REQUIRED libmicrohttpd>=0.9.55)
-PKG_CHECK_MODULES(openssl REQUIRED openssl)
-PKG_CHECK_MODULES(uuid REQUIRED uuid)
+PKG_CHECK_MODULES(libsystemd libsystemd>=222)
+PKG_CHECK_MODULES(libmicrohttpd libmicrohttpd>=0.9.55)
+PKG_CHECK_MODULES(openssl openssl)
+PKG_CHECK_MODULES(uuid uuid)
PKG_CHECK_MODULES(cynara cynara-client)
IF(AGL_DEVEL)
ADD_DEFINITIONS(-DBACKEND_PERMISSION_IS_CYNARA)
ENDIF(cynara_FOUND)
+IF(HAVE_LIBMAGIC AND libsystemd_FOUND AND libmicrohttpd_FOUND AND openssl_FOUND AND uuid_FOUND)
+ SET(WITH_BINDER TRUE)
+ ADD_DEFINITIONS(-DUSE_MAGIC_MIME_TYPE)
+ELSE()
+ IF(NOT HAVE_LIBMAGIC)
+ MESSAGE(WARNING "\"magic.h\" or \"libmagic.so\" missing.
+ Please install the \"file-devel\" or \"libmagic-dev\" package !")
+ ENDIF(NOT HAVE_LIBMAGIC)
+ IF(NOT libsystemd_FOUND)
+ MESSAGE(WARNING "Dependency to 'libsystemd' is missing")
+ ENDIF()
+ IF(NOT libmicrohttpd_FOUND)
+ MESSAGE(WARNING "Dependency to 'libmicrohttpd' is missing")
+ ENDIF()
+ IF(NOT openssl_FOUND)
+ MESSAGE(WARNING "Dependency to 'openssl' is missing")
+ ENDIF()
+ IF(NOT uuid_FOUND)
+ MESSAGE(WARNING "Dependency to 'uuid' is missing")
+ ENDIF()
+ IF(NOT ALLOW_NO_BINDER)
+ MESSAGE(FATAL_ERROR "Can't compile the binder, either define ALLOW_NO_BINDER or install dependencies")
+ ENDIF()
+ SET(WITH_BINDER FALSE)
+ENDIF()
+
ADD_DEFINITIONS(-DAFB_VERSION="${PROJECT_VERSION}")
INCLUDE_DIRECTORIES(
${uuid_LDFLAGS}
${openssl_LDFLAGS}
${cynara_LDFLAGS}
- -lmagic
+ ${LIBMAGIC_LDFLAGS}
-ldl
-lrt
)
###########################################################################
# activates the monitoring by default
-if(INCLUDE_MONITORING)
+if(INCLUDE_MONITORING AND WITH_BINDER)
add_definitions(-DWITH_MONITORING_OPTION)
INSTALL(DIRECTORY
${CMAKE_CURRENT_SOURCE_DIR}/test/monitoring
# limitations under the License.
###########################################################################
+IF(WITH_BINDER)
ADD_SUBDIRECTORY(intrinsics)
ADD_SUBDIRECTORY(samples)
ADD_SUBDIRECTORY(tutorial)
+ENDIF(WITH_BINDER)
+
int i, rc;
for (i = 0; apis[i] ; i++) {
- rc = afb_dynapi_new_api(dynapi, apis[i], NULL, build_api, (void*)apis[i]);
+ rc = afb_dynapi_new_api(dynapi, apis[i], NULL, 0, build_api, (void*)apis[i]);
if (rc < 0)
AFB_DYNAPI_ERROR(dynapi, "can't create API %s", apis[i]);
}
struct afb_req (*unstore_req)(void*closure, struct afb_stored_req *sreq);
int (*require_api)(void*closure, const char *name, int initialized);
int (*rename_api)(void*closure, const char *name);
- int (*new_api)(void *closure, const char *api, const char *info, int (*preinit)(void*, struct afb_dynapi *), void *preinit_closure);
+ int (*new_api)(void *closure, const char *api, const char *info, int noconcurrency, int (*preinit)(void*, struct afb_dynapi *), void *preinit_closure);
};
/*
struct afb_daemon daemon,
const char *api,
const char *info,
+ int noconcurrency,
int (*preinit)(void*, struct afb_dynapi *),
void *closure)
{
- return daemon.itf->new_api(daemon.closure, api, info, preinit, closure);
+ return daemon.itf->new_api(daemon.closure, api, info, noconcurrency, preinit, closure);
}
static inline int afb_daemon_new_api_v2(
const char *api,
const char *info,
+ int noconcurrency,
int (*preinit)(void*, struct afb_dynapi *),
void *closure)
{
- return afb_get_daemon_v2().itf->new_api(afb_get_daemon_v2().closure, api, info, preinit, closure);
+ return afb_get_daemon_v2().itf->new_api(afb_get_daemon_v2().closure, api, info, noconcurrency, preinit, closure);
}
void *dynapi,
const char *api,
const char *info,
+ int noconcurrency,
int (*preinit)(void*, struct afb_dynapi *),
void *closure);
struct afb_dynapi *dynapi,
const char *api,
const char *info,
+ int noconcurrency,
int (*preinit)(void*, struct afb_dynapi *),
void *closure)
{
- return dynapi->itf->api_new_api(dynapi, api, info, preinit, closure);
+ return dynapi->itf->api_new_api(dynapi, api, info, noconcurrency, preinit, closure);
}
static inline int afb_dynapi_set_verbs_v2(
INCLUDE(FindPkgConfig)
ADD_SUBDIRECTORY(genskel)
+
+IF(WITH_BINDER)
+###########################################
+
ADD_SUBDIRECTORY(tests)
ADD_DEFINITIONS(-DBINDING_INSTALL_DIR="${binding_install_dir}")
###########################################
# build and install libafbwsc
###########################################
-ADD_LIBRARY(afbwsc SHARED afb-ws.c afb-ws-client.c afb-wsj1.c websock.c afb-proto-ws.c)
+ADD_LIBRARY(afbwsc SHARED afb-ws.c afb-ws-client.c afb-wsj1.c websock.c afb-proto-ws.c jobs-fake.c)
SET_TARGET_PROPERTIES(afbwsc PROPERTIES
VERSION ${LIBAFBWSC_VERSION}
SOVERSION ${LIBAFBWSC_SOVERSION})
TARGET_LINK_LIBRARIES(afbwsc
${libsystemd_LDFLAGS}
${json-c_LDFLAGS}
+ -lpthread
-Wl,--version-script=${CMAKE_CURRENT_SOURCE_DIR}/export-afbwsc.map
-Wl,--as-needed
-Wl,--gc-sections
INSTALL(TARGETS afb-client-demo
RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR})
-
+###########################################
+ELSE(WITH_BINDER)
+ MESSAGE(WARNING "NOT compiling the binder! but tools are compiled")
+ENDIF(WITH_BINDER)
.describe = describe_cb
};
-int afb_api_dyn_add(struct afb_apiset *apiset, const char *name, const char *info, int (*preinit)(void*, struct afb_dynapi*), void *closure)
+int afb_api_dyn_add(struct afb_apiset *apiset, const char *name, const char *info, int noconcurrency, int (*preinit)(void*, struct afb_dynapi*), void *closure)
{
int rc;
struct afb_api_dyn *dynapi;
/* records the binding */
afb_api.closure = dynapi;
afb_api.itf = &dyn_api_itf;
- afb_api.group = NULL;
+ afb_api.group = noconcurrency ? dynapi : NULL;
if (afb_apiset_add(apiset, afb_export_apiname(dynapi->export), afb_api) < 0) {
ERROR("dynamic api %s can't be registered to set %s, ABORTING it!",
afb_export_apiname(dynapi->export),
struct afb_apiset *apiset,
const char *name,
const char *info,
+ int noconcurrency,
int (*preinit)(void*, struct afb_dynapi*),
void *closure);
int created;
struct afb_session *session;
- session = afb_session_get (uuid, &created);
+ session = afb_session_get (uuid, AFB_SESSION_TIMEOUT_DEFAULT, &created);
if (session == NULL)
return -1;
init_context(context, session, token);
#pragma once
struct afb_event;
+struct afb_eventid;
struct afb_evtid;
struct afb_session;
struct json_object;
static struct afb_event event_make_cb(void *closure, const char *name)
{
struct afb_eventid *eventid = eventid_make_cb(closure, name);
- return (struct afb_event){ .itf = eventid ? eventid->itf : NULL, .closure = eventid };
+ return afb_evt_event_from_evtid(afb_evt_eventid_to_evtid(eventid));
}
static int event_broadcast_cb(void *closure, const char *name, struct json_object *object)
void *closure,
const char *api,
const char *info,
+ int noconcurrency,
int (*preinit)(void*, struct afb_dynapi *),
void *preinit_closure)
{
struct afb_export *export = closure;
- return afb_api_dyn_add(export->apiset, api, info, preinit, preinit_closure);
+ return afb_api_dyn_add(export->apiset, api, info, noconcurrency, preinit, preinit_closure);
}
/**********************************************
void *closure,
const char *api,
const char *info,
+ int noconcurrency,
int (*preinit)(void*, struct afb_dynapi *),
void *preinit_closure)
{
/* TODO */
- return api_new_api_cb(closure, api, info, preinit, preinit_closure);
+ return api_new_api_cb(closure, api, info, noconcurrency, preinit, preinit_closure);
}
/**********************************************
* vectors
/* session shared with other exports */
if (common_session == NULL) {
- common_session = afb_session_create (NULL, 0);
+ common_session = afb_session_create (0);
if (common_session == NULL)
return NULL;
}
int afb_export_unshare_session(struct afb_export *export)
{
if (export->session == common_session) {
- export->session = afb_session_create (NULL, 0);
+ export->session = afb_session_create (0);
if (export->session)
afb_session_unref(common_session);
else {
#include "afb-ws.h"
#include "afb-msg-json.h"
#include "afb-proto-ws.h"
+#include "jobs.h"
struct afb_proto_ws;
void (*on_hangup)(void *closure);
};
+/******************* streaming objects **********************************/
+
+#define WRITEBUF_COUNT_MAX 32
+struct writebuf
+{
+ struct iovec iovec[WRITEBUF_COUNT_MAX];
+ uint32_t uints[WRITEBUF_COUNT_MAX];
+ int count;
+};
+
+struct readbuf
+{
+ char *base, *head, *end;
+};
+
+struct binary
+{
+ struct afb_proto_ws *protows;
+ struct readbuf rb;
+};
+
/******************* common useful tools **********************************/
/**
/******************* serialisation part **********************************/
-struct readbuf
-{
- char *base, *head, *end;
-};
-
-#define WRITEBUF_COUNT_MAX 32
-struct writebuf
-{
- struct iovec iovec[WRITEBUF_COUNT_MAX];
- uint32_t uints[WRITEBUF_COUNT_MAX];
- int count;
-};
-
static char *readbuf_get(struct readbuf *rb, uint32_t length)
{
char *before = rb->head;
{
int rc = -1;
struct writebuf wb = { .count = 0 };
+ struct afb_proto_ws *protows = call->protows;
if (writebuf_char(&wb, CHAR_FOR_ANSWER_SUCCESS)
&& writebuf_uint32(&wb, call->callid)
&& writebuf_string(&wb, info ?: "")
&& writebuf_object(&wb, obj)) {
- rc = afb_ws_binary_v(call->protows->ws, wb.iovec, wb.count);
+ pthread_mutex_lock(&protows->mutex);
+ rc = afb_ws_binary_v(protows->ws, wb.iovec, wb.count);
+ pthread_mutex_unlock(&protows->mutex);
if (rc >= 0) {
rc = 0;
goto success;
{
int rc = -1;
struct writebuf wb = { .count = 0 };
+ struct afb_proto_ws *protows = call->protows;
if (writebuf_char(&wb, CHAR_FOR_ANSWER_FAIL)
&& writebuf_uint32(&wb, call->callid)
&& writebuf_string(&wb, status)
&& writebuf_string(&wb, info ? : "")) {
- rc = afb_ws_binary_v(call->protows->ws, wb.iovec, wb.count);
+ pthread_mutex_lock(&protows->mutex);
+ rc = afb_ws_binary_v(protows->ws, wb.iovec, wb.count);
+ pthread_mutex_unlock(&protows->mutex);
if (rc >= 0) {
rc = 0;
goto success;
sc->callback = callback;
sc->closure = cb_closure;
- pthread_mutex_unlock(&protows->mutex);
+ pthread_mutex_lock(&protows->mutex);
sc->subcallid = ptr2id(sc);
do {
sc->subcallid++;
&& writebuf_string(&wb, api)
&& writebuf_string(&wb, verb)
&& writebuf_object(&wb, args)) {
+ pthread_mutex_lock(&protows->mutex);
rc = afb_ws_binary_v(protows->ws, wb.iovec, wb.count);
+ pthread_mutex_unlock(&protows->mutex);
if (rc >= 0) {
rc = 0;
goto success;
{
int rc = -1;
struct writebuf wb = { .count = 0 };
+ struct afb_proto_ws *protows = call->protows;
if (writebuf_char(&wb, CHAR_FOR_EVT_SUBSCRIBE)
&& writebuf_uint32(&wb, call->callid)
&& writebuf_uint32(&wb, (uint32_t)event_id)
&& writebuf_string(&wb, event_name)) {
- rc = afb_ws_binary_v(call->protows->ws, wb.iovec, wb.count);
+ pthread_mutex_lock(&protows->mutex);
+ rc = afb_ws_binary_v(protows->ws, wb.iovec, wb.count);
+ pthread_mutex_unlock(&protows->mutex);
if (rc >= 0) {
rc = 0;
goto success;
{
int rc = -1;
struct writebuf wb = { .count = 0 };
+ struct afb_proto_ws *protows = call->protows;
if (writebuf_char(&wb, CHAR_FOR_EVT_UNSUBSCRIBE)
&& writebuf_uint32(&wb, call->callid)
&& writebuf_uint32(&wb, (uint32_t)event_id)
&& writebuf_string(&wb, event_name)) {
- rc = afb_ws_binary_v(call->protows->ws, wb.iovec, wb.count);
+ pthread_mutex_lock(&protows->mutex);
+ rc = afb_ws_binary_v(protows->ws, wb.iovec, wb.count);
+ pthread_mutex_unlock(&protows->mutex);
if (rc >= 0) {
rc = 0;
goto success;
/******************* client part **********************************/
/* search a memorized call */
-static struct client_call *client_call_search(struct afb_proto_ws *protows, uint32_t callid)
+static struct client_call *client_call_search_locked(struct afb_proto_ws *protows, uint32_t callid)
{
struct client_call *call;
return call;
}
+static struct client_call *client_call_search_unlocked(struct afb_proto_ws *protows, uint32_t callid)
+{
+ struct client_call *result;
+
+ pthread_mutex_lock(&protows->mutex);
+ result = client_call_search_locked(protows, callid);
+ pthread_mutex_unlock(&protows->mutex);
+ return result;
+}
+
/* free and release the memorizing call */
static void client_call_destroy(struct client_call *call)
{
struct client_call **prv;
+ struct afb_proto_ws *protows = call->protows;
+ pthread_mutex_lock(&protows->mutex);
prv = &call->protows->calls;
while (*prv != NULL) {
if (*prv == call) {
}
prv = &(*prv)->next;
}
+ pthread_mutex_unlock(&protows->mutex);
free(call);
}
}
/* get the call */
- *call = client_call_search(protows, callid);
+ *call = client_call_search_unlocked(protows, callid);
if (*call == NULL) {
return 0;
}
if (!client_msg_call_get(protows, rb, &call))
return;
+
if (readbuf_string(rb, &status, NULL) && readbuf_string(rb, &info, NULL)) {
protows->client_itf->on_reply_fail(protows->closure, call->request, status, info);
{
struct writebuf wb = { .count = 0 };
char ie = status < 0;
+ int rc;
- return -!(writebuf_char(&wb, CHAR_FOR_SUBCALL_REPLY)
+ if (writebuf_char(&wb, CHAR_FOR_SUBCALL_REPLY)
&& writebuf_uint32(&wb, subcallid)
&& writebuf_char(&wb, ie)
- && writebuf_object(&wb, object)
- && afb_ws_binary_v(protows->ws, wb.iovec, wb.count) >= 0);
+ && writebuf_object(&wb, object)) {
+ pthread_mutex_lock(&protows->mutex);
+ rc = afb_ws_binary_v(protows->ws, wb.iovec, wb.count);
+ pthread_mutex_unlock(&protows->mutex);
+ if (rc >= 0)
+ return 0;
+ }
+ return -1;
}
/* callback for subcall reply */
struct json_object *object;
if (readbuf_uint32(rb, &descid)) {
+ pthread_mutex_lock(&protows->mutex);
prv = &protows->describes;
while ((desc = *prv) && desc->descid != descid)
prv = &desc->next;
- if (desc) {
+ if (!desc)
+ pthread_mutex_unlock(&protows->mutex);
+ else {
*prv = desc->next;
+ pthread_mutex_unlock(&protows->mutex);
if (!readbuf_object(rb, &object))
object = NULL;
desc->callback(desc->closure, object);
}
/* callback when receiving binary data */
-static void client_on_binary(void *closure, char *data, size_t size)
+static void client_on_binary_job(int sig, void *closure)
{
- struct afb_proto_ws *protows;
- struct readbuf rb;
-
- rb.base = data;
- if (size > 0) {
- rb.head = data;
- rb.end = data + size;
- protows = closure;
+ struct binary *binary = closure;
- pthread_mutex_lock(&protows->mutex);
- switch (*rb.head++) {
+ if (!sig) {
+ switch (*binary->rb.head++) {
case CHAR_FOR_ANSWER_SUCCESS: /* success */
- client_on_reply_success(protows, &rb);
+ client_on_reply_success(binary->protows, &binary->rb);
break;
case CHAR_FOR_ANSWER_FAIL: /* fail */
- client_on_reply_fail(protows, &rb);
+ client_on_reply_fail(binary->protows, &binary->rb);
break;
case CHAR_FOR_EVT_BROADCAST: /* broadcast */
- client_on_event_broadcast(protows, &rb);
+ client_on_event_broadcast(binary->protows, &binary->rb);
break;
case CHAR_FOR_EVT_ADD: /* creates the event */
- client_on_event_create(protows, &rb);
+ client_on_event_create(binary->protows, &binary->rb);
break;
case CHAR_FOR_EVT_DEL: /* removes the event */
- client_on_event_remove(protows, &rb);
+ client_on_event_remove(binary->protows, &binary->rb);
break;
case CHAR_FOR_EVT_PUSH: /* pushs the event */
- client_on_event_push(protows, &rb);
+ client_on_event_push(binary->protows, &binary->rb);
break;
case CHAR_FOR_EVT_SUBSCRIBE: /* subscribe event for a request */
- client_on_event_subscribe(protows, &rb);
+ client_on_event_subscribe(binary->protows, &binary->rb);
break;
case CHAR_FOR_EVT_UNSUBSCRIBE: /* unsubscribe event for a request */
- client_on_event_unsubscribe(protows, &rb);
+ client_on_event_unsubscribe(binary->protows, &binary->rb);
break;
case CHAR_FOR_SUBCALL_CALL: /* subcall */
- client_on_subcall(protows, &rb);
+ client_on_subcall(binary->protows, &binary->rb);
break;
case CHAR_FOR_DESCRIPTION: /* description */
- client_on_description(protows, &rb);
+ client_on_description(binary->protows, &binary->rb);
break;
default: /* unexpected message */
/* TODO: close the connection */
break;
}
- pthread_mutex_unlock(&protows->mutex);
}
- free(rb.base);
+ free(binary->rb.base);
+ free(binary);
+}
+
+/* callback when receiving binary data */
+static void client_on_binary(void *closure, char *data, size_t size)
+{
+ int rc;
+ struct binary *binary;
+
+ if (size) {
+ binary = malloc(sizeof *binary);
+ if (!binary) {
+ errno = ENOMEM;
+ } else {
+ binary->protows = closure;
+ binary->rb.base = data;
+ binary->rb.head = data;
+ binary->rb.end = data + size;
+ rc = jobs_queue(NULL, 0, client_on_binary_job, binary);
+ if (rc >= 0)
+ return;
+ free(binary);
+ }
+ }
+ free(data);
}
int afb_proto_ws_client_call(
/* init call data */
pthread_mutex_lock(&protows->mutex);
call->callid = ptr2id(call);
- while(client_call_search(protows, call->callid) != NULL)
+ while(client_call_search_locked(protows, call->callid) != NULL)
call->callid++;
call->protows = protows;
call->next = protows->calls;
protows->calls = call;
+ pthread_mutex_unlock(&protows->mutex);
/* creates the call message */
if (!writebuf_char(&wb, CHAR_FOR_CALL)
}
/* send */
+ pthread_mutex_lock(&protows->mutex);
rc = afb_ws_binary_v(protows->ws, wb.iovec, wb.count);
+ pthread_mutex_unlock(&protows->mutex);
if (rc >= 0) {
rc = 0;
goto end;
clean:
client_call_destroy(call);
end:
- pthread_mutex_unlock(&protows->mutex);
return rc;
}
desc->protows = protows;
desc->next = protows->describes;
protows->describes = desc;
- pthread_mutex_unlock(&protows->mutex);
/* send */
if (writebuf_char(&wb, CHAR_FOR_DESCRIBE)
&& writebuf_uint32(&wb, desc->descid)
- && afb_ws_binary_v(protows->ws, wb.iovec, wb.count) >= 0)
+ && afb_ws_binary_v(protows->ws, wb.iovec, wb.count) >= 0) {
+ pthread_mutex_unlock(&protows->mutex);
return 0;
+ }
- pthread_mutex_lock(&protows->mutex);
d = protows->describes;
if (d == desc)
protows->describes = desc->next;
if (d)
d->next = desc->next;
}
- free(desc);
pthread_mutex_unlock(&protows->mutex);
+ free(desc);
error:
/* TODO? callback(closure, NULL); */
return -1;
static int server_send_description(struct afb_proto_ws *protows, uint32_t descid, struct json_object *descobj)
{
+ int rc;
struct writebuf wb = { .count = 0 };
- return -!(writebuf_char(&wb, CHAR_FOR_DESCRIPTION)
- && writebuf_uint32(&wb, descid)
- && writebuf_object(&wb, descobj)
- && afb_ws_binary_v(protows->ws, wb.iovec, wb.count) >= 0);
+ if (writebuf_char(&wb, CHAR_FOR_DESCRIPTION)
+ && writebuf_uint32(&wb, descid)
+ && writebuf_object(&wb, descobj)) {
+ pthread_mutex_lock(&protows->mutex);
+ rc = afb_ws_binary_v(protows->ws, wb.iovec, wb.count);
+ pthread_mutex_unlock(&protows->mutex);
+ if (rc >= 0)
+ return 0;
+ }
+ return -1;
}
int afb_proto_ws_describe_put(struct afb_proto_ws_describe *describe, struct json_object *description)
{
int rc = server_send_description(describe->protows, describe->descid, description);
- afb_proto_ws_addref(describe->protows);
+ afb_proto_ws_unref(describe->protows);
free(describe);
return rc;
}
}
/* callback when receiving binary data */
-static void server_on_binary(void *closure, char *data, size_t size)
+static void server_on_binary_job(int sig, void *closure)
{
- struct afb_proto_ws *protows;
- struct readbuf rb;
-
- rb.base = data;
- if (size > 0) {
- rb.head = data;
- rb.end = data + size;
- protows = closure;
+ struct binary *binary = closure;
- switch (*rb.head++) {
+ if (!sig) {
+ switch (*binary->rb.head++) {
case CHAR_FOR_CALL:
- server_on_call(protows, &rb);
+ server_on_call(binary->protows, &binary->rb);
break;
case CHAR_FOR_SUBCALL_REPLY:
- server_on_subcall_reply(protows, &rb);
+ server_on_subcall_reply(binary->protows, &binary->rb);
break;
case CHAR_FOR_DESCRIBE:
- server_on_describe(protows, &rb);
+ server_on_describe(binary->protows, &binary->rb);
break;
default: /* unexpected message */
/* TODO: close the connection */
break;
}
}
- free(rb.base);
+ free(binary->rb.base);
+ free(binary);
+}
+
+static void server_on_binary(void *closure, char *data, size_t size)
+{
+ int rc;
+ struct binary *binary;
+
+ if (size) {
+ binary = malloc(sizeof *binary);
+ if (!binary) {
+ errno = ENOMEM;
+ } else {
+ binary->protows = closure;
+ binary->rb.base = data;
+ binary->rb.head = data;
+ binary->rb.end = data + size;
+ rc = jobs_queue(NULL, 0, server_on_binary_job, binary);
+ if (rc >= 0)
+ return;
+ free(binary);
+ }
+ }
+ free(data);
}
/******************* server part: manage events **********************************/
static int server_event_send(struct afb_proto_ws *protows, char order, const char *event_name, int event_id, struct json_object *data)
{
struct writebuf wb = { .count = 0 };
+ int rc;
- return -!(writebuf_char(&wb, order)
- && (order == CHAR_FOR_EVT_BROADCAST || writebuf_uint32(&wb, event_id))
- && writebuf_string(&wb, event_name)
- && (order == CHAR_FOR_EVT_ADD || order == CHAR_FOR_EVT_DEL || writebuf_object(&wb, data))
- && afb_ws_binary_v(protows->ws, wb.iovec, wb.count) >= 0);
+ if (writebuf_char(&wb, order)
+ && (order == CHAR_FOR_EVT_BROADCAST || writebuf_uint32(&wb, event_id))
+ && writebuf_string(&wb, event_name)
+ && (order == CHAR_FOR_EVT_ADD || order == CHAR_FOR_EVT_DEL || writebuf_object(&wb, data))) {
+ pthread_mutex_lock(&protows->mutex);
+ rc = afb_ws_binary_v(protows->ws, wb.iovec, wb.count);
+ pthread_mutex_unlock(&protows->mutex);
+ if (rc >= 0)
+ return 0;
+ }
+ return -1;
}
int afb_proto_ws_server_event_create(struct afb_proto_ws *protows, const char *event_name, int event_id)
}
if (protows->fd >= 0) {
+ close(protows->fd);
protows->fd = -1;
if (protows->on_hangup)
protows->on_hangup(protows->closure);
#include "afb-session.h"
#include "verbose.h"
-#define COOKEYCOUNT 8
-#define COOKEYMASK (COOKEYCOUNT - 1)
+#define SIZEUUID 37
+#define HEADCOUNT 16
+#define COOKEYCOUNT 8
+#define COOKEYMASK (COOKEYCOUNT - 1)
-#define NOW (time(NULL))
+#define _MAXEXP_ ((time_t)(~(time_t)0))
+#define _MAXEXP2_ ((time_t)((((unsigned long long)_MAXEXP_) >> 1)))
+#define MAX_EXPIRATION (_MAXEXP_ >= 0 ? _MAXEXP_ : _MAXEXP2_)
+#define NOW (time(NULL))
struct cookie
{
struct afb_session
{
+ struct afb_session *next; /* link to the next */
unsigned refcount;
int timeout;
- time_t expiration; // expiration time of the token
- time_t access;
+ time_t expiration; // expiration time of the token
pthread_mutex_t mutex;
- char uuid[37]; // long term authentication of remote client
- char token[37]; // short term authentication of remote client
struct cookie *cookies[COOKEYCOUNT];
+ char autoclose;
+ char idx;
+ char uuid[SIZEUUID]; // long term authentication of remote client
+ char token[SIZEUUID]; // short term authentication of remote client
};
// Session UUID are store in a simple array [for 10 sessions this should be enough]
static struct {
- pthread_mutex_t mutex; // declare a mutex to protect hash table
- struct afb_session **store; // sessions store
- int count; // current number of sessions
+ pthread_mutex_t mutex; // declare a mutex to protect hash table
+ struct afb_session *heads[HEADCOUNT]; // sessions
+ int count; // current number of sessions
int max;
int timeout;
- char initok[37];
+ char initok[SIZEUUID];
} sessions;
-/**
- * Get the index of the 'key' in the cookies array.
- * @param key the key to scan
- * @return the index of the list for key within cookies
- */
-static int cookeyidx(const void *key)
-{
- intptr_t x = (intptr_t)key;
- unsigned r = (unsigned)((x >> 5) ^ (x >> 15));
- return r & COOKEYMASK;
-}
-
/* generate a uuid */
-static void new_uuid(char uuid[37])
+static void new_uuid(char uuid[SIZEUUID])
{
uuid_t newuuid;
uuid_generate(newuuid);
}
// Free context [XXXX Should be protected again memory abort XXXX]
-static void free_data (struct afb_session *session)
+static void close_session(struct afb_session *session)
{
int idx;
- struct cookie *cookie, *next;
+ struct cookie *cookie;
- // free cookies
+ /* free cookies */
for (idx = 0 ; idx < COOKEYCOUNT ; idx++) {
- cookie = session->cookies[idx];
- session->cookies[idx] = NULL;
- while (cookie != NULL) {
- next = cookie->next;
+ while ((cookie = session->cookies[idx])) {
+ session->cookies[idx] = cookie->next;
if (cookie->freecb != NULL)
cookie->freecb(cookie->value);
free(cookie);
- cookie = next;
}
}
}
+/* tiny hash function inspired from pearson */
+static int pearson4(const char *text)
+{
+ static uint8_t T[16] = {
+ 4, 1, 6, 0, 9, 14, 11, 5,
+ 2, 3, 12, 15, 10, 7, 8, 13
+ };
+ uint8_t r, c;
+
+ for (r = 0; (c = (uint8_t)*text) ; text++) {
+ r = T[r ^ (15 & c)];
+ r = T[r ^ (c >> 4)];
+ }
+ return r; // % HEADCOUNT;
+}
+
// Create a new store in RAM, not that is too small it will be automatically extended
void afb_session_init (int max_session_count, int timeout, const char *initok)
{
- // let's create as store as hashtable does not have any
- sessions.store = calloc (1 + (unsigned)max_session_count, sizeof *sessions.store);
pthread_mutex_init(&sessions.mutex, NULL);
sessions.max = max_session_count;
sessions.timeout = timeout;
if (initok == NULL)
/* without token, a secret is made to forbid creation of sessions */
new_uuid(sessions.initok);
- else if (strlen(initok) < sizeof(sessions.store[0]->token))
+ else if (strlen(initok) < sizeof sessions.initok)
strcpy(sessions.initok, initok);
else {
- ERROR("initial token '%s' too long (max length 36)", initok);
+ ERROR("initial token '%s' too long (max length %d)", initok, ((int)(sizeof sessions.initok)) - 1);
exit(1);
}
}
return sessions.initok;
}
-static struct afb_session *search (const char* uuid)
+static struct afb_session *search (const char* uuid, int idx)
{
- int idx;
struct afb_session *session;
- assert (uuid != NULL);
-
- pthread_mutex_lock(&sessions.mutex);
-
- for (idx=0; idx < sessions.max; idx++) {
- session = sessions.store[idx];
- if (session && (0 == strcmp (uuid, session->uuid)))
- goto found;
- }
- session = NULL;
+ session = sessions.heads[idx];
+ while (session && strcmp(uuid, session->uuid))
+ session = session->next;
-found:
- pthread_mutex_unlock(&sessions.mutex);
return session;
}
-static int destroy (struct afb_session *session)
+static void destroy (struct afb_session *session)
{
- int idx;
- int status;
+ struct afb_session **prv;
assert (session != NULL);
+ close_session(session);
pthread_mutex_lock(&sessions.mutex);
-
- for (idx=0; idx < sessions.max; idx++) {
- if (sessions.store[idx] == session) {
- sessions.store[idx] = NULL;
+ prv = &sessions.heads[(int)session->idx];
+ while (*prv)
+ if (*prv != session)
+ prv = &((*prv)->next);
+ else {
+ *prv = session->next;
sessions.count--;
- status = 1;
- goto deleted;
+ pthread_mutex_destroy(&session->mutex);
+ free(session);
+ break;
}
- }
- status = 0;
-deleted:
pthread_mutex_unlock(&sessions.mutex);
- return status;
}
-static int add (struct afb_session *session)
+// Loop on every entry and remove old context sessions.hash
+static time_t cleanup ()
{
+ struct afb_session *session, *next;
int idx;
- int status;
-
- assert (session != NULL);
-
- pthread_mutex_lock(&sessions.mutex);
+ time_t now;
- for (idx=0; idx < sessions.max; idx++) {
- if (NULL == sessions.store[idx]) {
- sessions.store[idx] = session;
- sessions.count++;
- status = 1;
- goto added;
+ // Loop on Sessions Table and remove anything that is older than timeout
+ now = NOW;
+ for (idx = 0 ; idx < HEADCOUNT; idx++) {
+ session = sessions.heads[idx];
+ while (session) {
+ next = session->next;
+ if (session->expiration < now)
+ afb_session_close(session);
+ session = next;
}
}
- status = 0;
-added:
- pthread_mutex_unlock(&sessions.mutex);
- return status;
+ return now;
}
-// Check if context timeout or not
-static int is_expired (struct afb_session *ctx, time_t now)
+static void update_timeout(struct afb_session *session, time_t now, int timeout)
{
- assert (ctx != NULL);
- return ctx->expiration < now;
+ time_t expiration;
+
+ /* compute expiration */
+ if (timeout == AFB_SESSION_TIMEOUT_INFINITE)
+ expiration = MAX_EXPIRATION;
+ else {
+ if (timeout == AFB_SESSION_TIMEOUT_DEFAULT)
+ expiration = now + sessions.timeout;
+ else
+ expiration = now + timeout;
+ if (expiration < 0)
+ expiration = MAX_EXPIRATION;
+ }
+
+ /* record the values */
+ session->timeout = timeout;
+ session->expiration = expiration;
}
-// Check if context is active or not
-static int is_active (struct afb_session *ctx, time_t now)
+static void update_expiration(struct afb_session *session, time_t now)
{
- assert (ctx != NULL);
- return ctx->uuid[0] != 0 && ctx->expiration >= now;
+ update_timeout(session, now, session->timeout);
}
-// Loop on every entry and remove old context sessions.hash
-static void cleanup (time_t now)
+static struct afb_session *add_session (const char *uuid, int timeout, time_t now, int idx)
{
- struct afb_session *ctx;
- long idx;
+ struct afb_session *session;
- // Loop on Sessions Table and remove anything that is older than timeout
- for (idx=0; idx < sessions.max; idx++) {
- ctx = sessions.store[idx];
- if (ctx != NULL && is_expired(ctx, now)) {
- afb_session_close (ctx);
- }
+ /* check arguments */
+ if (!AFB_SESSION_TIMEOUT_IS_VALID(timeout)
+ || (uuid && strlen(uuid) >= sizeof session->uuid)) {
+ errno = EINVAL;
+ return NULL;
}
-}
-static struct afb_session *make_session (const char *uuid, int timeout, time_t now)
-{
- struct afb_session *session;
+ /* check session count */
+ if (sessions.count >= sessions.max) {
+ errno = EBUSY;
+ return NULL;
+ }
/* allocates a new one */
session = calloc(1, sizeof *session);
if (session == NULL) {
errno = ENOMEM;
- goto error;
- }
- pthread_mutex_init(&session->mutex, NULL);
-
- /* generate the uuid */
- if (uuid == NULL) {
- new_uuid(session->uuid);
- } else {
- if (strlen(uuid) >= sizeof session->uuid) {
- errno = EINVAL;
- goto error2;
- }
- strcpy(session->uuid, uuid);
+ return NULL;
}
- /* init the token */
+ /* initialize */
+ pthread_mutex_init(&session->mutex, NULL);
+ session->refcount = 1;
+ strcpy(session->uuid, uuid);
strcpy(session->token, sessions.initok);
- session->timeout = timeout;
- if (timeout != 0)
- session->expiration = now + timeout;
- else {
- session->expiration = (time_t)(~(time_t)0);
- if (session->expiration < 0)
- session->expiration = (time_t)(((unsigned long long)session->expiration) >> 1);
- }
- if (!add (session)) {
- errno = ENOMEM;
- goto error2;
- }
+ update_timeout(session, now, timeout);
+
+ /* link */
+ session->idx = (char)idx;
+ session->next = sessions.heads[idx];
+ sessions.heads[idx] = session;
+ sessions.count++;
- session->access = now;
- session->refcount = 1;
return session;
+}
+
+/* create a new session for the given timeout */
+static struct afb_session *new_session (int timeout, time_t now)
+{
+ int idx;
+ char uuid[SIZEUUID];
-error2:
- free(session);
-error:
- return NULL;
+ do {
+ new_uuid(uuid);
+ idx = pearson4(uuid);
+ } while(search(uuid, idx));
+ return add_session(uuid, timeout, now, idx);
}
-struct afb_session *afb_session_create (const char *uuid, int timeout)
+/* Creates a new session with 'timeout' */
+struct afb_session *afb_session_create (int timeout)
{
time_t now;
+ struct afb_session *session;
/* cleaning */
- now = NOW;
- cleanup (now);
+ pthread_mutex_lock(&sessions.mutex);
+ now = cleanup();
+ session = new_session(timeout, now);
+ pthread_mutex_unlock(&sessions.mutex);
- /* search for an existing one not too old */
- if (uuid != NULL && search(uuid) != NULL) {
- errno = EEXIST;
- return NULL;
- }
+ return session;
+}
+
+/* Searchs the session of 'uuid' */
+struct afb_session *afb_session_search (const char *uuid)
+{
+ struct afb_session *session;
+
+ /* cleaning */
+ pthread_mutex_lock(&sessions.mutex);
+ cleanup();
+ session = search(uuid, pearson4(uuid));
+ if (session)
+ __atomic_add_fetch(&session->refcount, 1, __ATOMIC_RELAXED);
+ pthread_mutex_unlock(&sessions.mutex);
+ return session;
- return make_session(uuid, timeout, now);
}
-// This function will return exiting session or newly created session
-struct afb_session *afb_session_get (const char *uuid, int *created)
+/* This function will return exiting session or newly created session */
+struct afb_session *afb_session_get (const char *uuid, int timeout, int *created)
{
+ int idx;
struct afb_session *session;
time_t now;
/* cleaning */
- now = NOW;
- cleanup (now);
+ pthread_mutex_lock(&sessions.mutex);
+ now = cleanup();
/* search for an existing one not too old */
- if (uuid != NULL) {
- session = search(uuid);
- if (!created)
- return session;
- if (session != NULL) {
- *created = 0;
- session->access = now;
- session->refcount++;
+ if (!uuid)
+ session = new_session(timeout, now);
+ else {
+ idx = pearson4(uuid);
+ session = search(uuid, idx);
+ if (session) {
+ __atomic_add_fetch(&session->refcount, 1, __ATOMIC_RELAXED);
+ pthread_mutex_unlock(&sessions.mutex);
+ if (created)
+ *created = 0;
return session;
}
+ session = add_session (uuid, timeout, now, idx);
}
+ pthread_mutex_unlock(&sessions.mutex);
if (created)
- *created = 1;
+ *created = !!session;
- return make_session(uuid, sessions.timeout, now);
+ return session;
}
+/* increase the use count on the session */
struct afb_session *afb_session_addref(struct afb_session *session)
{
if (session != NULL)
return session;
}
+/* decrease the use count of the session */
void afb_session_unref(struct afb_session *session)
{
if (session != NULL) {
assert(session->refcount != 0);
if (!__atomic_sub_fetch(&session->refcount, 1, __ATOMIC_RELAXED)) {
- if (session->uuid[0] == 0) {
+ pthread_mutex_lock(&session->mutex);
+ if (session->autoclose || session->uuid[0] == 0)
destroy (session);
- pthread_mutex_destroy(&session->mutex);
- free(session);
- }
+ else
+ pthread_mutex_unlock(&session->mutex);
}
}
}
-// Free Client Session Context
+// close Client Session Context
void afb_session_close (struct afb_session *session)
{
assert(session != NULL);
+ pthread_mutex_lock(&session->mutex);
if (session->uuid[0] != 0) {
session->uuid[0] = 0;
- free_data (session);
- if (session->refcount == 0) {
+ if (session->refcount)
+ close_session(session);
+ else {
destroy (session);
- free(session);
+ return;
}
}
+ pthread_mutex_unlock(&session->mutex);
+}
+
+/* set the autoclose flag */
+void afb_session_set_autoclose(struct afb_session *session, int autoclose)
+{
+ assert(session != NULL);
+ session->autoclose = (char)!!autoclose;
+}
+
+// is the session active?
+int afb_session_is_active (struct afb_session *session)
+{
+ assert(session != NULL);
+ return !!session->uuid[0];
+}
+
+// is the session closed?
+int afb_session_is_closed (struct afb_session *session)
+{
+ assert(session != NULL);
+ return !session->uuid[0];
}
// Sample Generic Ping Debug API
assert(session != NULL);
assert(token != NULL);
- // compare current token with previous one
- if (!is_active (session, NOW))
+ if (!session->uuid[0])
+ return 0;
+
+ if (session->expiration < NOW)
return 0;
if (session->token[0] && strcmp (token, session->token) != 0)
new_uuid(session->token);
// keep track of time for session timeout and further clean up
- if (session->timeout != 0)
- session->expiration = NOW + session->timeout;
+ update_expiration(session, NOW);
}
+/* Returns the uuid of 'session' */
const char *afb_session_uuid (struct afb_session *session)
{
assert(session != NULL);
return session->uuid;
}
+/* Returns the token of 'session' */
const char *afb_session_token (struct afb_session *session)
{
assert(session != NULL);
return session->token;
}
-static struct cookie *cookie_search(struct afb_session *session, const void *key, int *idx)
-{
- struct cookie *cookie;
-
- cookie = session->cookies[*idx = cookeyidx(key)];
- while(cookie != NULL && cookie->key != key)
- cookie = cookie->next;
- return cookie;
-}
-
-static struct cookie *cookie_add(struct afb_session *session, int idx, const void *key, void *value, void (*freecb)(void*))
+/**
+ * Get the index of the 'key' in the cookies array.
+ * @param key the key to scan
+ * @return the index of the list for key within cookies
+ */
+static int cookeyidx(const void *key)
{
- struct cookie *cookie;
-
- cookie = malloc(sizeof *cookie);
- if (!cookie)
- errno = ENOMEM;
- else {
- cookie->key = key;
- cookie->value = value;
- cookie->freecb = freecb;
- cookie->next = session->cookies[idx];
- session->cookies[idx] = cookie;
- }
- return cookie;
+ intptr_t x = (intptr_t)key;
+ unsigned r = (unsigned)((x >> 5) ^ (x >> 15));
+ return r & COOKEYMASK;
}
+/**
+ * Set, get, replace, remove a cookie of 'key' for the 'session'
+ *
+ * The behaviour of this function depends on its parameters:
+ *
+ * @param session the session
+ * @param key the key of the cookie
+ * @param makecb the creation function or NULL
+ * @param freecb the release function or NULL
+ * @param closure an argument for makecb or the value if makecb==NULL
+ * @param replace a boolean enforcing replecement of the previous value
+ *
+ * @return the value of the cookie
+ *
+ * The 'key' is a pointer and compared as pointers.
+ *
+ * For getting the current value of the cookie:
+ *
+ * afb_session_cookie(session, key, NULL, NULL, NULL, 0)
+ *
+ * For storing the value of the cookie
+ *
+ * afb_session_cookie(session, key, NULL, NULL, value, 1)
+ */
void *afb_session_cookie(struct afb_session *session, const void *key, void *(*makecb)(void *closure), void (*freecb)(void *item), void *closure, int replace)
{
int idx;
void *value;
- struct cookie *cookie;
+ struct cookie *cookie, **prv;
+ /* get key hashed index */
+ idx = cookeyidx(key);
+
+ /* lock session and search for the cookie of 'key' */
lock(session);
- cookie = cookie_search(session, key, &idx);
- if (cookie) {
- if (!replace)
- value = cookie->value;
- else {
+ prv = &session->cookies[idx];
+ for (;;) {
+ cookie = *prv;
+ if (!cookie) {
+ /* 'key' not found, create value using 'closure' and 'makecb' */
value = makecb ? makecb(closure) : closure;
- if (cookie->value != value && cookie->freecb)
- cookie->freecb(cookie->value);
- cookie->value = value;
- cookie->freecb = freecb;
- }
- } else {
- value = makecb ? makecb(closure) : closure;
- if (replace || makecb || freecb) {
- cookie = cookie_add(session, idx, key, value, freecb);
- if (!cookie) {
- if (makecb && freecb)
- freecb(value);
- value = NULL;
+ /* store the the only if it has some meaning */
+ if (replace || makecb || freecb) {
+ cookie = malloc(sizeof *cookie);
+ if (!cookie) {
+ errno = ENOMEM;
+ /* calling freecb if there is no makecb may have issue */
+ if (makecb && freecb)
+ freecb(value);
+ value = NULL;
+ } else {
+ cookie->key = key;
+ cookie->value = value;
+ cookie->freecb = freecb;
+ cookie->next = NULL;
+ *prv = cookie;
+ }
+ }
+ break;
+ } else if (cookie->key == key) {
+ /* cookie of key found */
+ if (!replace)
+ /* not replacing, get the value */
+ value = cookie->value;
+ else {
+ /* create value using 'closure' and 'makecb' */
+ value = makecb ? makecb(closure) : closure;
+
+ /* free previous value is needed */
+ if (cookie->value != value && cookie->freecb)
+ cookie->freecb(cookie->value);
+
+ /* store the value and its releaser */
+ cookie->value = value;
+ cookie->freecb = freecb;
+
+ /* but if both are NULL drop the cookie */
+ if (!value && !freecb) {
+ *prv = cookie->next;
+ free(cookie);
+ }
}
+ break;
+ } else {
+ prv = &(cookie->next);
}
}
+
+ /* unlock the session and return the value */
unlock(session);
return value;
}
void *afb_session_get_cookie(struct afb_session *session, const void *key)
{
- int idx;
- void *value;
- struct cookie *cookie;
-
- lock(session);
- cookie = cookie_search(session, key, &idx);
- value = cookie ? cookie->value : NULL;
- unlock(session);
- return value;
+ return afb_session_cookie(session, key, NULL, NULL, NULL, 0);
}
int afb_session_set_cookie(struct afb_session *session, const void *key, void *value, void (*freecb)(void*))
{
- int idx;
- struct cookie *cookie;
-
- lock(session);
- cookie = cookie_search(session, key, &idx);
- if (!cookie)
- cookie = cookie_add(session, idx, key, value, freecb);
- else {
- if (cookie->value != value && cookie->freecb)
- cookie->freecb(cookie->value);
- cookie->value = value;
- cookie->freecb = freecb;
- }
- unlock(session);
- return -!cookie;
+ return -(value != afb_session_cookie(session, key, NULL, freecb, value, 1));
}
struct afb_session;
+#define AFB_SESSION_TIMEOUT_INFINITE -1
+#define AFB_SESSION_TIMEOUT_DEFAULT -2
+#define AFB_SESSION_TIMEOUT_IS_VALID(x) ((x) >= AFB_SESSION_TIMEOUT_DEFAULT)
+
extern void afb_session_init(int max_session_count, int timeout, const char *initok);
extern const char *afb_session_initial_token();
-extern struct afb_session *afb_session_create (const char *uuid, int timeout);
-extern struct afb_session *afb_session_get (const char *uuid, int *created);
+extern struct afb_session *afb_session_create (int timeout);
+extern struct afb_session *afb_session_search (const char *uuid);
+extern struct afb_session *afb_session_get (const char *uuid, int timeout, int *created);
extern const char *afb_session_uuid (struct afb_session *session);
extern struct afb_session *afb_session_addref(struct afb_session *session);
extern void afb_session_unref(struct afb_session *session);
+extern void afb_session_set_autoclose(struct afb_session *session, int autoclose);
extern void afb_session_close(struct afb_session *session);
+extern int afb_session_is_active (struct afb_session *session);
+extern int afb_session_is_closed (struct afb_session *session);
extern int afb_session_check_token(struct afb_session *session, const char *token);
extern void afb_session_new_token(struct afb_session *session);
struct afb_proto_ws_describe *describe;
};
+/*
+ * structure for recording sessions
+ */
+struct server_session
+{
+ struct server_session *next;
+ struct afb_session *session;
+};
+
/******************* stub description for client or servers ******************/
struct afb_stub_ws
/* credentials (server side) */
struct afb_cred *cred;
+ /* sessions (server side) */
+ struct server_session *sessions;
+
/* apiset */
struct afb_apiset *apiset;
/*****************************************************/
+static void record_session(struct afb_stub_ws *stubws, struct afb_session *session)
+{
+ struct server_session *s, **prv;
+
+ /* search */
+ prv = &stubws->sessions;
+ while ((s = *prv)) {
+ if (s->session == session)
+ return;
+ if (afb_session_is_active(s->session))
+ prv = &s->next;
+ else {
+ *prv = s->next;
+ afb_session_addref(s->session);
+ free(s);
+ }
+ }
+
+ /* create */
+ s = malloc(sizeof *s);
+ if (s) {
+ s->session = afb_session_addref(session);
+ s->next = stubws->sessions;
+ stubws->sessions = s;
+ }
+}
+
+static void release_sessions(struct afb_stub_ws *stubws)
+{
+ struct server_session *s;
+
+ while((s = stubws->sessions)) {
+ stubws->sessions = s->next;
+ afb_session_unref(s->session);
+ free(s);
+ }
+}
+
+/*****************************************************/
+
static void on_call(void *closure, struct afb_proto_ws_call *call, const char *verb, struct json_object *args, const char *sessionid)
{
struct afb_stub_ws *stubws = closure;
if (afb_context_connect(&wreq->xreq.context, sessionid, NULL) < 0)
goto unconnected;
wreq->xreq.context.validated = 1;
+ record_session(stubws, wreq->xreq.context.session);
+ if (wreq->xreq.context.created)
+ afb_session_set_autoclose(wreq->xreq.context.session, 1);
/* makes the call */
wreq->xreq.cred = afb_cred_addref(stubws->cred);
if (stubws->on_hangup)
stubws->on_hangup(stubws);
+
+ release_sessions(stubws);
}
/*****************************************************/
if (!__atomic_sub_fetch(&stubws->refcount, 1, __ATOMIC_RELAXED)) {
drop_all_events(stubws);
afb_evt_listener_unref(stubws->listener);
+ release_sessions(stubws);
afb_proto_ws_unref(stubws->proto);
afb_cred_unref(stubws->cred);
afb_apiset_unref(stubws->apiset);
};
/* struct for sessions */
-struct session {
- struct session *next; /* link to the next session */
- struct afb_session *session; /* the session */
- struct afb_trace *trace; /* the tracer */
+struct cookie {
+ struct afb_session *session; /* the session */
+ struct afb_trace *trace; /* the tracer */
};
/* struct for recording hooks */
void *handler; /* the handler of the hook */
struct event *event; /* the associated event */
struct tag *tag; /* the associated tag */
- struct session *session; /* the associated session */
+ struct afb_session *session; /* the associated session */
};
/* types of hooks */
struct afb_session *bound; /* bound to session */
struct event *events; /* list of events */
struct tag *tags; /* list of tags */
- struct session *sessions; /* list of tags */
struct hook *hooks[Trace_Type_Count]; /* hooks */
};
/*******************************************************************************/
/* drop hooks of 'trace' matching 'tag' and 'event' and 'session' */
-static void trace_unhook(struct afb_trace *trace, struct tag *tag, struct event *event, struct session *session)
+static void trace_unhook(struct afb_trace *trace, struct tag *tag, struct event *event, struct afb_session *session)
{
int i;
struct hook *hook, **prev;
struct hook *hook;
struct tag *tag, **ptag;
struct event *event, **pevent;
- struct session *session, **psession;
- /* clean sessions */
- psession = &trace->sessions;
- while ((session = *psession)) {
- /* search for session */
- for (hook = NULL, i = 0 ; !hook && i < Trace_Type_Count ; i++)
- for (hook = trace->hooks[i] ; hook && hook->session != session ; hook = hook->next);
- /* keep or free whether used or not */
- if (hook)
- psession = &session->next;
- else {
- *psession = session->next;
- if (__atomic_exchange_n(&session->trace, NULL, __ATOMIC_RELAXED))
- afb_session_set_cookie(session->session, session, NULL, NULL);
- free(session);
- }
- }
/* clean tags */
ptag = &trace->tags;
while ((tag = *ptag)) {
}
}
-/* callback at end of traced session */
-static void free_session_cookie(void *cookie)
-{
- struct session *session = cookie;
- struct afb_trace *trace = __atomic_exchange_n(&session->trace, NULL, __ATOMIC_RELAXED);
- if (trace) {
- pthread_mutex_lock(&trace->mutex);
- trace_unhook(trace, NULL, NULL, session);
- trace_cleanup(trace);
- pthread_mutex_unlock(&trace->mutex);
- }
-}
-
/*
* Get the tag of 'name' within 'trace'.
* If 'alloc' isn't zero, create the tag and add it.
}
/*
- * Get the session of 'value' within 'trace'.
- * If 'alloc' isn't zero, create the session and add it.
+ * called on session closing
*/
-static struct session *trace_get_session(struct afb_trace *trace, struct afb_session *value, int alloc)
-{
- struct session *session;
-
- /* search the session */
- session = trace->sessions;
- while (session && session->session != value)
- session = session->next;
-
- if (!session && alloc) {
- session = malloc(sizeof * session);
- if (session) {
- session->session = value;
- session->trace = NULL;
- session->next = trace->sessions;
- trace->sessions = session;
- }
- }
- return session;
+static void session_closed(void *item)
+{
+ struct cookie *cookie = item;
+
+ pthread_mutex_lock(&cookie->trace->mutex);
+ trace_unhook(cookie->trace, NULL, NULL, cookie->session);
+ pthread_mutex_unlock(&cookie->trace->mutex);
+ free(cookie);
+}
+
+/*
+ * records the cookie of session for tracking close
+ */
+static void *session_open(void *closure)
+{
+ struct cookie *param = closure, *cookie;
+ cookie = malloc(sizeof *cookie);
+ if (cookie)
+ *cookie = *param;
+ return cookie;
}
/*
* Get the session of 'uuid' within 'trace'.
* If 'alloc' isn't zero, create the session and add it.
*/
-static struct session *trace_get_session_by_uuid(struct afb_trace *trace, const char *uuid, int alloc)
+static struct afb_session *trace_get_session_by_uuid(struct afb_trace *trace, const char *uuid, int alloc)
{
- struct afb_session *session;
- int created;
+ struct cookie cookie;
- session = afb_session_get(uuid, alloc ? &created : NULL);
- return session ? trace_get_session(trace, session, alloc) : NULL;
+ if (!alloc)
+ cookie.session = afb_session_search(uuid);
+ else {
+ cookie.session = afb_session_get(uuid, AFB_SESSION_TIMEOUT_DEFAULT, NULL);
+ if (cookie.session) {
+ cookie.trace = trace;
+ afb_session_cookie(cookie.session, cookie.trace, session_open, session_closed, &cookie, 0);
+ }
+ }
+ return cookie.session;
}
static struct hook *trace_make_detached_hook(struct afb_trace *trace, const char *event, const char *tag)
static void trace_attach_hook(struct afb_trace *trace, struct hook *hook, enum trace_type type)
{
- struct session *session = hook->session;
hook->next = trace->hooks[type];
trace->hooks[type] = hook;
- if (session && !session->trace) {
- session->trace = trace;
- afb_session_set_cookie(session->session, session, session, free_session_cookie);
- }
}
/*******************************************************************************/
static void addhook(struct desc *desc, enum trace_type type)
{
struct hook *hook;
- struct session *session;
+ struct afb_session *session;
struct afb_session *bind;
struct afb_trace *trace = desc->context->trace;
/* create the hook handler */
switch (type) {
case Trace_Type_Xreq:
- if (desc->session) {
+ if (!desc->session)
+ session = afb_session_addref(bind);
+ else {
session = trace_get_session_by_uuid(trace, desc->session, 1);
if (!session) {
ctxt_error(&desc->context->errors, "allocation of session failed");
free(hook);
return;
}
- bind = session->session;
}
- hook->handler = afb_hook_create_xreq(desc->api, desc->verb, bind,
+ hook->handler = afb_hook_create_xreq(desc->api, desc->verb, session,
desc->flags[type], &hook_xreq_itf, hook);
+ afb_session_unref(session);
break;
case Trace_Type_Ditf:
hook->handler = afb_hook_create_ditf(desc->api, desc->flags[type], &hook_ditf_itf, hook);
{
int rc;
struct context *context = closure;
- struct session *session;
+ struct afb_session *session;
const char *uuid;
rc = wrap_json_unpack(object, "s", &uuid);
session = trace_get_session_by_uuid(context->trace, uuid, 0);
if (!session)
ctxt_error(&context->errors, "session %s not found", uuid);
- else
+ else {
trace_unhook(context->trace, NULL, NULL, session);
+ afb_session_unref(session);
+ }
}
}
--- /dev/null
+/*
+ * Copyright (C) 2016, 2017 "IoT.bzh"
+ * Author José Bollo <jose.bollo@iot.bzh>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#define _GNU_SOURCE
+
+#include <stdlib.h>
+#include <stdint.h>
+#include <unistd.h>
+#include <signal.h>
+#include <time.h>
+#include <sys/syscall.h>
+#include <pthread.h>
+#include <errno.h>
+#include <assert.h>
+
+#include <systemd/sd-event.h>
+
+#include "jobs.h"
+#include "sig-monitor.h"
+#include "verbose.h"
+
+#include "jobs.h"
+
+struct jobloop;
+
+struct job
+{
+ struct job *next;
+ const void *group;
+ int timeout;
+ void (*callback)(int signum, void* arg);
+ void *closure;
+};
+
+static struct job *first, *last;
+static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
+
+static int add_job(const void *group, int timeout, void (*callback)(int signum, void *closure), void *closure)
+{
+ struct job *j;
+
+ j = malloc(sizeof*j);
+ if (!j) {
+ errno = ENOMEM;
+ return -1;
+ }
+
+ j->next = 0;
+ j->group = group;
+ j->timeout = timeout;
+ j->callback = callback;
+ j->closure = closure;
+
+ pthread_mutex_lock(&mutex);
+ if (first)
+ last->next = j;
+ else
+ first = j;
+ last = j;
+ pthread_mutex_unlock(&mutex);
+ return 0;
+}
+
+static void *thrrun(void *arg)
+{
+ struct job *j;
+
+ pthread_mutex_lock(&mutex);
+ j = first;
+ if (j)
+ first = j->next;
+ pthread_mutex_unlock(&mutex);
+ if (j) {
+ j->callback(0, j->closure);
+ free(j);
+ }
+ return 0;
+}
+
+int jobs_queue(
+ const void *group,
+ int timeout,
+ void (*callback)(int signum, void* arg),
+ void *arg)
+{
+ pthread_t tid;
+ int rc = add_job(group, timeout, callback, arg);
+ if (!rc) {
+ rc = pthread_create(&tid, NULL, thrrun, NULL);
+ if (rc)
+ rc = -1;
+ }
+ return rc;
+}
+
+#if 0
+int jobs_enter(
+ const void *group,
+ int timeout,
+ void (*callback)(int signum, void *closure, struct jobloop *jobloop),
+ void *closure)
+{
+ return 0;
+}
+
+int jobs_leave(struct jobloop *jobloop)
+{
+ return 0;
+}
+
+int jobs_call(
+ const void *group,
+ int timeout,
+ void (*callback)(int, void*),
+ void *arg)
+{
+ return 0;
+}
+
+struct sd_event *jobs_get_sd_event()
+{
+ struct sd_event *r;
+ int rc = sd_event_default(&r);
+ return rc < 0 ? NULL : r;
+}
+
+void jobs_terminate()
+{
+}
+
+int jobs_start(int allowed_count, int start_count, int waiter_count, void (*start)(int signum))
+{
+ start(0);
+ return 0;
+}
+#endif
struct events *next;
struct sd_event *event;
uint64_t timeout;
- unsigned used: 1;
- unsigned runs: 1;
+ enum {
+ Available,
+ Modifiable,
+ Locked
+ } state;
};
/** Description of threads */
struct thread *next; /**< next thread of the list */
struct thread *upper; /**< upper same thread */
struct job *job; /**< currently processed job */
- struct events *events; /**< currently processed job */
pthread_t tid; /**< the thread id */
unsigned stop: 1; /**< stop requested */
- unsigned lowered: 1; /**< has a lower same thread */
unsigned waits: 1; /**< is waiting? */
};
/* list of threads */
static struct thread *threads;
-static _Thread_local struct thread *current;
+static _Thread_local struct thread *current_thread;
+static _Thread_local struct events *current_events;
/* queue of pending jobs */
static struct job *first_job;
static inline struct events *events_get()
{
struct events *events = first_events;
- while (events && events->used)
+ while (events && events->state != Available)
events = events->next;
return events;
}
*/
static void events_call(int signum, void *arg)
{
+ int rc;
+ struct sd_event *se;
struct events *events = arg;
- if (!signum)
- sd_event_run(events->event, events->timeout);
+
+ if (!signum) {
+ se = events->event;
+ rc = sd_event_prepare(se);
+ if (rc < 0) {
+ errno = -rc;
+ ERROR("sd_event_prepare returned an error (state: %d): %m", sd_event_get_state(events->event));
+ } else {
+ if (rc == 0) {
+ rc = sd_event_wait(se, events->timeout);
+ if (rc < 0) {
+ errno = -rc;
+ ERROR("sd_event_wait returned an error (state: %d): %m", sd_event_get_state(events->event));
+ }
+ }
+
+ if (rc > 0) {
+ rc = sd_event_dispatch(se);
+ if (rc < 0) {
+ errno = -rc;
+ ERROR("sd_event_dispatch returned an error (state: %d): %m", sd_event_get_state(events->event));
+ }
+ }
+ }
+ }
}
/**
*/
static void thread_run(volatile struct thread *me)
{
- struct thread **prv, *thr;
+ struct thread **prv;
struct job *job;
struct events *events;
uint64_t evto;
/* initialize description of itself and link it in the list */
me->tid = pthread_self();
me->stop = 0;
- me->lowered = 0;
me->waits = 0;
- me->upper = current;
- if (current) {
- current->lowered = 1;
+ me->upper = current_thread;
+ if (current_thread) {
evto = EVENT_TIMEOUT_CHILD;
- me->events = current->events;
} else {
started++;
sig_monitor_init_timeouts();
evto = EVENT_TIMEOUT_TOP;
- me->events = NULL;
}
me->next = threads;
threads = (struct thread*)me;
- current = (struct thread*)me;
+ current_thread = (struct thread*)me;
/* loop until stopped */
while (!me->stop) {
sig_monitor(job->timeout, job->callback, job->arg);
pthread_mutex_lock(&mutex);
- /* release the run job */
- job_release(job);
-
/* release event if any */
- events = me->events;
- if (events) {
- events->used = 0;
- me->events = NULL;
+ events = current_events;
+ if (events && events->state == Modifiable) {
+ current_events = NULL;
+ events->state = Available;
}
+
+ /* release the run job */
+ job_release(job);
} else {
/* no job, check events */
- events = me->events;
- if (!events || events->runs)
+ events = current_events;
+ if (!events)
events = events_get();
+ else if (events->state == Locked) {
+ events = 0;
+ WARNING("Loosing an event loop because reentering");
+ }
if (events) {
/* run the events */
- events->used = 1;
- events->runs = 1;
+ events->state = Locked;
events->timeout = evto;
- me->events = events;
+ current_events = events;
pthread_mutex_unlock(&mutex);
sig_monitor(0, events_call, events);
pthread_mutex_lock(&mutex);
- events->used = 0;
- events->runs = 0;
- me->events = NULL;
- thr = me->upper;
- while (thr && thr->events == events) {
- thr->events = NULL;
- thr = thr->upper;
- }
+ current_events = NULL;
+ events->state = Available;
} else {
/* no job and not events */
waiting++;
while (*prv != me)
prv = &(*prv)->next;
*prv = me->next;
- current = me->upper;
- if (current) {
- current->lowered = 0;
- } else {
+ current_thread = me->upper;
+ if (!current_thread) {
sig_monitor_clean_timeouts();
started--;
}
struct sd_event *jobs_get_sd_event()
{
struct events *events;
- struct thread *me;
int rc;
pthread_mutex_lock(&mutex);
/* search events on stack */
- me = current;
- while (me && !me->events)
- me = me->upper;
- if (me)
- /* return the stacked events */
- events = me->events;
- else {
+ events = current_events;
+ if (!events) {
/* search an available events */
events = events_get();
if (!events) {
events = malloc(sizeof *events);
if (events && (rc = sd_event_new(&events->event)) >= 0) {
if (nevents < started || start_one_thread() >= 0) {
- events->used = 0;
- events->runs = 0;
+ events->state = Available;
events->next = first_events;
first_events = events;
} else {
}
}
if (events) {
- me = current;
- if (me) {
- events->used = 1;
- me->events = events;
- } else {
+ events->state = Modifiable;
+ if (!current_thread)
WARNING("event returned for unknown thread!");
- }
+ current_events = events;
}
}
pthread_mutex_unlock(&mutex);
pthread_mutex_lock(&mutex);
/* check whether already running */
- if (current || allowed) {
+ if (current_thread || allowed) {
ERROR("thread already started");
errno = EINVAL;
goto error;
head = job->next;
/* search if job is stacked for current */
- t = current;
+ t = current_thread;
while (t && t->job != job)
t = t->upper;
if (t) {
list = config->calls;
if (list) {
sreq = calloc(1, sizeof *sreq);
- sreq->session = afb_session_create("startup", 3600);
+ sreq->session = afb_session_create(3600);
sreq->current = list;
startup_call_current(sreq);
}
#include <errno.h>
#include <string.h>
#include <sys/uio.h>
+#include <pthread.h>
static const char *appname;
static const char chars[] = { '\n', '?', ':', ' ', '[', ',', ']' };
+static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
+
static void _vverbose_(int loglevel, const char *file, int line, const char *function, const char *fmt, va_list args)
{
char buffer[4000];
iov[n].iov_base = (void*)&chars[0];
iov[n++].iov_len = 1;
+ pthread_mutex_lock(&mutex);
writev(STDERR_FILENO, iov, n);
+ pthread_mutex_unlock(&mutex);
errno = saverr;
}
var initial = {
base: base.base || "api",
- token: base.token || initialtoken || "hello",
+ token: base.token || initialtoken || "HELLO",
host: base.host || window.location.host,
url: base.url || undefined
};
function onclose(event) {
for (var id in this.pendings) {
- var ferr = this.pendings[id].onerror;
- ferr && ferr(null, this);
+ try { this.pendings[id][1](); } catch (x) {/*TODO?*/}
}
this.pendings = {};
this.onclose && this.onclose();
if (id in pendings) {
var p = pendings[id];
delete pendings[id];
- var f = p[offset];
- f(ans);
+ try { p[offset](ans); } catch (x) {/*TODO?*/}
}
}
this.onabort = function(){};
}
- function call(method, request) {
+ function call(method, request, callid) {
return new Promise((function(resolve, reject){
var id, arr;
- do {
- id = String(this.counter = 4095 & (this.counter + 1));
- } while (id in this.pendings);
+ if (callid) {
+ id = String(callid);
+ if (id in this.pendings)
+ throw new Error("pending callid("+id+") exists");
+ } else {
+ do {
+ id = String(this.counter = 4095 & (this.counter + 1));
+ } while (id in this.pendings);
+ }
this.pendings[id] = [ resolve, reject ];
arr = [CALL, id, method, request ];
if (AFB_context.token) arr.push(AFB_context.token);
var initial = {
base: base.base || "api",
- token: base.token || initialtoken || "hello",
+ token: base.token || initialtoken || "HELLO",
host: base.host || window.location.host,
url: base.url || undefined
};
function onclose(event) {
for (var id in this.pendings) {
- var ferr = this.pendings[id].onerror;
- ferr && ferr(null, this);
+ try { this.pendings[id][1](); } catch (x) {/*TODO?*/}
}
this.pendings = {};
this.onclose && this.onclose();
if (id in pendings) {
var p = pendings[id];
delete pendings[id];
- var f = p[offset];
- f(ans);
+ try { p[offset](ans); } catch (x) {/*TODO?*/}
}
}
this.onabort = function(){};
}
- function call(method, request) {
+ function call(method, request, callid) {
return new Promise((function(resolve, reject){
var id, arr;
- do {
- id = String(this.counter = 4095 & (this.counter + 1));
- } while (id in this.pendings);
+ if (callid) {
+ id = String(callid);
+ if (id in this.pendings)
+ throw new Error("pending callid("+id+") exists");
+ } else {
+ do {
+ id = String(this.counter = 4095 & (this.counter + 1));
+ } while (id in this.pendings);
+ }
this.pendings[id] = [ resolve, reject ];
arr = [CALL, id, method, request ];
if (AFB_context.token) arr.push(AFB_context.token);
/*******************************************************************/
/* json format */
-.json.string { color: lightskyblue; }
+.json.string { color: teal; }
.json.number { color: darkorange; }
.json.boolean { color: deepskyblue; }
.json.null { color: magenta; }
<div id="params" class="clearfix">
<div>host: <input type="text" id="param-host" size="50" value="localhost"></input></div>
<div>port: <input type="text" id="param-port" size="10" value="1234"></input></div>
- <div>token: <input type="text" id="param-token" size="33" value="hello"></input></div>
+ <div>token: <input type="text" id="param-token" size="33" value="HELLO"></input></div>
</div>
<div class="-flex-fill -box-out">
<div id="trace-events" class="-box-in">
at("param-host").value = document.location.hostname;
at("param-port").value = document.location.port;
var args = new URLSearchParams(document.location.search.substring(1));
- at("param-token").value = args.get("x-afb-token") || args.get("token") || "hello";
+ at("param-token").value = args.get("x-afb-token") || args.get("token") || "HELLO";
document.onbeforeunload = on_disconnect;
Code Review / src / app-framework-binder.git / commitdiff