From c22f4d41c8d997915232d000cb1eab0e5ea9ff81 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Fri, 9 Oct 2026 17:08:22 -0700 Subject: [PATCH] Add an OBS plugin that pushes fragmented MP4 to a Streamplace node Installing the plugin adds Streamplace to OBS's Settings -> Stream list. On Start Streaming it asks the server for a stream configuration with OBS's multitrack video (GoLiveApi) request, creates the video renditions and audio tracks the server asks for, and pushes them as fragmented MP4 over a chunked HTTP POST. The plugin builds against the OBS 30.2 headers with the existing cross files, and the GitLab build jobs package and upload it for linux-amd64, linux-arm64, windows-amd64, darwin-amd64 and darwin-arm64. --- .gitlab-ci.yml | 2 +- Makefile | 18 + obs-plugin/README.md | 73 ++ obs-plugin/meson.build | 45 ++ obs-plugin/meson_options.txt | 1 + obs-plugin/obs.def | 96 +++ obs-plugin/package.sh | 62 ++ obs-plugin/src/fmp4.c | 505 ++++++++++++ obs-plugin/src/fmp4.h | 66 ++ obs-plugin/src/http.c | 338 ++++++++ obs-plugin/src/http.h | 36 + obs-plugin/src/plugin-main.c | 103 +++ obs-plugin/src/streamplace-output.c | 757 ++++++++++++++++++ obs-plugin/src/streamplace.h | 11 + obs-plugin/subprojects/.gitignore | 5 + obs-plugin/subprojects/obs-studio.wrap | 8 + .../packagefiles/obs-studio/meson.build | 3 + 17 files changed, 2128 insertions(+), 1 deletion(-) create mode 100644 obs-plugin/README.md create mode 100644 obs-plugin/meson.build create mode 100644 obs-plugin/meson_options.txt create mode 100644 obs-plugin/obs.def create mode 100755 obs-plugin/package.sh create mode 100644 obs-plugin/src/fmp4.c create mode 100644 obs-plugin/src/fmp4.h create mode 100644 obs-plugin/src/http.c create mode 100644 obs-plugin/src/http.h create mode 100644 obs-plugin/src/plugin-main.c create mode 100644 obs-plugin/src/streamplace-output.c create mode 100644 obs-plugin/src/streamplace.h create mode 100644 obs-plugin/subprojects/.gitignore create mode 100644 obs-plugin/subprojects/obs-studio.wrap create mode 100644 obs-plugin/subprojects/packagefiles/obs-studio/meson.build diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index e214cfe53..17faadfa8 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -26,7 +26,7 @@ variables: script: - rm -rf /usr/x86_64-w64-mingw32/include/d3dcompiler.h /usr/share/mingw-w64/include/d3dcompiler.h /usr/lib/gcc/x86_64-w64-mingw32/10-posix/libssp.dll.a /usr/lib/gcc/x86_64-w64-mingw32/10-posix/libssp-0.dll /usr/lib/gcc/x86_64-w64-mingw32/10-win32/libssp-0.dll /usr/lib/gcc/x86_64-w64-mingw32/10-win32/libssp.dll.a - git fetch --unshallow || echo 'already unshallow' - - bash -c "make version && make $TARGET && make test-vod && make desktop-$TARGET && make pkg-$TARGET && make ci-upload-node-$TARGET -j$(nproc)" + - bash -c "make version && make $TARGET && make test-vod && make desktop-$TARGET && make pkg-$TARGET && make obs-plugin-$TARGET && make ci-upload-node-$TARGET -j$(nproc) && make ci-upload-obs-plugin-$TARGET" artifacts: reports: dotenv: .ci/build.env diff --git a/Makefile b/Makefile index a200f1f83..f147c0ce7 100644 --- a/Makefile +++ b/Makefile @@ -723,6 +723,20 @@ darwin-arm64: && tar -czvf ../bin/streamplace-$(VERSION)-darwin-arm64.tar.gz ./streamplace \ && cd - +# The OBS plugin in obs-plugin/, cross-compiled with the same cross files as the node. +OBS_PLUGIN_TARGETS=linux-amd64 linux-arm64 windows-amd64 darwin-amd64 darwin-arm64 +OBS_PLUGIN_CROSS_FILE_linux-arm64=--cross-file ../util/linux-arm64-gnu.ini +OBS_PLUGIN_CROSS_FILE_windows-amd64=--cross-file ../util/windows-amd64-gnu.ini +OBS_PLUGIN_CROSS_FILE_darwin-amd64=--cross-file ../util/osxcross-darwin-amd64.ini +OBS_PLUGIN_CROSS_FILE_darwin-arm64=--cross-file ../util/osxcross-darwin-arm64.ini + +.PHONY: $(addprefix obs-plugin-,$(OBS_PLUGIN_TARGETS)) +$(addprefix obs-plugin-,$(OBS_PLUGIN_TARGETS)): obs-plugin-%: + cd obs-plugin \ + && meson setup --reconfigure build-$* $(OBS_PLUGIN_CROSS_FILE_$*) -Dplugin_version=$(VERSION) \ + && meson compile -C build-$* + ./obs-plugin/package.sh $* $(VERSION) $(OUT_DIR) + # unbuffer here is a workaround for wine trying to pop up a terminal window and failing .PHONY: windows-amd64-startup-test windows-amd64-startup-test: @@ -1021,6 +1035,10 @@ ci-upload-android-release: ci-upload-ios: ios $(MAKE) ci-upload-file upload_file=streamplace-$(VERSION)-ios-release.xcarchive.tar.gz +.PHONY: $(addprefix ci-upload-obs-plugin-,$(OBS_PLUGIN_TARGETS)) +$(addprefix ci-upload-obs-plugin-,$(OBS_PLUGIN_TARGETS)): ci-upload-obs-plugin-%: + $(MAKE) ci-upload-file upload_file=obs-streamplace-$(VERSION)-$*.$(if $(findstring windows,$*),zip,tar.gz) + upload_file?="" .PHONY: ci-upload-file ci-upload-file: diff --git a/obs-plugin/README.md b/obs-plugin/README.md new file mode 100644 index 000000000..718258aab --- /dev/null +++ b/obs-plugin/README.md @@ -0,0 +1,73 @@ +# obs-streamplace + +An OBS Studio plugin (OBS 30.2 or newer) that streams to a Streamplace node as +fragmented MP4 over HTTP. The node configures the stream through the same +request OBS sends for Enhanced RTMP multitrack video, so it can choose how many +video renditions and audio tracks the plugin encodes. + +## Using it + +Extract the archive for your platform into OBS's plugin directory: + +| Platform | Directory | +| -------- | --------------------------------------------------- | +| Linux | `~/.config/obs-studio/plugins/` | +| Windows | `C:\ProgramData\obs-studio\plugins\` | +| macOS | `~/Library/Application Support/obs-studio/plugins/` | + +Restart OBS, then pick **Streamplace** in Settings → Stream, the **Local +Streamplace node** server (`http://127.0.0.1:38080`), and your stream key. OBS +offers no free-form server field for listed services, so to use another +address, set `server` in the profile's `service.json`. The plugin only speaks +plain `http://`, and expects a node on the same machine. + +OBS's Settings page only lists services from `rtmp-services`, so on startup the +plugin adds a Streamplace entry to the `services.json` copy in OBS's config +directory (`plugin_config/rtmp-services/`). The entry names the plugin's output +protocol, so OBS hides it whenever the plugin is not loaded. If OBS's service +updater replaces that file, the entry comes back on the next start. + +## Protocol + +When you start streaming, the plugin does three things in order. + +1. It sends `POST {server}/api/ingest/client-configuration` with a JSON body in + the shape of OBS's + [GoLiveApi `PostData`](https://github.com/obsproject/obs-studio/blob/master/frontend/utility/models/multitrack-video.hpp). + `authentication` carries the stream key, and `client.supported_codecs` is + `["h264"]`. +2. It reads the response as a GoLiveApi `Config`: + - `status.result: "error"` stops the stream and shows `status.html_en_us`. + - `ingest_endpoints`: the plugin uses the first entry whose `protocol` is + `FMP4`, replacing `{stream_key}` in its `url_template`. + - `encoder_configurations` replaces the encoders from OBS's output settings + with one video encoder per entry (H.264 encoder types only). Each entry + becomes one video track. + - `audio_configurations.live` creates one audio encoder per entry. `codec` + is `aac` or `opus`, and `track_id` is the zero-based OBS audio mixer track + to encode. + - If either list is empty, the plugin uses OBS's own encoders for that kind + of track. +3. It sends `POST` to the ingest URL with `Transfer-Encoding: chunked` and a + `video/mp4` body: + - First comes an init segment (`ftyp` + `moov`) with every video track, + then every audio track. + - After that it sends one `moof` + `mdat` fragment per frame of the first + video track. Each fragment holds every sample completed since the last one. + - Sample durations come from the real gaps between timestamps, so frames + that OBS drops show up as longer samples. + - The plugin ends the body when streaming stops. + +## Building + +From the repository root, in the builder container (see the +`streamplace-docker` skill): + +```sh +make obs-plugin-linux-amd64 # or linux-arm64, windows-amd64, darwin-amd64, darwin-arm64 +``` + +The archive is written to `bin/`. The build only needs the libobs headers. A +Meson wrap downloads them, pinned to OBS 30.2.3, the oldest release the plugin +supports. On Windows, the build generates an import library for `obs.dll` from +`obs.def`. When the plugin starts calling a new libobs function, add it there. diff --git a/obs-plugin/meson.build b/obs-plugin/meson.build new file mode 100644 index 000000000..dcf349b16 --- /dev/null +++ b/obs-plugin/meson.build @@ -0,0 +1,45 @@ +project( + 'obs-streamplace', + 'c', + default_options: ['c_std=gnu11', 'warning_level=2', 'werror=true', 'buildtype=release'], + meson_version: '>=1.1', +) + +# Headers only: the plugin resolves libobs symbols from the OBS process that loads it. +libobs_dep = subproject('obs-studio').get_variable('libobs_dep') + +c_args = ['-DPLUGIN_VERSION="@0@"'.format(get_option('plugin_version'))] +link_args = [] +link_with = [] +deps = [libobs_dep, dependency('threads')] + +if host_machine.system() == 'windows' + # Windows needs an import library for obs.dll; obs.def lists what the plugin uses. + dlltool = find_program('x86_64-w64-mingw32-dlltool') + link_with += custom_target( + 'obs-import-lib', + input: 'obs.def', + output: 'libobs.dll.a', + command: [dlltool, '--input-def', '@INPUT@', '--output-lib', '@OUTPUT@', '--dllname', 'obs.dll'], + ) + deps += meson.get_compiler('c').find_library('ws2_32') + # One self-contained DLL, without libwinpthread or libgcc beside it. + link_args += ['-static'] +elif host_machine.system() == 'darwin' + c_args += ['-mmacosx-version-min=11.0'] + link_args += ['-mmacosx-version-min=11.0'] +endif + +shared_module( + 'obs-streamplace', + 'src/plugin-main.c', + 'src/streamplace-output.c', + 'src/fmp4.c', + 'src/http.c', + name_prefix: '', + c_args: c_args, + link_args: link_args, + link_with: link_with, + dependencies: deps, + gnu_symbol_visibility: 'hidden', +) diff --git a/obs-plugin/meson_options.txt b/obs-plugin/meson_options.txt new file mode 100644 index 000000000..1ba68cbc9 --- /dev/null +++ b/obs-plugin/meson_options.txt @@ -0,0 +1 @@ +option('plugin_version', type: 'string', value: 'dev', description: 'Version reported in logs and to the server') diff --git a/obs-plugin/obs.def b/obs-plugin/obs.def new file mode 100644 index 000000000..7b372ddc4 --- /dev/null +++ b/obs-plugin/obs.def @@ -0,0 +1,96 @@ +; libobs functions the plugin calls. The Windows build links against an import +; library generated from this list; a missing entry fails that link. +LIBRARY obs.dll +EXPORTS + array_output_serializer_free + array_output_serializer_init + astrcmpi + astrcmpi_n + astrstri + audio_output_get_channels + audio_output_get_sample_rate + bfree + blog + bmalloc + bmemdup + brealloc + dstr_copy + dstr_ncat + dstr_ncopy + dstr_printf + dstr_replace + obs_audio_encoder_create + obs_data_array_count + obs_data_array_create + obs_data_array_erase + obs_data_array_item + obs_data_array_push_back + obs_data_array_release + obs_data_create + obs_data_create_from_json + obs_data_create_from_json_file + obs_data_get_array + obs_data_get_int + obs_data_get_json + obs_data_get_obj + obs_data_get_string + obs_data_release + obs_data_save_json_pretty_safe + obs_data_set_array + obs_data_set_bool + obs_data_set_int + obs_data_set_obj + obs_data_set_string + obs_encoder_audio + obs_encoder_get_codec + obs_encoder_get_extra_data + obs_encoder_get_frame_rate_divisor + obs_encoder_get_height + obs_encoder_get_name + obs_encoder_get_sample_rate + obs_encoder_get_settings + obs_encoder_get_width + obs_encoder_group_create + obs_encoder_group_destroy + obs_encoder_packet_ref + obs_encoder_packet_release + obs_encoder_release + obs_encoder_set_audio + obs_encoder_set_frame_rate_divisor + obs_encoder_set_gpu_scale_type + obs_encoder_set_group + obs_encoder_set_preferred_video_format + obs_encoder_set_scaled_size + obs_encoder_set_video + obs_encoder_video + obs_enum_encoder_types + obs_find_module_file + obs_get_audio + obs_get_audio_info + obs_get_encoder_caps + obs_get_encoder_codec + obs_get_encoder_type + obs_get_module + obs_get_version_string + obs_get_video + obs_get_video_info + obs_module_get_config_path + obs_output_begin_data_capture + obs_output_can_begin_data_capture + obs_output_end_data_capture + obs_output_get_audio_encoder + obs_output_get_service + obs_output_get_video_encoder2 + obs_output_initialize_encoders + obs_output_set_audio_encoder + obs_output_set_last_error + obs_output_set_video_encoder2 + obs_output_signal_stop + obs_parse_avc_header + obs_parse_avc_packet + obs_register_output_s + obs_service_get_connect_info + obs_video_encoder_create + os_mkdirs + os_set_thread_name + video_output_get_info diff --git a/obs-plugin/package.sh b/obs-plugin/package.sh new file mode 100755 index 000000000..bde7b6ee4 --- /dev/null +++ b/obs-plugin/package.sh @@ -0,0 +1,62 @@ +#!/bin/bash +# Packages a built plugin in the layout OBS loads plugins from: +# Linux: extract into ~/.config/obs-studio/plugins/ +# Windows: extract into C:\ProgramData\obs-studio\plugins\ +# macOS: extract into ~/Library/Application Support/obs-studio/plugins/ +# Usage: package.sh + +set -euo pipefail + +TARGET="$1" +VERSION="$2" +OUT="$(realpath "$3")" +NAME=obs-streamplace +BUILD="$(dirname "$0")/build-$TARGET" +STAGE="$(mktemp -d)" +trap 'rm -rf "$STAGE"' EXIT + +case "$TARGET" in +linux-*) + mkdir -p "$STAGE/$NAME/bin/64bit" + cp "$BUILD/$NAME.so" "$STAGE/$NAME/bin/64bit/" + tar -czf "$OUT/$NAME-$VERSION-$TARGET.tar.gz" -C "$STAGE" "$NAME" + ;; +windows-*) + mkdir -p "$STAGE/$NAME/bin/64bit" + cp "$BUILD/$NAME.dll" "$STAGE/$NAME/bin/64bit/" + rm -f "$OUT/$NAME-$VERSION-$TARGET.zip" + (cd "$STAGE" && zip -qr "$OUT/$NAME-$VERSION-$TARGET.zip" "$NAME") + ;; +darwin-*) + BUNDLE="$STAGE/$NAME.plugin/Contents" + mkdir -p "$BUNDLE/MacOS" + cp "$BUILD/$NAME.dylib" "$BUNDLE/MacOS/$NAME" + cat >"$BUNDLE/Info.plist" < + + + + CFBundleName + $NAME + CFBundleIdentifier + place.stream.$NAME + CFBundleExecutable + $NAME + CFBundlePackageType + BNDL + CFBundleInfoDictionaryVersion + 6.0 + CFBundleShortVersionString + $VERSION + LSMinimumSystemVersion + 11.0 + + +EOF + tar -czf "$OUT/$NAME-$VERSION-$TARGET.tar.gz" -C "$STAGE" "$NAME.plugin" + ;; +*) + echo "unknown target $TARGET" >&2 + exit 1 + ;; +esac diff --git a/obs-plugin/src/fmp4.c b/obs-plugin/src/fmp4.c new file mode 100644 index 000000000..c7cd20763 --- /dev/null +++ b/obs-plugin/src/fmp4.c @@ -0,0 +1,505 @@ +#include "fmp4.h" + +#include + +#include + +#define TFHD_DEFAULT_BASE_IS_MOOF 0x020000 +/* data-offset, sample-duration, sample-size, sample-flags, composition-time-offset */ +#define TRUN_FLAGS 0x000F01 +#define SAMPLE_FLAGS_SYNC 0x02000000 +#define SAMPLE_FLAGS_NON_SYNC 0x01010000 + +static const uint32_t unity_matrix[9] = {0x00010000, 0, 0, 0, 0x00010000, 0, 0, 0, 0x40000000}; + +static int64_t box_begin(struct serializer *s, const char *type) +{ + int64_t start = serializer_get_pos(s); + s_wb32(s, 0); + s_write(s, type, 4); + return start; +} + +static int64_t fullbox_begin(struct serializer *s, const char *type, uint8_t version, uint32_t flags) +{ + int64_t start = box_begin(s, type); + s_w8(s, version); + s_wb24(s, flags); + return start; +} + +static void patch_u32(struct serializer *s, int64_t pos, uint32_t value) +{ + int64_t end = serializer_get_pos(s); + serializer_seek(s, pos, SERIALIZE_SEEK_START); + s_wb32(s, value); + serializer_seek(s, end, SERIALIZE_SEEK_START); +} + +static void box_end(struct serializer *s, int64_t start) +{ + patch_u32(s, start, (uint32_t)(serializer_get_pos(s) - start)); +} + +static void write_zeros(struct serializer *s, size_t count) +{ + while (count--) + s_w8(s, 0); +} + +static void write_matrix(struct serializer *s) +{ + for (size_t i = 0; i < 9; i++) + s_wb32(s, unity_matrix[i]); +} + +static bool is_video(const struct fmp4_track *track) +{ + return track->codec == FMP4_H264; +} + +void fmp4_free(struct fmp4_mux *mux) +{ + for (size_t i = 0; i < mux->tracks.num; i++) { + struct fmp4_track *track = &mux->tracks.array[i]; + if (track->has_pending) + obs_encoder_packet_release(&track->pending); + for (size_t j = 0; j < track->samples.num; j++) + obs_encoder_packet_release(&track->samples.array[j].packet); + da_free(track->samples); + bfree(track->config); + } + da_free(mux->tracks); + mux->sequence = 0; + mux->init_written = false; +} + +size_t fmp4_add_track(struct fmp4_mux *mux, const struct fmp4_track *info) +{ + struct fmp4_track *track = da_push_back_new(mux->tracks); + track->id = (uint32_t)mux->tracks.num; + track->codec = info->codec; + track->timescale = info->timescale; + track->timebase_num = info->timebase_num; + track->width = info->width; + track->height = info->height; + track->channels = info->channels; + track->sample_rate = info->sample_rate; + track->bitrate = info->bitrate; + track->config = bmemdup(info->config, info->config_size); + track->config_size = info->config_size; + return mux->tracks.num - 1; +} + +static void write_mvhd(struct serializer *s, uint32_t next_track_id) +{ + int64_t start = fullbox_begin(s, "mvhd", 0, 0); + s_wb32(s, 0); // creation_time + s_wb32(s, 0); // modification_time + s_wb32(s, 1000); // timescale + s_wb32(s, 0); // duration + s_wb32(s, 0x00010000); // rate + s_wb16(s, 0x0100); // volume + write_zeros(s, 10); // reserved + write_matrix(s); + write_zeros(s, 24); // pre_defined + s_wb32(s, next_track_id); + box_end(s, start); +} + +static void write_tkhd(struct serializer *s, const struct fmp4_track *track) +{ + int64_t start = fullbox_begin(s, "tkhd", 0, 0x000003); // enabled, in movie + s_wb32(s, 0); // creation_time + s_wb32(s, 0); // modification_time + s_wb32(s, track->id); + s_wb32(s, 0); // reserved + s_wb32(s, 0); // duration + write_zeros(s, 8); // reserved + s_wb16(s, 0); // layer + s_wb16(s, 0); // alternate_group + s_wb16(s, is_video(track) ? 0 : 0x0100); // volume + s_wb16(s, 0); // reserved + write_matrix(s); + s_wb32(s, track->width << 16); + s_wb32(s, track->height << 16); + box_end(s, start); +} + +static void write_mdhd(struct serializer *s, const struct fmp4_track *track) +{ + int64_t start = fullbox_begin(s, "mdhd", 0, 0); + s_wb32(s, 0); // creation_time + s_wb32(s, 0); // modification_time + s_wb32(s, track->timescale); + s_wb32(s, 0); // duration + s_wb16(s, 0x55C4); // language: und + s_wb16(s, 0); // pre_defined + box_end(s, start); +} + +static void write_hdlr(struct serializer *s, const struct fmp4_track *track) +{ + const char *name = is_video(track) ? "VideoHandler" : "SoundHandler"; + int64_t start = fullbox_begin(s, "hdlr", 0, 0); + s_wb32(s, 0); // pre_defined + s_write(s, is_video(track) ? "vide" : "soun", 4); + write_zeros(s, 12); // reserved + s_write(s, name, strlen(name) + 1); + box_end(s, start); +} + +static void write_dinf(struct serializer *s) +{ + int64_t dinf = box_begin(s, "dinf"); + int64_t dref = fullbox_begin(s, "dref", 0, 0); + s_wb32(s, 1); // entry_count + box_end(s, fullbox_begin(s, "url ", 0, 1)); // media is in this file + box_end(s, dref); + box_end(s, dinf); +} + +static void write_avc1(struct serializer *s, const struct fmp4_track *track) +{ + int64_t start = box_begin(s, "avc1"); + write_zeros(s, 6); // reserved + s_wb16(s, 1); // data_reference_index + write_zeros(s, 16); // pre_defined, reserved + s_wb16(s, (uint16_t)track->width); + s_wb16(s, (uint16_t)track->height); + s_wb32(s, 0x00480000); // horizresolution: 72 dpi + s_wb32(s, 0x00480000); // vertresolution: 72 dpi + s_wb32(s, 0); // reserved + s_wb16(s, 1); // frame_count + write_zeros(s, 32); // compressorname + s_wb16(s, 0x0018); // depth + s_wb16(s, 0xFFFF); // pre_defined + + int64_t avcc = box_begin(s, "avcC"); + s_write(s, track->config, track->config_size); + box_end(s, avcc); + box_end(s, start); +} + +static void write_audio_sample_entry(struct serializer *s, const struct fmp4_track *track) +{ + write_zeros(s, 6); // reserved + s_wb16(s, 1); // data_reference_index + write_zeros(s, 8); // reserved + s_wb16(s, (uint16_t)track->channels); + s_wb16(s, 16); // samplesize + s_wb16(s, 0); // pre_defined + s_wb16(s, 0); // reserved + s_wb32(s, track->sample_rate << 16); +} + +/* ISO/IEC 14496-1 expandable descriptor size, always in the 4-byte form. */ +static void write_descriptor(struct serializer *s, uint8_t tag, uint32_t size) +{ + s_w8(s, tag); + for (int i = 3; i > 0; i--) + s_w8(s, (uint8_t)((size >> (7 * i)) & 0x7F) | 0x80); + s_w8(s, size & 0x7F); +} + +static void write_mp4a(struct serializer *s, const struct fmp4_track *track) +{ + int64_t start = box_begin(s, "mp4a"); + write_audio_sample_entry(s, track); + + int64_t esds = fullbox_begin(s, "esds", 0, 0); + uint32_t specific_info_size = 5 + (uint32_t)track->config_size; + write_descriptor(s, 0x03, 3 + 5 + 13 + specific_info_size + 5 + 1); // ES_Descriptor + s_wb16(s, (uint16_t)track->id); + s_w8(s, 0); // flags + write_descriptor(s, 0x04, 13 + specific_info_size); // DecoderConfigDescriptor + s_w8(s, 0x40); // objectTypeIndication: AAC + s_w8(s, 0x15); // streamType: audio + s_wb24(s, 0); // bufferSizeDB + s_wb32(s, track->bitrate); // maxBitrate + s_wb32(s, track->bitrate); // avgBitrate + write_descriptor(s, 0x05, (uint32_t)track->config_size); // DecoderSpecificInfo + s_write(s, track->config, track->config_size); + write_descriptor(s, 0x06, 1); // SLConfigDescriptor + s_w8(s, 0x02); // predefined: MP4 + box_end(s, esds); + + box_end(s, start); +} + +static uint16_t read_le16(const uint8_t *p) +{ + return (uint16_t)(p[0] | (p[1] << 8)); +} + +static uint32_t read_le32(const uint8_t *p) +{ + return (uint32_t)p[0] | ((uint32_t)p[1] << 8) | ((uint32_t)p[2] << 16) | ((uint32_t)p[3] << 24); +} + +/* Opus in ISO BMFF: dOps is the OpusHead with its fields made big-endian. */ +static void write_opus(struct serializer *s, const struct fmp4_track *track) +{ + const uint8_t *head = track->config; + uint8_t channels = head[9]; + uint8_t mapping_family = head[18]; + + int64_t start = box_begin(s, "Opus"); + write_audio_sample_entry(s, track); + + int64_t dops = box_begin(s, "dOps"); + s_w8(s, 0); // version + s_w8(s, channels); + s_wb16(s, read_le16(head + 10)); // pre-skip + s_wb32(s, read_le32(head + 12)); // input sample rate + s_wb16(s, read_le16(head + 16)); // output gain + s_w8(s, mapping_family); + if (mapping_family && track->config_size >= 21u + channels) + s_write(s, head + 19, 2 + channels); // stream count, coupled count, mapping + box_end(s, dops); + + box_end(s, start); +} + +static void write_stbl(struct serializer *s, const struct fmp4_track *track) +{ + int64_t stbl = box_begin(s, "stbl"); + + int64_t stsd = fullbox_begin(s, "stsd", 0, 0); + s_wb32(s, 1); // entry_count + switch (track->codec) { + case FMP4_H264: + write_avc1(s, track); + break; + case FMP4_AAC: + write_mp4a(s, track); + break; + case FMP4_OPUS: + write_opus(s, track); + break; + } + box_end(s, stsd); + + /* Fragmented: the sample tables are empty and every sample lives in a trun. */ + int64_t box = fullbox_begin(s, "stts", 0, 0); + s_wb32(s, 0); + box_end(s, box); + box = fullbox_begin(s, "stsc", 0, 0); + s_wb32(s, 0); + box_end(s, box); + box = fullbox_begin(s, "stsz", 0, 0); + s_wb32(s, 0); // sample_size + s_wb32(s, 0); // sample_count + box_end(s, box); + box = fullbox_begin(s, "stco", 0, 0); + s_wb32(s, 0); + box_end(s, box); + + box_end(s, stbl); +} + +/* Encoders can start before zero (AAC and Opus priming): the track's media + * timeline is shifted to start at zero, and the edit list shifts it back. */ +static void write_edts(struct serializer *s, const struct fmp4_track *track) +{ + int64_t edts = box_begin(s, "edts"); + int64_t elst = fullbox_begin(s, "elst", 0, 0); + s_wb32(s, 1); // entry_count + s_wb32(s, 0); // segment_duration: the whole (fragmented) track + s_wb32(s, (uint32_t)(track->dts_offset * track->timebase_num)); // media_time + s_wb32(s, 0x00010000); // media_rate + box_end(s, elst); + box_end(s, edts); +} + +static void write_trak(struct serializer *s, const struct fmp4_track *track) +{ + int64_t trak = box_begin(s, "trak"); + write_tkhd(s, track); + if (track->dts_offset) + write_edts(s, track); + + int64_t mdia = box_begin(s, "mdia"); + write_mdhd(s, track); + write_hdlr(s, track); + + int64_t minf = box_begin(s, "minf"); + if (is_video(track)) { + int64_t vmhd = fullbox_begin(s, "vmhd", 0, 1); + write_zeros(s, 8); // graphicsmode, opcolor + box_end(s, vmhd); + } else { + int64_t smhd = fullbox_begin(s, "smhd", 0, 0); + write_zeros(s, 4); // balance, reserved + box_end(s, smhd); + } + write_dinf(s); + write_stbl(s, track); + box_end(s, minf); + + box_end(s, mdia); + box_end(s, trak); +} + +static void write_init(struct fmp4_mux *mux, struct serializer *s) +{ + int64_t ftyp = box_begin(s, "ftyp"); + s_write(s, "iso6", 4); // major_brand + s_wb32(s, 0); // minor_version + s_write(s, "iso6", 4); + s_write(s, "mp41", 4); + box_end(s, ftyp); + + int64_t moov = box_begin(s, "moov"); + write_mvhd(s, (uint32_t)mux->tracks.num + 1); + for (size_t i = 0; i < mux->tracks.num; i++) + write_trak(s, &mux->tracks.array[i]); + + int64_t mvex = box_begin(s, "mvex"); + for (size_t i = 0; i < mux->tracks.num; i++) { + int64_t trex = fullbox_begin(s, "trex", 0, 0); + s_wb32(s, mux->tracks.array[i].id); + s_wb32(s, 1); // default_sample_description_index + s_wb32(s, 0); // default_sample_duration + s_wb32(s, 0); // default_sample_size + s_wb32(s, 0); // default_sample_flags + box_end(s, trex); + } + box_end(s, mvex); + box_end(s, moov); +} + +static void complete_pending(struct fmp4_track *track, uint32_t duration) +{ + struct fmp4_sample *sample = da_push_back_new(track->samples); + sample->packet = track->pending; + sample->duration = duration; + track->last_duration = duration; + track->has_pending = false; +} + +void fmp4_push(struct fmp4_mux *mux, size_t track_idx, struct encoder_packet *packet) +{ + struct fmp4_track *track = &mux->tracks.array[track_idx]; + if (!track->started) { + track->started = true; + track->dts_offset = packet->dts < 0 ? -packet->dts : 0; + } + if (track->has_pending) { + int64_t delta = packet->dts - track->pending.dts; + complete_pending(track, delta > 0 ? (uint32_t)(delta * track->timebase_num) : 0); + } + track->pending = *packet; + track->has_pending = true; +} + +static uint32_t sample_flags(const struct fmp4_track *track, const struct encoder_packet *packet) +{ + if (!is_video(track) || packet->keyframe) + return SAMPLE_FLAGS_SYNC; + return SAMPLE_FLAGS_NON_SYNC; +} + +static void write_fragment(struct fmp4_mux *mux, struct serializer *s, bool force_init) +{ + if (!mux->init_written) { + bool all_started = true; + for (size_t i = 0; i < mux->tracks.num; i++) + all_started = all_started && mux->tracks.array[i].started; + if (!all_started && !force_init) + return; + write_init(mux, s); + mux->init_written = true; + } + + int64_t data_offset_pos[FMP4_MAX_TRACKS]; + bool any = false; + for (size_t i = 0; i < mux->tracks.num; i++) + any = any || mux->tracks.array[i].samples.num; + if (!any) + return; + + int64_t moof = box_begin(s, "moof"); + int64_t mfhd = fullbox_begin(s, "mfhd", 0, 0); + s_wb32(s, ++mux->sequence); + box_end(s, mfhd); + + for (size_t i = 0; i < mux->tracks.num; i++) { + struct fmp4_track *track = &mux->tracks.array[i]; + if (!track->samples.num) + continue; + + int64_t traf = box_begin(s, "traf"); + int64_t tfhd = fullbox_begin(s, "tfhd", 0, TFHD_DEFAULT_BASE_IS_MOOF); + s_wb32(s, track->id); + box_end(s, tfhd); + + uint64_t first_dts = (uint64_t)(track->samples.array[0].packet.dts + track->dts_offset); + int64_t tfdt = fullbox_begin(s, "tfdt", 1, 0); + s_wb64(s, first_dts * track->timebase_num); + box_end(s, tfdt); + + /* Version 1: signed composition offsets. */ + int64_t trun = fullbox_begin(s, "trun", 1, TRUN_FLAGS); + s_wb32(s, (uint32_t)track->samples.num); + data_offset_pos[i] = serializer_get_pos(s); + s_wb32(s, 0); // data_offset, patched below + for (size_t j = 0; j < track->samples.num; j++) { + const struct fmp4_sample *sample = &track->samples.array[j]; + int64_t offset = (sample->packet.pts - sample->packet.dts) * track->timebase_num; + s_wb32(s, sample->duration); + s_wb32(s, (uint32_t)sample->packet.size); + s_wb32(s, sample_flags(track, &sample->packet)); + s_wb32(s, (uint32_t)(int32_t)offset); + } + box_end(s, trun); + box_end(s, traf); + } + box_end(s, moof); + + /* Sample data follows the moof in track order; trun offsets are relative to the moof. */ + uint32_t data_offset = (uint32_t)(serializer_get_pos(s) - moof) + 8; + int64_t mdat = box_begin(s, "mdat"); + for (size_t i = 0; i < mux->tracks.num; i++) { + struct fmp4_track *track = &mux->tracks.array[i]; + if (!track->samples.num) + continue; + + patch_u32(s, data_offset_pos[i], data_offset); + for (size_t j = 0; j < track->samples.num; j++) { + struct encoder_packet *packet = &track->samples.array[j].packet; + s_write(s, packet->data, packet->size); + data_offset += (uint32_t)packet->size; + obs_encoder_packet_release(packet); + } + da_resize(track->samples, 0); + } + box_end(s, mdat); +} + +void fmp4_write_fragment(struct fmp4_mux *mux, struct serializer *s) +{ + write_fragment(mux, s, false); +} + +static uint32_t default_duration(const struct fmp4_track *track) +{ + switch (track->codec) { + case FMP4_AAC: + return 1024; + case FMP4_OPUS: + return 960; + default: + return track->timebase_num; // one frame + } +} + +void fmp4_finish(struct fmp4_mux *mux, struct serializer *s) +{ + for (size_t i = 0; i < mux->tracks.num; i++) { + struct fmp4_track *track = &mux->tracks.array[i]; + if (track->has_pending) + complete_pending(track, track->last_duration ? track->last_duration : default_duration(track)); + } + write_fragment(mux, s, true); +} diff --git a/obs-plugin/src/fmp4.h b/obs-plugin/src/fmp4.h new file mode 100644 index 000000000..3d08973bf --- /dev/null +++ b/obs-plugin/src/fmp4.h @@ -0,0 +1,66 @@ +#pragma once + +#include +#include +#include + +/* Fragmented MP4 muxer for a live push: one init segment (ftyp+moov) once every + * track has a sample, then a moof+mdat fragment each time the caller asks. + * Every track holds back its newest sample until the next one arrives, so each + * sample's duration is the real gap to its successor (OBS drops frames under + * load). */ + +#define FMP4_MAX_TRACKS (MAX_OUTPUT_VIDEO_ENCODERS + MAX_OUTPUT_AUDIO_ENCODERS) + +enum fmp4_codec { + FMP4_H264, + FMP4_AAC, + FMP4_OPUS, +}; + +struct fmp4_sample { + struct encoder_packet packet; + uint32_t duration; +}; + +struct fmp4_track { + uint32_t id; + enum fmp4_codec codec; + /* Media timescale is the encoder's timebase denominator; packet + * timestamps are multiplied by the numerator. */ + uint32_t timescale; + uint32_t timebase_num; + uint32_t width, height; + uint32_t channels, sample_rate; + uint32_t bitrate; + /* avcC payload for H.264, AudioSpecificConfig for AAC, OpusHead for Opus. */ + uint8_t *config; + size_t config_size; + + bool started; + /* Packets before zero (encoder priming), in timebase units; undone by an edit list. */ + int64_t dts_offset; + bool has_pending; + struct encoder_packet pending; + uint32_t last_duration; + DARRAY(struct fmp4_sample) samples; +}; + +struct fmp4_mux { + DARRAY(struct fmp4_track) tracks; + uint32_t sequence; + bool init_written; +}; + +void fmp4_free(struct fmp4_mux *mux); + +/* Copies the track description and its config bytes. Returns the track index. */ +size_t fmp4_add_track(struct fmp4_mux *mux, const struct fmp4_track *info); +/* Takes ownership of a packet whose data was ref'd or allocated by libobs. + * Video packets must already be length-prefixed (AVCC). */ +void fmp4_push(struct fmp4_mux *mux, size_t track_idx, struct encoder_packet *packet); +/* Writes the init segment once possible, then every sample with a known + * duration; no-op when there is nothing to write. */ +void fmp4_write_fragment(struct fmp4_mux *mux, struct serializer *s); +/* Writes the held-back samples, reusing each track's previous duration. */ +void fmp4_finish(struct fmp4_mux *mux, struct serializer *s); diff --git a/obs-plugin/src/http.c b/obs-plugin/src/http.c new file mode 100644 index 000000000..63cafd94e --- /dev/null +++ b/obs-plugin/src/http.c @@ -0,0 +1,338 @@ +#include "http.h" + +#include +#include +#include +#include + +#ifdef _WIN32 +#include +#include +typedef SOCKET socket_t; +#define INVALID_SOCK INVALID_SOCKET +#define close_socket closesocket +#else +#include +#include +#include +#include +#include +#include +#include +#include +typedef int socket_t; +#define INVALID_SOCK (-1) +#define close_socket close +#endif + +#ifdef MSG_NOSIGNAL +#define SEND_FLAGS MSG_NOSIGNAL +#else +#define SEND_FLAGS 0 +#endif + +#define CONNECT_TIMEOUT_SEC 5 +#define IO_TIMEOUT_SEC 10 +#define MAX_RESPONSE_SIZE (1024 * 1024) + +void http_init(void) +{ +#ifdef _WIN32 + WSADATA wsa; + WSAStartup(MAKEWORD(2, 2), &wsa); +#endif +} + +void http_free(void) +{ +#ifdef _WIN32 + WSACleanup(); +#endif +} + +bool http_parse_url(const char *url, struct http_url *out, struct dstr *err) +{ + static const char scheme[] = "http://"; + memset(out, 0, sizeof(*out)); + + if (astrcmpi_n(url, scheme, sizeof(scheme) - 1) != 0) { + dstr_printf(err, "Unsupported URL '%s': only http:// URLs are supported", url); + return false; + } + + const char *host = url + sizeof(scheme) - 1; + const char *path = strchr(host, '/'); + const char *host_end = path ? path : host + strlen(host); + const char *port = NULL; + + if (*host == '[') { + const char *close = memchr(host, ']', host_end - host); + if (close) { + dstr_ncopy(&out->host, host + 1, close - host - 1); + if (close[1] == ':') + port = close + 2; + } + } else { + const char *colon = memchr(host, ':', host_end - host); + dstr_ncopy(&out->host, host, (colon ? colon : host_end) - host); + if (colon) + port = colon + 1; + } + + if (port && port < host_end) + dstr_ncopy(&out->port, port, host_end - port); + else + dstr_copy(&out->port, "80"); + dstr_copy(&out->path, path ? path : "/"); + + if (dstr_is_empty(&out->host)) { + dstr_printf(err, "Invalid URL '%s'", url); + http_url_free(out); + return false; + } + return true; +} + +void http_url_free(struct http_url *url) +{ + dstr_free(&url->host); + dstr_free(&url->port); + dstr_free(&url->path); +} + +static void set_nonblocking(socket_t s, bool nonblocking) +{ +#ifdef _WIN32 + u_long mode = nonblocking; + ioctlsocket(s, FIONBIO, &mode); +#else + int flags = fcntl(s, F_GETFL, 0); + fcntl(s, F_SETFL, nonblocking ? flags | O_NONBLOCK : flags & ~O_NONBLOCK); +#endif +} + +static bool connect_with_timeout(socket_t s, const struct sockaddr *addr, socklen_t len) +{ + set_nonblocking(s, true); + if (connect(s, addr, len) != 0) { +#ifdef _WIN32 + if (WSAGetLastError() != WSAEWOULDBLOCK) + return false; +#else + if (errno != EINPROGRESS) + return false; +#endif + fd_set wfds, efds; + FD_ZERO(&wfds); + FD_ZERO(&efds); + FD_SET(s, &wfds); + FD_SET(s, &efds); + struct timeval tv = {CONNECT_TIMEOUT_SEC, 0}; + if (select((int)s + 1, NULL, &wfds, &efds, &tv) <= 0) + return false; + + int so_error = 0; + socklen_t so_len = sizeof(so_error); + if (getsockopt(s, SOL_SOCKET, SO_ERROR, (char *)&so_error, &so_len) != 0 || so_error != 0) + return false; + } + set_nonblocking(s, false); + return true; +} + +static void configure_socket(socket_t s) +{ +#ifdef _WIN32 + DWORD timeout = IO_TIMEOUT_SEC * 1000; +#else + struct timeval timeout = {IO_TIMEOUT_SEC, 0}; +#endif + setsockopt(s, SOL_SOCKET, SO_RCVTIMEO, (const char *)&timeout, sizeof(timeout)); + setsockopt(s, SOL_SOCKET, SO_SNDTIMEO, (const char *)&timeout, sizeof(timeout)); + + int one = 1; + setsockopt(s, IPPROTO_TCP, TCP_NODELAY, (const char *)&one, sizeof(one)); +#ifdef SO_NOSIGPIPE + setsockopt(s, SOL_SOCKET, SO_NOSIGPIPE, &one, sizeof(one)); +#endif +} + +static socket_t open_socket(const struct http_url *url, struct dstr *err) +{ + struct addrinfo hints = {0}; + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_STREAM; + + struct addrinfo *res = NULL; + if (getaddrinfo(url->host.array, url->port.array, &hints, &res) != 0) { + dstr_printf(err, "Could not resolve '%s'", url->host.array); + return INVALID_SOCK; + } + + socket_t s = INVALID_SOCK; + for (struct addrinfo *ai = res; ai; ai = ai->ai_next) { + s = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol); + if (s == INVALID_SOCK) + continue; + if (connect_with_timeout(s, ai->ai_addr, (socklen_t)ai->ai_addrlen)) + break; + close_socket(s); + s = INVALID_SOCK; + } + freeaddrinfo(res); + + if (s == INVALID_SOCK) + dstr_printf(err, "Could not connect to %s:%s", url->host.array, url->port.array); + else + configure_socket(s); + return s; +} + +static bool send_all(socket_t s, const void *data, size_t size) +{ + const char *p = data; + while (size) { + int n = send(s, p, size > INT_MAX ? INT_MAX : (int)size, SEND_FLAGS); + if (n <= 0) + return false; + p += n; + size -= (size_t)n; + } + return true; +} + +static bool send_request_head(socket_t s, const struct http_url *url, const char *content_type, + const char *extra_headers) +{ + struct dstr head = {0}; + bool ipv6 = strchr(url->host.array, ':') != NULL; + dstr_printf(&head, + "POST %s HTTP/1.1\r\n" + "Host: %s%s%s:%s\r\n" + "User-Agent: obs-streamplace\r\n" + "Content-Type: %s\r\n" + "%s\r\n", + url->path.array, ipv6 ? "[" : "", url->host.array, ipv6 ? "]" : "", url->port.array, content_type, + extra_headers); + bool ok = send_all(s, head.array, head.len); + dstr_free(&head); + return ok; +} + +static bool decode_chunked(const char *p, const char *end, struct dstr *out) +{ + while (p < end) { + char *size_end; + unsigned long size = strtoul(p, &size_end, 16); + const char *data = strstr(size_end, "\r\n"); + if (!data) + return false; + data += 2; + if (size == 0) + return true; + if ((size_t)(end - data) < size) + return false; + dstr_ncat(out, data, size); + p = data + size + 2; + } + return false; +} + +/* Reads until the server closes the connection, then splits the response. */ +static bool read_response(socket_t s, long *status, struct dstr *body) +{ + struct dstr raw = {0}; + char buf[4096]; + int n; + while (raw.len < MAX_RESPONSE_SIZE && (n = recv(s, buf, sizeof(buf), 0)) > 0) + dstr_ncat(&raw, buf, n); + + bool ok = false; + const char *header_end = raw.array ? strstr(raw.array, "\r\n\r\n") : NULL; + if (header_end && sscanf(raw.array, "HTTP/%*d.%*d %ld", status) == 1) { + struct dstr headers = {0}; + dstr_ncopy(&headers, raw.array, header_end - raw.array); + const char *start = header_end + 4; + const char *end = raw.array + raw.len; + + dstr_free(body); + if (astrstri(headers.array, "transfer-encoding: chunked")) { + ok = decode_chunked(start, end, body); + } else { + dstr_ncopy(body, start, end - start); + ok = true; + } + dstr_free(&headers); + } + + dstr_free(&raw); + return ok; +} + +bool http_post_json(const char *url_str, const char *body, long *status, struct dstr *response, struct dstr *err) +{ + struct http_url url; + if (!http_parse_url(url_str, &url, err)) + return false; + + socket_t s = open_socket(&url, err); + bool ok = false; + if (s != INVALID_SOCK) { + struct dstr headers = {0}; + dstr_printf(&headers, "Content-Length: %lu\r\nConnection: close\r\n", (unsigned long)strlen(body)); + ok = send_request_head(s, &url, "application/json", headers.array) && send_all(s, body, strlen(body)) && + read_response(s, status, response); + if (!ok) + dstr_printf(err, "Request to %s failed", url_str); + dstr_free(&headers); + close_socket(s); + } + + http_url_free(&url); + return ok; +} + +bool http_begin_chunked_post(struct http_conn *conn, const char *url_str, const char *content_type, struct dstr *err) +{ + struct http_url url; + if (!http_parse_url(url_str, &url, err)) + return false; + + socket_t s = open_socket(&url, err); + bool ok = s != INVALID_SOCK && + send_request_head(s, &url, content_type, "Transfer-Encoding: chunked\r\nConnection: close\r\n"); + if (ok) { + conn->sock = (intptr_t)s; + } else if (s != INVALID_SOCK) { + dstr_printf(err, "Could not send request to %s", url_str); + close_socket(s); + } + + http_url_free(&url); + return ok; +} + +bool http_write_chunk(struct http_conn *conn, const uint8_t *data, size_t size) +{ + char header[32]; + int len = snprintf(header, sizeof(header), "%lx\r\n", (unsigned long)size); + socket_t s = (socket_t)conn->sock; + return send_all(s, header, (size_t)len) && send_all(s, data, size) && send_all(s, "\r\n", 2); +} + +long http_end_chunked_post(struct http_conn *conn) +{ + socket_t s = (socket_t)conn->sock; + long status = 0; + struct dstr body = {0}; + if (send_all(s, "0\r\n\r\n", 5)) + read_response(s, &status, &body); + dstr_free(&body); + return status; +} + +void http_close(struct http_conn *conn) +{ + close_socket((socket_t)conn->sock); + conn->sock = (intptr_t)INVALID_SOCK; +} diff --git a/obs-plugin/src/http.h b/obs-plugin/src/http.h new file mode 100644 index 000000000..40a1eef38 --- /dev/null +++ b/obs-plugin/src/http.h @@ -0,0 +1,36 @@ +#pragma once + +#include +#include +#include + +#include + +/* A deliberately small HTTP/1.1 client over plain TCP: the plugin talks to a + * Streamplace node on the same machine, so there is no TLS. */ + +struct http_url { + struct dstr host; + struct dstr port; + struct dstr path; +}; + +struct http_conn { + intptr_t sock; +}; + +void http_init(void); +void http_free(void); + +bool http_parse_url(const char *url, struct http_url *out, struct dstr *err); +void http_url_free(struct http_url *url); + +/* POSTs a JSON body and returns the response body. */ +bool http_post_json(const char *url, const char *body, long *status, struct dstr *response, struct dstr *err); + +/* Opens a POST with a chunked request body. */ +bool http_begin_chunked_post(struct http_conn *conn, const char *url, const char *content_type, struct dstr *err); +bool http_write_chunk(struct http_conn *conn, const uint8_t *data, size_t size); +/* Terminates the body and returns the response status, or 0 if none arrived. */ +long http_end_chunked_post(struct http_conn *conn); +void http_close(struct http_conn *conn); diff --git a/obs-plugin/src/plugin-main.c b/obs-plugin/src/plugin-main.c new file mode 100644 index 000000000..d1e4f4aec --- /dev/null +++ b/obs-plugin/src/plugin-main.c @@ -0,0 +1,103 @@ +#include + +#include + +#include "http.h" +#include "streamplace.h" + +OBS_DECLARE_MODULE() + +MODULE_EXPORT const char *obs_module_description(void) +{ + return "Streams to a Streamplace node as fragmented MP4 over HTTP"; +} + +/* OBS's Settings → Stream page only knows the rtmp-services service types, so a + * third-party output is reached through a services.json entry whose "protocol" + * this plugin registers: rtmp-services hides the entry unless the protocol is + * registered, and the frontend starts whichever output registered it. */ +static const char service_entry[] = "{" + "\"name\": \"Streamplace\"," + "\"common\": true," + "\"protocol\": \"" STREAMPLACE_PROTOCOL "\"," + "\"more_info_link\": \"https://stream.place/about\"," + "\"stream_key_link\": \"https://stream.place/live\"," + "\"servers\": [{\"name\": \"Local Streamplace node\", \"url\": \"http://127.0.0.1:38080\"}]," + "\"supported video codecs\": [\"h264\"]," + "\"supported audio codecs\": [\"aac\", \"opus\"]" + "}"; + +/* Adds the entry to rtmp-services' services.json in the user's config + * directory, which it reads in preference to its bundled copy. When OBS's + * service updater replaces that file, the next startup adds the entry again. */ +static void install_service_entry(void) +{ + obs_module_t *rtmp_services = obs_get_module("rtmp-services"); + if (!rtmp_services) { + do_log(LOG_WARNING, "rtmp-services is not loaded; Streamplace will not be listed in Settings"); + return; + } + + char *config_dir = obs_module_get_config_path(rtmp_services, ""); + char *config_file = obs_module_get_config_path(rtmp_services, "services.json"); + char *bundled_file = obs_find_module_file(rtmp_services, "services.json"); + + obs_data_t *root = obs_data_create_from_json_file(config_file); + if (!root && bundled_file) + root = obs_data_create_from_json_file(bundled_file); + + obs_data_array_t *services = root ? obs_data_get_array(root, "services") : NULL; + if (services) { + obs_data_t *entry = obs_data_create_from_json(service_entry); + const char *entry_json = obs_data_get_json(entry); + bool current = false; + + for (size_t i = obs_data_array_count(services); i > 0; i--) { + obs_data_t *service = obs_data_array_item(services, i - 1); + if (strcmp(obs_data_get_string(service, "name"), "Streamplace") == 0) { + if (strcmp(obs_data_get_json(service), entry_json) == 0) + current = true; + else + obs_data_array_erase(services, i - 1); + } + obs_data_release(service); + } + + if (!current) { + obs_data_array_push_back(services, entry); + os_mkdirs(config_dir); + if (obs_data_save_json_pretty_safe(root, config_file, ".tmp", NULL)) + do_log(LOG_INFO, "Added the Streamplace service to %s", config_file); + else + do_log(LOG_WARNING, "Could not write %s", config_file); + } + + obs_data_release(entry); + obs_data_array_release(services); + } else { + do_log(LOG_WARNING, "Could not read rtmp-services' services.json"); + } + + obs_data_release(root); + bfree(bundled_file); + bfree(config_file); + bfree(config_dir); +} + +bool obs_module_load(void) +{ + http_init(); + obs_register_output(&streamplace_output_info); + do_log(LOG_INFO, "loaded version %s", PLUGIN_VERSION); + return true; +} + +void obs_module_post_load(void) +{ + install_service_entry(); +} + +void obs_module_unload(void) +{ + http_free(); +} diff --git a/obs-plugin/src/streamplace-output.c b/obs-plugin/src/streamplace-output.c new file mode 100644 index 000000000..8f14d008c --- /dev/null +++ b/obs-plugin/src/streamplace-output.c @@ -0,0 +1,757 @@ +#include +#include +#include + +#include +#include +#include +#include + +#include "fmp4.h" +#include "http.h" +#include "streamplace.h" + +/* Same request and response shapes as OBS's multitrack video ("GoLiveApi") + * configuration endpoint, so one server endpoint can configure both this + * plugin and OBS's built-in Enhanced RTMP multitrack output. */ +#define CONFIG_PATH "/api/ingest/client-configuration" +#define CONFIG_SCHEMA_VERSION "2025-01-25" +#define INGEST_PROTOCOL "FMP4" +#define MAX_QUEUED_BYTES (64 * 1024 * 1024) + +struct sp_output { + obs_output_t *output; + obs_encoder_group_t *encoder_group; + + pthread_t connect_thread; + bool connect_thread_active; + pthread_t send_thread; + bool send_thread_active; + struct http_conn conn; + + pthread_mutex_t mutex; + pthread_cond_t cond; + /* Everything below is guarded by mutex. */ + bool stopping; /* stop() was called */ + bool capturing; /* packets are muxed and queued for the send thread */ + bool finishing; /* the send thread should wrap up and exit */ + int error_code; /* why it is finishing, or OBS_OUTPUT_SUCCESS */ + bool mux_started; + size_t video_tracks, audio_tracks; + struct fmp4_mux mux; + struct serializer queue_s; + struct array_output_data queue; + uint64_t total_bytes; +}; + +static bool is_stopping(struct sp_output *sp) +{ + pthread_mutex_lock(&sp->mutex); + bool stopping = sp->stopping; + pthread_mutex_unlock(&sp->mutex); + return stopping; +} + +/* --------------------------------------------------------------------------- + * Stream configuration */ + +static obs_data_t *build_config_request(const char *key) +{ + obs_data_t *post = obs_data_create(); + obs_data_set_string(post, "service", "Streamplace"); + obs_data_set_string(post, "schema_version", CONFIG_SCHEMA_VERSION); + obs_data_set_string(post, "authentication", key); + + obs_data_t *client = obs_data_create(); + obs_data_set_string(client, "name", "obs-studio"); + obs_data_set_string(client, "version", obs_get_version_string()); + obs_data_set_string(client, "plugin_version", PLUGIN_VERSION); + obs_data_set_string(client, "supported_codecs", "h264"); + obs_data_set_obj(post, "client", client); + obs_data_release(client); + + obs_data_t *preferences = obs_data_create(); + obs_data_set_bool(preferences, "vod_track_audio", false); + struct obs_video_info ovi; + if (obs_get_video_info(&ovi)) { + obs_data_set_int(preferences, "composition_gpu_index", ovi.adapter); + obs_data_t *canvas = obs_data_create(); + obs_data_set_int(canvas, "width", ovi.output_width); + obs_data_set_int(canvas, "height", ovi.output_height); + obs_data_set_int(canvas, "canvas_width", ovi.base_width); + obs_data_set_int(canvas, "canvas_height", ovi.base_height); + obs_data_t *framerate = obs_data_create(); + obs_data_set_int(framerate, "numerator", ovi.fps_num); + obs_data_set_int(framerate, "denominator", ovi.fps_den); + obs_data_set_obj(canvas, "framerate", framerate); + obs_data_release(framerate); + obs_data_array_t *canvases = obs_data_array_create(); + obs_data_array_push_back(canvases, canvas); + obs_data_release(canvas); + obs_data_set_array(preferences, "canvases", canvases); + obs_data_array_release(canvases); + } + struct obs_audio_info oai; + if (obs_get_audio_info(&oai)) { + obs_data_set_int(preferences, "audio_samples_per_sec", oai.samples_per_sec); + obs_data_set_int(preferences, "audio_channels", get_audio_channels(oai.speakers)); + } + obs_data_set_obj(post, "preferences", preferences); + obs_data_release(preferences); + + return post; +} + +/* obs_data has no arrays of strings, so the codec list is patched in as JSON. */ +static char *config_request_json(const char *key) +{ + obs_data_t *post = build_config_request(key); + struct dstr json = {0}; + dstr_copy(&json, obs_data_get_json(post)); + dstr_replace(&json, "\"supported_codecs\":\"h264\"", "\"supported_codecs\":[\"h264\"]"); + obs_data_release(post); + return json.array; +} + +static obs_data_t *fetch_config(const char *server, const char *key, struct dstr *err) +{ + struct dstr url = {0}; + dstr_copy(&url, server); + while (url.len && url.array[url.len - 1] == '/') + dstr_resize(&url, url.len - 1); + dstr_cat(&url, CONFIG_PATH); + + char *body = config_request_json(key); + struct dstr response = {0}; + long status = 0; + obs_data_t *config = NULL; + + do_log(LOG_INFO, "Requesting stream configuration from %s", url.array); + if (http_post_json(url.array, body, &status, &response, err)) { + if (status != 200) { + dstr_printf(err, "Stream configuration request to %s failed with HTTP %ld: %s", url.array, status, + response.array ? response.array : ""); + } else if (!(config = obs_data_create_from_json(response.array))) { + dstr_printf(err, "Stream configuration from %s is not valid JSON", url.array); + } else { + do_log(LOG_INFO, "Stream configuration: %s", response.array); + } + } + + if (config) { + obs_data_t *result = obs_data_get_obj(config, "status"); + const char *outcome = obs_data_get_string(result, "result"); + const char *message = obs_data_get_string(result, "html_en_us"); + if (strcmp(outcome, "error") == 0) { + dstr_copy(err, *message ? message : "The server refused to configure the stream"); + obs_data_release(config); + config = NULL; + } else if (strcmp(outcome, "warning") == 0) { + do_log(LOG_WARNING, "Stream configuration warning: %s", message); + } + obs_data_release(result); + } + + bfree(body); + dstr_free(&response); + dstr_free(&url); + return config; +} + +static bool find_ingest_url(obs_data_t *config, const char *key, struct dstr *url, struct dstr *err) +{ + obs_data_array_t *endpoints = obs_data_get_array(config, "ingest_endpoints"); + for (size_t i = 0; i < obs_data_array_count(endpoints) && dstr_is_empty(url); i++) { + obs_data_t *endpoint = obs_data_array_item(endpoints, i); + if (astrcmpi(obs_data_get_string(endpoint, "protocol"), INGEST_PROTOCOL) == 0) { + dstr_copy(url, obs_data_get_string(endpoint, "url_template")); + dstr_replace(url, "{stream_key}", key); + } + obs_data_release(endpoint); + } + obs_data_array_release(endpoints); + + if (dstr_is_empty(url)) + dstr_printf(err, "The server offered no %s ingest endpoint", INGEST_PROTOCOL); + return !dstr_is_empty(url); +} + +static bool encoder_type_available(const char *type) +{ + const char *id; + for (size_t i = 0; obs_enum_encoder_types(i, &id); i++) { + if (strcmp(id, type) == 0) + return true; + } + return false; +} + +static const char *audio_encoder_for_codec(const char *codec) +{ + const char *id; + for (size_t i = 0; obs_enum_encoder_types(i, &id); i++) { + if (obs_get_encoder_type(id) == OBS_ENCODER_AUDIO && strcmp(obs_get_encoder_codec(id), codec) == 0 && + !(obs_get_encoder_caps(id) & (OBS_ENCODER_CAP_DEPRECATED | OBS_ENCODER_CAP_INTERNAL))) + return id; + } + return NULL; +} + +struct enum_name { + const char *name; + int value; +}; + +static const struct enum_name scale_types[] = { + {"OBS_SCALE_POINT", OBS_SCALE_POINT}, {"OBS_SCALE_BICUBIC", OBS_SCALE_BICUBIC}, + {"OBS_SCALE_BILINEAR", OBS_SCALE_BILINEAR}, {"OBS_SCALE_LANCZOS", OBS_SCALE_LANCZOS}, + {"OBS_SCALE_AREA", OBS_SCALE_AREA}, {NULL, 0}, +}; + +static const struct enum_name video_formats[] = { + {"VIDEO_FORMAT_NV12", VIDEO_FORMAT_NV12}, {"VIDEO_FORMAT_I420", VIDEO_FORMAT_I420}, + {"VIDEO_FORMAT_I444", VIDEO_FORMAT_I444}, {"VIDEO_FORMAT_I010", VIDEO_FORMAT_I010}, + {"VIDEO_FORMAT_P010", VIDEO_FORMAT_P010}, {NULL, 0}, +}; + +/* Unset fields get the same defaults OBS's multitrack output uses. */ +static int lookup_enum(const struct enum_name *names, const char *name, int fallback) +{ + for (; names->name; names++) { + if (strcmp(names->name, name) == 0) + return names->value; + } + return fallback; +} + +static obs_encoder_t *create_video_encoder(obs_data_t *config, size_t idx, const struct obs_video_info *ovi, + struct dstr *err) +{ + const char *type = obs_data_get_string(config, "type"); + if (!encoder_type_available(type)) { + dstr_printf(err, "Video encoder '%s' is not available in this OBS", type); + return NULL; + } + if (strcmp(obs_get_encoder_codec(type), "h264") != 0) { + dstr_printf(err, "Video encoder '%s' is not H.264, the only video codec Streamplace accepts", type); + return NULL; + } + + obs_data_t *settings = obs_data_get_obj(config, "settings"); + if (!settings) + settings = obs_data_create(); + /* Keeps keyframes aligned across renditions. */ + obs_data_set_bool(settings, "disable_scenecut", true); + + struct dstr name = {0}; + dstr_printf(&name, "streamplace video %d", (int)idx); + obs_encoder_t *encoder = obs_video_encoder_create(type, name.array, settings, NULL); + obs_data_release(settings); + dstr_free(&name); + if (!encoder) { + dstr_printf(err, "Could not create video encoder '%s'", type); + return NULL; + } + + obs_encoder_set_video(encoder, obs_get_video()); + uint32_t width = (uint32_t)obs_data_get_int(config, "width"); + uint32_t height = (uint32_t)obs_data_get_int(config, "height"); + if (width && height) + obs_encoder_set_scaled_size(encoder, width, height); + obs_encoder_set_gpu_scale_type(encoder, lookup_enum(scale_types, obs_data_get_string(config, "gpu_scale_type"), + OBS_SCALE_BICUBIC)); + obs_encoder_set_preferred_video_format( + encoder, lookup_enum(video_formats, obs_data_get_string(config, "format"), VIDEO_FORMAT_NV12)); + + obs_data_t *framerate = obs_data_get_obj(config, "framerate"); + uint64_t num = (uint64_t)obs_data_get_int(framerate, "numerator"); + uint64_t den = (uint64_t)obs_data_get_int(framerate, "denominator"); + if (num && den) { + uint64_t divisor = ((uint64_t)ovi->fps_num * den) / (num * ovi->fps_den); + if (divisor > 1) + obs_encoder_set_frame_rate_divisor(encoder, (uint32_t)divisor); + } + obs_data_release(framerate); + + return encoder; +} + +static bool create_video_encoders(struct sp_output *sp, obs_data_array_t *configs, struct dstr *err) +{ + size_t count = obs_data_array_count(configs); + if (count > MAX_OUTPUT_VIDEO_ENCODERS) { + dstr_printf(err, "The server asked for %d video tracks; at most %d are supported", (int)count, + MAX_OUTPUT_VIDEO_ENCODERS); + return false; + } + + struct obs_video_info ovi; + obs_get_video_info(&ovi); + + /* Grouped encoders start together, so their keyframes line up. */ + obs_encoder_group_t *group = obs_encoder_group_create(); + obs_encoder_t *encoders[MAX_OUTPUT_VIDEO_ENCODERS] = {0}; + bool ok = true; + for (size_t i = 0; ok && i < count; i++) { + obs_data_t *config = obs_data_array_item(configs, i); + encoders[i] = create_video_encoder(config, i, &ovi, err); + ok = encoders[i] && obs_encoder_set_group(encoders[i], group); + obs_data_release(config); + } + + if (ok) { + for (size_t i = 0; i < MAX_OUTPUT_VIDEO_ENCODERS; i++) + obs_output_set_video_encoder2(sp->output, encoders[i], i); + obs_encoder_group_destroy(sp->encoder_group); + sp->encoder_group = group; + } else { + obs_encoder_group_destroy(group); + } + for (size_t i = 0; i < count; i++) + obs_encoder_release(encoders[i]); + return ok; +} + +/* track_id is the zero-based OBS audio mixer index (the mixer's "Track 1" is 0). */ +static bool create_audio_encoders(struct sp_output *sp, obs_data_array_t *configs, struct dstr *err) +{ + size_t count = obs_data_array_count(configs); + if (count > MAX_OUTPUT_AUDIO_ENCODERS) { + dstr_printf(err, "The server asked for %d audio tracks; at most %d are supported", (int)count, + MAX_OUTPUT_AUDIO_ENCODERS); + return false; + } + + obs_encoder_t *encoders[MAX_OUTPUT_AUDIO_ENCODERS] = {0}; + bool ok = true; + for (size_t i = 0; ok && i < count; i++) { + obs_data_t *config = obs_data_array_item(configs, i); + const char *codec = obs_data_get_string(config, "codec"); + const char *type = audio_encoder_for_codec(*codec ? codec : "aac"); + int mixer = (int)obs_data_get_int(config, "track_id"); + + if (!type) { + dstr_printf(err, "No audio encoder for '%s' is available in this OBS", codec); + ok = false; + } else if (mixer < 0 || mixer >= MAX_AUDIO_MIXES) { + dstr_printf(err, "Audio track_id %d is not an OBS mixer track (0-%d)", mixer, MAX_AUDIO_MIXES - 1); + ok = false; + } else { + obs_data_t *settings = obs_data_get_obj(config, "settings"); + struct dstr name = {0}; + dstr_printf(&name, "streamplace audio %d", (int)i); + encoders[i] = obs_audio_encoder_create(type, name.array, settings, (size_t)mixer, NULL); + if (encoders[i]) + obs_encoder_set_audio(encoders[i], obs_get_audio()); + else + dstr_printf(err, "Could not create audio encoder '%s'", type); + ok = encoders[i] != NULL; + obs_data_release(settings); + dstr_free(&name); + } + obs_data_release(config); + } + + if (ok) { + for (size_t i = 0; i < MAX_OUTPUT_AUDIO_ENCODERS; i++) + obs_output_set_audio_encoder(sp->output, encoders[i], i); + } + for (size_t i = 0; i < count; i++) + obs_encoder_release(encoders[i]); + return ok; +} + +/* Without encoder configurations, the encoders from OBS's output settings stay. */ +static bool apply_encoder_configs(struct sp_output *sp, obs_data_t *config, struct dstr *err) +{ + obs_data_array_t *video = obs_data_get_array(config, "encoder_configurations"); + obs_data_t *audio_configs = obs_data_get_obj(config, "audio_configurations"); + obs_data_array_t *audio = obs_data_get_array(audio_configs, "live"); + + bool ok = (!obs_data_array_count(video) || create_video_encoders(sp, video, err)) && + (!obs_data_array_count(audio) || create_audio_encoders(sp, audio, err)); + + obs_data_array_release(audio); + obs_data_release(audio_configs); + obs_data_array_release(video); + return ok; +} + +/* --------------------------------------------------------------------------- + * Muxing, on OBS's output thread */ + +static bool add_video_track(struct sp_output *sp, obs_encoder_t *encoder, struct dstr *err) +{ + uint8_t *extra; + size_t extra_size; + if (strcmp(obs_encoder_get_codec(encoder), "h264") != 0) { + dstr_printf(err, "Streamplace only accepts H.264 video, not %s", obs_encoder_get_codec(encoder)); + return false; + } + if (!obs_encoder_get_extra_data(encoder, &extra, &extra_size)) { + dstr_printf(err, "Video encoder '%s' has no codec headers", obs_encoder_get_name(encoder)); + return false; + } + + const struct video_output_info *voi = video_output_get_info(obs_encoder_video(encoder)); + struct fmp4_track track = { + .codec = FMP4_H264, + .timescale = voi->fps_num, + .timebase_num = voi->fps_den * obs_encoder_get_frame_rate_divisor(encoder), + .width = obs_encoder_get_width(encoder), + .height = obs_encoder_get_height(encoder), + }; + track.config_size = obs_parse_avc_header(&track.config, extra, extra_size); + fmp4_add_track(&sp->mux, &track); + bfree(track.config); + return true; +} + +static bool add_audio_track(struct sp_output *sp, obs_encoder_t *encoder, struct dstr *err) +{ + const char *codec = obs_encoder_get_codec(encoder); + uint8_t *extra = NULL; + size_t extra_size = 0; + obs_encoder_get_extra_data(encoder, &extra, &extra_size); + + struct fmp4_track track = { + .timescale = (uint32_t)audio_output_get_sample_rate(obs_encoder_audio(encoder)), + .timebase_num = 1, + .channels = (uint32_t)audio_output_get_channels(obs_encoder_audio(encoder)), + .sample_rate = obs_encoder_get_sample_rate(encoder), + .config = extra, + .config_size = extra_size, + }; + if (strcmp(codec, "aac") == 0 && extra_size) { + track.codec = FMP4_AAC; + obs_data_t *settings = obs_encoder_get_settings(encoder); + track.bitrate = (uint32_t)obs_data_get_int(settings, "bitrate") * 1000; + obs_data_release(settings); + } else if (strcmp(codec, "opus") == 0 && extra_size >= 19 && memcmp(extra, "OpusHead", 8) == 0) { + track.codec = FMP4_OPUS; + track.sample_rate = 48000; + } else { + dstr_printf(err, "Audio encoder '%s' (%s) has no usable codec headers", obs_encoder_get_name(encoder), + codec); + return false; + } + + fmp4_add_track(&sp->mux, &track); + return true; +} + +static bool start_mux(struct sp_output *sp, struct dstr *err) +{ + obs_encoder_t *encoder; + while (sp->video_tracks < MAX_OUTPUT_VIDEO_ENCODERS && + (encoder = obs_output_get_video_encoder2(sp->output, sp->video_tracks))) { + if (!add_video_track(sp, encoder, err)) + return false; + sp->video_tracks++; + } + while (sp->audio_tracks < MAX_OUTPUT_AUDIO_ENCODERS && + (encoder = obs_output_get_audio_encoder(sp->output, sp->audio_tracks))) { + if (!add_audio_track(sp, encoder, err)) + return false; + sp->audio_tracks++; + } + + do_log(LOG_INFO, "Muxing %d video and %d audio tracks", (int)sp->video_tracks, (int)sp->audio_tracks); + sp->mux_started = true; + return true; +} + +static void finish_with_error(struct sp_output *sp, int code) +{ + sp->finishing = true; + sp->error_code = code; + pthread_cond_signal(&sp->cond); +} + +static void sp_encoded_packet(void *data, struct encoder_packet *packet) +{ + struct sp_output *sp = data; + pthread_mutex_lock(&sp->mutex); + if (!sp->capturing || sp->finishing) { + pthread_mutex_unlock(&sp->mutex); + return; + } + + if (!packet) { + do_log(LOG_ERROR, "An encoder failed"); + finish_with_error(sp, OBS_OUTPUT_ENCODE_ERROR); + pthread_mutex_unlock(&sp->mutex); + return; + } + + if (!sp->mux_started) { + struct dstr err = {0}; + if (!start_mux(sp, &err)) { + do_log(LOG_ERROR, "%s", err.array); + obs_output_set_last_error(sp->output, err.array); + finish_with_error(sp, OBS_OUTPUT_UNSUPPORTED); + } + dstr_free(&err); + } + + bool video = packet->type == OBS_ENCODER_VIDEO; + size_t track_count = video ? sp->video_tracks : sp->audio_tracks; + if (sp->mux_started && packet->track_idx < track_count) { + struct encoder_packet owned; + if (video) + obs_parse_avc_packet(&owned, packet); + else + obs_encoder_packet_ref(&owned, packet); + + size_t track = video ? packet->track_idx : sp->video_tracks + packet->track_idx; + fmp4_push(&sp->mux, track, &owned); + /* A fragment per sample on the first track keeps latency to one frame. */ + if (track == 0) + fmp4_write_fragment(&sp->mux, &sp->queue_s); + + if (sp->queue.bytes.num > MAX_QUEUED_BYTES) { + do_log(LOG_ERROR, "The server is not keeping up; over %d MiB are queued", + MAX_QUEUED_BYTES / (1024 * 1024)); + finish_with_error(sp, OBS_OUTPUT_DISCONNECTED); + } + pthread_cond_signal(&sp->cond); + } + pthread_mutex_unlock(&sp->mutex); +} + +/* --------------------------------------------------------------------------- + * Threads */ + +static void *send_thread(void *data) +{ + struct sp_output *sp = data; + os_set_thread_name("streamplace-send"); + + int code = OBS_OUTPUT_SUCCESS; + bool finished = false; + while (!finished) { + pthread_mutex_lock(&sp->mutex); + while (!sp->queue.bytes.num && !sp->finishing) + pthread_cond_wait(&sp->cond, &sp->mutex); + if (sp->finishing) { + finished = true; + code = sp->error_code; + sp->capturing = false; + if (code == OBS_OUTPUT_SUCCESS && sp->mux_started) + fmp4_finish(&sp->mux, &sp->queue_s); + } + struct darray bytes = sp->queue.bytes.da; + memset(&sp->queue, 0, sizeof(sp->queue)); + pthread_mutex_unlock(&sp->mutex); + + if (bytes.num && code == OBS_OUTPUT_SUCCESS) { + if (http_write_chunk(&sp->conn, bytes.array, bytes.num)) { + pthread_mutex_lock(&sp->mutex); + sp->total_bytes += bytes.num; + pthread_mutex_unlock(&sp->mutex); + } else { + do_log(LOG_WARNING, "Lost the connection to the server"); + code = OBS_OUTPUT_DISCONNECTED; + pthread_mutex_lock(&sp->mutex); + sp->capturing = false; + pthread_mutex_unlock(&sp->mutex); + finished = true; + } + } + darray_free(&bytes); + } + + if (code == OBS_OUTPUT_SUCCESS) + do_log(LOG_INFO, "Stream ended; server responded with HTTP %ld", http_end_chunked_post(&sp->conn)); + http_close(&sp->conn); + + if (code == OBS_OUTPUT_SUCCESS || is_stopping(sp)) + obs_output_end_data_capture(sp->output); + else + obs_output_signal_stop(sp->output, code); + return NULL; +} + +static bool start_streaming(struct sp_output *sp, const char *url, struct dstr *err) +{ + if (!http_begin_chunked_post(&sp->conn, url, "video/mp4", err)) + return false; + + if (!obs_output_can_begin_data_capture(sp->output, 0) || !obs_output_initialize_encoders(sp->output, 0)) { + dstr_copy(err, "Could not initialize the encoders"); + http_close(&sp->conn); + return false; + } + + pthread_mutex_lock(&sp->mutex); + if (!sp->stopping) + sp->send_thread_active = pthread_create(&sp->send_thread, NULL, send_thread, sp) == 0; + sp->capturing = sp->send_thread_active; + pthread_mutex_unlock(&sp->mutex); + if (!sp->send_thread_active) { + http_close(&sp->conn); + return false; + } + + do_log(LOG_INFO, "Streaming to %s", url); + if (!obs_output_begin_data_capture(sp->output, 0)) { + /* The send thread owns the connection now; it ends the stream. */ + pthread_mutex_lock(&sp->mutex); + finish_with_error(sp, OBS_OUTPUT_ERROR); + pthread_mutex_unlock(&sp->mutex); + } + return true; +} + +static bool connect_and_start(struct sp_output *sp, struct dstr *err) +{ + obs_service_t *service = obs_output_get_service(sp->output); + const char *server = obs_service_get_connect_info(service, OBS_SERVICE_CONNECT_INFO_SERVER_URL); + const char *key = obs_service_get_connect_info(service, OBS_SERVICE_CONNECT_INFO_STREAM_ID); + if (!server || !*server) { + dstr_copy(err, "No Streamplace server is configured"); + return false; + } + + obs_data_t *config = fetch_config(server, key ? key : "", err); + if (!config) + return false; + + struct dstr url = {0}; + bool ok = find_ingest_url(config, key ? key : "", &url, err) && !is_stopping(sp) && + apply_encoder_configs(sp, config, err) && !is_stopping(sp) && start_streaming(sp, url.array, err); + obs_data_release(config); + dstr_free(&url); + return ok; +} + +static void *connect_thread(void *data) +{ + struct sp_output *sp = data; + os_set_thread_name("streamplace-connect"); + + struct dstr err = {0}; + if (!connect_and_start(sp, &err) && !is_stopping(sp)) { + do_log(LOG_WARNING, "Could not start streaming: %s", err.array); + obs_output_set_last_error(sp->output, err.array); + obs_output_signal_stop(sp->output, OBS_OUTPUT_CONNECT_FAILED); + } + dstr_free(&err); + return NULL; +} + +static void join_threads(struct sp_output *sp) +{ + if (sp->connect_thread_active) { + pthread_join(sp->connect_thread, NULL); + sp->connect_thread_active = false; + } + if (sp->send_thread_active) { + pthread_join(sp->send_thread, NULL); + sp->send_thread_active = false; + } +} + +/* --------------------------------------------------------------------------- + * obs_output_info */ + +static const char *sp_get_name(void *type_data) +{ + UNUSED_PARAMETER(type_data); + return "Streamplace"; +} + +static void *sp_create(obs_data_t *settings, obs_output_t *output) +{ + UNUSED_PARAMETER(settings); + struct sp_output *sp = bzalloc(sizeof(*sp)); + sp->output = output; + pthread_mutex_init(&sp->mutex, NULL); + pthread_cond_init(&sp->cond, NULL); + array_output_serializer_init(&sp->queue_s, &sp->queue); + return sp; +} + +static void sp_destroy(void *data) +{ + struct sp_output *sp = data; + join_threads(sp); + fmp4_free(&sp->mux); + array_output_serializer_free(&sp->queue); + obs_encoder_group_destroy(sp->encoder_group); + pthread_cond_destroy(&sp->cond); + pthread_mutex_destroy(&sp->mutex); + bfree(sp); +} + +static bool sp_start(void *data) +{ + struct sp_output *sp = data; + if (!obs_output_can_begin_data_capture(sp->output, 0)) + return false; + + /* Threads from a previous run have already wound down by now. */ + join_threads(sp); + fmp4_free(&sp->mux); + array_output_serializer_free(&sp->queue); + array_output_serializer_init(&sp->queue_s, &sp->queue); + sp->stopping = false; + sp->capturing = false; + sp->finishing = false; + sp->error_code = OBS_OUTPUT_SUCCESS; + sp->mux_started = false; + sp->video_tracks = 0; + sp->audio_tracks = 0; + sp->total_bytes = 0; + + sp->connect_thread_active = pthread_create(&sp->connect_thread, NULL, connect_thread, sp) == 0; + return sp->connect_thread_active; +} + +static void sp_stop(void *data, uint64_t ts) +{ + UNUSED_PARAMETER(ts); + struct sp_output *sp = data; + + pthread_mutex_lock(&sp->mutex); + sp->stopping = true; + pthread_mutex_unlock(&sp->mutex); + + if (sp->connect_thread_active) { + pthread_join(sp->connect_thread, NULL); + sp->connect_thread_active = false; + } + + pthread_mutex_lock(&sp->mutex); + if (sp->capturing && !sp->finishing) + finish_with_error(sp, OBS_OUTPUT_SUCCESS); + pthread_mutex_unlock(&sp->mutex); + + /* With no send thread, nothing else will report the stop. */ + if (!sp->send_thread_active) + obs_output_signal_stop(sp->output, OBS_OUTPUT_SUCCESS); +} + +static uint64_t sp_get_total_bytes(void *data) +{ + struct sp_output *sp = data; + pthread_mutex_lock(&sp->mutex); + uint64_t total = sp->total_bytes; + pthread_mutex_unlock(&sp->mutex); + return total; +} + +struct obs_output_info streamplace_output_info = { + .id = "streamplace_output", + .flags = OBS_OUTPUT_AV | OBS_OUTPUT_ENCODED | OBS_OUTPUT_SERVICE | OBS_OUTPUT_MULTI_TRACK_AV, + .protocols = STREAMPLACE_PROTOCOL, + .encoded_video_codecs = "h264", + .encoded_audio_codecs = "aac;opus", + .get_name = sp_get_name, + .create = sp_create, + .destroy = sp_destroy, + .start = sp_start, + .stop = sp_stop, + .encoded_packet = sp_encoded_packet, + .get_total_bytes = sp_get_total_bytes, +}; diff --git a/obs-plugin/src/streamplace.h b/obs-plugin/src/streamplace.h new file mode 100644 index 000000000..c061fde7e --- /dev/null +++ b/obs-plugin/src/streamplace.h @@ -0,0 +1,11 @@ +#pragma once + +#include + +/* The output protocol the plugin registers. The services.json entry names it, + * so OBS lists that entry only while the plugin is loaded. */ +#define STREAMPLACE_PROTOCOL "Streamplace" + +#define do_log(level, format, ...) blog(level, "[obs-streamplace] " format, ##__VA_ARGS__) + +extern struct obs_output_info streamplace_output_info; diff --git a/obs-plugin/subprojects/.gitignore b/obs-plugin/subprojects/.gitignore new file mode 100644 index 000000000..f9d20a115 --- /dev/null +++ b/obs-plugin/subprojects/.gitignore @@ -0,0 +1,5 @@ +* +!.gitignore +!obs-studio.wrap +!packagefiles/ +!packagefiles/** diff --git a/obs-plugin/subprojects/obs-studio.wrap b/obs-plugin/subprojects/obs-studio.wrap new file mode 100644 index 000000000..6e6bcfbc7 --- /dev/null +++ b/obs-plugin/subprojects/obs-studio.wrap @@ -0,0 +1,8 @@ +; The oldest OBS the plugin supports. libobs refuses plugins built against newer +; struct layouts, so build against the minimum rather than the latest. +[wrap-file] +directory = obs-studio-30.2.3 +source_url = https://github.com/obsproject/obs-studio/archive/refs/tags/30.2.3.tar.gz +source_filename = obs-studio-30.2.3.tar.gz +source_hash = 3d5f37f564564b6e6fb32ba535285e062ee25d573b3a6a405e5a86d5ab029351 +patch_directory = obs-studio diff --git a/obs-plugin/subprojects/packagefiles/obs-studio/meson.build b/obs-plugin/subprojects/packagefiles/obs-studio/meson.build new file mode 100644 index 000000000..b3762e392 --- /dev/null +++ b/obs-plugin/subprojects/packagefiles/obs-studio/meson.build @@ -0,0 +1,3 @@ +project('obs-studio', 'c', version: '30.2.3') + +libobs_dep = declare_dependency(include_directories: include_directories('libobs', is_system: true)) -- 2.54.0