provideworker.h

Go to the documentation of this file.

/*---------------------------------------------------------------------\
|                          ____ _   __ __ ___                          |
|                         |__  / \ / / . \ . \                         |
|                           / / \ V /|  _/  _/                         |
|                          / /__ | | | | | |                           |
|                         /_____||_| |_| |_|                           |
|                                                                      |
\---------------------------------------------------------------------*/
 
#ifndef ZYPP_MEDIA_PROVIDE_WORKER_H_INCLUDED
#define ZYPP_MEDIA_PROVIDE_WORKER_H_INCLUDED
 
#include <zypp-core/zyppng/base/Base>
#include <zypp-core/zyppng/base/EventLoop>
#include <zypp-core/zyppng/base/Timer>
#include <zypp-core/zyppng/io/AsyncDataSource>
#include <zypp-core/zyppng/rpc/MessageStream>
#include <zypp-core/zyppng/pipelines/Expected>
#include <zypp-proto/media/provider.pb.h>
#include <zypp-media/ng/provide-configvars.h>
#include <zypp-media/ng/private/providemessage_p.h>
#include <zypp-media/ng/HeaderValueMap>
#include <zypp-media/MediaException>
#include <zypp-media/Mount>
 
#include <string_view>
#include <deque>
 
namespace zyppng::worker {
 
  using WorkerCaps     = zypp::proto::Capabilities;
  using Message        = zypp::proto::Envelope;
  using Configuration  = zypp::proto::Configuration;
 
  struct AuthInfo
  {
    std::string username;
    std::string password;
    int64_t last_auth_timestamp = 0;
    std::map<std::string, std::string> extraKeys = {};
  };
 
  class RequestCancelException : public zypp::media::MediaException
  {
  public:
    RequestCancelException();
  };
 
  ZYPP_FWD_DECL_TYPE_WITH_REFS (ProvideWorker);
  ZYPP_FWD_DECL_TYPE_WITH_REFS (ProvideWorkerItem);
 
  class ProvideWorkerItem : public zyppng::Base
  {
  public:
    enum State {
      Pending,
      Running,
      Finished
    };
 
    ProvideWorkerItem( ProvideMessage &&spec ) : _spec( std::move(spec) ) { }
 
    State _state = Pending;
    ProvideMessage _spec;
  };
 
  class ProvideWorker : public Base
  {
  public:
 
    enum ProvideNotificatioMode {
      ONLY_NEW_PROVIDES,  // provide is called only when new provide requests are added to the queue
      QUEUE_NOT_EMTPY     // provide is called continiously until the queue is empty
    };
 
    ProvideWorker( std::string_view workerName );
    virtual ~ProvideWorker();
 
    RpcMessageStream::Ptr messageStream() const;
 
    expected<void> run ( int recv = STDIN_FILENO, int send = STDOUT_FILENO );
 
    std::deque<ProvideWorkerItemRef> &requestQueue();
    virtual void immediateShutdown (){};
 
    enum MediaChangeRes {
      SUCCESS,
      ABORT,
      SKIP
    };
    MediaChangeRes requestMediaChange ( const uint32_t id, const std::string &label, const int32_t mediaNr, const std::vector<std::string> &devices, const std::optional<std::string> &desc = {} );
 
    expected<AuthInfo> requireAuthorization ( const uint32_t id, const zypp::Url &url, const std::string &lastTriedUsername = "", const int64_t lastTimestamp = -1, const std::map<std::string, std::string> &extraFields = {} );
 
    ProvideNotificatioMode provNotificationMode() const;
    void setProvNotificationMode(const ProvideNotificatioMode &provNotificationMode);
 
  protected:
    virtual void initLog();
    virtual expected<WorkerCaps> initialize ( const Configuration &conf ) = 0;
 
    virtual void provide ( ) = 0;
    virtual void cancel  ( const std::deque<ProvideWorkerItemRef>::iterator &request ) = 0;
 
    virtual ProvideWorkerItemRef makeItem (ProvideMessage &&spec );
 
    void provideStart   ( const uint32_t id, const zypp::Url &url, const zypp::Pathname &localFile, const zypp::Pathname &stagingFile = {}  );
 
    void provideSuccess (const uint32_t id, bool cacheHit, const zypp::Pathname &localFile, const HeaderValueMap extra = {} );
 
    void provideFailed  ( const uint32_t id, const uint code, const std::string &reason, const bool transient, const HeaderValueMap extra = {} );
 
    void provideFailed  ( const uint32_t id, const uint code, const bool transient, const zypp::Exception &e );
 
    void attachSuccess ( const uint32_t id );
 
    void detachSuccess ( const uint32_t id );
 
    void redirect       ( const uint32_t id, const zypp::Url &url, const zypp::Pathname &newPath );
 
    AsyncDataSource &controlIO ();
 
 
  private:
    expected<void> executeHandshake ();
    void maybeDelayedShutdown ();
    void messageLoop ( Timer & );
    void readFdClosed  ( uint, AsyncDataSource::ChannelCloseReason );
    void writeFdClosed ( AsyncDataSource::ChannelCloseReason );
    void messageReceived ();
    void onInvalidMessageReceived ( );
    void invalidMessageReceived ( std::exception_ptr p );
    void handleSingleMessage (const ProvideMessage &provide );
    void pushSingleMessage ( const RpcMessage &msg );
    expected<ProvideMessage> sendAndWaitForResponse ( const ProvideMessage &request, const std::vector<uint> &responseCodes );
    expected<ProvideMessage> parseReceivedMessage( const RpcMessage &m );
 
  private:
    ProvideNotificatioMode _provNotificationMode = QUEUE_NOT_EMTPY;
    bool _inControllerRequest = false; //< Used to signalize that we are currently in a blocking controller callback
    bool _isRunning = false;
    std::string_view _workerName;
    EventLoop::Ptr _loop = EventLoop::create();
    Timer::Ptr _msgAvail = Timer::create();
    Timer::Ptr _delayedShutdown = Timer::create();
    AsyncDataSource::Ptr _controlIO;
    RpcMessageStream::Ptr _stream;
    Configuration _workerConf;
 
    std::exception_ptr _fatalError; //< Error that caused the eventloop to stop
 
    std::deque<ProvideMessage> _pendingMessages;
    std::deque<ProvideWorkerItemRef> _pendingProvides;
  };
}
 
 
#endif