/**
* 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;
}