Bump version: 4.4 → 4.5

Fix h264 rtp packetization - the marked bit should only be set for th…

.bumpversion.cfg

@@ -1,14 +1,22 @@
 [bumpversion]
 commit = True
 tag = True
-current_version = 3.21
-parse = (?P<major>\d+)\.(?P<minor>\d+)(\.(?P<patch>\d+)(\-(?P<release>[a-z]+))?)?
+current_version = 4.5
+parse = (?P<major>\d+)\.(?P<minor>\d+)
 serialize = 
 	{major}.{minor}
 
 [bumpversion:file:src/libs/config.h]
-search = VERSION "{current_version}"
-replace = VERSION "{new_version}"
+parse = (?P<major>\d+)
+serialize = {major}
+search = VERSION_MAJOR {current_version}
+replace = VERSION_MAJOR {new_version}
+
+[bumpversion:file:./src/libs/config.h]
+parse = <major>\d+\.(?P<minor>\d+)
+serialize = {minor}
+search = VERSION_MINOR {current_version}
+replace = VERSION_MINOR {new_version}
 
 [bumpversion:file:python/setup.py]
 search = version="{current_version}"

.dockerignore

@@ -4,3 +4,6 @@
 # Allow source code
 !Makefile
 !src/**
+!python/**
+!janus/**
+!man/**

.editorconfig

@@ -0,0 +1,10 @@
+root = true
+
+[*]
+end_of_file = lf
+indent_style = tab
+indent_size = 4
+
+[*.{py,yaml}]
+indent_style = space
+indent_size = 4

.gitignore

@@ -2,14 +2,16 @@
 /linters/.mypy_cache/
 /pkg/arch/pkg/
 /pkg/arch/src/
-/pkg/arch/v*.tar.gz
-/pkg/arch/ustreamer-*.pkg.tar.xz
-/pkg/arch/ustreamer-*.pkg.tar.zst
-/build/
+/src/build/
 /python/build/
-/config.mk
-/vgcore.*
+/janus/build/
 /ustreamer
 /ustreamer-dump
-/*.so
-/*.sock
+/config.mk
+vgcore.*
+*.sock
+*.so
+*.bin
+*.pkg.tar.xz
+*.pkg.tar.zst
+*.tar.gz

Makefile

@@ -1,7 +1,5 @@
 -include config.mk
 
-USTR ?= ustreamer
-DUMP ?= ustreamer-dump
 DESTDIR ?=
 PREFIX ?= /usr/local
 MANPREFIX ?= $(PREFIX)/share/man
@@ -14,147 +12,82 @@ LDFLAGS ?=
 RPI_VC_HEADERS ?= /opt/vc/include
 RPI_VC_LIBS ?= /opt/vc/lib
 
-BUILD ?= build
+export
 
-LINTERS_IMAGE ?= $(USTR)-linters
+_LINTERS_IMAGE ?= ustreamer-linters
 
 
 # =====
-_CFLAGS = -MD -c -std=c11 -Wall -Wextra -D_GNU_SOURCE $(CFLAGS)
-_LDFLAGS = $(LDFLAGS)
-
-_COMMON_LIBS = -lm -ljpeg -pthread -lrt
-
-_USTR_LIBS = $(_COMMON_LIBS) -levent -levent_pthreads
-_USTR_SRCS = $(shell ls \
-	src/libs/*.c \
-	src/ustreamer/*.c \
-	src/ustreamer/http/*.c \
-	src/ustreamer/data/*.c \
-	src/ustreamer/encoders/cpu/*.c \
-	src/ustreamer/encoders/hw/*.c \
-)
-
-_DUMP_LIBS = $(_COMMON_LIBS)
-_DUMP_SRCS = $(shell ls \
-	src/libs/*.c \
-	src/dump/*.c \
-)
-
-
 define optbool
 $(filter $(shell echo $(1) | tr A-Z a-z), yes on 1)
 endef
 
 
-ifneq ($(call optbool,$(WITH_OMX)),)
-_USTR_LIBS += -lbcm_host -lvcos -lvcsm -lopenmaxil -lmmal -lmmal_core -lmmal_util -lmmal_vc_client -lmmal_components -L$(RPI_VC_LIBS)
-override _CFLAGS += -DWITH_OMX -DOMX_SKIP64BIT -I$(RPI_VC_HEADERS)
-_USTR_SRCS += $(shell ls \
-	src/ustreamer/encoders/omx/*.c \
-	src/ustreamer/h264/*.c \
-)
-endif
-
-
-ifneq ($(call optbool,$(WITH_GPIO)),)
-_USTR_LIBS += -lgpiod
-override _CFLAGS += -DWITH_GPIO
-_USTR_SRCS += $(shell ls src/ustreamer/gpio/*.c)
-endif
-
-
-WITH_PTHREAD_NP ?= 1
-ifneq ($(call optbool,$(WITH_PTHREAD_NP)),)
-override _CFLAGS += -DWITH_PTHREAD_NP
-endif
-
-
-WITH_SETPROCTITLE ?= 1
-ifneq ($(call optbool,$(WITH_SETPROCTITLE)),)
-ifeq ($(shell uname -s | tr A-Z a-z),linux)
-_USTR_LIBS += -lbsd
-endif
-override _CFLAGS += -DWITH_SETPROCTITLE
-endif
-
-
 # =====
-all: $(USTR) $(DUMP) python
-
-
-install: all
-	mkdir -p $(DESTDIR)$(PREFIX)/bin $(DESTDIR)$(MANPREFIX)/man1
-	install -m755 $(USTR) $(DESTDIR)$(PREFIX)/bin/$(USTR)
-	install -m755 $(DUMP) $(DESTDIR)$(PREFIX)/bin/$(DUMP)
-	install -m644 man/$(USTR).1 $(DESTDIR)$(MANPREFIX)/man1/$(USTR).1
-	install -m644 man/$(DUMP).1 $(DESTDIR)$(MANPREFIX)/man1/$(DUMP).1
-	gzip -f $(DESTDIR)$(MANPREFIX)/man1/$(USTR).1
-	gzip -f $(DESTDIR)$(MANPREFIX)/man1/$(DUMP).1
+all:
+	+ $(MAKE) apps
 ifneq ($(call optbool,$(WITH_PYTHON)),)
-	cd python && $(PY) setup.py install --prefix=$(PREFIX) --root=$(if $(DESTDIR),$(DESTDIR),/)
+	+ $(MAKE) python
+endif
+ifneq ($(call optbool,$(WITH_JANUS)),)
+	+ $(MAKE) janus
 endif
 
 
-install-strip: install
-	strip $(DESTDIR)$(PREFIX)/bin/$(USTR)
-	strip $(DESTDIR)$(PREFIX)/bin/$(DUMP)
-
-
-regen:
-	tools/make-jpeg-h.py src/ustreamer/data/blank.jpeg src/ustreamer/data/blank_jpeg.c BLANK
-	tools/make-html-h.py src/ustreamer/data/index.html src/ustreamer/data/index_html.c INDEX
-
-
-$(USTR): $(_USTR_SRCS:%.c=$(BUILD)/%.o)
-#	$(info ========================================)
-	$(info == LD $@)
-	@ $(CC) $^ -o $@ $(_LDFLAGS) $(_USTR_LIBS)
-#	$(info :: CC      = $(CC))
-#	$(info :: LIBS    = $(_USTR_LIBS))
-#	$(info :: CFLAGS  = $(_CFLAGS))
-#	$(info :: LDFLAGS = $(_LDFLAGS))
-
-
-$(DUMP): $(_DUMP_SRCS:%.c=$(BUILD)/%.o)
-#	$(info ========================================)
-	$(info == LD $@)
-	@ $(CC) $^ -o $@ $(_LDFLAGS) $(_DUMP_LIBS)
-#	$(info :: CC      = $(CC))
-#	$(info :: LIBS    = $(_DUMP_LIBS))
-#	$(info :: CFLAGS  = $(_CFLAGS))
-#	$(info :: LDFLAGS = $(_LDFLAGS))
-
-
-$(BUILD)/%.o: %.c
-	$(info -- CC $<)
-	@ mkdir -p $(dir $@) || true
-	@ $(CC) $< -o $@ $(_CFLAGS)
+apps:
+	$(MAKE) -C src
+	@ ln -sf src/ustreamer.bin ustreamer
+	@ ln -sf src/ustreamer-dump.bin ustreamer-dump
 
 
 python:
-ifneq ($(call optbool,$(WITH_PYTHON)),)
-	$(info == PY_BUILD ustreamer-*.so)
-	@ cd python && $(PY) setup.py build
+	$(MAKE) -C python
 	@ ln -sf python/build/lib.*/*.so .
-else
-	@ true
+
+
+janus:
+	$(MAKE) -C janus
+	@ ln -sf janus/*.so .
+
+
+install: all
+	$(MAKE) -C src install
+ifneq ($(call optbool,$(WITH_PYTHON)),)
+	$(MAKE) -C python install
 endif
+ifneq ($(call optbool,$(WITH_JANUS)),)
+	$(MAKE) -C janus install
+endif
+	mkdir -p $(DESTDIR)$(MANPREFIX)/man1
+	for man in $(shell ls man); do \
+		install -m644 man/$$man $(DESTDIR)$(MANPREFIX)/man1/$$man; \
+		gzip -f $(DESTDIR)$(MANPREFIX)/man1/$$man; \
+	done
+
+
+install-strip: install
+	$(MAKE) -C src install-strip
+
+
+regen:
+	tools/$(MAKE)-jpeg-h.py src/ustreamer/data/blank.jpeg src/ustreamer/data/blank_jpeg.c BLANK
+	tools/$(MAKE)-html-h.py src/ustreamer/data/index.html src/ustreamer/data/index_html.c INDEX
+
 
 release:
-	make clean
-	make tox
-	make push
-	make bump V=$(V)
-	make push
-	make clean
+	$(MAKE) clean
+	$(MAKE) tox
+	$(MAKE) push
+	$(MAKE) bump V=$(V)
+	$(MAKE) push
+	$(MAKE) clean
 
 
 tox: linters
 	time docker run --rm \
 			--volume `pwd`:/src:ro \
 			--volume `pwd`/linters:/src/linters:rw \
-		-t $(LINTERS_IMAGE) bash -c " \
+		-t $(_LINTERS_IMAGE) bash -c " \
 			cd /src \
 			&& tox -q -c linters/tox.ini $(if $(E),-e $(E),-p auto) \
 		"
@@ -164,7 +97,7 @@ linters:
 	docker build \
 			$(if $(call optbool,$(NC)),--no-cache,) \
 			--rm \
-			--tag $(LINTERS_IMAGE) \
+			--tag $(_LINTERS_IMAGE) \
 		-f linters/Dockerfile linters
 
 
@@ -180,14 +113,15 @@ push:
 clean-all: linters clean
 	- docker run --rm \
 			--volume `pwd`:/src \
-		-it $(LINTERS_IMAGE) bash -c "cd src && rm -rf linters/{.tox,.mypy_cache}"
+		-it $(_LINTERS_IMAGE) bash -c "cd src && rm -rf linters/{.tox,.mypy_cache}"
+
+
 clean:
 	rm -rf pkg/arch/pkg pkg/arch/src pkg/arch/v*.tar.gz pkg/arch/ustreamer-*.pkg.tar.{xz,zst}
-	rm -rf $(USTR) $(DUMP) $(BUILD) python/build vgcore.* *.sock *.so
+	rm -f ustreamer ustreamer-dump *.so
+	$(MAKE) -C src clean
+	$(MAKE) -C python clean
+	$(MAKE) -C janus clean
 
 
-.PHONY: python linters
-
-
-_OBJS = $(_USTR_SRCS:%.c=$(BUILD)/%.o) $(_DUMP_SRCS:%.c=$(BUILD)/%.o)
--include $(_OBJS:%.o=%.d)
+.PHONY: python janus linters

README.md

@@ -38,8 +38,7 @@ You'll need  ```make```, ```gcc```, ```libevent``` with ```pthreads``` support,
 
 * Arch: `sudo pacman -S libevent libjpeg-turbo libutil-linux libbsd`.
 * Raspbian: `sudo apt install libevent-dev libjpeg8-dev libbsd-dev`. Add `libraspberrypi-dev` for `WITH_OMX=1` and `libgpiod` for `WITH_GPIO=1`.
-* Debian: `sudo apt install build-essential libevent-dev libjpeg62-turbo-dev libbsd-dev`.
-* Ubuntu 20.04 x86_64: `sudo apt install build-essential libevent-dev libjpeg-dev libjpeg62-dev libbsd-dev make gcc libjpeg8 libjpeg-turbo8 libbsd0`.
+* Debian/Ubuntu: `sudo apt install build-essential libevent-dev libjpeg-dev libbsd-dev`.
 
 On Raspberry Pi you can build the program with OpenMAX IL. To do this pass option ```WITH_OMX=1``` to ```make```. To enable GPIO support install [libgpiod](https://git.kernel.org/pub/scm/libs/libgpiod/libgpiod.git/about) and pass option ```WITH_GPIO=1```. If the compiler reports about a missing function ```pthread_get_name_np()``` (or similar), add option ```WITH_PTHREAD_NP=0``` (it's enabled by default). For the similar error with ```setproctitle()``` add option ```WITH_SETPROCTITLE=0```.
 
@@ -100,8 +99,8 @@ v4l2 utilities provide the tools to manage USB webcam setting and information. S
 * List of available video devices: `v4l2-ctl --list-devices`.
 * List available control settings: `v4l2-ctl -d /dev/video0 --list-ctrls`.
 * List available video formats: `v4l2-ctl -d /dev/video0 --list-formats-ext`.
-* Read the current setting: `v4l2-ctl d /dev/video0 --get-ctrl=exposure_auto`.
-* Change the setting value: `v4l2-ctl d /dev/video0 --set-ctrl=exposure_auto=1`.
+* Read the current setting: `v4l2-ctl -d /dev/video0 --get-ctrl=exposure_auto`.
+* Change the setting value: `v4l2-ctl -d /dev/video0 --set-ctrl=exposure_auto=1`.
 
 [Here](https://www.kurokesu.com/main/2016/01/16/manual-usb-camera-settings-in-linux/) you can find more examples. Documentation is available in [`man v4l2-ctl`](https://www.mankier.com/1/v4l2-ctl).
 

README.ru.md

@@ -38,8 +38,7 @@
 
 * Arch: `sudo pacman -S libevent libjpeg-turbo libutil-linux libbsd`.
 * Raspbian: `sudo apt install libevent-dev libjpeg8-dev libbsd-dev`.  Добавьте `libraspberrypi-dev` для сборки с `WITH_OMX=1` и `libgpiod` для `WITH_GPIO=1`.
-* Debian: `sudo apt install build-essential libevent-dev libjpeg62-turbo-dev libbsd-dev`.
-* Ubuntu 20.04 x86_64: `sudo apt install build-essential libevent-dev libjpeg-dev libjpeg62-dev libbsd-dev make gcc libjpeg8 libjpeg-turbo8 libbsd0`.
+* Debian/Ubuntu: `sudo apt install build-essential libevent-dev libjpeg-dev libbsd-dev`.
 
 На Raspberry Pi программу можно собрать с поддержкой OpenMAX IL. Для этого передайте ```make``` параметр ```WITH_OMX=1```. Для включения сборки с поддержкой GPIO установите [libgpiod](https://git.kernel.org/pub/scm/libs/libgpiod/libgpiod.git/about) и добавьте параметр ```WITH_GPIO=1```. Если при сборке компилятор ругается на отсутствие функции ```pthread_get_name_np()``` или другой подобной, добавьте параметр ```WITH_PTHREAD_NP=0``` (по умолчанию он включен). При аналогичной ошибке с функцией ```setproctitle()``` добавьте параметр ```WITH_SETPROCTITLE=0```.
 
@@ -100,8 +99,8 @@ V4L2 предоставляет ряд официальных утилит дл
 * Вывести список видеоустройств: `v4l2-ctl --list-devices`.
 * Вывести список доступных контролов устройства: `v4l2-ctl -d /dev/video0 --list-ctrls`.
 * Вывести список доступных форматов видео: `v4l2-ctl -d /dev/video0 --list-formats-ext`.
-* Показать текущее значение контрола: `v4l2-ctl d /dev/video0 --get-ctrl=exposure_auto`.
-* Изменить значение контрола: `v4l2-ctl d /dev/video0 --set-ctrl=exposure_auto=1`.
+* Показать текущее значение контрола: `v4l2-ctl -d /dev/video0 --get-ctrl=exposure_auto`.
+* Изменить значение контрола: `v4l2-ctl -d /dev/video0 --set-ctrl=exposure_auto=1`.
 
 Больше примеров вы можете найти [здесь](https://www.kurokesu.com/main/2016/01/16/manual-usb-camera-settings-in-linux/), а документацию в [`man v4l2-ctl`](https://www.mankier.com/1/v4l2-ctl).
 

janus/Makefile

@@ -0,0 +1,54 @@
+DESTDIR ?=
+PREFIX ?= /usr/local
+
+CC ?= gcc
+CFLAGS ?= -O3
+LDFLAGS ?=
+
+
+# =====
+_PLUGIN = libjanus_ustreamer.so
+
+_CFLAGS = -fPIC -MD -c -std=c11 -Wall -Wextra -D_GNU_SOURCE $(shell pkg-config --cflags glib-2.0) $(CFLAGS)
+_LDFLAGS = -shared -lm -pthread -lrt -ljansson $(shell pkg-config --libs glib-2.0) $(LDFLAGS)
+
+_SRCS = $(shell ls src/*.c)
+
+_BUILD = build
+
+
+define optbool
+$(filter $(shell echo $(1) | tr A-Z a-z), yes on 1)
+endef
+
+
+WITH_PTHREAD_NP ?= 1
+ifneq ($(call optbool,$(WITH_PTHREAD_NP)),)
+override _CFLAGS += -DWITH_PTHREAD_NP
+endif
+
+
+# =====
+$(_PLUGIN): $(_SRCS:%.c=$(_BUILD)/%.o)
+	$(info == SO $@)
+	@ $(CC) $^ -o $@ $(_LDFLAGS)
+
+
+$(_BUILD)/%.o: %.c
+	$(info -- CC $<)
+	@ mkdir -p $(dir $@) || true
+	@ $(CC) $< -o $@ $(_CFLAGS)
+
+
+
+install: $(_PLUGIN)
+	mkdir -p $(DESTDIR)$(PREFIX)/lib/ustreamer/janus
+	install -m755 $(_PLUGIN) $(DESTDIR)$(PREFIX)/lib/ustreamer/janus/$(PLUGIN)
+
+
+clean:
+	rm -rf $(_PLUGIN) $(_BUILD)
+
+
+_OBJS = $(_SRCS:%.c=$(_BUILD)/%.o)
+-include $(_OBJS:%.o=%.d)

janus/src/base64.c

@@ -0,0 +1 @@
+../../src/libs/base64.c

janus/src/base64.h

@@ -0,0 +1 @@
+../../src/libs/base64.h

janus/src/config.h

@@ -0,0 +1 @@
+../../src/libs/config.h

janus/src/frame.c

@@ -0,0 +1 @@
+../../src/libs/frame.c

janus/src/frame.h

@@ -0,0 +1 @@
+../../src/libs/frame.h

janus/src/list.h

@@ -0,0 +1 @@
+../../src/libs/list.h

janus/src/memsinksh.h

@@ -0,0 +1 @@
+../../src/libs/memsinksh.h

janus/src/plugin.c

@@ -0,0 +1,522 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPG-HTTP streamer.                    #
+#                                                                            #
+#    Copyright (C) 2018-2021  Maxim Devaev <mdevaev@gmail.com>               #
+#                                                                            #
+#    This program is free software: you can redistribute it and/or modify    #
+#    it under the terms of the GNU General Public License as published by    #
+#    the Free Software Foundation, either version 3 of the License, or       #
+#    (at your option) any later version.                                     #
+#                                                                            #
+#    This program is distributed in the hope that it will be useful,         #
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of          #
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the           #
+#    GNU General Public License for more details.                            #
+#                                                                            #
+#    You should have received a copy of the GNU General Public License       #
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.  #
+#                                                                            #
+*****************************************************************************/
+
+
+#include <stdint.h>
+#include <stdbool.h>
+#include <stdatomic.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <assert.h>
+
+#include <sys/mman.h>
+#include <sys/stat.h>
+#include <linux/videodev2.h>
+
+#include <pthread.h>
+#include <jansson.h>
+#include <janus/config.h>
+#include <janus/plugins/plugin.h>
+
+#include "config.h"
+#include "tools.h"
+#include "threading.h"
+#include "list.h"
+#include "memsinksh.h"
+
+#include "rtp.h"
+
+
+static int _plugin_init(janus_callbacks *gw, const char *config_file_path);
+static void _plugin_destroy(void);
+
+static void _plugin_create_session(janus_plugin_session *session, int *error);
+static void _plugin_destroy_session(janus_plugin_session *session, int *error);
+static json_t *_plugin_query_session(janus_plugin_session *session);
+
+static void _plugin_setup_media(janus_plugin_session *session);
+static void _plugin_hangup_media(janus_plugin_session *session);
+
+static struct janus_plugin_result *_plugin_handle_message(
+	janus_plugin_session *session, char *transaction, json_t *msg, json_t *jsep);
+
+static int _plugin_get_api_compatibility(void);
+static int _plugin_get_version(void);
+static const char *_plugin_get_version_string(void);
+static const char *_plugin_get_description(void);
+static const char *_plugin_get_name(void);
+static const char *_plugin_get_author(void);
+static const char *_plugin_get_package(void);
+
+
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Woverride-init"
+static janus_plugin _plugin = JANUS_PLUGIN_INIT(
+	.init = _plugin_init,
+	.destroy = _plugin_destroy,
+
+	.create_session = _plugin_create_session,
+	.destroy_session = _plugin_destroy_session,
+	.query_session = _plugin_query_session,
+
+	.setup_media = _plugin_setup_media,
+	.hangup_media = _plugin_hangup_media,
+
+	.handle_message = _plugin_handle_message,
+
+	.get_api_compatibility = _plugin_get_api_compatibility,
+	.get_version = _plugin_get_version,
+	.get_version_string = _plugin_get_version_string,
+	.get_description = _plugin_get_description,
+	.get_name = _plugin_get_name,
+	.get_author = _plugin_get_author,
+	.get_package = _plugin_get_package,
+);
+#pragma GCC diagnostic pop
+
+janus_plugin *create(void) { // cppcheck-suppress unusedFunction
+	return &_plugin;
+}
+
+
+typedef struct _client_sx {
+	janus_plugin_session *session;
+	bool transmit;
+
+	LIST_STRUCT(struct _client_sx);
+} _client_s;
+
+
+static char				*_g_memsink_obj = NULL;
+const long double		_g_wait_timeout = 1;
+const long double		_g_lock_timeout = 1;
+const useconds_t		_g_lock_polling = 1000;
+const useconds_t		_g_watchers_polling = 100000;
+
+static _client_s		*_g_clients = NULL;
+static janus_callbacks	*_g_gw = NULL;
+static rtp_s			*_g_rtp = NULL;
+
+static pthread_t		_g_tid;
+static pthread_mutex_t	_g_lock = PTHREAD_MUTEX_INITIALIZER;
+static atomic_bool		_g_ready = ATOMIC_VAR_INIT(false);
+static atomic_bool		_g_stop = ATOMIC_VAR_INIT(false);
+static atomic_bool		_g_has_watchers = ATOMIC_VAR_INIT(false);
+
+
+#define JLOG_INFO(_msg, ...)	JANUS_LOG(LOG_INFO, "== %s -- " _msg "\n", _plugin_get_name(), ##__VA_ARGS__)
+#define JLOG_WARN(_msg, ...)	JANUS_LOG(LOG_WARN, "== %s -- " _msg "\n", _plugin_get_name(), ##__VA_ARGS__)
+#define JLOG_ERROR(_msg, ...)	JANUS_LOG(LOG_ERR, "== %s -- " _msg "\n", _plugin_get_name(), ##__VA_ARGS__)
+
+#define JLOG_PERROR(_msg, ...) { \
+		char _perror_buf[1024] = {0}; \
+		char *_perror_ptr = errno_to_string(errno, _perror_buf, 1023); \
+		JANUS_LOG(LOG_ERR, "[%s] " _msg ": %s\n", _plugin_get_name(), ##__VA_ARGS__, _perror_ptr); \
+	}
+
+#define LOCK			A_MUTEX_LOCK(&_g_lock)
+#define UNLOCK			A_MUTEX_UNLOCK(&_g_lock)
+#define READY			atomic_load(&_g_ready)
+#define STOP			atomic_load(&_g_stop)
+#define HAS_WATCHERS	atomic_load(&_g_has_watchers)
+
+
+static void _relay_rtp_clients(const uint8_t *datagram, size_t size) {
+	janus_plugin_rtp packet = {
+		.video = true,
+		.buffer = (char *)datagram,
+		.length = size,
+	};
+	janus_plugin_rtp_extensions_reset(&packet.extensions);
+	LIST_ITERATE(_g_clients, client, {
+		if (client->transmit) {
+			_g_gw->relay_rtp(client->session, &packet);
+		}
+	});
+}
+
+static int _wait_frame(int fd, memsink_shared_s* mem, uint64_t last_id) {
+	long double deadline_ts = get_now_monotonic() + _g_wait_timeout;
+	long double now;
+	do {
+		int retval = flock_timedwait_monotonic(fd, _g_lock_timeout);
+		now = get_now_monotonic();
+		if (retval < 0 && errno != EWOULDBLOCK) {
+			JLOG_PERROR("Can't lock memsink");
+			return -1;
+		} else if (retval == 0) {
+			if (mem->magic == MEMSINK_MAGIC && mem->version == MEMSINK_VERSION && mem->id != last_id) {
+				return 0;
+			}
+			if (flock(fd, LOCK_UN) < 0) {
+				JLOG_PERROR("Can't unlock memsink");
+				return -1;
+			}
+		}
+		usleep(_g_lock_polling);
+	} while (now < deadline_ts);
+	return -2;
+}
+
+static int _get_frame(int fd, memsink_shared_s *mem, frame_s *frame, uint64_t *frame_id) {
+	frame_set_data(frame, mem->data, mem->used);
+	FRAME_COPY_META(mem, frame);
+	*frame_id = mem->id;
+	mem->last_client_ts = get_now_monotonic();
+	int retval = 0;
+	if (frame->format != V4L2_PIX_FMT_H264) {
+		JLOG_ERROR("Got non-H264 frame from memsink");
+		retval = -1;
+	}
+	if (flock(fd, LOCK_UN) < 0) {
+		JLOG_PERROR("Can't unlock memsink");
+		retval = -1;
+	}
+	return retval;
+}
+
+static void *_clients_thread(UNUSED void *arg) {
+	A_THREAD_RENAME("us_clients");
+	atomic_store(&_g_ready, true);
+
+	frame_s *frame = frame_init();
+	uint64_t frame_id = 0;
+
+	unsigned error_reported = 0;
+
+#	define IF_NOT_REPORTED(_error, ...) { \
+			if (error_reported != _error) { __VA_ARGS__; error_reported = _error; } \
+		}
+
+	while (!STOP) {
+		if (!HAS_WATCHERS) {
+			IF_NOT_REPORTED(1, {
+				JLOG_INFO("No active watchers, memsink disconnected");
+			});
+			usleep(_g_watchers_polling);
+			continue;
+		}
+
+		int fd = -1;
+		memsink_shared_s *mem = NULL;
+
+		if ((fd = shm_open(_g_memsink_obj, O_RDWR, 0)) <= 0) {
+			IF_NOT_REPORTED(2, {
+				JLOG_PERROR("Can't open memsink");
+			});
+			goto close_memsink;
+		}
+
+		if ((mem = memsink_shared_map(fd)) == NULL) {
+			IF_NOT_REPORTED(3, {
+				JLOG_PERROR("Can't map memsink");
+			});
+			goto close_memsink;
+		}
+
+		error_reported = 0;
+
+		JLOG_INFO("Memsink opened; reading frames ...");
+		while (!STOP && HAS_WATCHERS) {
+			int result = _wait_frame(fd, mem, frame_id);
+			if (result == 0) {
+				if (_get_frame(fd, mem, frame, &frame_id) != 0) {
+					goto close_memsink;
+				}
+				LOCK;
+				rtp_wrap_h264(_g_rtp, frame, _relay_rtp_clients);
+				UNLOCK;
+			} else if (result == -1) {
+				goto close_memsink;
+			}
+		}
+
+		close_memsink:
+			if (mem != NULL) {
+				JLOG_INFO("Memsink closed");
+				memsink_shared_unmap(mem);
+				mem = NULL;
+			}
+			if (fd > 0) {
+				close(fd);
+				fd = -1;
+			}
+			sleep(1); // error_delay
+	}
+
+#	undef IF_NOT_REPORTED
+
+	frame_destroy(frame);
+	return NULL;
+}
+
+static int _read_config(const char *config_dir_path) {
+	char *config_file_path;
+	janus_config *config = NULL;
+
+	A_ASPRINTF(config_file_path, "%s/%s.jcfg", config_dir_path, _plugin_get_package());
+	JLOG_INFO("Reading config file '%s' ...", config_file_path);
+
+	config = janus_config_parse(config_file_path);
+	if (config == NULL) {
+		JLOG_ERROR("Can't read config");
+		goto error;
+	}
+	janus_config_print(config);
+
+	janus_config_category *config_memsink = janus_config_get_create(config, NULL, janus_config_type_category, "memsink");
+	janus_config_item *config_memsink_obj = janus_config_get(config, config_memsink, janus_config_type_item, "object");
+	if (config_memsink_obj == NULL || config_memsink_obj->value == NULL || config_memsink_obj->value[0] == '\0') {
+		JLOG_ERROR("Missing config value: memsink.object");
+		goto error;
+	}
+	_g_memsink_obj = strdup(config_memsink_obj->value);
+
+	int retval = 0;
+	goto ok;
+	error:
+		retval = -1;
+	ok:
+		if (config) {
+			janus_config_destroy(config);
+		}
+		free(config_file_path);
+		return retval;
+}
+
+static int _plugin_init(janus_callbacks *gw, const char *config_dir_path) {
+	// https://groups.google.com/g/meetecho-janus/c/xoWIQfaoJm8
+	// sysctl -w net.core.rmem_default=500000 
+	// sysctl -w net.core.wmem_default=500000 
+	// sysctl -w net.core.rmem_max=1000000 
+	// sysctl -w net.core.wmem_max=1000000
+
+	JLOG_INFO("Initializing plugin ...");
+	assert(!READY);
+	assert(!STOP);
+	if (gw == NULL || config_dir_path == NULL || _read_config(config_dir_path) < 0) {
+		return -1;
+	}
+	_g_gw = gw;
+	_g_rtp = rtp_init();
+	A_THREAD_CREATE(&_g_tid, _clients_thread, NULL);
+	return 0;
+}
+
+static void _plugin_destroy(void) {
+	JLOG_INFO("Destroying plugin ...");
+	atomic_store(&_g_stop, true);
+	if (READY) {
+		A_THREAD_JOIN(_g_tid);
+	}
+
+	LIST_ITERATE(_g_clients, client, {
+		LIST_REMOVE(_g_clients, client);
+		free(client);
+	});
+	_g_clients = NULL;
+
+	rtp_destroy(_g_rtp);
+	_g_rtp = NULL;
+
+	_g_gw = NULL;
+
+	if (_g_memsink_obj) {
+		free(_g_memsink_obj);
+		_g_memsink_obj = NULL;
+	}
+}
+
+#define IF_DISABLED(...) { if (!READY || STOP) { __VA_ARGS__ } }
+
+static void _plugin_create_session(janus_plugin_session *session, int *error) {
+	IF_DISABLED({ *error = -1; return; });
+	LOCK;
+	JLOG_INFO("Creating session %p ...", session);
+	_client_s *client;
+	A_CALLOC(client, 1);
+	client->session = session;
+	client->transmit = true;
+	LIST_APPEND(_g_clients, client);
+	atomic_store(&_g_has_watchers, true);
+	UNLOCK;
+}
+
+static void _plugin_destroy_session(janus_plugin_session* session, int *error) {
+	IF_DISABLED({ *error = -1; return; });
+	LOCK;
+	bool found = false;
+	bool has_watchers = false;
+	LIST_ITERATE(_g_clients, client, {
+		if (client->session == session) {
+			JLOG_INFO("Removing session %p ...", session);
+			LIST_REMOVE(_g_clients, client);
+			free(client);
+			found = true;
+		} else {
+			has_watchers = (has_watchers || client->transmit);
+		}
+	});
+	if (!found) {
+		JLOG_WARN("No session %p", session);
+		*error = -2;
+	}
+	atomic_store(&_g_has_watchers, has_watchers);
+	UNLOCK;
+}
+
+static json_t *_plugin_query_session(janus_plugin_session *session) {
+	IF_DISABLED({ return NULL; });
+	json_t *info = NULL;
+	LOCK;
+	LIST_ITERATE(_g_clients, client, {
+		if (client->session == session) {
+			info = json_string("session_found");
+			break;
+		}
+	});
+	UNLOCK;
+	return info;
+}
+
+static void _set_transmit(janus_plugin_session *session, UNUSED const char *msg, bool transmit) {
+	IF_DISABLED({ return; });
+	LOCK;
+	bool found = false;
+	bool has_watchers = false;
+	LIST_ITERATE(_g_clients, client, {
+		if (client->session == session) {
+			client->transmit = transmit;
+			//JLOG_INFO("%s session %p", msg, session);
+			found = true;
+		}
+		has_watchers = (has_watchers || client->transmit);
+	});
+	if (!found) {
+		JLOG_WARN("No session %p", session);
+	}
+	atomic_store(&_g_has_watchers, has_watchers);
+	UNLOCK;
+}
+
+#undef IF_DISABLED
+
+static void _plugin_setup_media(janus_plugin_session *session) { _set_transmit(session, "Unmuted", true); }
+static void _plugin_hangup_media(janus_plugin_session *session) { _set_transmit(session, "Muted", false); }
+
+static struct janus_plugin_result *_plugin_handle_message(
+	janus_plugin_session *session, char *transaction, json_t *msg, json_t *jsep) {
+
+	assert(transaction != NULL);
+
+#	define FREE_MSG_JSEP { \
+			if (msg) json_decref(msg); \
+			if (jsep) json_decref(jsep); \
+		}
+
+	if (session == NULL || msg == NULL) {
+		free(transaction);
+		FREE_MSG_JSEP;
+		return janus_plugin_result_new(JANUS_PLUGIN_ERROR, (msg ? "No session" : "No message"), NULL);
+	}
+
+#	define PUSH_ERROR(_error, _reason) { \
+			/*JLOG_ERROR("Message error in session %p: %s", session, _reason);*/ \
+			json_t *_event = json_object(); \
+			json_object_set_new(_event, "ustreamer", json_string("event")); \
+			json_object_set_new(_event, "error_code", json_integer(_error)); \
+			json_object_set_new(_event, "error", json_string(_reason)); \
+			_g_gw->push_event(session, &_plugin, transaction, _event, NULL); \
+			json_decref(_event); \
+		}
+
+	json_t *request_obj = json_object_get(msg, "request");
+	if (request_obj == NULL) {
+		PUSH_ERROR(400, "Request missing");
+		goto ok_wait;
+	}
+
+	const char *request_str = json_string_value(request_obj);
+	if (!request_str) {
+		PUSH_ERROR(400, "Request not a string");
+		goto ok_wait;
+	}
+	//JLOG_INFO("Message: %s", request_str);
+
+#	define PUSH_STATUS(_status, _jsep) { \
+			json_t *_event = json_object(); \
+			json_object_set_new(_event, "ustreamer", json_string("event")); \
+			json_t *_result = json_object(); \
+			json_object_set_new(_result, "status", json_string(_status)); \
+			json_object_set_new(_event, "result", _result); \
+			_g_gw->push_event(session, &_plugin, transaction, _event, _jsep); \
+			json_decref(_event); \
+		}
+
+	if (!strcmp(request_str, "start")) {
+		PUSH_STATUS("started", NULL);
+
+	} else if (!strcmp(request_str, "stop")) {
+		PUSH_STATUS("stopped", NULL);
+
+	} else if (!strcmp(request_str, "watch")) {
+		char *sdp = rtp_make_sdp(_g_rtp);
+		if (sdp == NULL) {
+			PUSH_ERROR(503, "Haven't received SPS/PPS from memsink yet");
+			goto ok_wait;
+		}
+		//JLOG_INFO("SDP generated:\n%s", sdp);
+		json_t *offer_jsep = json_pack("{ssss}", "type", "offer", "sdp", sdp);
+		free(sdp);
+		PUSH_STATUS("started", offer_jsep);
+		json_decref(offer_jsep);
+
+	} else {
+		PUSH_ERROR(405, "Not implemented");
+	}
+
+	ok_wait:
+		FREE_MSG_JSEP;
+		return janus_plugin_result_new(JANUS_PLUGIN_OK_WAIT, NULL, NULL);
+
+#	undef PUSH_ERROR
+#	undef FREE_MSG_JSEP
+}
+
+static int _plugin_get_api_compatibility(void)		{ return JANUS_PLUGIN_API_VERSION; }
+static int _plugin_get_version(void)				{ return VERSION_U; }
+static const char *_plugin_get_version_string(void)	{ return VERSION; }
+static const char *_plugin_get_description(void)	{ return "Pi-KVM uStreamer Janus plugin for H.264 video"; }
+static const char *_plugin_get_name(void)			{ return "ustreamer"; }
+static const char *_plugin_get_author(void)			{ return "Maxim Devaev <mdevaev@gmail.com>"; }
+static const char *_plugin_get_package(void)		{ return "janus.plugin.ustreamer"; }
+
+
+#undef STOP
+#undef READY
+#undef UNLOCK
+#undef LOCK
+
+#undef JLOG_PERROR
+#undef JLOG_ERROR
+#undef JLOG_WARN
+#undef JLOG_INFO

janus/src/rtp.c

@@ -0,0 +1,228 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPG-HTTP streamer.                    #
+#                                                                            #
+#    This source file is partially based on this code:                       #
+#      - https://github.com/catid/kvm/blob/master/kvm_pipeline/src           #
+#                                                                            #
+#    Copyright (C) 2018-2021  Maxim Devaev <mdevaev@gmail.com>               #
+#                                                                            #
+#    This program is free software: you can redistribute it and/or modify    #
+#    it under the terms of the GNU General Public License as published by    #
+#    the Free Software Foundation, either version 3 of the License, or       #
+#    (at your option) any later version.                                     #
+#                                                                            #
+#    This program is distributed in the hope that it will be useful,         #
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of          #
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the           #
+#    GNU General Public License for more details.                            #
+#                                                                            #
+#    You should have received a copy of the GNU General Public License       #
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.  #
+#                                                                            #
+*****************************************************************************/
+
+
+#include "rtp.h"
+
+
+#define PAYLOAD 96 // Payload type
+#define PRE 3 // Annex B prefix length
+
+
+void _rtp_process_nalu(rtp_s *rtp, const uint8_t *data, size_t size, uint32_t pts, bool marked, rtp_callback_f callback);
+static void _rtp_write_header(rtp_s *rtp, uint32_t pts, bool marked);
+
+static ssize_t _find_annexb(const uint8_t *data, size_t size);
+
+
+rtp_s *rtp_init(void) {
+	rtp_s *rtp;
+	A_CALLOC(rtp, 1);
+	rtp->ssrc = triple_u32(get_now_monotonic_u64());
+	rtp->sps = frame_init();
+	rtp->pps = frame_init();
+	A_MUTEX_INIT(&rtp->mutex);
+	return rtp;
+}
+
+void rtp_destroy(rtp_s *rtp) {
+	A_MUTEX_DESTROY(&rtp->mutex);
+	frame_destroy(rtp->pps);
+	frame_destroy(rtp->sps);
+	free(rtp);
+}
+
+char *rtp_make_sdp(rtp_s *rtp) {
+	A_MUTEX_LOCK(&rtp->mutex);
+
+	if (rtp->sps->used == 0 || rtp->pps->used == 0) {
+		A_MUTEX_UNLOCK(&rtp->mutex);
+		return NULL;
+	}
+
+	char *sps = NULL;
+	char *pps = NULL;
+	base64_encode(rtp->sps->data, rtp->sps->used, &sps, NULL);
+	base64_encode(rtp->pps->data, rtp->pps->used, &pps, NULL);
+
+	A_MUTEX_UNLOCK(&rtp->mutex);
+
+	// https://tools.ietf.org/html/rfc6184
+	// https://github.com/meetecho/janus-gateway/issues/2443
+	char *sdp;
+	A_ASPRINTF(sdp,
+		"v=0" RN
+		"o=- %" PRIu64 " 1 IN IP4 127.0.0.1" RN
+		"s=Pi-KVM uStreamer" RN
+		"t=0 0" RN
+		"m=video 1 RTP/SAVPF %d" RN
+		"c=IN IP4 0.0.0.0" RN
+		"a=rtpmap:%d H264/90000" RN
+		"a=fmtp:%d profile-level-id=42E01F" RN
+		"a=fmtp:%d packetization-mode=1" RN
+		"a=fmtp:%d sprop-sps=%s" RN
+		"a=fmtp:%d sprop-pps=%s" RN
+		"a=rtcp-fb:%d nack" RN
+		"a=rtcp-fb:%d nack pli" RN
+		"a=rtcp-fb:%d goog-remb" RN
+		"a=sendonly" RN,
+		get_now_id() >> 1, PAYLOAD, PAYLOAD, PAYLOAD, PAYLOAD,
+		PAYLOAD, sps,
+		PAYLOAD, pps,
+		PAYLOAD, PAYLOAD, PAYLOAD
+	);
+
+	free(sps);
+	free(pps);
+	return sdp;
+}
+
+void rtp_wrap_h264(rtp_s *rtp, const frame_s *frame, rtp_callback_f callback) {
+	// There is a complicated logic here but everything works as it should:
+	//   - https://github.com/pikvm/ustreamer/issues/115#issuecomment-893071775
+
+	assert(frame->format == V4L2_PIX_FMT_H264);
+
+	const uint32_t pts = get_now_monotonic_u64() * 9 / 100; // PTS units are in 90 kHz
+	ssize_t last_offset = -PRE;
+
+	while (true) { // Find and iterate by nalus
+		const size_t next_start = last_offset + PRE;
+		ssize_t offset = _find_annexb(frame->data + next_start, frame->used - next_start);
+		if (offset < 0) {
+			break;
+		}
+		offset += next_start;
+
+		if (last_offset >= 0) {
+			const uint8_t *data = frame->data + last_offset + PRE;
+			size_t size = offset - last_offset - PRE;
+			if (data[size - 1] == 0) { // Check for extra 00
+				--size;
+			}
+			_rtp_process_nalu(rtp, data, size, pts, false, callback);
+		}
+
+		last_offset = offset;
+	}
+
+	if (last_offset >= 0) {
+		const uint8_t *data = frame->data + last_offset + PRE;
+		size_t size = frame->used - last_offset - PRE;
+		_rtp_process_nalu(rtp, data, size, pts, true, callback);
+	}
+}
+
+void _rtp_process_nalu(rtp_s *rtp, const uint8_t *data, size_t size, uint32_t pts, bool marked, rtp_callback_f callback) {
+	const unsigned ref_idc = (data[0] >> 5) & 3;
+	const unsigned type = data[0] & 0x1F;
+
+	frame_s *ps = NULL;
+	switch (type) {
+		case 7: ps = rtp->sps; break;
+		case 8: ps = rtp->pps; break;
+	}
+	if (ps) {
+		A_MUTEX_LOCK(&rtp->mutex);
+		frame_set_data(ps, data, size);
+		A_MUTEX_UNLOCK(&rtp->mutex);
+	}
+
+#	define HEADER_SIZE 12
+
+	if (size + HEADER_SIZE <= RTP_DATAGRAM_SIZE) {
+		_rtp_write_header(rtp, pts, marked);
+		memcpy(rtp->datagram + HEADER_SIZE, data, size);
+		callback(rtp->datagram, size + HEADER_SIZE);
+		return;
+	}
+
+	const size_t fu_overhead = HEADER_SIZE + 2; // FU-A overhead
+
+	const uint8_t *src = data + 1;
+	ssize_t remaining = size - 1;
+
+	bool first = true;
+	while (remaining > 0) {
+		ssize_t frag_size = RTP_DATAGRAM_SIZE - fu_overhead;
+		const bool last = (remaining <= frag_size);
+		if (last) {
+			frag_size = remaining;
+		}
+
+		_rtp_write_header(rtp, pts, (marked && last));
+
+		rtp->datagram[HEADER_SIZE] = 28 | (ref_idc << 5);
+
+		uint8_t fu = type;
+		if (first) {
+			fu |= 0x80;
+		}
+		if (last) {
+			fu |= 0x40;
+		}
+		rtp->datagram[HEADER_SIZE + 1] = fu;
+
+		memcpy(rtp->datagram + fu_overhead,	src, frag_size);
+
+		callback(rtp->datagram, fu_overhead + frag_size);
+
+		src += frag_size;
+		remaining -= frag_size;
+		first = false;
+	}
+
+#	undef HEADER_SIZE
+}
+
+static void _rtp_write_header(rtp_s *rtp, uint32_t pts, bool marked) {
+	uint32_t word0 = 0x80000000;
+	if (marked) {
+		word0 |= 1 << 23;
+	}
+	word0 |= (PAYLOAD & 0x7F) << 16;
+	word0 |= rtp->seq;
+	++rtp->seq;
+
+#	define WRITE_BE_U32(_offset, _value) *((uint32_t *)(rtp->datagram + _offset)) = __builtin_bswap32(_value)
+	WRITE_BE_U32(0, word0);
+	WRITE_BE_U32(4, pts);
+	WRITE_BE_U32(8, rtp->ssrc);
+#	undef WRITE_BE_U32
+}
+
+static ssize_t _find_annexb(const uint8_t *data, size_t size) {
+	// Parses buffer for 00 00 01 start codes
+	if (size >= PRE) {
+		for (size_t index = 0; index <= size - PRE; ++index) {
+			if (data[index] == 0 && data[index + 1] == 0 && data[index + 2] == 1) {
+				return index;
+			}
+		}
+	}
+	return -1;
+}
+
+#undef PRE
+#undef PAYLOAD

janus/src/rtp.h

@@ -0,0 +1,69 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPG-HTTP streamer.                    #
+#                                                                            #
+#    This source file is partially based on this code:                       #
+#      - https://github.com/catid/kvm/blob/master/kvm_pipeline/src           #
+#                                                                            #
+#    Copyright (C) 2018-2021  Maxim Devaev <mdevaev@gmail.com>               #
+#                                                                            #
+#    This program is free software: you can redistribute it and/or modify    #
+#    it under the terms of the GNU General Public License as published by    #
+#    the Free Software Foundation, either version 3 of the License, or       #
+#    (at your option) any later version.                                     #
+#                                                                            #
+#    This program is distributed in the hope that it will be useful,         #
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of          #
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the           #
+#    GNU General Public License for more details.                            #
+#                                                                            #
+#    You should have received a copy of the GNU General Public License       #
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.  #
+#                                                                            #
+*****************************************************************************/
+
+
+#pragma once
+
+#include <stdlib.h>
+#include <stdbool.h>
+#include <stdint.h>
+#include <inttypes.h>
+#include <string.h>
+#include <assert.h>
+
+#include <sys/types.h>
+#include <linux/videodev2.h>
+
+#include <pthread.h>
+
+#include "tools.h"
+#include "threading.h"
+#include "frame.h"
+#include "base64.h"
+
+
+// https://stackoverflow.com/questions/47635545/why-webrtc-chose-rtp-max-packet-size-to-1200-bytes
+#define RTP_DATAGRAM_SIZE 1200
+
+
+typedef struct {
+	uint32_t ssrc;
+	uint16_t seq;
+
+	uint8_t datagram[RTP_DATAGRAM_SIZE];
+
+	frame_s *sps; // Actually not a frame, just a bytes storage
+	frame_s *pps;
+
+	pthread_mutex_t mutex;
+} rtp_s;
+
+typedef void (*rtp_callback_f)(const uint8_t *datagram, size_t size);
+
+
+rtp_s *rtp_init(void);
+void rtp_destroy(rtp_s *rtp);
+
+char *rtp_make_sdp(rtp_s *rtp);
+void rtp_wrap_h264(rtp_s *rtp, const frame_s *frame, rtp_callback_f callback);

janus/src/threading.h

@@ -0,0 +1 @@
+../../src/libs/threading.h

janus/src/tools.h

@@ -0,0 +1 @@
+../../src/libs/tools.h

linters/cppcheck.h

@@ -0,0 +1,4 @@
+#define CHAR_BIT 8
+#define WITH_OMX
+#define WITH_GPIO
+#define JANUS_PLUGIN_INIT(...) { __VA_ARGS__ }

linters/tox.ini

@@ -1,12 +1,12 @@
 [tox]
-envlist = cppcheck-src, cppcheck-python, flake8, pylint, mypy, vulture, htmlhint
+envlist = cppcheck, flake8, pylint, mypy, vulture, htmlhint
 skipsdist = true
 
 [testenv]
 basepython = python3.9
 changedir = /src
 
-[testenv:cppcheck-src]
+[testenv:cppcheck]
 whitelist_externals = cppcheck
 commands = cppcheck \
 			--force \
@@ -18,25 +18,8 @@ commands = cppcheck \
 			--suppress=variableScope \
 			--inline-suppr \
 			--library=python \
-			-DCHAR_BIT=8 \
-			-DWITH_OMX \
-			-DWITH_GPIO \
-		src python
-
-[testenv:cppcheck-python]
-whitelist_externals = cppcheck
-commands = cppcheck \
-			--force \
-			--std=c11 \
-			--error-exitcode=1 \
-			--quiet \
-			--enable=warning,unusedFunction,portability,performance,style \
-			--suppress=assignmentInAssert \
-			--suppress=variableScope \
-			--inline-suppr \
-			--library=python \
-			-DCHAR_BIT=8 \
-		python
+			--include=linters/cppcheck.h \
+		src python/ustreamer.c janus/rtp.h janus/rtp.c janus/plugin.c
 
 [testenv:flake8]
 whitelist_externals = bash

man/ustreamer-dump.1

@@ -1,6 +1,6 @@
 .\" Manpage for ustreamer-dump.
 .\" Open an issue or pull request to https://github.com/pikvm/ustreamer to correct errors or typos
-.TH USTREAMER-DUMP 1 "version 3.21" "January 2021"
+.TH USTREAMER-DUMP 1 "version 4.5" "January 2021"
 
 .SH NAME
 ustreamer-dump \- Dump uStreamer's memory sink to file

man/ustreamer.1

@@ -1,6 +1,6 @@
 .\" Manpage for ustreamer.
 .\" Open an issue or pull request to https://github.com/pikvm/ustreamer to correct errors or typos
-.TH USTREAMER 1 "version 3.21" "November 2020"
+.TH USTREAMER 1 "version 4.5" "November 2020"
 
 .SH NAME
 ustreamer \- stream MJPG video from any V4L2 device to the network

pkg/arch/PKGBUILD

@@ -3,7 +3,7 @@
 
 
 pkgname=ustreamer
-pkgver=3.21
+pkgver=4.5
 pkgrel=1
 pkgdesc="Lightweight and fast MJPG-HTTP streamer"
 url="https://github.com/pikvm/ustreamer"
@@ -26,6 +26,11 @@ if [ -e /opt/vc/include/IL/OMX_Core.h ]; then
 	makedepends+=(raspberrypi-firmware)
 	_options="$_options WITH_OMX=1"
 fi
+if [ -e /usr/include/janus/plugins/plugin.h ];then
+	depends+=(janus-gateway-pikvm)
+	makedepends+=(janus-gateway-pikvm)
+	_options="$_options WITH_JANUS=1"
+fi
 
 
 # LD does not link mmal with this option

pkg/openwrt/Makefile

@@ -6,7 +6,7 @@
 include $(TOPDIR)/rules.mk
 
 PKG_NAME:=ustreamer
-PKG_VERSION:=3.21
+PKG_VERSION:=4.5
 PKG_RELEASE:=1
 PKG_MAINTAINER:=Maxim Devaev <mdevaev@gmail.com>
 

python/Makefile

@@ -0,0 +1,20 @@
+-include ../config.mk
+
+DESTDIR ?=
+PREFIX ?= /usr/local
+
+PY ?= python3
+
+
+# =====
+all:
+	$(info == PY_BUILD ustreamer-*.so)
+	@ $(PY) setup.py build
+
+
+install:
+	$(PY) setup.py install --prefix=$(PREFIX) --root=$(if $(DESTDIR),$(DESTDIR),/)
+
+
+clean:
+	rm -rf build

python/setup.py

@@ -1,3 +1,5 @@
+import os
+
 from distutils.core import Extension
 from distutils.core import setup
 
@@ -6,7 +8,7 @@ from distutils.core import setup
 if __name__ == "__main__":
     setup(
         name="ustreamer",
-        version="3.21",
+        version="4.5",
         description="uStreamer tools",
         author="Maxim Devaev",
         author_email="mdevaev@gmail.com",
@@ -16,11 +18,8 @@ if __name__ == "__main__":
                 "ustreamer",
                 libraries=["rt", "m", "pthread"],
                 undef_macros=["NDEBUG"],
-                sources=["ustreamer.c"],
-                depends=[
-                    "../src/libs/tools.h",
-                    "../src/libs/memsinksh.h",
-                ],
+                sources=["src/" + name for name in os.listdir("src") if name.endswith(".c")],
+                depends=["src/" + name for name in os.listdir("src") if name.endswith(".h")],
             ),
         ],
     )

python/src/frame.c

@@ -0,0 +1 @@
+../../src/libs/frame.c

python/src/frame.h

@@ -0,0 +1 @@
+../../src/libs/frame.h

python/src/memsinksh.h

@@ -0,0 +1 @@
+../../src/libs/memsinksh.h

python/src/tools.h

@@ -0,0 +1 @@
+../../src/libs/tools.h

python/src/ustreamer.c

@@ -13,31 +13,11 @@
 
 #include <Python.h>
 
-#include "../src/libs/tools.h" // Just a header without C-sources
-#include "../src/libs/memsinksh.h" // No sources again
+#include "tools.h"
+#include "frame.h"
+#include "memsinksh.h"
 
 
-typedef struct {
-	uint64_t	id;
-	long double	ts;
-
-	uint8_t	*data;
-	size_t	used;
-	size_t	allocated;
-
-	unsigned width;
-	unsigned height;
-	unsigned format;
-	unsigned stride;
-
-	bool online;
-	bool key;
-
-	long double	grab_ts;
-	long double	encode_begin_ts;
-	long double	encode_end_ts;
-} tmp_frame_s;
-
 typedef struct {
 	PyObject_HEAD
 
@@ -49,29 +29,28 @@ typedef struct {
 	int					fd;
 	memsink_shared_s	*mem;
 
-	tmp_frame_s	*tmp_frame;
+	uint64_t	frame_id;
+	long double	frame_ts;
+	frame_s		*frame;
 } MemsinkObject;
 
 
-#define MEM(_next) self->mem->_next
-#define TMP(_next) self->tmp_frame->_next
+#define MEM(_next)		self->mem->_next
+#define FRAME(_next)	self->frame->_next
 
 
 static void MemsinkObject_destroy_internals(MemsinkObject *self) {
 	if (self->mem != NULL) {
-		munmap(self->mem, sizeof(memsink_shared_s));
+		memsink_shared_unmap(self->mem);
 		self->mem = NULL;
 	}
 	if (self->fd > 0) {
 		close(self->fd);
 		self->fd = -1;
 	}
-	if (self->tmp_frame) {
-		if (TMP(data)) {
-			free(TMP(data));
-		}
-		free(self->tmp_frame);
-		self->tmp_frame = NULL;
+	if (self->frame) {
+		frame_destroy(self->frame);
+		self->frame = NULL;
 	}
 }
 
@@ -99,29 +78,15 @@ static int MemsinkObject_init(MemsinkObject *self, PyObject *args, PyObject *kwa
 
 #	undef SET_DOUBLE
 
-	A_CALLOC(self->tmp_frame, 1);
-	TMP(allocated) = 512 * 1024;
-	A_REALLOC(TMP(data), TMP(allocated));
+	self->frame = frame_init();
 
 	if ((self->fd = shm_open(self->obj, O_RDWR, 0)) == -1) {
 		PyErr_SetFromErrno(PyExc_OSError);
 		goto error;
 	}
 
-	if ((self->mem = mmap(
-		NULL,
-		sizeof(memsink_shared_s),
-		PROT_READ | PROT_WRITE,
-		MAP_SHARED,
-		self->fd,
-		0
-	)) == MAP_FAILED) {
+	if ((self->mem = memsink_shared_map(self->fd)) == NULL) {
 		PyErr_SetFromErrno(PyExc_OSError);
-		self->mem = NULL;
-		goto error;
-	}
-	if (self->mem == NULL) {
-		PyErr_SetString(PyExc_RuntimeError, "Memory mapping is NULL"); \
 		goto error;
 	}
 
@@ -177,24 +142,16 @@ static int wait_frame(MemsinkObject *self) {
 			RETURN_OS_ERROR;
 
 		} else if (retval == 0) {
-			if (MEM(magic) == MEMSINK_MAGIC && MEM(version) == MEMSINK_VERSION && TMP(id) != MEM(id)) {
+			if (MEM(magic) == MEMSINK_MAGIC && MEM(version) == MEMSINK_VERSION && MEM(id) != self->frame_id) {
 				if (self->drop_same_frames > 0) {
-#					define CMP(_field) (TMP(_field) == MEM(_field))
 					if (
-						CMP(used)
-						&& CMP(width)
-						&& CMP(height)
-						&& CMP(format)
-						&& CMP(stride)
-						&& CMP(online)
-						&& CMP(key)
-						&& (TMP(ts) + self->drop_same_frames > now)
-						&& !memcmp(TMP(data), MEM(data), MEM(used))
+						FRAME_COMPARE_META_USED_NOTS(self->mem, self->frame)
+						&& (self->frame_ts + self->drop_same_frames > now)
+						&& !memcmp(FRAME(data), MEM(data), MEM(used))
 					) {
-						TMP(id) = MEM(id);
+						self->frame_id = MEM(id);
 						goto drop;
 					}
-#					undef CMP
 				}
 
 				Py_BLOCK_THREADS
@@ -236,30 +193,11 @@ static PyObject *MemsinkObject_wait_frame(MemsinkObject *self, PyObject *Py_UNUS
 		default: return NULL;
 	}
 
-#	define COPY(_field) TMP(_field) = MEM(_field)
-	COPY(width);
-	COPY(height);
-	COPY(format);
-	COPY(stride);
-	COPY(online);
-	COPY(key);
-	COPY(grab_ts);
-	COPY(encode_begin_ts);
-	COPY(encode_end_ts);
-	COPY(used);
-#	undef COPY
-
-	if (TMP(allocated) < MEM(used)) {
-		size_t size = MEM(used) + (512 * 1024);
-		A_REALLOC(TMP(data), size);
-		TMP(allocated) = size;
-	}
-	memcpy(TMP(data), MEM(data), MEM(used));
-	TMP(used) = MEM(used);
-
-	TMP(id) = MEM(id);
-	TMP(ts) = get_now_monotonic();
-	MEM(last_client_ts) = TMP(ts);
+	frame_set_data(self->frame, MEM(data), MEM(used));
+	FRAME_COPY_META(self->mem, self->frame);
+	self->frame_id = MEM(id);
+	self->frame_ts = get_now_monotonic();
+	MEM(last_client_ts) = self->frame_ts;
 
 	if (flock(self->fd, LOCK_UN) < 0) {
 		return PyErr_SetFromErrno(PyExc_OSError);
@@ -281,7 +219,7 @@ static PyObject *MemsinkObject_wait_frame(MemsinkObject *self, PyObject *Py_UNUS
 			} \
 			Py_DECREF(_tmp); \
 		}
-#	define SET_NUMBER(_key, _from, _to) SET_VALUE(#_key, Py##_to##_From##_from(TMP(_key)))
+#	define SET_NUMBER(_key, _from, _to) SET_VALUE(#_key, Py##_to##_From##_from(FRAME(_key)))
 
 	SET_NUMBER(width, Long, Long);
 	SET_NUMBER(height, Long, Long);
@@ -292,7 +230,7 @@ static PyObject *MemsinkObject_wait_frame(MemsinkObject *self, PyObject *Py_UNUS
 	SET_NUMBER(grab_ts, Double, Float);
 	SET_NUMBER(encode_begin_ts, Double, Float);
 	SET_NUMBER(encode_end_ts, Double, Float);
-	SET_VALUE("data", PyBytes_FromStringAndSize((const char *)TMP(data), TMP(used)));
+	SET_VALUE("data", PyBytes_FromStringAndSize((const char *)FRAME(data), FRAME(used)));
 
 #	undef SET_NUMBER
 #	undef SET_VALUE
@@ -376,5 +314,5 @@ PyMODINIT_FUNC PyInit_ustreamer(void) { // cppcheck-suppress unusedFunction
 	return module;
 }
 
-#undef TMP
+#undef FRAME
 #undef MEM

src/Makefile

@@ -0,0 +1,113 @@
+DESTDIR ?=
+PREFIX ?= /usr/local
+
+CC ?= gcc
+CFLAGS ?= -O3
+LDFLAGS ?=
+
+RPI_VC_HEADERS ?= /opt/vc/include
+RPI_VC_LIBS ?= /opt/vc/lib
+
+
+# =====
+_USTR = ustreamer.bin
+_DUMP = ustreamer-dump.bin
+
+_CFLAGS = -MD -c -std=c11 -Wall -Wextra -D_GNU_SOURCE $(CFLAGS)
+_LDFLAGS = $(LDFLAGS)
+
+_COMMON_LIBS = -lm -ljpeg -pthread -lrt
+
+_USTR_LIBS = $(_COMMON_LIBS) -levent -levent_pthreads
+_USTR_SRCS = $(shell ls \
+	libs/*.c \
+	ustreamer/*.c \
+	ustreamer/http/*.c \
+	ustreamer/data/*.c \
+	ustreamer/encoders/cpu/*.c \
+	ustreamer/encoders/hw/*.c \
+)
+
+_DUMP_LIBS = $(_COMMON_LIBS)
+_DUMP_SRCS = $(shell ls \
+	libs/*.c \
+	dump/*.c \
+)
+
+_BUILD = build
+
+
+define optbool
+$(filter $(shell echo $(1) | tr A-Z a-z), yes on 1)
+endef
+
+
+ifneq ($(call optbool,$(WITH_OMX)),)
+_USTR_LIBS += -lbcm_host -lvcos -lvcsm -lopenmaxil -lmmal -lmmal_core -lmmal_util -lmmal_vc_client -lmmal_components -L$(RPI_VC_LIBS)
+override _CFLAGS += -DWITH_OMX -DOMX_SKIP64BIT -I$(RPI_VC_HEADERS)
+_USTR_SRCS += $(shell ls \
+	ustreamer/encoders/omx/*.c \
+	ustreamer/h264/*.c \
+)
+endif
+
+
+ifneq ($(call optbool,$(WITH_GPIO)),)
+_USTR_LIBS += -lgpiod
+override _CFLAGS += -DWITH_GPIO
+_USTR_SRCS += $(shell ls ustreamer/gpio/*.c)
+endif
+
+
+WITH_PTHREAD_NP ?= 1
+ifneq ($(call optbool,$(WITH_PTHREAD_NP)),)
+override _CFLAGS += -DWITH_PTHREAD_NP
+endif
+
+
+WITH_SETPROCTITLE ?= 1
+ifneq ($(call optbool,$(WITH_SETPROCTITLE)),)
+ifeq ($(shell uname -s | tr A-Z a-z),linux)
+_USTR_LIBS += -lbsd
+endif
+override _CFLAGS += -DWITH_SETPROCTITLE
+endif
+
+
+# =====
+all: $(_USTR) $(_DUMP)
+
+
+install: all
+	mkdir -p $(DESTDIR)$(PREFIX)/bin
+	install -m755 $(_USTR) $(DESTDIR)$(PREFIX)/bin/$(subst .bin,,$(_USTR))
+	install -m755 $(_DUMP) $(DESTDIR)$(PREFIX)/bin/$(subst .bin,,$(_DUMP))
+
+
+install-strip: install
+	strip $(DESTDIR)$(PREFIX)/bin/$(subst .bin,,$(_USTR))
+	strip $(DESTDIR)$(PREFIX)/bin/$(subst .bin,,$(_DUMP))
+
+
+$(_USTR): $(_USTR_SRCS:%.c=$(_BUILD)/%.o)
+	$(info == LD $@)
+	@ $(CC) $^ -o $@ $(_LDFLAGS) $(_USTR_LIBS)
+
+
+$(_DUMP): $(_DUMP_SRCS:%.c=$(_BUILD)/%.o)
+	$(info == LD $@)
+	@ $(CC) $^ -o $@ $(_LDFLAGS) $(_DUMP_LIBS)
+
+
+$(_BUILD)/%.o: %.c
+	$(info -- CC $<)
+	@ mkdir -p $(dir $@) || true
+	@ $(CC) $< -o $@ $(_CFLAGS)
+
+
+clean:
+	rm -rf $(_USTR) $(_DUMP) $(_BUILD)
+
+
+_OBJS = $(_USTR_SRCS:%.c=$(_BUILD)/%.o) $(_DUMP_SRCS:%.c=$(_BUILD)/%.o)
+-include $(_OBJS:%.o=%.d)

src/dump/main.c

@@ -128,12 +128,12 @@ int main(int argc, char *argv[]) {
 			case _O_OUTPUT:			OPT_SET(output_path, optarg);
 			case _O_OUTPUT_JSON:	OPT_SET(output_json, true);
 
-			case _O_LOG_LEVEL:			OPT_NUMBER("--log-level", log_level, LOG_LEVEL_INFO, LOG_LEVEL_DEBUG, 0);
-			case _O_PERF:				OPT_SET(log_level, LOG_LEVEL_PERF);
-			case _O_VERBOSE:			OPT_SET(log_level, LOG_LEVEL_VERBOSE);
-			case _O_DEBUG:				OPT_SET(log_level, LOG_LEVEL_DEBUG);
-			case _O_FORCE_LOG_COLORS:	OPT_SET(log_colored, true);
-			case _O_NO_LOG_COLORS:		OPT_SET(log_colored, false);
+			case _O_LOG_LEVEL:			OPT_NUMBER("--log-level", us_log_level, LOG_LEVEL_INFO, LOG_LEVEL_DEBUG, 0);
+			case _O_PERF:				OPT_SET(us_log_level, LOG_LEVEL_PERF);
+			case _O_VERBOSE:			OPT_SET(us_log_level, LOG_LEVEL_VERBOSE);
+			case _O_DEBUG:				OPT_SET(us_log_level, LOG_LEVEL_DEBUG);
+			case _O_FORCE_LOG_COLORS:	OPT_SET(us_log_colored, true);
+			case _O_NO_LOG_COLORS:		OPT_SET(us_log_colored, false);
 
 			case _O_HELP:		_help(stdout); return 0;
 			case _O_VERSION:	puts(VERSION); return 0;
@@ -202,7 +202,7 @@ static void _install_signal_handlers(void) {
 }
 
 static int _dump_sink(const char *sink_name, unsigned sink_timeout, _output_context_s *ctx) {
-	frame_s *frame = frame_init("input");
+	frame_s *frame = frame_init();
 	memsink_s *sink = NULL;
 
 	if ((sink = memsink_init("input", sink_name, false, 0, false, 0, sink_timeout)) == NULL) {
@@ -287,7 +287,7 @@ static void _help(FILE *fp) {
 	SAY("    --log-level <N>  ──── Verbosity level of messages from 0 (info) to 3 (debug).");
 	SAY("                          Enabling debugging messages can slow down the program.");
 	SAY("                          Available levels: 0 (info), 1 (performance), 2 (verbose), 3 (debug).");
-	SAY("                          Default: %d.\n", log_level);
+	SAY("                          Default: %d.\n", us_log_level);
 	SAY("    --perf  ───────────── Enable performance messages (same as --log-level=1). Default: disabled.\n");
 	SAY("    --verbose  ────────── Enable verbose messages and lower (same as --log-level=2). Default: disabled.\n");
 	SAY("    --debug  ──────────── Enable debug messages and lower (same as --log-level=3). Default: disabled.\n");

src/libs/config.h

@@ -22,6 +22,11 @@
 
 #pragma once
 
-#ifndef VERSION
-#	define VERSION "3.21"
-#endif
+#define VERSION_MAJOR 4
+#define VERSION_MINOR 5
+
+#define MAKE_VERSION2(_major, _minor) #_major "." #_minor
+#define MAKE_VERSION1(_major, _minor) MAKE_VERSION2(_major, _minor)
+#define VERSION MAKE_VERSION1(VERSION_MAJOR, VERSION_MINOR)
+
+#define VERSION_U ((unsigned)(VERSION_MAJOR * 1000 + VERSION_MINOR))

src/libs/frame.c

@@ -23,17 +23,14 @@
 #include "frame.h"
 
 
-frame_s *frame_init(const char *name) {
+frame_s *frame_init(void) {
 	frame_s *frame;
 	A_CALLOC(frame, 1);
-	frame->name = name;
-	frame->managed = true;
 	frame_realloc_data(frame, 512 * 1024);
 	return frame;
 }
 
 void frame_destroy(frame_s *frame) {
-	assert(frame->managed);
 	if (frame->data) {
 		free(frame->data);
 	}
@@ -41,68 +38,36 @@ void frame_destroy(frame_s *frame) {
 }
 
 void frame_realloc_data(frame_s *frame, size_t size) {
-	assert(frame->managed);
 	if (frame->allocated < size) {
-		LOG_DEBUG("Increasing frame buffer '%s': %zu -> %zu (+%zu)",
-			frame->name, frame->allocated, size, size - frame->allocated);
 		A_REALLOC(frame->data, size);
 		frame->allocated = size;
 	}
 }
 
 void frame_set_data(frame_s *frame, const uint8_t *data, size_t size) {
-	assert(frame->managed);
 	frame_realloc_data(frame, size);
 	memcpy(frame->data, data, size);
 	frame->used = size;
 }
 
 void frame_append_data(frame_s *frame, const uint8_t *data, size_t size) {
-	assert(frame->managed);
 	size_t new_used = frame->used + size;
 	frame_realloc_data(frame, new_used);
 	memcpy(frame->data + frame->used, data, size);
 	frame->used = new_used;
 }
 
-#define COPY(_field) dest->_field = src->_field
-
 void frame_copy(const frame_s *src, frame_s *dest) {
-	assert(dest->managed);
 	frame_set_data(dest, src->data, src->used);
-	COPY(used);
-	frame_copy_meta(src, dest);
+	FRAME_COPY_META(src, dest);
 }
 
-void frame_copy_meta(const frame_s *src, frame_s *dest) {
-	// Don't copy the name
-	COPY(width);
-	COPY(height);
-	COPY(format);
-	COPY(stride);
-	COPY(online);
-	COPY(key);
-	COPY(grab_ts);
-	COPY(encode_begin_ts);
-	COPY(encode_end_ts);
-}
-
-#undef COPY
-
 bool frame_compare(const frame_s *a, const frame_s *b) {
-#	define CMP(_field) (a->_field == b->_field)
 	return (
 		a->allocated && b->allocated
-		&& CMP(used)
-		&& CMP(width)
-		&& CMP(height)
-		&& CMP(format)
-		&& CMP(stride)
-		&& CMP(online)
-		&& CMP(key)
+		&& FRAME_COMPARE_META_USED_NOTS(a, b)
 		&& !memcmp(a->data, b->data, b->used)
 	);
-#	undef CMP
 }
 
 unsigned frame_get_padding(const frame_s *frame) {

src/libs/frame.h

@@ -32,12 +32,9 @@
 #include <linux/videodev2.h>
 
 #include "tools.h"
-#include "logging.h"
 
 
 typedef struct {
-	const char *name;
-
 	uint8_t	*data;
 	size_t	used;
 	size_t	allocated;
@@ -56,12 +53,37 @@ typedef struct {
 	long double	grab_ts;
 	long double	encode_begin_ts;
 	long double	encode_end_ts;
-
-	bool managed;
 } frame_s;
 
 
-frame_s *frame_init(const char *name);
+#define FRAME_COPY_META(_src, _dest) { \
+		_dest->width = _src->width; \
+		_dest->height = _src->height; \
+		_dest->format = _src->format; \
+		_dest->stride = _src->stride; \
+		_dest->online = _src->online; \
+		_dest->key = _src->key; \
+		_dest->grab_ts = _src->grab_ts; \
+		_dest->encode_begin_ts = _src->encode_begin_ts; \
+		_dest->encode_end_ts = _src->encode_end_ts; \
+	}
+
+inline void frame_copy_meta(const frame_s *src, frame_s *dest) {
+	FRAME_COPY_META(src, dest);
+}
+
+#define FRAME_COMPARE_META_USED_NOTS(_a, _b) ( \
+		_a->used == _b->used \
+		&& _a->width == _b->width \
+		&& _a->height == _b->height \
+		&& _a->format == _b->format \
+		&& _a->stride == _b->stride \
+		&& _a->online == _b->online \
+		&& _a->key == _b->key \
+	)
+
+
+frame_s *frame_init(void);
 void frame_destroy(frame_s *frame);
 
 void frame_realloc_data(frame_s *frame, size_t size);
@@ -69,7 +91,6 @@ void frame_set_data(frame_s *frame, const uint8_t *data, size_t size);
 void frame_append_data(frame_s *frame, const uint8_t *data, size_t size);
 
 void frame_copy(const frame_s *src, frame_s *dest);
-void frame_copy_meta(const frame_s *src, frame_s *dest);
 bool frame_compare(const frame_s *a, const frame_s *b);
 
 unsigned frame_get_padding(const frame_s *frame);

src/libs/list.h

@@ -0,0 +1,71 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPG-HTTP streamer.                    #
+#                                                                            #
+#    Copyright (C) 2018-2021  Maxim Devaev <mdevaev@gmail.com>               #
+#                                                                            #
+#    This program is free software: you can redistribute it and/or modify    #
+#    it under the terms of the GNU General Public License as published by    #
+#    the Free Software Foundation, either version 3 of the License, or       #
+#    (at your option) any later version.                                     #
+#                                                                            #
+#    This program is distributed in the hope that it will be useful,         #
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of          #
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the           #
+#    GNU General Public License for more details.                            #
+#                                                                            #
+#    You should have received a copy of the GNU General Public License       #
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.  #
+#                                                                            #
+*****************************************************************************/
+
+
+#pragma once
+
+#include <assert.h>
+
+
+#define LIST_STRUCT(...) \
+	__VA_ARGS__ *prev; \
+	__VA_ARGS__ *next;
+
+#define LIST_ITERATE(_first, _item, ...) { \
+		for (__typeof__(_first) _item = _first; _item;) { \
+			__typeof__(_first) _next = _item->next; \
+			__VA_ARGS__ \
+			_item = _next; \
+		} \
+	}
+
+#define LIST_APPEND(_first, _item) { \
+		if (_first == NULL) { \
+			_first = _item; \
+		} else { \
+			__typeof__(_first) _last = _first; \
+			for (; _last->next; _last = _last->next); \
+			_item->prev = _last; \
+			_last->next = _item; \
+		} \
+	}
+
+#define LIST_APPEND_C(_first, _item, _count) { \
+		LIST_APPEND(_first, _item); \
+		++(_count); \
+	}
+
+#define LIST_REMOVE(_first, _item) { \
+		if (_item->prev == NULL) { \
+			_first = _item->next; \
+		} else { \
+			_item->prev->next = _item->next; \
+		} \
+		if (_item->next != NULL) { \
+			_item->next->prev = _item->prev; \
+		} \
+	}
+
+#define LIST_REMOVE_C(_first, _item, _count) { \
+		LIST_REMOVE(_first, _item); \
+		assert((_count) >= 1); \
+		--(_count); \
+	}

src/libs/logging.c

@@ -23,8 +23,8 @@
 #include "logging.h"
 
 
-enum log_level_t log_level;
+enum log_level_t us_log_level;
 
-bool log_colored;
+bool us_log_colored;
 
-pthread_mutex_t log_mutex;
+pthread_mutex_t us_log_mutex;

src/libs/logging.h

@@ -45,23 +45,23 @@ enum log_level_t {
 };
 
 
-extern enum log_level_t log_level;
+extern enum log_level_t us_log_level;
 
-extern bool log_colored;
+extern bool us_log_colored;
 
-extern pthread_mutex_t log_mutex;
+extern pthread_mutex_t us_log_mutex;
 
 
 #define LOGGING_INIT { \
-		log_level = LOG_LEVEL_INFO; \
-		log_colored = isatty(2); \
-		A_MUTEX_INIT(&log_mutex); \
+		us_log_level = LOG_LEVEL_INFO; \
+		us_log_colored = isatty(2); \
+		A_MUTEX_INIT(&us_log_mutex); \
 	}
 
-#define LOGGING_DESTROY A_MUTEX_DESTROY(&log_mutex)
+#define LOGGING_DESTROY A_MUTEX_DESTROY(&us_log_mutex)
 
-#define LOGGING_LOCK	A_MUTEX_LOCK(&log_mutex)
-#define LOGGING_UNLOCK	A_MUTEX_UNLOCK(&log_mutex)
+#define LOGGING_LOCK	A_MUTEX_LOCK(&us_log_mutex)
+#define LOGGING_UNLOCK	A_MUTEX_UNLOCK(&us_log_mutex)
 
 
 #define COLOR_GRAY		"\x1b[30;1m"
@@ -84,7 +84,7 @@ extern pthread_mutex_t log_mutex;
 	}
 
 #define SEP_DEBUG(_ch) { \
-		if (log_level >= LOG_LEVEL_DEBUG) { \
+		if (us_log_level >= LOG_LEVEL_DEBUG) { \
 			SEP_INFO(_ch); \
 		} \
 	}
@@ -93,7 +93,7 @@ extern pthread_mutex_t log_mutex;
 #define LOG_PRINTF_NOLOCK(_label_color, _label, _msg_color, _msg, ...) { \
 		char _tname_buf[MAX_THREAD_NAME] = {0}; \
 		thread_get_name(_tname_buf); \
-		if (log_colored) { \
+		if (us_log_colored) { \
 			fprintf(stderr, COLOR_GRAY "-- " _label_color _label COLOR_GRAY \
 				" [%.03Lf %9s]" " -- " COLOR_RESET _msg_color _msg COLOR_RESET, \
 				get_now_monotonic(), _tname_buf, ##__VA_ARGS__); \
@@ -130,25 +130,25 @@ extern pthread_mutex_t log_mutex;
 	}
 
 #define LOG_PERF(_msg, ...) { \
-		if (log_level >= LOG_LEVEL_PERF) { \
+		if (us_log_level >= LOG_LEVEL_PERF) { \
 			LOG_PRINTF(COLOR_CYAN, "PERF ", COLOR_CYAN, _msg, ##__VA_ARGS__); \
 		} \
 	}
 
 #define LOG_PERF_FPS(_msg, ...) { \
-		if (log_level >= LOG_LEVEL_PERF) { \
+		if (us_log_level >= LOG_LEVEL_PERF) { \
 			LOG_PRINTF(COLOR_YELLOW, "PERF ", COLOR_YELLOW, _msg, ##__VA_ARGS__); \
 		} \
 	}
 
 #define LOG_VERBOSE(_msg, ...) { \
-		if (log_level >= LOG_LEVEL_VERBOSE) { \
+		if (us_log_level >= LOG_LEVEL_VERBOSE) { \
 			LOG_PRINTF(COLOR_BLUE, "VERB ", COLOR_BLUE, _msg, ##__VA_ARGS__); \
 		} \
 	}
 
 #define LOG_VERBOSE_PERROR(_msg, ...) { \
-		if (log_level >= LOG_LEVEL_VERBOSE) { \
+		if (us_log_level >= LOG_LEVEL_VERBOSE) { \
 			char _perror_buf[1024] = {0}; \
 			char *_perror_ptr = errno_to_string(errno, _perror_buf, 1023); \
 			LOG_PRINTF(COLOR_BLUE, "VERB ", COLOR_BLUE, _msg ": %s", ##__VA_ARGS__, _perror_ptr); \
@@ -156,17 +156,7 @@ extern pthread_mutex_t log_mutex;
 	}
 
 #define LOG_DEBUG(_msg, ...) { \
-		if (log_level >= LOG_LEVEL_DEBUG) { \
+		if (us_log_level >= LOG_LEVEL_DEBUG) { \
 			LOG_PRINTF(COLOR_GRAY, "DEBUG", COLOR_GRAY, _msg, ##__VA_ARGS__); \
 		} \
 	}
-
-
-INLINE char *errno_to_string(int error, char *buf, size_t size) {
-#	if defined(__GLIBC__) && defined(_GNU_SOURCE)
-	return strerror_r(error, buf, size);
-#	else
-	strerror_r(error, buf, size);
-	return buf;
-#	endif
-}

src/libs/memsink.c

@@ -37,6 +37,7 @@ memsink_s *memsink_init(
 	sink->timeout = timeout;
 	sink->fd = -1;
 	sink->mem = MAP_FAILED;
+	atomic_init(&sink->has_clients, false);
 
 	LOG_INFO("Using %s-sink: %s", name, obj);
 
@@ -54,14 +55,7 @@ memsink_s *memsink_init(
 		goto error;
 	}
 
-	if ((sink->mem = mmap(
-		NULL,
-		sizeof(memsink_shared_s),
-		PROT_READ | PROT_WRITE,
-		MAP_SHARED,
-		sink->fd,
-		0
-	)) == MAP_FAILED) {
+	if ((sink->mem = memsink_shared_map(sink->fd)) == NULL) {
 		LOG_PERROR("%s-sink: Can't mmap shared memory", name);
 		goto error;
 	}
@@ -75,7 +69,7 @@ memsink_s *memsink_init(
 
 void memsink_destroy(memsink_s *sink) {
 	if (sink->mem != MAP_FAILED) {
-		if (munmap(sink->mem, sizeof(memsink_shared_s)) < 0) {
+		if (memsink_shared_unmap(sink->mem) < 0) {
 			LOG_PERROR("%s-sink: Can't unmap shared memory", sink->name);
 		}
 	}
@@ -93,30 +87,35 @@ void memsink_destroy(memsink_s *sink) {
 }
 
 bool memsink_server_check(memsink_s *sink, const frame_s *frame) {
-	// Возвращает true, если если клиенты ИЛИ изменились метаданные
+	// Return true (the need to write to memsink) on any of these conditions:
+	//   - EWOULDBLOCK - we have an active client;
+	//   - Incorrect magic or version - need to first write;
+	//   - We have some active clients by last_client_ts;
+	//   - Frame meta differs (like size, format, but not timestamp).
 
 	assert(sink->server);
 
 	if (flock(sink->fd, LOCK_EX | LOCK_NB) < 0) {
 		if (errno == EWOULDBLOCK) {
-			sink->has_clients = true;
+			atomic_store(&sink->has_clients, true);
 			return true;
 		}
 		LOG_PERROR("%s-sink: Can't lock memory", sink->name);
 		return false;
 	}
 
-	sink->has_clients = (sink->mem->last_client_ts + sink->client_ttl > get_now_monotonic());
+	if (sink->mem->magic != MEMSINK_MAGIC || sink->mem->version != MEMSINK_VERSION) {
+		return true;
+	}
 
-#	define NEQ(_field) (sink->mem->_field != frame->_field)
-	bool retval = (sink->has_clients || NEQ(width) || NEQ(height) || NEQ(format) || NEQ(stride) || NEQ(online) || NEQ(key));
-#	undef NEQ
+	bool has_clients = (sink->mem->last_client_ts + sink->client_ttl > get_now_monotonic());
+	atomic_store(&sink->has_clients, has_clients);
 
 	if (flock(sink->fd, LOCK_UN) < 0) {
 		LOG_PERROR("%s-sink: Can't unlock memory", sink->name);
 		return false;
 	}
-	return retval;
+	return (has_clients || !FRAME_COMPARE_META_USED_NOTS(sink->mem, frame));;
 }
 
 int memsink_server_put(memsink_s *sink, const frame_s *frame) {
@@ -133,24 +132,16 @@ int memsink_server_put(memsink_s *sink, const frame_s *frame) {
 	if (flock_timedwait_monotonic(sink->fd, 1) == 0) {
 		LOG_VERBOSE("%s-sink: >>>>> Exposing new frame ...", sink->name);
 
-#		define COPY(_field) sink->mem->_field = frame->_field
 		sink->last_id = get_now_id();
 		sink->mem->id = sink->last_id;
-		COPY(used);
-		COPY(width);
-		COPY(height);
-		COPY(format);
-		COPY(stride);
-		COPY(online);
-		COPY(key);
-		COPY(grab_ts);
-		COPY(encode_begin_ts);
-		COPY(encode_end_ts);
-		sink->has_clients = (sink->mem->last_client_ts + sink->client_ttl > get_now_monotonic());
+
 		memcpy(sink->mem->data, frame->data, frame->used);
+		sink->mem->used = frame->used;
+		FRAME_COPY_META(frame, sink->mem);
+
 		sink->mem->magic = MEMSINK_MAGIC;
 		sink->mem->version = MEMSINK_VERSION;
-#		undef COPY
+		atomic_store(&sink->has_clients, (sink->mem->last_client_ts + sink->client_ttl > get_now_monotonic()));
 
 		if (flock(sink->fd, LOCK_UN) < 0) {
 			LOG_PERROR("%s-sink: Can't unlock memory", sink->name);
@@ -189,19 +180,9 @@ int memsink_client_get(memsink_s *sink, frame_s *frame) { // cppcheck-suppress u
 			goto done;
 		}
 		if (sink->mem->id != sink->last_id) { // When updated
-#			define COPY(_field) frame->_field = sink->mem->_field
 			sink->last_id = sink->mem->id;
-			COPY(width);
-			COPY(height);
-			COPY(format);
-			COPY(stride);
-			COPY(online);
-			COPY(key);
-			COPY(grab_ts);
-			COPY(encode_begin_ts);
-			COPY(encode_end_ts);
 			frame_set_data(frame, sink->mem->data, sink->mem->used);
-#			undef COPY
+			FRAME_COPY_META(sink->mem, frame);
 			retval = 0;
 		}
 		sink->mem->last_client_ts = get_now_monotonic();

src/libs/memsink.h

@@ -24,6 +24,7 @@
 
 #include <stdint.h>
 #include <stdbool.h>
+#include <stdatomic.h>
 #include <string.h>
 #include <fcntl.h>
 #include <errno.h>
@@ -51,7 +52,7 @@ typedef struct {
 	int					fd;
 	memsink_shared_s	*mem;
 	uint64_t			last_id;
-	bool				has_clients; // Only for server
+	atomic_bool			has_clients; // Only for server
 } memsink_s;
 
 

src/libs/memsinksh.h

@@ -26,6 +26,7 @@
 #include <stdbool.h>
 
 #include <sys/types.h>
+#include <sys/mman.h>
 
 
 #define MEMSINK_MAGIC	((uint64_t)0xCAFEBABECAFEBABE)
@@ -59,3 +60,25 @@ typedef struct {
 
 	uint8_t		data[MEMSINK_MAX_DATA];
 } memsink_shared_s;
+
+
+INLINE memsink_shared_s *memsink_shared_map(int fd) {
+	memsink_shared_s *mem = mmap(
+		NULL,
+		sizeof(memsink_shared_s),
+		PROT_READ | PROT_WRITE,
+		MAP_SHARED,
+		fd,
+		0
+	);
+	if (mem == MAP_FAILED) {
+		return NULL;
+	}
+	assert(mem != NULL);
+	return mem;
+}
+
+INLINE int memsink_shared_unmap(memsink_shared_s *mem) {
+	assert(mem != NULL);
+	return munmap(mem, sizeof(memsink_shared_s));
+}

src/libs/tools.h

@@ -26,7 +26,9 @@
 #include <stdint.h>
 #include <stdlib.h>
 #include <stdbool.h>
+#include <string.h>
 #include <unistd.h>
+#include <locale.h>
 #include <errno.h>
 #include <math.h>
 #include <time.h>
@@ -35,17 +37,21 @@
 #include <sys/file.h>
 
 
-#define A_CALLOC(_dest, _nmemb)		assert((_dest = calloc(_nmemb, sizeof(*(_dest)))))
-#define A_REALLOC(_dest, _nmemb)	assert((_dest = realloc(_dest, _nmemb * sizeof(*(_dest)))))
-#define MEMSET_ZERO(_obj)			memset(&(_obj), 0, sizeof(_obj))
-
-#define ARRAY_LEN(_array) (sizeof(_array) / sizeof(_array[0]))
+#define RN "\r\n"
 
 #define INLINE inline __attribute__((always_inline))
 #define UNUSED __attribute__((unused))
 
+#define A_CALLOC(_dest, _nmemb)		assert((_dest = calloc(_nmemb, sizeof(*(_dest)))))
+#define A_REALLOC(_dest, _nmemb)	assert((_dest = realloc(_dest, _nmemb * sizeof(*(_dest)))))
+#define MEMSET_ZERO(_obj)			memset(&(_obj), 0, sizeof(_obj))
 
-INLINE char *bool_to_string(bool flag) {
+#define A_ASPRINTF(_dest, _fmt, ...) assert(asprintf(&(_dest), _fmt, ##__VA_ARGS__) >= 0)
+
+#define ARRAY_LEN(_array) (sizeof(_array) / sizeof(_array[0]))
+
+
+INLINE const char *bool_to_string(bool flag) {
 	return (flag ? "true" : "false");
 }
 
@@ -104,10 +110,14 @@ INLINE long double get_now_monotonic(void) {
 	return (long double)sec + ((long double)msec) / 1000;
 }
 
-INLINE uint64_t get_now_id(void) {
+INLINE uint64_t get_now_monotonic_u64(void) {
 	struct timespec ts;
 	assert(!clock_gettime(X_CLOCK_MONOTONIC, &ts));
-	uint64_t now = (uint64_t)(ts.tv_nsec / 1000) + (uint64_t)ts.tv_sec * 1000000;
+	return (uint64_t)(ts.tv_nsec / 1000) + (uint64_t)ts.tv_sec * 1000000;
+}
+
+INLINE uint64_t get_now_id(void) {
+	uint64_t now = get_now_monotonic_u64();
 	return (uint64_t)triple_u32(now) | ((uint64_t)triple_u32(now + 12345) << 32);
 }
 
@@ -141,3 +151,16 @@ INLINE int flock_timedwait_monotonic(int fd, long double timeout) {
 	}
 	return retval;
 }
+
+INLINE char *errno_to_string(int error, char *buf, size_t size) {
+	assert(buf);
+	assert(size > 0);
+	locale_t locale = newlocale(LC_MESSAGES_MASK, "C", NULL);
+	char *str = "!!! newlocale() error !!!";
+	strncpy(buf, (locale ? strerror_l(error, locale) : str), size - 1);
+	buf[size - 1] = '\0';
+	if (locale) {
+		freelocale(locale);
+	}
+	return buf;
+}

src/libs/unjpeg.c

@@ -87,6 +87,6 @@ static void _jpeg_error_handler(j_common_ptr jpeg) {
 	char msg[JMSG_LENGTH_MAX];
 
 	(*jpeg_error->mgr.format_message)(jpeg, msg);
-	LOG_ERROR("Can't decompress %s JPEG: %s", jpeg_error->frame->name, msg);
+	LOG_ERROR("Can't decompress JPEG: %s", msg);
 	longjmp(jpeg_error->jmp, -1);
 }

src/ustreamer/blank.c

@@ -44,7 +44,7 @@ frame_s *blank_frame_init(const char *path) {
 }
 
 static frame_s *_init_internal(void) {
-	frame_s *blank = frame_init("blank_internal");
+	frame_s *blank = frame_init();
 	frame_set_data(blank, BLANK_JPEG_DATA, BLANK_JPEG_DATA_SIZE);
 	blank->width = BLANK_JPEG_WIDTH;
 	blank->height = BLANK_JPEG_HEIGHT;
@@ -55,7 +55,7 @@ static frame_s *_init_internal(void) {
 static frame_s *_init_external(const char *path) {
 	FILE *fp = NULL;
 
-	frame_s *blank = frame_init("blank_external");
+	frame_s *blank = frame_init();
 	blank->format = V4L2_PIX_FMT_JPEG;
 
 	if ((fp = fopen(path, "rb")) == NULL) {
@@ -83,7 +83,7 @@ static frame_s *_init_external(const char *path) {
 	}
 #	undef CHUNK_SIZE
 
-	frame_s *decoded = frame_init("blank_external_decoded");
+	frame_s *decoded = frame_init();
 	if (unjpeg(blank, decoded, false) < 0) {
 		frame_destroy(decoded);
 		goto error;

src/ustreamer/device.c

@@ -852,6 +852,7 @@ static void _device_apply_controls(device_s *dev) {
 	CONTROL_AUTO_CID	(V4L2_CID_AUTO_WHITE_BALANCE,	V4L2_CID_WHITE_BALANCE_TEMPERATURE,	white_balance);
 	CONTROL_AUTO_CID	(V4L2_CID_AUTOGAIN,				V4L2_CID_GAIN,						gain);
 	CONTROL_MANUAL_CID	(								V4L2_CID_COLORFX,					color_effect);
+	CONTROL_MANUAL_CID	(								V4L2_CID_ROTATE,	   				rotate);
 	CONTROL_MANUAL_CID	(								V4L2_CID_VFLIP,						flip_vertical);
 	CONTROL_MANUAL_CID	(								V4L2_CID_HFLIP,						flip_horizontal);
 

src/ustreamer/device.h

@@ -122,6 +122,7 @@ typedef struct {
 	control_s white_balance;
 	control_s gain;
 	control_s color_effect;
+	control_s rotate;
 	control_s flip_vertical;
 	control_s flip_horizontal;
 } controls_s;

src/ustreamer/encoder.c

@@ -36,7 +36,7 @@ static const struct {
 };
 
 
-static void *_worker_job_init(worker_s *wr, void *v_enc);
+static void *_worker_job_init(void *v_enc);
 static void _worker_job_destroy(void *v_job);
 static bool _worker_run_job(worker_s *wr);
 
@@ -212,23 +212,17 @@ void encoder_get_runtime_params(encoder_s *enc, encoder_type_e *type, unsigned *
 	A_MUTEX_UNLOCK(&ER(mutex));
 }
 
-static void *_worker_job_init(worker_s *wr, void *v_enc) {
+static void *_worker_job_init(void *v_enc) {
 	encoder_job_s *job;
 	A_CALLOC(job, 1);
 	job->enc = (encoder_s *)v_enc;
-
-	const size_t dest_role_len = strlen(wr->name) + 16;
-	A_CALLOC(job->dest_role, dest_role_len);
-	snprintf(job->dest_role, dest_role_len, "%s_dest", wr->name);
-	job->dest = frame_init(job->dest_role);
-
+	job->dest = frame_init();
 	return (void *)job;
 }
 
 static void _worker_job_destroy(void *v_job) {
 	encoder_job_s *job = (encoder_job_s *)v_job;
 	frame_destroy(job->dest);
-	free(job->dest_role);
 	free(job);
 }
 

src/ustreamer/gpio/gpio.c

@@ -23,7 +23,7 @@
 #include "gpio.h"
 
 
-gpio_s gpio = {
+gpio_s us_gpio = {
 	.path = "/dev/gpiochip0",
 	.consumer_prefix = "ustreamer",
 
@@ -51,42 +51,42 @@ static void _gpio_output_destroy(gpio_output_s *output);
 
 
 void gpio_init(void) {
-	assert(gpio.chip == NULL);
+	assert(us_gpio.chip == NULL);
 	if (
-		gpio.prog_running.pin >= 0
-		|| gpio.stream_online.pin >= 0
-		|| gpio.has_http_clients.pin >= 0
+		us_gpio.prog_running.pin >= 0
+		|| us_gpio.stream_online.pin >= 0
+		|| us_gpio.has_http_clients.pin >= 0
 	) {
-		A_MUTEX_INIT(&gpio.mutex);
-		LOG_INFO("GPIO: Using chip device: %s", gpio.path);
-		if ((gpio.chip = gpiod_chip_open(gpio.path)) != NULL) {
-			_gpio_output_init(&gpio.prog_running);
-			_gpio_output_init(&gpio.stream_online);
-			_gpio_output_init(&gpio.has_http_clients);
+		A_MUTEX_INIT(&us_gpio.mutex);
+		LOG_INFO("GPIO: Using chip device: %s", us_gpio.path);
+		if ((us_gpio.chip = gpiod_chip_open(us_gpio.path)) != NULL) {
+			_gpio_output_init(&us_gpio.prog_running);
+			_gpio_output_init(&us_gpio.stream_online);
+			_gpio_output_init(&us_gpio.has_http_clients);
 		} else {
-			LOG_PERROR("GPIO: Can't initialize chip device %s", gpio.path);
+			LOG_PERROR("GPIO: Can't initialize chip device %s", us_gpio.path);
 		}
 	}
 }
 
 void gpio_destroy(void) {
-	_gpio_output_destroy(&gpio.prog_running);
-	_gpio_output_destroy(&gpio.stream_online);
-	_gpio_output_destroy(&gpio.has_http_clients);
-	if (gpio.chip) {
-		gpiod_chip_close(gpio.chip);
-		gpio.chip = NULL;
-		A_MUTEX_DESTROY(&gpio.mutex);
+	_gpio_output_destroy(&us_gpio.prog_running);
+	_gpio_output_destroy(&us_gpio.stream_online);
+	_gpio_output_destroy(&us_gpio.has_http_clients);
+	if (us_gpio.chip) {
+		gpiod_chip_close(us_gpio.chip);
+		us_gpio.chip = NULL;
+		A_MUTEX_DESTROY(&us_gpio.mutex);
 	}
 }
 
 int gpio_inner_set(gpio_output_s *output, bool state) {
 	int retval = 0;
 
-	assert(gpio.chip);
+	assert(us_gpio.chip);
 	assert(output->line);
 	assert(output->state != state); // Must be checked in macro for the performance
-	A_MUTEX_LOCK(&gpio.mutex);
+	A_MUTEX_LOCK(&us_gpio.mutex);
 
 	if (gpiod_line_set_value(output->line, (int)state) < 0) { \
 		LOG_PERROR("GPIO: Can't write value %d to line %s (will be disabled)", state, output->consumer); \
@@ -94,19 +94,18 @@ int gpio_inner_set(gpio_output_s *output, bool state) {
 		retval = -1;
 	}
 
-	A_MUTEX_UNLOCK(&gpio.mutex);
+	A_MUTEX_UNLOCK(&us_gpio.mutex);
 	return retval;
 }
 
 static void _gpio_output_init(gpio_output_s *output) {
-	assert(gpio.chip);
+	assert(us_gpio.chip);
 	assert(output->line == NULL);
 
-	A_CALLOC(output->consumer, strlen(gpio.consumer_prefix) + strlen(output->role) + 16);
-	sprintf(output->consumer, "%s::%s", gpio.consumer_prefix, output->role);
+	A_ASPRINTF(output->consumer, "%s::%s", us_gpio.consumer_prefix, output->role);
 
 	if (output->pin >= 0) {
-		if ((output->line = gpiod_chip_get_line(gpio.chip, output->pin)) != NULL) {
+		if ((output->line = gpiod_chip_get_line(us_gpio.chip, output->pin)) != NULL) {
 			if (gpiod_line_request_output(output->line, output->consumer, 0) < 0) {
 				LOG_PERROR("GPIO: Can't request pin=%d as %s", output->pin, output->consumer);
 				_gpio_output_destroy(output);

src/ustreamer/gpio/gpio.h

@@ -56,7 +56,7 @@ typedef struct {
 } gpio_s;
 
 
-extern gpio_s gpio;
+extern gpio_s us_gpio;
 
 
 void gpio_init(void);
@@ -73,15 +73,15 @@ int gpio_inner_set(gpio_output_s *output, bool state);
 	}
 
 INLINE void gpio_set_prog_running(bool state) {
-	SET_STATE(gpio.prog_running, state);
+	SET_STATE(us_gpio.prog_running, state);
 }
 
 INLINE void gpio_set_stream_online(bool state) {
-	SET_STATE(gpio.stream_online, state);
+	SET_STATE(us_gpio.stream_online, state);
 }
 
 INLINE void gpio_set_has_http_clients(bool state) {
-	SET_STATE(gpio.has_http_clients, state);
+	SET_STATE(us_gpio.has_http_clients, state);
 }
 
 #undef SET_STATE

src/ustreamer/h264/stream.c

@@ -27,12 +27,14 @@ h264_stream_s *h264_stream_init(memsink_s *sink, unsigned bitrate, unsigned gop)
 	h264_stream_s *h264;
 	A_CALLOC(h264, 1);
 	h264->sink = sink;
-	h264->tmp_src = frame_init("h264_tmp_src");
-	h264->dest = frame_init("h264_dest");
+	h264->tmp_src = frame_init();
+	h264->dest = frame_init();
 	atomic_init(&h264->online, false);
 
-	// FIXME: 30 or 0? https://github.com/6by9/yavta/blob/master/yavta.c#L210
-	if ((h264->enc = h264_encoder_init(bitrate, gop, 0)) == NULL) {
+	// FIXME: 30 or 0? https://github.com/6by9/yavta/blob/master/yavta.c#L2100
+	// По логике вещей правильно 0, но почему-то на низких разрешениях типа 640x480
+	// енкодер через несколько секунд перестает производить корректные фреймы.
+	if ((h264->enc = h264_encoder_init(bitrate, gop, 30)) == NULL) {
 		goto error;
 	}
 

src/ustreamer/http/server.c

@@ -51,7 +51,7 @@ static char *_http_get_client_hostport(struct evhttp_request *request);
 server_s *server_init(stream_s *stream) {
 	exposed_s *exposed;
 	A_CALLOC(exposed, 1);
-	exposed->frame = frame_init("http_exposed");
+	exposed->frame = frame_init();
 
 	server_runtime_s *run;
 	A_CALLOC(run, 1);
@@ -93,13 +93,11 @@ void server_destroy(server_s *server) {
 	libevent_global_shutdown();
 #	endif
 
-	for (stream_client_s *client = RUN(stream_clients); client != NULL;) {
-		stream_client_s *next = client->next;
+	LIST_ITERATE(RUN(stream_clients), client, {
 		free(client->key);
 		free(client->hostport);
 		free(client);
-		client = next;
-	}
+	});
 
 	if (RUN(auth_token)) {
 		free(RUN(auth_token));
@@ -145,16 +143,14 @@ int server_listen(server_s *server) {
 	evhttp_set_timeout(RUN(http), server->timeout);
 
 	if (server->user[0] != '\0') {
-		char *raw_token;
 		char *encoded_token = NULL;
 
-		A_CALLOC(raw_token, strlen(server->user) + strlen(server->passwd) + 16);
-		sprintf(raw_token, "%s:%s", server->user, server->passwd);
+		char *raw_token;
+		A_ASPRINTF(raw_token, "%s:%s", server->user, server->passwd);
 		base64_encode((uint8_t *)raw_token, strlen(raw_token), &encoded_token, NULL);
 		free(raw_token);
 
-		A_CALLOC(RUN(auth_token), strlen(encoded_token) + 16);
-		sprintf(RUN(auth_token), "Basic %s", encoded_token);
+		A_ASPRINTF(RUN(auth_token), "Basic %s", encoded_token);
 		free(encoded_token);
 
 		LOG_INFO("Using HTTP basic auth");
@@ -346,6 +342,7 @@ static void _http_callback_state(struct evhttp_request *request, void *v_server)
 		encoder_type_to_string(enc_type),
 		enc_quality
 	));
+
 #	ifdef WITH_OMX
 	if (STREAM(run->h264)) {
 		assert(evbuffer_add_printf(buf,
@@ -356,6 +353,32 @@ static void _http_callback_state(struct evhttp_request *request, void *v_server)
 		));
 	}
 #	endif
+
+	if (
+		STREAM(sink)
+#	ifdef WITH_OMX
+		|| STREAM(h264_sink)
+#	endif
+	) {
+		assert(evbuffer_add_printf(buf, " \"sinks\": {"));
+		if (STREAM(sink)) {
+			assert(evbuffer_add_printf(buf,
+				"\"jpeg\": {\"has_clients\": %s}",
+				bool_to_string(atomic_load(&STREAM(sink->has_clients)))
+			));
+		}
+#	ifdef WITH_OMX
+		if (STREAM(h264_sink)) {
+			assert(evbuffer_add_printf(buf,
+				"%s\"h264\": {\"has_clients\": %s}",
+				(STREAM(sink) ? ", " : ""),
+				bool_to_string(atomic_load(&STREAM(h264_sink->has_clients)))
+			));
+		}
+#	endif
+		assert(evbuffer_add_printf(buf, "},"));
+	}
+
 	assert(evbuffer_add_printf(buf,
 		" \"source\": {\"resolution\": {\"width\": %u, \"height\": %u},"
 		" \"online\": %s, \"desired_fps\": %u, \"captured_fps\": %u},"
@@ -369,7 +392,7 @@ static void _http_callback_state(struct evhttp_request *request, void *v_server)
 		RUN(stream_clients_count)
 	));
 
-	for (stream_client_s * client = RUN(stream_clients); client != NULL; client = client->next) {
+	LIST_ITERATE(RUN(stream_clients), client, {
 		assert(evbuffer_add_printf(buf,
 			"\"%" PRIx64 "\": {\"fps\": %u, \"extra_headers\": %s, \"advance_headers\": %s,"
 			" \"dual_final_frames\": %s, \"zero_data\": %s}%s",
@@ -381,7 +404,7 @@ static void _http_callback_state(struct evhttp_request *request, void *v_server)
 			bool_to_string(client->zero_data),
 			(client->next ? ", " : "")
 		));
-	}
+	});
 
 	assert(evbuffer_add_printf(buf, "}}}}"));
 
@@ -479,16 +502,7 @@ static void _http_callback_stream(struct evhttp_request *request, void *v_server
 		client->hostport = _http_get_client_hostport(request);
 		client->id = get_now_id();
 
-		if (RUN(stream_clients) == NULL) {
-			RUN(stream_clients) = client;
-		} else {
-			stream_client_s *last = RUN(stream_clients);
-
-			for (; last->next != NULL; last = last->next);
-			client->prev = last;
-			last->next = client;
-		}
-		RUN(stream_clients_count) += 1;
+		LIST_APPEND_C(RUN(stream_clients), client, RUN(stream_clients_count));
 
 		if (RUN(stream_clients_count) == 1) {
 			atomic_store(&VID(has_clients), true);
@@ -522,7 +536,6 @@ static void _http_callback_stream(struct evhttp_request *request, void *v_server
 
 static void _http_callback_stream_write(struct bufferevent *buf_event, void *v_client) {
 #	define BOUNDARY "boundarydonotcross"
-#	define RN "\r\n"
 
 	stream_client_s *client = (stream_client_s *)v_client;
 	server_s *server = client->server;
@@ -647,7 +660,6 @@ static void _http_callback_stream_write(struct bufferevent *buf_event, void *v_c
 	bufferevent_enable(buf_event, EV_READ);
 
 #	undef ADD_ADVANCE_HEADERS
-#	undef RN
 #	undef BOUNDARY
 }
 
@@ -655,10 +667,7 @@ static void _http_callback_stream_error(UNUSED struct bufferevent *buf_event, UN
 	stream_client_s *client = (stream_client_s *)v_client;
 	server_s *server = client->server;
 
-	char *reason = bufferevent_my_format_reason(what);
-
-	assert(RUN(stream_clients_count) > 0);
-	RUN(stream_clients_count) -= 1;
+	LIST_REMOVE_C(RUN(stream_clients), client, RUN(stream_clients_count));
 
 	if (RUN(stream_clients_count) == 0) {
 		atomic_store(&VID(has_clients), false);
@@ -667,34 +676,26 @@ static void _http_callback_stream_error(UNUSED struct bufferevent *buf_event, UN
 #		endif
 	}
 
+	char *reason = bufferevent_my_format_reason(what);
 	LOG_INFO("HTTP: Disconnected client: %s, id=%" PRIx64 ", %s; clients now: %u",
 		client->hostport, client->id, reason, RUN(stream_clients_count));
+	free(reason);
 
 	struct evhttp_connection *conn = evhttp_request_get_connection(client->request);
 	if (conn) {
 		evhttp_connection_free(conn);
 	}
 
-	if (client->prev == NULL) {
-		RUN(stream_clients) = client->next;
-	} else {
-		client->prev->next = client->next;
-	}
-	if (client->next != NULL) {
-		client->next->prev = client->prev;
-	}
 	free(client->key);
 	free(client->hostport);
 	free(client);
-
-	free(reason);
 }
 
 static void _http_queue_send_stream(server_s *server, bool stream_updated, bool frame_updated) {
 	bool has_clients = false;
 	bool queued = false;
 
-	for (stream_client_s *client = RUN(stream_clients); client != NULL; client = client->next) {
+	LIST_ITERATE(RUN(stream_clients), client, {
 		struct evhttp_connection *conn = evhttp_request_get_connection(client->request);
 		if (conn) {
 			// Фикс для бага WebKit. При включенной опции дропа одинаковых фреймов,
@@ -725,7 +726,7 @@ static void _http_queue_send_stream(server_s *server, bool stream_updated, bool
 
 			has_clients = true;
 		}
-	}
+	});
 
 	if (queued) {
 		static unsigned queued_fps_accum = 0;
@@ -857,10 +858,7 @@ static char *_http_get_client_hostport(struct evhttp_request *request) {
 	}
 
 	char *hostport;
-	size_t hostport_len = strlen(addr) + 64;
-	A_CALLOC(hostport, hostport_len);
-
-	snprintf(hostport, hostport_len, "[%s]:%u", addr, port);
+	A_ASPRINTF(hostport, "[%s]:%u", addr, port);
 	free(addr);
 	return hostport;
 }

src/ustreamer/http/server.h

@@ -58,6 +58,7 @@
 #include "../../libs/process.h"
 #include "../../libs/frame.h"
 #include "../../libs/base64.h"
+#include "../../libs/list.h"
 #include "../data/index_html.h"
 #include "../encoder.h"
 #include "../stream.h"
@@ -91,8 +92,7 @@ typedef struct stream_client_sx {
 	unsigned		fps_accum;
 	long long		fps_accum_second;
 
-	struct stream_client_sx *prev;
-	struct stream_client_sx *next;
+	LIST_STRUCT(struct stream_client_sx);
 } stream_client_s;
 
 typedef struct {

src/ustreamer/http/static.c

@@ -32,7 +32,7 @@ char *find_static_file_path(const char *root_path, const char *request_path) {
 		goto error;
 	}
 
-	A_CALLOC(path, strlen(root_path) + strlen(simplified_path) + 16);
+	A_CALLOC(path, strlen(root_path) + strlen(simplified_path) + 16); // + reserved for /index.html
 	sprintf(path, "%s/%s", root_path, simplified_path);
 
 	struct stat st;

src/ustreamer/options.c

@@ -72,6 +72,7 @@ enum _OPT_VALUES {
 	_O_WHITE_BALANCE,
 	_O_GAIN,
 	_O_COLOR_EFFECT,
+	_O_ROTATE,
 	_O_FLIP_VERTICAL,
 	_O_FLIP_HORIZONTAL,
 
@@ -158,6 +159,7 @@ static const struct option _LONG_OPTS[] = {
 	{"white-balance",			required_argument,	NULL,	_O_WHITE_BALANCE},
 	{"gain",					required_argument,	NULL,	_O_GAIN},
 	{"color-effect",			required_argument,	NULL,	_O_COLOR_EFFECT},
+	{"rotate",				required_argument,	NULL,	_O_ROTATE},
 	{"flip-vertical",			required_argument,	NULL,	_O_FLIP_VERTICAL},
 	{"flip-horizontal",			required_argument,	NULL,	_O_FLIP_HORIZONTAL},
 
@@ -394,6 +396,7 @@ int options_parse(options_s *options, device_s *dev, encoder_s *enc, stream_s *s
 				OPT_CTL_DEFAULT_NOBREAK(white_balance);
 				OPT_CTL_DEFAULT_NOBREAK(gain);
 				OPT_CTL_DEFAULT_NOBREAK(color_effect);
+				OPT_CTL_DEFAULT_NOBREAK(rotate);
 				OPT_CTL_DEFAULT_NOBREAK(flip_vertical);
 				OPT_CTL_DEFAULT_NOBREAK(flip_horizontal);
 				break;
@@ -407,6 +410,7 @@ int options_parse(options_s *options, device_s *dev, encoder_s *enc, stream_s *s
 			case _O_WHITE_BALANCE:			OPT_CTL_AUTO(white_balance);
 			case _O_GAIN:					OPT_CTL_AUTO(gain);
 			case _O_COLOR_EFFECT:			OPT_CTL_MANUAL(color_effect);
+			case _O_ROTATE:			 	OPT_CTL_MANUAL(rotate);
 			case _O_FLIP_VERTICAL:			OPT_CTL_MANUAL(flip_vertical);
 			case _O_FLIP_HORIZONTAL:		OPT_CTL_MANUAL(flip_horizontal);
 
@@ -440,11 +444,11 @@ int options_parse(options_s *options, device_s *dev, encoder_s *enc, stream_s *s
 #			undef ADD_SINK
 
 #			ifdef WITH_GPIO
-			case _O_GPIO_DEVICE:			OPT_SET(gpio.path, optarg);
-			case _O_GPIO_CONSUMER_PREFIX:	OPT_SET(gpio.consumer_prefix, optarg);
-			case _O_GPIO_PROG_RUNNING:		OPT_NUMBER("--gpio-prog-running", gpio.prog_running.pin, 0, 256, 0);
-			case _O_GPIO_STREAM_ONLINE:		OPT_NUMBER("--gpio-stream-online", gpio.stream_online.pin, 0, 256, 0);
-			case _O_GPIO_HAS_HTTP_CLIENTS:	OPT_NUMBER("--gpio-has-http-clients", gpio.has_http_clients.pin, 0, 256, 0);
+			case _O_GPIO_DEVICE:			OPT_SET(us_gpio.path, optarg);
+			case _O_GPIO_CONSUMER_PREFIX:	OPT_SET(us_gpio.consumer_prefix, optarg);
+			case _O_GPIO_PROG_RUNNING:		OPT_NUMBER("--gpio-prog-running", us_gpio.prog_running.pin, 0, 256, 0);
+			case _O_GPIO_STREAM_ONLINE:		OPT_NUMBER("--gpio-stream-online", us_gpio.stream_online.pin, 0, 256, 0);
+			case _O_GPIO_HAS_HTTP_CLIENTS:	OPT_NUMBER("--gpio-has-http-clients", us_gpio.has_http_clients.pin, 0, 256, 0);
 #			endif
 
 #			ifdef HAS_PDEATHSIG
@@ -459,12 +463,12 @@ int options_parse(options_s *options, device_s *dev, encoder_s *enc, stream_s *s
 #			endif
 			case _O_NOTIFY_PARENT:			OPT_SET(server->notify_parent, true);
 
-			case _O_LOG_LEVEL:			OPT_NUMBER("--log-level", log_level, LOG_LEVEL_INFO, LOG_LEVEL_DEBUG, 0);
-			case _O_PERF:				OPT_SET(log_level, LOG_LEVEL_PERF);
-			case _O_VERBOSE:			OPT_SET(log_level, LOG_LEVEL_VERBOSE);
-			case _O_DEBUG:				OPT_SET(log_level, LOG_LEVEL_DEBUG);
-			case _O_FORCE_LOG_COLORS:	OPT_SET(log_colored, true);
-			case _O_NO_LOG_COLORS:		OPT_SET(log_colored, false);
+			case _O_LOG_LEVEL:			OPT_NUMBER("--log-level", us_log_level, LOG_LEVEL_INFO, LOG_LEVEL_DEBUG, 0);
+			case _O_PERF:				OPT_SET(us_log_level, LOG_LEVEL_PERF);
+			case _O_VERBOSE:			OPT_SET(us_log_level, LOG_LEVEL_VERBOSE);
+			case _O_DEBUG:				OPT_SET(us_log_level, LOG_LEVEL_DEBUG);
+			case _O_FORCE_LOG_COLORS:	OPT_SET(us_log_colored, true);
+			case _O_NO_LOG_COLORS:		OPT_SET(us_log_colored, false);
 
 			case _O_HELP:		_help(stdout, dev, enc, stream, server); return 1;
 			case _O_VERSION:	puts(VERSION); return 1;
@@ -637,6 +641,7 @@ static void _help(FILE *fp, device_s *dev, encoder_s *enc, stream_s *stream, ser
 	SAY("    --white-balance <N|auto|default>  ───── Set white balance. Default: no change.\n");
 	SAY("    --gain <N|auto|default>  ────────────── Set gain. Default: no change.\n");
 	SAY("    --color-effect <N|default>  ─────────── Set color effect. Default: no change.\n");
+	SAY("    --rotate <N|default>  ───────────────── Set rotation. Default: no change.\n");
 	SAY("    --flip-vertical <1|0|default>  ──────── Set vertical flip. Default: no change.\n");
 	SAY("    --flip-horizontal <1|0|default>  ────── Set horizontal flip. Default: no change.\n");
 	SAY("    Hint: use v4l2-ctl --list-ctrls-menus to query available controls of the device.\n");
@@ -679,8 +684,8 @@ static void _help(FILE *fp, device_s *dev, encoder_s *enc, stream_s *stream, ser
 #	ifdef WITH_GPIO
 	SAY("GPIO options:");
 	SAY("═════════════");
-	SAY("    --gpio-device </dev/path>  ───── Path to GPIO character device. Default: %s.\n", gpio.path);
-	SAY("    --gpio-consumer-prefix <str>  ── Consumer prefix for GPIO outputs. Default: %s.\n", gpio.consumer_prefix);
+	SAY("    --gpio-device </dev/path>  ───── Path to GPIO character device. Default: %s.\n", us_gpio.path);
+	SAY("    --gpio-consumer-prefix <str>  ── Consumer prefix for GPIO outputs. Default: %s.\n", us_gpio.consumer_prefix);
 	SAY("    --gpio-prog-running <pin>  ───── Set 1 on GPIO pin while uStreamer is running. Default: disabled.\n");
 	SAY("    --gpio-stream-online <pin>  ──── Set 1 while streaming. Default: disabled.\n");
 	SAY("    --gpio-has-http-clients <pin>  ─ Set 1 while stream has at least one client. Default: disabled.\n");
@@ -703,7 +708,7 @@ static void _help(FILE *fp, device_s *dev, encoder_s *enc, stream_s *stream, ser
 	SAY("    --log-level <N>  ──── Verbosity level of messages from 0 (info) to 3 (debug).");
 	SAY("                          Enabling debugging messages can slow down the program.");
 	SAY("                          Available levels: 0 (info), 1 (performance), 2 (verbose), 3 (debug).");
-	SAY("                          Default: %d.\n", log_level);
+	SAY("                          Default: %d.\n", us_log_level);
 	SAY("    --perf  ───────────── Enable performance messages (same as --log-level=1). Default: disabled.\n");
 	SAY("    --verbose  ────────── Enable verbose messages and lower (same as --log-level=2). Default: disabled.\n");
 	SAY("    --debug  ──────────── Enable debug messages and lower (same as --log-level=3). Default: disabled.\n");

src/ustreamer/stream.c

@@ -52,7 +52,7 @@ stream_s *stream_init(device_s *dev, encoder_s *enc) {
 
 	video_s *video;
 	A_CALLOC(video, 1);
-	video->frame = frame_init("stream_video");
+	video->frame = frame_init();
 	atomic_init(&video->updated, false);
 	A_MUTEX_INIT(&video->mutex);
 	atomic_init(&video->has_clients, false);
@@ -136,9 +136,9 @@ void stream_loop(stream_s *stream) {
 					&& !atomic_load(&RUN(stop))
 					&& !atomic_load(&RUN(video->has_clients))
 					// has_clients синков НЕ обновляются в реальном времени
-					&& (stream->sink == NULL || !stream->sink->has_clients)
+					&& (stream->sink == NULL || !atomic_load(&stream->sink->has_clients))
 #					ifdef WITH_OMX
-					&& (RUN(h264) == NULL || /*RUN(h264->sink) == NULL ||*/ !RUN(h264->sink->has_clients))
+					&& (RUN(h264) == NULL || /*RUN(h264->sink) == NULL ||*/ !atomic_load(&RUN(h264->sink->has_clients)))
 #					endif
 				) {
 					usleep(100000);

src/ustreamer/workers.c

@@ -28,9 +28,9 @@ static void *_worker_thread(void *v_worker);
 
 workers_pool_s *workers_pool_init(
 	const char *name, const char *wr_prefix, unsigned n_workers, long double desired_interval,
-	void *(*job_init)(worker_s *wr, void *arg), void *job_init_arg,
-	void (*job_destroy)(void *),
-	bool (*run_job)(worker_s *)) {
+	workers_pool_job_init_f job_init, void *job_init_arg,
+	workers_pool_job_destroy_f job_destroy,
+	workers_pool_run_job_f run_job) {
 
 	LOG_INFO("Creating pool %s with %u workers ...", name, n_workers);
 
@@ -49,21 +49,18 @@ workers_pool_s *workers_pool_init(
 	A_MUTEX_INIT(&pool->free_workers_mutex);
 	A_COND_INIT(&pool->free_workers_cond);
 
-	const size_t wr_name_len = strlen(wr_prefix) + 64;
-
 	for (unsigned number = 0; number < pool->n_workers; ++number) {
 #		define WR(_next) pool->workers[number]._next
 
 		WR(number) = number;
-		A_CALLOC(WR(name), wr_name_len);
-		snprintf(WR(name), wr_name_len, "%s-%u", wr_prefix, number);
+		A_ASPRINTF(WR(name), "%s-%u", wr_prefix, number);
 
 		A_MUTEX_INIT(&WR(has_job_mutex));
 		atomic_init(&WR(has_job), false);
 		A_COND_INIT(&WR(has_job_cond));
 
 		WR(pool) = pool;
-		WR(job) = job_init(&pool->workers[number], job_init_arg);
+		WR(job) = job_init(job_init_arg);
 
 		A_THREAD_CREATE(&WR(tid), _worker_thread, (void *)&(pool->workers[number]));
 		pool->free_workers += 1;

src/ustreamer/workers.h

@@ -55,12 +55,16 @@ typedef struct worker_sx {
 	struct workers_pool_sx *pool;
 } worker_s;
 
+typedef void *(*workers_pool_job_init_f)(void *arg);
+typedef void (*workers_pool_job_destroy_f)(void *job);
+typedef bool (*workers_pool_run_job_f)(worker_s *wr);
+
 typedef struct workers_pool_sx {
 	const char		*name;
 	long double		desired_interval;
 
-	bool (*run_job)(worker_s *wr);
-	void (*job_destroy)(void *job);
+	workers_pool_job_destroy_f	job_destroy;
+	workers_pool_run_job_f		run_job;
 
 	unsigned		n_workers;
 	worker_s		*workers;
@@ -79,9 +83,9 @@ typedef struct workers_pool_sx {
 
 workers_pool_s *workers_pool_init(
 	const char *name, const char *wr_prefix, unsigned n_workers, long double desired_interval,
-	void *(*job_init)(worker_s *wr, void *arg), void *job_init_arg,
-	void (*job_destroy)(void *job),
-	bool (*run_job)(worker_s *));
+	workers_pool_job_init_f job_init, void *job_init_arg,
+	workers_pool_job_destroy_f job_destroy,
+	workers_pool_run_job_f run_job);
 
 void workers_pool_destroy(workers_pool_s *pool);