asyncqueue.h

Go to the documentation of this file.

/*---------------------------------------------------------------------\
|                          ____ _   __ __ ___                          |
|                         |__  / \ / / . \ . \                         |
|                           / / \ V /|  _/  _/                         |
|                          / /__ | | | | | |                           |
|                         /_____||_| |_| |_|                           |
|                                                                      |
----------------------------------------------------------------------*/
#ifndef ZYPP_NG_THREAD_ASYNCQUEUE_H_INCLUDED
#define ZYPP_NG_THREAD_ASYNCQUEUE_H_INCLUDED
 
#include <zypp-core/zyppng/base/AbstractEventSource>
 
#include <queue>
#include <set>
#include <mutex>
#include <memory>
#include <optional>
#include <condition_variable>
 
namespace zyppng {
 
  class AsyncQueueWatch;
 
  class AsyncQueueBase {
    public:
      virtual ~AsyncQueueBase();
 
      void addWatch ( AsyncQueueWatch &watch );
      void removeWatch (  AsyncQueueWatch &watch );
      void notifyWatches ( );
 
    private:
      std::set<AsyncQueueWatch *> _watches;
      std::recursive_mutex _watchLock;
  };

  template< class Message >
  class AsyncQueue : public AsyncQueueBase {
 
  public:
 
    using Ptr = std::shared_ptr<AsyncQueue>;
 
    static Ptr create () {
      return Ptr( new AsyncQueue() );
    }
 
    AsyncQueue(const AsyncQueue&) = delete;
    AsyncQueue& operator=(const AsyncQueue&) = delete;

    template< typename T = Message >
    void pushUnlocked ( T &&value ) {
      _messages.push( std::forward<T>(value) );
    }

    template< typename T = Message >
    void push ( T &&value ) {
      {
        std::lock_guard lk( _mut );
        pushUnlocked( std::forward<T>(value) );
      }
      notify();
    }

    Message pop () {
      std::unique_lock<std::mutex> lk( _mut );
      _cv.wait( lk, [this](){ return _messages.size() > 0; } );
      Message msg = std::move( _messages.front() );
      _messages.pop();
      return msg;
    }

    std::optional<Message> tryPop () {
      std::lock_guard lk( _mut );
      return tryPopUnlocked();
    }

    std::optional<Message> tryPopUnlocked () {
      if ( _messages.size() ) {
        Message msg = std::move( _messages.front() );
        _messages.pop();
        return msg;
      }
      return {};
    }

    void lock   () {
      _mut.lock();
    }

    void unlock () {
      _mut.unlock();
    }

    void notify () {
      _cv.notify_all();
      notifyWatches();
    }
 
  private:
    AsyncQueue() = default;
    std::queue<Message> _messages;
    std::mutex _mut;
    std::condition_variable _cv;
  };
 
  class AsyncQueueWatchPrivate;
  class LIBZYPP_NG_EXPORT AsyncQueueWatch : public AbstractEventSource
  {
    ZYPP_DECLARE_PRIVATE(AsyncQueueWatch)
  public:
 
    static std::shared_ptr<AsyncQueueWatch> create ( std::shared_ptr<AsyncQueueBase> queue );
    ~AsyncQueueWatch() override;
 
    void postNotifyEvent ();
 
    SignalProxy<void()> sigMessageAvailable();
 
    // AbstractEventSource interface
    void onFdReady(int fd, int events) override;
    void onSignal(int signal) override;
 
  protected:
    AsyncQueueWatch( std::shared_ptr<AsyncQueueBase> &&queue );
    AsyncQueueWatch( AsyncQueueWatchPrivate &dd );
  };
 
}
 
 
#endif