rsyncAsyncProcess.hpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_common_libdaq
 * @copyright (c) Copyright ESO 2022
 * All Rights Reserved
 * ESO (eso.org) is an Intergovernmental Organisation, and therefore special legal conditions apply.
 *
 * @brief daq::RsyncAsyncProcess and related class declarations.
 */
#ifndef DAQ_PROCESS_RSYNC_ASYNC_PROCESS_HPP
#define DAQ_PROCESS_RSYNC_ASYNC_PROCESS_HPP
#include <daq/config.hpp>
 
#include <chrono>
#include <optional>
#include <string>
#include <vector>
 
#include <boost/signals2/signal.hpp>
 
#include <daq/process/asyncProcess.hpp>
 
namespace daq {
 
/**
 * Options controlling rsync invocation.
 */
struct RsyncOptions {
    /**
     * Binary to use.
     */
    std::optional<std::string> rsync;
    /**
     * Enables rate-limiting in kb/s.
     */
    std::optional<unsigned> bw_limit;
    /**
     * Transfer with/without delta xfer algorithm.
     */
    std::optional<bool> whole_file;
 
    /**
     * --inplace/--no-inplace
     */
    std::optional<bool> inplace;
    /**
     * I/O timeout
     */
    std::optional<std::chrono::seconds> timeout;
};
 
/**
 * Describes file transfer progress,
 */
struct RsyncProgress {
    /**
     * Number of transferred bytes.
     */
    uint64_t transferred;
    /**
     * Progress as fraction of 1 (complete == 1.0)
     */
    float progress;
    /**
     * Transfer speed in bytes/sec.
     */
    float speed;
    /**
     * Estimated remaining time.
     */
    std::chrono::seconds remaining;
};
 
/**
 * More specialized version for `rsync` which also monitors transfer progress.
 */
class RsyncAsyncProcessIf : public virtual AsyncProcessIf {
public:
    enum class DryRun { Disabled = 0, Enabled };
    virtual ~RsyncAsyncProcessIf() {
    }
 
    /**
     * Progress update signal.
     */
    using SigProgress = boost::signals2::signal<void(pid_t, RsyncProgress const&)>;
 
    /**
     * Connect to progress signal.
     */
    virtual boost::signals2::connection ConnectProgress(SigProgress::slot_type const& slot) = 0;
};
 
/**
 * Parse progress update from rsync.
 *
 * @param line a single line in the format rsync outputs with option `--info=progress2`.
 * @returns optional with value if parsing suceeds, empty optional otherwise.
 */
std::optional<RsyncProgress> ParseRsyncProgress(std::string const& line) noexcept;
 
/**
 * Represents an rsync process as an asynchronous operation.
 *
 * Once constructed the operation is initiated (only once) with `Initiate()` which starts the
 * process and returns a boost::future object that will receive exit code when process terminates
 * *and* all output has been read.
 *
 * @note No signals will be emitted after future has received the value or exception.
 *
 * Operation can be aborted with `Abort()` which will terminate process and set future with
 * exceptional result.
 *
 * @ingroup daq_common_libdaq
 */
class RsyncAsyncProcess : public virtual RsyncAsyncProcessIf, public AsyncProcess {
public:
    virtual ~RsyncAsyncProcess();
 
    /**
     * Construct async operation.
     *
     * @note Does not start the process or any other asynchronous operations. This is done in
     * AsyncProcess::Initiate().
     *
     * @param ctx io_context instance to use.
     * @param source rsync file source.
     * @param dest rsync file dest.
     * @param opts rsync options.
     * @param flag Control whether rsync should do a dry-run (useful to test if connection is
     * possible).
     */
    RsyncAsyncProcess(boost::asio::io_context& ctx,
                      std::string source,
                      std::string dest,
                      RsyncOptions const& opts = {},
                      DryRun flag = DryRun::Disabled);
 
    /**
     * @name Signals
     */
    /// @{
    /**
     * Progress update signal.
     */
    using SigProgress = RsyncAsyncProcessIf::SigProgress;
 
    [[nodiscard]] boost::future<int> Initiate() override;
 
    /**
     * Connect to progress signal.
     */
    boost::signals2::connection ConnectProgress(SigProgress::slot_type const& slot) override;
    /// @}
private:
    SigProgress m_progress;
};
 
}  // namespace daq
 
#endif  // #ifndef DAQ_PROCESS_RSYNC_ASYNC_PROCESS_HPP