// FileManagementSystem.cxx

#include "dataset_file/FileManagementSystem.h"
#include <iostream>
#include <sstream>
#include <unistd.h>
#include <iomanip>
#include "dataset_util/getcwd.h"
#include "dataset_util/mkdir.h"
#include "dataset_util/FileStatus.h"
#include "dataset_util/FileName.h"
#include "dataset_util/Text.h"
#include "dataset_util/SystemCommand.h"
#include "dataset_credential/SubprocessCredential.h"

using std::string;
using std::ostream;
using std::cerr;
using std::endl;
using std::istringstream;
using std::ostringstream;
using std::setw;
using std::setfill;
using dset::Url;
using dset::SubprocessCredential;
using dset::FileCatalog;
using dset::FileManagementSystem;

typedef FileManagementSystem::Name Name;
typedef FileManagementSystem::NameList NameList;
typedef FileManagementSystem::FCList FCList;
typedef FileCatalog::UrlList UrlList;

//**********************************************************************
// Local definitions.
//**********************************************************************
namespace {
//**********************************************************************

// Convert comma separated string to name list.

NameList string_to_names(string line) {
  NameList names;
  Text::split(line, names, ",");
  return names;
}

//**********************************************************************

// Convert a list of names into a comma-separated string.

string names_to_string(const NameList& names) {
  string line;
  for ( NameList::const_iterator inam=names.begin();
        inam!=names.end(); ++inam) {
    if ( inam != names.begin() ) line += ",";
    line += *inam;
  }
  return line;
}

//**********************************************************************

// Use a list of names to fill a list of file catalogs.
// Returns nonzero for error.

int names_to_fcs(const NameList& names, FCList& fcs) {
  string prefix = "FileManagementSystem::names_to_fcs: ";
  for ( NameList::const_iterator inam=names.begin();
        inam!=names.end(); ++inam) {
    Name type = *inam;
    Name name = "";
    string::size_type ipos = type.find(":");
    if ( ipos != string::npos ) {
      name = type.substr(ipos+1);
      type = type.substr(0, ipos);
    }
    FileCatalog* pfc = FileCatalog::locate(type, name);
    if ( pfc == 0 ) {
      cerr << prefix << "Unable to locate file catalog:" << endl;
      cerr << "  " << *inam << endl;
      return 2;
    }
    fcs.push_back(pfc);
  }
  return 0;
}

//**********************************************************************

// Filter the list of protocols.
NameList
filtered_protocols(const NameList allowed, const NameList& order) {
  string prefix = "FileManagementSystem::filtered_protocols: ";
  NameList prots;
  for ( NameList::const_iterator ipro=order.begin();
        ipro!=order.end(); ++ipro ) {
    Name ext_prot = *ipro;
    Name base_prot = *ipro;
    // If the extended protocol has a colon, strip off the extension.
    string::size_type ipos = ext_prot.find(":");
    if ( ipos != string::npos ) {
      base_prot = ext_prot.substr(0,ipos);
    }
    // Remove aliases.
    if ( base_prot == "copy" ) base_prot = "file";
    // Check if the base protocol is allowed.
    if ( find(allowed.begin(), allowed.end(), base_prot)
         != allowed.end() ) {
      prots.push_back(ext_prot);
    }
  }
  if ( prots.size() == 0 ) {
    cerr << prefix << "No matching protocols found" << endl;
  }
  return prots;
}

//**********************************************************************

// Return if a URL matches an extended protocol.

bool match(Url url, string xprot) {
  if ( url.size() < xprot.size() ) return false;
  return url.to_string().substr(0, xprot.size()) == xprot;
}

//**********************************************************************

// Get/set the default instance.

FileManagementSystem& default_fms(FileManagementSystem* pfms =0) {
  static FileManagementSystem* pval = 0;
  if ( pfms != 0 ) {
    pval = pfms;
  }
  if ( pval == 0 ) {
    pval = new FileManagementSystem(Environment::current());
  }
  return *pval;
}
  
//**********************************************************************
}  // end unnamed namespace
//**********************************************************************
// Static methods.
//**********************************************************************

// Simple instance.

int FileManagementSystem::create_simple_instance() {
  string sedir = getcwd();
  sedir += "/fms_se";
  mkdir(sedir);
  if ( ! FileStatus(sedir).is_directory() ) {
    return 1;
  }
  Environment env;
  env["DIAL_FMS_PROTOCOLS"] = "file";
  env["DIAL_FMS_PROTOCOL_ORDER"] = "file";
  env["DIAL_FMS_SE"] = "file:" + sedir;
  env["DIAL_FMS_VERBOSE"] =
    Environment::current().value("DIAL_FMS_VERBOSE");
  static FileManagementSystem fms(env);
  fms.set_default();
  return 0;
}
  
//**********************************************************************

// Default instance.

FileManagementSystem& FileManagementSystem::default_instance() {
  return default_fms();
}
  
//**********************************************************************
// Private methods.
//**********************************************************************

// Put entry in cache.
void FileManagementSystem::cache_put(Url purl, Url lurl) {
  // Fetch the modification time for purl.
  string pfn = purl.fullpath();
  if ( ! pfn.size() ) return;
  Time modtime = Time(FileStatus(pfn).modtime());
  if ( ! modtime.is_valid() ) return;
  m_urls[lurl] = purl;
  m_mods[lurl] = modtime;
}

//**********************************************************************

// Get entry from cache.
Url FileManagementSystem::cache_get(Url lurl) {
  UrlMap::iterator iurl = m_urls.find(lurl);
  TimeMap::iterator itim = m_mods.find(lurl);
  if ( iurl == m_urls.end() ) return Url();
  if ( itim == m_mods.end() ) return Url();
  Url purl = iurl->second;
  string pfn = purl.fullpath();
  if ( ! pfn.size() ) return Url();
  Time modtime = Time(FileStatus(pfn).modtime());
  Time oldmodtime = itim->second;
  if ( modtime.unix_time() != oldmodtime.unix_time() ) {
    m_urls.erase(iurl);
    m_mods.erase(itim);
    return Url();
  }
  return purl;
}

//**********************************************************************

// Retrieve a file.
// Implementation shared by get and copy.

Url FileManagementSystem::
internal_get(Url inurl, string prefix, const NameList& prots,
             bool make_copy) {
  if ( ! is_valid() ) {
    cerr << prefix << error_message() << endl;
    return Url();
  }
  m_err = 0;
  bool dbg = verbose();
  // Fetch the first extended protocol. We can exit as soon as we match
  // this.
  if ( prots.size() == 0 ) {
    m_err = 101;
    cerr << long_error_message() << endl;
    return Url();
  }
  if ( ! inurl.is_valid() ) {
    m_err = 102;
    cerr << long_error_message() << endl;
    return Url();
  }
  // Extract the primary protocol (first in the list). As soon as we
  // match this, we are done.
  string xprot = prots.front();
  if ( dbg ) {
    if ( make_copy ) {
      cerr << "FMS: Begin copy" << endl;
    } else {
      cerr << "FMS: Begin get/copy" << endl;
    }
    cerr << "FMS:   Input: " << inurl << endl;
    cerr << "FMS:   First XPROT: " << xprot << endl;
  }
  if ( ! make_copy ) {
    if ( dbg ) {
      cerr << "FMS:   Checking cache.." << endl;
    }
    // If the input URL has the primary protocol, use it.
    if ( match(inurl, xprot) ) return inurl;
    // If there is a cached URL, use it.
    Url curl = cache_get(inurl);
    if ( curl.size() ) return curl;
    if ( dbg ) {
      cerr << "FMS:     No match" << endl;
    }
  }
  // Create list of candidate urls.
  UrlList canurls;
  canurls.push_back(inurl);
  string prot = inurl.prefix();
  string name = inurl.suffix();
  // Resolve LFN or GUID.
  if ( dbg ) {
    cerr << "FMS:   Resolving LFN's and GUID's" << endl;
  }
  if ( prot == "lfn" || prot == "guid" ) {
    for ( FCList::const_iterator ifc=file_catalogs().begin();
          ifc!=file_catalogs().end(); ++ifc ) {
      FileCatalog* pfc = *ifc;
      if ( pfc == 0 ) continue;
      UrlList newurls = pfc->replicas(inurl);
      for ( UrlList::const_iterator iurl=newurls.begin();
            iurl!=newurls.end(); ++iurl ) {
        if ( dbg ) {
          cerr << "FMS:     New: " << *iurl << endl;
        }
        canurls.push_back(*iurl);
        Url outurl = *iurl;
        if ( match(outurl, xprot) ) {
          cache_put(outurl, inurl);
          return outurl;
        }
      }
    }
  }

  // Filter candidate list.
  string com = input_filter();
  if ( com.size() ) {
    if ( dbg ) {
      cerr << "FMS:   Filtering" << endl;
    }
    UrlList newurls;
    for ( UrlList::size_type iurl=0; iurl<canurls.size(); ++iurl ) {
      Url curl = canurls[iurl];
      if ( dbg ) {
        cerr << "FMS:     " << curl << endl;
      }
      // Run command to filter URL.
      SystemCommand fcom(com + " " + curl.to_string() );
      int fstat = fcom.runerr();
      if ( fstat ) {
        cerr << prefix << "Error running filter:" << endl;
        cerr << fcom << endl;
        return Url();
      }
      // Loop over output lines.
      Text out = fcom.out();
      for ( int iline=0; iline<out.size(); ++iline ) {
        string surl = out.line(iline);
        if ( dbg ) {
          cerr << "FMS:     " << surl << endl;
        }
        Url url(surl);
        if ( ! url.is_valid() ) {
          cerr << prefix << "Invalid filter URL:" << endl;
          cerr << "  " << surl << endl;
          cerr << fcom << endl;
          return Url();
        }
        newurls.push_back(url);
      }
    }
    canurls = newurls;
  }
        
  // For srm or gsiftp, resolve to dcap if the path begins with
  // /pnfs and is locally accessible.
  if ( dbg ) {
    cerr << "FMS:   Resolving SRM" << endl;
  }
  for ( UrlList::size_type iurl=0; iurl<canurls.size(); ++iurl ) {
    Url url = canurls[iurl];
    if ( url.prefix() == "srm" || url.prefix() == "gsiftp") {
      string path = url.fullpath();
      if ( path.size()>5  && path.substr(0,5)=="/pnfs" ) {
        string pname = url.fullpath();
        if ( FileStatus(pname).exists() ) {
          Url newurl("dcap:" + pname);
          if ( dbg ) {
            cerr << "FMS:     New: " << newurl << endl;
          }
          if ( match(newurl, xprot) ) {
            cache_put(newurl, inurl);
            return newurl;
          }
          canurls.push_back(newurl);
        }
      }
    }
  }
  // Next resolve dcap.
  // Next resolve gsiftp.
  // Next resolve file.
  // Now loop over extended protocols and candidate URL's and
  // look for matches.

  if ( dbg ) {
    cerr << "FMS:   Loop over XPROT's and resolved URL's" << endl;
  }
  for ( NameList::const_iterator ixpr=prots.begin();
        ixpr!=prots.end(); ++ixpr ) {
    Name xprot = *ixpr;
    Name xprot_prot = xprot;
    Name copy_dir = "";
    string::size_type ipos = xprot.find(":");
    if ( ipos != string::npos ) {
      xprot_prot = xprot.substr(0, ipos);
      copy_dir = xprot.substr(ipos+1);
    }
    for ( UrlList::const_iterator iurl=canurls.begin();
          iurl!=canurls.end(); ++iurl ) {
      Url url = *iurl;
      if ( dbg ) {
        cerr << "FMS:     XPR: " << xprot << endl;
        cerr << "FMS:     URL: " << url << endl;
      }
      // If the output protocol is copy, check the associated directory and
      // build command to copy the file. This command depends on the input
      // protocol.
      if ( xprot_prot == "copy" ) {
        if ( dbg ) cerr << "FMS:       Trying copy..." << endl;
        if ( copy_dir.size() == 0 ) {
          copy_dir = getcwd();
        }
        FileStatus fs(copy_dir);
        if ( fs.is_directory() && fs.is_writeable() ) {
          string url_prot = url.prefix();
          string url_name = url.suffix();
          string::size_type kpos = url_name.rfind("/");
          string copy_file = copy_dir + "/" + url_name.substr(kpos+1);
          // If we are making a copy for modification or if the file
          // already exists, then append a copy suffix to the destination
          // file name.
          if ( make_copy || FileStatus(copy_file).exists() ) {
            string copy_file_base = copy_file;
            copy_file = "";
            // If a copy suffix is already present, drop it from the base
            // name and uset it to set the new copy number.
            int icopy = 1;
            string::size_type isuf = copy_file_base.size() - 9;
            if ( copy_file_base.substr(isuf,5) == ".copy" ) {
              istringstream ssicopy(copy_file_base.substr(isuf+5));
              ssicopy >> icopy;
              ++icopy;
              copy_file_base = copy_file_base.substr(0, isuf);
            }
            // Append suffix to ensure file does not exist.
            for ( int i=icopy; i<9999; ++i ) {
              ostringstream sssuff;
              sssuff << ".copy" << setw(4) << setfill('0') << i;
              string filename = copy_file_base + sssuff.str();
              if ( ! FileStatus(filename).exists() ) {
                copy_file = filename;
                break;
              }
            }
            if ( copy_file.size() == 0 ) {
              cerr << prefix << "Too many files with base " << endl;
              cerr << "  " << copy_file_base << endl;
              return Url();
            }
          }
          string copy_url = "file://" + copy_file;
          Url outurl("file:" + copy_file);
          if ( dbg ) {
            cerr << "FMS:         From: " << url << endl;
            cerr << "FMS:           To: " << outurl << endl;
          }
          string command;
          if ( url_prot == "file" ) {
            command = "cp " + url_name + " " + copy_file;
          } else if ( url_prot == "gsiftp" ) {
            command = "globus-url-copy " + url.to_string() + " " + copy_url;
          } else if ( url_prot == "dcap" ) {
            command = "dccp " + url_name + " " + copy_file;
          } else if ( url_prot == "rfio" ) {
            command = "rfcp " + url_name + " " + copy_file;
          } else if ( url_prot == "srm" ) {
            string srmurl = url.to_string();
            if ( url.port() == "" ) {
              string::size_type ipos = srmurl.find("/", 6);
              if ( srmurl.find(":", 6, ipos-1) == string::npos ) {
                srmurl = srmurl.substr(0,ipos) + ":8443" +
                         srmurl.substr(ipos);
              }
            }
            command = "srmcp " + srmurl + " file:///" + copy_file;
          }
          if ( command.size() ) {
            if ( dbg ) {
              cerr << "FMS:         " << command << endl;
              cerr << "FMS:         " << copy_file << endl;
            }
            // Make sure destination file is not the same as the original.
            if ( copy_file == url_name ) {
               cerr << prefix
                    << "Copy source and destination are the same:" << endl;
               cerr << "  " << copy_file << endl;
            } else {
              // Make sure destination file is not already present.
              unlink(copy_file.c_str());
            }
            if ( FileStatus(copy_file).exists() ) {
              cerr << prefix << "Unable to remove existing file:" << endl;
              cerr << "  " << copy_file << endl;
            } else {
              SystemCommand scom(command);
              scom.runout();
              if ( FileStatus(copy_file).exists() ) {
                cache_put(outurl, inurl);
                return outurl;
              } else {
                cerr << prefix << "Unable to copy file:" << endl;
                cerr << scom << endl;
              }
            }
          } else {
            if ( dbg ) {
              cerr << "FMS:         Unable to find copy command"
                   << endl;
            }
          }
        } else {
          cerr << prefix << "Unable to write to copy directory:"
               << endl;
          cerr << "  " << copy_dir << endl;
        }
      // Check if the URL is a direct match to the protocol.
      } else {
        if ( dbg ) cerr << "FMS:       Trying direct match..." << endl;
        if ( match(url, xprot) ) {
          cache_put(url, inurl);
          return url;
        }
      }
    }
  }
  // No match found.
  m_err = 103;
  return Url();
}

//**********************************************************************
// Public methods.
//**********************************************************************

// Default constructor.

FileManagementSystem::FileManagementSystem()
: m_err(1), m_verbose(0) { }

//**********************************************************************

// Environment constructor.

FileManagementSystem::FileManagementSystem(const Environment& env) 
: m_err(0) {
  string sfcs = env.value("DIAL_FMS_CATALOG");
  string sprots = env.value("DIAL_FMS_PROTOCOLS");
  string sorder = env.value("DIAL_FMS_PROTOCOL_ORDER");
  m_filter = env.value("DIAL_FMS_FILTER");
  m_se = Url(env.value("DIAL_FMS_SE"));
  m_se_alias = env.value("DIAL_FMS_SE_ALIAS");
  string ssefcs = env.value("DIAL_FMS_SE_CATALOG");
  if ( ssefcs.find(",") != string::npos ) {
    cerr << "FileManagementSystem::ctor: "
         << "Ignoring invalid value for DIAL_FMS_SE_CATALOG" << endl;
    ssefcs = "";
  }
  m_verbose = env.value("DIAL_FMS_VERBOSE").size();
  // Locate input file catalogs.
  NameList fcs = string_to_names(sfcs);
  m_err = names_to_fcs(fcs, m_fcs);
  // Locate output file catalogs.
  NameList sefcs = string_to_names(ssefcs);
  m_err = names_to_fcs(sefcs, m_sefcs);
  if ( m_err ) return;
  assert( m_sefcs.size() < 2 );
  // Build protocol lists.
  m_allowed = string_to_names(sprots);
  m_order = string_to_names(sorder);
  m_prots = filtered_protocols(m_allowed, m_order);
}

//**********************************************************************

// Set default.

int FileManagementSystem::set_default() {
  default_fms(this);
  return 0;
}

//**********************************************************************

// Verbose.

bool FileManagementSystem::verbose() const {
  return m_verbose || FileStatus("debug_FileManagementSystem").exists();
}

//**********************************************************************

// Validity.

bool FileManagementSystem::is_valid() const {
  return m_err == 0 || m_err >10;
}

//**********************************************************************

// Error code.

int FileManagementSystem::error() const {
  return m_err;
}

//**********************************************************************

// Error string.

string FileManagementSystem::error_message() const {
  switch(m_err) {
  case 0: return "";
  case 1: return "ctor: Default FMS";
  case 2: return "ctor: Invalid file catalog name";
  case 101: return "get: Filtered protocol list is empty";
  case 102: return "get: Input URL is invalid";
  case 103: return "get: No match found";
  case 201: return "put: Storage element is invalid";
  case 202: return "put: Input URL is invalid";
  case 203: return "put: Unable to extract input basename";
  case 204: return "put: Unable to find a transfer protocol";
  case 205: return "put: Destination file already exists";
  case 206: return "put: File transfer failed";
  case 207: return "put: Unable to find file after transfer";
  }
  return "no_method: Unknown error";
}

//**********************************************************************

// Long error string.

string FileManagementSystem::long_error_message() const {
  if ( m_err != 0 ) {
    return "FileManagementSystem::" + error_message();
  }
  return "";
}

//**********************************************************************

// Replica catalogs.

const FCList& FileManagementSystem::file_catalogs() const {
  return m_fcs;
}

//**********************************************************************

// Allowed protocols.

const NameList& FileManagementSystem::allowed_protocols() const {
  return m_allowed;
}

//**********************************************************************

// Allowed protcols string.

string FileManagementSystem::allowed_protocols_string() const {
  return names_to_string(m_allowed);
}

//**********************************************************************

// Protcol order.

const NameList& FileManagementSystem::protocol_order() const {
  return m_order;
}

//**********************************************************************

// Protcol order string.

string FileManagementSystem::protocol_order_string() const {
  return names_to_string(m_order);
}

//**********************************************************************

// Actual protocols.

const NameList& FileManagementSystem::protocols() const {
  return m_prots;
}

//**********************************************************************

// Input filter.

Name FileManagementSystem::input_filter() const {
  return m_filter;
}

//**********************************************************************

// Storage element.

Url FileManagementSystem::storage_element() const {
  return m_se;
}

//**********************************************************************

// SE alias.

Name FileManagementSystem::se_alias() const {
  return m_se_alias;
}

//**********************************************************************

// SE file catalog.

FileCatalog* FileManagementSystem::se_file_catalog() const {
  if ( m_sefcs.size() == 0 ) return 0;
  assert( m_sefcs.size() == 1 );
  return m_sefcs.front();
}

//**********************************************************************

// Get.

Url FileManagementSystem::get(Url inurl) {
  string prefix = "FileManagementSystem::get ";
  return internal_get(inurl, prefix, protocols(), false);
}

//**********************************************************************

// Copy.

Url FileManagementSystem::copy(Url inurl, Url outurl) {
  string prefix = "FileManagementSystem::copy ";
  bool dbg = verbose();
  if ( ! outurl.is_valid() ) {
    cerr << prefix << "Invalid output directory URL" << endl;
    return Url();
  }
  if ( outurl.prefix() != "file" ) {
    cerr << prefix << "Invalid output protocol:" << endl;
    cerr << "  " << outurl << endl;
    return Url();
  }
  // The output URL must point to an existing directory or a 
  // nonexisting file in an existing directory.
  // an existing directory.
  string path = outurl.fullpath();
  string basename;
  if ( ! FileStatus(path).exists() ) {
    FileName fname(path);
    path = fname.directory_name();
    basename = fname.nopath_name();
  }
  if ( ! FileStatus(path).is_directory() ) {
    cerr << prefix << "URL path is not an existing directory:" << endl;
    cerr << "  " << outurl << endl;
    return Url();
  }
  Url desturl("copy:" + path);
  NameList prots;
  prots.push_back(desturl.to_string());
  Url geturl = internal_get(inurl, prefix, prots, true);
  // If a filename was provided, then rename the file.
  if ( geturl.is_valid() && basename.size() ) {
    if ( dbg ) {
      cerr << "FMS:   Renaming outpur file" << endl;
    }
    string oldpath = geturl.fullpath();
    string newpath = path + "/" + basename;
    int lstat = link(oldpath.c_str(), newpath.c_str());
    if ( lstat ) {
      cerr << prefix << "Link failed" << endl;
      cerr << "  " << oldpath << endl;
      cerr << "  " << newpath << endl;
    } else {
      int ustat = unlink(oldpath.c_str());
      if ( ustat ) {
        cerr << prefix << "Unlink failed" << endl;
        cerr << "  " << oldpath << endl;
      }
    }
    geturl = Url("file:" + newpath);
  }
  return geturl;
}

//**********************************************************************

// Put a file into the SE.

Url FileManagementSystem::put(Url inurl) {

  string prefix = "FileManagementSystem::put:";
  if ( ! is_valid() ) {
    cerr << prefix << error_message() << endl;
    return Url();
  }
  Url se = storage_element();
  if ( ! se.is_valid() ) {
    m_err = 201;
    cerr << long_error_message() << endl;
    return Url();
  }
  if ( ! inurl.is_valid() ) {
    m_err = 202;
    cerr << long_error_message() << endl;
    return Url();
  }
  m_err = 0;
  bool dbg = verbose();
  if ( dbg ) {
    cerr << "FMS: Begin put" << endl;
    cerr << "FMS:   Input: " << inurl << endl;
    cerr << "FMS:   SE: " << se << endl;
  }

  // If there is an SE file catalog, temporarily register the file
  // there to generate a logical URL.
  Url lurl;
  Url inlurl;
  unsigned int lifetime;
  FileCatalog* pfc = se_file_catalog();
  if ( pfc != 0 ) {
    lurl = pfc->insert(inurl, inlurl, lifetime);
    if ( ! lurl.is_valid() ) {
      cerr << prefix << "Initial FC registration failed" << endl;
      return Url();
    }
  }
  bool err = false;

  // Input protocol.
  string inprot = inurl.prefix();
  // SE protocol.
  string seprot = se.prefix();
  // Output URL.
  string bname = inurl.basename();
  if ( ! bname.size() ) {
    m_err = 203;
    cerr << long_error_message() << endl;
    cerr << "  Input URL: " << inurl << endl;
    if ( pfc != 0 ) pfc->remove(inurl, lurl);
    return Url();
  }
  Url outurl(se.to_string() + "/" + bname);
  assert( outurl.is_valid() );

  // Fetch subprocess credential.
  SubprocessCredential scred;

  // Build commands to list and check destination file and choose
  // a transfer protocol.
  string slist;
  string scheck;
  string xfprot;
  if ( seprot == "file" ) {
    slist = "ls " + outurl.fullpath();
    scheck = slist;
    if ( inprot == "file" ) xfprot = "file";
  } else if ( seprot == "dcap" ) {
    slist = "ls " + outurl.fullpath();
    scheck = slist;
    if ( inprot == "file" ) xfprot = "dcap";
  } else if ( seprot == "gsiftp" ) {
    slist = scred.env() + "edg-gridftp-exists " + outurl.to_string();
    scheck = "echo " + outurl.to_string() + "; " + slist;
    if ( inprot == "file" ) xfprot = "gsiftp";
  } else if ( seprot == "srm" ) {
    slist = scred.env() + "srm-get-metadata -retry_num=0 ";
    slist += outurl.to_string();
    scheck = scred.env() + "srm-get-metadata ";
    scheck += outurl.to_string();
    if ( inprot == "file" ) xfprot = "srm";
  }
  // Error if no transfer protocol is found.
  if ( xfprot == "" ) {
    cerr << prefix << "Unable to to transfer " << inprot
         << " to " << seprot << endl;
    m_err = 204;
    if ( pfc != 0 ) pfc->remove(inurl, lurl);
    return Url();
  }
  if ( dbg ) {
    cerr << "FMS:   Transfer protocol is " << xfprot << endl;
    cerr << "FMS:   Destination list command:" << endl;
    cerr << "         " << slist << endl;
  }
  assert( slist.size() );

  // Build commands to transfer file--depends on transfer protocol.
  string sxfer;
  // ...file
  if ( xfprot == "file" ) {
    sxfer = "cp " + inurl.fullpath() + " " + outurl.fullpath();
  } else if ( xfprot == "dcap" ) {
    sxfer = "dccp " + inurl.fullpath() + " " + outurl.fullpath();
  } else if ( xfprot == "gsiftp" ) {
    sxfer = scred.env() + "globus-url-copy -p 5 " +
            inurl.to_string() + " " + outurl.to_string();
  } else if ( xfprot == "srm" ) {
    sxfer= scred.env() + "srmcp ";
    if ( inprot == "file" ) {
      sxfer += "file:///" + inurl.fullpath();
    } else {
      sxfer += inurl.to_string();
    }
    sxfer += " " + outurl.to_string();
  }
  if ( dbg ) {
    cerr << "FMS:   Transfer command:" << endl;
    cerr << "         " << sxfer << endl;
  }

  // Check destination file does not exist.
  // We assume missing file returns 1 and has no stdout.
  // OK for posix and srm.
  SystemCommand list(slist);
  list.runerr();
  if ( list.status()!=1 || list.out().size() ) {
    m_err = 205;
    cerr << long_error_message() << endl;
    cerr << list << endl;
    if ( pfc != 0 ) pfc->remove(inurl, lurl);
    return Url();
  }

  // Run transfer command.
  SystemCommand xfer(sxfer);
  xfer.runout();
  if ( xfer.status() != 0 ) {
    m_err = 206;
    cerr << long_error_message() << endl;
    cerr << xfer << endl;
    if ( pfc != 0 ) pfc->remove(inurl, lurl);
    return Url();
  }

  // Check destination file now does exist.
  // Assume exiting file returns 0 and has stdout.
  SystemCommand check(scheck);
  check.runerr();
  if ( check.status()!=0 || check.out().size()==0 ) {
    m_err = 207;
    cerr << long_error_message() << endl;
    cerr << check << endl;
    if ( pfc != 0 ) pfc->remove(inurl, lurl);
    return Url();
  }

  // Update URL with alias.
  string alias = se_alias();
  if ( alias.size() ) {
    string sout = outurl.to_string();
    string::size_type osize = sout.size();
    string::size_type ssize = se.size();
    if ( osize < ssize || sout.substr(0,ssize) != se.to_string() ) {
      cerr << prefix << "Unable to update SE URL with alias." << endl;
      cerr << "    URL: " << outurl << endl;
      cerr << "     SE: " << se << endl;
      cerr << "  alias: " << alias << endl;
    } else {
      if ( dbg ) {
        cerr << "FMS:   Setting SE alias URL" << endl;
        cerr << "FMS:     SE URL: " << outurl << endl;
      }
      sout.replace(0, ssize, alias);
      outurl = Url(sout);
      if ( dbg ) {
        cerr << "FMS:      alias: " << outurl << endl;
      }
    }
  }

  // If SE file catalog is defined, replace registration of the
  // source URL with the destination URL.
  if ( pfc != 0 ) {
    outurl = pfc->insert(outurl, lurl, lifetime);
    if ( ! lurl.is_valid() ) {
      cerr << prefix << "Initial FC registration failed" << endl;
      return Url();
    }
    pfc->remove(inurl, lurl);
  }

  // Cache the associateion beteen the input and output URL's.
  cache_put(inurl, outurl);

  // Return the output URL.
  return outurl;
}

//**********************************************************************

// Release.

int FileManagementSystem::release(Url purl) {
  bool dbg = verbose();
  if ( dbg ) {
    cerr << "FMS: Begin release" << endl;
    cerr << "FMS:   " << purl << endl;
  }
  return 0;
}

//**********************************************************************

// Display cache.

void FileManagementSystem::display_cache(ostream& sout) {
  sout << "FMS cache has " << m_urls.size() << " entr";
  if ( m_urls.size() == 1 ) {
    sout << "y:";
  } else if ( m_urls.size() == 0 ) {
    sout << "ies";
  } else {
    sout << "ies:";
  }
  sout << endl;
  for ( UrlMap::const_iterator iurl=m_urls.begin();
        iurl!=m_urls.end(); ++iurl ) {
    Url lurl = iurl->first;
    Url purl = iurl->second;
    TimeMap::const_iterator itim = m_mods.find(lurl);
    Time modtime;
    if ( itim != m_mods.end() ) {
      modtime = itim->second;
    }
    sout << "  " << lurl << endl;
    sout << "    " << purl << endl;
    sout << "    Mod time: " << modtime << endl;
  }
}

//**********************************************************************
// Free functions.
//**********************************************************************

// Output stream.

ostream& operator<<(ostream& lhs, const FileManagementSystem& rhs) {
  if ( ! rhs.is_valid() ) {
    lhs << "Invalid FMS: " << rhs.error_message();
    return lhs;
  }
  lhs << "FileManagementSystem";
  lhs << "\n  Allowed protocols:";
  for ( NameList::const_iterator inam=rhs.allowed_protocols().begin();
        inam!=rhs.allowed_protocols().end(); ++inam ) {
     lhs << "\n    " << *inam;
  }
  lhs << "\n  Protocol order:";
  for ( NameList::const_iterator inam=rhs.protocol_order().begin();
        inam!=rhs.protocol_order().end(); ++inam ) {
     lhs << "\n    " << *inam;
  }
  lhs << "\n  Combined protocols:";
  for ( NameList::const_iterator inam=rhs.protocols().begin();
        inam!=rhs.protocols().end(); ++inam ) {
     lhs << "\n    " << *inam;
  }
  lhs << "\n  Input replica file catalogs";
  if ( rhs.file_catalogs().size() == 0 ) {
    lhs << " are not defined";
  } else {
    lhs << ":";
    for ( FCList::const_iterator ifc=rhs.file_catalogs().begin();
          ifc!=rhs.file_catalogs().end(); ++ifc ) {
      lhs << "\n    " << **ifc;
    }
  }
  lhs << "\n  Input filter";
  if ( rhs.input_filter().size() ) {
    lhs << ": " << rhs.input_filter();
  } else {
    lhs << " is not defined";
  }
  lhs << "\n  Storage element";
  Url se = rhs.storage_element();
  if ( se.is_valid() ) {
    lhs << ":\n    " << se;
  } else {
    lhs << " is invalid";
  }
  lhs << "\n  SE alias";
  if ( rhs.se_alias().size() ) {
    lhs << ": " << rhs.se_alias();
  } else {
    lhs << " is not defined";
  }
  lhs << "\n  SE replica file catalog";
  if ( rhs.se_file_catalog() == 0 ) {
    lhs << " is not defined";
  } else {
    lhs << ": ";
    lhs << "\n    " << *rhs.se_file_catalog();
  }
  return lhs;
}

//**********************************************************************