http ping + snapshot

.gitignore

@@ -1,2 +1,3 @@
+/vgcore.*
 /src/*.o
 /ustreamer

Makefile

@@ -1,4 +1,4 @@
-LIBS = -lm -ljpeg -pthread -levent
+LIBS = -lm -ljpeg -pthread -levent -levent_pthreads
 CC = gcc
 CFLAGS = -c -O3 -Wall -Wextra
 LDFLAGS =
@@ -19,4 +19,4 @@ $(PROG): $(OBJECTS)
 
 
 clean:
-	rm -f src/*.o $(PROG)
+	rm -f src/*.o vgcore.* $(PROG)

src/capture.c

@@ -89,8 +89,11 @@ void capture_loop(struct device_t *dev, struct captured_picture_t *captured) {
 		LOG_DEBUG("Allocation memory for captured (result) picture ...");
 		A_CALLOC(captured->picture.data, dev->run->max_picture_size, sizeof(*captured->picture.data));
 
+		A_PTHREAD_M_LOCK(&captured->mutex);
 		captured->width = dev->run->width;
 		captured->height = dev->run->height;
+		captured->online = true;
+		A_PTHREAD_M_UNLOCK(&captured->mutex);
 
 		while (!dev->stop) {
 			SEP_DEBUG('-');
@@ -103,6 +106,7 @@ void capture_loop(struct device_t *dev, struct captured_picture_t *captured) {
 			if (last_worker && !last_worker->has_job && dev->run->pictures[last_worker->ctx.index].data) {
 				A_PTHREAD_M_LOCK(&captured->mutex);
 				captured->picture.size = dev->run->pictures[last_worker->ctx.index].size;
+				captured->picture.allocated = dev->run->pictures[last_worker->ctx.index].allocated;
 				memcpy(
 					captured->picture.data,
 					dev->run->pictures[last_worker->ctx.index].data,
@@ -242,6 +246,7 @@ void capture_loop(struct device_t *dev, struct captured_picture_t *captured) {
 		A_PTHREAD_M_LOCK(&captured->mutex);
 		captured->picture.size = 0;
 		free(captured->picture.data);
+		captured->online = false;
 		A_PTHREAD_M_UNLOCK(&captured->mutex);
 	}
 
@@ -391,8 +396,9 @@ static void *_capture_worker_thread(void *v_ctx) {
 }
 
 static void _capture_destroy_workers(struct device_t *dev, struct workers_pool_t *pool) {
-	LOG_INFO("Destroying workers ...");
 	if (pool->workers) {
+		LOG_INFO("Destroying workers ...");
+
 		*pool->workers_stop = true;
 		for (unsigned index = 0; index < dev->run->n_buffers; ++index) {
 			A_PTHREAD_M_LOCK(&pool->workers[index].has_job_mutex);

src/capture.h

@@ -55,6 +55,7 @@ struct captured_picture_t {
 	struct picture_t	picture;
 	unsigned			width;
 	unsigned			height;
+	bool				online;
 	bool				updated;
 	pthread_mutex_t		mutex;
 };

src/device.c

@@ -49,16 +49,15 @@ static const char *_standard_to_string(const v4l2_std_id standard);
 
 
 struct device_t *device_init() {
-	struct device_t *dev;
 	struct device_runtime_t *run;
-
-	A_CALLOC(dev, 1, sizeof(*dev));
-	MEMSET_ZERO_PTR(dev);
+	struct device_t *dev;
 
 	A_CALLOC(run, 1, sizeof(*run));
 	MEMSET_ZERO_PTR(run);
-	dev->run = run;
-	dev->run->fd = -1;
+	run->fd = -1;
+
+	A_CALLOC(dev, 1, sizeof(*dev));
+	MEMSET_ZERO_PTR(dev);
 
 	dev->path = (char *)DEFAULT_DEVICE;
 	dev->width = 640;
@@ -69,6 +68,7 @@ struct device_t *device_init() {
 	dev->jpeg_quality = 80;
 	dev->timeout = 1;
 	dev->error_timeout = 1;
+	dev->run = run;
 	return dev;
 }
 
@@ -392,6 +392,7 @@ static void _device_open_alloc_picbufs(struct device_t *dev) {
 	for (unsigned index = 0; index < dev->run->n_buffers; ++index) {
 		LOG_DEBUG("Allocating picture buffer %d ...", index);
 		A_CALLOC(dev->run->pictures[index].data, dev->run->max_picture_size, sizeof(*dev->run->pictures[index].data));
+		dev->run->pictures[index].allocated = dev->run->max_picture_size;
 	}
 }
 

src/device.h

@@ -18,6 +18,7 @@ struct hw_buffer_t {
 struct picture_t {
 	unsigned char	*data;
 	unsigned long	size;
+	unsigned long	allocated;
 };
 
 struct device_runtime_t {
@@ -28,7 +29,7 @@ struct device_runtime_t {
 	unsigned			n_buffers;
 	struct hw_buffer_t	*hw_buffers;
 	struct picture_t	*pictures;
-	unsigned			max_picture_size;
+	unsigned long		max_picture_size;
 	bool				capturing;
 };
 

src/http.c

@@ -1,31 +1,179 @@
-#include "capture.h"
+#include <stdlib.h>
+#include <stdbool.h>
+#include <string.h>
+#include <time.h>
+#include <assert.h>
+#include <event2/event.h>
+#include <event2/event-config.h>
+#include <event2/thread.h>
+#include <event2/http.h>
+#include <event2/buffer.h>
+
+#ifndef EVTHREAD_USE_PTHREADS_IMPLEMENTED
+#	error Required libevent-pthreads support
+#endif
+
 #include "tools.h"
+#include "capture.h"
 #include "http.h"
 
 
 static const char DEFAULT_HOST[] = "localhost";
 
 
-struct http_server_t *http_server_init() {
+static void _http_callback_root(struct evhttp_request *request, void *arg);
+static void _http_callback_stream_ping(struct evhttp_request *request, void *v_server);
+static void _http_callback_stream_snapshot(struct evhttp_request *request, void *v_server);
+static void _http_update_exposed(struct http_server_t *server);
+static void _http_add_header(struct evhttp_request *request, const char *key, const char *value);
+
+
+struct http_server_t *http_server_init(struct captured_picture_t *captured) {
+	struct captured_picture_t *exposed;
+	struct http_server_runtime_t *run;
 	struct http_server_t *server;
 
+	exposed = captured_picture_init();
+
+	A_CALLOC(run, 1, sizeof(*run));
+	MEMSET_ZERO_PTR(run);
+	run->captured = captured;
+	run->exposed = exposed;
+
 	A_CALLOC(server, 1, sizeof(*server));
 	MEMSET_ZERO_PTR(server);
-
 	server->host = (char *)DEFAULT_HOST;
 	server->port = 8080;
+	server->run = run;
+
+	assert(!evthread_use_pthreads());
+	assert((run->base = event_base_new()));
+	assert((run->http = evhttp_new(run->base)));
+	evhttp_set_allowed_methods(run->http, EVHTTP_REQ_GET); // TODO: HEAD
+
+	assert(!evhttp_set_cb(run->http, "/", _http_callback_root, NULL));
+	assert(!evhttp_set_cb(run->http, "/ping", _http_callback_stream_ping, (void *)server));
+	assert(!evhttp_set_cb(run->http, "/snapshot", _http_callback_stream_snapshot, (void *)server));
 	return server;
 }
 
 void http_server_destroy(struct http_server_t *server) {
+	evhttp_free(server->run->http);
+	event_base_free(server->run->base);
+	free(server->run->exposed->picture.data);
+	captured_picture_destroy(server->run->exposed);
+	free(server->run);
 	free(server);
+	libevent_global_shutdown();
 }
 
-void http_server_loop(struct http_server_t *server, struct captured_picture_t *captured) {
-	// TODO: implement server here
+int http_server_listen(struct http_server_t *server) {
+	LOG_DEBUG("Binding HTTP to [%s]:%d ...", server->host, server->port);
+	if (evhttp_bind_socket(server->run->http, server->host, server->port) != 0) {
+		LOG_PERROR("Can't listen HTTP on [%s]:%d", server->host, server->port)
+		return -1;
+	}
+	LOG_INFO("Listening HTTP on [%s]:%d", server->host, server->port);
+	return 0;
+}
+
+void http_server_loop(struct http_server_t *server) {
+	LOG_INFO("Starting HTTP eventloop ...");
+	event_base_dispatch(server->run->base);
+	LOG_INFO("HTTP eventloop stopped");
 }
 
 void http_server_loop_break(struct http_server_t *server) {
-	// TODO: implement stop here
+	LOG_INFO("Stopping HTTP eventloop ...");
+	event_base_loopbreak(server->run->base);
 }
 
+
+static void _http_callback_root(struct evhttp_request *request, UNUSED void *arg) {
+	struct evbuffer *buf;
+
+	assert((buf = evbuffer_new()));
+	assert(evbuffer_add_printf(buf,
+		"<!DOCTYPE html><html><head><meta charset=\"utf-8\">"
+		"<title>uStreamer</title></head><body><ul>"
+		"<li><a href=\"/ping\">/ping</a></li>"
+		"<li><a href=\"/snapshot\">/snapshot</a></li>"
+		"</body></html>"
+	));
+	_http_add_header(request, "Content-Type", "text/html");
+	evhttp_send_reply(request, 200, "OK", buf);
+	evbuffer_free(buf);
+}
+
+static void _http_callback_stream_ping(struct evhttp_request *request, void *v_server) {
+	struct http_server_t *server = (struct http_server_t *)v_server;
+	struct evbuffer *buf;
+
+	_http_update_exposed(server);
+
+	assert((buf = evbuffer_new()));
+	assert(evbuffer_add_printf(buf,
+		"{\"stream\": {\"resolution\":"
+		" {\"width\": %u, \"height\": %u},"
+		" \"online\": %s}}",
+		server->run->exposed->width,
+		server->run->exposed->height,
+		(server->run->exposed->online ? "true" : "false")
+	));
+	_http_add_header(request, "Content-Type", "application/json");
+	evhttp_send_reply(request, 200, "OK", buf);
+	evbuffer_free(buf);
+}
+
+static void _http_callback_stream_snapshot(struct evhttp_request *request, void *v_server) {
+	struct http_server_t *server = (struct http_server_t *)v_server;
+	struct evbuffer *buf;
+	struct timespec now_spec;
+	char now_str[64];
+
+	_http_update_exposed(server);
+
+	assert((buf = evbuffer_new()));
+	assert(!evbuffer_add(buf, (const void *)server->run->exposed->picture.data, server->run->exposed->picture.size));
+
+	assert(!clock_gettime(CLOCK_REALTIME, &now_spec));
+	sprintf(now_str, "%u.%06u", (unsigned)now_spec.tv_sec, (unsigned)(now_spec.tv_nsec / 1000)); // TODO: round?
+
+	_http_add_header(request, "Access-Control-Allow-Origin:", "*"); // TODO: need this?
+	_http_add_header(request, "Cache-Control", "no-store, no-cache, must-revalidate, pre-check=0, post-check=0, max-age=0");
+	_http_add_header(request, "Pragma", "no-cache");
+	_http_add_header(request, "Expires", "Mon, 3 Jan 2000 12:34:56 GMT");
+	_http_add_header(request, "X-Timestamp", now_str);
+	_http_add_header(request, "Content-Type", "image/jpeg");
+
+	evhttp_send_reply(request, 200, "OK", buf);
+	evbuffer_free(buf);
+}
+
+static void _http_update_exposed(struct http_server_t *server) {
+	if (server->run->captured->updated) {
+		A_PTHREAD_M_LOCK(&server->run->captured->mutex);
+		if (server->run->captured->picture.allocated > server->run->exposed->picture.allocated) {
+			A_REALLOC(
+				server->run->exposed->picture.data,
+				server->run->captured->picture.allocated * sizeof(*server->run->exposed->picture.data)
+			);
+			server->run->exposed->picture.allocated = server->run->captured->picture.allocated;
+		}
+		memcpy(
+			server->run->exposed->picture.data,
+			server->run->captured->picture.data,
+			server->run->captured->picture.size * sizeof(*server->run->exposed->picture.data)
+		);
+		server->run->exposed->picture.size = server->run->captured->picture.size;
+		server->run->exposed->width = server->run->captured->width;
+		server->run->exposed->height = server->run->captured->height;
+		server->run->exposed->online = server->run->captured->online;
+		server->run->captured->updated = false;
+		A_PTHREAD_M_UNLOCK(&server->run->captured->mutex);
+	}
+}
+
+static void _http_add_header(struct evhttp_request *request, const char *key, const char *value) {
+	assert(!evhttp_add_header(evhttp_request_get_output_headers(request), key, value));
+}

src/http.h

@@ -1,15 +1,29 @@
+#include <stdbool.h>
+#include <event2/event.h>
+#include <event2/http.h>
+
 #include "tools.h"
 #include "capture.h"
 
 
+struct http_server_runtime_t {
+	struct event_base			*base;
+	struct evhttp				*http;
+	struct captured_picture_t	*captured;
+	struct captured_picture_t	*exposed; // updated and mutex are not used
+};
+
 struct http_server_t {
 	char		*host;
 	unsigned	port;
+
+	struct http_server_runtime_t *run;
 };
 
 
-struct http_server_t *http_server_init();
+struct http_server_t *http_server_init(struct captured_picture_t *captured);
 void http_server_destroy(struct http_server_t *server);
 
-void http_server_loop(struct http_server_t *server, struct captured_picture_t *captured);
+int http_server_listen(struct http_server_t *server);
+void http_server_loop(struct http_server_t *server);
 void http_server_loop_break(struct http_server_t *server);

src/main.c

@@ -127,7 +127,7 @@ static void *_capture_loop_thread(UNUSED void *_) {
 
 static void *_server_loop_thread(UNUSED void *_) {
 	_block_thread_signals();
-	http_server_loop(_ctx->server, _ctx->captured);
+	http_server_loop(_ctx->server);
 	return NULL;
 }
 
@@ -151,18 +151,22 @@ static void _install_signal_handlers() {
 
 	LOG_INFO("Installing SIGTERM handler ...");
 	assert(!sigaction(SIGTERM, &sig_act, NULL));
+
+	LOG_INFO("Ignoring SIGPIPE ...");
+	assert(signal(SIGPIPE, SIG_IGN) != SIG_ERR);
 }
 
 int main(int argc, char *argv[]) {
 	struct device_t *dev;
 	struct captured_picture_t *captured;
 	struct http_server_t *server;
+	int exit_code = 0;
 
 	dev = device_init();
 	captured = captured_picture_init();
-	server = http_server_init();
+	server = http_server_init(captured);
 
-	if (_parse_options(argc, argv, dev, server) == 0) {
+	if ((exit_code = _parse_options(argc, argv, dev, server)) == 0) {
 		_install_signal_handlers();
 
 		pthread_t capture_loop_tid;
@@ -174,14 +178,16 @@ int main(int argc, char *argv[]) {
 		ctx.server = server;
 		_ctx = &ctx;
 
-		A_PTHREAD_CREATE(&capture_loop_tid, _capture_loop_thread, NULL);
-		A_PTHREAD_CREATE(&server_loop_tid, _server_loop_thread, NULL);
-		A_PTHREAD_JOIN(capture_loop_tid);
-		A_PTHREAD_JOIN(server_loop_tid);
+		if ((exit_code = http_server_listen(server)) == 0) {
+			A_PTHREAD_CREATE(&capture_loop_tid, _capture_loop_thread, NULL);
+			A_PTHREAD_CREATE(&server_loop_tid, _server_loop_thread, NULL);
+			A_PTHREAD_JOIN(capture_loop_tid);
+			A_PTHREAD_JOIN(server_loop_tid);
+		}
 	}
 
 	http_server_destroy(server);
 	captured_picture_destroy(captured);
 	device_destroy(dev);
-	return 0;
+	return abs(exit_code);
 }

src/tools.h

@@ -67,6 +67,7 @@ unsigned log_level;
 
 
 #define A_CALLOC(_dest, _nmemb, _size)	assert((_dest = calloc(_nmemb, _size)))
+#define A_REALLOC(_dest, _size)			assert((_dest = realloc(_dest, _size)))
 #define MEMSET_ZERO(_x_obj)				memset(&(_x_obj), 0, sizeof(_x_obj))
 #define MEMSET_ZERO_PTR(_x_ptr)			memset(_x_ptr, 0, sizeof(*(_x_ptr)))