#include "FileServiceProxy.h"

Classes
struct	FileEntity

Public Member Functions
	FileServiceProxy (std::vector< std::string > &&fileNames, size_t attempts=5, double waitBetweenAttempts=5.0)

	~FileServiceProxy ()

std::string	next ()

void	finish ()

Private Member Functions
std::string	obtainURI_ ()

std::string	obtainFileFromURI_ ()

Private Attributes
ServiceHandle< CatalogInterface >	ci_ {}

ServiceHandle< FileTransfer >	ft_ {}

struct art::detail::FileServiceProxy::FileEntity	currentItem_

size_t const	attemptsPerPhase_

double const	waitBetweenAttempts_

Detailed Description

Definition at line 17 of file FileServiceProxy.h.

Constructor & Destructor Documentation

art::detail::FileServiceProxy::FileServiceProxy	(	std::vector< std::string > &&	fileNames,
		size_t	attempts = `5`,
		double	waitBetweenAttempts = `5.0`
	)

explicit

Definition at line 4 of file FileServiceProxy.cc.

References attemptsPerPhase_, ci_, and waitBetweenAttempts_.

   : currentItem_{attempts}
   , attemptsPerPhase_{attempts}
   , waitBetweenAttempts_{waitBetweenAttempts}
 {
   ci_->configure(std::move(fileNames));
 }

art::detail::FileServiceProxy::~FileServiceProxy ( )

Definition at line 15 of file FileServiceProxy.cc.

References ci_, currentItem_, art::detail::FileServiceProxy::FileEntity::ftStatus, art::INCOMPLETE, art::SKIPPED, art::SUCCESS, and art::detail::FileServiceProxy::FileEntity::uri.

 {
   if (!currentItem_.uri.empty()) {
     if (currentItem_.ftStatus == FileTransferStatus::SUCCESS) {
       ci_->updateStatus(currentItem_.uri, FileDisposition::INCOMPLETE);
     } else {
       ci_->updateStatus(currentItem_.uri, FileDisposition::SKIPPED);
     }
   }
 }

Member Function Documentation

void art::detail::FileServiceProxy::finish ( )

Definition at line 27 of file FileServiceProxy.cc.

References attemptsPerPhase_, ci_, art::CONSUMED, currentItem_, art::detail::FileServiceProxy::FileEntity::ftStatus, art::SUCCESS, and art::detail::FileServiceProxy::FileEntity::uri.

Referenced by next().

 {
   if (currentItem_.ftStatus == FileTransferStatus::SUCCESS) {
     // File is complete.
     ci_->updateStatus(currentItem_.uri, FileDisposition::CONSUMED);
   }
   currentItem_ = FileEntity(attemptsPerPhase_);
 }

std::string art::detail::FileServiceProxy::next ( )

inline

Definition at line 48 of file FileServiceProxy.h.

References finish(), and obtainURI_().

 {
   finish();
   return obtainURI_();
 }

std::string art::detail::FileServiceProxy::obtainFileFromURI_ ( )

private

Definition at line 70 of file FileServiceProxy.cc.

References attemptsPerPhase_, art::detail::FileServiceProxy::FileEntity::attemptsRemaining, art::errors::CatalogServiceError, ci_, currentItem_, ft_, art::detail::FileServiceProxy::FileEntity::ftStatus, art::PENDING, art::detail::FileServiceProxy::FileEntity::pfn, art::SUCCESS, art::TRANSFERRED, art::translateFileTransferStatus(), art::detail::FDS::UNAVAILABLE, art::detail::FileServiceProxy::FileEntity::uri, art::detail::FileServiceProxy::FileEntity::uriStatus, and waitBetweenAttempts_.

Referenced by obtainURI_().

 {
   assert(currentItem_.uriStatus == FileDeliveryStatus::SUCCESS);
   switch (currentItem_.ftStatus) {
   case FileTransferStatus::UNAVAILABLE:
     sleep(waitBetweenAttempts_);
     [[fallthrough]];
   case FileTransferStatus::PENDING:
     if (!currentItem_.attemptsRemaining--) {
       throw Exception(errors::CatalogServiceError)
         << "Unable to obtain URI " << currentItem_.uri
         << " from FileTransfer service after " << attemptsPerPhase_
         << " attempts.\n";
     }
     currentItem_.ftStatus = static_cast<FileTransferStatus>(
       ft_->translateToLocalFilename(currentItem_.uri, currentItem_.pfn));
     return obtainFileFromURI_();
   case FileTransferStatus::SUCCESS:
     ci_->updateStatus(currentItem_.uri, FileDisposition::TRANSFERRED);
     return currentItem_.pfn; // Done.
   default:
     throw Exception(errors::CatalogServiceError)
       << "FileTransfer service returned failure code "
       << translateFileTransferStatus(currentItem_.ftStatus) << " ("
       << static_cast<int>(currentItem_.ftStatus) << ") for URI "
       << currentItem_.uri << ".\n";
   }
 }

std::string art::detail::FileServiceProxy::obtainURI_ ( )

private

Definition at line 37 of file FileServiceProxy.cc.

References attemptsPerPhase_, art::detail::FileServiceProxy::FileEntity::attemptsRemaining, art::errors::CatalogServiceError, ci_, currentItem_, art::NO_MORE_FILES, obtainFileFromURI_(), art::PENDING, art::SUCCESS, art::translateFileDeliveryStatus(), art::detail::FDS::TRY_AGAIN_LATER, art::detail::FDS::UNAVAILABLE, art::detail::FileServiceProxy::FileEntity::uri, art::detail::FileServiceProxy::FileEntity::uriStatus, and waitBetweenAttempts_.

Referenced by next().

 {
   double wait = 0.0;
   switch (currentItem_.uriStatus) {
   case FileDeliveryStatus::TRY_AGAIN_LATER:
     [[fallthrough]];
   case FileDeliveryStatus::UNAVAILABLE:
     wait = waitBetweenAttempts_;
     [[fallthrough]];
   case FileDeliveryStatus::PENDING:
     if (!currentItem_.attemptsRemaining--) {
       throw Exception(errors::CatalogServiceError)
         << "Unable to obtain URI from CatalogInterface service after "
         << attemptsPerPhase_ << " attempts.\n";
     }
     currentItem_.uriStatus = static_cast<FileDeliveryStatus>(
       ci_->getNextFileURI(currentItem_.uri, wait));
     return obtainURI_();
   case FileDeliveryStatus::SUCCESS:
     currentItem_.attemptsRemaining = attemptsPerPhase_; // Reset.
     return obtainFileFromURI_();
     break;
   case FileDeliveryStatus::NO_MORE_FILES:
     return std::string(); // Done.
   default:
     throw Exception(errors::CatalogServiceError)
       << "CatalogInterface service returned failure code "
       << translateFileDeliveryStatus(currentItem_.uriStatus) << " ("
       << static_cast<int>(currentItem_.uriStatus) << ").\n";
   }
 }