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