#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 20 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 ci_.

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

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

Definition at line 17 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 29 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 55 of file FileServiceProxy.h.

References finish(), and obtainURI_().

 {
   finish();
   return obtainURI_();
 }

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

private

Definition at line 72 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 39 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";
   }
 }