rtmp-stream.c 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872
  1. /******************************************************************************
  2. Copyright (C) 2014 by Hugh Bailey <[email protected]>
  3. This program is free software: you can redistribute it and/or modify
  4. it under the terms of the GNU General Public License as published by
  5. the Free Software Foundation, either version 2 of the License, or
  6. (at your option) any later version.
  7. This program is distributed in the hope that it will be useful,
  8. but WITHOUT ANY WARRANTY; without even the implied warranty of
  9. MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  10. GNU General Public License for more details.
  11. You should have received a copy of the GNU General Public License
  12. along with this program. If not, see <http://www.gnu.org/licenses/>.
  13. ******************************************************************************/
  14. #include <obs-module.h>
  15. #include <obs-avc.h>
  16. #include <util/platform.h>
  17. #include <util/circlebuf.h>
  18. #include <util/dstr.h>
  19. #include <util/threading.h>
  20. #include <inttypes.h>
  21. #include "librtmp/rtmp.h"
  22. #include "librtmp/log.h"
  23. #include "flv-mux.h"
  24. #ifdef _WIN32
  25. #include <Iphlpapi.h>
  26. #else
  27. #include <sys/ioctl.h>
  28. #endif
  29. #define do_log(level, format, ...) \
  30. blog(level, "[rtmp stream: '%s'] " format, \
  31. obs_output_get_name(stream->output), ##__VA_ARGS__)
  32. #define warn(format, ...) do_log(LOG_WARNING, format, ##__VA_ARGS__)
  33. #define info(format, ...) do_log(LOG_INFO, format, ##__VA_ARGS__)
  34. #define debug(format, ...) do_log(LOG_DEBUG, format, ##__VA_ARGS__)
  35. #define OPT_DROP_THRESHOLD "drop_threshold_ms"
  36. #define OPT_MAX_SHUTDOWN_TIME_SEC "max_shutdown_time_sec"
  37. //#define TEST_FRAMEDROPS
  38. struct rtmp_stream {
  39. obs_output_t *output;
  40. pthread_mutex_t packets_mutex;
  41. struct circlebuf packets;
  42. bool sent_headers;
  43. volatile bool connecting;
  44. pthread_t connect_thread;
  45. volatile bool active;
  46. volatile bool disconnected;
  47. pthread_t send_thread;
  48. int max_shutdown_time_sec;
  49. os_sem_t *send_sem;
  50. os_event_t *stop_event;
  51. struct dstr path, key;
  52. struct dstr username, password;
  53. struct dstr encoder_name;
  54. /* frame drop variables */
  55. int64_t drop_threshold_usec;
  56. int64_t min_drop_dts_usec;
  57. int min_priority;
  58. int64_t last_dts_usec;
  59. uint64_t total_bytes_sent;
  60. int dropped_frames;
  61. RTMP rtmp;
  62. };
  63. static const char *rtmp_stream_getname(void *unused)
  64. {
  65. UNUSED_PARAMETER(unused);
  66. return obs_module_text("RTMPStream");
  67. }
  68. static void log_rtmp(int level, const char *format, va_list args)
  69. {
  70. if (level > RTMP_LOGWARNING)
  71. return;
  72. blogva(LOG_INFO, format, args);
  73. }
  74. static inline size_t num_buffered_packets(struct rtmp_stream *stream);
  75. static inline void free_packets(struct rtmp_stream *stream)
  76. {
  77. size_t num_packets;
  78. pthread_mutex_lock(&stream->packets_mutex);
  79. num_packets = num_buffered_packets(stream);
  80. if (num_packets)
  81. info("Freeing %d remaining packets", (int)num_packets);
  82. while (stream->packets.size) {
  83. struct encoder_packet packet;
  84. circlebuf_pop_front(&stream->packets, &packet, sizeof(packet));
  85. obs_free_encoder_packet(&packet);
  86. }
  87. pthread_mutex_unlock(&stream->packets_mutex);
  88. }
  89. static inline bool stopping(struct rtmp_stream *stream)
  90. {
  91. return os_event_try(stream->stop_event) != EAGAIN;
  92. }
  93. static inline bool connecting(struct rtmp_stream *stream)
  94. {
  95. return os_atomic_load_bool(&stream->connecting);
  96. }
  97. static inline bool active(struct rtmp_stream *stream)
  98. {
  99. return os_atomic_load_bool(&stream->active);
  100. }
  101. static inline bool disconnected(struct rtmp_stream *stream)
  102. {
  103. return os_atomic_load_bool(&stream->disconnected);
  104. }
  105. static void rtmp_stream_destroy(void *data)
  106. {
  107. struct rtmp_stream *stream = data;
  108. if (stopping(stream) && !connecting(stream)) {
  109. pthread_join(stream->send_thread, NULL);
  110. } else if (connecting(stream) || active(stream)) {
  111. if (stream->connecting)
  112. pthread_join(stream->connect_thread, NULL);
  113. os_event_signal(stream->stop_event);
  114. if (active(stream)) {
  115. os_sem_post(stream->send_sem);
  116. obs_output_end_data_capture(stream->output);
  117. pthread_join(stream->send_thread, NULL);
  118. }
  119. }
  120. if (stream) {
  121. free_packets(stream);
  122. dstr_free(&stream->path);
  123. dstr_free(&stream->key);
  124. dstr_free(&stream->username);
  125. dstr_free(&stream->password);
  126. dstr_free(&stream->encoder_name);
  127. os_event_destroy(stream->stop_event);
  128. os_sem_destroy(stream->send_sem);
  129. pthread_mutex_destroy(&stream->packets_mutex);
  130. circlebuf_free(&stream->packets);
  131. bfree(stream);
  132. }
  133. }
  134. static void *rtmp_stream_create(obs_data_t *settings, obs_output_t *output)
  135. {
  136. struct rtmp_stream *stream = bzalloc(sizeof(struct rtmp_stream));
  137. stream->output = output;
  138. pthread_mutex_init_value(&stream->packets_mutex);
  139. RTMP_Init(&stream->rtmp);
  140. RTMP_LogSetCallback(log_rtmp);
  141. RTMP_LogSetLevel(RTMP_LOGWARNING);
  142. if (pthread_mutex_init(&stream->packets_mutex, NULL) != 0)
  143. goto fail;
  144. if (os_event_init(&stream->stop_event, OS_EVENT_TYPE_MANUAL) != 0)
  145. goto fail;
  146. UNUSED_PARAMETER(settings);
  147. return stream;
  148. fail:
  149. rtmp_stream_destroy(stream);
  150. return NULL;
  151. }
  152. static void rtmp_stream_stop(void *data)
  153. {
  154. struct rtmp_stream *stream = data;
  155. if (stopping(stream))
  156. return;
  157. if (connecting(stream))
  158. pthread_join(stream->connect_thread, NULL);
  159. os_event_signal(stream->stop_event);
  160. if (active(stream)) {
  161. os_sem_post(stream->send_sem);
  162. obs_output_end_data_capture(stream->output);
  163. }
  164. }
  165. static inline void set_rtmp_str(AVal *val, const char *str)
  166. {
  167. bool valid = (str && *str);
  168. val->av_val = valid ? (char*)str : NULL;
  169. val->av_len = valid ? (int)strlen(str) : 0;
  170. }
  171. static inline void set_rtmp_dstr(AVal *val, struct dstr *str)
  172. {
  173. bool valid = !dstr_is_empty(str);
  174. val->av_val = valid ? str->array : NULL;
  175. val->av_len = valid ? (int)str->len : 0;
  176. }
  177. static inline bool get_next_packet(struct rtmp_stream *stream,
  178. struct encoder_packet *packet)
  179. {
  180. bool new_packet = false;
  181. pthread_mutex_lock(&stream->packets_mutex);
  182. if (stream->packets.size) {
  183. circlebuf_pop_front(&stream->packets, packet,
  184. sizeof(struct encoder_packet));
  185. new_packet = true;
  186. }
  187. pthread_mutex_unlock(&stream->packets_mutex);
  188. return new_packet;
  189. }
  190. static bool discard_recv_data(struct rtmp_stream *stream, size_t size)
  191. {
  192. RTMP *rtmp = &stream->rtmp;
  193. uint8_t buf[512];
  194. #ifdef _WIN32
  195. int ret;
  196. #else
  197. ssize_t ret;
  198. #endif
  199. do {
  200. size_t bytes = size > 512 ? 512 : size;
  201. size -= bytes;
  202. #ifdef _WIN32
  203. ret = recv(rtmp->m_sb.sb_socket, buf, (int)bytes, 0);
  204. #else
  205. ret = recv(rtmp->m_sb.sb_socket, buf, bytes, 0);
  206. #endif
  207. if (ret <= 0) {
  208. #ifdef _WIN32
  209. int error = WSAGetLastError();
  210. #else
  211. int error = errno;
  212. #endif
  213. if (ret < 0) {
  214. do_log(LOG_ERROR, "recv error: %d (%d bytes)",
  215. error, (int)size);
  216. }
  217. return false;
  218. }
  219. } while (size > 0);
  220. return true;
  221. }
  222. static int send_packet(struct rtmp_stream *stream,
  223. struct encoder_packet *packet, bool is_header, size_t idx)
  224. {
  225. uint8_t *data;
  226. size_t size;
  227. int recv_size = 0;
  228. int ret = 0;
  229. #ifdef _WIN32
  230. ret = ioctlsocket(stream->rtmp.m_sb.sb_socket, FIONREAD,
  231. (u_long*)&recv_size);
  232. #else
  233. ret = ioctl(stream->rtmp.m_sb.sb_socket, FIONREAD, &recv_size);
  234. #endif
  235. if (ret >= 0 && recv_size > 0) {
  236. if (!discard_recv_data(stream, (size_t)recv_size))
  237. return -1;
  238. }
  239. flv_packet_mux(packet, &data, &size, is_header);
  240. #ifdef TEST_FRAMEDROPS
  241. os_sleep_ms(rand() % 40);
  242. #endif
  243. ret = RTMP_Write(&stream->rtmp, (char*)data, (int)size, (int)idx);
  244. bfree(data);
  245. obs_free_encoder_packet(packet);
  246. stream->total_bytes_sent += size;
  247. return ret;
  248. }
  249. static inline void send_headers(struct rtmp_stream *stream);
  250. static bool send_remaining_packets(struct rtmp_stream *stream)
  251. {
  252. struct encoder_packet packet;
  253. uint64_t max_ns = (uint64_t)stream->max_shutdown_time_sec * 1000000000;
  254. uint64_t begin_time_ns = os_gettime_ns();
  255. if (!stream->sent_headers)
  256. send_headers(stream);
  257. while (get_next_packet(stream, &packet)) {
  258. if (send_packet(stream, &packet, false, packet.track_idx) < 0)
  259. return false;
  260. /* Just disconnect if it takes too long to shut down */
  261. if ((os_gettime_ns() - begin_time_ns) > max_ns) {
  262. info("Took longer than %d second(s) to shut down, "
  263. "automatically stopping connection",
  264. stream->max_shutdown_time_sec);
  265. return false;
  266. }
  267. }
  268. return true;
  269. }
  270. static void *send_thread(void *data)
  271. {
  272. struct rtmp_stream *stream = data;
  273. os_set_thread_name("rtmp-stream: send_thread");
  274. while (os_sem_wait(stream->send_sem) == 0) {
  275. struct encoder_packet packet;
  276. if (stopping(stream))
  277. break;
  278. if (!get_next_packet(stream, &packet))
  279. continue;
  280. if (!stream->sent_headers)
  281. send_headers(stream);
  282. if (send_packet(stream, &packet, false, packet.track_idx) < 0) {
  283. os_atomic_set_bool(&stream->disconnected, true);
  284. break;
  285. }
  286. }
  287. if (!disconnected(stream) && !send_remaining_packets(stream))
  288. os_atomic_set_bool(&stream->disconnected, true);
  289. if (disconnected(stream)) {
  290. info("Disconnected from %s", stream->path.array);
  291. free_packets(stream);
  292. } else {
  293. info("User stopped the stream");
  294. }
  295. RTMP_Close(&stream->rtmp);
  296. if (!stopping(stream)) {
  297. pthread_detach(stream->send_thread);
  298. obs_output_signal_stop(stream->output, OBS_OUTPUT_DISCONNECTED);
  299. }
  300. os_event_reset(stream->stop_event);
  301. os_atomic_set_bool(&stream->active, false);
  302. stream->sent_headers = false;
  303. return NULL;
  304. }
  305. static bool send_meta_data(struct rtmp_stream *stream, size_t idx)
  306. {
  307. uint8_t *meta_data;
  308. size_t meta_data_size;
  309. bool success = flv_meta_data(stream->output, &meta_data,
  310. &meta_data_size, false, idx);
  311. if (success) {
  312. RTMP_Write(&stream->rtmp, (char*)meta_data,
  313. (int)meta_data_size, (int)idx);
  314. bfree(meta_data);
  315. }
  316. return success;
  317. }
  318. static bool send_audio_header(struct rtmp_stream *stream, size_t idx)
  319. {
  320. obs_output_t *context = stream->output;
  321. obs_encoder_t *aencoder = obs_output_get_audio_encoder(context, idx);
  322. uint8_t *header;
  323. struct encoder_packet packet = {
  324. .type = OBS_ENCODER_AUDIO,
  325. .timebase_den = 1
  326. };
  327. if (!aencoder)
  328. return false;
  329. obs_encoder_get_extra_data(aencoder, &header, &packet.size);
  330. packet.data = bmemdup(header, packet.size);
  331. send_packet(stream, &packet, true, idx);
  332. return true;
  333. }
  334. static void send_video_header(struct rtmp_stream *stream)
  335. {
  336. obs_output_t *context = stream->output;
  337. obs_encoder_t *vencoder = obs_output_get_video_encoder(context);
  338. uint8_t *header;
  339. size_t size;
  340. struct encoder_packet packet = {
  341. .type = OBS_ENCODER_VIDEO,
  342. .timebase_den = 1,
  343. .keyframe = true
  344. };
  345. obs_encoder_get_extra_data(vencoder, &header, &size);
  346. packet.size = obs_parse_avc_header(&packet.data, header, size);
  347. send_packet(stream, &packet, true, 0);
  348. }
  349. static inline void send_headers(struct rtmp_stream *stream)
  350. {
  351. stream->sent_headers = true;
  352. size_t i = 0;
  353. send_audio_header(stream, i++);
  354. send_video_header(stream);
  355. while (send_audio_header(stream, i++));
  356. }
  357. static inline bool reset_semaphore(struct rtmp_stream *stream)
  358. {
  359. os_sem_destroy(stream->send_sem);
  360. return os_sem_init(&stream->send_sem, 0) == 0;
  361. }
  362. #ifdef _WIN32
  363. #define socklen_t int
  364. #endif
  365. #define MIN_SENDBUF_SIZE 65535
  366. static void adjust_sndbuf_size(struct rtmp_stream *stream, int new_size)
  367. {
  368. int cur_sendbuf_size = new_size;
  369. socklen_t int_size = sizeof(int);
  370. getsockopt(stream->rtmp.m_sb.sb_socket, SOL_SOCKET, SO_SNDBUF,
  371. (char*)&cur_sendbuf_size, &int_size);
  372. if (cur_sendbuf_size < new_size) {
  373. cur_sendbuf_size = new_size;
  374. setsockopt(stream->rtmp.m_sb.sb_socket, SOL_SOCKET, SO_SNDBUF,
  375. (const char*)&cur_sendbuf_size, int_size);
  376. }
  377. }
  378. static int init_send(struct rtmp_stream *stream)
  379. {
  380. int ret;
  381. size_t idx = 0;
  382. #if defined(_WIN32)
  383. adjust_sndbuf_size(stream, MIN_SENDBUF_SIZE);
  384. #endif
  385. reset_semaphore(stream);
  386. ret = pthread_create(&stream->send_thread, NULL, send_thread, stream);
  387. if (ret != 0) {
  388. RTMP_Close(&stream->rtmp);
  389. warn("Failed to create send thread");
  390. return OBS_OUTPUT_ERROR;
  391. }
  392. os_atomic_set_bool(&stream->active, true);
  393. while (send_meta_data(stream, idx++));
  394. obs_output_begin_data_capture(stream->output, 0);
  395. return OBS_OUTPUT_SUCCESS;
  396. }
  397. #ifdef _WIN32
  398. static void win32_log_interface_type(struct rtmp_stream *stream)
  399. {
  400. RTMP *rtmp = &stream->rtmp;
  401. MIB_IPFORWARDROW route;
  402. uint32_t dest_addr, source_addr;
  403. char hostname[256];
  404. HOSTENT *h;
  405. if (rtmp->Link.hostname.av_len >= sizeof(hostname) - 1)
  406. return;
  407. strncpy(hostname, rtmp->Link.hostname.av_val, sizeof(hostname));
  408. hostname[rtmp->Link.hostname.av_len] = 0;
  409. h = gethostbyname(hostname);
  410. if (!h)
  411. return;
  412. dest_addr = *(uint32_t*)h->h_addr_list[0];
  413. if (rtmp->m_bindIP.addrLen == 0)
  414. source_addr = 0;
  415. else if (rtmp->m_bindIP.addr.ss_family = AF_INET)
  416. source_addr = (*(struct sockaddr_in*)&rtmp->m_bindIP)
  417. .sin_addr.S_un.S_addr;
  418. else
  419. return;
  420. if (!GetBestRoute(dest_addr, source_addr, &route)) {
  421. MIB_IFROW row;
  422. memset(&row, 0, sizeof(row));
  423. row.dwIndex = route.dwForwardIfIndex;
  424. if (!GetIfEntry(&row)) {
  425. uint32_t speed =row.dwSpeed / 1000000;
  426. char *type;
  427. struct dstr other = {0};
  428. if (row.dwType == IF_TYPE_ETHERNET_CSMACD) {
  429. type = "ethernet";
  430. } else if (row.dwType == IF_TYPE_IEEE80211) {
  431. type = "802.11";
  432. } else {
  433. dstr_printf(&other, "type %lu", row.dwType);
  434. type = other.array;
  435. }
  436. info("Interface: %s (%s, %lu mbps)", row.bDescr, type,
  437. speed);
  438. dstr_free(&other);
  439. }
  440. }
  441. }
  442. #endif
  443. static int try_connect(struct rtmp_stream *stream)
  444. {
  445. if (dstr_is_empty(&stream->path)) {
  446. warn("URL is empty");
  447. return OBS_OUTPUT_BAD_PATH;
  448. }
  449. info("Connecting to RTMP URL %s...", stream->path.array);
  450. memset(&stream->rtmp.Link, 0, sizeof(stream->rtmp.Link));
  451. if (!RTMP_SetupURL(&stream->rtmp, stream->path.array))
  452. return OBS_OUTPUT_BAD_PATH;
  453. RTMP_EnableWrite(&stream->rtmp);
  454. dstr_copy(&stream->encoder_name, "FMLE/3.0 (compatible; obs-studio/");
  455. #ifdef HAVE_OBSCONFIG_H
  456. dstr_cat(&stream->encoder_name, OBS_VERSION);
  457. #else
  458. dstr_catf(&stream->encoder_name, "%d.%d.%d",
  459. LIBOBS_API_MAJOR_VER,
  460. LIBOBS_API_MINOR_VER,
  461. LIBOBS_API_PATCH_VER);
  462. #endif
  463. dstr_cat(&stream->encoder_name, "; FMSc/1.0)");
  464. set_rtmp_dstr(&stream->rtmp.Link.pubUser, &stream->username);
  465. set_rtmp_dstr(&stream->rtmp.Link.pubPasswd, &stream->password);
  466. set_rtmp_dstr(&stream->rtmp.Link.flashVer, &stream->encoder_name);
  467. stream->rtmp.Link.swfUrl = stream->rtmp.Link.tcUrl;
  468. RTMP_AddStream(&stream->rtmp, stream->key.array);
  469. for (size_t idx = 1;; idx++) {
  470. obs_encoder_t *encoder = obs_output_get_audio_encoder(
  471. stream->output, idx);
  472. const char *encoder_name;
  473. if (!encoder)
  474. break;
  475. encoder_name = obs_encoder_get_name(encoder);
  476. RTMP_AddStream(&stream->rtmp, encoder_name);
  477. }
  478. stream->rtmp.m_outChunkSize = 4096;
  479. stream->rtmp.m_bSendChunkSizeInfo = true;
  480. stream->rtmp.m_bUseNagle = true;
  481. #ifdef _WIN32
  482. win32_log_interface_type(stream);
  483. #endif
  484. if (!RTMP_Connect(&stream->rtmp, NULL))
  485. return OBS_OUTPUT_CONNECT_FAILED;
  486. if (!RTMP_ConnectStream(&stream->rtmp, 0))
  487. return OBS_OUTPUT_INVALID_STREAM;
  488. info("Connection to %s successful", stream->path.array);
  489. return init_send(stream);
  490. }
  491. static bool init_connect(struct rtmp_stream *stream)
  492. {
  493. obs_service_t *service;
  494. obs_data_t *settings;
  495. if (stopping(stream))
  496. pthread_join(stream->send_thread, NULL);
  497. free_packets(stream);
  498. service = obs_output_get_service(stream->output);
  499. if (!service)
  500. return false;
  501. os_atomic_set_bool(&stream->disconnected, false);
  502. stream->total_bytes_sent = 0;
  503. stream->dropped_frames = 0;
  504. stream->min_drop_dts_usec= 0;
  505. stream->min_priority = 0;
  506. settings = obs_output_get_settings(stream->output);
  507. dstr_copy(&stream->path, obs_service_get_url(service));
  508. dstr_copy(&stream->key, obs_service_get_key(service));
  509. dstr_copy(&stream->username, obs_service_get_username(service));
  510. dstr_copy(&stream->password, obs_service_get_password(service));
  511. stream->drop_threshold_usec =
  512. (int64_t)obs_data_get_int(settings, OPT_DROP_THRESHOLD) * 1000;
  513. stream->max_shutdown_time_sec =
  514. (int)obs_data_get_int(settings, OPT_MAX_SHUTDOWN_TIME_SEC);
  515. obs_data_release(settings);
  516. return true;
  517. }
  518. static void *connect_thread(void *data)
  519. {
  520. struct rtmp_stream *stream = data;
  521. int ret;
  522. os_set_thread_name("rtmp-stream: connect_thread");
  523. if (!init_connect(stream)) {
  524. obs_output_signal_stop(stream->output, OBS_OUTPUT_BAD_PATH);
  525. return NULL;
  526. }
  527. ret = try_connect(stream);
  528. if (ret != OBS_OUTPUT_SUCCESS) {
  529. obs_output_signal_stop(stream->output, ret);
  530. info("Connection to %s failed: %d", stream->path.array, ret);
  531. }
  532. if (!stopping(stream))
  533. pthread_detach(stream->connect_thread);
  534. os_atomic_set_bool(&stream->connecting, false);
  535. return NULL;
  536. }
  537. static bool rtmp_stream_start(void *data)
  538. {
  539. struct rtmp_stream *stream = data;
  540. if (!obs_output_can_begin_data_capture(stream->output, 0))
  541. return false;
  542. if (!obs_output_initialize_encoders(stream->output, 0))
  543. return false;
  544. os_atomic_set_bool(&stream->connecting, true);
  545. return pthread_create(&stream->connect_thread, NULL, connect_thread,
  546. stream) == 0;
  547. }
  548. static inline bool add_packet(struct rtmp_stream *stream,
  549. struct encoder_packet *packet)
  550. {
  551. circlebuf_push_back(&stream->packets, packet,
  552. sizeof(struct encoder_packet));
  553. stream->last_dts_usec = packet->dts_usec;
  554. return true;
  555. }
  556. static inline size_t num_buffered_packets(struct rtmp_stream *stream)
  557. {
  558. return stream->packets.size / sizeof(struct encoder_packet);
  559. }
  560. static void drop_frames(struct rtmp_stream *stream)
  561. {
  562. struct circlebuf new_buf = {0};
  563. int drop_priority = 0;
  564. uint64_t last_drop_dts_usec = 0;
  565. int num_frames_dropped = 0;
  566. debug("Previous packet count: %d", (int)num_buffered_packets(stream));
  567. circlebuf_reserve(&new_buf, sizeof(struct encoder_packet) * 8);
  568. while (stream->packets.size) {
  569. struct encoder_packet packet;
  570. circlebuf_pop_front(&stream->packets, &packet, sizeof(packet));
  571. last_drop_dts_usec = packet.dts_usec;
  572. /* do not drop audio data or video keyframes */
  573. if (packet.type == OBS_ENCODER_AUDIO ||
  574. packet.drop_priority == OBS_NAL_PRIORITY_HIGHEST) {
  575. circlebuf_push_back(&new_buf, &packet, sizeof(packet));
  576. } else {
  577. if (drop_priority < packet.drop_priority)
  578. drop_priority = packet.drop_priority;
  579. num_frames_dropped++;
  580. obs_free_encoder_packet(&packet);
  581. }
  582. }
  583. circlebuf_free(&stream->packets);
  584. stream->packets = new_buf;
  585. stream->min_priority = drop_priority;
  586. stream->min_drop_dts_usec = last_drop_dts_usec;
  587. stream->dropped_frames += num_frames_dropped;
  588. debug("New packet count: %d", (int)num_buffered_packets(stream));
  589. }
  590. static void check_to_drop_frames(struct rtmp_stream *stream)
  591. {
  592. struct encoder_packet first;
  593. int64_t buffer_duration_usec;
  594. if (num_buffered_packets(stream) < 5)
  595. return;
  596. circlebuf_peek_front(&stream->packets, &first, sizeof(first));
  597. /* do not drop frames if frames were just dropped within this time */
  598. if (first.dts_usec < stream->min_drop_dts_usec)
  599. return;
  600. /* if the amount of time stored in the buffered packets waiting to be
  601. * sent is higher than threshold, drop frames */
  602. buffer_duration_usec = stream->last_dts_usec - first.dts_usec;
  603. if (buffer_duration_usec > stream->drop_threshold_usec) {
  604. drop_frames(stream);
  605. debug("dropping %" PRId64 " worth of frames",
  606. buffer_duration_usec);
  607. }
  608. }
  609. static bool add_video_packet(struct rtmp_stream *stream,
  610. struct encoder_packet *packet)
  611. {
  612. check_to_drop_frames(stream);
  613. /* if currently dropping frames, drop packets until it reaches the
  614. * desired priority */
  615. if (packet->priority < stream->min_priority) {
  616. stream->dropped_frames++;
  617. return false;
  618. } else {
  619. stream->min_priority = 0;
  620. }
  621. return add_packet(stream, packet);
  622. }
  623. static void rtmp_stream_data(void *data, struct encoder_packet *packet)
  624. {
  625. struct rtmp_stream *stream = data;
  626. struct encoder_packet new_packet;
  627. bool added_packet = false;
  628. if (disconnected(stream))
  629. return;
  630. if (packet->type == OBS_ENCODER_VIDEO)
  631. obs_parse_avc_packet(&new_packet, packet);
  632. else
  633. obs_duplicate_encoder_packet(&new_packet, packet);
  634. pthread_mutex_lock(&stream->packets_mutex);
  635. if (!disconnected(stream)) {
  636. added_packet = (packet->type == OBS_ENCODER_VIDEO) ?
  637. add_video_packet(stream, &new_packet) :
  638. add_packet(stream, &new_packet);
  639. }
  640. pthread_mutex_unlock(&stream->packets_mutex);
  641. if (added_packet)
  642. os_sem_post(stream->send_sem);
  643. else
  644. obs_free_encoder_packet(&new_packet);
  645. }
  646. static void rtmp_stream_defaults(obs_data_t *defaults)
  647. {
  648. obs_data_set_default_int(defaults, OPT_DROP_THRESHOLD, 600);
  649. obs_data_set_default_int(defaults, OPT_MAX_SHUTDOWN_TIME_SEC, 5);
  650. }
  651. static obs_properties_t *rtmp_stream_properties(void *unused)
  652. {
  653. UNUSED_PARAMETER(unused);
  654. obs_properties_t *props = obs_properties_create();
  655. obs_properties_add_int(props, OPT_DROP_THRESHOLD,
  656. obs_module_text("RTMPStream.DropThreshold"),
  657. 200, 10000, 100);
  658. return props;
  659. }
  660. static uint64_t rtmp_stream_total_bytes_sent(void *data)
  661. {
  662. struct rtmp_stream *stream = data;
  663. return stream->total_bytes_sent;
  664. }
  665. static int rtmp_stream_dropped_frames(void *data)
  666. {
  667. struct rtmp_stream *stream = data;
  668. return stream->dropped_frames;
  669. }
  670. struct obs_output_info rtmp_output_info = {
  671. .id = "rtmp_output",
  672. .flags = OBS_OUTPUT_AV |
  673. OBS_OUTPUT_ENCODED |
  674. OBS_OUTPUT_SERVICE |
  675. OBS_OUTPUT_MULTI_TRACK,
  676. .get_name = rtmp_stream_getname,
  677. .create = rtmp_stream_create,
  678. .destroy = rtmp_stream_destroy,
  679. .start = rtmp_stream_start,
  680. .stop = rtmp_stream_stop,
  681. .encoded_packet = rtmp_stream_data,
  682. .get_defaults = rtmp_stream_defaults,
  683. .get_properties = rtmp_stream_properties,
  684. .get_total_bytes = rtmp_stream_total_bytes_sent,
  685. .get_dropped_frames = rtmp_stream_dropped_frames
  686. };