diff --git a/NEWS.md b/NEWS.md index e732d7a..4a82532 100644 --- a/NEWS.md +++ b/NEWS.md @@ -100,6 +100,7 @@ Version 4.0.0 (Oct 19, 2021): is outside the band of the channel filter (thx @charlie-foxtrot). * New output type `udp_stream` for sending uncompressed audio to another host via UDP/IP (thx @charlie-foxtrot). +* New output type `srt` for streaming uncompressed audio over the SRT protocol. * Added `multiple_output_threads` global option. When set to `true`, a separate output thread is spawned for each device (thx @charlie-foxtrot). * Modulation in scan mode is now configurable per channel (thx diff --git a/SRT.md b/SRT.md new file mode 100644 index 0000000..a876acb --- /dev/null +++ b/SRT.md @@ -0,0 +1,68 @@ +# SRT output + +RTLSDR-Airband can send audio over the [SRT protocol](https://www.srtalliance.org/). +Building with SRT support requires the development files for **libsrt**. +When configuring with CMake leave the `-DSRT` option enabled (default) +and ensure `pkg-config` can locate the library. If libsrt is missing the +feature is disabled automatically. + +The SRT output supports three audio formats controlled by the `format` +setting in the configuration: + +- `pcm` (default) – raw 16‑bit signed PCM (standard format) +- `mp3` – encoded using libmp3lame +- `wav` – 16‑bit PCM with WAV header so players like VLC auto-detect the format + +For low latency playback with ffplay: + +```bash +# For mp3 or wav formats (auto-detected): +ffplay -fflags nobuffer -flags low_delay srt://: + +# For pcm format (match the configured sample_rate, default 8kHz mono): +ffplay -fflags nobuffer -flags low_delay -f s16le -ar 8000 -ac 1 srt://: +``` + +## Configuration + +``` +outputs: ( + { + type = "srt"; + listen_address = "0.0.0.0"; + listen_port = 8890; + format = "mp3"; # pcm|mp3|wav + mode = "live"; # live|raw (default: live) + sample_rate = 24000; # optional, default: native (8000 or 16000) + continuous = true; # optional, default false + } +); +``` + +`continuous` controls whether the stream pauses when the squelch is +closed. Set it to `true` if the receiving application does not handle +frequent reconnects well. + +## SRT Mode + +The `mode` setting controls SRT protocol behavior: + +- `live` (default) – Standard SRT live mode with TSBPD (Timestamp-Based + Packet Delivery), packet drop, and NAK reports enabled. Compatible with + all SRT clients including gosrt, OBS, and other strict implementations. + Adds approximately 120ms latency. + +- `raw` – Minimal latency mode with TSBPD disabled. Only works with lenient + clients like ffplay/ffmpeg. Use this if you need the absolute lowest + latency and only use ffplay for playback. + +## Sample Rate + +The `sample_rate` setting resamples audio output to the specified rate +using linear interpolation. This applies to `pcm` and `wav` formats only +(`mp3` uses its own encoding rate). + +The default is the native rate (`8000` Hz, or `16000` Hz with NFM). +Set to `24000` for use with the OpenAI Realtime API which requires +24 kHz mono 16-bit PCM. + diff --git a/config/srt_example.conf b/config/srt_example.conf new file mode 100644 index 0000000..fb622a2 --- /dev/null +++ b/config/srt_example.conf @@ -0,0 +1,33 @@ +# Example configuration demonstrating SRT output +# Listens on 0.0.0.0:8890 for clients using the SRT protocol +# Clients can connect with: +# ffplay -ac 1 -ar 8000 -analyzeduration 0 -probesize 32 -f f32le srt://:8890 +# Refer to https://github.com/rtl-airband/RTLSDR-Airband/wiki for config syntax + +devices: +( + { + type = "rtlsdr"; + index = 0; + gain = 25; + centerfreq = 120.0; + correction = 80; + channels: + ( + { + freq = 118.15; + outputs: ( + { + type = "srt"; + listen_address = "0.0.0.0"; + listen_port = 8890; + # format can be "pcm" (default), "mp3" or "wav" + format = "mp3"; + # stream continuously even when squelched + continuous = true; + } + ); + } + ); + } +); diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 16a0da8..fe996af 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -127,6 +127,9 @@ set(WITH_SOAPYSDR FALSE) option(PULSEAUDIO "Enable PulseAudio support" ON) set(WITH_PULSEAUDIO FALSE) +option(SRT "Enable SRT output support" ON) +set(WITH_SRT FALSE) + option(PROFILING "Enable profiling with gperftools") set(WITH_PROFILING FALSE) @@ -177,6 +180,17 @@ if(PULSEAUDIO) endif() endif() +if(SRT) + pkg_check_modules(SRT srt) + if(SRT_FOUND) + list(APPEND rtl_airband_extra_sources srt_stream.cpp) + list(APPEND rtl_airband_extra_libs ${SRT_LIBRARIES}) + list(APPEND rtl_airband_include_dirs ${SRT_INCLUDE_DIRS}) + list(APPEND link_dirs ${SRT_LIBRARY_DIRS}) + set(WITH_SRT TRUE) + endif() +endif() + if(PROFILING) pkg_check_modules(PROFILING libprofiler) if(PROFILING_FOUND) @@ -261,6 +275,7 @@ message(STATUS " - Build Unit Tests:\t${BUILD_UNITTESTS}") message(STATUS " - Broadcom VideoCore GPU:\t${WITH_BCM_VC}") message(STATUS " - NFM support:\t\t${NFM}") message(STATUS " - PulseAudio:\t\trequested: ${PULSEAUDIO}, enabled: ${WITH_PULSEAUDIO}") +message(STATUS " - SRT output:\t\trequested: ${SRT}, enabled: ${WITH_SRT}") message(STATUS " - Profiling:\t\trequested: ${PROFILING}, enabled: ${WITH_PROFILING}") message(STATUS " - Icecast TLS support:\t${LIBSHOUT_HAS_TLS}") diff --git a/src/config.cpp b/src/config.cpp index 7acdf6f..23f8633 100644 --- a/src/config.cpp +++ b/src/config.cpp @@ -121,6 +121,17 @@ static int parse_outputs(libconfig::Setting& outs, channel_t* channel, int i, in fdata->append = (!outs[o].exists("append")) || (bool)(outs[o]["append"]); fdata->split_on_transmission = outs[o].exists("split_on_transmission") ? (bool)(outs[o]["split_on_transmission"]) : false; fdata->include_freq = outs[o].exists("include_freq") ? (bool)(outs[o]["include_freq"]) : false; + if (fdata->split_on_transmission) { + fdata->min_rx_seconds = outs[o].exists("min_rx_seconds") ? (double)(outs[o]["min_rx_seconds"]) : 0.0; + if (outs[o].exists("post_write_script")) { + fdata->post_write_script = outs[o]["post_write_script"].c_str(); + } + } else { + if (outs[o].exists("min_rx_seconds") || outs[o].exists("post_write_script")) { + cerr << "Configuration error: devices.[" << i << "] channels.[" << j << "] outputs.[" << o << "]: min_rx_seconds and post_write_script require split_on_transmission\n"; + error(); + } + } channel->outputs[oo].has_mp3_output = true; @@ -159,6 +170,8 @@ static int parse_outputs(libconfig::Setting& outs, channel_t* channel, int i, in fdata->append = (!outs[o].exists("append")) || (bool)(outs[o]["append"]); fdata->split_on_transmission = outs[o].exists("split_on_transmission") ? (bool)(outs[o]["split_on_transmission"]) : false; fdata->include_freq = outs[o].exists("include_freq") ? (bool)(outs[o]["include_freq"]) : false; + fdata->min_rx_seconds = 0.0; + fdata->post_write_script.clear(); channel->needs_raw_iq = channel->has_iq_outputs = 1; if (fdata->continuous && fdata->split_on_transmission) { @@ -229,6 +242,90 @@ static int parse_outputs(libconfig::Setting& outs, channel_t* channel, int i, in cerr << "missing dest_port\n"; error(); } + } else if (!strncmp(outs[o]["type"], "srt", 3)) { + channel->outputs[oo].data = XCALLOC(1, sizeof(struct srt_stream_data)); + channel->outputs[oo].type = O_SRT; + + srt_stream_data* sdata = (srt_stream_data*)channel->outputs[oo].data; + + sdata->continuous = outs[o].exists("continuous") ? (bool)(outs[o]["continuous"]) : false; + + if (outs[o].exists("listen_address")) { + sdata->listen_address = strdup(outs[o]["listen_address"]); + } else { + if (parsing_mixers) { + cerr << "Configuration error: mixers.[" << i << "] outputs.[" << o << "]: "; + } else { + cerr << "Configuration error: devices.[" << i << "] channels.[" << j << "] outputs.[" << o << "]: "; + } + cerr << "missing listen_address\n"; + error(); + } + + if (outs[o].exists("listen_port")) { + if (outs[o]["listen_port"].getType() == libconfig::Setting::TypeInt) { + char buffer[12]; + sprintf(buffer, "%d", (int)outs[o]["listen_port"]); + sdata->listen_port = strdup(buffer); + } else { + sdata->listen_port = strdup(outs[o]["listen_port"]); + } + } else { + if (parsing_mixers) { + cerr << "Configuration error: mixers.[" << i << "] outputs.[" << o << "]: "; + } else { + cerr << "Configuration error: devices.[" << i << "] channels.[" << j << "] outputs.[" << o << "]: "; + } + cerr << "missing listen_port\n"; + error(); + } + + if (outs[o].exists("format")) { + const char* fmt = outs[o]["format"]; + if (!strcmp(fmt, "mp3")) { + sdata->format = SRT_STREAM_MP3; + channel->outputs[oo].has_mp3_output = true; + } else if (!strcmp(fmt, "raw") || !strcmp(fmt, "pcm")) { + sdata->format = SRT_STREAM_PCM; + } else if (!strcmp(fmt, "wav")) { + sdata->format = SRT_STREAM_WAV; + } else { + if (parsing_mixers) { + cerr << "Configuration error: mixers.[" << i << "] outputs.[" << o << "]: "; + } else { + cerr << "Configuration error: devices.[" << i << "] channels.[" << j << "] outputs.[" << o << "]: "; + } + cerr << "invalid SRT format, must be 'pcm', 'mp3' or 'wav'\n"; + error(); + } + } else { + sdata->format = SRT_STREAM_PCM; + } + + if (outs[o].exists("sample_rate")) { + sdata->sample_rate = (int)outs[o]["sample_rate"]; + } else { + sdata->sample_rate = WAVE_RATE; + } + + if (outs[o].exists("mode")) { + const char* m = outs[o]["mode"]; + if (!strcmp(m, "live")) { + sdata->srt_mode = SRT_MODE_LIVE; + } else if (!strcmp(m, "raw")) { + sdata->srt_mode = SRT_MODE_RAW; + } else { + if (parsing_mixers) { + cerr << "Configuration error: mixers.[" << i << "] outputs.[" << o << "]: "; + } else { + cerr << "Configuration error: devices.[" << i << "] channels.[" << j << "] outputs.[" << o << "]: "; + } + cerr << "invalid SRT mode, must be 'live' or 'raw'\n"; + error(); + } + } else { + sdata->srt_mode = SRT_MODE_LIVE; + } #ifdef WITH_PULSEAUDIO } else if (!strncmp(outs[o]["type"], "pulse", 5)) { channel->outputs[oo].data = XCALLOC(1, sizeof(struct pulse_data)); diff --git a/src/config.h.in b/src/config.h.in index 6f441f0..95e9cd3 100644 --- a/src/config.h.in +++ b/src/config.h.in @@ -25,6 +25,7 @@ #cmakedefine WITH_SOAPYSDR #cmakedefine WITH_PROFILING #cmakedefine WITH_PULSEAUDIO +#cmakedefine WITH_SRT #cmakedefine NFM #cmakedefine WITH_BCM_VC #cmakedefine LIBSHOUT_HAS_TLS diff --git a/src/output.cpp b/src/output.cpp index 3f7e3a7..38ebff0 100644 --- a/src/output.cpp +++ b/src/output.cpp @@ -24,6 +24,7 @@ #include #include #include +#include #include #include @@ -157,6 +158,8 @@ lame_t airlame_init(mix_modes mixmode, int highpass, int lowpass) { lame_set_quality(lame, 7); lame_set_lowpassfreq(lame, lowpass); lame_set_highpassfreq(lame, highpass); + /* Disable the bit reservoir to reduce encoder latency */ + lame_set_disable_reservoir(lame, 1); lame_set_out_samplerate(lame, MP3_RATE); if (mixmode == MM_STEREO) { lame_set_num_channels(lame, 2); @@ -317,6 +320,8 @@ static void close_file(output_t* output) { return; } + double duration_sec = delta_sec(&fdata->open_time, &fdata->last_write_time); + // close all mp3 files for every output that has a lame context if (fdata->type == O_FILE && fdata->f && output->lame) { int encoded = lame_encode_flush_nogap(output->lame, output->lamebuf, LAMEBUF_SIZE); @@ -337,7 +342,31 @@ static void close_file(output_t* output) { if (fdata->f) { fclose(fdata->f); fdata->f = NULL; - rename_if_exists(fdata->file_path_tmp.c_str(), fdata->file_path.c_str()); + bool keep = true; + if (fdata->split_on_transmission && fdata->min_rx_seconds > 0.0 && duration_sec < fdata->min_rx_seconds) { + keep = false; + } + + if (keep) { + rename_if_exists(fdata->file_path_tmp.c_str(), fdata->file_path.c_str()); + if (fdata->split_on_transmission && !fdata->post_write_script.empty()) { + pid_t pid = fork(); + if (pid < 0) { + log(LOG_ERR, "Cannot fork for post_write_script: %s\n", strerror(errno)); + } else if (pid == 0) { + pid_t pid2 = fork(); + if (pid2 == 0) { + execl("/bin/sh", "sh", fdata->post_write_script.c_str(), fdata->file_path.c_str(), (char*)NULL); + _exit(1); + } + _exit(0); + } else { + waitpid(pid, NULL, 0); + } + } + } else { + unlink(fdata->file_path_tmp.c_str()); + } } fdata->file_path.clear(); fdata->file_path_tmp.clear(); @@ -568,11 +597,30 @@ void process_outputs(channel_t* channel, int cur_scan_freq) { if (sdata->continuous == false && channel->axcindicate == NO_SIGNAL) { continue; } + } else if (channel->outputs[k].type == O_SRT) { + srt_stream_data* sdata = (srt_stream_data*)channel->outputs[k].data; - if (channel->mode == MM_MONO) { - udp_stream_write(sdata, channel->waveout, (size_t)WAVE_BATCH * sizeof(float)); + if (sdata->continuous == false && channel->axcindicate == NO_SIGNAL) + continue; + + if (sdata->format == SRT_STREAM_MP3) { + const auto& lame = channel->outputs[k].lame; + const auto& lamebuf = channel->outputs[k].lamebuf; + int mp3_bytes = lame_encode_buffer_ieee_float( + lame, channel->waveout, + (channel->mode == MM_STEREO ? channel->waveout_r : NULL), + WAVE_BATCH, lamebuf, LAMEBUF_SIZE); + if (mp3_bytes < 0) { + log(LOG_WARNING, "lame_encode_buffer_ieee_float: %d\n", mp3_bytes); + } else if (mp3_bytes > 0) { + srt_stream_send_bytes(sdata, lamebuf, mp3_bytes); + } } else { - udp_stream_write(sdata, channel->waveout, channel->waveout_r, (size_t)WAVE_BATCH * sizeof(float)); + if (channel->mode == MM_MONO) { + srt_stream_write(sdata, channel->waveout, (size_t)WAVE_BATCH * sizeof(float)); + } else { + srt_stream_write(sdata, channel->waveout, channel->waveout_r, (size_t)WAVE_BATCH * sizeof(float)); + } } #ifdef WITH_PULSEAUDIO @@ -607,6 +655,9 @@ void disable_channel_outputs(channel_t* channel) { } else if (output->type == O_UDP_STREAM) { udp_stream_data* sdata = (udp_stream_data*)output->data; udp_stream_shutdown(sdata); + } else if (output->type == O_SRT) { + srt_stream_data* sdata = (srt_stream_data*)output->data; + srt_stream_shutdown(sdata); #ifdef WITH_PULSEAUDIO } else if (output->type == O_PULSE) { pulse_data* pdata = (pulse_data*)(output->data); @@ -989,6 +1040,12 @@ void* output_check_thread(void*) { if (dev->input->state == INPUT_FAILED) { udp_stream_shutdown(sdata); } + } else if (dev->channels[j].outputs[k].type == O_SRT) { + srt_stream_data* sdata = (srt_stream_data*)dev->channels[j].outputs[k].data; + + if (dev->input->state == INPUT_FAILED) { + srt_stream_shutdown(sdata); + } #ifdef WITH_PULSEAUDIO } else if (dev->channels[j].outputs[k].type == O_PULSE) { pulse_data* pdata = (pulse_data*)(dev->channels[j].outputs[k].data); diff --git a/src/rtl_airband.cpp b/src/rtl_airband.cpp index b9334aa..053088f 100644 --- a/src/rtl_airband.cpp +++ b/src/rtl_airband.cpp @@ -277,6 +277,11 @@ bool init_output(channel_t* channel, output_t* output) { if (!udp_stream_init(sdata, channel->mode, (size_t)WAVE_BATCH * sizeof(float))) { return false; } + } else if (output->type == O_SRT) { + srt_stream_data* sdata = (srt_stream_data*)(output->data); + if (!srt_stream_init(sdata, channel->mode, (size_t)WAVE_BATCH * sizeof(float))) { + return false; + } #ifdef WITH_PULSEAUDIO } else if (output->type == O_PULSE) { pulse_init(); diff --git a/src/rtl_airband.h b/src/rtl_airband.h index 4563e34..b4a861b 100644 --- a/src/rtl_airband.h +++ b/src/rtl_airband.h @@ -30,6 +30,7 @@ #include #include #include +#include #include "config.h" @@ -43,6 +44,9 @@ #include #include #endif /* WITH_PULSEAUDIO */ +#ifdef WITH_SRT +#include +#endif /* WITH_SRT */ #include "filters.h" #include "input-common.h" // input_t @@ -106,7 +110,8 @@ enum output_type { O_FILE, O_RAWFILE, O_MIXER, - O_UDP_STREAM + O_UDP_STREAM, + O_SRT #ifdef WITH_PULSEAUDIO , O_PULSE @@ -140,6 +145,8 @@ struct file_data { bool append; bool split_on_transmission; bool include_freq; + double min_rx_seconds; + std::string post_write_script; timeval open_time; timeval last_write_time; FILE* f; @@ -159,6 +166,47 @@ struct udp_stream_data { socklen_t dest_sockaddr_len; }; +enum srt_stream_format { + SRT_STREAM_PCM, + SRT_STREAM_MP3, + SRT_STREAM_WAV +}; + +enum srt_stream_mode { + SRT_MODE_LIVE, // Standard SRT with TSBPD, compatible with all clients + SRT_MODE_RAW // Minimal latency, TSBPD disabled (ffplay only) +}; + +struct srt_client { + SRTSOCKET sock; + bool header_sent; +}; + +struct srt_stream_data { + float* stereo_buffer; + size_t stereo_buffer_len; + + int16_t* pcm_buffer; + size_t pcm_buffer_len; + + int16_t* resample_buffer; + size_t resample_buffer_len; + + int payload_size; + int sample_rate; + + srt_stream_format format; + srt_stream_mode srt_mode; + mix_modes mode; + + bool continuous; + const char* listen_address; + const char* listen_port; + + SRTSOCKET listen_socket; + std::vector clients; +}; + #ifdef WITH_PULSEAUDIO struct pulse_data { const char* server; @@ -394,6 +442,13 @@ void udp_stream_write(udp_stream_data* sdata, const float* data, size_t len); void udp_stream_write(udp_stream_data* sdata, const float* data_left, const float* data_right, size_t len); void udp_stream_shutdown(udp_stream_data* sdata); +// srt_stream.cpp +bool srt_stream_init(srt_stream_data* sdata, mix_modes mode, size_t len); +void srt_stream_write(srt_stream_data* sdata, const float* data, size_t len); +void srt_stream_write(srt_stream_data* sdata, const float* data_left, const float* data_right, size_t len); +void srt_stream_send_bytes(srt_stream_data* sdata, const unsigned char* data, size_t len); +void srt_stream_shutdown(srt_stream_data* sdata); + #ifdef WITH_PULSEAUDIO #define PULSE_STREAM_LATENCY_LIMIT 10000000UL // pulse.cpp diff --git a/src/srt_stream.cpp b/src/srt_stream.cpp new file mode 100644 index 0000000..8442b20 --- /dev/null +++ b/src/srt_stream.cpp @@ -0,0 +1,335 @@ +#include +#include +#include +#include +#include +#include +#include +#include + +#include "rtl_airband.h" + +#ifdef WITH_SRT + +static bool srt_initialized = false; +static void srt_stream_accept(srt_stream_data* sdata); +static void srt_stream_send_header(srt_stream_data* sdata, srt_client& client); + +static size_t resample_linear(const int16_t* in, size_t in_count, + int16_t* out, int in_rate, int out_rate) { + size_t out_count = (size_t)((uint64_t)in_count * out_rate / in_rate); + for (size_t i = 0; i < out_count; i++) { + double pos = (double)i * in_rate / out_rate; + size_t idx = (size_t)pos; + double frac = pos - idx; + if (idx + 1 < in_count) { + out[i] = (int16_t)(in[idx] * (1.0 - frac) + in[idx + 1] * frac); + } else { + out[i] = in[idx]; + } + } + return out_count; +} + +static void srt_log_dummy(void*, int, const char*, int, const char*, const char*) {} + +static void srt_try_startup() { + if (!srt_initialized) { + if (srt_startup() != 0) { + log(LOG_ERR, "srt_stream: srt_startup failed: %s\n", srt_getlasterror_str()); + } else { + srt_initialized = true; + /* Reduce noise from libsrt by disabling log output */ + srt_setloglevel(LOG_CRIT); + srt_setloghandler(NULL, srt_log_dummy); + } + } +} + +static void srt_stream_send(srt_stream_data* sdata, const char* data, size_t len) { + if (sdata->listen_socket == SRT_INVALID_SOCK) + return; + + srt_stream_accept(sdata); + for (auto it = sdata->clients.begin(); it != sdata->clients.end();) { + if (sdata->format == SRT_STREAM_WAV && !it->header_sent) { + srt_stream_send_header(sdata, *it); + it->header_sent = true; + } + + const char* ptr = data; + size_t remaining = len; + while (remaining > 0) { + int chunk = remaining > (size_t)sdata->payload_size ? sdata->payload_size : remaining; + int ret = srt_send(it->sock, ptr, chunk); + if (ret == SRT_ERROR) { + int serr; + srt_getlasterror(&serr); + if (serr == SRT_EASYNCSND) { + break; /* Buffer full, drop remaining data for this client */ + } + srt_close(it->sock); + it = sdata->clients.erase(it); + goto next_client; + } + ptr += chunk; + remaining -= chunk; + } + ++it; + continue; + next_client:; + } +} + +bool srt_stream_init(srt_stream_data* sdata, mix_modes mode, size_t len) { + srt_try_startup(); + if (!srt_initialized) { + return false; + } + + sdata->mode = mode; + + if (sdata->format != SRT_STREAM_MP3 && mode == MM_STEREO) { + sdata->stereo_buffer_len = (len / sizeof(float)) * 2; + sdata->stereo_buffer = (float*)XCALLOC(sdata->stereo_buffer_len, sizeof(float)); + } else { + sdata->stereo_buffer_len = 0; + sdata->stereo_buffer = NULL; + } + + if (sdata->format == SRT_STREAM_WAV || sdata->format == SRT_STREAM_PCM) { + sdata->pcm_buffer_len = (len / sizeof(float)) * (mode == MM_STEREO ? 2 : 1); + sdata->pcm_buffer = (int16_t*)XCALLOC(sdata->pcm_buffer_len, sizeof(int16_t)); + } else { + sdata->pcm_buffer_len = 0; + sdata->pcm_buffer = NULL; + } + + if (sdata->sample_rate != WAVE_RATE && + (sdata->format == SRT_STREAM_WAV || sdata->format == SRT_STREAM_PCM)) { + sdata->resample_buffer_len = sdata->pcm_buffer_len * + sdata->sample_rate / WAVE_RATE + 1; + sdata->resample_buffer = (int16_t*)XCALLOC(sdata->resample_buffer_len, sizeof(int16_t)); + } else { + sdata->resample_buffer_len = 0; + sdata->resample_buffer = NULL; + } + + int len_tmp; + int blocking = 0; + + sdata->listen_socket = srt_create_socket(); + if (sdata->listen_socket == SRT_INVALID_SOCK) { + log(LOG_ERR, "srt_stream: socket failed: %s\n", srt_getlasterror_str()); + goto fail; + } + + len_tmp = sizeof(sdata->payload_size); + if (srt_getsockopt(sdata->listen_socket, 0, SRTO_PAYLOADSIZE, &sdata->payload_size, &len_tmp) == SRT_ERROR) { + sdata->payload_size = SRT_LIVE_DEF_PLSIZE; + } + srt_setsockopt(sdata->listen_socket, 0, SRTO_SNDSYN, &blocking, sizeof(blocking)); + srt_setsockopt(sdata->listen_socket, 0, SRTO_RCVSYN, &blocking, sizeof(blocking)); + + if (sdata->srt_mode == SRT_MODE_LIVE) { + /* Standard SRT live mode - compatible with all SRT clients */ + int yes = 1; + srt_setsockopt(sdata->listen_socket, 0, SRTO_TSBPDMODE, &yes, sizeof(yes)); + srt_setsockopt(sdata->listen_socket, 0, SRTO_TLPKTDROP, &yes, sizeof(yes)); + srt_setsockopt(sdata->listen_socket, 0, SRTO_NAKREPORT, &yes, sizeof(yes)); + int latency = 120; /* 120ms default latency for live mode */ + srt_setsockopt(sdata->listen_socket, 0, SRTO_LATENCY, &latency, sizeof(latency)); + } else { + /* Raw mode - minimal latency, TSBPD disabled (ffplay only) */ + int tsbpd = 0; + srt_setsockopt(sdata->listen_socket, 0, SRTO_TSBPDMODE, &tsbpd, sizeof(tsbpd)); + int zero = 0; + srt_setsockopt(sdata->listen_socket, 0, SRTO_LATENCY, &zero, sizeof(zero)); + } + + struct sockaddr_in addr; + memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_port = htons(atoi(sdata->listen_port)); + if (inet_aton(sdata->listen_address, &addr.sin_addr) == 0) { + log(LOG_ERR, "srt_stream: invalid listen address %s\n", sdata->listen_address); + goto fail; + } + + if (srt_bind(sdata->listen_socket, (struct sockaddr*)&addr, sizeof(addr)) == SRT_ERROR) { + log(LOG_ERR, "srt_stream: bind failed: %s\n", srt_getlasterror_str()); + goto fail; + } + if (srt_listen(sdata->listen_socket, 5) == SRT_ERROR) { + log(LOG_ERR, "srt_stream: listen failed: %s\n", srt_getlasterror_str()); + goto fail; + } + + sdata->clients.clear(); + log(LOG_INFO, "srt_stream: listening on %s:%s\n", sdata->listen_address, sdata->listen_port); + return true; + +fail: + free(sdata->stereo_buffer); + sdata->stereo_buffer = NULL; + free(sdata->pcm_buffer); + sdata->pcm_buffer = NULL; + free(sdata->resample_buffer); + sdata->resample_buffer = NULL; + if (sdata->listen_socket != SRT_INVALID_SOCK) { + srt_close(sdata->listen_socket); + sdata->listen_socket = SRT_INVALID_SOCK; + } + return false; +} + +static void srt_stream_send_header(srt_stream_data* sdata, srt_client& client) { + if (sdata->format != SRT_STREAM_WAV) + return; + + const int channels = (sdata->mode == MM_STEREO) ? 2 : 1; + const int sample_rate = sdata->sample_rate; + const int bits_per_sample = 16; + uint32_t byte_rate = sample_rate * channels * bits_per_sample / 8; + uint16_t block_align = channels * bits_per_sample / 8; + + char header[44]; + memcpy(header, "RIFF", 4); + /* use 0xFFFFFFFF for unknown/streaming length per WAV spec */ + uint32_t sz = 0xFFFFFFFFu; + memcpy(header + 4, &sz, 4); + memcpy(header + 8, "WAVEfmt ", 8); + uint32_t fmt_size = 16; + memcpy(header + 16, &fmt_size, 4); + uint16_t audio_format = 1; /* PCM */ + memcpy(header + 20, &audio_format, 2); + uint16_t ch = channels; + memcpy(header + 22, &ch, 2); + uint32_t sr = sample_rate; + memcpy(header + 24, &sr, 4); + memcpy(header + 28, &byte_rate, 4); + memcpy(header + 32, &block_align, 2); + uint16_t bps = bits_per_sample; + memcpy(header + 34, &bps, 2); + memcpy(header + 36, "data", 4); + memcpy(header + 40, &sz, 4); + + srt_send(client.sock, header, sizeof(header)); +} + +static void srt_stream_accept(srt_stream_data* sdata) { + while (true) { + struct sockaddr_storage rem; + int len = sizeof(rem); + SRTSOCKET s = srt_accept(sdata->listen_socket, (struct sockaddr*)&rem, &len); + if (s == SRT_INVALID_SOCK) { + int serr; + srt_getlasterror(&serr); + if (serr == SRT_EASYNCRCV) + break; + break; + } + int blocking = 0; + srt_setsockopt(s, 0, SRTO_SNDSYN, &blocking, sizeof(blocking)); + srt_setsockopt(s, 0, SRTO_RCVSYN, &blocking, sizeof(blocking)); + if (sdata->srt_mode == SRT_MODE_LIVE) { + int yes = 1; + srt_setsockopt(s, 0, SRTO_TSBPDMODE, &yes, sizeof(yes)); + srt_setsockopt(s, 0, SRTO_TLPKTDROP, &yes, sizeof(yes)); + srt_setsockopt(s, 0, SRTO_NAKREPORT, &yes, sizeof(yes)); + int latency = 120; + srt_setsockopt(s, 0, SRTO_LATENCY, &latency, sizeof(latency)); + } else { + int tsbpd = 0; + srt_setsockopt(s, 0, SRTO_TSBPDMODE, &tsbpd, sizeof(tsbpd)); + int zero = 0; + srt_setsockopt(s, 0, SRTO_LATENCY, &zero, sizeof(zero)); + } + srt_client c{s, false}; + sdata->clients.push_back(c); + } +} + +void srt_stream_write(srt_stream_data* sdata, const float* data, size_t len) { + if (sdata->format == SRT_STREAM_WAV || sdata->format == SRT_STREAM_PCM) { + size_t sample_count = len / sizeof(float); + if (sample_count > sdata->pcm_buffer_len) + return; + for (size_t i = 0; i < sample_count; ++i) { + float v = data[i]; + if (v > 1.0f) + v = 1.0f; + else if (v < -1.0f) + v = -1.0f; + sdata->pcm_buffer[i] = (int16_t)(v * 32767.0f); + } + if (sdata->resample_buffer) { + size_t out_count = resample_linear(sdata->pcm_buffer, sample_count, + sdata->resample_buffer, + WAVE_RATE, sdata->sample_rate); + srt_stream_send(sdata, (const char*)sdata->resample_buffer, out_count * sizeof(int16_t)); + } else { + srt_stream_send(sdata, (const char*)sdata->pcm_buffer, sample_count * sizeof(int16_t)); + } + } else { + srt_stream_send(sdata, (const char*)data, len); + } +} + +void srt_stream_send_bytes(srt_stream_data* sdata, const unsigned char* data, size_t len) { + srt_stream_send(sdata, (const char*)data, len); +} + +void srt_stream_write(srt_stream_data* sdata, const float* left, const float* right, size_t len) { + if (!sdata->stereo_buffer) + return; + size_t sample_count = len / sizeof(float); + if (sample_count * 2 > sdata->stereo_buffer_len) + return; + for (size_t i = 0; i < sample_count; ++i) { + sdata->stereo_buffer[2 * i] = left[i]; + sdata->stereo_buffer[2 * i + 1] = right[i]; + } + if (sdata->format == SRT_STREAM_WAV || sdata->format == SRT_STREAM_PCM) { + size_t total = sample_count * 2; + if (total > sdata->pcm_buffer_len) + return; + for (size_t i = 0; i < total; ++i) { + float v = sdata->stereo_buffer[i]; + if (v > 1.0f) + v = 1.0f; + else if (v < -1.0f) + v = -1.0f; + sdata->pcm_buffer[i] = (int16_t)(v * 32767.0f); + } + if (sdata->resample_buffer) { + size_t out_count = resample_linear(sdata->pcm_buffer, total, + sdata->resample_buffer, + WAVE_RATE, sdata->sample_rate); + srt_stream_send(sdata, (const char*)sdata->resample_buffer, out_count * sizeof(int16_t)); + } else { + srt_stream_send(sdata, (const char*)sdata->pcm_buffer, total * sizeof(int16_t)); + } + } else { + srt_stream_send(sdata, (const char*)sdata->stereo_buffer, sample_count * 2 * sizeof(float)); + } +} + +void srt_stream_shutdown(srt_stream_data* sdata) { + for (auto& c : sdata->clients) { + srt_close(c.sock); + } + sdata->clients.clear(); + free(sdata->stereo_buffer); + sdata->stereo_buffer = NULL; + free(sdata->pcm_buffer); + sdata->pcm_buffer = NULL; + free(sdata->resample_buffer); + sdata->resample_buffer = NULL; + if (sdata->listen_socket != SRT_INVALID_SOCK) { + srt_close(sdata->listen_socket); + sdata->listen_socket = SRT_INVALID_SOCK; + } +} + +#endif // WITH_SRT diff --git a/src/udp_stream.cpp b/src/udp_stream.cpp index 106762f..9708a55 100644 --- a/src/udp_stream.cpp +++ b/src/udp_stream.cpp @@ -27,10 +27,12 @@ #include "rtl_airband.h" bool udp_stream_init(udp_stream_data* sdata, mix_modes mode, size_t len) { - // pre-allocate the stereo buffer + // `len` is provided in bytes. For stereo streams allocate a buffer + // large enough to hold two channels worth of samples. if (mode == MM_STEREO) { - sdata->stereo_buffer_len = len * 2; - sdata->stereo_buffer = (float*)XCALLOC(sdata->stereo_buffer_len, sizeof(float)); + sdata->stereo_buffer_len = (len / sizeof(float)) * 2; + sdata->stereo_buffer = + (float*)XCALLOC(sdata->stereo_buffer_len, sizeof(float)); } else { sdata->stereo_buffer_len = 0; sdata->stereo_buffer = NULL; @@ -90,14 +92,19 @@ void udp_stream_write(udp_stream_data* sdata, const float* data, size_t len) { } } -void udp_stream_write(udp_stream_data* sdata, const float* data_left, const float* data_right, size_t len) { +void udp_stream_write(udp_stream_data* sdata, + const float* data_left, + const float* data_right, + size_t len) { if (sdata->send_socket != -1) { - assert(len * 2 <= sdata->stereo_buffer_len); - for (size_t i = 0; i < len; ++i) { + size_t sample_count = len / sizeof(float); + assert(sample_count * 2 <= sdata->stereo_buffer_len); + for (size_t i = 0; i < sample_count; ++i) { sdata->stereo_buffer[2 * i] = data_left[i]; sdata->stereo_buffer[2 * i + 1] = data_right[i]; } - udp_stream_write(sdata, sdata->stereo_buffer, len * 2); + udp_stream_write(sdata, sdata->stereo_buffer, + sample_count * 2 * sizeof(float)); } }