WIP

9 files changed, 282 insertions, 11 deletions

diff --git a/v3/glougloud/Makefile b/v3/glougloud/Makefile
index abb0098..8cf198b 100644
--- a/v3/glougloud/Makefile
+++ b/v3/glougloud/Makefile
@@ -2,9 +2,9 @@ PROG = glougloud
 SOURCES = glougloud.c  probes.c  redis.c  viz.c
 OBJECTS = $(SOURCES:.c=.o)
 CFLAGS+=-Wall -g
-LDFLAGS=-levent -ldnet -lglouglou -ldl -lhiredis
-USER_PROBES = _glougloud_probe
-USER_VIZ = _glougloud_viz
+LDFLAGS=-levent -ldnet -lglouglou -lsendbuf -ldl -lhiredis
+USER_PROBES = _ggdprobe
+USER_VIZ = _ggdviz
 GLOUGLOUD_HOME = "/var/lib/glougloud"
 GLOUGLOUD_CHROOT = "$(GLOUGLOUD_HOME)/chroot"
 
diff --git a/v3/glougloud/glougloud.h b/v3/glougloud/glougloud.h
index 7249183..3835fac 100644
--- a/v3/glougloud/glougloud.h
+++ b/v3/glougloud/glougloud.h
@@ -5,8 +5,8 @@
 #include <hiredis/hiredis.h>
 #include <hiredis/async.h>
 
-#define GLOUGLOUD_USER_PROBES "_glougloud_probe"
-#define GLOUGLOUD_USER_VIZ "_glougloud_viz"
+#define GLOUGLOUD_USER_PROBES "_ggdprobe"
+#define GLOUGLOUD_USER_VIZ "_ggdviz"
 #define GLOUGLOUD_LOGFILE "/var/log/glougloud.log"
 #define GLOUGLOUD_MOD_PATH "/lib/glougloud/modules/"
 
diff --git a/v3/glougloud/probes.c b/v3/glougloud/probes.c
index 3e7d657..37fa8ca 100644
--- a/v3/glougloud/probes.c
+++ b/v3/glougloud/probes.c
@@ -56,18 +56,23 @@ probes_init(struct glougloud *ggd) {
 	_probes->pid = fork();
 	if (_probes->pid > 0)
 		return 0;
-	droppriv(GLOUGLOUD_USER_PROBES, 1, NULL);
 	setprocname("probes");
+	droppriv(GLOUGLOUD_USER_PROBES, 1, NULL);
 
 	_probes->evb = event_base_new();
 	_probes->mods = modules_load(GLOUGLOUD_MOD_PATH, NULL);
 
 	_probes->rc = redis_connect(_probes->evb, cb_connect, cb_disconnect);
+	if (_probes->rc->err)
+		return -1;
 
-	// XXX libglouglou gg_server
 	_probes->server = gg_server_start(_probes->evb,
 		&ggd->probes.serv_ip, ggd->probes.serv_port,
 		prb_handle_conn, prb_handle_packet, NULL);
+	if (!_probes->server) {
+		log_warn("probes: gg_server_start failed");
+		return -1;
+	}
 
 	event_base_dispatch(_probes->evb);
 
diff --git a/v3/glougloud/viz.c b/v3/glougloud/viz.c
index 46f3f77..5a0d372 100644
--- a/v3/glougloud/viz.c
+++ b/v3/glougloud/viz.c
@@ -61,18 +61,22 @@ viz_init(struct glougloud *ggd)
 	_viz->pid = fork();
 	if (_viz->pid > 0)
 		return 0;
-	droppriv(GLOUGLOUD_USER_VIZ, 1, NULL);
 	setprocname("viz");
+	droppriv(GLOUGLOUD_USER_VIZ, 1, NULL);
 
 	_viz->evb = event_base_new();
 	_viz->mods = modules_load(GLOUGLOUD_MOD_PATH, NULL);
 
 	_viz->rc = redis_connect(_viz->evb, cb_connect, cb_disconnect);
+	if (_viz->rc->err)
+		return -1;
 	redisAsyncCommand(_viz->rc, cb_notification, "event",
 		"SUBSCRIBE __keyevent@ggd__:*");
 
 	_viz->srv_tcp.ev = tcp_server_create(_viz->evb, &_ggd->viz.serv_ip,
 		_ggd->viz.serv_port, cb_srv_conn, NULL);
+	if (!_viz->srv_tcp.ev)
+		return -1;
 
 	event_base_dispatch(_viz->evb);
 
diff --git a/v3/libglouglou/libglouglou.h b/v3/libglouglou/libglouglou.h
index cd85c5f..f3a9d46 100644
--- a/v3/libglouglou/libglouglou.h
+++ b/v3/libglouglou/libglouglou.h
@@ -35,7 +35,7 @@ enum client_status {
 };
 
 struct gg_client {
-	struct event_base *ev_base;
+	struct event_base *evb;
 	struct addr *ip;
 	int	port;
 	struct sockaddr_in addr;
@@ -68,7 +68,7 @@ struct gg_user {
 };
 
 struct gg_server {
-	struct event_base *ev_base;
+	struct event_base *evb;
 	struct addr *ip;
 	int	 port;
 	struct sockaddr_in addr;
diff --git a/v3/libglouglou/server.c b/v3/libglouglou/server.c
index eaa4135..d3870fe 100644
--- a/v3/libglouglou/server.c
+++ b/v3/libglouglou/server.c
@@ -1,10 +1,65 @@
+#include <stdlib.h>
+#include <unistd.h>
+#include <string.h>
+#include <errno.h>
+
+#include <libsendbuf.h>
 #include "libglouglou.h"
 
+void
+cb_srv_receive(evutil_socket_t fd, short what, void *arg)
+{
+
+}
+
+int
+user_send(struct gg_user *usr, void *data, int size)
+{
+	int sent;
+
+	sent = sendto(usr->sock, data, size, 0, (struct sockaddr *)&usr->addr,
+			sizeof(struct sockaddr_in));
+	if (sent == -1)
+		log_warn("failed: %s", strerror(errno));
+	return sent;
+}
+
+void
+user_del(struct gg_server *srv, struct gg_user *usr)
+{
+	log_debug("Del user %d !", usr->id);
+	if (usr->sbuf)
+		sendbuf_free(usr->sbuf);
+	user_send(usr, "", 0);
+	LIST_REMOVE(usr, entry);
+	srv->user_count--;
+	free(usr);
+}
+
+
 struct gg_server *gg_server_start(struct event_base *evb,
 			struct addr *ip, int port,
 			int (*handle_conn)(struct gg_server *srv, struct gg_user *usr),
-			int (*handle_pkt)(struct gg_server *srv, struct gg_user *usr, struct gg_packet *pkt), void *usrdata)
+			int (*handle_packet)(struct gg_server *srv, struct gg_user *usr, struct gg_packet *pkt), void *usrdata)
 {
+	struct gg_server *srv;
+
+	srv = xcalloc(1, sizeof(struct gg_server));
+	srv->evb = evb;
+	srv->ip = ip;
+	srv->port = port;
+	srv->handle_conn = handle_conn;
+	srv->handle_packet = handle_packet;
+	srv->usrdata = usrdata;
+	srv->ev = udp_server_create(evb, ip, port, cb_srv_receive, srv);
+	if (!srv->ev)
+		goto err;
+
+	return srv;
+
+err:
+	log_warn("gg_server_start: %s", strerror(errno));
+	gg_server_stop(srv);
 	return NULL;
 }
 
@@ -12,6 +67,15 @@ struct gg_server *gg_server_start(struct event_base *evb,
 void
 gg_server_stop(struct gg_server *srv)
 {
+	struct gg_user *usr;
 
+	while ((usr = LIST_FIRST(&srv->user_list))) {
+		user_del(srv, usr);
+	}
+	if (srv->ev) {
+		close(event_get_fd(srv->ev));
+		event_del(srv->ev);
+	}
+	free(srv);
 }
 
diff --git a/v3/libsendbuf/Makefile b/v3/libsendbuf/Makefile
new file mode 100644
index 0000000..1f4849a
--- /dev/null
+++ b/v3/libsendbuf/Makefile
@@ -0,0 +1,32 @@
+#CFLAGS += -Wall -O2 -fPIC -shared -g
+CFLAGS += -Wall -O0 -fPIC -shared -g
+
+PREFIX=/usr/local
+INCLUDEDIR=$(PREFIX)/include
+LIBDIR=$(PREFIX)/lib
+LIBNAME=libsendbuf
+
+TARGET  = ${LIBNAME}.so
+SOURCES = libsendbuf.c
+HEADERS = libsendbuf.h
+OBJECTS = $(SOURCES:.c=.o)
+
+all: $(TARGET)
+
+$(TARGET): $(OBJECTS)
+	$(CC) $(CFLAGS) -o $(TARGET) $(OBJECTS)
+
+install:
+	@echo "installation of $(LIBNAME)"
+	mkdir -p $(LIBDIR)
+	mkdir -p $(INCLUDEDIR)
+	install -m 0644 $(TARGET) $(LIBDIR)
+	install -m 0644 $(HEADERS) $(INCLUDEDIR)
+
+tests:
+	$(MAKE) -C tests run
+
+clean:
+	rm -f $(TARGET) $(OBJECTS)
+
+.PHONY: tests
diff --git a/v3/libsendbuf/libsendbuf.c b/v3/libsendbuf/libsendbuf.c
new file mode 100644
index 0000000..12dc95b
--- /dev/null
+++ b/v3/libsendbuf/libsendbuf.c
@@ -0,0 +1,142 @@
+#include <stdlib.h>
+#include <string.h>
+
+#include "libsendbuf.h"
+
+static void cb_timer(evutil_socket_t, short, void *);
+
+/*
+ * Public
+ */
+
+/*
+ * Create a sendbuf
+ */
+struct sendbuf *
+sendbuf_new(struct event_base *ev_base, int buffer_size, int msec_max,
+	    int (*send_func)(void *, int, void *), void *usrdata)
+{
+	struct sendbuf *sbuf = NULL;
+	struct event *ev_timer;
+
+	sbuf = calloc(1, sizeof(struct sendbuf));
+	if (!sbuf)
+		return NULL;
+	sbuf->ev_base = ev_base;
+	sbuf->msec_max = msec_max;
+	sbuf->buffer_size = buffer_size;
+	sbuf->send_func = send_func;
+	sbuf->usrdata = usrdata;
+	sbuf->buffer = malloc(sbuf->buffer_size);
+	if (!sbuf->buffer)
+		goto err;
+
+	ev_timer = evtimer_new(ev_base, cb_timer, sbuf);
+	sbuf->ev_timer = ev_timer;
+	sbuf->ev_timer_tv.tv_usec = msec_max * 1000;
+	evtimer_add(ev_timer, &sbuf->ev_timer_tv);
+
+	return sbuf;
+
+err:
+	sendbuf_free(sbuf);
+	return NULL;
+}
+
+void
+sendbuf_free(struct sendbuf *sbuf)
+{
+	if (sbuf->ev_timer)
+		event_del(sbuf->ev_timer);
+	if (sbuf->buffer && sbuf->send_func)
+		sendbuf_flush(sbuf);
+	if (sbuf->buffer)
+		free(sbuf->buffer);
+	free(sbuf);
+}
+
+/*
+ * Append to the token buffer data to be sent
+ * uses a memcpy, in contrary to sendbuf_gettoken().
+ * return size on success, -1 on error
+ */
+int
+sendbuf_append(struct sendbuf *sbuf, void *token, int size)
+{
+	if (sbuf->buffer_pos + size >= sbuf->buffer_size)
+		if (sendbuf_flush(sbuf) == -1)
+			return -1;
+
+	memcpy(sbuf->buffer + sbuf->buffer_pos, token, size);
+	sbuf->buffer_pos = sbuf->buffer_pos + size;
+
+	return size;
+}
+
+/*
+ * Returns a token buffer to write data to be sent
+ * avoids memcpy, in contrary to sendbuf_append().
+ * might return NULL if the sendbuf is temporary full
+ */
+void *
+sendbuf_gettoken(struct sendbuf *sbuf, int size)
+{
+	void *token;
+
+	if (sbuf->buffer_pos + size >= sbuf->buffer_size)
+		if (sendbuf_flush(sbuf) == -1)
+			return NULL;
+
+	token = sbuf->buffer + sbuf->buffer_pos;
+	sbuf->buffer_pos = sbuf->buffer_pos + size;
+
+	return token;
+}
+
+/*
+ * Send buffer immediatly
+ * Note that you can still add data to the buffer even if flushing is in
+ * progress
+ * returns 0 on success or -1 on error
+ */
+int
+sendbuf_flush(struct sendbuf *sbuf)
+{
+	int tosend, sent;
+
+	if (sbuf->buffer_pos == 0)
+		return 0;
+
+	sbuf->flushing = 1;
+
+	tosend = sbuf->buffer_pos - sbuf->flushing_pos;
+	sent = sbuf->send_func(sbuf->buffer + sbuf->flushing_pos,
+			tosend, sbuf->usrdata);
+	if (sent == -1) {
+		// XXX handle error
+		return -1;
+	} else if (sent < tosend) {
+		sbuf->flushing_pos = sbuf->flushing_pos + sent;
+		return -1;
+	}
+	sbuf->buffer_pos = 0;
+
+	sbuf->flushing = 0;
+	sbuf->flushing_pos = 0;
+	return 0;
+}
+
+
+/*
+ * Private
+ */
+
+static void
+cb_timer(evutil_socket_t fd, short what, void *arg)
+{
+	struct sendbuf *sbuf;
+
+	sbuf = arg;
+	sendbuf_flush(sbuf);
+	evtimer_add(sbuf->ev_timer, &sbuf->ev_timer_tv);
+}
diff --git a/v3/libsendbuf/libsendbuf.h b/v3/libsendbuf/libsendbuf.h
new file mode 100644
index 0000000..f0a7689
--- /dev/null
+++ b/v3/libsendbuf/libsendbuf.h
@@ -0,0 +1,24 @@
+#include <event.h>
+#include <sys/queue.h>
+
+struct sendbuf {
+	struct event_base *ev_base;
+	struct event      *ev_timer;
+	struct timeval     ev_timer_tv;
+	int   msec_max;
+	int   buffer_size;
+	void *buffer;
+	int   buffer_pos; /* next to use in buffer */
+	int   flushing;
+	int   flushing_pos; /* next to send in buffer */
+	int (*send_func)(void *, int, void *);
+	void *usrdata;
+};
+
+struct sendbuf	*sendbuf_new(struct event_base *, int, int,
+			int (*send_func)(void *, int, void *),
+			void *);
+void		 sendbuf_free(struct sendbuf *);
+int		 sendbuf_append(struct sendbuf *, void *, int);
+void		*sendbuf_gettoken(struct sendbuf *, int);
+int		 sendbuf_flush(struct sendbuf *);