[streaming] support sending ICY metadata (title) (#732)

* [streaming] obey Icy-MetaData directive and send icy title

* [streaming] fix 'streaming_sessions' race conditions/dbl frees on shutdown

* [streaming] simplify icy meta sending logic

* [streaming] 'icy request' variable cleanup

* [streaming] icy-meta comment/logging cleanup

src/httpd_streaming.c

@@ -28,6 +28,7 @@
 
 #include <uninorm.h>
 #include <unistd.h>
+#include <pthread.h>
 
 #include <event2/event.h>
 
@@ -37,6 +38,7 @@
 #include "transcode.h"
 #include "player.h"
 #include "listener.h"
+#include "db.h"
 
 /* httpd event base, from httpd.c */
 extern struct event_base *evbase_httpd;
@@ -56,7 +58,11 @@ extern struct event_base *evbase_httpd;
 struct streaming_session {
   struct evhttp_request *req;
   struct streaming_session *next;
+
+  bool     require_icy; // Client requested icy meta
+  size_t   bytes_sent;  // Audio bytes sent since last metablock
 };
+static pthread_mutex_t streaming_sessions_lck;
 static struct streaming_session *streaming_sessions;
 
 // Means we're not able to encode to mp3
@@ -78,17 +84,34 @@ static int streaming_player_changed;
 static int streaming_pipe[2];
 static int streaming_meta[2];
 
+#define STREAMING_ICY_METALEN_MAX 4080  // 255*16
+static const short STREAMING_ICY_METAINT = 8192;
+static unsigned streaming_icy_clients;
+static char *streaming_icy_title;
+
 
 static void
-streaming_fail_cb(struct evhttp_connection *evcon, void *arg)
+streaming_close_cb(struct evhttp_connection *evcon, void *arg)
 {
   struct streaming_session *this;
   struct streaming_session *session;
   struct streaming_session *prev;
+  char *address;
+  ev_uint16_t port;
 
   this = (struct streaming_session *)arg;
 
-  DPRINTF(E_WARN, L_STREAMING, "Connection failed; stopping mp3 streaming to client\n");
+  evhttp_connection_get_peer(evcon, &address, &port);
+  DPRINTF(E_INFO, L_STREAMING, "Stopping mp3 streaming to %s:%d\n", address, (int)port);
+
+  pthread_mutex_lock(&streaming_sessions_lck);
+  if (streaming_sessions == NULL)
+    {
+      // This close comes duing deinit() - we don't free `this` since it is
+      // already a dangling ptr (free'd in deinit()) at this stage
+      pthread_mutex_unlock(&streaming_sessions_lck);
+      return;
+    }
 
   prev = NULL;
   for (session = streaming_sessions; session; session = session->next)
@@ -101,8 +124,9 @@ streaming_fail_cb(struct evhttp_connection *evcon, void *arg)
 
   if (!session)
     {
-      DPRINTF(E_LOG, L_STREAMING, "Bug! Got a failure callback for an unknown stream\n");
+      DPRINTF(E_LOG, L_STREAMING, "Bug! Got a failure callback for an unknown stream (%s:%d)\n", address, (int)port);
       free(this);
+    pthread_mutex_unlock(&streaming_sessions_lck);
       return;
     }
 
@@ -111,6 +135,9 @@ streaming_fail_cb(struct evhttp_connection *evcon, void *arg)
   else
     prev->next = session->next;
 
+  if (session->require_icy)
+    --streaming_icy_clients;
+
   free(session);
 
   if (!streaming_sessions)
@@ -119,6 +146,7 @@ streaming_fail_cb(struct evhttp_connection *evcon, void *arg)
       event_del(streamingev);
       event_del(metaev);
     }
+    pthread_mutex_unlock(&streaming_sessions_lck);
 }
 
 static void
@@ -126,17 +154,25 @@ streaming_end(void)
 {
   struct streaming_session *session;
   struct evhttp_connection *evcon;
+  char *address;
+  ev_uint16_t port;
 
+  pthread_mutex_lock(&streaming_sessions_lck);
   for (session = streaming_sessions; streaming_sessions; session = streaming_sessions)
     {
       evcon = evhttp_request_get_connection(session->req);
       if (evcon)
-	evhttp_connection_set_closecb(evcon, NULL, NULL);
+	{
+	  evhttp_connection_set_closecb(evcon, NULL, NULL);
+	  evhttp_connection_get_peer(evcon, &address, &port);
+	  DPRINTF(E_INFO, L_STREAMING, "Force close stream to %s:%d\n", address, (int)port);
+	}
       evhttp_send_reply_end(session->req);
 
       streaming_sessions = session->next;
       free(session);
     }
+  pthread_mutex_unlock(&streaming_sessions_lck);
 
   event_del(streamingev);
   event_del(metaev);
@@ -221,6 +257,115 @@ encode_buffer(uint8_t *buffer, size_t size)
   return ret;
 }
 
+/* We know that the icymeta is limited to 1+255*16 (ie 4081) bytes so caller must
+ * provide a buf of this size to avoid needless mallocs
+ *
+ * The icy meta block is defined by a single byte indicating how many double byte
+ * words used for the actual meta.  Unused bytes are null padded
+ *
+ * https://stackoverflow.com/questions/4911062/pulling-track-info-from-an-audio-stream-using-php/4914538#4914538
+ * http://www.smackfu.com/stuff/programming/shoutcast.html
+ */
+static uint8_t *
+streaming_icy_meta_create(uint8_t buf[STREAMING_ICY_METALEN_MAX+1], const char *title, unsigned *buflen)
+{
+  unsigned titlelen = 0;
+  unsigned metalen = 0;
+  uint8_t no16s;
+
+  *buflen = 0;
+
+  if (title == NULL)
+    {
+      no16s = 0;
+      memcpy(buf, &no16s, 1);
+
+      *buflen = 1;
+    }
+  else
+    {
+      titlelen = strlen(title);
+      if (titlelen > STREAMING_ICY_METALEN_MAX)
+	titlelen = STREAMING_ICY_METALEN_MAX;  // dont worry about the null byte
+
+      // [0]    1x byte N, indicate the total number of 16 bytes words required
+      //        to represent the meta data
+      // [1..N] meta data book ended by "StreamTitle='" and "';"
+      //
+      // The '15' is strlen of StreamTitle=' + ';
+      no16s = (15 + titlelen)/16 +1;
+      metalen = 1 + no16s*16;
+      memset(buf, 0, metalen);
+
+      memcpy(buf,             &no16s, 1);
+      memcpy(buf+1,           (const uint8_t*)"StreamTitle='", 13);
+      memcpy(buf+14,          title, titlelen);
+      memcpy(buf+14+titlelen, (const uint8_t*)"';", 2);
+
+      *buflen = metalen;
+    }
+
+  return buf;
+}
+
+static uint8_t *
+streaming_icy_meta_splice(const uint8_t *data, size_t datalen, off_t offset, size_t *len)
+{
+  uint8_t  meta[STREAMING_ICY_METALEN_MAX+1];  // Buffer, of max sz, for the created icymeta
+  unsigned metalen;     // How much of the buffer is in use
+  uint8_t *buf;         // Client returned buffer; contains the audio (from data) spliced w/meta (from meta)
+
+  if (data == NULL || datalen == 0)
+    return NULL;
+
+  memset(meta, 0, sizeof(meta));
+  streaming_icy_meta_create(meta, streaming_icy_title, &metalen);
+
+  *len = datalen + metalen;
+  // DPRINTF(E_DBG, L_STREAMING, "splicing meta, audio block=%d bytes, offset=%d, metalen=%d new buflen=%d\n", datalen, offset, metalen, *len);
+  buf = malloc(*len);
+  memcpy(buf,                data, offset);
+  memcpy(buf+offset,         &meta[0], metalen);
+  memcpy(buf+offset+metalen, data+offset, datalen-offset);
+
+  return buf;
+}
+
+static void
+streaming_player_status_update()
+{
+  unsigned x, y;
+  struct db_queue_item *queue_item = NULL;
+  struct player_status  tmp;
+
+  tmp.id = streaming_player_status.id;
+  player_get_status(&streaming_player_status);
+
+  if (tmp.id != streaming_player_status.id && streaming_icy_clients)
+    {
+      free(streaming_icy_title);
+      if ( (queue_item = db_queue_fetch_byfileid(streaming_player_status.id)) == NULL)
+	{
+	  streaming_icy_title = NULL;
+	}
+      else
+	{
+	  x = strlen(queue_item->title);
+	  y = strlen(queue_item->artist);
+	  if (x && y)
+	    {
+	      streaming_icy_title = malloc(x+y+4);
+	      snprintf(streaming_icy_title, x+y+4, "%s - %s", queue_item->title, queue_item->artist);
+	    }
+	  else
+	    {
+	      streaming_icy_title = strdup( x ? queue_item->title : queue_item->artist);
+	    }
+	  free_queue_item(queue_item, 0);
+	}
+    }
+}
+
 static void
 streaming_send_cb(evutil_socket_t fd, short event, void *arg)
 {
@@ -228,6 +373,10 @@ streaming_send_cb(evutil_socket_t fd, short event, void *arg)
   struct evbuffer *evbuf;
   uint8_t rawbuf[STREAMING_READ_SIZE];
   uint8_t *buf;
+  uint8_t *splice_buf = NULL;
+  size_t splice_len;
+  size_t count;
+  int overflow;
   int len;
   int ret;
 
@@ -240,6 +389,12 @@ streaming_send_cb(evutil_socket_t fd, short event, void *arg)
 	  if (ret <= 0)
 	    break;
 
+	  if (streaming_player_changed)
+	    {
+	      streaming_player_changed = 0;
+	      streaming_player_status_update();
+	    }
+
 	  ret = encode_buffer(rawbuf, ret);
 	  if (ret < 0)
 	    return;
@@ -251,7 +406,7 @@ streaming_send_cb(evutil_socket_t fd, short event, void *arg)
       if (streaming_player_changed)
 	{
 	  streaming_player_changed = 0;
-	  player_get_status(&streaming_player_status);
+	  streaming_player_status_update();
 	}
 
       if (streaming_player_status.status != PLAY_PAUSED)
@@ -269,17 +424,52 @@ streaming_send_cb(evutil_socket_t fd, short event, void *arg)
 
   // Send data
   evbuf = evbuffer_new();
+  pthread_mutex_lock(&streaming_sessions_lck);
   for (session = streaming_sessions; session; session = session->next)
     {
-      if (session->next)
+      // Does this session want ICY meta data and is it time to send?
+      count = session->bytes_sent+len;
+      if (session->require_icy && count > STREAMING_ICY_METAINT)
 	{
+	  overflow = count%STREAMING_ICY_METAINT;
 	  buf = evbuffer_pullup(streaming_encoded_data, -1);
-	  evbuffer_add(evbuf, buf, len);
+
+	  // DPRINTF(E_DBG, L_STREAMING, "session=%x sent=%ld len=%ld overflow=%ld\n", session, session->bytes_sent, len, overflow);
+
+	  // Splice the 'icy title' in with the encoded audio data
+	  splice_len = 0;
+	  splice_buf = streaming_icy_meta_splice(buf, len, len-overflow, &splice_len);
+
+	  evbuffer_add(evbuf, splice_buf, splice_len);
+
+	  free(splice_buf);
+	  splice_buf = NULL;
+
 	  evhttp_send_reply_chunk(session->req, evbuf);
+
+	  if (session->next == NULL)
+	    {
+	      // We're the last session, drop the contents of the encoded buffer
+	      evbuffer_drain(streaming_encoded_data, len);
+	    }
+	  session->bytes_sent = overflow;
 	}
       else
-	evhttp_send_reply_chunk(session->req, streaming_encoded_data);
+	{
+	  if (session->next)
+	    {
+	      buf = evbuffer_pullup(streaming_encoded_data, -1);
+	      evbuffer_add(evbuf, buf, len);
+	      evhttp_send_reply_chunk(session->req, evbuf);
+	    }
+	  else
+	    {
+	      evhttp_send_reply_chunk(session->req, streaming_encoded_data);
+	    }
+	  session->bytes_sent += len;
+	}
     }
+  pthread_mutex_unlock(&streaming_sessions_lck);
 
   evbuffer_free(evbuf);
 }
@@ -297,6 +487,7 @@ streaming_write(struct output_buffer *obuf)
 {
   int ret;
 
+  // Explicit no-lock - let the write to pipes fail if during deinit
   if (!streaming_sessions)
     return;
 
@@ -305,7 +496,10 @@ streaming_write(struct output_buffer *obuf)
       ret = write(streaming_meta[1], &obuf->data[0].quality, sizeof(struct media_quality));
       if (ret < 0)
 	{
-	  DPRINTF(E_LOG, L_STREAMING, "Error writing to streaming pipe: %s\n", strerror(errno));
+	  if (errno == EBADF)
+	    DPRINTF(E_LOG, L_STREAMING, "streaming pipe already closed\n");
+	  else
+	    DPRINTF(E_LOG, L_STREAMING, "Error writing to streaming pipe: %s\n", strerror(errno));
 	  return;
 	}
     }
@@ -316,7 +510,12 @@ streaming_write(struct output_buffer *obuf)
       if (errno == EAGAIN)
 	DPRINTF(E_WARN, L_STREAMING, "Streaming pipe full, skipping write\n");
       else
-	DPRINTF(E_LOG, L_STREAMING, "Error writing to streaming pipe: %s\n", strerror(errno));
+	{
+	  if (errno == EBADF)
+	    DPRINTF(E_LOG, L_STREAMING, "Streaming pipe already closed\n");
+	  else
+	    DPRINTF(E_LOG, L_STREAMING, "Error writing to streaming pipe: %s\n", strerror(errno));
+	}
     }
 }
 
@@ -330,6 +529,9 @@ streaming_request(struct evhttp_request *req, struct httpd_uri_parsed *uri_parse
   const char *name;
   char *address;
   ev_uint16_t port;
+  const char *param;
+  bool require_icy = false;
+  char buf[9];
 
   if (streaming_not_supported)
     {
@@ -341,8 +543,11 @@ streaming_request(struct evhttp_request *req, struct httpd_uri_parsed *uri_parse
 
   evcon = evhttp_request_get_connection(req);
   evhttp_connection_get_peer(evcon, &address, &port);
+  param = evhttp_find_header( evhttp_request_get_input_headers(req), "Icy-MetaData");
+  if (param && strcmp(param, "1") == 0)
+    require_icy = true;
 
-  DPRINTF(E_INFO, L_STREAMING, "Beginning mp3 streaming to %s:%d\n", address, (int)port);
+  DPRINTF(E_INFO, L_STREAMING, "Beginning mp3 streaming (with icy=%d) to %s:%d\n", require_icy, address, (int)port);
 
   lib = cfg_getsec(cfg, "library");
   name = cfg_getstr(lib, "name");
@@ -353,14 +558,19 @@ streaming_request(struct evhttp_request *req, struct httpd_uri_parsed *uri_parse
   evhttp_add_header(output_headers, "Cache-Control", "no-cache");
   evhttp_add_header(output_headers, "Pragma", "no-cache");
   evhttp_add_header(output_headers, "Expires", "Mon, 31 Aug 2015 06:00:00 GMT");
-  evhttp_add_header(output_headers, "icy-name", name);
+  if (require_icy)
+    {
+      ++streaming_icy_clients;
+      evhttp_add_header(output_headers, "icy-name", name);
+      snprintf(buf, sizeof(buf)-1, "%d", STREAMING_ICY_METAINT);
+      evhttp_add_header(output_headers, "icy-metaint", buf);
+    }
   evhttp_add_header(output_headers, "Access-Control-Allow-Origin", "*");
   evhttp_add_header(output_headers, "Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS");
 
-  // TODO ICY metaint
   evhttp_send_reply_start(req, HTTP_OK, "OK");
 
-  session = malloc(sizeof(struct streaming_session));
+  session = calloc(1, sizeof(struct streaming_session));
   if (!session)
     {
       DPRINTF(E_LOG, L_STREAMING, "Out of memory for streaming request\n");
@@ -369,6 +579,8 @@ streaming_request(struct evhttp_request *req, struct httpd_uri_parsed *uri_parse
       return -1;
     }
 
+  pthread_mutex_lock(&streaming_sessions_lck);
+
   if (!streaming_sessions)
     {
       event_add(streamingev, &streaming_silence_tv);
@@ -377,9 +589,13 @@ streaming_request(struct evhttp_request *req, struct httpd_uri_parsed *uri_parse
 
   session->req = req;
   session->next = streaming_sessions;
+  session->require_icy = require_icy;
+  session->bytes_sent = 0;
   streaming_sessions = session;
 
-  evhttp_connection_set_closecb(evcon, streaming_fail_cb, session);
+  pthread_mutex_unlock(&streaming_sessions_lck);
+
+  evhttp_connection_set_closecb(evcon, streaming_close_cb, session);
 
   return 0;
 }
@@ -401,6 +617,8 @@ streaming_init(void)
 {
   int ret;
 
+  pthread_mutex_init(&streaming_sessions_lck, NULL);
+
   // Non-blocking because otherwise httpd and player thread may deadlock
 #ifdef HAVE_PIPE2
   ret = pipe2(streaming_pipe, O_CLOEXEC | O_NONBLOCK);
@@ -448,6 +666,9 @@ streaming_init(void)
   CHECK_NULL(L_STREAMING, streamingev = event_new(evbase_httpd, streaming_pipe[0], EV_TIMEOUT | EV_READ | EV_PERSIST, streaming_send_cb, NULL));
   CHECK_NULL(L_STREAMING, metaev = event_new(evbase_httpd, streaming_meta[0], EV_READ | EV_PERSIST, streaming_meta_cb, NULL));
 
+  streaming_icy_clients = 0;
+  streaming_icy_title = NULL;
+
   return 0;
 
  error:
@@ -462,22 +683,10 @@ streaming_init(void)
 void
 streaming_deinit(void)
 {
-  struct streaming_session *session;
+  streaming_end();
-  struct streaming_session *next;
-
-  session = streaming_sessions;
-  streaming_sessions = NULL; // Stops writing and sending
-
-  next = NULL;
-  while (session)
-    {
-      evhttp_send_reply_end(session->req);
-      next = session->next;
-      free(session);
-      session = next;
-    }
 
   event_free(streamingev);
+  streamingev = NULL;
 
   listener_remove(player_change_cb);
 
@@ -488,4 +697,7 @@ streaming_deinit(void)
 
   transcode_encode_cleanup(&streaming_encode_ctx);
   evbuffer_free(streaming_encoded_data);
+  free(streaming_icy_title);
+
+  pthread_mutex_destroy(&streaming_sessions_lck);
 }