followRequestProvider.h

Go to the documentation of this file.

#ifndef __followRequestProvider_h__
#define __followRequestProvider_h__
#include "copyRequestTypes.h"
#include "requestProvider.h"
#include "errMsgQueue.h"
#include "timer.h"
 
class followRequestProvider: public requestProvider {
  protected:
    static options::single<bool> mayDelete;
    static options::single<bool> nullDelimiter;
 
 
 
    virtual void followStream(std::istream& stream) = 0;
    template <typename ... Types> void enqueueOrAppend(const std::string& srcPath,
            copyRequest::fileInWork::slotTypes type,
            copyRequest::clock_type::time_point timestamp,
            Types ... args) {
        timerInst(enqueue);
        auto now = copyRequest::clock_type::now();
        if (timestamp > now) { // cannot be true, probably ntp uncertainty
            if (timestamp - now > std::chrono::milliseconds(10)) {
                errMsg::emit(errMsg::level::debug, errMsg::location(),
                             srcPath, "createRequest",
                             "fixing future timestamp ",
                             std::chrono::duration_cast<std::chrono::duration<double>>(timestamp - now).count(), "s");
            }
            timestamp = now;
        }
 
        copyRequest::fileInWork::inserter inserter(srcPath);
        if (inserter.getFileInWork().getLatestRequestType() == type) { // just update the existing requests start time
            std::string suffix;
            copyRequest::base::getSuffix(srcPath, suffix);
            auto& fileInWork = inserter.getFileInWork();
            copyRequest::base::adviseDelay(timestamp - fileInWork.getEarliestprocessTime(), suffix);
            fileInWork.updateExecTime(timestamp);
        } else { // create a new request
            std::unique_ptr<copyRequest::base> request(new copyRequest::base(InputHandler, srcPath, args..., timestamp));
            if (inserter.getFileInWork().getLatestRequestType() == copyRequest::fileInWork::slotTypes::none) { // no resquest for this file yet
                inserter.markForQueueing(request);
                auto advisedDelay = request->getAdvisedDelay();
                if (request->getSlotType() != copyRequest::fileInWork::slotTypes::unlink
                        && advisedDelay != copyRequest::clock_type::duration::zero()) {
                    inserter.getFileInWork().setWaitTime(advisedDelay);
                    delayedRequests.enqueue(request, advisedDelay);
                } else {
                    requests.enqueue(request);
                }
            } else { // add request to per-file list
                inserter.enRegister(request);
            }
        }
    }
    virtual void handleMove(const std::string& srcPath,
                            const std::string& dstPath,
                            const std::string& fromPath,
                            const std::string& origPath,
                            const singleMap& mapEntry,
                            copyRequest::clock_type::time_point timestamp);
    virtual void handleOther(const std::string& srcPath,
                             const std::string& dstPath,
                             const singleMap& mapEntry,
                             copyRequest::clock_type::time_point timestamp,
                             bool requestForRemoval);
    followRequestProvider(decltype(requests) aRequests,
                          decltype(delayedRequests) aDelayedRequests,
                          decltype(parents) aParents,
                          decltype(InputHandler) aInputHandler,
                          decltype(OutputHandler) aOutputHandler);
  public:
    ~followRequestProvider() override = default;
    bool isFollowMode() const override {
        return true;
    };
    void processSources(const std::vector<std::string>& sources) override;
};
 
 
#endif