From f2a54dd00c3e5058bf4e48bde5c862b6affd866c Mon Sep 17 00:00:00 2001 From: Mike Brady Date: Mon, 22 Sep 2014 18:31:30 +0100 Subject: [PATCH] Send SIGUSR2 to shairport sync and it will send a pause request to the source. Also keep checking buffers. --- player.c | 73 ++++++++++++++++++++++++++++++++++++++++++++++------- rtp.c | 73 ++++++++++++++++++++++++++++++++++++++++++++++++----- rtp.h | 5 +++- rtsp.c | 12 ++++++++- shairport.c | 9 +++++++ 5 files changed, 155 insertions(+), 17 deletions(-) diff --git a/player.c b/player.c index 4d8e8a7d..f315f2ef 100644 --- a/player.c +++ b/player.c @@ -135,8 +135,10 @@ static uint64_t missing_packets,late_packets,too_late_packets,resend_requests; static void ab_resync(void) { int i; - for (i=0; i(BUFFER_FRAMES*7)/8) - debug(1,"An interval of %u frames has opened, with ab_read: %u, ab_write: %u and seqno: %u.",seq_diff(ab_read,seqno),ab_read,ab_write,seqno); + //if (ORDINATE(seqno)>(BUFFER_FRAMES*7)/8) + // debug(1,"An interval of %u frames has opened, with ab_read: %u, ab_write: %u and seqno: %u.",seq_diff(ab_read,seqno),ab_read,ab_write,seqno); int32_t gap = seq_diff(ab_write,PREDECESSOR(seqno))+1; if (gap<=0) debug(1,"Unexpected gap size: %d.",gap); @@ -390,9 +413,41 @@ static abuf_t *buffer_get_frame(void) { flush_requested=0; } pthread_mutex_unlock(&flush_mutex); - + uint32_t flush_limit = 0; if (ab_synced) { - + do { + curframe = audio_buffer + BUFIDX(ab_read); + if (curframe->ready) { + + if (curframe->sequence_number!=ab_read) { + // some kind of sync problem has occurred. + if (BUFIDX(curframe->sequence_number)==BUFIDX(ab_read)) { + // it looks like some kind of aliasing has happened + if (seq_order(ab_read,curframe->sequence_number)) { + ab_read=curframe->sequence_number; + debug(1,"Aliasing of buffer index -- reset."); + } + } else { + debug(1,"Inconsistent sequence numbers detected"); + } + } + + if ((flush_rtp_timestamp!=0x7fffffff) && ((curframe->timestamp==flush_rtp_timestamp) || seq32_order(curframe->timestamp,flush_rtp_timestamp))) { + debug(1,"Dropping flushed packet seqno %u, timestamp %u",curframe->sequence_number,curframe->timestamp); + curframe->ready=0; + flush_limit++; + ab_read==SUCCESSOR(ab_read); + } + if ((flush_rtp_timestamp!=0x7fffffff) && (!seq32_order(curframe->timestamp,flush_rtp_timestamp))) // if we have gone past the flush boundary time + flush_rtp_timestamp=0x7fffffff; + } + } while ((flush_rtp_timestamp!=0x7fffffff) && (flush_limit<=8820) && (curframe->ready!=0)); + + if (flush_limit==8820) { + debug(1,"Flush hit the 8820 frame limit!"); + flush_limit=0; + } + dac_delay = 0; curframe = audio_buffer + BUFIDX(ab_read); if (config.output->delay) { @@ -717,7 +772,7 @@ static void *player_thread_func(void *arg) { late_packet_message_sent=0; missing_packets=late_packets=too_late_packets=resend_requests=0; - flush_rtp_timestamp=0; + flush_rtp_timestamp=0x7fffffff; // it seems this number has a special significance -- it seems to be used as a null operand, so we'll use it like that too int sync_error_out_of_bounds = 0; // number of times in a row that there's been a serious sync error while (!please_stop) { abuf_t *inframe = buffer_get_frame(); @@ -955,8 +1010,8 @@ void player_flush(uint32_t timestamp) { // debug(1,"Flush requested up to %u. It seems as if 2147483647 is special.",timestamp); pthread_mutex_lock(&flush_mutex); flush_requested=1; - if (timestamp!=0x7fffffff) - flush_rtp_timestamp=timestamp; // flush all packets up to (and including?) this + //if (timestamp!=0x7fffffff) + flush_rtp_timestamp=timestamp; // flush all packets up to (and including?) this pthread_mutex_unlock(&flush_mutex); } diff --git a/rtp.c b/rtp.c index 9e34cc63..8b4378c0 100644 --- a/rtp.c +++ b/rtp.c @@ -56,6 +56,10 @@ typedef struct time_ping_record { static int running = 0; static int please_shutdown; +static char client_ip_string[INET6_ADDRSTRLEN]; // the ip string pointing to the client +static short client_ip_family; // AF_INET / AF_INET6 +static uint32_t client_active_remote; // used when you want to control the client... + static SOCKADDR rtp_client_control_socket; // a socket pointing to the control port of the client static SOCKADDR rtp_client_timing_socket; // a socket pointing to the timing port of the client static int audio_socket; // our local [server] audio socket @@ -421,18 +425,20 @@ static int bind_port(SOCKADDR *remote,int *sock, int desired_port ) { } -void rtp_setup(SOCKADDR *remote, int cport, int tport, int *lsport, int *lcport, int *ltport) { +void rtp_setup(SOCKADDR *remote, int cport, int tport, uint32_t active_remote, int *lsport, int *lcport, int *ltport) { if (running) die("rtp_setup called with active stream!"); debug(2, "rtp_setup: cport=%d tport=%d.", cport, tport); + client_active_remote = active_remote; + // print out what we know about the client void *addr; char *ipver; int port; - char ipstr[INET6_ADDRSTRLEN]; char portstr[20]; + client_ip_family = remote->SAFAMILY; // keep information about the kind of ip of the client #ifdef AF_INET6 if (remote->SAFAMILY == AF_INET6) { struct sockaddr_in6 *sa6 = (struct sockaddr_in6*)remote; @@ -447,8 +453,8 @@ void rtp_setup(SOCKADDR *remote, int cport, int tport, int *lsport, int *lcport, port = ntohs(sa4->sin_port); ipver = "IPv4"; } - inet_ntop(remote->SAFAMILY,addr,ipstr,sizeof(ipstr)); - debug(1,"Connection from %s: %s:%d",ipver,ipstr,port); + inet_ntop(remote->SAFAMILY,addr,client_ip_string,sizeof(client_ip_string)); // keep the client's ip number + debug(1,"Connection from %s: %s:%d",ipver,client_ip_string,port); @@ -461,7 +467,7 @@ void rtp_setup(SOCKADDR *remote, int cport, int tport, int *lsport, int *lcport, hints.ai_family = remote->SAFAMILY; hints.ai_socktype = SOCK_DGRAM; snprintf(portstr, 20, "%d", cport); - if (getaddrinfo(ipstr,portstr,&hints,&servinfo)!=0) + if (getaddrinfo(client_ip_string,portstr,&hints,&servinfo)!=0) die("Can't get address of client's control port"); #ifdef AF_INET6 @@ -478,7 +484,7 @@ void rtp_setup(SOCKADDR *remote, int cport, int tport, int *lsport, int *lcport, hints.ai_family = remote->SAFAMILY; hints.ai_socktype = SOCK_DGRAM; snprintf(portstr, 20, "%d", tport); - if (getaddrinfo(ipstr,portstr,&hints,&servinfo)!=0) + if (getaddrinfo(client_ip_string,portstr,&hints,&servinfo)!=0) die("Can't get address of client's timing port"); #ifdef AF_INET6 if (servinfo->ai_family == AF_INET6) @@ -566,3 +572,58 @@ void rtp_request_resend(seq_t first, uint32_t count) { } } } + +void rtp_request_client_pause() { + if (running) { + debug(1,"Send a request to pause to client at %s on port 3689 with active remote %u.",client_ip_string,client_active_remote); + + + struct addrinfo hints, *res; + int sockfd; + + char message[1000] , server_reply[2000]; + + // first, load up address structs with getaddrinfo(): + + memset(&hints, 0, sizeof hints); + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_STREAM; + + getaddrinfo(client_ip_string, "3689", &hints, &res); + + // make a socket: + + sockfd = socket(res->ai_family, res->ai_socktype, res->ai_protocol); + + if (sockfd == -1) { + die("Could not create socket"); + } + debug(1,"Socket created"); + + // connect! + + if (connect(sockfd, res->ai_addr, res->ai_addrlen) < 0) { + die("connect failed. Error"); + } + // debug(1,"Connect successful"); + + sprintf(message,"GET /ctrl-int/1/pause HTTP/1.1\r\nHost: %s\r\nActive-Remote: %u\r\n\r\n",client_ip_string,client_active_remote); + // debug(1,"Sending this message: \"%s\".",message); + + //Send some data + if( send(sockfd , message , strlen(message) , 0) < 0) { + debug(1,"Send failed"); + } + + //Receive a reply from the server + if( recv(sockfd , server_reply , 2000 , 0) < 0) { + debug(1,"recv failed"); + } + + // debug(1,"Server replied: \"%s\".",server_reply); + + close(sockfd); + } else { + debug(1,"Request to pause non-existent play stream -- ignored."); + } +} diff --git a/rtp.h b/rtp.h index 4e6189d5..5b679a60 100644 --- a/rtp.h +++ b/rtp.h @@ -3,9 +3,12 @@ #include -void rtp_setup(SOCKADDR *remote, int controlport, int timingport, int *local_server_port, int *local_control_port, int *local_timing_port); +#include "player.h" + +void rtp_setup(SOCKADDR *remote, int controlport, int timingport, uint32_t active_remote, int *local_server_port, int *local_control_port, int *local_timing_port); void rtp_shutdown(void); void rtp_request_resend(seq_t first, uint32_t count); +void rtp_request_client_pause(void); // ask the client to pause void get_reference_timestamp_stuff(uint32_t *timestamp,uint64_t *timestamp_time); void clear_reference_timestamp(void); diff --git a/rtsp.c b/rtsp.c index 67968788..652f7869 100644 --- a/rtsp.c +++ b/rtsp.c @@ -436,6 +436,16 @@ static void handle_setup(rtsp_conn_info *conn, rtsp_message *req, rtsp_message *resp) { int cport, tport; int lsport,lcport,ltport; + uint32_t active_remote=0; + + char * ar = msg_get_header(req,"Active-Remote"); + if (ar) { + debug(1,"Active-Remote string seen: \"%s\".",ar); + // get the active remote + char *p; + active_remote = strtoul(ar,&p,10); + debug(1,"Active Remote is %u.",active_remote); + } if (config.userSuppliedLatency==0) { char * ua = msg_get_header(req,"User-Agent"); @@ -473,7 +483,7 @@ static void handle_setup(rtsp_conn_info *conn, tport = atoi(p); rtsp_take_player(); - rtp_setup(&conn->remote, cport, tport,&lsport,&lcport,<port); + rtp_setup(&conn->remote, cport, tport, active_remote, &lsport,&lcport,<port); if (!lsport) return; char *q; diff --git a/shairport.c b/shairport.c index c6ac3f1a..5079cbd7 100644 --- a/shairport.c +++ b/shairport.c @@ -46,6 +46,7 @@ #include "common.h" #include "rtsp.h" +#include "rtp.h" #include "mdns.h" #include @@ -91,6 +92,10 @@ static void sig_logrotate(int foo, siginfo_t *bar, void *baz) { // log_setup(); } +static void sig_pause_client(int foo, siginfo_t *bar, void *baz) { + rtp_request_client_pause(); +} + void print_version(void) { char version_string[200]; strcpy(version_string,PACKAGE_VERSION); @@ -276,6 +281,7 @@ void signal_setup(void) { sigdelset(&set, SIGHUP); sigdelset(&set, SIGSTOP); sigdelset(&set, SIGCHLD); + sigdelset(&set, SIGUSR2); pthread_sigmask(SIG_BLOCK, &set, NULL); // setting this to SIG_IGN would prevent signalling any threads. @@ -293,6 +299,9 @@ void signal_setup(void) { sa.sa_sigaction = &sig_logrotate; sigaction(SIGHUP, &sa, NULL); + sa.sa_sigaction = &sig_pause_client; + sigaction(SIGUSR2, &sa, NULL); + sa.sa_sigaction = &sig_child; sigaction(SIGCHLD, &sa, NULL); }