/*=========================================================================

  Copyright (c) Kitware, Inc.
  All rights reserved.
  See Copyright.txt or http://www.kitware.com/VolViewCopyright.htm for details.

     This software is distributed WITHOUT ANY WARRANTY; without even
     the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
     PURPOSE.  See the above copyright notice for more information.

=========================================================================*/
#include "vtkObject.h"
#include "vtkObjectFactory.h"
#include "vtkKWRemoteIOManager.h"
#include "vtkURIHandler.h"
#include "vtkMutexLock.h"
#include "vtkMultiThreader.h"
#include "vtkKWRemoteIOTask.h"
#include "vtkKWDataTransfer.h"
#include "vtkCallbackCommand.h"
#include "vtkKWRemoteIOUtilities.h"

#include "vtksys/SystemTools.hxx"

#ifdef linux 
#include "unistd.h"
#endif

#ifdef VTK_USE_WIN32_THREADS
#include <windows.h>
#include <winbase.h>
#endif 

#include <list>
#include <string>
#include <algorithm>
#include <set>
#include <queue>

vtkStandardNewMacro ( vtkKWRemoteIOManager );
vtkCxxRevisionMacro(vtkKWRemoteIOManager, "$Revision: 1.20 $");

void vtkKWRemoteIOManagerTimerCallback( ClientData arg )
{
  vtkKWRemoteIOManager *self = reinterpret_cast< vtkKWRemoteIOManager * >(arg);
  self->TimerCallback();
}

class ProcessingTaskQueue : public 
      std::queue<vtkSmartPointer<vtkKWRemoteIOTask> > {};

//----------------------------------------------------------------------------
vtkKWRemoteIOManager::vtkKWRemoteIOManager()
{
  this->DataTransferCollection = vtkCollection::New();
  this->CacheManager           = vtkKWCacheManager::New();

  // set up callback
  this->TransferUpdateCommand = vtkCallbackCommand::New();
  this->TransferUpdateCommand->SetClientData (
      reinterpret_cast<void *>(this) );
  this->AddObserver ( vtkKWRemoteIOManager::TransferUpdateEvent,  
                      this->TransferUpdateCommand);
  this->TransferStatusChangedCommand = vtkCallbackCommand::New();
  this->TransferStatusChangedCommand->SetClientData ( 
      reinterpret_cast<void *>(this) );
  this->AddObserver ( vtkKWRemoteIOManager::TransferStatusChangedEvent,  
                      this->TransferStatusChangedCommand);

  this->ProcessingThreader         = vtkMultiThreader::New();
  this->ProcessingThreadId         = -1;
  this->ProcessingThreadActive     = false;
  this->ProcessingThreadActiveLock = vtkMutexLock::New();
  this->ProcessingTaskQueueLock    = vtkMutexLock::New();
  this->InternalTaskQueue          = new ProcessingTaskQueue;  
  this->TimerToken                 = NULL;
  this->TimerDelay                 = 1000; // 1 second (1000 ms)
}

//----------------------------------------------------------------------------
vtkKWRemoteIOManager::~vtkKWRemoteIOManager()
{
  this->RemoveObservers( 
      vtkKWRemoteIOManager::TransferStatusChangedEvent,  
      this->TransferStatusChangedCommand);
  this->RemoveObservers( 
      vtkKWRemoteIOManager::TransferUpdateEvent,  
      this->TransferUpdateCommand);

  if ( this->TransferUpdateCommand )
    {
    this->TransferUpdateCommand->Delete();
    this->TransferUpdateCommand = NULL;
    }
  
  if ( this->TransferStatusChangedCommand )
    {
    this->TransferStatusChangedCommand->Delete();
    this->TransferStatusChangedCommand = NULL;
    }
  
  if ( this->DataTransferCollection )
    {
    this->DataTransferCollection->RemoveAllItems();
    this->DataTransferCollection->Delete();
    this->DataTransferCollection = NULL;
    }

  if ( this->CacheManager )
    {
    this->CacheManager->Delete();
    this->CacheManager = NULL;
    }

  // Note that TerminateThread does not kill a thread, it only waits
  // for the thread to finish.  We need to signal the thread that we
  // want to terminate

  this->TerminateProcessingThread();
  
  delete this->InternalTaskQueue;
  this->InternalTaskQueue = 0;

  if (this->ProcessingThreader)
    {  
    this->ProcessingThreader->Delete();
    }
  if (this->ProcessingThreadActiveLock)
    {
    this->ProcessingThreadActiveLock->Delete();
    }
  if (this->ProcessingTaskQueueLock)
    {
    this->ProcessingTaskQueueLock->Delete();
    }
}

//----------------------------------------------------------------------------
void vtkKWRemoteIOManager::PrintSelf(ostream& os, vtkIndent indent)
{
  this->vtkObject::PrintSelf(os, indent);
  os << indent << "DataTransferCollection: " << this->GetDataTransferCollection() << "\n";
  os << indent << "CacheManager: " << this->GetCacheManager() << "\n";
}

//----------------------------------------------------------------------------
void vtkKWRemoteIOManager::SetTransferStatus(
  vtkKWDataTransfer *transfer, int status)
{
  if (transfer != NULL)
    {
    if (transfer->GetTransferStatus() != status)
      {
      transfer->SetTransferStatus(status);
      }
    }
}

//----------------------------------------------------------------------------
int vtkKWRemoteIOManager::GetNumberOfDataTransfers()
{
  if (this->DataTransferCollection == NULL)
    {
    return 0;;
    }
  return this->DataTransferCollection->GetNumberOfItems();
}

//----------------------------------------------------------------------------
int vtkKWRemoteIOManager::GetTransferStatus( vtkKWDataTransfer *transfer)
{
  return ( transfer->GetTransferStatus() );
}

//----------------------------------------------------------------------------
vtkKWDataTransfer *vtkKWRemoteIOManager::CreateNewDataTransfer()
{
  vtkKWDataTransfer *transfer = vtkKWDataTransfer::New();
  transfer->SetTransferID(this->GetUniqueTransferID());
  return transfer;
}

//----------------------------------------------------------------------------
void vtkKWRemoteIOManager::AllTransfersClearedFromCache()
{
  if (this->DataTransferCollection == NULL)
    {
    return;
    }

  const int n = this->DataTransferCollection->GetNumberOfItems();
  for ( int i=0; i < n; i++ )
    {
    if (vtkKWDataTransfer *dt = vtkKWDataTransfer::SafeDownCast(
              this->DataTransferCollection->GetItemAsObject(i) ))
      {
      dt->SetTransferCached(0);
      }
    }
}

//----------------------------------------------------------------------------
void vtkKWRemoteIOManager::AddDataTransfer(vtkKWDataTransfer *transfer)
{
  if (transfer == NULL)
    {
    vtkErrorMacro("AddDataTransfer: can't add a null transfer");
    return;
    }

  if ( this->DataTransferCollection == NULL )
    {
    this->DataTransferCollection = vtkCollection::New();
    }

  vtkDebugMacro("AddDataTransfer: adding item");
  this->DataTransferCollection->AddItem(transfer);
  this->Modified();
}

//----------------------------------------------------------------------------
void vtkKWRemoteIOManager::RemoveDataTransfer ( vtkKWDataTransfer *transfer)
{
  if (this->DataTransferCollection == NULL)
    {
    return;
    }

  if (transfer != NULL)
    {
    this->DataTransferCollection->RemoveItem(transfer);
    this->Modified();
    }
}

//----------------------------------------------------------------------------
void vtkKWRemoteIOManager::RemoveDataTransfer( int transferID )
{
  if (this->DataTransferCollection == NULL)
    {
    return;
    }

  int n = this->DataTransferCollection->GetNumberOfItems();
  for ( int i=0; i < n; i++ )
    {
    if (vtkKWDataTransfer *dt = vtkKWDataTransfer::SafeDownCast(
              this->DataTransferCollection->GetItemAsObject(i) ))
      {
      if ( dt->GetTransferID() == transferID )
        {
        this->DataTransferCollection->RemoveItem ( i );
        this->Modified();
        break;
        }
      }
    }
}

//----------------------------------------------------------------------------
vtkKWDataTransfer* vtkKWRemoteIOManager::GetDataTransferByTransferID(
  int transferID)
{
  if (this->DataTransferCollection)
    {
    int n = this->DataTransferCollection->GetNumberOfItems();
    for (int i = 0; i < n; i++)
      {
      vtkKWDataTransfer *dt = vtkKWDataTransfer::SafeDownCast( 
        this->DataTransferCollection->GetItemAsObject(i));
      if (dt && transferID == dt->GetTransferID())
        {
        return dt;
        }
      }
    }
  
  return NULL;
}

//----------------------------------------------------------------------------
vtkKWDataTransfer* vtkKWRemoteIOManager::GetDataTransferByIdentifier(
  const char *identifier)
{
  if (this->DataTransferCollection && identifier)
    {
    int n = this->DataTransferCollection->GetNumberOfItems();
    for (int i = 0; i < n; i++)
      {
      vtkKWDataTransfer *dt = vtkKWDataTransfer::SafeDownCast( 
        this->DataTransferCollection->GetItemAsObject(i));
      if (dt && dt->GetIdentifier() && !strcmp(identifier, dt->GetIdentifier()))
        {
        return dt;
        }
      }
    }
  
  return NULL;
}

//----------------------------------------------------------------------------
void vtkKWRemoteIOManager::ClearDataTransfers( )
{
  if (this->DataTransferCollection == NULL)
    {
    return;
    }

  this->DataTransferCollection->RemoveAllItems();
  this->Modified();
}

//----------------------------------------------------------------------------
int vtkKWRemoteIOManager::QueueRead( vtkKWDataTransfer *transfer )
{
  if (transfer == NULL)
    {
    vtkErrorMacro("QueueRead: null input node!");
    return 0;
    }

  vtkKWCacheManager *cm = this->GetCacheManager();
  const char *source = transfer->GetSourceURI();

  vtkstd::string dest = transfer->GetDestinationURI();

  // If the destination file is a relative filename, make it absolute, by 
  // prepending the cache directory.
  if (!vtksys::SystemTools::FileIsFullPath(dest.c_str()))
    {
    vtkstd::string dest = cm->GetRemoteCacheDirectory();
    dest += "/";

    if (transfer->GetDestinationURI())
      {
      dest += transfer->GetDestinationURI();
      }
    else
      {
      const char *destFname = cm->GetFilenameFromURI(source);
      if (destFname == NULL)
        {
        vtkErrorMacro("Unable to get file name from source URI " << source);
        return 0;
        }
      dest += destFname;
      }
    }

  // Create the path where the file will reside.
  if (!vtksys::SystemTools::MakeDirectory(
       vtksys::SystemTools::GetFilenamePath(dest).c_str()))
    {
    vtkErrorMacro("Failed to create the path " <<
        vtksys::SystemTools::GetFilenamePath(dest).c_str() );
    return 0;
    }

  //--- check to see if RemoteCacheLimit is exceeded
  //--- check to see if FreeBufferSize is exceeded.
 
  //--- if force redownload is enabled, remove the old file from cache.
  if (cm->GetEnableForceRedownload () )
    {
    vtkDebugMacro("QueueRead: Calling remove from cache");
    this->GetCacheManager()->DeleteFromCache(dest.c_str());
    }
  
  //--- trigger logic to download, if there's cache space.
  //--- need to convert to bytes to get a more conservative guess.
  if ( (cm->GetCurrentCacheSize()*1000000.0) < 
      ((float)(cm->GetRemoteCacheLimit())*1000000.0) )
    {
    if ( cm->CachedFileExists(dest.c_str()) )
      {
      vtkDebugMacro( << dest.c_str() << " already exists in the cache.");
      return 1;
      }

    // This is where the final file will be downloaded.
    transfer->SetDestinationURI( dest.c_str() );

    if ( transfer->GetAsynchronous() )
      {
      vtkDebugMacro("QueueRead: Schedule an ASYNCHRONOUS data transfer");

      vtkKWRemoteIOTask *task = vtkKWRemoteIOTask::New();
      task->SetTypeToNetworking();

      // Pass the current data transfer, which has a pointer 
      // to the associated mrml node, as client data to the task.
      if ( !task )
        {
        transfer->Delete();
        return 0;
        }
      transfer->SetTransferStatus ( vtkKWDataTransfer::Pending );
      task->SetTaskFunction(this, (vtkKWRemoteIOTask::TaskFunctionPointer)
                            &vtkKWRemoteIOManager::ApplyTransfer, transfer);
    
      // Schedule the transfer
      if ( ! this->ScheduleTask( task ) )
        {
        transfer->SetTransferStatus( vtkKWDataTransfer::CompletedWithErrors);
        task->Delete();
        return 0;      
        }
      task->Delete();
      }
    else
      {
      vtkDebugMacro("QueueRead: Schedule a SYNCHRONOUS data transfer");

      transfer->SetTransferStatus( vtkKWDataTransfer::Running);

      // Add an observer for possible progress reporting
      vtkURIHandler *handler = transfer->GetHandler();
      unsigned long tag = 0;
      if (handler)
        {
        tag = transfer->AddObserver( 
          vtkKWRemoteIOManager::TransferUpdateEvent, this->TransferUpdateCommand);
        }

      
      // This transfers the data.
      this->ApplyTransfer ( transfer );

      // Remove the observer, now that the transfer is done.
      if (tag)
        {
        transfer->RemoveObserver(tag);
        }

      transfer->SetTransferStatus( vtkKWDataTransfer::Completed);
      }
    }

  return 1;
}

//----------------------------------------------------------------------------
int vtkKWRemoteIOManager::GetUniqueTransferID()
{
  
  //--- keep looping until we find an id that is unique
  int id = 1;
  int exists = 0;
  vtkKWDataTransfer *dt;
    
  if ( this->DataTransferCollection == NULL )
    {
    this->DataTransferCollection = vtkCollection::New();
    }

  // loop until found or return.
  while ( !exists )
    {
    
    // loop thru the existing data transfers
    int n = this->DataTransferCollection->GetNumberOfItems();
    vtkDebugMacro("GetUniqueTransferID: in loop, id = " << id << ", n = " << n);
    for ( int i=0; i < n; i++ )
      {
      dt = vtkKWDataTransfer::SafeDownCast(
          this->DataTransferCollection->GetItemAsObject ( i ) );
      if  ( dt != NULL && id == dt->GetTransferID() )
        {
        exists = 1;
        break;
        }
      }
    // finished looping -- did we find the id?
    if ( exists )
      {
      // if so, try a new id
      id++;
      exists = 0;
      }
    else
      {
      vtkDebugMacro("GetUniqueTransferID: in loop, returning id = " << id);
      return (id);
      }
    }
  vtkDebugMacro("GetUniqueTransferID: returning id = " << id);
  return ( id );
}

//----------------------------------------------------------------------------
void vtkKWRemoteIOManager::CreateProcessingThread()
{
  if (this->ProcessingThreadId == -1)
    {
    this->ProcessingThreadActiveLock->Lock();
    this->ProcessingThreadActive = true;
    this->ProcessingThreadActiveLock->Unlock();
    
    this->ProcessingThreadId = this->ProcessingThreader->SpawnThread(
              vtkKWRemoteIOManager::ProcessingThreaderCallback, this);

    // Start four network threads (TODO: make the number of threads a setting)
    this->NetworkingThreadIDs.push_back( this->ProcessingThreader
      ->SpawnThread(vtkKWRemoteIOManager::NetworkingThreaderCallback, this));

    /*
     * TODO: it looks like curl is not thread safe by default
     * - maybe there's a setting that cmcurl can have
     *   similar to the --enable-threading of the standard curl build
     *
    this->NetworkingThreadIDs.push_back ( this->ProcessingThreader
          ->SpawnThread(vtkKWRemoteIOManager::NetworkingThreaderCallback,
                    this) );
    this->NetworkingThreadIDs.push_back ( this->ProcessingThreader
          ->SpawnThread(vtkKWRemoteIOManager::NetworkingThreaderCallback,
                    this) );
    this->NetworkingThreadIDs.push_back ( this->ProcessingThreader
          ->SpawnThread(vtkKWRemoteIOManager::NetworkingThreaderCallback,
                    this) );
    */

    // Let's create a timer to notify us about the IO taking place on this
    // other thread, we've created.
    if (!this->TimerToken)
      {
      this->TimerToken = Tcl_CreateTimerHandler( 
                  this->TimerDelay, /* every 1000 ms */
                  vtkKWRemoteIOManagerTimerCallback,
                  (ClientData)this);
      }
    }
}

//----------------------------------------------------------------------------
void vtkKWRemoteIOManager::TerminateProcessingThread()
{
  if (this->ProcessingThreadId != -1 && this->ProcessingThreader)
    {
    this->ProcessingThreadActiveLock->Lock();
    this->ProcessingThreadActive = false;
    this->ProcessingThreadActiveLock->Unlock();

    this->ProcessingThreader->TerminateThread( this->ProcessingThreadId );
    this->ProcessingThreadId = -1;

    std::vector<int>::const_iterator idIterator;
    idIterator = this->NetworkingThreadIDs.begin();
    while (idIterator != this->NetworkingThreadIDs.end())
      {
      this->ProcessingThreader->TerminateThread( *idIterator );
      ++idIterator;
      }
    this->NetworkingThreadIDs.clear();

    // Remote timer callbacks.
    if ( this->TimerToken )
      {
      Tcl_DeleteTimerHandler( this->TimerToken );
      this->TimerToken = NULL;
      }    
    }
}


//----------------------------------------------------------------------------
VTK_THREAD_RETURN_TYPE vtkKWRemoteIOManager
::ProcessingThreaderCallback( void *arg )
{ 
#ifdef VTK_USE_WIN32_THREADS
  // Adjust the priority of this thread
  SetThreadPriority(GetCurrentThread(),
                    THREAD_PRIORITY_BELOW_NORMAL);
#endif

#ifdef VTK_USE_PTHREADS
  // Adjust the priority of all PROCESS level threads.  Not a perfect solution.
  nice(20);
#endif
    
  // pull out the reference to the appLogic
  vtkKWRemoteIOManager *self = (vtkKWRemoteIOManager*)
    (((vtkMultiThreader::ThreadInfo *)(arg))->UserData);

  // Tell the app to start processing any tasks slated for the
  // processing thread
  self->ProcessProcessingTasks();

  return VTK_THREAD_RETURN_VALUE;
}

//----------------------------------------------------------------------------
void vtkKWRemoteIOManager::ProcessProcessingTasks()
{
  int active = true;
  vtkSmartPointer<vtkKWRemoteIOTask> task = 0;
  
  while (active)
    {
    // Check to see if we should be shutting down
    this->ProcessingThreadActiveLock->Lock();
    active = this->ProcessingThreadActive;
    this->ProcessingThreadActiveLock->Unlock();

    if (active)
      {
      // pull a task off the queue
      this->ProcessingTaskQueueLock->Lock();
      if ((*this->InternalTaskQueue).size() > 0)
        {
        //std::cout << "Number of queued tasks: " 
        //  << (*this->InternalTaskQueue).size() << std::endl;
        
        // only handle processing tasks in this thread
        task = (*this->InternalTaskQueue).front();
        if ( task->GetType() == vtkKWRemoteIOTask::Processing )
          {
          (*this->InternalTaskQueue).pop();
          }
        else 
          {
          task = NULL;
          }
        }
      this->ProcessingTaskQueueLock->Unlock();
      
      // process the task (should this be in a separate thread?)
      if (task)
        {
        task->Execute();
        task = 0;
        }
      }

    // busy wait
    vtksys::SystemTools::Delay(100);
    }
}

//----------------------------------------------------------------------------
VTK_THREAD_RETURN_TYPE vtkKWRemoteIOManager
::NetworkingThreaderCallback( void *arg )
{
  
#ifdef VTK_USE_WIN32_THREADS
  // Adjust the priority of this thread
  SetThreadPriority(GetCurrentThread(),
                    THREAD_PRIORITY_BELOW_NORMAL);
#endif

#ifdef VTK_USE_PTHREADS
  // Adjust the priority of all PROCESS level threads.  Not a perfect solution.
  nice(20);
#endif
    
  // pull out the reference to the appLogic
  vtkKWRemoteIOManager *self = (vtkKWRemoteIOManager*)
    (((vtkMultiThreader::ThreadInfo *)(arg))->UserData);

  // Tell the app to start processing any tasks slated for the
  // processing thread
  self->ProcessNetworkingTasks();

  return VTK_THREAD_RETURN_VALUE;
}

//----------------------------------------------------------------------------
void vtkKWRemoteIOManager::ProcessNetworkingTasks()
{
  int active = true;
  vtkSmartPointer<vtkKWRemoteIOTask> task = 0;
  
  while (active)
    {
    // Check to see if we should be shutting down
    this->ProcessingThreadActiveLock->Lock();
    active = this->ProcessingThreadActive;
    this->ProcessingThreadActiveLock->Unlock();

    if (active)
      {
      // pull a task off the queue
      this->ProcessingTaskQueueLock->Lock();
      if ((*this->InternalTaskQueue).size() > 0)
        {
        // std::cout << "Number of queued tasks: " 
        //  << (*this->InternalTaskQueue).size() << std::endl;
        task = (*this->InternalTaskQueue).front();
        if ( task->GetType() == vtkKWRemoteIOTask::Networking )
          {
          (*this->InternalTaskQueue).pop();
          }
        else 
          {
          task = NULL;
          }
        }
      this->ProcessingTaskQueueLock->Unlock();
      
      // process the task (should this be in a separate thread?)
      if (task)
        {
        task->Execute();
        task = 0;
        }
      }

    // busy wait
    vtksys::SystemTools::Delay(100);
    }
}

//----------------------------------------------------------------------------
int vtkKWRemoteIOManager::ScheduleTask( vtkKWRemoteIOTask *task )
{
  //std::cout << "Scheduling a task ";

  // only schedule a task if the processing task is up
  this->ProcessingThreadActiveLock->Lock();
  int active = this->ProcessingThreadActive;
  this->ProcessingThreadActiveLock->Unlock();

  if (!active)
    {
    // We might not have created the processing thread yet. Let's try to
    // create it.
    this->CreateProcessingThread();
    }

  // only schedule a task if the processing task is up
  this->ProcessingThreadActiveLock->Lock();
  active = this->ProcessingThreadActive;
  this->ProcessingThreadActiveLock->Unlock();

  // If active is still false, we failed to create the processing thread.
  // Let's return false.

  if (active)
    {
    this->ProcessingTaskQueueLock->Lock();
    (*this->InternalTaskQueue).push( task );
    //std::cout << (*this->InternalTaskQueue).size() << std::endl;
    this->ProcessingTaskQueueLock->Unlock();
    
    return true;
    }

  // could not schedule the task
  return false;
}

//----------------------------------------------------------------------------
// The actual work of transferring data is done here.
void vtkKWRemoteIOManager::ApplyTransfer( void *clientdata )
{
  //--- get the DataTransfer from the clientdata
  vtkKWDataTransfer *dt = reinterpret_cast < vtkKWDataTransfer*> (clientdata);
  if ( dt == NULL )
    {
    vtkErrorMacro("ApplyTransfer: data transfer is null");
    return;
    }

  int asynchIO = dt->GetAsynchronous();

  const char *source = dt->GetSourceURI();
  vtkstd::string dest = dt->GetDestinationURI();
  if ( dt->GetTransferType() == vtkKWDataTransfer::RemoteDownload  )
    {
    //---
    //--- Download data
    //---
    vtkURIHandler *handler = dt->GetHandler();
    if ( handler != NULL && source != NULL )
      {
      int async_pending = 
        asynchIO && dt->GetTransferStatus() == vtkKWDataTransfer::Pending;
      if (async_pending)
        {
        dt->SetTransferStatusNoModify ( vtkKWDataTransfer::Running );
        }
      else
        {
        vtkDebugMacro(
          "ApplyTransfer: stage file read on the handler..., source = " 
            << source << ", dest = " << dest.c_str());
        }
      
      handler->StageFileRead( source, dest.c_str() );

      // If the downloaded files are .tar files, untar them.
      //
      size_t posTar = dest.rfind(".tar");
      size_t posTarGz = dest.rfind(".tar.gz");
      const bool isTar = 
        (posTar != vtkstd::string::npos && posTar == (dest.length()-4));
      const bool isTarGz = 
        (posTarGz != vtkstd::string::npos && posTarGz == (dest.length()-7));
      if ((isTar || isTarGz) &&
          !dt->GetDisableTar() && 
          vtksys::SystemTools::FileExists(dest.c_str()))
        {
        vtkstd::string destPath = 
          vtksys::SystemTools::GetFilenamePath(dest.c_str());

        // Extract the tar / tar.gz archive
        if (vtkKWRemoteIOUtilities::ExtractTar( 
              dest.c_str(), destPath.c_str(), isTarGz ))
          {
          // Remove the tar archive after untarring
          vtksys::SystemTools::RemoveFile(dest.c_str());
          }
        }

      if (async_pending)
        {
        dt->SetTransferStatusNoModify ( vtkKWDataTransfer::Completed );
        }
      }
    }

  else if ( dt->GetTransferType() == vtkKWDataTransfer::RemoteUpload  )
    {
    }
  else
    {
    vtkErrorMacro(
      "ApplyTransfer: unknonw transfer type " <<  dt->GetTransferType() );
    }
}

//----------------------------------------------------------------------------
void vtkKWRemoteIOManager::TimerCallback()
{
  if (this->TimerToken)
    {
    Tcl_DeleteTimerHandler( this->TimerToken );
    this->TimerToken = NULL;
    }

  const int n = this->DataTransferCollection->GetNumberOfItems();
  for ( int i=0; i < n; i++ )
    {
    vtkKWDataTransfer *dt = static_cast< vtkKWDataTransfer *>(
            this->DataTransferCollection->GetItemAsObject(i) );
    
   // Check if the transfer status changed.
    if (dt->CheckAndUpdateTransferStatus())
      {
      this->InvokeEvent( 
          vtkKWRemoteIOManager::TransferStatusChangedEvent, dt );
      }

    else if (dt->GetTransferStatus() == vtkKWDataTransfer::Running)
      {
      // Report some progress event
      this->InvokeEvent( 
          vtkKWRemoteIOManager::TransferUpdateEvent, dt );
      }
    }

  if (!this->TimerToken)
    {
    this->TimerToken = Tcl_CreateTimerHandler( 
                this->TimerDelay,
                vtkKWRemoteIOManagerTimerCallback,
                (ClientData)this);
    }
}

//----------------------------------------------------------------------------
void vtkKWRemoteIOManager::SetTransferStatusChangedCallback( 
      void (*f)(vtkObject *caller, unsigned long eid,
      void *clientdata, void *calldata), vtkObject *obj)
{
  this->TransferStatusChangedCommand->SetCallback(f);
  this->TransferStatusChangedCommand->SetClientData(obj);
}

//----------------------------------------------------------------------------
void vtkKWRemoteIOManager::SetTransferUpdateCallback( 
      void (*f)(vtkObject *caller, unsigned long eid,
      void *clientdata, void *calldata), vtkObject *obj)
{
  this->TransferUpdateCommand->SetCallback(f);
  this->TransferUpdateCommand->SetClientData(obj);
}