/** * GMyth Library * * @file gmyth/gmyth_file_transfer.c * * @brief

GMythFileTransfer deals with the file streaming media remote/local * transfering to the MythTV frontend. * * Copyright (C) 2006 INdT - Instituto Nokia de Tecnologia. * @author Rosfran Lins Borges * *//* * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU Lesser General Public License as published by * the Free Software Foundation; either version 2 of the License, or * (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU Lesser General Public License * along with this program; if not, write to the Free Software * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA * * GStreamer MythTV plug-in properties: * - location (backend server hostname/URL) [ex.: myth://192.168.1.73:28722/1000_1092091.nuv] * - path (qurl - remote file to be opened) * - port number * */ #ifdef HAVE_CONFIG_H #include "config.h" #endif #include "gmyth_file_transfer.h" #include "gmyth_livetv.h" #include "gmyth_util.h" #include "gmyth_socket.h" #include "gmyth_stringlist.h" #include "gmyth_debug.h" #include "gmyth_uri.h" #include "gmyth_marshal.h" #include #include #include #include #include #include #include #include #include #define GMYTHTV_QUERY_HEADER "QUERY_FILETRANSFER " /* default values to the file transfer parameters */ #define GMYTHTV_USER_READ_AHEAD TRUE #define GMYTHTV_RETRIES -1 #define GMYTHTV_FILE_SIZE 0 #define GMYTHTV_BUFFER_SIZE 64*1024 #define GMYTHTV_VERSION 30 #define GMYTHTV_TRANSFER_MAX_WAITS 700 #ifdef GMYTHTV_ENABLE_DEBUG #define GMYTHTV_ENABLE_DEBUG 1 #else #undef GMYTHTV_ENABLE_DEBUG #endif /* this NDEBUG is to maintain compatibility with GMyth library */ #ifndef NDEBUG #define GMYTHTV_ENABLE_DEBUG 1 #endif enum myth_sock_types { GMYTH_PLAYBACK_TYPE = 0, GMYTH_MONITOR_TYPE, GMYTH_FILETRANSFER_TYPE, GMYTH_RINGBUFFER_TYPE }; #define GMYTH_FILE_TRANSFER_GET_PRIVATE(obj) (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GMYTH_FILE_TRANSFER_TYPE, GMythFileTransferPrivate)) struct _GMythFileTransferPrivate { GMythLiveTV *livetv; gboolean do_next_program_chain; }; static void gmyth_file_transfer_class_init (GMythFileTransferClass *klass); static void gmyth_file_transfer_init (GMythFileTransfer *object); static void gmyth_file_transfer_dispose (GObject *object); static void gmyth_file_transfer_finalize (GObject *object); static void gmyth_file_transfer_program_info_changed( GMythFileTransfer *transfer, gint msg_code, gpointer livetv_transfer ); static gboolean gmyth_connect_to_backend (GMythFileTransfer *transfer); void gmyth_file_transfer_close( GMythFileTransfer *transfer ); static gboolean myth_control_acquire_context( gboolean do_wait ); static gboolean myth_control_release_context( ); G_DEFINE_TYPE(GMythFileTransfer, gmyth_file_transfer, G_TYPE_OBJECT) static void gmyth_file_transfer_class_init (GMythFileTransferClass *klass) { GObjectClass *gobject_class; GMythFileTransferClass *gtransfer_class; gobject_class = (GObjectClass *) klass; gtransfer_class = (GMythFileTransferClass *) gobject_class; gobject_class->dispose = gmyth_file_transfer_dispose; gobject_class->finalize = gmyth_file_transfer_finalize; g_type_class_add_private (gobject_class, sizeof (GMythFileTransferPrivate)); gtransfer_class->program_info_changed_handler = gmyth_file_transfer_program_info_changed; gtransfer_class->program_info_changed_handler_signal_id = g_signal_new ( "program-info-changed", G_TYPE_FROM_CLASS (gtransfer_class), G_SIGNAL_RUN_LAST | G_SIGNAL_NO_RECURSE | G_SIGNAL_NO_HOOKS, 0, NULL, NULL, gmyth_marshal_VOID__INT_POINTER, G_TYPE_NONE, 2, G_TYPE_INT, G_TYPE_POINTER ); } static void gmyth_file_transfer_init (GMythFileTransfer *transfer) { g_return_if_fail( transfer != NULL ); transfer->readposition = 0; transfer->filesize = GMYTHTV_FILE_SIZE; transfer->priv = GMYTH_FILE_TRANSFER_GET_PRIVATE(transfer); transfer->priv->do_next_program_chain = FALSE; transfer->priv->livetv = NULL; transfer->control_sock = NULL; transfer->sock = NULL; transfer->mutex = g_mutex_new(); g_signal_connect ( G_OBJECT (transfer), "program-info-changed", (GCallback)(GMYTH_FILE_TRANSFER_GET_CLASS(transfer)->program_info_changed_handler), NULL ); } static void gmyth_file_transfer_dispose (GObject *object) { GMythFileTransfer *transfer = GMYTH_FILE_TRANSFER (object); if ( transfer->mutex != NULL ) { g_mutex_free( transfer->mutex ); transfer->mutex = NULL; } if ( transfer->control_sock != NULL ) { g_object_unref( transfer->control_sock ); transfer->control_sock = NULL; } if ( transfer->sock != NULL ) { g_object_unref( transfer->sock ); transfer->sock = NULL; } if ( transfer->filename != NULL ) { g_free( transfer->filename ); transfer->filename = NULL; } G_OBJECT_CLASS (gmyth_file_transfer_parent_class)->dispose (object); } static void gmyth_file_transfer_finalize (GObject *object) { g_signal_handlers_destroy (object); G_OBJECT_CLASS (gmyth_file_transfer_parent_class)->finalize (object); } // fixme: do we need the card_id???? GMythFileTransfer* gmyth_file_transfer_new ( const GMythBackendInfo *backend_info) { GMythFileTransfer *transfer = GMYTH_FILE_TRANSFER (g_object_new (GMYTH_FILE_TRANSFER_TYPE, NULL)); transfer->backend_info = (GMythBackendInfo *)backend_info; g_object_ref (transfer->backend_info); return transfer; } GMythFileTransfer* gmyth_file_transfer_new_with_uri (const gchar* uri_str) { GMythFileTransfer *transfer = GMYTH_FILE_TRANSFER (g_object_new (GMYTH_FILE_TRANSFER_TYPE, NULL)); transfer->backend_info = gmyth_backend_info_new_with_uri (uri_str); return transfer; } gboolean gmyth_file_transfer_open (GMythFileTransfer *transfer, const gchar* filename) { gboolean ret = TRUE; g_return_val_if_fail (transfer != NULL, FALSE); g_return_val_if_fail (filename != NULL, FALSE); if (transfer->filename != NULL) { g_free (transfer->filename); transfer->filename = NULL; } transfer->filename = g_strdup (filename); /* configure the control socket */ if (transfer->control_sock == NULL) { if (!gmyth_connect_to_backend (transfer)) { gmyth_debug ("Connection to backend failed (Control Socket).\n"); ret = FALSE; } } else { g_warning("Remote transfer control socket already created.\n"); } gmyth_debug ("Got file with size = %lld.\n", transfer->filesize); return ret; } static gboolean gmyth_connect_to_backend (GMythFileTransfer *transfer) { GString *base_str = NULL; GString *hostname = NULL; GMythStringList *strlist = NULL; gboolean ret = TRUE; g_return_val_if_fail (transfer != NULL, FALSE ); base_str = g_string_new (""); /* Creates the control socket */ if (transfer->control_sock != NULL) { g_object_unref (transfer->control_sock); transfer->control_sock = NULL; } transfer->control_sock = gmyth_socket_new(); // Connects the socket, send Mythtv ANN command and verify Mythtv protocol version if (!gmyth_socket_connect_to_backend (transfer->control_sock, transfer->backend_info->hostname, transfer->backend_info->port, TRUE)) { g_object_unref (transfer->control_sock); transfer->control_sock = NULL; return FALSE; } /* Creates the data socket */ if (transfer->sock != NULL) { g_object_unref (transfer->sock); transfer->sock = NULL; } transfer->sock = gmyth_socket_new (); gmyth_socket_connect (transfer->sock, transfer->backend_info->hostname, transfer->backend_info->port); strlist = gmyth_string_list_new(); hostname = gmyth_socket_get_local_hostname(); gmyth_debug( "[%s] MythTV version (from backend) = %d.\n", __FUNCTION__, transfer->control_sock->mythtv_version ); if ( transfer->control_sock->mythtv_version > 26 ) g_string_printf( base_str, "ANN FileTransfer %s 1 -1", hostname->str); else g_string_printf( base_str, "ANN FileTransfer %s", hostname->str); gmyth_string_list_append_string (strlist, base_str ); gmyth_string_list_append_char_array (strlist, transfer->filename); gmyth_socket_write_stringlist (transfer->sock, strlist ); gmyth_socket_read_stringlist (transfer->sock, strlist ); /* file identification used in future file transfer requests to backend */ transfer->file_id = gmyth_string_list_get_int( strlist, 1 ); /* Myth URI stream file size - decoded using two 8-bytes sequences (64 bits/long long types) */ transfer->filesize = gmyth_util_decode_long_long( strlist, 2 ); gmyth_debug ( "[%s] ***** Received: recordernum = %d, filesize = %" G_GUINT64_FORMAT "\n", __FUNCTION__, transfer->file_id, transfer->filesize ); if (transfer->filesize < 0 ) { gmyth_debug ( "[%s] Got filesize equals to %llu is lesser than 0 [invalid stream file]\n", __FUNCTION__, transfer->filesize ); g_object_unref (transfer->sock); transfer->sock = NULL; ret = FALSE; goto cleanup; } cleanup: if ( strlist != NULL ) g_object_unref( strlist ); g_string_free (base_str, TRUE); g_string_free (hostname, TRUE); return ret; } void gmyth_file_transfer_emit_program_info_changed_signal ( GMythFileTransfer *transfer, gint msg_code, gpointer live_tv ) { /* g_signal_emit_by_name ( G_OBJECT(transfer), "program-info-changed", msg_code, live_tv ); */ gmyth_debug( "Calling signal handler... [FILE_TRANSFER]" ); g_signal_emit ( transfer, GMYTH_FILE_TRANSFER_GET_CLASS (transfer)->program_info_changed_handler_signal_id, 0, /* details */ msg_code, live_tv ); } gboolean gmyth_file_transfer_is_open (GMythFileTransfer *transfer) { GMythStringList *strlist; GString *query; g_return_val_if_fail (transfer->control_sock != NULL, FALSE); g_return_val_if_fail (transfer->sock != NULL, FALSE); strlist = gmyth_string_list_new(); query = g_string_new (GMYTHTV_QUERY_HEADER); g_string_append_printf (query, "%d", transfer->file_id ); gmyth_string_list_append_string (strlist, query ); gmyth_string_list_append_char_array( strlist, "IS_OPEN" ); gmyth_socket_write_stringlist( transfer->control_sock, strlist ); gmyth_socket_read_stringlist( transfer->control_sock, strlist ); g_string_free (query, TRUE); g_object_unref (strlist); return ( strlist != NULL && gmyth_string_list_get_int( strlist, 0 ) == 1 ); } void gmyth_file_transfer_close( GMythFileTransfer *transfer ) { GMythStringList *strlist; GString *query; g_return_if_fail (transfer->control_sock != NULL); strlist = gmyth_string_list_new( ); query = g_string_new (GMYTHTV_QUERY_HEADER); g_string_append_printf( query, "%d", transfer->file_id); gmyth_string_list_append_string( strlist, query ); gmyth_string_list_append_char_array( strlist, "DONE" ); if ( gmyth_socket_sendreceive_stringlist(transfer->control_sock, strlist) <= 0 ) { // fixme: time out??? g_printerr( "Remote file timeout.\n" ); } g_string_free (query, TRUE); g_object_unref (strlist); if (transfer->sock) { g_object_unref( transfer->sock ); transfer->sock = NULL; } if (transfer->control_sock) { g_object_unref( transfer->control_sock ); transfer->control_sock = NULL; } } gint64 gmyth_file_transfer_seek(GMythFileTransfer *transfer, guint64 pos, gint whence) { GMythStringList *strlist = gmyth_string_list_new(); GString *query; g_return_val_if_fail (transfer->sock != NULL, -1); g_return_val_if_fail (transfer->control_sock != NULL, -1); strlist = gmyth_string_list_new(); query = g_string_new (GMYTHTV_QUERY_HEADER); g_string_append_printf (query, "%d", transfer->file_id); myth_control_acquire_context( TRUE ); gmyth_string_list_append_string( strlist, query ); gmyth_string_list_append_char_array( strlist, "SEEK" ); gmyth_string_list_append_uint64( strlist, pos ); gmyth_string_list_append_int( strlist, whence ); if (pos > 0 ) gmyth_string_list_append_uint64( strlist, pos ); else gmyth_string_list_append_uint64( strlist, transfer->readposition ); gmyth_socket_sendreceive_stringlist( transfer->control_sock, strlist ); gint64 retval = gmyth_string_list_get_int64(strlist, 0); transfer->readposition = retval; gmyth_debug ( "[%s] got reading position pointer from the streaming = %lld\n", __FUNCTION__, retval ); myth_control_release_context( ); return retval; } static gboolean myth_control_acquire_context( gboolean do_wait ) { gboolean ret = TRUE; //guint max_iter = 50; //g_mutex_lock( mutex ); //while ( !has_io_access ) // g_cond_wait( io_watcher_cond, mutex ); //has_io_access = FALSE; //myth_control_acquire_context (FALSE); /* if ( do_wait ) { while ( --max_iter > 0 && !g_main_context_wait( io_watcher_context, io_watcher_cond, mutex ) ) ret = FALSE; } else if ( !g_main_context_acquire( io_watcher_context ) ) ret = FALSE; */ //g_static_mutex_lock( &st_mutex ); return ret; } static gboolean myth_control_release_context( ) { gboolean ret = TRUE; //g_static_mutex_unlock( &st_mutex ); //g_main_context_release( io_watcher_context ); //g_main_context_wakeup( io_watcher_context ); //has_io_access = TRUE; //g_cond_broadcast( io_watcher_cond ); //g_mutex_unlock( mutex ); return ret; } gint gmyth_file_transfer_read(GMythFileTransfer *transfer, GByteArray *data, gint size, gboolean read_unlimited) { gint bytes_sent = 0; gsize bytes_read = 0; gsize total_read = 0; GError *error = NULL; GIOChannel *io_channel; GIOChannel *io_channel_control; GIOCondition io_cond; GIOCondition io_cond_control; GIOStatus io_status = G_IO_STATUS_NORMAL, io_status_control = G_IO_STATUS_NORMAL; gboolean ret = TRUE; GString *query; g_return_val_if_fail ( data != NULL, -2 ); io_channel = transfer->sock->sd_io_ch; io_channel_control = transfer->control_sock->sd_io_ch; io_status = g_io_channel_set_encoding( io_channel, NULL, &error ); if ( io_status == G_IO_STATUS_NORMAL ) gmyth_debug ( "[%s] Setting encoding to binary data socket).\n", __FUNCTION__ ); io_cond = g_io_channel_get_buffer_condition( io_channel ); io_cond_control = g_io_channel_get_buffer_condition( io_channel ); if ( transfer->sock == NULL || ( io_status == G_IO_STATUS_ERROR ) ) { g_printerr( "gmyth_file_transfer_read(): Called with no raw socket.\n" ); //exit(0); // fixme remove this return -1; } if ( transfer->control_sock == NULL || ( io_status_control == G_IO_STATUS_ERROR ) ) { g_printerr( "gmyth_file_transfer_read(): Called with no control socket.\n" ); //exit(0); // fixme remove this return -1; } query = g_string_new (GMYTHTV_QUERY_HEADER); g_string_append_printf ( query, "%d", transfer->file_id ); gmyth_debug ("[%s] Transfer_query = %s\n", __FUNCTION__, query->str ); /* send requests to the maximum number of REQUEST_BLOCK messages */ guint max_tries = 5; myth_control_acquire_context( TRUE ); while (total_read == 0 && --max_tries > 0) { GMythStringList *strlist = gmyth_string_list_new(); gmyth_string_list_append_char_array( strlist, query->str ); gmyth_string_list_append_char_array( strlist, "REQUEST_BLOCK" ); gmyth_string_list_append_int( strlist, size ); // Request the block to the backend gmyth_socket_write_stringlist( transfer->control_sock, strlist ); // Receives the backand answer gmyth_socket_read_stringlist( transfer->control_sock, strlist ); if ( strlist != NULL && gmyth_string_list_length( strlist ) > 0 ) { bytes_sent = gmyth_string_list_get_int( strlist, 0 ); // -1 on backend error gmyth_debug ( "[%s] got SENT buffer message = %d\n", __FUNCTION__, bytes_sent ); if ( bytes_sent >= 0 ) { gchar *data_buffer = g_new0 ( gchar, bytes_sent ); while ( 0 != bytes_sent ) { io_status = g_io_channel_read_chars( io_channel, data_buffer + total_read, (gsize) bytes_sent, &bytes_read, &error ); total_read += bytes_read; bytes_sent -= bytes_read; /* append new data to the increasing byte array */ data = g_byte_array_append (data, (const guint8*)data_buffer, bytes_read); gmyth_debug ("Total transfer data read: %d\n", total_read); } g_free (data_buffer); } /* if */ } else if ( !(transfer->priv != NULL && transfer->priv->livetv != NULL && transfer->priv->do_next_program_chain) ) { total_read = GMYTHTV_FILE_TRANSFER_READ_ERROR; g_object_unref (strlist); strlist = NULL; break; } g_object_unref (strlist); strlist = NULL; } if ( bytes_sent == 0 && max_tries == 0 ) { gmyth_debug( "Trying to move to the next program chain..." ); transfer->priv = GMYTH_FILE_TRANSFER_GET_PRIVATE(transfer); if ( transfer->priv != NULL && transfer->priv->livetv != NULL && transfer->priv->do_next_program_chain ) { total_read = GMYTHTV_FILE_TRANSFER_NEXT_PROG_CHAIN; g_mutex_lock( transfer->mutex ); ret = gmyth_livetv_next_program_chain( transfer->priv->livetv ); g_mutex_unlock( transfer->mutex ); if ( !ret ) gmyth_debug( "Cannot change to the next program chain!" ); else gmyth_debug( "OK!!! MOVED to the next program chain [%s]!", (gmyth_tvchain_get_id( transfer->priv->livetv->tvchain ))->str ); } } /* if */ myth_control_release_context( ); g_string_free (query, TRUE); if ( error != NULL ) { gmyth_debug ("Cleaning-up ERROR: %s [msg = %s, code = %d]\n", __FUNCTION__, error->message, error->code); g_error_free (error); } if ( total_read > 0 ) transfer->readposition += total_read; return total_read; } static void gmyth_file_transfer_program_info_changed( GMythFileTransfer *transfer, gint msg_code, gpointer livetv_transfer ) { GMythLiveTV *livetv = GMYTH_LIVETV( livetv_transfer ); gmyth_debug( "Program info changed! ( file transfer orig. = %p, ptr. = [%s], user data = [%s] )", transfer, livetv_transfer != NULL ? "[NOT NULL]" : "[NULL]", livetv != NULL ? "[NOT NULL]" : "[NULL]" ); if ( livetv != NULL && transfer == livetv->file_transfer ) { gmyth_debug( "YES, the requested program info movement on the LiveTV transfer is authentical!" ); } transfer->priv = GMYTH_FILE_TRANSFER_GET_PRIVATE(transfer); transfer->priv->livetv = livetv; transfer->priv->do_next_program_chain = TRUE; //g_object_unref( transfer ); } gboolean gmyth_file_transfer_settimeout( GMythFileTransfer *transfer, gboolean fast ) { GMythStringList *strlist = NULL; g_return_val_if_fail (transfer->sock != NULL, FALSE); g_return_val_if_fail (transfer->control_sock != NULL, FALSE); // if ( transfer->timeoutisfast == fast ) // return; strlist = gmyth_string_list_new(); gmyth_string_list_append_char_array( strlist, GMYTHTV_QUERY_HEADER ); gmyth_string_list_append_char_array( strlist, "SET_TIMEOUT" ); gmyth_string_list_append_int( strlist, fast ); gmyth_socket_write_stringlist( transfer->control_sock, strlist ); gmyth_socket_read_stringlist( transfer->control_sock, strlist ); // transfer->timeoutisfast = fast; return TRUE; } guint64 gmyth_file_transfer_get_filesize (GMythFileTransfer *transfer) { g_return_val_if_fail (transfer != NULL, 0); return transfer->filesize; }