/****************************************************************************** Copyright (C) 2014 by Hugh Bailey This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 2 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this program. If not, see . ******************************************************************************/ #include #include #include #include #include #include #include #include "librtmp/rtmp.h" #include "librtmp/log.h" #include "flv-mux.h" #ifdef _WIN32 #include #else #include #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_MAX_SHUTDOWN_TIME_SEC "max_shutdown_time_sec" //#define TEST_FRAMEDROPS 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; struct dstr path, key; struct dstr username, password; struct dstr encoder_name; /* frame drop variables */ int64_t drop_threshold_usec; int64_t min_drop_dts_usec; int min_priority; int64_t last_dts_usec; uint64_t total_bytes_sent; int dropped_frames; RTMP rtmp; }; static const char *rtmp_stream_getname(void *unused) { UNUSED_PARAMETER(unused); return obs_module_text("RTMPStream"); } static void log_rtmp(int level, const char *format, va_list args) { if (level > RTMP_LOGWARNING) return; blogva(LOG_INFO, format, args); } static inline size_t num_buffered_packets(struct rtmp_stream *stream); static inline void free_packets(struct rtmp_stream *stream) { size_t num_packets; pthread_mutex_lock(&stream->packets_mutex); num_packets = num_buffered_packets(stream); if (num_packets) info("Freeing %d remaining packets", (int)num_packets); while (stream->packets.size) { struct encoder_packet packet; circlebuf_pop_front(&stream->packets, &packet, sizeof(packet)); obs_free_encoder_packet(&packet); } pthread_mutex_unlock(&stream->packets_mutex); } static inline bool stopping(struct rtmp_stream *stream) { return os_event_try(stream->stop_event) != EAGAIN; } static inline bool connecting(struct rtmp_stream *stream) { return os_atomic_load_bool(&stream->connecting); } static inline bool active(struct rtmp_stream *stream) { return os_atomic_load_bool(&stream->active); } static inline bool disconnected(struct rtmp_stream *stream) { return os_atomic_load_bool(&stream->disconnected); } static void rtmp_stream_destroy(void *data) { struct rtmp_stream *stream = data; if (stopping(stream) && !connecting(stream)) { pthread_join(stream->send_thread, NULL); } else if (connecting(stream) || active(stream)) { if (stream->connecting) pthread_join(stream->connect_thread, NULL); os_event_signal(stream->stop_event); if (active(stream)) { os_sem_post(stream->send_sem); obs_output_end_data_capture(stream->output); pthread_join(stream->send_thread, NULL); } } 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); os_event_destroy(stream->stop_event); os_sem_destroy(stream->send_sem); pthread_mutex_destroy(&stream->packets_mutex); circlebuf_free(&stream->packets); bfree(stream); } } static void *rtmp_stream_create(obs_data_t *settings, obs_output_t *output) { struct rtmp_stream *stream = bzalloc(sizeof(struct rtmp_stream)); stream->output = output; pthread_mutex_init_value(&stream->packets_mutex); RTMP_Init(&stream->rtmp); RTMP_LogSetCallback(log_rtmp); RTMP_LogSetLevel(RTMP_LOGWARNING); if (pthread_mutex_init(&stream->packets_mutex, NULL) != 0) goto fail; if (os_event_init(&stream->stop_event, OS_EVENT_TYPE_MANUAL) != 0) goto fail; UNUSED_PARAMETER(settings); return stream; fail: rtmp_stream_destroy(stream); return NULL; } static void rtmp_stream_stop(void *data) { struct rtmp_stream *stream = data; if (stopping(stream)) return; if (connecting(stream)) pthread_join(stream->connect_thread, NULL); os_event_signal(stream->stop_event); if (active(stream)) { os_sem_post(stream->send_sem); obs_output_end_data_capture(stream->output); } } static inline void set_rtmp_str(AVal *val, const char *str) { bool valid = (str && *str); val->av_val = valid ? (char*)str : NULL; val->av_len = valid ? (int)strlen(str) : 0; } static inline void set_rtmp_dstr(AVal *val, struct dstr *str) { bool valid = !dstr_is_empty(str); val->av_val = valid ? str->array : NULL; val->av_len = valid ? (int)str->len : 0; } static inline bool get_next_packet(struct rtmp_stream *stream, struct encoder_packet *packet) { bool new_packet = false; pthread_mutex_lock(&stream->packets_mutex); if (stream->packets.size) { circlebuf_pop_front(&stream->packets, packet, sizeof(struct encoder_packet)); new_packet = true; } pthread_mutex_unlock(&stream->packets_mutex); return new_packet; } static bool discard_recv_data(struct rtmp_stream *stream, size_t size) { RTMP *rtmp = &stream->rtmp; uint8_t buf[512]; #ifdef _WIN32 int ret; #else ssize_t ret; #endif do { size_t bytes = size > 512 ? 512 : size; size -= bytes; #ifdef _WIN32 ret = recv(rtmp->m_sb.sb_socket, buf, (int)bytes, 0); #else ret = recv(rtmp->m_sb.sb_socket, buf, bytes, 0); #endif if (ret <= 0) { #ifdef _WIN32 int error = WSAGetLastError(); #else int error = errno; #endif if (ret < 0) { do_log(LOG_ERROR, "recv error: %d (%d bytes)", error, (int)size); } return false; } } while (size > 0); return true; } static int send_packet(struct rtmp_stream *stream, struct encoder_packet *packet, bool is_header, size_t idx) { uint8_t *data; size_t size; int recv_size = 0; int ret = 0; #ifdef _WIN32 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); #endif 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); #ifdef TEST_FRAMEDROPS os_sleep_ms(rand() % 40); #endif ret = RTMP_Write(&stream->rtmp, (char*)data, (int)size, (int)idx); bfree(data); obs_free_encoder_packet(packet); stream->total_bytes_sent += size; return ret; } static inline void send_headers(struct rtmp_stream *stream); static bool send_remaining_packets(struct rtmp_stream *stream) { struct encoder_packet packet; uint64_t max_ns = (uint64_t)stream->max_shutdown_time_sec * 1000000000; uint64_t begin_time_ns = os_gettime_ns(); if (!stream->sent_headers) send_headers(stream); while (get_next_packet(stream, &packet)) { if (send_packet(stream, &packet, false, packet.track_idx) < 0) return false; /* Just disconnect if it takes too long to shut down */ if ((os_gettime_ns() - begin_time_ns) > max_ns) { info("Took longer than %d second(s) to shut down, " "automatically stopping connection", stream->max_shutdown_time_sec); return false; } } return true; } static void *send_thread(void *data) { struct rtmp_stream *stream = data; os_set_thread_name("rtmp-stream: send_thread"); while (os_sem_wait(stream->send_sem) == 0) { struct encoder_packet packet; if (stopping(stream)) break; if (!get_next_packet(stream, &packet)) continue; if (!stream->sent_headers) send_headers(stream); if (send_packet(stream, &packet, false, packet.track_idx) < 0) { os_atomic_set_bool(&stream->disconnected, true); break; } } if (!disconnected(stream) && !send_remaining_packets(stream)) os_atomic_set_bool(&stream->disconnected, true); if (disconnected(stream)) { info("Disconnected from %s", stream->path.array); free_packets(stream); } else { info("User stopped the stream"); } RTMP_Close(&stream->rtmp); if (!stopping(stream)) { pthread_detach(stream->send_thread); obs_output_signal_stop(stream->output, OBS_OUTPUT_DISCONNECTED); } os_event_reset(stream->stop_event); os_atomic_set_bool(&stream->active, false); stream->sent_headers = false; return NULL; } static bool send_meta_data(struct rtmp_stream *stream, size_t idx) { uint8_t *meta_data; size_t meta_data_size; bool success = flv_meta_data(stream->output, &meta_data, &meta_data_size, false, idx); if (success) { RTMP_Write(&stream->rtmp, (char*)meta_data, (int)meta_data_size, (int)idx); bfree(meta_data); } return success; } static bool send_audio_header(struct rtmp_stream *stream, size_t idx) { obs_output_t *context = stream->output; obs_encoder_t *aencoder = obs_output_get_audio_encoder(context, idx); uint8_t *header; struct encoder_packet packet = { .type = OBS_ENCODER_AUDIO, .timebase_den = 1 }; if (!aencoder) return false; obs_encoder_get_extra_data(aencoder, &header, &packet.size); packet.data = bmemdup(header, packet.size); send_packet(stream, &packet, true, idx); return true; } static void send_video_header(struct rtmp_stream *stream) { obs_output_t *context = stream->output; obs_encoder_t *vencoder = obs_output_get_video_encoder(context); uint8_t *header; size_t size; struct encoder_packet packet = { .type = OBS_ENCODER_VIDEO, .timebase_den = 1, .keyframe = true }; obs_encoder_get_extra_data(vencoder, &header, &size); packet.size = obs_parse_avc_header(&packet.data, header, size); send_packet(stream, &packet, true, 0); } static inline void send_headers(struct rtmp_stream *stream) { stream->sent_headers = true; size_t i = 0; send_audio_header(stream, i++); send_video_header(stream); while (send_audio_header(stream, i++)); } static inline bool reset_semaphore(struct rtmp_stream *stream) { os_sem_destroy(stream->send_sem); return os_sem_init(&stream->send_sem, 0) == 0; } #ifdef _WIN32 #define socklen_t int #endif #define MIN_SENDBUF_SIZE 65535 static void adjust_sndbuf_size(struct rtmp_stream *stream, int new_size) { int cur_sendbuf_size = new_size; socklen_t int_size = sizeof(int); getsockopt(stream->rtmp.m_sb.sb_socket, SOL_SOCKET, SO_SNDBUF, (char*)&cur_sendbuf_size, &int_size); if (cur_sendbuf_size < new_size) { cur_sendbuf_size = new_size; setsockopt(stream->rtmp.m_sb.sb_socket, SOL_SOCKET, SO_SNDBUF, (const char*)&cur_sendbuf_size, int_size); } } static int init_send(struct rtmp_stream *stream) { int ret; size_t idx = 0; #if defined(_WIN32) adjust_sndbuf_size(stream, MIN_SENDBUF_SIZE); #endif reset_semaphore(stream); ret = pthread_create(&stream->send_thread, NULL, send_thread, stream); if (ret != 0) { RTMP_Close(&stream->rtmp); warn("Failed to create send thread"); return OBS_OUTPUT_ERROR; } os_atomic_set_bool(&stream->active, true); while (send_meta_data(stream, idx++)); obs_output_begin_data_capture(stream->output, 0); return OBS_OUTPUT_SUCCESS; } #ifdef _WIN32 static void win32_log_interface_type(struct rtmp_stream *stream) { RTMP *rtmp = &stream->rtmp; MIB_IPFORWARDROW route; uint32_t dest_addr, source_addr; char hostname[256]; HOSTENT *h; if (rtmp->Link.hostname.av_len >= sizeof(hostname) - 1) return; strncpy(hostname, rtmp->Link.hostname.av_val, sizeof(hostname)); hostname[rtmp->Link.hostname.av_len] = 0; h = gethostbyname(hostname); if (!h) return; dest_addr = *(uint32_t*)h->h_addr_list[0]; if (rtmp->m_bindIP.addrLen == 0) source_addr = 0; else if (rtmp->m_bindIP.addr.ss_family = AF_INET) source_addr = (*(struct sockaddr_in*)&rtmp->m_bindIP) .sin_addr.S_un.S_addr; else return; if (!GetBestRoute(dest_addr, source_addr, &route)) { MIB_IFROW row; memset(&row, 0, sizeof(row)); row.dwIndex = route.dwForwardIfIndex; if (!GetIfEntry(&row)) { uint32_t speed =row.dwSpeed / 1000000; char *type; struct dstr other = {0}; if (row.dwType == IF_TYPE_ETHERNET_CSMACD) { type = "ethernet"; } else if (row.dwType == IF_TYPE_IEEE80211) { type = "802.11"; } else { dstr_printf(&other, "type %lu", row.dwType); type = other.array; } info("Interface: %s (%s, %lu mbps)", row.bDescr, type, speed); dstr_free(&other); } } } #endif static int try_connect(struct rtmp_stream *stream) { if (dstr_is_empty(&stream->path)) { warn("URL is empty"); return OBS_OUTPUT_BAD_PATH; } info("Connecting to RTMP URL %s...", stream->path.array); memset(&stream->rtmp.Link, 0, sizeof(stream->rtmp.Link)); if (!RTMP_SetupURL(&stream->rtmp, stream->path.array)) return OBS_OUTPUT_BAD_PATH; RTMP_EnableWrite(&stream->rtmp); dstr_copy(&stream->encoder_name, "FMLE/3.0 (compatible; obs-studio/"); #ifdef HAVE_OBSCONFIG_H dstr_cat(&stream->encoder_name, OBS_VERSION); #else dstr_catf(&stream->encoder_name, "%d.%d.%d", LIBOBS_API_MAJOR_VER, LIBOBS_API_MINOR_VER, LIBOBS_API_PATCH_VER); #endif dstr_cat(&stream->encoder_name, "; FMSc/1.0)"); set_rtmp_dstr(&stream->rtmp.Link.pubUser, &stream->username); set_rtmp_dstr(&stream->rtmp.Link.pubPasswd, &stream->password); set_rtmp_dstr(&stream->rtmp.Link.flashVer, &stream->encoder_name); stream->rtmp.Link.swfUrl = stream->rtmp.Link.tcUrl; RTMP_AddStream(&stream->rtmp, stream->key.array); for (size_t idx = 1;; idx++) { obs_encoder_t *encoder = obs_output_get_audio_encoder( stream->output, idx); const char *encoder_name; if (!encoder) break; encoder_name = obs_encoder_get_name(encoder); RTMP_AddStream(&stream->rtmp, encoder_name); } stream->rtmp.m_outChunkSize = 4096; stream->rtmp.m_bSendChunkSizeInfo = true; stream->rtmp.m_bUseNagle = true; #ifdef _WIN32 win32_log_interface_type(stream); #endif if (!RTMP_Connect(&stream->rtmp, NULL)) return OBS_OUTPUT_CONNECT_FAILED; if (!RTMP_ConnectStream(&stream->rtmp, 0)) return OBS_OUTPUT_INVALID_STREAM; info("Connection to %s successful", stream->path.array); return init_send(stream); } static bool init_connect(struct rtmp_stream *stream) { obs_service_t *service; obs_data_t *settings; if (stopping(stream)) pthread_join(stream->send_thread, NULL); free_packets(stream); service = obs_output_get_service(stream->output); if (!service) return false; os_atomic_set_bool(&stream->disconnected, false); stream->total_bytes_sent = 0; stream->dropped_frames = 0; stream->min_drop_dts_usec= 0; stream->min_priority = 0; settings = obs_output_get_settings(stream->output); dstr_copy(&stream->path, obs_service_get_url(service)); dstr_copy(&stream->key, obs_service_get_key(service)); dstr_copy(&stream->username, obs_service_get_username(service)); dstr_copy(&stream->password, obs_service_get_password(service)); stream->drop_threshold_usec = (int64_t)obs_data_get_int(settings, OPT_DROP_THRESHOLD) * 1000; stream->max_shutdown_time_sec = (int)obs_data_get_int(settings, OPT_MAX_SHUTDOWN_TIME_SEC); obs_data_release(settings); return true; } static void *connect_thread(void *data) { struct rtmp_stream *stream = data; int ret; os_set_thread_name("rtmp-stream: connect_thread"); if (!init_connect(stream)) { obs_output_signal_stop(stream->output, OBS_OUTPUT_BAD_PATH); return NULL; } ret = try_connect(stream); if (ret != OBS_OUTPUT_SUCCESS) { obs_output_signal_stop(stream->output, ret); info("Connection to %s failed: %d", stream->path.array, ret); } if (!stopping(stream)) pthread_detach(stream->connect_thread); os_atomic_set_bool(&stream->connecting, false); return NULL; } static bool rtmp_stream_start(void *data) { struct rtmp_stream *stream = data; if (!obs_output_can_begin_data_capture(stream->output, 0)) return false; if (!obs_output_initialize_encoders(stream->output, 0)) return false; os_atomic_set_bool(&stream->connecting, true); return pthread_create(&stream->connect_thread, NULL, connect_thread, stream) == 0; } static inline bool add_packet(struct rtmp_stream *stream, struct encoder_packet *packet) { circlebuf_push_back(&stream->packets, packet, sizeof(struct encoder_packet)); stream->last_dts_usec = packet->dts_usec; return true; } static inline size_t num_buffered_packets(struct rtmp_stream *stream) { return stream->packets.size / sizeof(struct encoder_packet); } static void drop_frames(struct rtmp_stream *stream) { struct circlebuf new_buf = {0}; int drop_priority = 0; uint64_t last_drop_dts_usec = 0; int num_frames_dropped = 0; debug("Previous packet count: %d", (int)num_buffered_packets(stream)); circlebuf_reserve(&new_buf, sizeof(struct encoder_packet) * 8); while (stream->packets.size) { struct encoder_packet packet; circlebuf_pop_front(&stream->packets, &packet, sizeof(packet)); last_drop_dts_usec = packet.dts_usec; /* do not drop audio data or video keyframes */ if (packet.type == OBS_ENCODER_AUDIO || packet.drop_priority == OBS_NAL_PRIORITY_HIGHEST) { circlebuf_push_back(&new_buf, &packet, sizeof(packet)); } else { if (drop_priority < packet.drop_priority) drop_priority = packet.drop_priority; num_frames_dropped++; obs_free_encoder_packet(&packet); } } circlebuf_free(&stream->packets); stream->packets = new_buf; stream->min_priority = drop_priority; stream->min_drop_dts_usec = last_drop_dts_usec; stream->dropped_frames += num_frames_dropped; debug("New packet count: %d", (int)num_buffered_packets(stream)); } static void check_to_drop_frames(struct rtmp_stream *stream) { struct encoder_packet first; int64_t buffer_duration_usec; if (num_buffered_packets(stream) < 5) return; circlebuf_peek_front(&stream->packets, &first, sizeof(first)); /* do not drop frames if frames were just dropped within this time */ if (first.dts_usec < stream->min_drop_dts_usec) return; /* if the amount of time stored in the buffered packets waiting to be * sent is higher than threshold, drop frames */ buffer_duration_usec = stream->last_dts_usec - first.dts_usec; if (buffer_duration_usec > stream->drop_threshold_usec) { drop_frames(stream); debug("dropping %" PRId64 " worth of frames", buffer_duration_usec); } } static bool add_video_packet(struct rtmp_stream *stream, struct encoder_packet *packet) { check_to_drop_frames(stream); /* if currently dropping frames, drop packets until it reaches the * desired priority */ if (packet->priority < stream->min_priority) { stream->dropped_frames++; return false; } else { stream->min_priority = 0; } return add_packet(stream, packet); } static void rtmp_stream_data(void *data, struct encoder_packet *packet) { struct rtmp_stream *stream = data; struct encoder_packet new_packet; bool added_packet = false; if (disconnected(stream)) return; if (packet->type == OBS_ENCODER_VIDEO) obs_parse_avc_packet(&new_packet, packet); else obs_duplicate_encoder_packet(&new_packet, packet); pthread_mutex_lock(&stream->packets_mutex); if (!disconnected(stream)) { added_packet = (packet->type == OBS_ENCODER_VIDEO) ? add_video_packet(stream, &new_packet) : add_packet(stream, &new_packet); } pthread_mutex_unlock(&stream->packets_mutex); if (added_packet) os_sem_post(stream->send_sem); else obs_free_encoder_packet(&new_packet); } static void rtmp_stream_defaults(obs_data_t *defaults) { obs_data_set_default_int(defaults, OPT_DROP_THRESHOLD, 600); obs_data_set_default_int(defaults, OPT_MAX_SHUTDOWN_TIME_SEC, 5); } static obs_properties_t *rtmp_stream_properties(void *unused) { UNUSED_PARAMETER(unused); obs_properties_t *props = obs_properties_create(); obs_properties_add_int(props, OPT_DROP_THRESHOLD, obs_module_text("RTMPStream.DropThreshold"), 200, 10000, 100); return props; } static uint64_t rtmp_stream_total_bytes_sent(void *data) { struct rtmp_stream *stream = data; return stream->total_bytes_sent; } static int rtmp_stream_dropped_frames(void *data) { struct rtmp_stream *stream = data; return stream->dropped_frames; } struct obs_output_info rtmp_output_info = { .id = "rtmp_output", .flags = OBS_OUTPUT_AV | OBS_OUTPUT_ENCODED | OBS_OUTPUT_SERVICE | OBS_OUTPUT_MULTI_TRACK, .get_name = rtmp_stream_getname, .create = rtmp_stream_create, .destroy = rtmp_stream_destroy, .start = rtmp_stream_start, .stop = rtmp_stream_stop, .encoded_packet = rtmp_stream_data, .get_defaults = rtmp_stream_defaults, .get_properties = rtmp_stream_properties, .get_total_bytes = rtmp_stream_total_bytes_sent, .get_dropped_frames = rtmp_stream_dropped_frames };