gst-plugins-mythtv/src/myth_file_transfer.c
author rosfran
Mon Oct 23 14:36:26 2006 +0100 (2006-10-23)
branchtrunk
changeset 37 324e04989738
parent 34 c71d37b93734
permissions -rwxr-xr-x
[svn r38] Added program chain backend events.
     1 /* vim: set sw=2: -*- Mode: C; tab-width: 2; indent-tabs-mode: t; c-basic-offset: 2; c-indent-level: 2-*- */
     2 /**
     3  * GStreamer plug-in properties:
     4  * - location (backend server hostname/URL) [ex.: myth://192.168.1.73:28722/1000_1092091.nuv]
     5  * - path (qurl - remote file to be opened)
     6  * - port number
     7  *   @author Rosfran Lins Borges <rosfran.borges@indt.org.br>
     8  */
     9 
    10 #include "myth_file_transfer.h"
    11 #include "myth_uri.h"
    12 #include "myth_livetv.h"
    13 #include <gmyth/gmyth_util.h>
    14 #include <gmyth/gmyth_socket.h>
    15 #include <gmyth/gmyth_stringlist.h>
    16 
    17 #include <unistd.h>
    18 #include <glib.h>
    19 
    20 #include <arpa/inet.h>
    21 #include <sys/types.h>
    22 #include <sys/socket.h>
    23 #include <netdb.h>
    24 #include <errno.h>
    25 #include <stdlib.h>
    26 
    27 #define MYTHTV_QUERY_HEADER "QUERY_FILETRANSFER"
    28 #define MYTHTV_RECORDER_HEADER "QUERY_RECORDER"
    29 
    30 /* default values to the file transfer parameters */
    31 #define MYTHTV_USER_READ_AHEAD	FALSE
    32 #define MYTHTV_RETRIES			1
    33 #define MYTHTV_FILE_SIZE		-1
    34 
    35 #define MYTHTV_BUFFER_SIZE		8*1024
    36 
    37 #define MYTHTV_VERSION			30
    38 
    39 #define MYTHTV_TRANSFER_MAX_WAITS	700
    40 
    41 #ifdef MYTHTV_ENABLE_DEBUG
    42 #define MYTHTV_ENABLE_DEBUG	1
    43 #else
    44 #undef MYTHTV_ENABLE_DEBUG
    45 #endif
    46 
    47 /* this NDEBUG is to maintain compatibility with GMyth library */
    48 #ifndef NDEBUG
    49 #define MYTHTV_ENABLE_DEBUG	1
    50 #endif
    51 
    52 static guint wait_to_transfer = 0;
    53 
    54 enum myth_sock_types {
    55   MYTH_PLAYBACK_TYPE = 0,
    56   MYTH_MONITOR_TYPE,
    57   MYTH_FILETRANSFER_TYPE,
    58   MYTH_RINGBUFFER_TYPE
    59 };
    60 
    61 static GStaticMutex mutex = G_STATIC_MUTEX_INIT;
    62 
    63 static GMainContext *io_watcher_context = NULL;
    64 
    65 static void myth_file_transfer_class_init          (MythFileTransferClass *klass);
    66 static void myth_file_transfer_init                (MythFileTransfer *object);
    67 
    68 static void myth_file_transfer_dispose  (GObject *object);
    69 static void myth_file_transfer_finalize (GObject *object);
    70 
    71 static GMythSocket *myth_connect_to_transfer_backend( MythFileTransfer **transfer, guint sock_type );
    72 static void* myth_init_io_watchers( void *data );
    73 
    74 void myth_file_transfer_close( MythFileTransfer *transfer );
    75 
    76 G_DEFINE_TYPE(MythFileTransfer, myth_file_transfer, G_TYPE_OBJECT)
    77 
    78 #if 0
    79 static guint64
    80 mmyth_util_decode_long_long( GMythStringList *strlist, guint offset  )
    81 {
    82 
    83   guint64 ret_value = 0LL;
    84 
    85   g_return_val_if_fail( strlist != NULL, ret_value );
    86 
    87   if ( offset < gmyth_string_list_length( strlist ))
    88     g_printerr( "[%s] Offset is lower than the GMythStringList (offset = %d)!\n", __FUNCTION__, offset );
    89   g_return_val_if_fail( offset < gmyth_string_list_length( strlist ), ret_value );
    90 
    91   gint l1 = gmyth_string_list_get_int( strlist, offset );
    92   gint l2 = gmyth_string_list_get_int( strlist, offset + 1 );
    93 
    94   ret_value = ((guint64)(l2) & 0xffffffffLL) | ((guint64)(l1) << 32);
    95 
    96   return ret_value;
    97 
    98 }
    99 #endif
   100 
   101 static void
   102 myth_file_transfer_class_init (MythFileTransferClass *klass)
   103 {
   104   GObjectClass *gobject_class;
   105 
   106   gobject_class = (GObjectClass *) klass;
   107 
   108   gobject_class->dispose  = myth_file_transfer_dispose;
   109   gobject_class->finalize = myth_file_transfer_finalize;
   110 }
   111 
   112   static void
   113 myth_file_transfer_init (MythFileTransfer *myth_file_transfer)
   114 { 
   115   g_return_if_fail( myth_file_transfer != NULL );
   116   myth_file_transfer->mythtv_version = MYTHTV_VERSION;
   117 }
   118 
   119 static void
   120 myth_file_transfer_dispose  (GObject *object)
   121 {
   122   MythFileTransfer *myth_file_transfer = MYTH_FILE_TRANSFER(object);
   123 
   124   myth_file_transfer_close( myth_file_transfer );
   125 
   126   G_OBJECT_CLASS (myth_file_transfer_parent_class)->dispose (object);
   127 }
   128 
   129   static void
   130 myth_file_transfer_finalize (GObject *object)
   131 {
   132   g_signal_handlers_destroy (object);
   133 
   134   G_OBJECT_CLASS (myth_file_transfer_parent_class)->finalize (object);
   135 }
   136 
   137   MythFileTransfer*
   138 myth_file_transfer_new (gint num, GString *uri_str, gshort port, gint mythtv_version)
   139 {
   140   MythFileTransfer *transfer = MYTH_FILE_TRANSFER ( g_object_new (
   141 	MYTH_FILE_TRANSFER_TYPE, FALSE ));
   142 
   143   if ( mythtv_version > 0 )
   144     transfer->mythtv_version = mythtv_version;
   145 
   146   transfer->card_id = num;
   147 
   148   transfer->rec_id = -1;
   149 
   150   transfer->recordernum = num;
   151   transfer->uri = myth_uri_new ( uri_str->str );
   152 
   153   transfer->hostname = g_string_new( myth_uri_gethost(transfer->uri) );
   154   g_print( "\t--> transfer->hostname = %s\n", transfer->hostname->str );
   155 
   156   if ( port >= 0 )
   157     transfer->port = port;
   158   else
   159     transfer->port = myth_uri_getport( transfer->uri );
   160 
   161   g_print( "\t--> transfer->port = %d\n", transfer->port );
   162 
   163   transfer->readposition = 0;
   164   transfer->filesize = MYTHTV_FILE_SIZE;
   165   transfer->timeoutisfast = FALSE;
   166 
   167   transfer->userreadahead = MYTHTV_USER_READ_AHEAD;
   168   transfer->retries = MYTHTV_RETRIES;  
   169 
   170   transfer->live_tv = FALSE;
   171 
   172   transfer->query = g_string_new( MYTHTV_QUERY_HEADER );
   173   g_string_append_printf ( transfer->query, " %d", transfer->recordernum );
   174   g_print( "\t--> transfer->query = %s\n", transfer->query->str );
   175 
   176   transfer->control_sock = NULL;
   177   transfer->event_sock = NULL;
   178   transfer->sock = NULL;
   179 
   180   return transfer;
   181 }
   182 
   183 gboolean
   184 myth_file_transfer_livetv_setup( MythFileTransfer **transfer, GMythSocket *live_socket )
   185 {
   186 	(*transfer)->sock = live_socket;
   187 	g_object_ref( live_socket );
   188 
   189 	return TRUE;
   190 }
   191 
   192 gboolean
   193 myth_file_transfer_playback_setup( MythFileTransfer **transfer, gboolean live_tv )
   194 {
   195 
   196   gboolean ret = TRUE;
   197 
   198   (*transfer)->live_tv = live_tv;
   199 
   200   printf("[%s] Running config to the %s...\n", __FUNCTION__, live_tv ? "LiveTV" : "FileTransfer" );
   201 
   202   /* configure the control socket */
   203   if ((*transfer)->control_sock == NULL) { 
   204 
   205     if ( myth_connect_to_transfer_backend ( transfer, MYTH_PLAYBACK_TYPE ) == NULL ) {
   206       g_printerr( "Connection to backend failed (Control Socket).\n" );
   207       ret = FALSE;
   208     }
   209 
   210   } else {
   211     g_warning("Remote transfer control socket already created.\n");
   212   }
   213 
   214   return ret;
   215 
   216 }
   217 
   218 gboolean
   219 myth_file_transfer_setup( MythFileTransfer **transfer, gboolean live_tv )
   220 {
   221   GMythStringList *strlist = NULL;
   222 
   223   gboolean ret = TRUE;
   224 
   225   (*transfer)->live_tv = live_tv;
   226 
   227   printf("[%s] Running config to the %s...\n", __FUNCTION__, live_tv ? "LiveTV" : "FileTransfer" );
   228 
   229 #if 0
   230   /* configure the control socket */
   231   if ((*transfer)->event_sock == NULL) { 
   232 
   233     if ( myth_connect_to_transfer_backend ( transfer, MYTH_MONITOR_TYPE ) == NULL ) {
   234       g_printerr( "Connection to backend failed (Event Socket).\n" );
   235       ret = FALSE;
   236     }
   237 
   238   } else {
   239     g_warning("Remote transfer control socket already created.\n");
   240   }
   241 #endif
   242 
   243   /* configure the socket */
   244   if ( (*transfer)->sock == NULL ) { 
   245 
   246     //if ( live_tv == FALSE ) {
   247 
   248     if ( myth_connect_to_transfer_backend ( transfer, MYTH_FILETRANSFER_TYPE ) == NULL ) {
   249       g_printerr ("Connection to backend failed (Raw Transfer Socket).\n");
   250       ret = FALSE;
   251     }
   252 
   253     if ( !(*transfer)->live_tv && (*transfer)->control_sock != NULL) {
   254       strlist = gmyth_string_list_new();
   255       g_string_printf ( (*transfer)->query, "%s %d", MYTHTV_QUERY_HEADER, (*transfer)->recordernum );
   256 
   257       gmyth_string_list_append_string( strlist, (*transfer)->query );
   258       gmyth_string_list_append_char_array( strlist, "IS_OPEN" );
   259 
   260       gmyth_socket_write_stringlist( (*transfer)->control_sock, strlist );
   261       gmyth_socket_read_stringlist( (*transfer)->control_sock, strlist );
   262 
   263       if ( strlist!=NULL && gmyth_string_list_get_int( strlist, 0 ) == 1 ) {
   264 	g_print( "[%s] Remote Myth FileTransfer socket is open!\n", __FUNCTION__ );
   265       } else {
   266 	g_print( "[%s] Remote Myth FileTransfer socket is CLOSED! See the MythTV Server Backend for configuration details...\n", __FUNCTION__ );
   267 	ret = FALSE;
   268       }
   269     }
   270 
   271   } else {
   272     g_warning("Remote transfer (raw) socket already created.\n");
   273   }
   274 
   275   return ret;
   276 }
   277 
   278 static GMythSocket *
   279 myth_connect_to_transfer_backend( MythFileTransfer **transfer, guint sock_type )
   280 {
   281   GMythSocket *sock = NULL;
   282 
   283   g_return_val_if_fail( transfer != NULL && *transfer != NULL, NULL );
   284   g_return_val_if_fail( (*transfer)->uri != NULL, NULL );
   285 
   286   g_static_mutex_lock (&mutex);
   287 
   288   gchar *path_dir = myth_uri_getpath( (*transfer)->uri );
   289   //g_print( "\t--> %s: path_dir = %s\n", __FUNCTION__, path_dir );
   290 
   291   gchar *stype = g_strdup( "" );
   292 
   293   //  if ( (*transfer)->live_tv == FALSE ) {
   294 
   295   sock = gmyth_socket_new();
   296 
   297   gmyth_socket_connect( &sock, (*transfer)->hostname->str, (*transfer)->port );
   298 
   299   /*
   300      } else {
   301      sock = (*transfer)->sock;
   302      }
   303      */
   304 #ifdef MYTHTV_ENABLE_DEBUG
   305 
   306   g_print( "[%s] --> Creating socket... (%s, %d)\n", __FUNCTION__, (*transfer)->hostname->str, (*transfer)->port );
   307 #endif
   308 
   309   GMythStringList *strlist = NULL;
   310 
   311   GString *hostname = g_string_new( myth_uri_gethost( (*transfer)->uri ) );
   312   GString *base_str = g_string_new( "" );
   313 
   314   if ( gmyth_socket_check_protocol_version_number (sock, (*transfer)->mythtv_version) ) {
   315 
   316     if (sock == NULL) {
   317       stype = (sock_type==MYTH_PLAYBACK_TYPE) ? "control socket" : "file data socket";
   318       g_printerr( "FileTransfer, open_socket(%s): \n"
   319 	  "\t\t\tCould not connect to server \"%s\" @ port %d\n", stype, 
   320 	  (*transfer)->hostname->str, (*transfer)->port );
   321       g_object_unref(sock);
   322       g_static_mutex_unlock (&mutex);
   323       return NULL;
   324     }
   325 
   326     hostname = gmyth_socket_get_local_hostname();
   327 
   328     g_print( "[%s] local hostname = %s\n", __FUNCTION__, hostname->str  );
   329 
   330     if ( sock_type == MYTH_PLAYBACK_TYPE )
   331     {
   332       (*transfer)->control_sock = sock;
   333       g_string_printf( base_str, "ANN Playback %s %d", hostname->str, TRUE );
   334 
   335       gmyth_socket_send_command( (*transfer)->control_sock, base_str );
   336       GString *resp = gmyth_socket_receive_response( (*transfer)->control_sock );
   337       g_print( "[%s] Got Playback response from %s: %s\n", __FUNCTION__, base_str->str, resp->str );
   338     }
   339     else if ( sock_type == MYTH_MONITOR_TYPE )
   340     {
   341       (*transfer)->event_sock = sock;
   342       g_string_printf( base_str, "ANN Monitor %s %d", hostname->str, TRUE );
   343 
   344       gmyth_socket_send_command( (*transfer)->event_sock, base_str );
   345       GString *resp = gmyth_socket_receive_response( (*transfer)->event_sock );
   346       g_print( "[%s] Got Monitor response from %s: %s\n", __FUNCTION__, base_str->str, resp->str );
   347       //g_thread_create( myth_init_io_watchers, (void*)(*transfer), FALSE, NULL );
   348       myth_init_io_watchers ( (void*)(*transfer) );
   349 
   350       g_printerr( "[%s] Watch listener function to the IO control channel on thread %p.\n", __FUNCTION__, g_thread_self() );
   351 
   352     }
   353     else if ( sock_type == MYTH_FILETRANSFER_TYPE )
   354     {
   355       (*transfer)->sock = sock;
   356       strlist = gmyth_string_list_new();
   357       //g_string_printf( base_str, "ANN FileTransfer %s %d %d", hostname->str,
   358       //  		transfer->userreadahead, transfer->retries );
   359       g_string_printf( base_str, "ANN FileTransfer %s", hostname->str );
   360 
   361       gmyth_string_list_append_string( strlist, base_str );
   362       gmyth_string_list_append_char_array( strlist, path_dir );
   363 
   364       gmyth_socket_write_stringlist( (*transfer)->sock, strlist );
   365       gmyth_socket_read_stringlist( (*transfer)->sock, strlist );
   366 
   367       /* socket number, where all the stream data comes from - got from the MythTV remote backend */
   368       (*transfer)->recordernum = gmyth_string_list_get_int( strlist, 1 );
   369 
   370       /* Myth URI stream file size - decoded using two 8-bytes sequences (64 bits/long long types) */
   371       (*transfer)->filesize = gmyth_util_decode_long_long( strlist, 2 );
   372 
   373       printf( "[%s] ***** Received: recordernum = %d, filesize = %" G_GUINT64_FORMAT "\n", __FUNCTION__,
   374 	  (*transfer)->recordernum, (*transfer)->filesize );
   375 
   376       if ( (*transfer)->filesize <= 0 ) {
   377 	g_print( "[%s] Got filesize equals to %llu is lesser than 0 [invalid stream file]\n", __FUNCTION__, (*transfer)->filesize );
   378 	g_object_unref(sock);
   379 	sock = NULL; 
   380       }
   381     }
   382     else if ( sock_type == MYTH_RINGBUFFER_TYPE )
   383     {
   384       (*transfer)->sock = sock;
   385       //myth_file_transfer_spawntv( (*transfer), NULL );      
   386 
   387       strlist = gmyth_string_list_new();
   388       g_string_printf( base_str, "ANN RingBuffer %s %d", hostname->str, (*transfer)->card_id );
   389 
   390       gmyth_socket_send_command( (*transfer)->sock, base_str );
   391       GString *resp = gmyth_socket_receive_response( (*transfer)->sock );
   392       g_print( "[%s] Got RingBuffer response from %s: %s\n", __FUNCTION__, base_str->str, resp->str );
   393 
   394     }
   395 
   396   }
   397 
   398   printf("[%s] ANN %s sent: %s\n", (sock_type==MYTH_PLAYBACK_TYPE) ? "Playback" : (sock_type==MYTH_FILETRANSFER_TYPE) ? "FileTransfer" : "Monitor", __FUNCTION__, base_str->str);
   399 
   400   if ( strlist != NULL )
   401     g_object_unref( strlist );
   402 
   403   g_static_mutex_unlock (&mutex);
   404 
   405   return sock;
   406 }    
   407 
   408 void
   409 myth_file_transfer_spawntv ( MythFileTransfer *file_transfer, 
   410     GString *tvchain_id )
   411 {
   412   GMythStringList *str_list;
   413 
   414   g_debug ("myth_file_transfer_spawntv.\n");
   415 
   416   str_list = gmyth_string_list_new ();
   417 
   418   g_string_printf( file_transfer->query, "%s %d", MYTHTV_RECORDER_HEADER, 
   419       file_transfer->card_id );
   420   gmyth_string_list_append_string (str_list, file_transfer->query);
   421   gmyth_string_list_append_string (str_list, g_string_new ("SPAWN_LIVETV"));
   422   if (tvchain_id!=NULL) {
   423     gmyth_string_list_append_string (str_list, tvchain_id);
   424     gmyth_string_list_append_int (str_list, FALSE); // PIP = FALSE (0)
   425   }
   426 
   427   gmyth_socket_sendreceive_stringlist ( file_transfer->sock, str_list );
   428 
   429   //GString *str = NULL;
   430 
   431   //if (str_list!=NULL && (str = gmyth_string_list_get_string( str_list, 0 )) != NULL && strcasecmp( str->str, "ok" ) != 0 ) {
   432   //  g_print( "[%s]\t\tSpawnLiveTV is OK!\n", __FUNCTION__ );
   433   //}
   434   if (str_list!=NULL)
   435     g_object_unref (str_list);
   436 
   437 }
   438 
   439 gboolean
   440 myth_file_transfer_is_recording ( MythFileTransfer *file_transfer )
   441 {
   442   gboolean ret = TRUE;
   443   
   444   GMythStringList *str_list = gmyth_string_list_new ();
   445 
   446   g_debug ( "[%s]\n", __FUNCTION__ );
   447   g_static_mutex_lock (&mutex);
   448 
   449   g_string_printf( file_transfer->query, "%s %d", MYTHTV_RECORDER_HEADER, 
   450       file_transfer->rec_id >= 0 ? file_transfer->rec_id : file_transfer->card_id );
   451   gmyth_string_list_append_string (str_list, file_transfer->query);
   452   gmyth_string_list_append_string (str_list, g_string_new ("IS_RECORDING"));
   453 
   454   gmyth_socket_sendreceive_stringlist ( file_transfer->control_sock, str_list );
   455 
   456   if ( str_list != NULL && gmyth_string_list_length(str_list) > 0 ) 
   457   {
   458     GString *str = NULL;
   459     if ( ( str = gmyth_string_list_get_string( str_list, 0 ) ) != NULL && strcmp( str->str, "bad" )!= 0 ) {
   460       gint is_rec = gmyth_string_list_get_int( str_list, 0 );
   461       if ( is_rec != 0 )
   462 	ret = TRUE;
   463       else
   464 	ret = FALSE;
   465     }
   466   }  
   467   g_print( "[%s] %s, stream is %s being recorded!\n", __FUNCTION__, ret ? "YES" : "NO", ret ? "" : "NOT" );
   468   g_static_mutex_unlock (&mutex);
   469 
   470   if ( str_list != NULL )
   471     g_object_unref (str_list);
   472 
   473   return ret;
   474 
   475 }
   476 
   477 guint64
   478 myth_file_transfer_get_file_position ( MythFileTransfer *file_transfer )
   479 {
   480   gint64 pos = 0;
   481 
   482   GMythStringList *str_list = gmyth_string_list_new ();
   483 
   484   g_debug ( "[%s]\n", __FUNCTION__ );
   485   g_static_mutex_lock (&mutex);
   486 
   487   g_string_printf( file_transfer->query, "%s %d", MYTHTV_RECORDER_HEADER, 
   488       file_transfer->rec_id >= 0 ? file_transfer->rec_id : file_transfer->card_id );
   489 
   490   gmyth_string_list_append_string (str_list, file_transfer->query);
   491   gmyth_string_list_append_string (str_list, g_string_new ("GET_FILE_POSITION"));
   492 
   493   gmyth_socket_sendreceive_stringlist ( file_transfer->control_sock, str_list );
   494 
   495   if ( str_list != NULL && gmyth_string_list_length(str_list) > 0 ) 
   496   {
   497     GString *str = NULL;
   498     if ( ( str = gmyth_string_list_get_string( str_list, 0 ) ) != NULL && strcmp ( str->str, "bad" ) != 0 )
   499       pos = gmyth_util_decode_long_long( str_list, 0 );
   500   } 
   501   g_static_mutex_unlock (&mutex);
   502 
   503 #ifndef MYTHTV_ENABLE_DEBUG
   504 
   505   g_print( "[%s] Got file position = %llu\n", __FUNCTION__, pos );
   506 #endif
   507   if (str_list!=NULL)
   508     g_object_unref (str_list);
   509 
   510   return pos;
   511 
   512 }
   513 
   514   glong
   515 myth_file_transfer_get_recordernum( MythFileTransfer *transfer )
   516 {
   517   return transfer->recordernum;
   518 }
   519 
   520   glong
   521 myth_file_transfer_get_filesize( MythFileTransfer *transfer )
   522 {
   523   return transfer->filesize;
   524 }
   525 
   526   gboolean
   527 myth_file_transfer_isopen( MythFileTransfer *transfer )
   528 {
   529   return (transfer->sock != NULL && transfer->control_sock != NULL);
   530 }
   531 
   532   void
   533 myth_file_transfer_close( MythFileTransfer *transfer )
   534 {
   535   GMythStringList *strlist;
   536 
   537   if (transfer->control_sock == NULL)
   538     return;
   539 
   540   strlist = gmyth_string_list_new( );
   541 
   542   g_string_printf( transfer->query, "%s %d", MYTHTV_QUERY_HEADER, 
   543       transfer->recordernum );
   544   gmyth_string_list_append_string( strlist, transfer->query );
   545   gmyth_string_list_append_char_array( strlist, "DONE" );
   546 
   547 
   548   if ( gmyth_socket_sendreceive_stringlist(transfer->control_sock, strlist) <= 0 )
   549   {
   550     g_printerr( "Remote file timeout.\n" );
   551   }
   552 
   553   if (transfer->sock)
   554   {
   555     g_object_unref( transfer->sock );
   556     transfer->sock = NULL;
   557   }
   558 
   559   if (transfer->control_sock)
   560   {
   561     g_object_unref( transfer->control_sock );
   562     transfer->control_sock = NULL;
   563   } 
   564 
   565 }
   566 
   567   void
   568 myth_file_transfer_reset_controlsock( MythFileTransfer *transfer )
   569 {
   570   if (transfer->control_sock == NULL)
   571   {
   572     g_printerr( "myth_file_transfer_reset_controlsock(): Called with no control socket" );
   573     return;
   574   }
   575 
   576   GString *str = gmyth_socket_receive_response( transfer->control_sock );
   577 
   578   g_string_free( str, TRUE );
   579 }
   580 
   581 void
   582 myth_file_transfer_reset_sock( MythFileTransfer *transfer )
   583 {
   584   if ( transfer->sock == NULL )
   585   {
   586     g_printerr( "myth_file_transfer_reset_sock(): Called with no raw socket" );
   587     return;
   588   }
   589 
   590   GString *str = gmyth_socket_receive_response( transfer->sock );
   591 
   592   g_string_free( str, TRUE );
   593 }
   594 
   595 void
   596 myth_file_transfer_reset( MythFileTransfer *transfer )
   597 {
   598   myth_file_transfer_reset_controlsock( transfer );
   599   myth_file_transfer_reset_sock( transfer );
   600 }
   601 
   602 guint64
   603 myth_file_transfer_seek(MythFileTransfer *transfer, guint64 pos, gint whence)
   604 {
   605   if (transfer->sock == NULL)
   606   {
   607     g_printerr( "[%s] myth_file_transfer_seek(): Called with no socket", __FUNCTION__ );
   608     return 0;
   609   }
   610 
   611   if (transfer->control_sock == NULL)
   612     return 0;
   613 
   614   // if (!controlSock->isOpen() || controlSock->error())
   615   //   return 0;
   616 
   617   GMythStringList *strlist = gmyth_string_list_new();
   618   g_string_printf (transfer->query, "%s %d", MYTHTV_QUERY_HEADER, transfer->recordernum);
   619   gmyth_string_list_append_string( strlist, transfer->query );
   620   gmyth_string_list_append_char_array( strlist, "SEEK" );
   621   gmyth_string_list_append_uint64( strlist, pos );
   622   
   623   gmyth_string_list_append_int( strlist, whence );  
   624 
   625   if (pos > 0 )
   626     gmyth_string_list_append_uint64( strlist, pos );
   627   else
   628     gmyth_string_list_append_uint64( strlist, transfer->readposition );
   629   
   630   gmyth_socket_sendreceive_stringlist( transfer->control_sock, strlist );
   631 
   632   guint64 retval = gmyth_string_list_get_uint64(strlist, 0);
   633   transfer->readposition = retval;
   634   g_print( "[%s] got reading position pointer from the streaming = %llu\n", 
   635       __FUNCTION__, retval );
   636 
   637   //myth_file_transfer_reset( transfer );
   638 
   639   return retval;
   640 }
   641 
   642 static gboolean
   643 myth_control_sock_listener( GIOChannel *source, GIOCondition condition, gpointer data )
   644 {
   645 
   646   GIOStatus ret;
   647   GError *err = NULL;
   648   gchar *msg = g_strdup("");
   649   
   650   g_static_mutex_lock( &mutex );  
   651 
   652   gsize len;
   653   if (condition & G_IO_HUP)
   654     g_error ("Read end of pipe died!\n");
   655   ret = g_io_channel_read_line ( source, &msg, &len, NULL, &err);
   656   if ( ret == G_IO_STATUS_ERROR )
   657     g_error ("[%s] Error reading: %s\n", __FUNCTION__, err != NULL ? err->message : "" );
   658   g_print ("\n\n\n\n\n\n[%s]\t\tEVENT: Read %u bytes: %s\n\n\n\n\n", __FUNCTION__, len, msg != NULL ? msg : "" );
   659   if ( msg != NULL )
   660     g_free (msg);
   661     
   662   g_static_mutex_unlock( &mutex );
   663 
   664   return TRUE;
   665 
   666 }
   667 
   668 static void*
   669 myth_init_io_watchers( void *data )
   670 {
   671   MythFileTransfer *transfer = (MythFileTransfer*)data;
   672   io_watcher_context = g_main_context_new();
   673   GMainLoop *loop = g_main_loop_new( NULL, FALSE );
   674 
   675   GSource *source = NULL;
   676 
   677   if ( transfer->event_sock->sd_io_ch != NULL )
   678     source = g_io_create_watch( transfer->event_sock->sd_io_ch, G_IO_IN | G_IO_HUP );
   679   else
   680   	goto cleanup;
   681 
   682   g_source_set_callback ( source, (GSourceFunc)myth_control_sock_listener, NULL, NULL );
   683 
   684   g_source_attach( source, io_watcher_context );
   685 
   686   if (source==NULL) {
   687     g_printerr( "[%s] Error adding watch listener function to the IO control channel!\n", __FUNCTION__ );
   688     goto cleanup;
   689   }
   690   
   691   g_print( "[%s]\tOK! Starting listener on the MONITOR event socket...\n", __FUNCTION__ );
   692 
   693   g_main_loop_run( loop );
   694 
   695 cleanup:
   696   if ( source != NULL )
   697     g_source_unref( source );
   698 
   699   g_main_loop_unref( loop );
   700 
   701   g_main_context_unref( io_watcher_context );
   702 
   703   return NULL;
   704 }
   705 
   706 
   707 gint 
   708 myth_file_transfer_read(MythFileTransfer *transfer, void *data, gint size, gboolean read_unlimited)
   709 {
   710   gint recv = 0;
   711   gsize bytes_read = 0;
   712   gint sent = 0;
   713   guint remaining = 0;  
   714   gboolean response = FALSE;
   715 
   716   GIOChannel *io_channel;
   717   GIOChannel *io_channel_control;
   718 
   719   GIOCondition io_cond;
   720   GIOCondition io_cond_control;
   721   GIOStatus io_status = G_IO_STATUS_NORMAL, io_status_control = G_IO_STATUS_NORMAL;   
   722 
   723   gint buf_len = MYTHTV_BUFFER_SIZE;
   724 
   725   GMythStringList *strlist = NULL;
   726   GError *error = NULL;
   727 
   728   gchar *trash = g_strdup("");
   729 
   730   g_return_val_if_fail ( data != NULL, -2 );
   731 
   732   /* gets the size of the entire file, if the size requested is lesser than 0 */
   733   if ( size <= 0 )
   734     size = transfer->filesize;
   735 
   736   io_channel = transfer->sock->sd_io_ch;
   737   io_channel_control = transfer->control_sock->sd_io_ch;
   738 
   739   //g_io_channel_set_flags( io_channel, G_IO_FLAG_APPEND | 
   740   //    G_IO_STATUS_AGAIN | G_IO_FLAG_IS_READABLE | G_IO_FLAG_IS_WRITEABLE | 
   741   //    G_IO_FLAG_IS_SEEKABLE, NULL );
   742 
   743   io_status = g_io_channel_set_encoding( io_channel, NULL, &error );
   744   if ( io_status == G_IO_STATUS_NORMAL )
   745     g_print( "[%s] Setting encoding to binary data socket).\n", __FUNCTION__ );
   746 
   747   io_cond = g_io_channel_get_buffer_condition( io_channel );  
   748 
   749   io_cond_control = g_io_channel_get_buffer_condition( io_channel );
   750 
   751   if ( transfer->sock == NULL || ( io_status == G_IO_STATUS_ERROR ) )
   752   {
   753     g_printerr( "myth_file_transfer_read(): Called with no raw socket.\n" );
   754     recv = -1;
   755     goto cleanup;
   756   }
   757 
   758   if ( transfer->control_sock == NULL || ( io_status_control == G_IO_STATUS_ERROR ) )
   759   {
   760     g_printerr( "myth_file_transfer_read(): Called with no control socket.\n" );
   761     recv = -1;
   762     goto cleanup;
   763   }
   764 
   765   /*
   766      if (!controlSock->isOpen() || controlSock->error())
   767      return -1;
   768      */
   769 
   770   if ( ( io_cond & G_IO_IN ) != 0 ) {
   771     do 
   772     {
   773       trash = g_new0( gchar, MYTHTV_BUFFER_SIZE );
   774 
   775       io_status = g_io_channel_read_chars( io_channel, trash, 
   776       		MYTHTV_BUFFER_SIZE, &bytes_read, &error);
   777 
   778       g_print( "[%s] cleaning buffer on IO binary channel... %d bytes gone!\n", 
   779       		__FUNCTION__, bytes_read );
   780       
   781       if ( trash != NULL )
   782       	g_free( trash );
   783       	
   784       io_cond = g_io_channel_get_buffer_condition( io_channel );
   785 
   786     } while ( ( io_cond & G_IO_IN ) != 0 && ( io_status != G_IO_STATUS_ERROR ) && (error == NULL) );
   787 
   788     //if ( trash!= NULL )
   789     //  g_free( trash );
   790   }
   791 
   792   if ( ( io_cond_control & G_IO_IN ) != 0 ) {
   793     GMythStringList *strlist_tmp = gmyth_string_list_new();
   794     gmyth_socket_read_stringlist( transfer->control_sock, strlist_tmp );
   795     g_object_unref( strlist_tmp );
   796   }
   797 
   798   wait_to_transfer = 0;
   799 
   800   //while ( transfer->live_tv && ( myth_file_transfer_get_file_position( transfer ) < 4096 ) &&
   801   //		wait_to_transfer++ < MYTHTV_TRANSFER_MAX_WAITS )
   802   //	g_usleep( 1000*50 ); /* waits just for 2/10 second */
   803 
   804   //g_thread_create( myth_init_io_watchers, (void*)transfer, FALSE, NULL );
   805   //g_printerr( "[%s] Watch listener function to the IO control channel on thread %p.\n", __FUNCTION__, g_thread_self() );
   806 
   807   //g_static_mutex_lock (&mutex);
   808   //strlist = gmyth_string_list_new();
   809 
   810   g_string_printf ( transfer->query, "%s %d",
   811   	  /*transfer->live_tv ? MYTHTV_RECORDER_HEADER :*/  MYTHTV_QUERY_HEADER, 
   812       /* transfer->live_tv ? transfer->card_id :*/ transfer->recordernum ); // transfer->recordernum
   813   g_print( "\t[%s] Transfer_query = %s\n", __FUNCTION__, transfer->query->str );
   814   
   815   sent = size;
   816   remaining = size - recv;  
   817   //g_static_mutex_unlock( &mutex );
   818   //data = (void*)g_new0( gchar, size );
   819 
   820   //g_io_channel_flush( io_channel, NULL );
   821 
   822   //g_static_mutex_lock( &mutex );
   823 
   824   io_cond = g_io_channel_get_buffer_condition( io_channel );
   825 
   826   while ( recv < size  && !response )//&& ( io_cond & G_IO_IN ) != 0 )
   827   {
   828   	g_io_channel_flush( io_channel_control, NULL );
   829   	
   830     strlist = gmyth_string_list_new();
   831     gmyth_string_list_append_char_array( strlist, transfer->query->str );
   832     gmyth_string_list_append_char_array( strlist, 
   833     		/*transfer->live_tv ? "REQUEST_BLOCK_RINGBUF" :*/ "REQUEST_BLOCK" );
   834     gmyth_string_list_append_int( strlist, remaining );
   835 	gmyth_socket_write_stringlist( transfer->control_sock, strlist );
   836 	
   837 	guint count_bytes = 0;
   838       
   839     do
   840     {
   841       //buf_len = ( sent - recv ) > MYTHTV_BUFFER_SIZE ? MYTHTV_BUFFER_SIZE : ( sent - recv );
   842       if ( remaining > MYTHTV_BUFFER_SIZE ) {
   843       	buf_len = MYTHTV_BUFFER_SIZE;
   844       } else {
   845       	buf_len = remaining;
   846       }
   847 	  
   848       bytes_read = 0;
   849 
   850       io_status = g_io_channel_read_chars( io_channel, data + recv, 
   851 	  		buf_len, &bytes_read, &error );
   852 	  
   853 	  //g_static_mutex_unlock( &mutex );
   854       /*
   855 	 GString *sss = g_string_new("");
   856 	 sss = g_string_append_len( sss, (gchar*)data+recv, bytes_read );
   857 
   858 	 g_print( "[%s] Reading buffer (length = %d)\n", __FUNCTION__, bytes_read);
   859 	 */
   860       if ( bytes_read > 0 )
   861       {
   862 	//if ( bytes_read <= buf_len )
   863 		  recv += bytes_read;
   864 		  count_bytes += bytes_read;
   865 		  remaining -= bytes_read;
   866 		  g_print( "[%s] Reading buffer (bytes read = %d, remaining = %d)\n", __FUNCTION__, bytes_read, remaining );
   867 		  if ( remaining == 0 ) {
   868 		  	break;
   869 		  }
   870       } else {
   871       	break;
   872       }
   873       
   874       //if ( remaining > 0 ) {
   875 
   876       if ( io_status == G_IO_STATUS_EOF ) {
   877 	g_print( "[%s] got EOS!", __FUNCTION__ );
   878 	break;
   879       } else if ( io_status == G_IO_STATUS_ERROR ) {
   880 	g_print( "[%s] myth_file_transfer_read(): socket error.\n", __FUNCTION__ );
   881 	break;
   882       }
   883       //}
   884 
   885       /* increase buffer size, to allow get more data (do not obey to the buffer size) */
   886       if ( read_unlimited == TRUE ) {
   887       	// FOR NOW, DO NOTHING!!!
   888 	//if ( recv > buf_len )
   889 	// sent += (bytes_read - buf_len) + 1;
   890       }
   891 
   892       /* verify if the input (read) buffer is ready to receive data */
   893       io_cond = g_io_channel_get_buffer_condition( io_channel );
   894 
   895       g_print( "[%s]\t io_cond %s prepared for reading! (G_IO_IN) !!!\n\n", __FUNCTION__, 
   896       		( ( io_cond & G_IO_IN ) != 0 ) ? "IS" : "IS NOT" );
   897       		
   898       //if ( recv == size )
   899 		//break;
   900 
   901     } while ( remaining > 0 );//&& ( io_status == G_IO_STATUS_NORMAL ) );
   902     
   903    // if ( ( recv < size ) ) {
   904     //	finish_read = FALSE;
   905     //}
   906 
   907     io_cond_control = g_io_channel_get_buffer_condition( io_channel_control );
   908     if ( remaining == 0 )//( io_cond_control & G_IO_IN ) != 0  )
   909     {
   910       gmyth_socket_read_stringlist( transfer->control_sock, strlist );
   911       if ( strlist != NULL && gmyth_string_list_length( strlist ) > 0 ) 
   912       {
   913       	sent = gmyth_string_list_get_int( strlist,  0 ); // -1 on backend error
   914       	g_print( "[%s] got SENT buffer message = %d\n", __FUNCTION__, sent );
   915       	if ( sent != 0 )
   916       	{
   917       		g_print( "[%s]\t received = %d bytes, backend says %d bytes sent, "\
   918       			"io_cond %s prepared for reading! (G_IO_IN) !!!\n\n", __FUNCTION__, 
   919 	  			recv, sent, ( ( io_cond & G_IO_IN ) != 0 ) ? "IS" : "IS NOT" );
   920 
   921 	  		if ( sent == count_bytes )
   922 	  		{
   923 	  			response = ( recv == size ); 
   924 	  			g_print( "[%s]\t\tsent %d, which is equals to bytes_read = %d\n\n", 
   925 	  						__FUNCTION__, sent, count_bytes );
   926 	  			if ( response == TRUE )
   927       				break;
   928 	  		}
   929 	  		else
   930 	  		{
   931 	  			g_print( "[%s]\t\tsent %d, which is NOT equals to bytes_read = %d\n\n", 
   932 	  						__FUNCTION__, sent, count_bytes );
   933 	  			goto cleanup;
   934 	  			//response = FALSE;
   935 	  			//break;
   936 	  		}
   937       	} else {
   938       		break;
   939       		//goto cleanup;
   940       	} // if
   941       } // if - reading control response from backend
   942     } else {
   943     	response = FALSE;
   944     } // if - stringlist response
   945     
   946   } // while
   947   
   948   io_cond_control = g_io_channel_get_buffer_condition( io_channel_control );
   949  // io_cond = g_io_channel_get_buffer_condition( io_channel );
   950 
   951   if ( ( ( io_cond_control & G_IO_IN ) != 0 ) &&  		 
   952   		( response || ( recv == size ) ) )
   953   {
   954     if ( gmyth_socket_read_stringlist( transfer->control_sock, strlist ) > 0 )
   955     {
   956       if ( strlist != NULL && gmyth_string_list_length(strlist) > 0 ) 
   957       {
   958 		sent = gmyth_string_list_get_int( strlist, 0 ); // -1 on backend error
   959 		g_print( "[%s]\t received = %d bytes -\tNOW returning from reading buffer I/O socket "\
   960 				 "[%s prepared for reading]! (G_IO_IN) !!!\n\n", __FUNCTION__, 
   961 	    	sent, ( ( io_cond & G_IO_IN ) != 0 ) ? "IS" : "IS NOT" );
   962       }
   963     }
   964     else
   965     {
   966       g_printerr ( "myth_file_transfer_read(): No response from control socket.");
   967       recv = -1;
   968     }
   969     
   970   } 
   971   else if ( error != NULL )
   972   {
   973       g_printerr( "[%s] Error occurred: (%d, %s)\n", __FUNCTION__, error->code, error->message );
   974   }
   975 
   976 cleanup:
   977   //g_static_mutex_unlock (&mutex);
   978 
   979   if ( trash != NULL )
   980     g_free( trash );
   981 
   982   if ( strlist != NULL )
   983     g_object_unref( strlist );
   984 
   985   g_print( "myth_file_transfer_read(): reqd=%d, rcvd=%d, rept=%d, "\
   986       "(rcvd and rept MUST be the same!)\n", size, 
   987       recv, sent );
   988 
   989   //if ( ( recv != size ) || ( sent != size ) ) {
   990     //recv = size;
   991   //}
   992 
   993   if ( error != NULL ) {
   994     g_printerr( "Cleaning-up ERROR: %s [msg = %s, code = %d]\n", __FUNCTION__, error->message, 
   995 	error->code );
   996 	g_error_free( error );
   997   }
   998 
   999   return recv;
  1000 }
  1001 
  1002 void 
  1003 myth_file_transfer_settimeout( MythFileTransfer *transfer, gboolean fast )
  1004 {
  1005 
  1006   GMythStringList *strlist = NULL;
  1007 
  1008   if ( transfer->timeoutisfast == fast )
  1009     return;
  1010 
  1011   if ( transfer->sock == NULL )
  1012   {
  1013     g_printerr( "myth_file_transfer_settimeout(): Called with no socket" );
  1014     return;
  1015   }
  1016 
  1017   if ( transfer->control_sock == NULL )
  1018     return;
  1019 
  1020   strlist = gmyth_string_list_new(); 
  1021   gmyth_string_list_append_string( strlist, transfer->query );
  1022   gmyth_string_list_append_char_array( strlist, "SET_TIMEOUT" );
  1023   gmyth_string_list_append_int( strlist, fast ); 
  1024 
  1025   gmyth_socket_write_stringlist( transfer->control_sock, strlist );
  1026   gmyth_socket_read_stringlist( transfer->control_sock, strlist );
  1027 
  1028   transfer->timeoutisfast = fast;
  1029 
  1030 }
  1031 
  1032 #ifdef DO_TESTING
  1033 
  1034   int
  1035 main( int argc, char *argv[] )
  1036 {
  1037   g_type_init();
  1038 
  1039   MythFileTransfer *file_transfer = myth_file_transfer_new( 1, 
  1040       g_string_new("myth://192.168.1.109:6543/jshks.nuv"), -1, MYTHTV_VERSION );
  1041   myth_file_transfer_setup( &file_transfer );
  1042   gchar *data = g_strdup("");
  1043 
  1044   gint num = myth_file_transfer_read( file_transfer, data, -1 );
  1045 
  1046   return 0;
  1047 
  1048 }
  1049 
  1050 #endif