Bump version: 4.7 → 4.8

Document how to integrate uStreamer with nginx

.bumpversion.cfg

@@ -1,7 +1,7 @@
 [bumpversion]
 commit = True
 tag = True
-current_version = 3.25
+current_version = 4.8
 parse = (?P<major>\d+)\.(?P<minor>\d+)
 serialize = 
 	{major}.{minor}

.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,15 +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
 /src/build/
-/src/*.bin
 /python/build/
+/janus/build/
 /ustreamer
 /ustreamer-dump
 /config.mk
-/vgcore.*
-/*.sock
-/*.so
+vgcore.*
+*.sock
+*.so
+*.bin
+*.pkg.tar.xz
+*.pkg.tar.zst
+*.tar.gz

Makefile

@@ -29,6 +29,9 @@ all:
 ifneq ($(call optbool,$(WITH_PYTHON)),)
 	+ $(MAKE) python
 endif
+ifneq ($(call optbool,$(WITH_JANUS)),)
+	+ $(MAKE) janus
+endif
 
 
 apps:
@@ -42,10 +45,18 @@ python:
 	@ ln -sf python/build/lib.*/*.so .
 
 
+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 \
@@ -110,6 +121,7 @@ clean:
 	rm -f ustreamer ustreamer-dump *.so
 	$(MAKE) -C src clean
 	$(MAKE) -C python clean
+	$(MAKE) -C janus clean
 
 
-.PHONY: python linters
+.PHONY: python janus linters

README.md

@@ -92,6 +92,21 @@ $ ./ustreamer --host :: -m jpeg --device-timeout=5 --buffers=3 -r 2592x1944
 $ modprobe bcm2835-v4l2 max_video_width=2592 max_video_height=1944
 ```
 
+-----
+# Integrations
+
+## Nginx
+When uStreamer is behind an Nginx proxy, it's buffering behavior introduces latency into the video stream. It's possible to disable Nginx's buffering to eliminate the additional latency:
+
+```nginx
+location /stream {
+    postpone_output 0;
+    proxy_buffering off;
+    proxy_ignore_headers X-Accel-Buffering;
+    proxy_pass http://ustreamer;
+}
+```
+
 -----
 # Tips & tricks for v4l2
 v4l2 utilities provide the tools to manage USB webcam setting and information. Scripts can be use to make adjustments and run manually or with cron. Running in cron for example to change the exposure settings at certain times of day. The package is available in all Linux distributions and is usually called `v4l-utils`.
@@ -107,7 +122,6 @@ v4l2 utilities provide the tools to manage USB webcam setting and information. S
 -----
 # See also
 * [Running uStreamer via systemd service](https://github.com/pikvm/ustreamer/issues/16).
-* [uStreamer Ansible Role](https://github.com/mtlynch/ansible-role-ustreamer): Use [Ansible](https://docs.ansible.com/ansible/latest/index.html) to compile uStreamer and install it as a systemd service automatically.
 
 -----
 # License

README.ru.md

@@ -92,6 +92,21 @@ $ ./ustreamer --host :: -m jpeg --device-timeout=5 --buffers=3 -r 2592x1944
 $ modprobe bcm2835-v4l2 max_video_width=2592 max_video_height=1944
 ```
 
+-----
+# Интеграция
+
+## Nginx
+Если uStreamer находится на Nginx, то последний будет буферизировать поток и создавать дополнительную задержку в стриме. Чтобы задержки не было, буферизацию можно отключить:
+
+```nginx
+location /stream {
+    postpone_output 0;
+    proxy_buffering off;
+    proxy_ignore_headers X-Accel-Buffering;
+    proxy_pass http://ustreamer;
+}
+```
+
 -----
 # Утилиты V4L2
 V4L2 предоставляет ряд официальных утилит для управления USB-вебкамерами и получения информации об устройствах. С их помощью можно писать всякие настроечные скрипты и запускать их по крону, если, например, вам требуется изменять настройки экспозиции в зависимости от времени суток. Пакет с этими утилитами доступен на всех дистрибутивах Linux и обычно называется `v4l-utils`.
@@ -107,7 +122,6 @@ V4L2 предоставляет ряд официальных утилит дл
 -----
 # Смотрите также
 * [Запуск с помощью systemd-сервиса](https://github.com/pikvm/ustreamer/issues/16).
-* [uStreamer Ansible Role](https://github.com/mtlynch/ansible-role-ustreamer): Использование [Ansible](https://docs.ansible.com/ansible/latest/index.html) для сборки и установки стримера как systemd-сервиса.
 
 -----
 # Лицензия

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

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

linters/pylint.ini

@@ -30,6 +30,7 @@ disable =
 	too-many-ancestors,
 	no-else-return,
 	len-as-condition,
+	unspecified-encoding,
 
 [REPORTS]
 msg-template = {symbol} -- {path}:{line}({obj}): {msg}

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.25" "January 2021"
+.TH USTREAMER-DUMP 1 "version 4.8" "January 2021"
 
 .SH NAME
 ustreamer-dump \- Dump uStreamer's memory sink to file
@@ -38,6 +38,12 @@ Filename to dump output to. Use '-' for stdout. Default: just consume the sink.
 .TP
 .BR \-j ", " \-\-output-json
 Format output as JSON. Required option --output. Default: disabled.
+.TP
+.BR \-c ", " \-\-count\ \fIN
+Limit the number of frames. Default: 0 (infinite).
+.TP
+.BR \-i ", "\-\-interval\ \fIsec
+Delay between reading frames (float). Default: 0.
 
 .SS "Logging options"
 .TP

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.25" "November 2020"
+.TH USTREAMER 1 "version 4.8" "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.25
+pkgver=4.8
 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.25
+PKG_VERSION:=4.8
 PKG_RELEASE:=1
 PKG_MAINTAINER:=Maxim Devaev <mdevaev@gmail.com>
 

python/frame.c

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

python/frame.h

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

python/memsinksh.h

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

python/setup.py

@@ -8,7 +8,7 @@ from distutils.core import setup
 if __name__ == "__main__":
     setup(
         name="ustreamer",
-        version="3.25",
+        version="4.8",
         description="uStreamer tools",
         author="Maxim Devaev",
         author_email="mdevaev@gmail.com",
@@ -18,8 +18,8 @@ if __name__ == "__main__":
                 "ustreamer",
                 libraries=["rt", "m", "pthread"],
                 undef_macros=["NDEBUG"],
-                sources=[name for name in os.listdir(".") if name.endswith(".c")],
-                depends=[name for name in os.listdir(".") if name.endswith(".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/tools.h

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

src/dump/main.c

@@ -25,6 +25,8 @@
 #include <stdbool.h>
 #include <unistd.h>
 #include <signal.h>
+#include <limits.h>
+#include <float.h>
 #include <getopt.h>
 #include <errno.h>
 #include <assert.h>
@@ -44,6 +46,8 @@ enum _OPT_VALUES {
 	_O_SINK_TIMEOUT = 't',
 	_O_OUTPUT = 'o',
 	_O_OUTPUT_JSON = 'j',
+	_O_COUNT = 'c',
+	_O_INTERVAL = 'i',
 
 	_O_HELP = 'h',
 	_O_VERSION = 'v',
@@ -61,6 +65,8 @@ static const struct option _LONG_OPTS[] = {
 	{"sink-timeout",		required_argument,	NULL,	_O_SINK_TIMEOUT},
 	{"output",				required_argument,	NULL,	_O_OUTPUT},
 	{"output-json",			no_argument,		NULL,	_O_OUTPUT_JSON},
+	{"count",				required_argument,	NULL,	_O_COUNT},
+	{"interval",			required_argument,	NULL,	_O_INTERVAL},
 
 	{"log-level",			required_argument,	NULL,	_O_LOG_LEVEL},
 	{"perf",				no_argument,		NULL,	_O_PERF},
@@ -89,7 +95,10 @@ typedef struct {
 static void _signal_handler(int signum);
 static void _install_signal_handlers(void);
 
-static int _dump_sink(const char *sink_name, unsigned sink_timeout, _output_context_s *ctx);
+static int _dump_sink(
+	const char *sink_name, unsigned sink_timeout,
+	long long count, long double interval,
+	_output_context_s *ctx);
 
 static void _help(FILE *fp);
 
@@ -102,6 +111,8 @@ int main(int argc, char *argv[]) {
 	unsigned sink_timeout = 1;
 	char *output_path = NULL;
 	bool output_json = false;
+	long long count = 0;
+	long double interval = 0;
 
 #	define OPT_SET(_dest, _value) { \
 			_dest = _value; \
@@ -118,6 +129,16 @@ int main(int argc, char *argv[]) {
 			break; \
 		}
 
+#	define OPT_LDOUBLE(_name, _dest, _min, _max) { \
+			errno = 0; char *_end = NULL; long double _tmp = strtold(optarg, &_end); \
+			if (errno || *_end || _tmp < _min || _tmp > _max) { \
+				printf("Invalid value for '%s=%s': min=%Lf, max=%Lf\n", _name, optarg, (long double)_min, (long double)_max); \
+				return 1; \
+			} \
+			_dest = _tmp; \
+			break; \
+		}
+
 	char short_opts[128];
 	build_short_options(_LONG_OPTS, short_opts, 128);
 
@@ -127,6 +148,8 @@ int main(int argc, char *argv[]) {
 			case _O_SINK_TIMEOUT:	OPT_NUMBER("--sink-timeout", sink_timeout, 1, 60, 0);
 			case _O_OUTPUT:			OPT_SET(output_path, optarg);
 			case _O_OUTPUT_JSON:	OPT_SET(output_json, true);
+			case _O_COUNT:			OPT_NUMBER("--count", count, 0, LLONG_MAX, 0);
+			case _O_INTERVAL:		OPT_LDOUBLE("--interval", interval, 0, 60);
 
 			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);
@@ -143,6 +166,7 @@ int main(int argc, char *argv[]) {
 		}
 	}
 
+#	undef OPT_LDOUBLE
 #	undef OPT_NUMBER
 #	undef OPT_SET
 
@@ -163,7 +187,7 @@ int main(int argc, char *argv[]) {
 	}
 
 	_install_signal_handlers();
-	int retval = abs(_dump_sink(sink_name, sink_timeout, &ctx));
+	int retval = abs(_dump_sink(sink_name, sink_timeout, count, interval, &ctx));
 	if (ctx.v_output && ctx.destroy) {
 		ctx.destroy(ctx.v_output);
 	}
@@ -201,7 +225,17 @@ static void _install_signal_handlers(void) {
 	assert(!sigaction(SIGPIPE, &sig_act, NULL));
 }
 
-static int _dump_sink(const char *sink_name, unsigned sink_timeout, _output_context_s *ctx) {
+static int _dump_sink(
+	const char *sink_name, unsigned sink_timeout,
+	long long count, long double interval,
+	_output_context_s *ctx) {
+
+	if (count == 0) {
+		count = -1;
+	}
+
+	useconds_t interval_us = interval * 1000000;
+
 	frame_s *frame = frame_init();
 	memsink_s *sink = NULL;
 
@@ -243,6 +277,17 @@ static int _dump_sink(const char *sink_name, unsigned sink_timeout, _output_cont
 			if (ctx->v_output) {
 				ctx->write(ctx->v_output, frame);
 			}
+
+			if (count >= 0) {
+				--count;
+				if (count <= 0) {
+					break;
+				}
+			}
+
+			if (interval_us > 0) {
+				usleep(interval_us);
+			}
 		} else if (error == -2) {
 			usleep(1000);
 		} else {
@@ -282,6 +327,8 @@ static void _help(FILE *fp) {
 	SAY("    -t|--sink-timeout <sec>  ─ Timeout for the upcoming frame. Default: 1.\n");
 	SAY("    -o|--output <filename> ─── Filename to dump output to. Use '-' for stdout. Default: just consume the sink.\n");
 	SAY("    -j|--output-json  ──────── Format output as JSON. Required option --output. Default: disabled.\n");
+	SAY("    -c|--count  <N>  ───────── Limit the number of frames. Default: 0 (infinite).\n");
+	SAY("    -i|--interval <sec>  ───── Delay between reading frames (float). Default: 0.\n");
 	SAY("Logging options:");
 	SAY("════════════════");
 	SAY("    --log-level <N>  ──── Verbosity level of messages from 0 (info) to 3 (debug).");

src/libs/config.h

@@ -22,8 +22,8 @@
 
 #pragma once
 
-#define VERSION_MAJOR 3
-#define VERSION_MINOR 25
+#define VERSION_MAJOR 4
+#define VERSION_MINOR 8
 
 #define MAKE_VERSION2(_major, _minor) #_major "." #_minor
 #define MAKE_VERSION1(_major, _minor) MAKE_VERSION2(_major, _minor)

src/libs/frame.h

@@ -68,7 +68,7 @@ typedef struct {
 		_dest->encode_end_ts = _src->encode_end_ts; \
 	}
 
-inline void frame_copy_meta(const frame_s *src, frame_s *dest) {
+static inline void frame_copy_meta(const frame_s *src, frame_s *dest) {
 	FRAME_COPY_META(src, dest);
 }
 
@@ -97,6 +97,6 @@ unsigned frame_get_padding(const frame_s *frame);
 
 const char *fourcc_to_string(unsigned format, char *buf, size_t size);
 
-inline bool is_jpeg(unsigned format) {
+static inline bool is_jpeg(unsigned format) {
 	return (format == V4L2_PIX_FMT_JPEG || format == V4L2_PIX_FMT_MJPEG);
 }

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);
 
@@ -86,28 +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;
+	}
 
-	bool retval = (sink->has_clients || !FRAME_COMPARE_META_USED_NOTS(sink->mem, frame));
+	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,7 +141,7 @@ int memsink_server_put(memsink_s *sink, const frame_s *frame) {
 
 		sink->mem->magic = MEMSINK_MAGIC;
 		sink->mem->version = MEMSINK_VERSION;
-		sink->has_clients = (sink->mem->last_client_ts + sink->client_ttl > get_now_monotonic());
+		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);

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/tools.h

@@ -110,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);
 }
 

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/http/server.c

@@ -25,6 +25,8 @@
 
 static int _http_preprocess_request(struct evhttp_request *request, server_s *server);
 
+static int _http_check_run_compat_action(struct evhttp_request *request, void *v_server);
+
 static void _http_callback_root(struct evhttp_request *request, void *v_server);
 static void _http_callback_static(struct evhttp_request *request, void *v_server);
 static void _http_callback_state(struct evhttp_request *request, void *v_server);
@@ -220,41 +222,63 @@ static int _http_preprocess_request(struct evhttp_request *request, server_s *se
 		} \
 	}
 
-static void _http_callback_root(struct evhttp_request *request, void *v_server) {
-	server_s *server = (server_s *)v_server;
+static int _http_check_run_compat_action(struct evhttp_request *request, void *v_server) {
+	// MJPG-Streamer compatibility layer
 
-	PREPROCESS_REQUEST;
+	struct evkeyvalq params;
+	int error = 0;
 
-	struct evkeyvalq params; // For mjpg-streamer compatibility
 	evhttp_parse_query(evhttp_request_get_uri(request), &params);
 	const char *action = evhttp_find_header(&params, "action");
 
 	if (action && !strcmp(action, "snapshot")) {
 		_http_callback_snapshot(request, v_server);
+		goto ok;
 	} else if (action && !strcmp(action, "stream")) {
 		_http_callback_stream(request, v_server);
-	} else {
-		struct evbuffer *buf;
-		assert((buf = evbuffer_new()));
-		assert(evbuffer_add_printf(buf, "%s", HTML_INDEX_PAGE));
-		ADD_HEADER("Content-Type", "text/html");
-		evhttp_send_reply(request, HTTP_OK, "OK", buf);
-		evbuffer_free(buf);
+		goto ok;
 	}
 
-	evhttp_clear_headers(&params);
+	error = -1;
+	ok:
+		evhttp_clear_headers(&params);
+		return error;
+}
+
+#define COMPAT_REQUEST { \
+		if (_http_check_run_compat_action(request, v_server) == 0) { \
+			return; \
+		} \
+	}
+
+static void _http_callback_root(struct evhttp_request *request, void *v_server) {
+	server_s *server = (server_s *)v_server;
+
+	PREPROCESS_REQUEST;
+	COMPAT_REQUEST;
+
+	struct evbuffer *buf;
+
+	assert((buf = evbuffer_new()));
+	assert(evbuffer_add_printf(buf, "%s", HTML_INDEX_PAGE));
+	ADD_HEADER("Content-Type", "text/html");
+	evhttp_send_reply(request, HTTP_OK, "OK", buf);
+
+	evbuffer_free(buf);
 }
 
 static void _http_callback_static(struct evhttp_request *request, void *v_server) {
 	server_s *server = (server_s *)v_server;
+
+	PREPROCESS_REQUEST;
+	COMPAT_REQUEST;
+
 	struct evbuffer *buf = NULL;
 	struct evhttp_uri *uri = NULL;
 	char *decoded_path = NULL;
 	char *static_path = NULL;
 	int fd = -1;
 
-	PREPROCESS_REQUEST;
-
 	{
 		char *uri_path;
 
@@ -324,6 +348,8 @@ static void _http_callback_static(struct evhttp_request *request, void *v_server
 		}
 }
 
+#undef COMPAT_REQUEST
+
 static void _http_callback_state(struct evhttp_request *request, void *v_server) {
 	server_s *server = (server_s *)v_server;
 
@@ -342,6 +368,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,
@@ -352,6 +379,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},"

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);
 
@@ -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");

src/ustreamer/stream.c

@@ -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);