tomcast reads mpeg transport streams over http or udp (multicast or unicast) and outputs them to chosen multicast group. Basically a simple http2multicast daemon designed to work 24/7.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

tomcast.c 36KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142
  1. /*
  2. * tomcast
  3. * Copyright (C) 2010-2013 Unix Solutions Ltd.
  4. *
  5. * This program is free software; you can redistribute it and/or modify
  6. * it under the terms of the GNU General Public License version 2
  7. * as published by the Free Software Foundation.
  8. *
  9. * This program is distributed in the hope that it will be useful,
  10. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. * GNU General Public License (COPYING file) for more details.
  13. *
  14. */
  15. #include <stdio.h>
  16. #include <stdlib.h>
  17. #include <unistd.h>
  18. #include <string.h>
  19. #include <pthread.h>
  20. #include <errno.h>
  21. #include <arpa/inet.h>
  22. #include <netinet/in.h>
  23. #include <sys/types.h>
  24. #include <sys/socket.h>
  25. #include <sys/types.h>
  26. #include <sys/stat.h>
  27. #include <sys/time.h>
  28. #include <signal.h>
  29. #include <fcntl.h>
  30. #include <regex.h>
  31. #include <netdb.h> // for uint32_t
  32. #include "libfuncs/libfuncs.h"
  33. #include "config.h"
  34. #include "web_server.h"
  35. #define DNS_RESOLVER_TIMEOUT 5000
  36. #define FDGETLINE_TIMEOUT 500
  37. #define FDGETLINE_RETRIES 30
  38. #define FDREAD_TIMEOUT 1500
  39. #define FDREAD_RETRIES 7
  40. #define FDWRITE_TIMEOUT 1500
  41. #define FDWRITE_RETRIES 7
  42. /* How much to wait for connection to be established with channel source (miliseconds) */
  43. #define PROXY_CONNECT_TIMEOUT 1000
  44. /* Seconds to sleep between retries (miliseconds) */
  45. #define PROXY_RETRY_TIMEOUT 1000
  46. #define TRANSPORT_PACKET_SIZE 188
  47. #define TRANSPORT_EXTEDED_PACKET_SIZE 192
  48. #define TRANSPORT_PACKETS_PER_NETWORK_PACKET 7
  49. #define TRANSPORT_SYNC_BYTE 0x47
  50. #define FRAME_PACKET_SIZE (TRANSPORT_PACKET_SIZE * TRANSPORT_PACKETS_PER_NETWORK_PACKET)
  51. #ifndef FREE
  52. #define FREE(x) if(x) { free(x); x=NULL; }
  53. #endif
  54. #ifndef POLLRDHUP
  55. #define POLLRDHUP 0
  56. #endif
  57. char *server_sig = "tomcast";
  58. char *server_ver = "1.15";
  59. char *copyright = "Copyright (C) 2010-2013 Unix Solutions Ltd.";
  60. static struct config config;
  61. channel_source get_sproto(char *url) {
  62. return strncmp(url, "http", 4)==0 ? tcp_sock : udp_sock;
  63. }
  64. CHANSRC *init_chansrc(char *url) {
  65. regex_t re;
  66. regmatch_t res[5];
  67. regcomp(&re, "^([a-z]+)://([^:/?]+):?([0-9]*)/?(.*)", REG_EXTENDED);
  68. if (regexec(&re,url,5,res,0)==0) {
  69. char *data = strdup(url);
  70. char *proto, *host, *port, *path;
  71. int iport;
  72. proto= data+res[1].rm_so; data[res[1].rm_eo]=0;
  73. host = data+res[2].rm_so; data[res[2].rm_eo]=0;
  74. port = data+res[3].rm_so; data[res[3].rm_eo]=0;
  75. path = data+res[4].rm_so; data[res[4].rm_eo]=0;
  76. iport = atoi(port);
  77. /* Setup */
  78. CHANSRC *src = calloc(1, sizeof(CHANSRC));
  79. src->proto = strdup(proto);
  80. src->sproto= get_sproto(url);
  81. src->host = strdup(host);
  82. src->port = iport ? iport : 80;
  83. src->path = strdup(path);
  84. FREE(data);
  85. regfree(&re);
  86. return src;
  87. }
  88. regfree(&re);
  89. return NULL;
  90. }
  91. void free_chansrc(CHANSRC *url) {
  92. if (url) {
  93. FREE(url->proto);
  94. FREE(url->host);
  95. FREE(url->path);
  96. FREE(url);
  97. }
  98. };
  99. int is_valid_url(char *url) {
  100. regex_t re;
  101. regmatch_t res[5];
  102. int ret;
  103. regcomp(&re, "^([a-z]+)://([^:/?]+):?([0-9]*)/?(.*)", REG_EXTENDED);
  104. ret = regexec(&re,url,5,res,0);
  105. regfree(&re);
  106. return ret == 0;
  107. }
  108. void add_channel_source(CHANNEL *c, char *src) {
  109. if (c->num_src >= MAX_CHANNEL_SOURCES-1)
  110. return;
  111. c->sources[c->num_src] = strdup(src);
  112. if (c->num_src == 0) /* Set default source to first one */
  113. c->source = c->sources[c->num_src];
  114. c->num_src++;
  115. }
  116. void next_channel_source(CHANNEL *c) {
  117. if (c->num_src <= 1)
  118. return;
  119. // uint8_t old_src = c->curr_src;
  120. c->curr_src++;
  121. if (c->curr_src >= MAX_CHANNEL_SOURCES-1 || c->sources[c->curr_src] == NULL)
  122. c->curr_src = 0;
  123. c->source = c->sources[c->curr_src];
  124. // LOGf("CHAN : Switch source | Channel: %s OldSrc: %d %s NewSrc: %d %s\n", c->name, old_src, c->sources[old_src], c->curr_src, c->source);
  125. }
  126. void set_channel_source(CHANNEL *c, uint8_t src_id) {
  127. if (src_id >= MAX_CHANNEL_SOURCES-1 || c->sources[src_id] == NULL)
  128. return;
  129. // uint8_t old_src = c->curr_src;
  130. c->curr_src = src_id;
  131. c->source = c->sources[c->curr_src];
  132. // LOGf("CHAN : Set source | Channel: %s OldSrc: %d %s NewSrc: %d %s\n", c->name, old_src, c->sources[old_src], c->curr_src, c->source);
  133. }
  134. CHANNEL * new_channel(char *name, char *source, char *dest, int port) {
  135. CHANNEL *c = calloc(1, sizeof(CHANNEL));
  136. c->name = strdup(name);
  137. c->dest_host = strdup(dest);
  138. c->dest_port = port;
  139. add_channel_source(c, source);
  140. return c;
  141. }
  142. void free_channel(CHANNEL *c) {
  143. int i;
  144. for (i=c->num_src-1; i>=0; i--) {
  145. FREE(c->sources[i]);
  146. }
  147. FREE(c->name);
  148. FREE(c->dest_host);
  149. c->source = NULL;
  150. FREE(c);
  151. }
  152. void free_channel_p(void *c) {
  153. free_channel(c);
  154. }
  155. int send_reset_opt = 0;
  156. int multicast_ttl = 1;
  157. struct in_addr output_intf = { .s_addr = INADDR_ANY };
  158. int connect_multicast(struct sockaddr_in send_to) {
  159. int sendsock = socket(AF_INET, SOCK_DGRAM, 0);
  160. if (sendsock < 0) {
  161. LOGf("socket(SOCK_DGRAM): %s\n", strerror(errno));
  162. return -1;
  163. }
  164. int on = 1;
  165. setsockopt(sendsock, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
  166. // subscribe to multicast group
  167. // LOGf("Using ttl %d\n", multicast_ttl);
  168. if (IN_MULTICAST(ntohl(send_to.sin_addr.s_addr))) {
  169. if (setsockopt(sendsock, IPPROTO_IP, IP_MULTICAST_TTL, &multicast_ttl, sizeof(multicast_ttl)) < 0) {
  170. LOGf("setsockopt(IP_MUTICAST_TTL): %s\n", strerror(errno));
  171. close(sendsock);
  172. return -1;
  173. }
  174. if (setsockopt(sendsock, IPPROTO_IP, IP_MULTICAST_IF, &output_intf, sizeof(output_intf)) < 0) {
  175. LOGf("setsockopt(IP_MUTICAST_IF, %s): %s\n", strerror(errno), inet_ntoa(output_intf));
  176. close(sendsock);
  177. return -1;
  178. }
  179. }
  180. int writebuflen = FRAME_PACKET_SIZE * 1000;
  181. if (setsockopt(sendsock, SOL_SOCKET, SO_SNDBUF, (const char *)&writebuflen, sizeof(writebuflen)) < 0)
  182. log_perror("setsockopt(): setsockopt(SO_SNDBUF)", errno);
  183. // call connect to get errors
  184. if (connect(sendsock, (struct sockaddr *)&send_to, sizeof send_to)) {
  185. LOGf("udp_connect() error: %s\n", strerror(errno));
  186. close(sendsock);
  187. return -1;
  188. }
  189. return sendsock;
  190. }
  191. void proxy_set_status(RESTREAMER *r, const char *proxy_status) {
  192. pthread_rwlock_wrlock(&r->lock);
  193. snprintf(r->status, sizeof(r->status), "%s", proxy_status);
  194. pthread_rwlock_unlock(&r->lock);
  195. }
  196. void connect_destination(RESTREAMER *r) {
  197. CHANNEL *c = r->channel;
  198. if (r->clientsock >= 0)
  199. shutdown_fd(&(r->clientsock));
  200. r->clientsock = connect_multicast(r->dst_sockname);
  201. LOGf("CONN : Connected dst_fd: %i | Chan: %s Dest: udp://%s:%d\n", r->clientsock, c->name, c->dest_host, c->dest_port);
  202. proxy_set_status(r, "Connected udp");
  203. }
  204. RESTREAMER * new_restreamer(const char *name, CHANNEL *channel) {
  205. int active = 1;
  206. struct sockaddr_in sockname;
  207. int dret = async_resolve_host(channel->dest_host, channel->dest_port, &sockname, DNS_RESOLVER_TIMEOUT, &active);
  208. if (dret != 0) {
  209. if (dret == 1)
  210. LOGf("ERR : Can't resolve host | Chan: %s Dest: udp://%s:%d\n", channel->name, channel->dest_host, channel->dest_port);
  211. if (dret == 2)
  212. LOGf("ERR : DNS timeout | Chan: %s Dest: udp://%s:%d\n", channel->name, channel->dest_host, channel->dest_port);
  213. return NULL;
  214. }
  215. RESTREAMER *r = calloc(1, sizeof(RESTREAMER));
  216. r->name = strdup(name);
  217. r->sock = -1;
  218. r->channel = channel;
  219. r->clientsock = -1;
  220. r->dst_sockname = sockname;
  221. pthread_rwlock_init(&r->lock, NULL);
  222. connect_destination(r);
  223. return r;
  224. }
  225. void free_restreamer(RESTREAMER *r) {
  226. if (r->sock > -1)
  227. shutdown_fd(&(r->sock));
  228. if (r->freechannel)
  229. free_channel(r->channel);
  230. FREE(r->name);
  231. FREE(r);
  232. }
  233. char TS_NULL_FRAME[FRAME_PACKET_SIZE];
  234. regex_t http_response;
  235. void proxy_log(RESTREAMER *r, char *msg, char *info) {
  236. LOGf("%s: %sChan: %s Src: %s Dst: udp://%s:%d SrcIP: %s SrcFD: %i DstFD: %i\n",
  237. msg,
  238. info,
  239. r->channel->name,
  240. r->channel->source,
  241. r->channel->dest_host,
  242. r->channel->dest_port,
  243. inet_ntoa(r->src_sockname.sin_addr),
  244. r->sock,
  245. r->clientsock
  246. );
  247. }
  248. int load_channels_config(struct config *cfg) {
  249. regex_t re;
  250. regmatch_t res[5];
  251. char line[1024];
  252. int fd;
  253. int num_channels = 0;
  254. if (pthread_mutex_trylock(&cfg->channels_lock) != 0)
  255. return -1;
  256. fd = open(cfg->channels_file, O_RDONLY);
  257. if (fd != -1) {
  258. struct timeval tv;
  259. gettimeofday(&tv, NULL);
  260. unsigned int randstate = tv.tv_usec;
  261. int cookie = rand_r(&randstate);
  262. regcomp(&re, "^([A-Za-z0-9]+)\t+([0-9.]+):([0-9]+)\t+(.*)", REG_EXTENDED);
  263. LIST *old_chanconf;
  264. LIST *new_chanconf = list_new("chanconf");
  265. while (fdgetline(fd,line,sizeof(line)) > 0) {
  266. chomp(line);
  267. if (regexec(&re,line,5,res,0)==0) {
  268. char *name, *dest_host, *dest_port, *source;
  269. char *org = strdup(line);
  270. name = org+res[1].rm_so; org[res[1].rm_eo]=0;
  271. dest_host = org+res[2].rm_so; org[res[2].rm_eo]=0;
  272. dest_port = org+res[3].rm_so; org[res[3].rm_eo]=0;
  273. source = org+res[4].rm_so; org[res[4].rm_eo]=0;
  274. if (!is_valid_url(source)) {
  275. LOGf("CONF : Invalid url: %s\n", source);
  276. FREE(org);
  277. goto report_error;
  278. }
  279. /* Search for already added channel */
  280. LNODE *l, *tmp;
  281. CHANNEL *chan = NULL;
  282. list_for_each_reverse(new_chanconf, l, tmp) {
  283. if (strcmp(name, ((CHANNEL *)l->data)->name)==0) {
  284. chan = l->data;
  285. break;
  286. }
  287. }
  288. if (!chan) {
  289. list_add(new_chanconf, new_channel(name, source, dest_host, atoi(dest_port)));
  290. num_channels++;
  291. } else {
  292. add_channel_source(chan, source);
  293. }
  294. FREE(org);
  295. } else {
  296. report_error:
  297. if (strlen(line) > 2 && line[0] != '#') {
  298. LOGf("CONF : Invalid config line: %s\n", line);
  299. }
  300. }
  301. }
  302. regfree(&re);
  303. shutdown_fd(&fd);
  304. /* Save current chanconf */
  305. old_chanconf = cfg->chanconf;
  306. /* Switch chanconf */
  307. cfg->chanconf = new_chanconf;
  308. /* Rewrite restreamer channels */
  309. LNODE *lc, *lr, *lctmp, *lrtmp;
  310. CHANNEL *chan;
  311. list_lock(cfg->restreamer); // Unlocked after second list_for_each(restreamer)
  312. list_lock(cfg->chanconf);
  313. list_for_each(cfg->chanconf, lc, lctmp) {
  314. chan = lc->data;
  315. list_for_each(cfg->restreamer, lr, lrtmp) {
  316. if (strcmp(chan->name, ((RESTREAMER *)lr->data)->name)==0) {
  317. RESTREAMER *restr = lr->data;
  318. /* Mark the restreamer as valid */
  319. restr->cookie = cookie;
  320. /* Check if current source exists in new channel configuration */
  321. int i, src_found = -1;
  322. char *old_source = restr->channel->source;
  323. for (i=0; i<chan->num_src; i++) {
  324. if (strcmp(old_source, chan->sources[i]) == 0) {
  325. src_found = i;
  326. }
  327. }
  328. if (src_found > -1) {
  329. /* New configuration contains existing source, just update the reference */
  330. set_channel_source(chan, src_found);
  331. restr->channel = chan;
  332. } else {
  333. /* New configuration *DO NOT* contain existing source. Force reconnect */
  334. LOGf("PROXY: Source changed | Channel: %s srv_fd: %d Old:%s New:%s\n", chan->name, restr->sock, restr->channel->source, chan->source);
  335. /* The order is important! */
  336. set_channel_source(chan, chan->num_src-1); /* Set source to last one. On reconnect next source will be used. */
  337. restr->channel = chan;
  338. restr->reconnect = 1;
  339. }
  340. break;
  341. }
  342. }
  343. }
  344. list_unlock(cfg->chanconf);
  345. /* Kill restreamers that serve channels that no longer exist */
  346. list_for_each(cfg->restreamer, lr, lrtmp) {
  347. RESTREAMER *r = lr->data;
  348. /* This restreamer should no longer serve clients */
  349. if (r->cookie != cookie) {
  350. proxy_log(r, "CLEAR", "Channel removed ");
  351. /* Replace channel reference with real object and instruct free_restreamer to free it */
  352. r->channel = new_channel(r->channel->name, r->channel->source, r->channel->dest_host, r->channel->dest_port);
  353. r->freechannel = 1;
  354. r->dienow = 1;
  355. }
  356. }
  357. list_unlock(cfg->restreamer);
  358. /* Free old_chanconf */
  359. list_free(&old_chanconf, free_channel_p, NULL);
  360. } else {
  361. num_channels = -1;
  362. }
  363. pthread_mutex_unlock(&cfg->channels_lock);
  364. if (num_channels == -1)
  365. LOGf("CONF : Error loading channels!\n");
  366. else
  367. LOGf("CONF : %d channels loaded\n", num_channels);
  368. return num_channels;
  369. }
  370. void proxy_close(RESTREAMER *r) {
  371. proxy_log(r, "STOP ","");
  372. // If there are no clients left, no "Timeout" messages will be logged
  373. list_del_entry(config.restreamer, r);
  374. free_restreamer(r);
  375. }
  376. /*
  377. On the last try, send no-signal to clients and exit
  378. otherwise wait a little bit before trying again
  379. */
  380. #define DO_RECONNECT do \
  381. { \
  382. free_chansrc(src); \
  383. if (retries == 0) { \
  384. return -1; \
  385. } else { \
  386. if (errno != EHOSTUNREACH) /* When host is unreachable there is already a delay of ~4 secs per try so no sleep is needed */ \
  387. usleep(PROXY_RETRY_TIMEOUT * 1000); \
  388. return 1; \
  389. } \
  390. } while(0)
  391. #define FATAL_ERROR do \
  392. { \
  393. free_chansrc(src); \
  394. return -1; \
  395. } while (0)
  396. /*
  397. Returns:
  398. -1 = exit thread
  399. 1 = retry
  400. 0 = connected ok
  401. */
  402. int connect_source(RESTREAMER *r, int retries, int readbuflen, int *http_code) {
  403. CHANSRC *src = init_chansrc(r->channel->source);
  404. if (!src) {
  405. LOGf("ERR : Can't parse channel source | Channel: %s Source: %s\n", r->channel->name, r->channel->source);
  406. FATAL_ERROR;
  407. }
  408. r->connected = 0;
  409. r->reconnect = 0;
  410. int active = 1;
  411. int dret = async_resolve_host(src->host, src->port, &(r->src_sockname), DNS_RESOLVER_TIMEOUT, &active);
  412. if (dret != 0) {
  413. if (dret == 1) {
  414. proxy_log(r, "ERR ","Can't resolve src host");
  415. proxy_set_status(r, "ERROR: Can not resolve source host");
  416. }
  417. if (dret == 2) {
  418. proxy_log(r, "ERR ","Timeout resolving src host");
  419. proxy_set_status(r, "ERROR: Dns resolve timeout");
  420. }
  421. DO_RECONNECT;
  422. }
  423. char buf[1024];
  424. *http_code = 0;
  425. if (src->sproto == tcp_sock) {
  426. r->sock = socket(PF_INET, SOCK_STREAM, 0);
  427. if (r->sock < 0) {
  428. log_perror("play(): Could not create SOCK_STREAM socket.", errno);
  429. FATAL_ERROR;
  430. }
  431. proxy_log(r, "NEW ","");
  432. if (do_connect(r->sock, (struct sockaddr *)&(r->src_sockname), sizeof(r->src_sockname), PROXY_CONNECT_TIMEOUT) < 0) {
  433. LOGf("ERR : Error connecting to %s srv_fd: %i err: %s\n", r->channel->source, r->sock, strerror(errno));
  434. proxy_set_status(r, "ERROR: Can not connect to source");
  435. DO_RECONNECT;
  436. }
  437. snprintf(buf,sizeof(buf)-1, "GET /%s HTTP/1.0\r\nHost: %s:%u\r\nX-Smart-Client: yes\r\nUser-Agent: %s %s (%s)\r\n\r\n",
  438. src->path, src->host, src->port, server_sig, server_ver, config.ident);
  439. buf[sizeof(buf)-1] = 0;
  440. fdwrite(r->sock, buf, strlen(buf));
  441. char xresponse[128];
  442. memset(xresponse, 0, sizeof(xresponse));
  443. memset(buf, 0, sizeof(buf));
  444. regmatch_t res[4];
  445. while (fdgetline(r->sock,buf,sizeof(buf)-1)) {
  446. if (buf[0] == '\n' || buf[0] == '\r')
  447. break;
  448. if (strstr(buf,"HTTP/1.") != NULL) {
  449. if (regexec(&http_response,buf,3,res,0) != REG_NOMATCH) {
  450. char codestr[4];
  451. if ((unsigned long)(res[1].rm_eo - res[1].rm_so) < sizeof(xresponse)) {
  452. strncpy(xresponse, &buf[res[1].rm_so], res[1].rm_eo-res[1].rm_so);
  453. xresponse[res[1].rm_eo-res[1].rm_so] = '\0';
  454. chomp(xresponse);
  455. strncpy(codestr, &buf[res[2].rm_so], res[2].rm_eo-res[2].rm_so);
  456. codestr[3] = 0;
  457. *http_code = atoi(codestr);
  458. }
  459. }
  460. }
  461. if (*http_code == 504) { // Extract extra error code
  462. if (strstr(buf, "X-ErrorCode: ") != NULL) {
  463. *http_code = atoi(buf+13);
  464. break;
  465. }
  466. }
  467. }
  468. if (*http_code == 0) { // No valid HTTP response, retry
  469. LOGf("DEBUG: Server returned not valid HTTP code | srv_fd: %i\n", r->sock);
  470. proxy_set_status(r, "ERROR: Source returned invalid HTTP code");
  471. DO_RECONNECT;
  472. }
  473. if (*http_code == 504) { // No signal, exit
  474. LOGf("ERR : Get no-signal for %s from %s on srv_fd: %i\n", r->channel->name, r->channel->source, r->sock);
  475. proxy_set_status(r, "ERROR: Source returned no-signal");
  476. FATAL_ERROR;
  477. }
  478. if (*http_code > 300) { // Unhandled or error codes, exit
  479. LOGf("ERR : Get code %i for %s from %s on srv_fd: %i exiting.\n", *http_code, r->channel->name, r->channel->source, r->sock);
  480. proxy_set_status(r, "ERROR: Source returned unhandled error code");
  481. FATAL_ERROR;
  482. }
  483. // connected ok, continue
  484. } else {
  485. if (!IN_MULTICAST(ntohl(r->src_sockname.sin_addr.s_addr))) {
  486. LOGf("ERR : %s is not multicast address\n", r->channel->source);
  487. FATAL_ERROR;
  488. }
  489. struct ip_mreq mreq;
  490. struct sockaddr_in receiving_from;
  491. r->sock = socket(PF_INET, SOCK_DGRAM, 0);
  492. if (r->sock < 0) {
  493. log_perror("play(): Could not create SOCK_DGRAM socket.", errno);
  494. FATAL_ERROR;
  495. }
  496. LOGf("CONN : Listening on multicast socket %s srv_fd: %i retries left: %i\n", r->channel->source, r->sock, retries);
  497. int on = 1;
  498. setsockopt(r->sock, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
  499. // subscribe to multicast group
  500. memcpy(&mreq.imr_multiaddr, &(r->src_sockname.sin_addr), sizeof(struct in_addr));
  501. mreq.imr_interface.s_addr = htonl(INADDR_ANY);
  502. if (setsockopt(r->sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq)) < 0) {
  503. LOGf("ERR : Failed to add IP membership on %s srv_fd: %i\n", r->channel->source, r->sock);
  504. FATAL_ERROR;
  505. }
  506. // bind to the socket so data can be read
  507. memset(&receiving_from, 0, sizeof(receiving_from));
  508. receiving_from.sin_family = AF_INET;
  509. receiving_from.sin_addr = r->src_sockname.sin_addr;
  510. receiving_from.sin_port = htons(src->port);
  511. if (bind(r->sock, (struct sockaddr *) &receiving_from, sizeof(receiving_from)) < 0) {
  512. LOGf("ERR : Failed to bind to %s srv_fd: %i\n", r->channel->source, r->sock);
  513. FATAL_ERROR;
  514. }
  515. }
  516. if (setsockopt(r->sock, SOL_SOCKET, SO_RCVBUF, (const char *)&readbuflen, sizeof(readbuflen)) < 0)
  517. log_perror("play(): setsockopt(SO_RCVBUF)", errno);
  518. proxy_set_status(r, "Connected");
  519. r->connected = 1;
  520. free_chansrc(src);
  521. return 0;
  522. }
  523. int check_restreamer_state(RESTREAMER *r) {
  524. if (r->dienow) {
  525. // LOGf("PROXY: Forced disconnect on srv_fd: %i | Channel: %s Source: %s\n", r->sock, r->channel->name, r->channel->source);
  526. proxy_set_status(r, "Dying");
  527. return 2;
  528. }
  529. if (r->reconnect) {
  530. LOGf("PROXY: Forced reconnect on srv_fd: %i | Channel: %s Source: %s\n", r->sock, r->channel->name, r->channel->source);
  531. proxy_set_status(r, "Forced reconnect");
  532. return 1;
  533. }
  534. return 0;
  535. }
  536. #define MAX_ZERO_READS 3
  537. /* Start: 3 seconds on connect */
  538. /* In connection: Max UDP timeout == 3 seconds (read) + 2 seconds (connect) == 5 seconds */
  539. #define UDP_READ_RETRIES 3
  540. #define UDP_READ_TIMEOUT 1000
  541. /* Start: 1/4 seconds on connect */
  542. /* In connection: Max TCP timeout == 5 seconds (read) + 2 seconds (connect) == 7 seconds */
  543. /* In connection: Max TCP timeout == 5 seconds (read) + 8 seconds (connect, host unrch) == 13 seconds */
  544. #define TCP_READ_RETRIES 5
  545. #define TCP_READ_TIMEOUT 1000
  546. /*
  547. Returns:
  548. 0 = synced ok
  549. 1 = not synced, reconnect
  550. */
  551. int mpeg_sync(RESTREAMER *r, int proxysock, char *channel, channel_source source_proto) {
  552. time_t sync_start = time(NULL);
  553. unsigned int sync_packets = 0;
  554. unsigned int read_bytes = 0;
  555. char syncframe[188];
  556. int _timeout = TCP_READ_TIMEOUT;
  557. int _retries = TCP_READ_RETRIES;
  558. if (source_proto == udp_sock) {
  559. _timeout = UDP_READ_TIMEOUT;
  560. _retries = UDP_READ_RETRIES;
  561. }
  562. do {
  563. resync:
  564. if (fdread_ex(proxysock, syncframe, 1, _timeout, _retries, 1) != 1) {
  565. LOGf("DEBUG: mpeg_sync fdread() timeout | Channel: %s\n", channel);
  566. proxy_set_status(r, "ERROR: fdread() timeout while syncing mpeg");
  567. return 1; // reconnect
  568. }
  569. // LOGf("DEBUG: Read 0x%02x Offset %u Sync: %u\n", (uint8_t)syncframe[0], read_bytes, sync_packets);
  570. read_bytes++;
  571. if (syncframe[0] == 0x47) {
  572. ssize_t rdsz = fdread_ex(proxysock, syncframe, 188-1, _timeout, _retries, 1);
  573. if (rdsz != 188-1) {
  574. LOGf("DEBUG: mpeg_sync fdread() timeout | Channel: %s\n", channel);
  575. proxy_set_status(r, "ERROR: fdread() timeout while syncing mpeg");
  576. return 1; // reconnect
  577. }
  578. read_bytes += 188-1;
  579. if (++sync_packets == 7) // sync 7 packets
  580. break;
  581. goto resync;
  582. } else {
  583. sync_packets = 0;
  584. }
  585. if (read_bytes > FRAME_PACKET_SIZE) { // Can't sync in 1316 bytes
  586. LOGf("DEBUG: Can't sync after %d bytes | Channel: %s\n", FRAME_PACKET_SIZE, channel);
  587. proxy_set_status(r, "ERROR: Can not sync mpeg");
  588. return 1; // reconnect
  589. }
  590. if (sync_start+2 <= time(NULL)) { // Do not sync in two seconds
  591. LOGf("DEBUG: Timeout while syncing (read %u bytes) | Channel: %s\n", read_bytes, channel);
  592. proxy_set_status(r, "ERROR: Timeout while syncing mpeg");
  593. return 1; // reconnect
  594. }
  595. } while (1);
  596. pthread_rwlock_wrlock(&r->lock);
  597. r->conn_ts = time(NULL);
  598. r->read_bytes = read_bytes;
  599. pthread_rwlock_unlock(&r->lock);
  600. LOGf("SYNC : TS synced after %u bytes | Channel: %s\n", read_bytes-FRAME_PACKET_SIZE, channel);
  601. proxy_set_status(r, "Synced");
  602. return 0;
  603. }
  604. char reset[FRAME_PACKET_SIZE] = {
  605. 0x47,0x40,0x00,0x10,0x00,0x00,0xB0,0x09,0x27,0x10,0xC1,0x00,
  606. 0x00,0x3C,0xDD,0xFF,0xB8,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  607. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  608. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  609. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  610. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  611. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  612. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  613. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  614. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  615. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  616. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  617. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  618. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  619. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  620. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0x47,0x40,0x00,0x11,
  621. 0x00,0x00,0xB0,0x0D,0x27,0x10,0xC3,0x00,0x00,0x4E,0x20,0xE0,
  622. 0x64,0xD8,0x46,0x8F,0xCB,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  623. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  624. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  625. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  626. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  627. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  628. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  629. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  630. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  631. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  632. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  633. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  634. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  635. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  636. 0xFF,0xFF,0xFF,0xFF,0x47,0x40,0x11,0x12,0x00,0x42,0xF0,0x0C,
  637. 0x27,0x10,0xC1,0x00,0x00,0x9C,0x40,0xFF,0x1F,0xA4,0x9D,0xBA,
  638. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  639. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  640. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  641. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  642. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  643. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  644. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  645. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  646. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  647. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  648. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  649. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  650. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  651. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  652. 0x47,0x40,0x11,0x13,0x00,0x42,0xF0,0x0C,0x27,0x10,0xC3,0x00,
  653. 0x00,0x9C,0x40,0xFF,0x29,0xF4,0x87,0x4A,0xFF,0xFF,0xFF,0xFF,
  654. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  655. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  656. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  657. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  658. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  659. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  660. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  661. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  662. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  663. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  664. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  665. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  666. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  667. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0x47,0x40,0x64,0x14,
  668. 0x00,0x02,0xB0,0x0D,0x4E,0x20,0xC1,0x00,0x00,0xE0,0x6E,0xF0,
  669. 0x00,0x30,0xB6,0x9F,0x1A,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  670. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  671. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  672. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  673. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  674. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  675. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  676. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  677. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  678. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  679. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  680. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  681. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  682. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  683. 0xFF,0xFF,0xFF,0xFF,0x47,0x40,0x64,0x15,0x00,0x02,0xB0,0x0D,
  684. 0x4E,0x20,0xC3,0x00,0x00,0xE0,0x78,0xF0,0x00,0xB7,0x41,0x6C,
  685. 0x5A,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  686. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  687. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  688. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  689. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  690. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  691. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  692. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  693. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  694. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  695. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  696. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  697. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  698. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  699. 0x47,0x1F,0xFF,0x10,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  700. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  701. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  702. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  703. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  704. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  705. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  706. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  707. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  708. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  709. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  710. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  711. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  712. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  713. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,
  714. 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF
  715. };
  716. void * proxy_ts_stream(void *self) {
  717. RESTREAMER *r = self;
  718. char buf[FRAME_PACKET_SIZE];
  719. signal(SIGPIPE, SIG_IGN);
  720. int http_code = 0;
  721. while (1) {
  722. r->conn_ts = 0;
  723. r->read_bytes = 0;
  724. int result = connect_source(self, 1, FRAME_PACKET_SIZE * 1000, &http_code);
  725. if (result > 0)
  726. goto RECONNECT;
  727. channel_source sproto = get_sproto(r->channel->source);
  728. int mpgsync = mpeg_sync(r, r->sock, r->channel->name, sproto);
  729. if (mpgsync == 1) // Timeout
  730. goto RECONNECT;
  731. ssize_t readen, written;
  732. int max_zero_reads = MAX_ZERO_READS;
  733. int send_reset = send_reset_opt;
  734. for (;;) {
  735. switch (check_restreamer_state(r)) {
  736. case 1: goto RECONNECT; // r->reconnect is on
  737. case 2: goto QUIT; // r->dienow is on
  738. }
  739. if (sproto == tcp_sock) {
  740. readen = fdread_ex(r->sock, buf, FRAME_PACKET_SIZE, TCP_READ_TIMEOUT, TCP_READ_RETRIES, 1);
  741. } else {
  742. readen = fdread_ex(r->sock, buf, FRAME_PACKET_SIZE, UDP_READ_TIMEOUT, UDP_READ_RETRIES, 0);
  743. }
  744. if (readen < 0)
  745. goto RECONNECT;
  746. if (readen == 0) { // ho, hum, wtf is going on here?
  747. LOGf("PROXY: zero read on srv_fd: %i | Channel: %s Source: %s\n", r->sock, r->channel->name, r->channel->source);
  748. if (--max_zero_reads == 0) {
  749. LOGf("PROXY: %d zero reads on srv_fd: %i | Channel: %s Source: %s\n", MAX_ZERO_READS, r->sock, r->channel->name, r->channel->source);
  750. proxy_set_status(r, "ERROR: Too many zero reads");
  751. break;
  752. }
  753. continue;
  754. }
  755. max_zero_reads = MAX_ZERO_READS;
  756. // Fill short frame with NULL packets
  757. if (readen < FRAME_PACKET_SIZE) {
  758. //LOGf("DEBUG: Short read (%d) on retreamer srv_fd: %i | Channel: %s\n", readen, sock, chan->name);
  759. memcpy(buf+readen, TS_NULL_FRAME+readen, FRAME_PACKET_SIZE - readen);
  760. }
  761. pthread_rwlock_wrlock(&r->lock);
  762. r->read_bytes += readen;
  763. pthread_rwlock_unlock(&r->lock);
  764. if (send_reset) {
  765. send_reset = 0;
  766. fdwrite(r->clientsock, reset, FRAME_PACKET_SIZE);
  767. }
  768. written = fdwrite(r->clientsock, buf, FRAME_PACKET_SIZE);
  769. if (written == -1) {
  770. LOGf("PROXY: Error writing to dst_fd: %i on srv_fd: %i | Channel: %s Source: %s\n", r->clientsock, r->sock, r->channel->name, r->channel->source);
  771. connect_destination(r);
  772. }
  773. }
  774. LOGf("DEBUG: fdread timeout restreamer srv_fd: %i | Channel: %s\n", r->sock, r->channel->name);
  775. proxy_set_status(r, "ERROR: Read timeout");
  776. RECONNECT:
  777. pthread_rwlock_wrlock(&r->lock);
  778. r->conn_ts = 0;
  779. pthread_rwlock_unlock(&r->lock);
  780. LOGf("DEBUG: reconnect srv_fd: %i | Channel: %s\n", r->sock, r->channel->name);
  781. proxy_set_status(r, "Reconnecting");
  782. shutdown_fd(&(r->sock));
  783. next_channel_source(r->channel);
  784. continue;
  785. QUIT:
  786. LOGf("DEBUG: quit srv_fd: %i | Channel: %s\n", r->sock, r->channel->name);
  787. break;
  788. }
  789. proxy_close(r);
  790. return 0;
  791. }
  792. static int copyright_shown = 0;
  793. void show_usage(int ident_only) {
  794. if (!copyright_shown) {
  795. printf("%s %s\n", server_sig, server_ver);
  796. puts(copyright);
  797. puts("");
  798. copyright_shown = 1;
  799. }
  800. if (ident_only)
  801. return;
  802. puts("Usage: tomcast -c config_file");
  803. puts("");
  804. puts("\t-c file\t\tChannels configuration file");
  805. puts("\t-i ident\tServer ident. Must be formated as PROVIDER/SERVER");
  806. puts("\t-d pidfile\tDaemonize and write daemon pid into pidfile");
  807. puts("\t-t ttl\t\tSet multicast ttl (default: 1)");
  808. puts("\t-o ip\t\tOutput interface address (default: 0.0.0.0)");
  809. puts("\t-l host\t\tSyslog host (default: disabled)");
  810. puts("\t-L port\t\tSyslog port (default: 514)");
  811. puts("\t-R\t\tSend reset packets when changing sources.");
  812. puts("Server settings:");
  813. puts("\t-b addr\t\tLocal IP address to bind. (default: 0.0.0.0)");
  814. puts("\t-p port\t\tPort to listen. (default: 0)");
  815. puts("");
  816. }
  817. void set_ident(char *new_ident, struct config *cfg) {
  818. cfg->ident = new_ident;
  819. cfg->logident = strdup(new_ident);
  820. char *c = cfg->logident;
  821. while (*c) {
  822. if (*c=='/')
  823. *c='-';
  824. c++;
  825. }
  826. }
  827. void parse_options(int argc, char **argv, struct config *cfg) {
  828. int j, ttl;
  829. cfg->server_socket = -1;
  830. pthread_mutex_init(&cfg->channels_lock, NULL);
  831. while ((j = getopt(argc, argv, "i:b:p:c:d:t:o:l:L:RHh")) != -1) {
  832. switch (j) {
  833. case 'b':
  834. cfg->server_addr = optarg;
  835. break;
  836. case 'p':
  837. cfg->server_port = atoi(optarg);
  838. break;
  839. case 'i':
  840. set_ident(optarg, cfg);
  841. break;
  842. case 'c':
  843. cfg->channels_file = optarg;
  844. break;
  845. case 'd':
  846. cfg->pidfile = optarg;
  847. break;
  848. case 'o':
  849. if (inet_aton(optarg, &output_intf) == 0) {
  850. fprintf(stderr, "Invalid interface address: %s\n", optarg);
  851. exit(1);
  852. }
  853. break;
  854. case 't':
  855. ttl = atoi(optarg);
  856. multicast_ttl = (ttl && ttl < 127) ? ttl : 1;
  857. break;
  858. case 'l':
  859. cfg->loghost = optarg;
  860. cfg->syslog_active = 1;
  861. break;
  862. case 'L':
  863. cfg->logport = atoi(optarg);
  864. break;
  865. case 'R':
  866. send_reset_opt = 1;
  867. break;
  868. case 'H':
  869. case 'h':
  870. show_usage(0);
  871. exit(0);
  872. break;
  873. }
  874. }
  875. if (!cfg->channels_file) {
  876. show_usage(0);
  877. fprintf(stderr, "ERROR: No channels file is set (use -c option).\n");
  878. exit(1);
  879. }
  880. if (!cfg->ident) {
  881. set_ident("unixsol/tomcast", cfg);
  882. }
  883. printf("Configuration:\n");
  884. printf("\tServer ident : %s\n", cfg->ident);
  885. printf("\tChannels file : %s\n", cfg->channels_file);
  886. printf("\tOutput iface addr : %s\n", inet_ntoa(output_intf));
  887. printf("\tMulticast ttl : %d\n", multicast_ttl);
  888. if (cfg->syslog_active) {
  889. printf("\tSyslog host : %s\n", cfg->loghost);
  890. printf("\tSyslog port : %d\n", cfg->logport);
  891. } else {
  892. printf("\tSyslog disabled.\n");
  893. }
  894. if (send_reset_opt)
  895. printf("\tSend reset packets.\n");
  896. if (cfg->pidfile) {
  897. printf("\tDaemonize : %s\n", cfg->pidfile);
  898. } else {
  899. printf("\tDo not daemonize.\n");
  900. }
  901. if (cfg->server_port) {
  902. init_server_socket(cfg->server_addr, cfg->server_port, &cfg->server, &cfg->server_socket);
  903. printf("\tStarting web srv : http://%s:%d/status (sock: %d)\n", cfg->server_addr, cfg->server_port, cfg->server_socket);
  904. } else {
  905. printf("\tNo web server\n");
  906. }
  907. }
  908. void init_vars(struct config *cfg) {
  909. cfg->restreamer = list_new("restreamer");
  910. regcomp(&http_response, "^HTTP/1.[0-1] (([0-9]{3}) .*)", REG_EXTENDED);
  911. memset(&TS_NULL_FRAME, 0xff, FRAME_PACKET_SIZE);
  912. int i;
  913. for (i=0; i<FRAME_PACKET_SIZE; i=i+188) {
  914. TS_NULL_FRAME[i+0] = 0x47;
  915. TS_NULL_FRAME[i+1] = 0x1f;
  916. TS_NULL_FRAME[i+2] = 0xff;
  917. TS_NULL_FRAME[i+3] = 0x00;
  918. }
  919. }
  920. void spawn_proxy_threads(struct config *cfg) {
  921. LNODE *lc, *lctmp;
  922. LNODE *lr, *lrtmp;
  923. int spawned = 0;
  924. list_for_each(cfg->chanconf, lc, lctmp) {
  925. CHANNEL *c = lc->data;
  926. int restreamer_active = 0;
  927. list_lock(cfg->restreamer);
  928. list_for_each(cfg->restreamer, lr, lrtmp) {
  929. RESTREAMER *r = lr->data;
  930. if (strcmp(r->name, c->name)==0) {
  931. restreamer_active = 1;
  932. break;
  933. }
  934. }
  935. list_unlock(cfg->restreamer);
  936. if (!restreamer_active) {
  937. RESTREAMER *nr = new_restreamer(c->name, c);
  938. if (nr->clientsock < 0) {
  939. LOGf("Error creating proxy socket for %s\n", c->name);
  940. free_restreamer(nr);
  941. } else {
  942. list_add(cfg->restreamer, nr);
  943. if (pthread_create(&nr->thread, NULL, &proxy_ts_stream, nr) == 0) {
  944. spawned++;
  945. pthread_detach(nr->thread);
  946. } else {
  947. LOGf("Error creating proxy for %s\n", c->name);
  948. }
  949. }
  950. }
  951. }
  952. LOGf("INFO : %d proxy threads spawned\n", spawned);
  953. }
  954. void kill_proxy_threads(struct config *cfg) {
  955. LNODE *l, *tmp;
  956. int killed = 0;
  957. list_lock(cfg->restreamer);
  958. list_for_each(cfg->restreamer, l, tmp) {
  959. RESTREAMER *r = l->data;
  960. r->dienow = 1;
  961. killed++;
  962. }
  963. list_unlock(cfg->restreamer);
  964. LOGf("INFO : %d proxy threads killed\n", killed);
  965. }
  966. int keep_going = 1;
  967. void signal_quit(int sig) {
  968. keep_going = 0;
  969. kill_proxy_threads(&config);
  970. usleep(500000);
  971. LOGf("KILL : Signal %i | %s %s (%s)\n", sig, server_sig, server_ver, config.ident);
  972. usleep(100000);
  973. log_close();
  974. if (config.pidfile && strlen(config.pidfile))
  975. unlink(config.pidfile);
  976. signal(sig, SIG_DFL);
  977. raise(sig);
  978. }
  979. struct config *get_config(void) {
  980. return &config;
  981. }
  982. void do_reconnect() {
  983. LNODE *l, *tmp;
  984. list_lock(config.restreamer);
  985. list_for_each(config.restreamer, l, tmp) {
  986. RESTREAMER *r = l->data;
  987. r->reconnect = 1;
  988. }
  989. list_unlock(config.restreamer);
  990. }
  991. void do_reconf() {
  992. load_channels_config(&config);
  993. spawn_proxy_threads(&config);
  994. }
  995. void init_signals() {
  996. signal(SIGCHLD, SIG_IGN);
  997. signal(SIGPIPE, SIG_IGN);
  998. signal(SIGHUP , do_reconf);
  999. signal(SIGUSR1, do_reconnect);
  1000. signal(SIGINT , signal_quit);
  1001. signal(SIGTERM, signal_quit);
  1002. }
  1003. void do_daemonize(struct config *cfg) {
  1004. if (!cfg->pidfile)
  1005. return;
  1006. fprintf(stderr, "Daemonizing.\n");
  1007. pid_t pid = fork();
  1008. if (pid > 0) {
  1009. FILE *F = fopen(cfg->pidfile,"w");
  1010. if (F) {
  1011. fprintf(F,"%i\n",pid);
  1012. fclose(F);
  1013. }
  1014. exit(0);
  1015. }
  1016. // Child process continues...
  1017. setsid(); // request a new session (job control)
  1018. freopen("/dev/null", "r", stdin);
  1019. freopen("/dev/null", "w", stdout);
  1020. freopen("/dev/null", "w", stderr);
  1021. }
  1022. /* Must be called after daemonize! */
  1023. void init_logger(struct config *cfg) {
  1024. if (cfg->syslog_active)
  1025. fprintf(stderr, "Logging to %s:%d\n", cfg->loghost, cfg->logport);
  1026. log_init(cfg->logident, cfg->syslog_active, cfg->pidfile == NULL, cfg->loghost, cfg->logport);
  1027. }
  1028. int main(int argc, char **argv) {
  1029. set_http_response_server_ident(server_sig, server_ver);
  1030. show_usage(1); // Show copyright and version
  1031. init_vars(&config);
  1032. parse_options(argc, argv, &config);
  1033. do_daemonize(&config);
  1034. init_logger(&config);
  1035. init_signals();
  1036. LOGf("INIT : %s %s (%s)\n" , server_sig, server_ver, config.ident);
  1037. load_channels_config(&config);
  1038. spawn_proxy_threads(&config);
  1039. web_server_start(&config);
  1040. do {
  1041. sleep(60);
  1042. } while(1);
  1043. signal_quit(15);
  1044. exit(0);
  1045. }