lagertonne/yolobs-studio
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

New upstream version 18.0.1+dfsg1

Browse source
This commit is contained in:
Sebastian Ramacher 2017-04-19 21:54:15 +02:00
parent 6efda2859e
commit f2cf6cce50
1337 changed files with 41178 additions and 84670 deletions
Download patch file Download diff file Expand all files Collapse all files

2
plugins/obs-outputs/CMakeLists.txt
View file

@ -62,6 +62,7 @@ endif()
set(obs-outputs_HEADERS
obs-output-ver.h
rtmp-helpers.h
rtmp-stream.h
net-if.h
flv-mux.h
flv-output.h
@ -69,6 +70,7 @@ set(obs-outputs_HEADERS
set(obs-outputs_SOURCES
obs-outputs.c
rtmp-stream.c
rtmp-windows.c
flv-output.c
flv-mux.c
net-if.c)

1
plugins/obs-outputs/data/locale/da-DK.ini
View file

@ -2,4 +2,5 @@ RTMPStream="RTMP Strøm"
RTMPStream.DropThreshold="Drop Tærskel (millisekunder)"
FLVOutput="FLV File Output"
FLVOutput.FilePath="Filsti"
Default="Standard"

1
plugins/obs-outputs/data/locale/el-GR.ini
View file

@ -2,4 +2,5 @@ RTMPStream="Ροή RTMP"
RTMPStream.DropThreshold="Όριο απωλειών (χιλιοστά δευτερολέπτου)"
FLVOutput="FLV Αρχείο Εξόδου"
FLVOutput.FilePath="Διαδρομή Αρχείου"
Default="Προεπιλογή"

2
plugins/obs-outputs/data/locale/et-EE.ini Normal file
View file

@ -0,0 +1,2 @@
RTMPStream="RTMP voogedastus"

1
plugins/obs-outputs/data/locale/nb-NO.ini
View file

@ -2,4 +2,5 @@ RTMPStream="RTMP-strøm"
RTMPStream.DropThreshold="Senk terskelen (millisekunder)"
FLVOutput="FLV-filutgang"
FLVOutput.FilePath="Filbane"
Default="Standard"

1
plugins/obs-outputs/data/locale/sk-SK.ini
View file

@ -1,4 +1,5 @@
RTMPStream="RTMP stream"
FLVOutput="Výstup do súboru FLV"
FLVOutput.FilePath="Cesta k súboru"
Default="Základné"

1
plugins/obs-outputs/data/locale/tr-TR.ini
View file

@ -2,4 +2,5 @@ RTMPStream="RTMP Yayın"
RTMPStream.DropThreshold="Düşürme Eşiği (milisaniye)"
FLVOutput="FLV Dosyası Çıkışı"
FLVOutput.FilePath="Dosya Yolu"
Default="Varsayılan"

3
plugins/obs-outputs/data/locale/vi-VN.ini Normal file
View file

@ -0,0 +1,3 @@
FLVOutput.FilePath="Đường dẫn tệp"
Default="Mặc định"

9
plugins/obs-outputs/flv-output.c
View file

@ -73,11 +73,12 @@ static void flv_output_stop(void *data, uint64_t ts)
struct flv_output *stream = data;
if (stream->active) {
if (stream->file)
if (stream->file) {
write_file_info(stream->file, stream->last_packet_ts,
os_ftelli64(stream->file));
fclose(stream->file);
fclose(stream->file);
}
obs_output_end_data_capture(stream->output);
stream->active = false;
stream->sent_headers = false;
@ -100,7 +101,7 @@ static int write_packet(struct flv_output *stream,
flv_packet_mux(packet, &data, &size, is_header);
fwrite(data, 1, size, stream->file);
bfree(data);
obs_free_encoder_packet(packet);
obs_encoder_packet_release(packet);
return ret;
}
@ -200,7 +201,7 @@ static void flv_output_data(void *data, struct encoder_packet *packet)
if (packet->type == OBS_ENCODER_VIDEO) {
obs_parse_avc_packet(&parsed_packet, packet);
write_packet(stream, &parsed_packet, false);
obs_free_encoder_packet(&parsed_packet);
obs_encoder_packet_release(&parsed_packet);
} else {
write_packet(stream, packet, false);
}

21
plugins/obs-outputs/librtmp/rtmp.c
View file

@ -650,7 +650,7 @@ int RTMP_AddStream(RTMP *r, const char *playpath)
}
static int
add_addr_info(struct sockaddr_storage *service, socklen_t *addrlen, AVal *host, int port)
add_addr_info(struct sockaddr_storage *service, socklen_t *addrlen, AVal *host, int port, socklen_t addrlen_hint)
{
char *hostname;
int ret = TRUE;
@ -698,7 +698,7 @@ add_addr_info(struct sockaddr_storage *service, socklen_t *addrlen, AVal *host,
// prefer ipv4 results, since lots of ISPs have broken ipv6 connectivity
for (ptr = result; ptr != NULL; ptr = ptr->ai_next)
{
if (ptr->ai_family == AF_INET)
if (ptr->ai_family == AF_INET && (!addrlen_hint || ptr->ai_addrlen == addrlen_hint))
{
memcpy(service, ptr->ai_addr, ptr->ai_addrlen);
*addrlen = (socklen_t)ptr->ai_addrlen;
@ -710,7 +710,7 @@ add_addr_info(struct sockaddr_storage *service, socklen_t *addrlen, AVal *host,
{
for (ptr = result; ptr != NULL; ptr = ptr->ai_next)
{
if (ptr->ai_family == AF_INET6)
if (ptr->ai_family == AF_INET6 && (!addrlen_hint || ptr->ai_addrlen == addrlen_hint))
{
memcpy(service, ptr->ai_addr, ptr->ai_addrlen);
*addrlen = (socklen_t)ptr->ai_addrlen;
@ -905,6 +905,7 @@ RTMP_Connect(RTMP *r, RTMPPacket *cp)
#endif
struct sockaddr_storage service;
socklen_t addrlen = 0;
socklen_t addrlen_hint = 0;
if (!r->Link.hostname.av_len)
return FALSE;
@ -920,16 +921,19 @@ RTMP_Connect(RTMP *r, RTMPPacket *cp)
memset(&service, 0, sizeof(service));
if (r->m_bindIP.addrLen)
addrlen_hint = r->m_bindIP.addrLen;
if (r->Link.socksport)
{
/* Connect via SOCKS */
if (!add_addr_info(&service, &addrlen, &r->Link.sockshost, r->Link.socksport))
if (!add_addr_info(&service, &addrlen, &r->Link.sockshost, r->Link.socksport, addrlen_hint))
return FALSE;
}
else
{
/* Connect directly */
if (!add_addr_info(&service, &addrlen, &r->Link.hostname, r->Link.port))
if (!add_addr_info(&service, &addrlen, &r->Link.hostname, r->Link.port, addrlen_hint))
return FALSE;
}
@ -949,7 +953,7 @@ SocksNegotiate(RTMP *r)
socklen_t addrlen = 0;
memset(&service, 0, sizeof(service));
add_addr_info(&service, &addrlen, &r->Link.hostname, r->Link.port);
add_addr_info(&service, &addrlen, &r->Link.hostname, r->Link.port, 0);
// not doing IPv6 socks
if (service.ss_family == AF_INET6)
@ -4248,6 +4252,11 @@ RTMP_Close(RTMP *r)
r->Link.lFlags ^= RTMP_LF_FTCU;
}
memset (&r->m_bindIP, 0, sizeof(r->m_bindIP));
r->m_bCustomSend = 0;
r->m_customSendFunc = NULL;
r->m_customSendParam = NULL;
#if defined(CRYPTO) || defined(USE_ONLY_MD5)
if (!(r->Link.protocol & RTMP_FEATURE_WRITE) || (r->Link.pFlags & RTMP_PUB_CLEAN))
{

1
plugins/obs-outputs/net-if.h
View file

@ -29,6 +29,7 @@
# ifdef __linux__
# include <linux/if_link.h>
# elif __FreeBSD__
# include <netinet/in.h>
# ifndef _GNU_SOURCE
# define _GNU_SOURCE
# define __NET_IF_GNU_SOURCE__

346
plugins/obs-outputs/rtmp-stream.c
View file

@ -15,95 +15,7 @@
along with this program. If not, see <http://www.gnu.org/licenses/>.
******************************************************************************/
#include <obs-module.h>
#include <obs-avc.h>
#include <util/platform.h>
#include <util/circlebuf.h>
#include <util/dstr.h>
#include <util/threading.h>
#include <inttypes.h>
#include "librtmp/rtmp.h"
#include "librtmp/log.h"
#include "flv-mux.h"
#include "net-if.h"
#ifdef _WIN32
#include <Iphlpapi.h>
#else
#include <sys/ioctl.h>
#endif
#define do_log(level, format, ...) \
blog(level, "[rtmp stream: '%s'] " format, \
obs_output_get_name(stream->output), ##__VA_ARGS__)
#define warn(format, ...) do_log(LOG_WARNING, format, ##__VA_ARGS__)
#define info(format, ...) do_log(LOG_INFO, format, ##__VA_ARGS__)
#define debug(format, ...) do_log(LOG_DEBUG, format, ##__VA_ARGS__)
#define OPT_DROP_THRESHOLD "drop_threshold_ms"
#define OPT_PFRAME_DROP_THRESHOLD "pframe_drop_threshold_ms"
#define OPT_MAX_SHUTDOWN_TIME_SEC "max_shutdown_time_sec"
#define OPT_BIND_IP "bind_ip"
//#define TEST_FRAMEDROPS
#ifdef TEST_FRAMEDROPS
#define DROPTEST_MAX_KBPS 3000
#define DROPTEST_MAX_BYTES (DROPTEST_MAX_KBPS * 1000 / 8)
struct droptest_info {
uint64_t ts;
size_t size;
};
#endif
struct rtmp_stream {
obs_output_t *output;
pthread_mutex_t packets_mutex;
struct circlebuf packets;
bool sent_headers;
volatile bool connecting;
pthread_t connect_thread;
volatile bool active;
volatile bool disconnected;
pthread_t send_thread;
int max_shutdown_time_sec;
os_sem_t *send_sem;
os_event_t *stop_event;
uint64_t stop_ts;
uint64_t shutdown_timeout_ts;
struct dstr path, key;
struct dstr username, password;
struct dstr encoder_name;
struct dstr bind_ip;
/* frame drop variables */
int64_t drop_threshold_usec;
int64_t min_drop_dts_usec;
int64_t pframe_drop_threshold_usec;
int64_t pframe_min_drop_dts_usec;
int min_priority;
int64_t last_dts_usec;
uint64_t total_bytes_sent;
int dropped_frames;
#ifdef TEST_FRAMEDROPS
struct circlebuf droptest_info;
size_t droptest_size;
#endif
RTMP rtmp;
};
#include "rtmp-stream.h"
static const char *rtmp_stream_getname(void *unused)
{
@ -134,7 +46,7 @@ static inline void free_packets(struct rtmp_stream *stream)
while (stream->packets.size) {
struct encoder_packet packet;
circlebuf_pop_front(&stream->packets, &packet, sizeof(packet));
obs_free_encoder_packet(&packet);
obs_encoder_packet_release(&packet);
}
pthread_mutex_unlock(&stream->packets_mutex);
}
@ -180,23 +92,30 @@ static void rtmp_stream_destroy(void *data)
}
}
if (stream) {
free_packets(stream);
dstr_free(&stream->path);
dstr_free(&stream->key);
dstr_free(&stream->username);
dstr_free(&stream->password);
dstr_free(&stream->encoder_name);
dstr_free(&stream->bind_ip);
os_event_destroy(stream->stop_event);
os_sem_destroy(stream->send_sem);
pthread_mutex_destroy(&stream->packets_mutex);
circlebuf_free(&stream->packets);
free_packets(stream);
dstr_free(&stream->path);
dstr_free(&stream->key);
dstr_free(&stream->username);
dstr_free(&stream->password);
dstr_free(&stream->encoder_name);
dstr_free(&stream->bind_ip);
os_event_destroy(stream->stop_event);
os_sem_destroy(stream->send_sem);
pthread_mutex_destroy(&stream->packets_mutex);
circlebuf_free(&stream->packets);
#ifdef TEST_FRAMEDROPS
circlebuf_free(&stream->droptest_info);
circlebuf_free(&stream->droptest_info);
#endif
bfree(stream);
}
os_event_destroy(stream->buffer_space_available_event);
os_event_destroy(stream->buffer_has_data_event);
os_event_destroy(stream->socket_available_event);
os_event_destroy(stream->send_thread_signaled_exit);
pthread_mutex_destroy(&stream->write_buf_mutex);
if (stream->write_buf)
bfree(stream->write_buf);
bfree(stream);
}
static void *rtmp_stream_create(obs_data_t *settings, obs_output_t *output)
@ -214,6 +133,32 @@ static void *rtmp_stream_create(obs_data_t *settings, obs_output_t *output)
if (os_event_init(&stream->stop_event, OS_EVENT_TYPE_MANUAL) != 0)
goto fail;
if (pthread_mutex_init(&stream->write_buf_mutex, NULL) != 0) {
warn("Failed to initialize write buffer mutex");
goto fail;
}
if (os_event_init(&stream->buffer_space_available_event,
OS_EVENT_TYPE_AUTO) != 0) {
warn("Failed to initialize write buffer event");
goto fail;
}
if (os_event_init(&stream->buffer_has_data_event,
OS_EVENT_TYPE_AUTO) != 0) {
warn("Failed to initialize data buffer event");
goto fail;
}
if (os_event_init(&stream->socket_available_event,
OS_EVENT_TYPE_AUTO) != 0) {
warn("Failed to initialize socket buffer event");
goto fail;
}
if (os_event_init(&stream->send_thread_signaled_exit,
OS_EVENT_TYPE_MANUAL) != 0) {
warn("Failed to initialize socket exit event");
goto fail;
}
UNUSED_PARAMETER(settings);
return stream;
@ -346,6 +291,38 @@ static void droptest_cap_data_rate(struct rtmp_stream *stream, size_t size)
}
#endif
static int socket_queue_data(RTMPSockBuf *sb, const char *data, int len, void *arg)
{
struct rtmp_stream *stream = arg;
retry_send:
if (!RTMP_IsConnected(&stream->rtmp))
return 0;
pthread_mutex_lock(&stream->write_buf_mutex);
if (stream->write_buf_len + len > stream->write_buf_size) {
pthread_mutex_unlock(&stream->write_buf_mutex);
if (os_event_wait(stream->buffer_space_available_event)) {
return 0;
}
goto retry_send;
}
memcpy(stream->write_buf + stream->write_buf_len, data, len);
stream->write_buf_len += len;
pthread_mutex_unlock(&stream->write_buf_mutex);
os_event_signal (stream->buffer_has_data_event);
return len;
}
static int send_packet(struct rtmp_stream *stream,
struct encoder_packet *packet, bool is_header, size_t idx)
{
@ -354,16 +331,18 @@ static int send_packet(struct rtmp_stream *stream,
int recv_size = 0;
int ret = 0;
if (!stream->new_socket_loop) {
#ifdef _WIN32
ret = ioctlsocket(stream->rtmp.m_sb.sb_socket, FIONREAD,
(u_long*)&recv_size);
ret = ioctlsocket(stream->rtmp.m_sb.sb_socket, FIONREAD,
(u_long*)&recv_size);
#else
ret = ioctl(stream->rtmp.m_sb.sb_socket, FIONREAD, &recv_size);
ret = ioctl(stream->rtmp.m_sb.sb_socket, FIONREAD, &recv_size);
#endif
if (ret >= 0 && recv_size > 0) {
if (!discard_recv_data(stream, (size_t)recv_size))
return -1;
if (ret >= 0 && recv_size > 0) {
if (!discard_recv_data(stream, (size_t)recv_size))
return -1;
}
}
flv_packet_mux(packet, &data, &size, is_header);
@ -375,7 +354,10 @@ static int send_packet(struct rtmp_stream *stream,
ret = RTMP_Write(&stream->rtmp, (char*)data, (int)size, (int)idx);
bfree(data);
obs_free_encoder_packet(packet);
if (is_header)
bfree(packet->data);
else
obs_encoder_packet_release(packet);
stream->total_bytes_sent += size;
return ret;
@ -414,7 +396,7 @@ static void *send_thread(void *data)
if (stopping(stream)) {
if (can_shutdown_stream(stream, &packet)) {
obs_free_encoder_packet(&packet);
obs_encoder_packet_release(&packet);
break;
}
}
@ -438,6 +420,14 @@ static void *send_thread(void *data)
info("User stopped the stream");
}
if (stream->new_socket_loop) {
os_event_signal(stream->send_thread_signaled_exit);
os_event_signal(stream->buffer_has_data_event);
pthread_join(stream->socket_thread, NULL);
stream->socket_thread_active = false;
stream->rtmp.m_bCustomSend = false;
}
RTMP_Close(&stream->rtmp);
if (!stopping(stream)) {
@ -577,6 +567,78 @@ static int init_send(struct rtmp_stream *stream)
return OBS_OUTPUT_ERROR;
}
if (stream->new_socket_loop) {
int one = 1;
#ifdef _WIN32
if (ioctlsocket(stream->rtmp.m_sb.sb_socket, FIONBIO, &one)) {
#else
if (ioctl(stream->rtmp.m_sb.sb_socket, FIONBIO, &one)) {
#endif
warn("Failed to set non-blocking socket");
return OBS_OUTPUT_ERROR;
}
os_event_reset(stream->send_thread_signaled_exit);
info("New socket loop enabled by user");
if (stream->low_latency_mode)
info("Low latency mode enabled by user");
if (stream->write_buf)
bfree(stream->write_buf);
int total_bitrate = 0;
obs_output_t *context = stream->output;
obs_encoder_t *vencoder = obs_output_get_video_encoder(context);
if (vencoder) {
obs_data_t *params = obs_encoder_get_settings(vencoder);
if (params) {
int bitrate = obs_data_get_int(params, "bitrate");
total_bitrate += bitrate;
obs_data_release(params);
}
}
obs_encoder_t *aencoder = obs_output_get_audio_encoder(context, 0);
if (aencoder) {
obs_data_t *params = obs_encoder_get_settings(aencoder);
if (params) {
int bitrate = obs_data_get_int(params, "bitrate");
total_bitrate += bitrate;
obs_data_release(params);
}
}
// to bytes/sec
int ideal_buffer_size = total_bitrate * 128;
if (ideal_buffer_size < 131072)
ideal_buffer_size = 131072;
stream->write_buf_size = ideal_buffer_size;
stream->write_buf = bmalloc(ideal_buffer_size);
#ifdef _WIN32
ret = pthread_create(&stream->socket_thread, NULL,
socket_thread_windows, stream);
#else
warn("New socket loop not supported on this platform");
return OBS_OUTPUT_ERROR;
#endif
if (ret != 0) {
RTMP_Close(&stream->rtmp);
warn("Failed to create socket thread");
return OBS_OUTPUT_ERROR;
}
stream->socket_thread_active = true;
stream->rtmp.m_bCustomSend = true;
stream->rtmp.m_customSendFunc = socket_queue_data;
stream->rtmp.m_customSendParam = stream;
}
os_atomic_set_bool(&stream->active, true);
while (next) {
if (!send_meta_data(stream, idx++, &next)) {
@ -676,8 +738,11 @@ static int try_connect(struct rtmp_stream *stream)
bool success = netif_str_to_addr(&stream->rtmp.m_bindIP.addr,
&stream->rtmp.m_bindIP.addrLen,
stream->bind_ip.array);
if (success)
info("Binding to IP");
if (success) {
int len = stream->rtmp.m_bindIP.addrLen;
bool ipv6 = len == sizeof(struct sockaddr_in6);
info("Binding to IPv%d", ipv6 ? 6 : 4);
}
}
RTMP_AddStream(&stream->rtmp, stream->key.array);
@ -720,8 +785,9 @@ static bool init_connect(struct rtmp_stream *stream)
int64_t drop_p;
int64_t drop_b;
if (stopping(stream))
if (stopping(stream)) {
pthread_join(stream->send_thread, NULL);
}
free_packets(stream);
@ -756,6 +822,11 @@ static bool init_connect(struct rtmp_stream *stream)
bind_ip = obs_data_get_string(settings, OPT_BIND_IP);
dstr_copy(&stream->bind_ip, bind_ip);
stream->new_socket_loop = obs_data_get_bool(settings,
OPT_NEWSOCKETLOOP_ENABLED);
stream->low_latency_mode = obs_data_get_bool(settings,
OPT_LOWLATENCY_ENABLED);
obs_data_release(settings);
return true;
}
@ -842,7 +913,7 @@ static void drop_frames(struct rtmp_stream *stream, const char *name,
} else {
num_frames_dropped++;
obs_free_encoder_packet(&packet);
obs_encoder_packet_release(&packet);
}
}
@ -878,8 +949,11 @@ static void check_to_drop_frames(struct rtmp_stream *stream, bool pframes)
stream->pframe_drop_threshold_usec :
stream->drop_threshold_usec;
if (num_packets < 5)
if (num_packets < 5) {
if (!pframes)
stream->congestion = 0.0f;
return;
}
circlebuf_peek_front(&stream->packets, &first, sizeof(first));
@ -891,8 +965,13 @@ static void check_to_drop_frames(struct rtmp_stream *stream, bool pframes)
* sent is higher than threshold, drop frames */
buffer_duration_usec = stream->last_dts_usec - first.dts_usec;
if (!pframes) {
stream->congestion = (float)buffer_duration_usec /
(float)drop_threshold;
}
if (buffer_duration_usec > drop_threshold) {
debug("buffer_duration_usec: %lld", buffer_duration_usec);
debug("buffer_duration_usec: %" PRId64, buffer_duration_usec);
drop_frames(stream, name, priority, p_min_dts_usec);
}
}
@ -905,7 +984,7 @@ static bool add_video_packet(struct rtmp_stream *stream,
/* if currently dropping frames, drop packets until it reaches the
* desired priority */
if (packet->priority < stream->min_priority) {
if (packet->drop_priority < stream->min_priority) {
stream->dropped_frames++;
return false;
} else {
@ -927,7 +1006,7 @@ static void rtmp_stream_data(void *data, struct encoder_packet *packet)
if (packet->type == OBS_ENCODER_VIDEO)
obs_parse_avc_packet(&new_packet, packet);
else
obs_duplicate_encoder_packet(&new_packet, packet);
obs_encoder_packet_ref(&new_packet, packet);
pthread_mutex_lock(&stream->packets_mutex);
@ -942,15 +1021,17 @@ static void rtmp_stream_data(void *data, struct encoder_packet *packet)
if (added_packet)
os_sem_post(stream->send_sem);
else
obs_free_encoder_packet(&new_packet);
obs_encoder_packet_release(&new_packet);
}
static void rtmp_stream_defaults(obs_data_t *defaults)
{
obs_data_set_default_int(defaults, OPT_DROP_THRESHOLD, 500);
obs_data_set_default_int(defaults, OPT_PFRAME_DROP_THRESHOLD, 800);
obs_data_set_default_int(defaults, OPT_DROP_THRESHOLD, 700);
obs_data_set_default_int(defaults, OPT_PFRAME_DROP_THRESHOLD, 900);
obs_data_set_default_int(defaults, OPT_MAX_SHUTDOWN_TIME_SEC, 30);
obs_data_set_default_string(defaults, OPT_BIND_IP, "default");
obs_data_set_default_bool(defaults, OPT_NEWSOCKETLOOP_ENABLED, false);
obs_data_set_default_bool(defaults, OPT_LOWLATENCY_ENABLED, false);
}
static obs_properties_t *rtmp_stream_properties(void *unused)
@ -978,6 +1059,11 @@ static obs_properties_t *rtmp_stream_properties(void *unused)
}
netif_saddr_data_free(&addrs);
obs_properties_add_bool(props, OPT_NEWSOCKETLOOP_ENABLED,
obs_module_text("RTMPStream.NewSocketLoop"));
obs_properties_add_bool(props, OPT_LOWLATENCY_ENABLED,
obs_module_text("RTMPStream.LowLatencyMode"));
return props;
}
@ -993,6 +1079,17 @@ static int rtmp_stream_dropped_frames(void *data)
return stream->dropped_frames;
}
static float rtmp_stream_congestion(void *data)
{
struct rtmp_stream *stream = data;
if (stream->new_socket_loop)
return (float)stream->write_buf_len /
(float)stream->write_buf_size;
else
return stream->min_priority > 0 ? 1.0f : stream->congestion;
}
struct obs_output_info rtmp_output_info = {
.id = "rtmp_output",
.flags = OBS_OUTPUT_AV |
@ -1008,5 +1105,6 @@ struct obs_output_info rtmp_output_info = {
.get_defaults = rtmp_stream_defaults,
.get_properties = rtmp_stream_properties,
.get_total_bytes = rtmp_stream_total_bytes_sent,
.get_congestion = rtmp_stream_congestion,
.get_dropped_frames = rtmp_stream_dropped_frames
};

110
plugins/obs-outputs/rtmp-stream.h Normal file
View file

@ -0,0 +1,110 @@
#include <obs-module.h>
#include <obs-avc.h>
#include <util/platform.h>
#include <util/circlebuf.h>
#include <util/dstr.h>
#include <util/threading.h>
#include <inttypes.h>
#include "librtmp/rtmp.h"
#include "librtmp/log.h"
#include "flv-mux.h"
#include "net-if.h"
#ifdef _WIN32
#include <Iphlpapi.h>
#else
#include <sys/ioctl.h>
#endif
#define do_log(level, format, ...) \
blog(level, "[rtmp stream: '%s'] " format, \
obs_output_get_name(stream->output), ##__VA_ARGS__)
#define warn(format, ...) do_log(LOG_WARNING, format, ##__VA_ARGS__)
#define info(format, ...) do_log(LOG_INFO, format, ##__VA_ARGS__)
#define debug(format, ...) do_log(LOG_DEBUG, format, ##__VA_ARGS__)
#define OPT_DROP_THRESHOLD "drop_threshold_ms"
#define OPT_PFRAME_DROP_THRESHOLD "pframe_drop_threshold_ms"
#define OPT_MAX_SHUTDOWN_TIME_SEC "max_shutdown_time_sec"
#define OPT_BIND_IP "bind_ip"
#define OPT_NEWSOCKETLOOP_ENABLED "new_socket_loop_enabled"
#define OPT_LOWLATENCY_ENABLED "low_latency_mode_enabled"
//#define TEST_FRAMEDROPS
#ifdef TEST_FRAMEDROPS
#define DROPTEST_MAX_KBPS 3000
#define DROPTEST_MAX_BYTES (DROPTEST_MAX_KBPS * 1000 / 8)
struct droptest_info {
uint64_t ts;
size_t size;
};
#endif
struct rtmp_stream {
obs_output_t *output;
pthread_mutex_t packets_mutex;
struct circlebuf packets;
bool sent_headers;
volatile bool connecting;
pthread_t connect_thread;
volatile bool active;
volatile bool disconnected;
pthread_t send_thread;
int max_shutdown_time_sec;
os_sem_t *send_sem;
os_event_t *stop_event;
uint64_t stop_ts;
uint64_t shutdown_timeout_ts;
struct dstr path, key;
struct dstr username, password;
struct dstr encoder_name;
struct dstr bind_ip;
/* frame drop variables */
int64_t drop_threshold_usec;
int64_t min_drop_dts_usec;
int64_t pframe_drop_threshold_usec;
int64_t pframe_min_drop_dts_usec;
int min_priority;
float congestion;
int64_t last_dts_usec;
uint64_t total_bytes_sent;
int dropped_frames;
#ifdef TEST_FRAMEDROPS
struct circlebuf droptest_info;
size_t droptest_size;
#endif
RTMP rtmp;
bool new_socket_loop;
bool low_latency_mode;
bool disable_send_window_optimization;
bool socket_thread_active;
pthread_t socket_thread;
uint8_t *write_buf;
size_t write_buf_len;
size_t write_buf_size;
pthread_mutex_t write_buf_mutex;
os_event_t *buffer_space_available_event;
os_event_t *buffer_has_data_event;
os_event_t *socket_available_event;
os_event_t *send_thread_signaled_exit;
};
#ifdef _WIN32
void *socket_thread_windows(void *data);
#endif

351
plugins/obs-outputs/rtmp-windows.c Normal file
View file

@ -0,0 +1,351 @@
#ifdef _WIN32
#include "rtmp-stream.h"
#include <winsock2.h>
static void fatal_sock_shutdown(struct rtmp_stream *stream)
{
closesocket(stream->rtmp.m_sb.sb_socket);
stream->rtmp.m_sb.sb_socket = -1;
stream->write_buf_len = 0;
os_event_signal(stream->buffer_space_available_event);
}
static bool socket_event(struct rtmp_stream *stream, bool *can_write,
uint64_t last_send_time)
{
WSANETWORKEVENTS net_events;
bool success;
success = !WSAEnumNetworkEvents(stream->rtmp.m_sb.sb_socket, NULL,
&net_events);
if (!success) {
blog(LOG_ERROR, "socket_thread_windows: Aborting due to "
"WSAEnumNetworkEvents failure, %d",
WSAGetLastError());
fatal_sock_shutdown(stream);
return false;
}
if (net_events.lNetworkEvents & FD_WRITE)
*can_write = true;
if (net_events.lNetworkEvents & FD_CLOSE) {
if (last_send_time) {
uint32_t diff =
(os_gettime_ns() / 1000000) - last_send_time;
blog(LOG_ERROR, "socket_thread_windows: Received "
"FD_CLOSE, %u ms since last send "
"(buffer: %d / %d)",
diff,
stream->write_buf_len,
stream->write_buf_size);
}
if (os_event_try(stream->stop_event) != EAGAIN)
blog(LOG_ERROR, "socket_thread_windows: Aborting due "
"to FD_CLOSE during shutdown, "
"%d bytes lost, error %d",
stream->write_buf_len,
net_events.iErrorCode[FD_CLOSE_BIT]);
else
blog(LOG_ERROR, "socket_thread_windows: Aborting due "
"to FD_CLOSE, error %d",
net_events.iErrorCode[FD_CLOSE_BIT]);
fatal_sock_shutdown(stream);
return false;
}
if (net_events.lNetworkEvents & FD_READ) {
char discard[16384];
int err_code;
bool fatal = false;
for (;;) {
int ret = recv(stream->rtmp.m_sb.sb_socket,
discard, sizeof(discard), 0);
if (ret == -1) {
err_code = WSAGetLastError();
if (err_code == WSAEWOULDBLOCK)
break;
fatal = true;
} else if (ret == 0) {
err_code = 0;
fatal = true;
}
if (fatal) {
blog(LOG_ERROR, "socket_thread_windows: "
"Socket error, recv() returned "
"%d, GetLastError() %d",
ret, err_code);
fatal_sock_shutdown(stream);
return false;
}
}
}
return true;
}
static void ideal_send_backlog_event(struct rtmp_stream *stream,
bool *can_write)
{
ULONG ideal_send_backlog;
int ret;
ret = idealsendbacklogquery(
stream->rtmp.m_sb.sb_socket,
&ideal_send_backlog);
if (ret == 0) {
int cur_tcp_bufsize;
int size = sizeof(cur_tcp_bufsize);
ret = getsockopt(stream->rtmp.m_sb.sb_socket,
SOL_SOCKET,
SO_SNDBUF,
(char *)&cur_tcp_bufsize,
&size);
if (ret == 0) {
if (cur_tcp_bufsize < (int)ideal_send_backlog) {
int bufsize = (int)ideal_send_backlog;
setsockopt(stream->rtmp.m_sb.sb_socket,
SOL_SOCKET,
SO_SNDBUF,
(const char *)&bufsize,
sizeof(bufsize));
blog(LOG_INFO, "socket_thread_windows: "
"Increasing send buffer to "
"ISB %d (buffer: %d / %d)",
ideal_send_backlog,
stream->write_buf_len,
stream->write_buf_size);
}
} else {
blog(LOG_ERROR, "socket_thread_windows: Got "
"send_backlog_event but "
"getsockopt() returned %d",
WSAGetLastError());
}
} else {
blog(LOG_ERROR, "socket_thread_windows: Got "
"send_backlog_event but WSAIoctl() "
"returned %d",
WSAGetLastError());
}
}
enum data_ret {
RET_BREAK,
RET_FATAL,
RET_CONTINUE
};
static enum data_ret write_data(struct rtmp_stream *stream, bool *can_write,
uint64_t *last_send_time, size_t latency_packet_size,
int delay_time)
{
bool exit_loop = false;
pthread_mutex_lock(&stream->write_buf_mutex);
if (!stream->write_buf_len) {
/* this is now an expected occasional condition due to use of
* auto-reset events, we could end up emptying the buffer as
* it's filled in a previous loop cycle, especially if using
* low latency mode. */
pthread_mutex_unlock(&stream->write_buf_mutex);
/* blog(LOG_DEBUG, "socket_thread_windows: Trying to send, "
"but no data available"); */
return RET_BREAK;
}
int ret;
if (stream->low_latency_mode) {
size_t send_len =
min(latency_packet_size, stream->write_buf_len);
ret = send(stream->rtmp.m_sb.sb_socket,
(const char *)stream->write_buf,
(int)send_len, 0);
} else {
ret = send(stream->rtmp.m_sb.sb_socket,
(const char *)stream->write_buf,
(int)stream->write_buf_len, 0);
}
if (ret > 0) {
if (stream->write_buf_len - ret)
memmove(stream->write_buf,
stream->write_buf + ret,
stream->write_buf_len - ret);
stream->write_buf_len -= ret;
*last_send_time = os_gettime_ns() / 1000000;
os_event_signal(stream->buffer_space_available_event);
} else {
int err_code;
bool fatal_err = false;
if (ret == -1) {
err_code = WSAGetLastError();
if (err_code == WSAEWOULDBLOCK) {
*can_write = false;
pthread_mutex_unlock(&stream->write_buf_mutex);
return RET_BREAK;
}
fatal_err = true;
} else if (ret == 0) {
err_code = 0;
fatal_err = true;
}
if (fatal_err) {
/* connection closed, or connection was aborted /
* socket closed / etc, that's a fatal error. */
blog(LOG_ERROR, "socket_thread_windows: "
"Socket error, send() returned %d, "
"GetLastError() %d",
ret, err_code);
pthread_mutex_unlock(&stream->write_buf_mutex);
fatal_sock_shutdown(stream);
return RET_FATAL;
}
}
/* finish writing for now */
if (stream->write_buf_len <= 1000)
exit_loop = true;
pthread_mutex_unlock(&stream->write_buf_mutex);
if (delay_time)
os_sleep_ms(delay_time);
return exit_loop ? RET_BREAK : RET_CONTINUE;
}
#define LATENCY_FACTOR 20
static inline void socket_thread_windows_internal(struct rtmp_stream *stream)
{
bool can_write = false;
int delay_time;
size_t latency_packet_size;
uint64_t last_send_time = 0;
HANDLE send_backlog_event;
OVERLAPPED send_backlog_overlapped;
SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_ABOVE_NORMAL);
WSAEventSelect(stream->rtmp.m_sb.sb_socket,
stream->socket_available_event,
FD_READ|FD_WRITE|FD_CLOSE);
send_backlog_event = CreateEvent(NULL, true, false, NULL);
if (stream->low_latency_mode) {
delay_time = 1000 / LATENCY_FACTOR;
latency_packet_size = stream->write_buf_size / (LATENCY_FACTOR - 2);
} else {
latency_packet_size = stream->write_buf_size;
delay_time = 0;
}
if (!stream->disable_send_window_optimization) {
memset(&send_backlog_overlapped, 0,
sizeof(send_backlog_overlapped));
send_backlog_overlapped.hEvent = send_backlog_event;
idealsendbacklognotify(stream->rtmp.m_sb.sb_socket,
&send_backlog_overlapped, NULL);
} else {
blog(LOG_INFO, "socket_thread_windows: Send window "
"optimization disabled by user.");
}
HANDLE objs[3];
objs[0] = stream->socket_available_event;
objs[1] = stream->buffer_has_data_event;
objs[2] = send_backlog_event;
for (;;) {
if (os_event_try(stream->send_thread_signaled_exit) != EAGAIN) {
pthread_mutex_lock(&stream->write_buf_mutex);
if (stream->write_buf_len == 0) {
//blog(LOG_DEBUG, "Exiting on empty buffer");
pthread_mutex_unlock(&stream->write_buf_mutex);
os_event_reset(stream->send_thread_signaled_exit);
break;
}
pthread_mutex_unlock(&stream->write_buf_mutex);
}
int status = WaitForMultipleObjects(3, objs, false, INFINITE);
if (status == WAIT_ABANDONED || status == WAIT_FAILED) {
blog(LOG_ERROR, "socket_thread_windows: Aborting due "
"to WaitForMultipleObjects failure");
fatal_sock_shutdown(stream);
return;
}
if (status == WAIT_OBJECT_0) {
/* Socket event */
if (!socket_event(stream, &can_write, last_send_time))
return;
} else if (status == WAIT_OBJECT_0 + 2) {
/* Ideal send backlog event */
ideal_send_backlog_event(stream, &can_write);
ResetEvent(send_backlog_event);
idealsendbacklognotify(stream->rtmp.m_sb.sb_socket,
&send_backlog_overlapped, NULL);
continue;
}
if (can_write) {
for (;;) {
enum data_ret ret = write_data(
stream,
&can_write,
&last_send_time,
latency_packet_size,
delay_time);
switch (ret) {
case RET_BREAK:
goto exit_write_loop;
case RET_FATAL:
return;
case RET_CONTINUE:;
}
}
}
exit_write_loop:;
}
if (stream->rtmp.m_sb.sb_socket != INVALID_SOCKET)
WSAEventSelect(stream->rtmp.m_sb.sb_socket,
stream->socket_available_event, 0);
blog(LOG_INFO, "socket_thread_windows: Normal exit");
}
void *socket_thread_windows(void *data)
{
struct rtmp_stream *stream = data;
socket_thread_windows_internal(stream);
return NULL;
}
#endif