Implemented thumbnail worker

config.h

+#define CONFIG_APIKEY "key"
+#define CONFIG_APIBASE "http://localhost:5000"
+#define CONFIG_VIDEOS_RELEASED "/mnt/released"
+#define CONFIG_VIDEOS_TMP "/mnt/video-main/kodiert"
+#define CONFIG_VIDEOS_RAW "/mnt/raw"

thumbnail.c

+#include <libavcodec/avcodec.h>
+#include <libavformat/avformat.h>
+#include <libswscale/swscale.h>
+
+#include "util.h"
+#include "config.h"
+
+AVFrame *scale_frame(AVFrame *frame, enum AVPixelFormat pix_fmt,
+		int width, int height)
+{
+	AVFrame *scaled;
+	struct SwsContext *swsctx;
+	scaled = av_frame_alloc();
+	scaled->format = pix_fmt;
+	scaled->width = width;
+	scaled->height = height;
+	av_frame_get_buffer(scaled, 32);
+	swsctx = sws_getContext(frame->width, frame->height, frame->format,
+			scaled->width, scaled->height, scaled->format, 0, 0, 0, 0);
+	sws_scale(swsctx, (const uint8_t **) frame->data, frame->linesize, 0,
+			frame->height, scaled->data, scaled->linesize);
+	return scaled;
+}
+
+int main(int argc, char *argv[])
+{
+	int jobid, vidx;
+	char lectureid[JSTR_SIZE], src[100], tmp[100], dest[100];
+	AVFormatContext *demux, *mux;
+	AVCodecContext *dec, *enc;
+	AVPacket pkt;
+	AVFrame *frame;
+	if (argc != 5)
+		return 1;
+	av_register_all();
+	av_log_set_callback(avlogbuf_callback);
+
+	/* Prepare arguments */
+	jobid = atoi(argv[1]);
+	jstrb(jlookup(argv[4], "lectureid"), "-1", lectureid);
+	snprintf(BL(src), "%s/%s", CONFIG_VIDEOS_RELEASED,
+			jstr(jlookup(argv[4], "path"), ""));
+	src[sizeof(src)-1] = 0;
+	snprintf(BL(tmp), "%s/.tmp-%i", CONFIG_VIDEOS_TMP, jobid);
+	tmp[sizeof(tmp)-1] = 0;
+	snprintf(BL(dest), "%s/thumbnail/l_%s.jpg", CONFIG_VIDEOS_RELEASED,
+			lectureid);
+	dest[sizeof(dest)-1] = 0;
+	ping_job(jobid, "running", 0);
+
+	/* Open src */
+	demux = 0;
+	if (avformat_open_input(&demux, src, 0, 0))
+		goto fail;
+	avformat_find_stream_info(demux, 0);
+	if ((vidx = av_find_best_stream(demux, AVMEDIA_TYPE_VIDEO, -1, -1, 0, 0)) < 0)
+		goto fail;
+	dec = avcodec_alloc_context3(avcodec_find_decoder(demux->streams[vidx]->codecpar->codec_id));
+	avcodec_parameters_to_context(dec, demux->streams[vidx]->codecpar);
+	avcodec_open2(dec, dec->codec, 0);
+
+	/* Open temporary dest */
+	avformat_alloc_output_context2(&mux, 0, "image2", tmp);
+	avformat_new_stream(mux, 0);
+	enc = avcodec_alloc_context3(avcodec_find_encoder(AV_CODEC_ID_MJPEG));
+	enc->width = 640;
+	enc->height = (enc->width*dec->height)/dec->width;
+	if (enc->codec->pix_fmts)
+		enc->pix_fmt = enc->codec->pix_fmts[0];
+	else
+		enc->pix_fmt = dec->pix_fmt;
+	enc->sample_aspect_ratio = dec->sample_aspect_ratio;
+	enc->time_base = AV_TIME_BASE_Q;
+	avio_open(&mux->pb, tmp, AVIO_FLAG_WRITE);
+	avcodec_open2(enc, enc->codec, 0);
+	avcodec_parameters_from_context(mux->streams[0]->codecpar, enc);
+	if (avformat_write_header(mux, 0) < 0)
+		goto fail;
+
+	/* Create thumbnail */
+	av_seek_frame(demux, -1, (2L*demux->duration)/5L, 0);
+	memset(&pkt, 0, sizeof(pkt));
+	av_init_packet(&pkt);
+	frame = av_frame_alloc();
+	do
+	{
+		if (av_read_frame(demux, &pkt))
+			goto fail;
+		if (pkt.stream_index == vidx)
+			avcodec_send_packet(dec, &pkt);
+	}
+	while (avcodec_receive_frame(dec, frame));
+	avcodec_send_frame(enc, scale_frame(frame, enc->pix_fmt, enc->width,
+				enc->height));
+	avcodec_send_frame(enc, 0);
+	while (!avcodec_receive_packet(enc, &pkt))
+	{
+		pkt.stream_index = 0;
+		pkt.pts = pkt.dts = 0;
+		av_interleaved_write_frame(mux, &pkt);
+	}
+	av_interleaved_write_frame(mux, 0);
+	if (av_write_trailer(mux))
+		goto fail;
+	avio_closep(&mux->pb);
+	rename(tmp, dest);
+	unlink(tmp);
+	ping_job(jobid, "finished", "{\"log\": \"%s\"}", get_avlogbuf());
+	return 0;
+
+fail:
+	unlink(tmp);
+	ping_job(jobid, "failed", "{\"log\": \"%s\"}", get_avlogbuf());
+	return 1;
+}

util.h

+#include <unistd.h>
+#include <stdarg.h>
+
+/* Generic */
+#define SL(s) (s), (sizeof(s)-1)
+#define BL(b) (b), (sizeof(b))
+size_t xvsnprintf(char *p, size_t len, const char *fmt, va_list ap);
+size_t xsnprintf(char *p, size_t len, const char *fmt, ...);
+
+void avlogbuf_callback(void *classp, int level, const char *fmt, va_list ap);
+char *get_avlogbuf(void);
+
+/* API */
+int ping_job(int id, char *state, char *status, ...);
+
+/* JSON parser */
+#define JSTR_SIZE 100
+
+ssize_t jbin(char *s, char *buf, size_t len);
+char *jstrb(char *s, char *err, char *buf);
+char *jstr(char *s, char *err);
+int jint(char *s, int err);
+char *jenter(char *s);
+char *jnext(char *s);
+char *jvalue(char *s);
+char *jlookup(char *s, char *key);
+

util/api.c

+#include <curl/curl.h>
+#include <stdio.h>
+#include <stdarg.h>
+#include <unistd.h>
+#include <string.h>
+
+#include "../util.h"
+#include "../config.h"
+
+static size_t curl_write_cb(char *ptr, size_t size, size_t nmemb, void *user)
+{
+	return size;
+}
+
+int ping_job(int id, char *state, char *status, ...)
+{
+	int ret;
+	CURL *curl;
+	va_list ap;
+	char *e_host, *e_status, *e_apikey, *e_state;
+	char url[100], buf[500];
+	if (!(curl = curl_easy_init()))
+		return -1;
+	curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, curl_write_cb);
+	e_apikey = curl_easy_escape(curl, CONFIG_APIKEY, 0);
+	e_state = curl_easy_escape(curl, state, 0);
+	if (gethostname(BL(buf)))
+		strcpy(buf, "localhost");
+	buf[sizeof(buf)-1] = 0;
+	e_host = curl_easy_escape(curl, buf, 0);
+	if (status)
+	{
+		va_start(ap, status);
+		xvsnprintf(BL(buf), status, ap);
+		va_end(ap);
+	}
+	else
+		strcpy(buf, "{}");
+	e_status = curl_easy_escape(curl, buf, 0);
+	xsnprintf(BL(buf), "apikey=%s&state=%s&host=%s&status=%s",
+			e_apikey, e_state, e_host, e_status);
+	curl_free(e_apikey);
+	curl_free(e_state);
+	curl_free(e_host);
+	curl_free(e_status);
+	curl_easy_setopt(curl, CURLOPT_POSTFIELDS, buf);
+	xsnprintf(BL(url), "%s/internal/jobs/api/job/%i/ping", CONFIG_APIBASE, id);
+	curl_easy_setopt(curl, CURLOPT_URL, url);
+	ret = curl_easy_perform(curl);
+	curl_easy_cleanup(curl);
+	if (ret)
+		return -1;
+	else
+		return 0;
+}

util/avlogbuf.c

+#include <pthread.h>
+#include <libavutil/log.h>
+
+#include "../util.h"
+
+static char *logbuffer = 0;
+static size_t logoffset = 0, logsize = 0;
+static pthread_mutex_t loglock = PTHREAD_MUTEX_INITIALIZER;;
+
+void avlogbuf_callback(void *class, int level, const char *fmt, va_list ap)
+{
+	int len, print_prefix;
+	if (level >= AV_LOG_VERBOSE)
+		return;
+	av_log_default_callback(class, level, fmt, ap);
+	if (pthread_mutex_lock(&loglock))
+		return;
+	if (logsize-logoffset < 1024)
+		logbuffer = realloc(logbuffer, (logsize += 1024));
+	if (!logbuffer)
+		return;
+	print_prefix = 1;
+	len = av_log_format_line2(class, level, fmt, ap, logbuffer+logoffset,
+			logsize-logoffset, &print_prefix);
+	if (len < logsize-logoffset)
+		logoffset += len;
+	else
+		logoffset = logsize-1;
+	pthread_mutex_unlock(&loglock);
+}
+
+char *get_avlogbuf(void)
+{
+	char *ret, *p;
+	if (!logbuffer || pthread_mutex_lock(&loglock))
+		return strdup("");
+	ret = strdup(logbuffer);
+	pthread_mutex_unlock(&loglock);
+	for (p = ret; *p; p ++)
+		if (*p == '"')
+			*p = '\'';
+	return ret;
+}

util/json.c

+#include <stdio.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "../util.h"
+
+static char *skip_ws(char *s)
+{
+	for (; *s == ' ' || *s == '\t' || *s == '\n' || *s == '\r'; s ++);
+	return s;
+}
+
+static char *skip(char *s)
+{
+	s = skip_ws(s);
+	if (*s == '"')
+	{
+		for (s++; *s && *s != '"'; s ++)
+			if (s[0] == '\\' && s[1] == '"')
+				s += 1;
+		if (*s == '"')
+			s ++;
+	}
+	else if (*s == '-' || (*s >= '1' && *s <= '9'))
+		for (s ++; (*s >= '0' && *s <= '9') || *s == 'e' || *s == 'E' || *s == '.' ||
+				*s == '-' || *s == '+'; s ++);
+	else if (*s == '[')
+	{
+		do
+			s = skip_ws(skip(s+1));
+		while (*s == ',');
+		if (*s == ']')
+			s ++;
+	}
+	else if (*s == '{')
+	{
+		do
+		{
+			s = skip_ws(skip(s+1));
+			if (*s == ':')
+				s ++;
+			s = skip_ws(skip(s));
+		}
+		while (*s == ',');
+		if (*s == '}')
+			s ++;
+	}
+	else if (!strncmp(s, SL("true")))
+		return s+4;
+	else if (!strncmp(s, SL("false")))
+		return s+5;
+	else if (!strncmp(s, SL("null")))
+		return s+4;
+	return s;
+}
+
+size_t parse_escape(char *s, uint32_t *code)
+{
+	char buf[4] = {};
+	if (*s == 'u')
+	{
+		if (!(buf[0] = s[1]) ||
+				!(buf[1] = s[2]) ||
+				!(buf[2] = s[3]) ||
+				!(buf[3] = s[4]))
+			return 0;
+		*code = strtol(buf, 0, 16);
+		return 5;
+	}
+	switch (*s)
+	{
+		case 'b': *code = '\b'; break;
+		case 'f': *code = '\f'; break;
+		case 'n': *code = '\n'; break;
+		case 'r': *code = '\r'; break;
+		case 't': *code = '\t'; break;
+		default:
+			*code = *s;
+	}
+	return 1;
+}
+
+#define utf8_len(code) utf8_enc(code, 0)
+size_t utf8_enc(uint32_t code, char *buf)
+{
+	char _buf[4];
+	if (!buf)
+		buf = _buf;
+	if (code <= 0x7f)
+	{
+		buf[0] = code;
+		return 1;
+	}
+	else if (code <= 0x7ff)
+	{
+		buf[0] = 0xc0 | (code >> 6);
+		buf[1] = 0x80 | (code & 0x3f);
+		return 2;
+	}
+	else if (code <= 0xffff)
+	{
+		buf[0] = 0xe0 | (code >> 12);
+		buf[1] = 0x80 | ((code >> 6) & 0x3f);
+		buf[2] = 0x80 | (code & 0x3f);
+		return 3;
+	}
+	else
+		return 0;
+}
+
+ssize_t jbin(char *s, char *buf, size_t len)
+{
+	int i, tmp;
+	uint32_t code;
+	if (!s)
+		return -1;
+	s = skip_ws(s);
+	if (*s != '"')
+		return -1;
+	for (i = 0, s ++; *s && *s != '"' && i+1 < len;)
+		if (*s == '\\')
+		{
+			if (!(tmp = parse_escape(++s, &code)))
+				return -1;
+			s += tmp;
+			if (i+utf8_len(code) >= len)
+				return -1;
+			i += utf8_enc(code, buf+i);
+		}
+		else
+			buf[i ++] = *(s ++);
+	if (*s != '"')
+		return -1;
+	return i;
+}
+
+char *jstrb(char *s, char *err, char *buf)
+{
+	ssize_t res;
+	res = jbin(s, buf, JSTR_SIZE);
+	if (res == -1 || res+1 >= JSTR_SIZE)
+		return err;
+	buf[res] = 0;
+	return buf;
+}
+
+char *jstr(char *s, char *err)
+{
+	static __thread char buf[JSTR_SIZE];
+	return jstrb(s, err, buf);
+}
+
+int jint(char *s, int err)
+{
+	if (!s)
+		return err;
+	if (!strncmp(s, SL("true")))
+		return 1;
+	if (!strncmp(s, SL("false")))
+		return 0;
+	if (!strncmp(s, SL("null")))
+		return 0;
+	return strtol(s, 0, 10);
+}
+
+char *jenter(char *s)
+{
+	if (!s)
+		return 0;
+	s = skip_ws(s);
+	if (*s == '[' || *s == '{')
+		return s+1;
+	return 0;
+}
+
+char *jnext(char *s)
+{
+	if (!s)
+		return 0;
+	s = skip_ws(skip(s));
+	if (*s == ':')
+		s = skip_ws(skip(s+1));
+	if (*s == ',')
+		return s+1;
+	else
+		return 0;
+}
+
+char *jvalue(char *s)
+{
+	if (!s)
+		return 0;
+	s = skip_ws(skip(s));
+	if (*s != ':')
+		return 0;
+	return s+1;
+}
+
+char *jlookup(char *s, char *key)
+{
+	static __thread char buf[JSTR_SIZE];
+	if (!s)
+		return 0;
+	s = skip_ws(s);
+	if (*s != '{')
+		return 0;
+	for (s = skip_ws(s+1); s; s = jnext(s))
+		if (!strcmp(jstrb(s, "INVALID KEY", buf), key))
+			return jvalue(s);
+	return 0;
+}

util/xprintf.c

+#include <stdio.h>
+#include <stdarg.h>
+
+#include "../util.h"
+
+size_t xvsnprintf(char *p, size_t len, const char *fmt, va_list ap)
+{
+	int ret;
+	if ((ret = vsnprintf(p, len, fmt, ap)) >= len)
+	{
+		p[len-1] = 0;
+		return len;
+	}
+	return ret;
+}
+
+size_t xsnprintf(char *p, size_t len, const char *fmt, ...)
+{
+	size_t ret;
+	va_list ap;
+	va_start(ap, fmt);
+	ret = vsnprintf(p, len, fmt, ap);
+	va_end(ap);
+	return ret;
+}