/* vim: set sw=2: -*- Mode: C; tab-width: 2; indent-tabs-mode: t; c-basic-offset: 2; c-indent-level: 2-*- */

/**

 * GMyth Library

 *

 * @file gmyth/gmyth_file_transfer.c

 * 

 * @brief <p> 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 <rosfran.borges@indt.org.br>

 *

 *//*

 * 

 * 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 *   

 */

#include "gmyth_file_transfer.h"

#include "gmyth_uri.h"

#include "gmyth_livetv.h"

#include "gmyth_util.h"

#include "gmyth_socket.h"

#include "gmyth_stringlist.h"

#include "gmyth_debug.h"

#include <unistd.h>

#include <glib.h>

#include <arpa/inet.h>

#include <sys/types.h>

#include <sys/socket.h>

#include <netdb.h>

#include <errno.h>

#include <stdlib.h>

#include <assert.h>

#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					128*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

GThread *monitor_th = NULL;

enum myth_sock_types {

  GMYTH_PLAYBACK_TYPE = 0,

  GMYTH_MONITOR_TYPE,

  GMYTH_FILETRANSFER_TYPE,

  GMYTH_RINGBUFFER_TYPE

};

static GStaticMutex st_mutex = G_STATIC_MUTEX_INIT;

static gboolean* myth_control_sock_listener( GIOChannel *io_channel );

/* static gboolean myth_control_sock_listener( GIOChannel *io_channel, GIOCondition condition, gpointer data ); */

static GMutex*				mutex 					 = NULL;

static GCond*					io_watcher_cond  = NULL;

static GMainContext*	io_watcher_context  = NULL;

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 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;

  gobject_class = (GObjectClass *) klass;

  gobject_class->dispose  = gmyth_file_transfer_dispose;

  gobject_class->finalize = gmyth_file_transfer_finalize;

}

static void

gmyth_file_transfer_init (GMythFileTransfer *transfer)

{ 

  g_return_if_fail( transfer != NULL );

  //transfer->card_id = num;

  //transfer->rec_id = -1;

  //transfer->recordernum = num;

  transfer->readposition = 0;

  transfer->filesize = GMYTHTV_FILE_SIZE;

  //transfer->timeoutisfast = FALSE;

  //transfer->userreadahead = GMYTHTV_USER_READ_AHEAD;

  //transfer->retries = GMYTHTV_RETRIES;  

  transfer->control_sock = NULL;

  transfer->sock = NULL;

  /* it is used for signalizing the event socket consumer thread */

  io_watcher_cond = g_cond_new();

  /* mutex to control access to the event socket consumer thread */

  mutex = g_mutex_new();

}

static void

gmyth_file_transfer_dispose  (GObject *object)

{

  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 ()

{

  GMythFileTransfer *transfer = GMYTH_FILE_TRANSFER ( g_object_new (

	GMYTH_FILE_TRANSFER_TYPE, FALSE ));

  return transfer;

}

gboolean

gmyth_file_transfer_open (GMythFileTransfer *transfer, GString *uri_str)

{

  gboolean ret = TRUE;

  transfer->uri = gmyth_uri_new_with_value ( uri_str->str );

  transfer->hostname = g_string_new( gmyth_uri_get_host(transfer->uri) );

  transfer->port = gmyth_uri_get_port( transfer->uri );

  gmyth_debug ("[%s] URI: %s\n", __FUNCTION__, uri_str->str);

  gmyth_debug ("hostname: %s, port %d\n", transfer->hostname->str, transfer->port);

  /* configure the control socket */

  if (transfer->control_sock == NULL) { 

    if ( !gmyth_connect_to_backend (transfer)) {

      g_printerr( "Connection to backend failed (Control Socket).\n" );

      ret = FALSE;

    }

  } else {

    g_warning("Remote transfer control socket already created.\n");

  }

  return ret;

}

static gboolean*

//myth_control_sock_listener( GIOChannel *io_channel, GIOCondition condition, gpointer data )

myth_control_sock_listener( GIOChannel *io_channel )

{

  GIOStatus io_status;

  GError *error = NULL;

  GIOCondition io_cond;

  GIOCondition condition;

  gchar *trash = g_new0( gchar, GMYTHTV_BUFFER_SIZE*10 );

  guint recv = 0;

  gboolean* ret = g_new0( gboolean, 1 );

  //gboolean ret = TRUE;

  gsize len = 0;

  *ret = TRUE;

  //myth_control_acquire_context (TRUE);

  if ( io_channel == NULL ) {

  	g_debug ("Monitor socket is NULL!\n");

  	*ret = FALSE;

  	goto clean_up;

  }

  gmyth_debug ("Listening on Monitor socket...!\n");

  while (TRUE) {

	  condition = g_io_channel_get_buffer_condition( io_channel );

	  //myth_control_acquire_context (TRUE);

	  //while ( !has_io_access ) 

	  //	g_cond_wait( io_watcher_cond, mutex );

	  //myth_control_acquire_context (TRUE);  

	  if (condition & G_IO_HUP) {

	    gmyth_debug ("Read end of pipe died!\n");

	    *ret = FALSE;

	    goto clean_up;

	  }

	  if ( ( condition & G_IO_IN ) != 0 ) {

	  	io_status = g_io_channel_set_encoding( io_channel, NULL, &error );

	    do 

	    {

	      //trash = g_new0( gchar, GMYTHTV_BUFFER_SIZE );

	      io_status = g_io_channel_read_chars( io_channel, trash + recv, 

	      		GMYTHTV_BUFFER_SIZE, &len, &error);

	      gmyth_debug ( "[%s] Received data buffer from IO binary channel... %d bytes gone!\n", 

	      		__FUNCTION__, len );

	      recv += len;

	      //msg = g_strconcat( msg, g_strdup(trash), NULL );

	      //if ( trash != NULL )

	      //	g_free( trash );

	      io_cond = g_io_channel_get_buffer_condition( io_channel );

	    } while ( ( io_cond & G_IO_IN ) != 0 && ( io_status != G_IO_STATUS_ERROR ) );

	  }

	  g_usleep( 300 );

  }

  //ret = g_io_channel_read_chars ( source, &msg, &len, NULL, &err);

  if ( io_status == G_IO_STATUS_ERROR ) {

    gmyth_debug ("[%s] Error reading: %s\n", __FUNCTION__, error != NULL ? error->message : "" );

    *ret = FALSE;

  }

  gmyth_debug ("\n[%s]\tEVENT: Read %d bytes: %s\n\n", __FUNCTION__, len, trash != NULL ? trash : "[no event data]" );

  //g_cond_signal( io_watcher_cond );

clean_up:

  //myth_control_release_context( );

  return ret;

}

static gboolean

gmyth_connect_to_backend_monitor (GMythFileTransfer *transfer, GString *hostname )

{

	gboolean ret = TRUE;

	GString *base_str = g_string_new( "" );	

	//guint source_id = 0;

  transfer->event_sock = gmyth_socket_new();

  gmyth_socket_connect ( transfer->event_sock, transfer->hostname->str, transfer->port);

  g_string_printf( base_str, "ANN Monitor %s %d", hostname->str, TRUE );

  gmyth_socket_send_command( transfer->event_sock, base_str );

  GString *resp = gmyth_socket_receive_response( transfer->event_sock );

  gmyth_debug ( "[%s] Got Monitor response from %s: %s\n", __FUNCTION__, base_str->str, resp->str );

  monitor_th = g_thread_create( (GThreadFunc)myth_control_sock_listener, transfer->event_sock->sd_io_ch, FALSE, NULL );

  /*

  io_watcher_context = g_main_context_default();

  GMainLoop *loop = g_main_loop_new( NULL, TRUE );

  GSource *source;

  if ( transfer->event_sock->sd_io_ch != NULL ) {

    source = g_io_create_watch( transfer->event_sock->sd_io_ch, G_IO_IN );

    //transfer->event_sock->sd_io_ch

  } else {

  	ret = FALSE;

  	goto cleanup;

  }

  g_source_set_callback ( source, (GIOFunc)myth_control_sock_listener, NULL, NULL );

  g_source_attach( source, io_watcher_context );

  */

/*

  if (source==NULL) {

    g_printerr( "[%s] Error adding watch listener function to the IO control channel!\n", __FUNCTION__ );

    goto cleanup;

  }

*/

  gmyth_debug ( "[%s]\tOK! Starting listener on the MONITOR event socket...\n", __FUNCTION__ );

  //g_main_loop_run( loop );

cleanup:

	/*

  if ( source != NULL )

    g_source_unref( source );

  */

  gmyth_debug( "[%s] Watch listener function over the IO control channel? %s!!!\n", 

  			__FUNCTION__, ( ret == TRUE ? "YES" : "NO" ) );

  return ret;

}

static gboolean

gmyth_connect_to_backend (GMythFileTransfer *transfer)

{

  GString *base_str = g_string_new( "" );

  GString *hostname = NULL;

  GMythStringList *strlist;

  g_return_val_if_fail (transfer != NULL, FALSE );

  g_return_val_if_fail (transfer->uri != NULL, FALSE );

  g_return_val_if_fail (transfer->hostname != NULL, FALSE);

  g_return_val_if_fail (transfer->port > 0, FALSE);

  gchar *path_dir = gmyth_uri_get_path (transfer->uri);

  /* Creates the control socket */

  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->hostname->str, transfer->port, TRUE)) {

    g_object_unref (transfer->control_sock);

    transfer->control_sock = NULL;

    return FALSE;

  }

  /* Creates the data socket */

  transfer->sock = gmyth_socket_new ();

  gmyth_socket_connect (transfer->sock, transfer->hostname->str, transfer->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, path_dir );

  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;

      return FALSE;

  }

  //gmyth_connect_to_backend_monitor( transfer, hostname );

  if ( strlist != NULL )

    g_object_unref( strlist );

  g_string_free (base_str, TRUE);

  g_string_free (hostname, TRUE);

  return TRUE;

}    

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)

{

  gsize 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;   

  GMythStringList *strlist = NULL;

  GString *query;

  gchar *data_buffer = NULL;

  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 );

  myth_control_acquire_context( TRUE );

  while (total_read == 0) {

    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 ) {

      		data_buffer = g_new0( gchar, bytes_sent );    

     	    while (total_read != bytes_sent) { 

                io_status = g_io_channel_read_chars( io_channel, data_buffer + total_read, 

	  								(gsize) bytes_sent, // buffer_len

		       					&bytes_read, &error );

	        total_read += bytes_read;

	        /* append new data to the increasing byte array */

		  		data = g_byte_array_append ( data, g_memdup( data_buffer, bytes_read ), bytes_read );

		printf ("Total file transfer data read: %d\n", total_read);

	    }

	}

    }

  }

  myth_control_release_context( );

    g_object_unref( strlist );

    g_string_free (query, TRUE);

    if ( error != NULL ) {

        g_printerr( "Cleaning-up ERROR: %s [msg = %s, code = %d]\n", __FUNCTION__, error->message, 

			error->code );

        g_error_free( error );

    }

  transfer->readposition += total_read;

  return total_read;

}

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)

{

    assert (transfer);

    return transfer->filesize;

}