2
0

stream.c 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768
  1. /*
  2. * Squeezelite - lightweight headless squeezebox emulator
  3. *
  4. * (c) Adrian Smith 2012-2015, triode1@btinternet.com
  5. * Ralph Irving 2015-2017, ralph_irving@hotmail.com
  6. *
  7. * This program is free software: you can redistribute it and/or modify
  8. * it under the terms of the GNU General Public License as published by
  9. * the Free Software Foundation, either version 3 of the License, or
  10. * (at your option) any later version.
  11. *
  12. * This program is distributed in the hope that it will be useful,
  13. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  15. * GNU General Public License for more details.
  16. *
  17. * You should have received a copy of the GNU General Public License
  18. * along with this program. If not, see <http://www.gnu.org/licenses/>.
  19. *
  20. */
  21. // stream thread
  22. #ifndef _GNU_SOURCE
  23. #define _GNU_SOURCE
  24. #endif
  25. #include "squeezelite.h"
  26. #include <fcntl.h>
  27. #if USE_SSL
  28. #include "openssl/ssl.h"
  29. #include "openssl/err.h"
  30. #endif
  31. #if SUN
  32. #include <signal.h>
  33. #endif
  34. static log_level loglevel;
  35. static struct buffer buf;
  36. struct buffer *streambuf = &buf;
  37. #define LOCK mutex_lock(streambuf->mutex)
  38. #define UNLOCK mutex_unlock(streambuf->mutex)
  39. /*
  40. When LMS sends a close/open sequence very quickly, the stream thread might
  41. still be waiting in the poll() on the closed socket. It is never recommended
  42. to have a thread closing a socket used by another thread but it works, as
  43. opposed to an infinite select().
  44. In stream_sock() a new socket is created and full OS will allocate a different
  45. one but on RTOS and simple IP stack, the same might be re-used and that causes
  46. an exception as a thread is already waiting on a newly allocated socket
  47. A simple variable that forces stream_sock() to wait until we are out of poll()
  48. is enough and much faster than a mutex
  49. */
  50. static bool polling;
  51. static sockfd fd;
  52. struct EXT_RAM_ATTR streamstate stream;
  53. static EXT_RAM_ATTR struct {
  54. bool flac;
  55. u64_t serial;
  56. enum { OGG_OFF, OGG_SYNC, OGG_HEADER, OGG_SEGMENTS, OGG_PAGE } state;
  57. size_t want, miss, match;
  58. u8_t* data, segments[255];
  59. #pragma pack(push, 1)
  60. struct {
  61. char pattern[4];
  62. u8_t version, type;
  63. u64_t granule;
  64. u32_t serial, page, checksum;
  65. u8_t count;
  66. } header;
  67. #pragma pack(pop)
  68. } ogg;
  69. #if USE_SSL
  70. static SSL_CTX *SSLctx;
  71. SSL *ssl;
  72. #endif
  73. #if !USE_SSL
  74. #define _recv(ssl, fc, buf, n, opt) recv(fd, buf, n, opt)
  75. #define _send(ssl, fd, buf, n, opt) send(fd, buf, n, opt)
  76. #define _poll(ssl, pollinfo, timeout) poll(pollinfo, 1, timeout)
  77. #define _last_error() last_error()
  78. #else
  79. #define _last_error() ERROR_WOULDBLOCK
  80. static int _recv(SSL *ssl, int fd, void *buffer, size_t bytes, int options) {
  81. int n;
  82. if (!ssl) return recv(fd, buffer, bytes, options);
  83. n = SSL_read(ssl, (u8_t*) buffer, bytes);
  84. if (n <= 0 && SSL_get_error(ssl, n) == SSL_ERROR_ZERO_RETURN) return 0;
  85. return n;
  86. }
  87. static int _send(SSL *ssl, int fd, void *buffer, size_t bytes, int options) {
  88. int n;
  89. if (!ssl) return send(fd, buffer, bytes, options);
  90. while (1) {
  91. int err;
  92. ERR_clear_error();
  93. if ((n = SSL_write(ssl, (u8_t*) buffer, bytes)) >= 0) return n;
  94. err = SSL_get_error(ssl, n);
  95. if (err == SSL_ERROR_WANT_READ || err == SSL_ERROR_WANT_WRITE) continue;
  96. LOG_INFO("SSL write error %d", err );
  97. return n;
  98. }
  99. }
  100. /*
  101. can't mimic exactly poll as SSL is a real pain. Even if SSL_pending returns
  102. 0, there might be bytes to read but when select (poll) return > 0, there might
  103. be no frame available. As well select (poll) < 0 does not mean that there is
  104. no data pending
  105. */
  106. static int _poll(SSL *ssl, struct pollfd *pollinfo, int timeout) {
  107. if (!ssl) return poll(pollinfo, 1, timeout);
  108. if (pollinfo->events & POLLIN && SSL_pending(ssl)) {
  109. if (pollinfo->events & POLLOUT) poll(pollinfo, 1, 0);
  110. pollinfo->revents = POLLIN;
  111. return 1;
  112. }
  113. return poll(pollinfo, 1, timeout);
  114. }
  115. #endif
  116. static bool send_header(void) {
  117. char *ptr = stream.header;
  118. int len = stream.header_len;
  119. unsigned try = 0;
  120. ssize_t n;
  121. while (len) {
  122. n = _send(ssl, fd, ptr, len, MSG_NOSIGNAL);
  123. if (n <= 0) {
  124. if (n < 0 && _last_error() == ERROR_WOULDBLOCK && try < 10) {
  125. LOG_SDEBUG("retrying (%d) writing to socket", ++try);
  126. usleep(1000);
  127. continue;
  128. }
  129. LOG_INFO("failed writing to socket: %s", strerror(last_error()));
  130. stream.disconnect = LOCAL_DISCONNECT;
  131. stream.state = DISCONNECT;
  132. wake_controller();
  133. return false;
  134. }
  135. LOG_SDEBUG("wrote %d bytes to socket", n);
  136. ptr += n;
  137. len -= n;
  138. }
  139. LOG_SDEBUG("wrote header");
  140. return true;
  141. }
  142. static bool running = true;
  143. static void _disconnect(stream_state state, disconnect_code disconnect) {
  144. stream.state = state;
  145. stream.disconnect = disconnect;
  146. if (ogg.state == OGG_PAGE && ogg.data) free(ogg.data);
  147. ogg.data = NULL;
  148. #if USE_SSL
  149. if (ssl) {
  150. SSL_shutdown(ssl);
  151. SSL_free(ssl);
  152. ssl = NULL;
  153. }
  154. #endif
  155. closesocket(fd);
  156. fd = -1;
  157. wake_controller();
  158. }
  159. static size_t memfind(const u8_t* haystack, size_t n, const char* needle, size_t len, size_t* offset) {
  160. size_t i;
  161. for (i = 0; i < n && *offset != len; i++) *offset = (haystack[i] == needle[*offset]) ? *offset + 1 : 0;
  162. return i;
  163. }
  164. /* https://xiph.org/ogg/doc/framing.html
  165. * https://xiph.org/flac/ogg_mapping.html
  166. * https://xiph.org/vorbis/doc/Vorbis_I_spec.html#x1-610004.2 */
  167. static void stream_ogg(size_t n) {
  168. if (ogg.state == OGG_OFF) return;
  169. u8_t* p = streambuf->writep;
  170. while (n) {
  171. size_t consumed = min(ogg.miss, n);
  172. // copy as many bytes as possible and come back later if we do'nt have enough
  173. if (ogg.data) {
  174. memcpy(ogg.data + ogg.want - ogg.miss, p, consumed);
  175. ogg.miss -= consumed;
  176. if (ogg.miss) return;
  177. }
  178. // we have what we want, let's parse
  179. switch (ogg.state) {
  180. case OGG_SYNC: {
  181. ogg.miss -= consumed;
  182. if (consumed) break;
  183. // we have to memorize position in case any of last 3 bytes match...
  184. size_t pos = memfind(p, n, "OggS", 4, &ogg.match);
  185. if (ogg.match == 4) {
  186. consumed = pos - ogg.match;
  187. ogg.state = OGG_HEADER;
  188. ogg.miss = ogg.want = sizeof(ogg.header);
  189. ogg.data = (u8_t*) &ogg.header;
  190. ogg.match = 0;
  191. } else {
  192. if (!ogg.match) LOG_INFO("OggS not at expected position %zu/%zu", pos, n);
  193. LOG_INFO("OggS not at expected position %zu/%zu", pos, n);
  194. return;
  195. }
  196. break;
  197. }
  198. case OGG_HEADER:
  199. if (!memcmp(ogg.header.pattern, "OggS", 4)) {
  200. ogg.miss = ogg.want = ogg.header.count;
  201. ogg.data = ogg.segments;
  202. ogg.state = OGG_SEGMENTS;
  203. } else {
  204. ogg.state = OGG_SYNC;
  205. ogg.data = NULL;
  206. }
  207. break;
  208. case OGG_SEGMENTS:
  209. // calculate size of page using lacing values
  210. for (size_t i = 0; i < ogg.want; i++) ogg.miss += ogg.data[i];
  211. ogg.want = ogg.miss;
  212. // acquire serial number when we are looking for headers and hit a bos
  213. if (ogg.serial == ULLONG_MAX && (ogg.header.type & 0x02)) ogg.serial = ogg.header.serial;
  214. // we have overshot and missed header, reset serial number to restart search (O and -1 are le/be)
  215. if (ogg.header.serial == ogg.serial && ogg.header.granule && ogg.header.granule != -1) ogg.serial = ULLONG_MAX;
  216. // not our serial (the above protected us from granule > 0)
  217. if (ogg.header.serial != ogg.serial) {
  218. // otherwise, jump over data
  219. ogg.state = OGG_SYNC;
  220. ogg.data = NULL;
  221. } else {
  222. ogg.state = OGG_PAGE;
  223. ogg.data = malloc(ogg.want);
  224. }
  225. break;
  226. case OGG_PAGE: {
  227. char** tag = (char* []){ "\x3vorbis", "OpusTags", NULL };
  228. size_t ofs = 0;
  229. /* with OggFlac, we need the next page (packet) - VorbisComment is wrapped into a FLAC_METADATA
  230. * and except with vorbis, comment packet starts a new page but even in vorbis, it won't span
  231. * accross multiple pages */
  232. if (ogg.flac) ofs = 4;
  233. else if (!memcmp(ogg.data, "\x7f""FLAC", 5)) ogg.flac = true;
  234. else for (size_t n = 0; *tag; tag++, ofs = 0) if ((ofs = memfind(ogg.data, ogg.want, *tag, strlen(*tag), &n)) && n == strlen(*tag)) break;
  235. if (ofs) {
  236. // u32:len,char[]:vendorId, u32:N, N x (u32:len,char[]:comment)
  237. char* p = (char*) ogg.data + ofs;
  238. p += *p + 4;
  239. u32_t count = *p;
  240. p += 4;
  241. // LMS metadata format for Ogg is "Ogg", N x (u16:len,char[]:comment)
  242. memcpy(stream.header, "Ogg", 3);
  243. stream.header_len = 3;
  244. for (u32_t len; count--; p += len) {
  245. len = *p;
  246. p += 4;
  247. // only report what we use and don't overflow (network byte order)
  248. if (!strncasecmp(p, "TITLE=", 6) || !strncasecmp(p, "ARTIST=", 7) || !strncasecmp(p, "ALBUM=", 6)) {
  249. if (stream.header_len + len > MAX_HEADER) break;
  250. stream.header[stream.header_len++] = len >> 8;
  251. stream.header[stream.header_len++] = len;
  252. memcpy(stream.header + stream.header_len, p, len);
  253. stream.header_len += len;
  254. LOG_INFO("metadata: %.*s", len, p);
  255. }
  256. }
  257. ogg.flac = false;
  258. ogg.serial = ULLONG_MAX;
  259. stream.meta_send = true;
  260. wake_controller();
  261. LOG_INFO("Ogg metadata length: %u", stream.header_len - 3);
  262. }
  263. free(ogg.data);
  264. ogg.data = NULL;
  265. ogg.state = OGG_SYNC;
  266. break;
  267. }
  268. default:
  269. break;
  270. }
  271. p += consumed;
  272. n -= consumed;
  273. }
  274. }
  275. static void *stream_thread() {
  276. while (running) {
  277. struct pollfd pollinfo;
  278. size_t space;
  279. LOCK;
  280. space = min(_buf_space(streambuf), _buf_cont_write(streambuf));
  281. if (fd < 0 || !space || stream.state <= STREAMING_WAIT) {
  282. UNLOCK;
  283. usleep(space ? 100000 : 25000);
  284. continue;
  285. }
  286. if (stream.state == STREAMING_FILE) {
  287. int n = read(fd, streambuf->writep, space);
  288. if (n == 0) {
  289. LOG_INFO("end of stream");
  290. _disconnect(DISCONNECT, DISCONNECT_OK);
  291. }
  292. if (n > 0) {
  293. _buf_inc_writep(streambuf, n);
  294. stream.bytes += n;
  295. LOG_SDEBUG("streambuf read %d bytes", n);
  296. }
  297. if (n < 0) {
  298. LOG_WARN("error reading: %s", strerror(last_error()));
  299. _disconnect(DISCONNECT, REMOTE_DISCONNECT);
  300. }
  301. UNLOCK;
  302. continue;
  303. } else {
  304. pollinfo.fd = fd;
  305. pollinfo.events = POLLIN;
  306. if (stream.state == SEND_HEADERS) {
  307. pollinfo.events |= POLLOUT;
  308. }
  309. }
  310. UNLOCK;
  311. // no mutex needed - we just want to know if we are inside poll()
  312. polling = true;
  313. if (_poll(ssl, &pollinfo, 100)) {
  314. polling = false;
  315. LOCK;
  316. // check socket has not been closed while in poll
  317. if (fd < 0) {
  318. UNLOCK;
  319. continue;
  320. }
  321. if ((pollinfo.revents & POLLOUT) && stream.state == SEND_HEADERS) {
  322. if (send_header()) stream.state = RECV_HEADERS;
  323. stream.header_len = 0;
  324. UNLOCK;
  325. continue;
  326. }
  327. if (pollinfo.revents & (POLLIN | POLLHUP)) {
  328. // get response headers
  329. if (stream.state == RECV_HEADERS) {
  330. // read one byte at a time to catch end of header
  331. char c;
  332. static int endtok;
  333. int n = _recv(ssl, fd, &c, 1, 0);
  334. if (n <= 0) {
  335. if (n < 0 && _last_error() == ERROR_WOULDBLOCK) {
  336. UNLOCK;
  337. continue;
  338. }
  339. LOG_INFO("error reading headers: %s", n ? strerror(last_error()) : "closed");
  340. _disconnect(STOPPED, LOCAL_DISCONNECT);
  341. UNLOCK;
  342. continue;
  343. }
  344. *(stream.header + stream.header_len) = c;
  345. stream.header_len++;
  346. if (stream.header_len > MAX_HEADER - 1) {
  347. LOG_ERROR("received headers too long: %u", stream.header_len);
  348. _disconnect(DISCONNECT, LOCAL_DISCONNECT);
  349. }
  350. if (stream.header_len > 1 && (c == '\r' || c == '\n')) {
  351. endtok++;
  352. if (endtok == 4) {
  353. *(stream.header + stream.header_len) = '\0';
  354. LOG_INFO("headers: len: %d\n%s", stream.header_len, stream.header);
  355. stream.state = stream.cont_wait ? STREAMING_WAIT : STREAMING_BUFFERING;
  356. wake_controller();
  357. }
  358. } else {
  359. endtok = 0;
  360. }
  361. UNLOCK;
  362. continue;
  363. }
  364. // receive icy meta data
  365. if (stream.meta_interval && stream.meta_next == 0) {
  366. if (stream.meta_left == 0) {
  367. // read meta length
  368. u8_t c;
  369. int n = _recv(ssl, fd, &c, 1, 0);
  370. if (n <= 0) {
  371. if (n < 0 && _last_error() == ERROR_WOULDBLOCK) {
  372. UNLOCK;
  373. continue;
  374. }
  375. LOG_INFO("error reading icy meta: %s", n ? strerror(last_error()) : "closed");
  376. _disconnect(STOPPED, LOCAL_DISCONNECT);
  377. UNLOCK;
  378. continue;
  379. }
  380. stream.meta_left = 16 * c;
  381. stream.header_len = 0; // amount of received meta data
  382. // MAX_HEADER must be more than meta max of 16 * 255
  383. }
  384. if (stream.meta_left) {
  385. int n = _recv(ssl, fd, stream.header + stream.header_len, stream.meta_left, 0);
  386. if (n <= 0) {
  387. if (n < 0 && _last_error() == ERROR_WOULDBLOCK) {
  388. UNLOCK;
  389. continue;
  390. }
  391. LOG_INFO("error reading icy meta: %s", n ? strerror(last_error()) : "closed");
  392. _disconnect(STOPPED, LOCAL_DISCONNECT);
  393. UNLOCK;
  394. continue;
  395. }
  396. stream.meta_left -= n;
  397. stream.header_len += n;
  398. }
  399. if (stream.meta_left == 0) {
  400. if (stream.header_len) {
  401. *(stream.header + stream.header_len) = '\0';
  402. LOG_INFO("icy meta: len: %u\n%s", stream.header_len, stream.header);
  403. stream.meta_send = true;
  404. wake_controller();
  405. }
  406. stream.meta_next = stream.meta_interval;
  407. UNLOCK;
  408. continue;
  409. }
  410. // stream body into streambuf
  411. } else {
  412. int n;
  413. space = min(_buf_space(streambuf), _buf_cont_write(streambuf));
  414. if (stream.meta_interval) {
  415. space = min(space, stream.meta_next);
  416. }
  417. n = _recv(ssl, fd, streambuf->writep, space, 0);
  418. if (n == 0) {
  419. LOG_INFO("end of stream (%u bytes)", stream.bytes);
  420. _disconnect(DISCONNECT, DISCONNECT_OK);
  421. }
  422. if (n < 0 && _last_error() != ERROR_WOULDBLOCK) {
  423. LOG_INFO("error reading: %s", strerror(last_error()));
  424. _disconnect(DISCONNECT, REMOTE_DISCONNECT);
  425. }
  426. if (n > 0) {
  427. stream_ogg(n);
  428. _buf_inc_writep(streambuf, n);
  429. stream.bytes += n;
  430. if (stream.meta_interval) {
  431. stream.meta_next -= n;
  432. }
  433. } else {
  434. UNLOCK;
  435. continue;
  436. }
  437. if (stream.state == STREAMING_BUFFERING && stream.bytes > stream.threshold) {
  438. stream.state = STREAMING_HTTP;
  439. wake_controller();
  440. }
  441. LOG_SDEBUG("streambuf read %d bytes", n);
  442. }
  443. }
  444. UNLOCK;
  445. } else {
  446. polling = false;
  447. LOG_SDEBUG("poll timeout");
  448. }
  449. }
  450. #if USE_SSL
  451. if (SSLctx) {
  452. SSL_CTX_free(SSLctx);
  453. }
  454. #endif
  455. return 0;
  456. }
  457. static thread_type thread;
  458. void stream_init(log_level level, unsigned stream_buf_size) {
  459. loglevel = level;
  460. LOG_INFO("init stream");
  461. LOG_DEBUG("streambuf size: %u", stream_buf_size);
  462. buf_init(streambuf, stream_buf_size);
  463. if (streambuf->buf == NULL) {
  464. LOG_ERROR("unable to malloc buffer");
  465. exit(2);
  466. }
  467. #if USE_SSL
  468. #if !LINKALL && !NO_SSLSYM
  469. if (ssl_loaded) {
  470. #endif
  471. SSL_library_init();
  472. SSLctx = SSL_CTX_new(SSLv23_client_method());
  473. if (SSLctx == NULL) {
  474. LOG_ERROR("unable to allocate SSL context");
  475. exit(3);
  476. }
  477. SSL_CTX_set_options(SSLctx, SSL_OP_NO_SSLv2);
  478. #if !LINKALL && !NO_SSLSYM
  479. }
  480. #endif
  481. ssl = NULL;
  482. #endif
  483. #if SUN
  484. signal(SIGPIPE, SIG_IGN); /* Force sockets to return -1 with EPIPE on pipe signal */
  485. #endif
  486. stream.state = STOPPED;
  487. stream.header = malloc(MAX_HEADER);
  488. *stream.header = '\0';
  489. fd = -1;
  490. #if LINUX || FREEBSD
  491. touch_memory(streambuf->buf, streambuf->size);
  492. #endif
  493. #if LINUX || OSX || FREEBSD || EMBEDDED
  494. pthread_attr_t attr;
  495. pthread_attr_init(&attr);
  496. #ifdef PTHREAD_STACK_MIN
  497. pthread_attr_setstacksize(&attr, PTHREAD_STACK_MIN + STREAM_THREAD_STACK_SIZE);
  498. #endif
  499. pthread_create_name(&thread, &attr, stream_thread, NULL, "stream");
  500. pthread_attr_destroy(&attr);
  501. #endif
  502. #if WIN
  503. thread = CreateThread(NULL, STREAM_THREAD_STACK_SIZE, (LPTHREAD_START_ROUTINE)&stream_thread, NULL, 0, NULL);
  504. #endif
  505. }
  506. void stream_close(void) {
  507. LOG_INFO("close stream");
  508. LOCK;
  509. running = false;
  510. UNLOCK;
  511. #if LINUX || OSX || FREEBSD || EMBEDDED
  512. pthread_join(thread, NULL);
  513. #endif
  514. free(stream.header);
  515. buf_destroy(streambuf);
  516. }
  517. void stream_file(const char *header, size_t header_len, unsigned threshold) {
  518. buf_flush(streambuf);
  519. LOCK;
  520. stream.header_len = header_len;
  521. memcpy(stream.header, header, header_len);
  522. *(stream.header+header_len) = '\0';
  523. LOG_INFO("opening local file: %s", stream.header);
  524. #if WIN
  525. fd = open(stream.header, O_RDONLY | O_BINARY);
  526. #else
  527. fd = open(stream.header, O_RDONLY);
  528. #endif
  529. stream.state = STREAMING_FILE;
  530. if (fd < 0) {
  531. LOG_INFO("can't open file: %s", stream.header);
  532. stream.state = DISCONNECT;
  533. }
  534. wake_controller();
  535. stream.cont_wait = false;
  536. stream.meta_interval = 0;
  537. stream.meta_next = 0;
  538. stream.meta_left = 0;
  539. stream.meta_send = false;
  540. stream.sent_headers = false;
  541. stream.bytes = 0;
  542. stream.threshold = threshold;
  543. UNLOCK;
  544. }
  545. void stream_sock(u32_t ip, u16_t port, bool use_ssl, bool use_ogg, const char *header, size_t header_len, unsigned threshold, bool cont_wait) {
  546. struct sockaddr_in addr;
  547. #if EMBEDDED
  548. // wait till we are not polling anymore
  549. while (polling && running) { usleep(10000); }
  550. #endif
  551. int sock = socket(AF_INET, SOCK_STREAM, 0);
  552. if (sock < 0) {
  553. LOG_ERROR("failed to create socket");
  554. return;
  555. }
  556. memset(&addr, 0, sizeof(addr));
  557. addr.sin_family = AF_INET;
  558. addr.sin_addr.s_addr = ip;
  559. addr.sin_port = port;
  560. LOG_INFO("connecting to %s:%d", inet_ntoa(addr.sin_addr), ntohs(addr.sin_port));
  561. set_nonblock(sock);
  562. set_nosigpipe(sock);
  563. if (connect_timeout(sock, (struct sockaddr *) &addr, sizeof(addr), 10) < 0) {
  564. LOG_INFO("unable to connect to server");
  565. LOCK;
  566. stream.state = DISCONNECT;
  567. stream.disconnect = UNREACHABLE;
  568. UNLOCK;
  569. return;
  570. }
  571. #if USE_SSL
  572. if (ntohs(port) == 443) {
  573. char server[256], *p;
  574. ssl = SSL_new(SSLctx);
  575. SSL_set_fd(ssl, sock);
  576. // add SNI
  577. sscanf(header, "Host:%255s", server);
  578. if (server) {
  579. if ((p = strchr(server, ':')) != NULL) *p = '\0';
  580. SSL_set_tlsext_host_name(ssl, server);
  581. }
  582. while (1) {
  583. int status, err = 0;
  584. ERR_clear_error();
  585. status = SSL_connect(ssl);
  586. // successful negotiation
  587. if (status == 1) break;
  588. // error or non-blocking requires more time
  589. if (status < 0) {
  590. err = SSL_get_error(ssl, status);
  591. if (err == SSL_ERROR_WANT_READ || err == SSL_ERROR_WANT_WRITE) continue;
  592. }
  593. LOG_WARN("unable to open SSL socket %d (%d)", status, err);
  594. closesocket(sock);
  595. SSL_free(ssl);
  596. ssl = NULL;
  597. LOCK;
  598. stream.state = DISCONNECT;
  599. stream.disconnect = UNREACHABLE;
  600. UNLOCK;
  601. return;
  602. }
  603. } else {
  604. ssl = NULL;
  605. }
  606. #endif
  607. buf_flush(streambuf);
  608. LOCK;
  609. fd = sock;
  610. stream.state = SEND_HEADERS;
  611. stream.cont_wait = cont_wait;
  612. stream.meta_interval = 0;
  613. stream.meta_next = 0;
  614. stream.meta_left = 0;
  615. stream.meta_send = false;
  616. stream.header_len = header_len;
  617. memcpy(stream.header, header, header_len);
  618. *(stream.header+header_len) = '\0';
  619. LOG_INFO("header: %s", stream.header);
  620. stream.sent_headers = false;
  621. stream.bytes = 0;
  622. stream.threshold = threshold;
  623. ogg.miss = ogg.match = 0;
  624. ogg.state = use_ogg ? OGG_SYNC : OGG_OFF;
  625. ogg.flac = false;
  626. ogg.serial = ULLONG_MAX;
  627. UNLOCK;
  628. }
  629. bool stream_disconnect(void) {
  630. bool disc = false;
  631. LOCK;
  632. #if USE_SSL
  633. if (ssl) {
  634. SSL_shutdown(ssl);
  635. SSL_free(ssl);
  636. ssl = NULL;
  637. }
  638. #endif
  639. if (fd != -1) {
  640. closesocket(fd);
  641. fd = -1;
  642. disc = true;
  643. }
  644. stream.state = STOPPED;
  645. if (ogg.state == OGG_PAGE && ogg.data) free(ogg.data);
  646. ogg.data = NULL;
  647. UNLOCK;
  648. return disc;
  649. }