rtmp-stream.c 22 KB

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