#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;
}
/* 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;
- 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;
}
}
- 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(
}
/* 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 **********************************/
--- /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
Code Review / src / app-framework-binder.git / commitdiff