code cleanup and network debug (WIP)

1 files changed, 387 insertions, 306 deletions

diff --git a/src/hdsync_cli.cpp b/src/hdsync_cli.cpp
index 88109a8..915ced3 100644
--- a/src/hdsync_cli.cpp
+++ b/src/hdsync_cli.cpp
@@ -27,7 +27,7 @@
 #include <sys/stat.h>
 #include <sys/types.h>
 #include <sys/socket.h>
-#include <netdb.h>
+//#include <netdb.h>
 
 #include <errno.h>
 
@@ -39,6 +39,7 @@
 #include <avremote.h>
 #include <parsers.h>
 #include <zmq.h>
+#include <zmq_utils.h>
 
 // our exit codes are shell style: 1 is error, 0 is success
 #define ERR 1
@@ -72,13 +73,396 @@ char IPv4[24];
 // 0mq
 void *zcontext;
 void *sock_in = NULL;
-void *sock_out = NULL;
+void *sock_shout = NULL;
 char pgmaddr[64];
 zmq_msg_t request;
 
 // we use only getopt, no _long
 static const char *short_options = "-hvts:p:";
 
+void cmdline(int argc, char **argv);
+
+
+int get_address();
+
+
+int shout(const char *message, int flags) {
+ int res;
+ // zmq_connect(sock_out, pgmaddr);
+ zmq_msg_init_size (&request, strlen(message));
+ memcpy (zmq_msg_data (&request), message, strlen(message));
+ printf ("Shouting \"%s\"\n", message);
+ res = zmq_send (sock_shout, &request, flags);
+ zmq_msg_close (&request);
+ // zmq_close(sock_out);
+ return(res);
+}
+
+// extract a string from a message received:
+// PREFIX;MESSAGE;
+// message should be a char of 512 bytes 
+int expect(const char *prefix, char *message, int flags) {
+ zmq_msg_t reply;
+ char res[512];
+ char *p, *pp;
+ int pfxlen = strlen(prefix);
+
+ printf("Expecting prefix %s\n",prefix);
+
+ // listen for answers
+ zmq_msg_init (&reply);
+ zmq_recv (sock_in, &reply, flags);
+ // printf ("Received response\n");
+ snprintf(res,511,"%s",zmq_msg_data(&reply));
+ zmq_msg_close (&reply);
+ 
+ // parse the response and if ACK insert it in uniq list
+ if( strncmp(res,prefix,pfxlen) == 0) {
+ // quick and secure string parsing
+ p = res; while(*p != ';') p++; p++;
+ pp = p; while(*pp != ';') pp++; *pp='\0';
+ 
+ // *p has the ip string
+ snprintf(message,511,"%s",p);
+ printf("Expected prefix %s found: %s\n",prefix,message);
+ return(0);
+ }
+ 
+ return(1);
+}
+
+int handshake() {
+ int request_nbr;
+ char message[256];
+ int listindex = 0;
+ int listmax = 0;
+ char res[512];
+ int c, i;
+ 
+ if(chanID==1) { // first channel is the one to offer
+ 
+ ////////
+ // OFFER
+ //////////////////////////////
+ printf ("Offering on %s from address %s\n",pgmaddr, IPv4);
+ 
+ listmax = chanTOT; // number of desired peers
+ listmax--; // we count outrselves in
+ printf("looking to connect %u peers\n",listmax);
+ // allocate space for listener array
+ listeners = (char**) calloc(listmax,sizeof(char*));
+ for(i=0; i<listmax; i++) listeners[i] = (char*)calloc(16, sizeof(char));
+ 
+ // offer subscribes to all broadcasts by default
+ // somehow listening to ACK only doesn't works
+ snprintf(message,255,"OFFER;%s;",IPv4);
+ while(1) {
+ 
+ shout(message, 0);
+
+ if( expect("ACK",res, 0) == 0) {
+ printf("Got ACK from %s\n",res);
+ // parse the response and if ACK insert it in uniq list
+ for(c=0; c<listindex; c++)
+ if(strncmp(listeners[c], res, 16) == 0)
+ break; // found a duplicate
+ if(c==listindex) { // there was no duplicate
+ snprintf(listeners[c], 16, "%s", res);
+ printf("New listener: %s\n",listeners[c]);
+ listindex++;
+ }
+ 
+ if(listindex==listmax) { // goal reached
+ printf("Sending ready signals to listeners\n");
+ ///////////////////////////////////////////////
+ shout("READY", 0);
+ break; // break offer loop
+ } // if listmax reached
+ 
+ } // if ACK response
+
+ zmq_sleep(1);
+
+ } // offer while loop
+ 
+ } else { // all other channels listen
+
+ /////////
+ // LISTEN
+
+ printf ("Listening on %s from address %s\n",pgmaddr, IPv4);
+
+ snprintf(message,255,"ACK;%s;",IPv4);
+
+ while (1) {
+ // Wait for offer
+ while( expect("OFFER", res, 0) !=0); // blocking
+ // reply with ACK
+ shout(message, 0);
+ 
+ while( expect("READY", res, 0) !=0); // blocking
+ printf ("Received ready signal\n");
+ break; // quit loop on ready signal
+ 
+ }
+
+ }
+ 
+ return(1);
+}
+
+int main(int argc, char **argv) {
+ upnp_t *upnp;
+ int found;
+
+ //////////
+ ///// INIT
+
+ cmdline(argc, argv);
+
+ if(!filename[0]) {
+ fprintf(stderr,"not enough args specified on commandline, see help.");
+ exit(ERR);
+ }
+
+ fprintf(stderr,"HDSync starting for channel %u of %u\n",chanID, chanTOT);
+ fprintf(stderr,"will sync and play video: %s\n", filename);
+
+ upnp = create_upnp();
+ 
+ // no server specified, force localhost
+ if(!server[0]) sprintf(server,"localhost");
+
+ // commandline or detection found explicit addresses
+ snprintf(upnp->hostname, MAX_HOSTNAME_SIZE-1,"%s",server);
+ upnp->port = port;
+
+ if(!dryrun) {
+ if ( connect_upnp (upnp) < 0 ) {
+ fprintf(stderr,"can't connect to %s:%u: operation aborted.\n", server, port);
+ free_upnp(upnp);
+ exit(ERR);
+ } else {
+ fprintf(stderr,"UPNP connected to %s:%u\n",server, port);
+ close(upnp->sockfd);
+ upnp->sockfd = 0;
+ }
+ }
+
+ if( ! get_address() ) {
+ fprintf(stderr,"Error getting own address.\n");
+ exit(ERR);
+ }
+
+ // initialize 0mq
+ zcontext = zmq_init (1);
+
+ // create broadcast socket
+ sock_shout = zmq_socket (zcontext, ZMQ_PUB);
+ zmq_connect(sock_shout, pgmaddr);
+
+ // create input socket
+ sock_in = zmq_socket (zcontext, ZMQ_SUB);
+ zmq_bind (sock_in, pgmaddr);
+ zmq_setsockopt(sock_in,ZMQ_SUBSCRIBE,"",0);
+
+
+ /// END OF INIT
+ //////////////// 
+
+
+ ////////////////////
+ if(! handshake()) {
+ // no error return so far, just running endless untill contact is established
+ fprintf(stderr,"error establishing contact with all desired channels, aborting.\n");
+ exit(ERR);
+ }
+
+ printf("Handshake successful\n");
+
+ //////////////////////
+ if(dryrun) { // DRYRUN
+ printf("Dry run\n");
+ 
+ // testing sync with no real playback
+ zmq_sleep(1);
+
+ if(chanID==1) { // offer should send sync
+
+ printf("Prepare to offer sync\n");
+ 
+ zmq_sleep(3);
+
+ shout("SYNC;GOGO;", 0);
+
+ } else { // listener should prepare to receive sync
+
+ printf("Prepare to receive sync\n");
+
+ // TODO
+ zmq_sleep(6);
+ }
+ 
+ //////////////////////
+
+ } else { // RUN FOR REAL
+
+ //////////////////////
+ 
+ // load, play and pause in sequence
+ // break and reopen connection in between
+ 
+
+ if(chanID==1) { // offer should send sync
+ zmq_sleep(1);
+ // create output socket
+ // sock_out = zmq_socket (zcontext, ZMQ_PUB);
+ // zmq_connect(sock_out, pgmaddr);
+ // prepare signal
+ zmq_msg_init_size (&request, 4);
+ memcpy (zmq_msg_data (&request), "SYNC", 4);
+
+ // stop player
+ connect_upnp(upnp);
+ render_upnp(upnp,"Stop","");
+ send_upnp(upnp);
+ recv_upnp(upnp, 1000);
+ close(upnp->sockfd);
+ upnp->sockfd = 0;
+
+
+ zmq_sleep(1);
+
+
+ // prepare player in position
+ render_uri_meta(upnp,filename);
+ render_upnp(upnp,"SetAVTransportURI", upnp->meta);
+ send_upnp(upnp);
+ recv_upnp(upnp, 1000);
+ close(upnp->sockfd);
+ upnp->sockfd = 0;
+
+ zmq_sleep(1);
+
+ // play
+ connect_upnp(upnp);
+ render_upnp(upnp,"Play","<Speed>1</Speed>");
+ send_upnp(upnp);
+ recv_upnp(upnp, 1000);
+ close(upnp->sockfd);
+ upnp->sockfd = 0;
+ // pause
+ connect_upnp(upnp);
+ render_upnp(upnp,"Pause","");
+ send_upnp(upnp);
+ recv_upnp(upnp, 1000);
+ close(upnp->sockfd);
+ upnp->sockfd = 0;
+
+ zmq_sleep(1);
+ 
+ // prepare upnp play message
+ connect_upnp(upnp);
+ render_upnp(upnp,"Play","<Speed>1</Speed>");
+
+ // send sync signal
+ // zmq_send (sock_out, &request, 0);
+
+ // start player
+ send_upnp(upnp);
+ recv_upnp(upnp, 1000);
+
+ } else { // this is a listener
+
+ // create input socket
+ sock_in = zmq_socket (zcontext, ZMQ_SUB);
+ zmq_bind (sock_in, pgmaddr); 
+
+
+ // stop player
+ connect_upnp(upnp);
+ render_upnp(upnp,"Stop","");
+ send_upnp(upnp);
+ recv_upnp(upnp, 1000);
+ close(upnp->sockfd);
+ upnp->sockfd = 0;
+
+
+ zmq_sleep(1);
+
+
+ // prepare player in position
+ render_uri_meta(upnp,filename);
+ render_upnp(upnp,"SetAVTransportURI", upnp->meta);
+ send_upnp(upnp);
+ recv_upnp(upnp, 1000);
+ close(upnp->sockfd);
+ upnp->sockfd = 0;
+
+ zmq_sleep(1);
+
+ // play
+ connect_upnp(upnp);
+ render_upnp(upnp,"Play","<Speed>1</Speed>");
+ send_upnp(upnp);
+ recv_upnp(upnp, 1000);
+ close(upnp->sockfd);
+ upnp->sockfd = 0;
+ // pause
+ connect_upnp(upnp);
+ render_upnp(upnp,"Pause","");
+ send_upnp(upnp);
+ recv_upnp(upnp, 1000);
+ close(upnp->sockfd);
+ upnp->sockfd = 0;
+
+ zmq_sleep(1);
+ 
+ // prepare upnp play message
+ connect_upnp(upnp);
+ render_upnp(upnp,"Play","<Speed>1</Speed>");
+ 
+ // Wait for next request from client
+ zmq_msg_init (&request);
+ zmq_recv (sock_in, &request, 0);
+ if(strncmp((char*)zmq_msg_data(&request),"SYNC",4)==0) {
+ // start player
+ send_upnp(upnp);
+ recv_upnp(upnp, 1000);
+ }
+
+ } // if(offer|listen)
+
+ zmq_msg_close (&request);
+ free_upnp(upnp);
+
+ } // if(test|real)
+
+ fprintf(stderr,"Software shutdown\n");
+
+ // if(sock_out) {
+ // zmq_close (sock_out);
+ // sock_out = NULL;
+ // }
+ // if(sock_in) {
+ // zmq_close (sock_in);
+ // sock_in = NULL;
+ // }
+
+ // somehow offer blocks here
+ zmq_term (zcontext);
+
+ if(listeners) {
+ for(int c; c<chanTOT; c++)
+ free(listeners[c]);
+ free(listeners);
+ }
+ 
+ fprintf(stderr,"Done.\n");
+ exit(0);
+}
+
+
 void cmdline(int argc, char **argv) {
 filename[0] = 0;
 server[0] = 0;
@@ -156,14 +540,7 @@ void cmdline(int argc, char **argv) {
 } while(res != -1);
 }
 
-int handshake() {
- int request_nbr;
- zmq_msg_t reply;
- char message[256];
- int listindex = 0;
- int listmax = 0;
- char res[128];
- 
+int get_address() {
 // discover own IP address
 struct ifaddrs * ifAddrStruct=NULL;
 struct ifaddrs * ifa=NULL;
@@ -194,301 +571,5 @@ int handshake() {
 printf("Error: ethernet interface %s is not configured\n",ETHDEV);
 return(0);
 }
-
-
- // create output socket
- sock_out = zmq_socket (zcontext, ZMQ_PUB);
- zmq_connect(sock_out, pgmaddr);
-
- // create input socket
- sock_in = zmq_socket (zcontext, ZMQ_SUB);
- zmq_bind (sock_in, pgmaddr);
-
- if(chanID==1) { // first channel is the one to offer
- int i;
-
- ////////
- // OFFER
- //////////////////////////////
- printf ("Offering on %s from address %s\n",pgmaddr, IPv4);
-
- listmax = chanTOT; // number of desired peers
- listmax--; // we count outrselves in
- printf("looking to connect %u peers\n",listmax);
- // allocate space for listener array
- listeners = (char**) calloc(listmax,sizeof(char*));
- for(i=0; i<listmax; i++) listeners[i] = (char*)calloc(16, sizeof(char));
- 
- // offer subscribes to all broadcasts by default
- // somehow listening to ACK only doesn't works
- zmq_setsockopt(sock_in,ZMQ_SUBSCRIBE,"",0);
- snprintf(message,255,"OFFER;%s;",IPv4);
- while(1) {
- 
- // broadcast offer
- zmq_msg_init_size (&request, strlen(message));
- memcpy (zmq_msg_data (&request), message, strlen(message));
- // printf ("Sending offer...\n", request_nbr);
- zmq_send (sock_out, &request, 0);
- zmq_msg_close (&request);
- 
- sleep(1);
- 
- // listen for answers
- zmq_msg_init (&reply);
- zmq_recv (sock_in, &reply, 0);
- // printf ("Received response\n");
- snprintf(res,127,"%s",zmq_msg_data(&reply));
- zmq_msg_close (&reply);
- // parse the response and if ACK insert it in uniq list
- if( strncmp(res,"ACK",3) == 0) {
- int c;
- char *p, *pp;
- // quick and secure string parsing
- p = res; while(*p != ';') p++; p++;
- pp = p; while(*pp != ';') pp++; *pp='\0';
-
- // *p has the ip string
- for(c=0; c<listindex; c++)
- if(strncmp(listeners[c], p, 16) == 0)
- break; // found a duplicate
- if(c==listindex) { // there was no duplicate
- snprintf(listeners[c], 16, "%s", p);
- printf("New listener: %s\n",p);
- listindex++;
- }
- if(listindex==listmax) { // goal reached
- printf("Sending ready signals to listeners:\n");
- ///////////////////////////////////////////////
- for(c=0;c<listmax;c++) {
- char tmpaddr[256];
- snprintf(tmpaddr,255, "tcp://%s:%s",listeners[c],PGMPORT);
- printf (" %s\n", tmpaddr);
- zmq_connect(sock_out, tmpaddr);
- zmq_msg_init_size (&request, 5);
- memcpy (zmq_msg_data (&request), "READY", 5);
- zmq_send (sock_out, &request, 0);
- zmq_msg_close (&request);
- }
- break; // break offer loop
- } // if listmax reached
- 
- } // if ACK response
-
- } // offer while loop
-
- } else { // all other channels listen
-
- /////////
- // LISTEN
-
-
- printf ("Listening on %s from address %s\n",pgmaddr, IPv4);
-
- // listen subscribes only to offers and ready
- zmq_setsockopt(sock_in,ZMQ_SUBSCRIBE,"OFFER",5);
- zmq_setsockopt(sock_in,ZMQ_SUBSCRIBE,"READY",5);
-
-
- snprintf(message,255,"ACK;%s;",IPv4);
-
- while (1) {
- // Wait for next request from client
- zmq_msg_init (&request);
- zmq_recv (sock_in, &request, 0);
-
- snprintf(res,127,"%s",zmq_msg_data(&request));
- zmq_msg_close (&request);
- if(strncmp(res,"READY",5) == 0) {
- printf ("Received ready signal\n");
- break; // quit loop on ready signal
- }
- sleep (1);
-
- // Send reply back to client
- zmq_msg_init_size (&reply, strlen(message));
- memcpy (zmq_msg_data (&reply), message, strlen(message));
- zmq_send (sock_out, &reply, 0);
- zmq_msg_close (&reply);
-
- }
-
- }
-
- if(sock_out) {
- zmq_close (sock_out);
- sock_out = NULL;
- }
- if(sock_in) {
- zmq_close (sock_in);
- sock_in = NULL;
- }
- 
 return(1);
 }
-
-int main(int argc, char **argv) {
- upnp_t *upnp;
- int found;
-
- cmdline(argc, argv);
-
- if(!filename[0]) {
- fprintf(stderr,"not enough args specified on commandline, see help.");
- exit(ERR);
- }
-
- fprintf(stderr,"HDSync starting for channel %u of %u\n",chanID, chanTOT);
- fprintf(stderr,"will sync and play video: %s\n", filename);
-
- upnp = create_upnp();
- 
- // no server specified, force localhost
- if(!server[0]) sprintf(server,"localhost");
-
- // commandline or detection found explicit addresses
- snprintf(upnp->hostname, MAX_HOSTNAME_SIZE-1,"%s",server);
- upnp->port = port;
-
- if(!dryrun) {
- if ( connect_upnp (upnp) < 0 )
- {
- fprintf(stderr,"can't connect to %s:%u: operation aborted.\n", server, port);
- free_upnp(upnp);
- exit(ERR);
- }
- }
-
- // initialize 0mq
- zcontext = zmq_init (1);
- 
- if(! handshake()) {
- // might never get here since while(1) tries endlessly
- fprintf(stderr,"error establishing contact with all desired channels, aborting.\n");
- exit(ERR);
- }
-
- //////////////////////
- if(dryrun) { // DRYRUN
- // testing sync with no real playback
- sleep(1);
-
- if(chanID==1) { // offer should send sync
- sleep(3);
- // create output socket
- sock_out = zmq_socket (zcontext, ZMQ_PUB);
- zmq_connect(sock_out, pgmaddr);
- // broadcast offer
- zmq_msg_init_size (&request, 4);
- memcpy (zmq_msg_data (&request), "SYNC", 4);
- // printf ("Sending offer...\n", request_nbr);
- zmq_send (sock_out, &request, 0);
- fprintf(stderr,"SYNC\n");
- zmq_msg_close (&request);
- 
- } else { // listener should prepare to receive sync
-
- // create input socket
- sock_in = zmq_socket (zcontext, ZMQ_SUB);
- zmq_bind (sock_in, pgmaddr); 
- zmq_setsockopt(sock_in,ZMQ_SUBSCRIBE,"",0);
- // Wait for next request from client
- zmq_msg_init (&request);
- zmq_recv (sock_in, &request, 0);
- if(strncmp((char*)zmq_msg_data(&request),"SYNC",4)==0) {
- fprintf(stderr,"SYNC\n");
- }
- zmq_msg_close (&request);
-
- }
- 
- //////////////////////
- } else { // RUN FOR REAL
-
-
- // Prepare
- 
- // load, play and pause in sequence
- // break and reopen connection in between
- 
- // was connected already
- 
- // load
- render_uri_meta(upnp,filename);
- render_upnp(upnp,"SetAVTransportURI", upnp->meta);
- 
- // must re-connect socket between commands
- send_upnp(upnp);
- recv_upnp(upnp, 1000);
- close(upnp->sockfd);
- upnp->sockfd = 0;
- 
- connect_upnp(upnp);
- render_upnp(upnp,"Play","<Speed>1</Speed>");
- 
- send_upnp(upnp);
- recv_upnp(upnp, 1000);
- close(upnp->sockfd);
- upnp->sockfd = 0;
- 
- connect_upnp(upnp);
- render_upnp(upnp,"Pause","");
- send_upnp(upnp);
- recv_upnp(upnp, 1000);
- 
- // START SYNC
- 
- // was connected already
- render_upnp(upnp,"Play","<Speed>1</Speed>");
- send_upnp(upnp);
- recv_upnp(upnp, 1000);
-
- /*
- case 'g': // dump a parsable full state of the device 
- render_upnp(upnp,"GetTransportInfo","");
- parser = GetTransportInfo;
- 
- break;
-
- case 'm': // set the playmode:
- // "NORMAL", "REPEAT_ONE", "REPEAT_ALL", "RANDOM"
- {
- char tmp[256];
- snprintf(tmp,255,"<NewPlayMode>%s</NewPlayMode>",filename);
- render_upnp(upnp,"SetPlayMode",tmp);
- }
- break;
-
- case 'j': // jump aka seek
- // <SeekMode> and <SeekTarget>
- {
- char tmp[512];
- snprintf(tmp,511,"<Unit>REL_TIME</Unit><Target>%s</Target>",filename);
- render_upnp(upnp,"Seek",tmp);
- }
- break;
- */
-
- free_upnp(upnp);
- }
-
- if(sock_out) {
- zmq_close (sock_out);
- sock_out = NULL;
- }
- if(sock_in) {
- zmq_close (sock_in);
- sock_in = NULL;
- }
- 
- zmq_term (zcontext);
-
- if(listeners) {
- for(int c; c<chanTOT; c++)
- free(listeners[c]);
- free(listeners);
- }
- 
- 
- exit(0);
-}