ExpirableQueue.h

Go to the documentation of this file.

// $Id: ExpirableQueue.h,v 1.8 2011/03/07 15:31:31 mommsen Exp $


#ifndef EventFilter_StorageManager_ExpirableQueue_h
#define EventFilter_StorageManager_ExpirableQueue_h

#include "EventFilter/StorageManager/interface/ConcurrentQueue.h"
#include "EventFilter/StorageManager/interface/EnquingPolicyTag.h"
#include "EventFilter/StorageManager/interface/QueueID.h"
#include "EventFilter/StorageManager/interface/Utils.h"

namespace stor
{
  template <class T, class Policy>
  class ExpirableQueue
  {
  public:
    typedef Policy PolicyType; // publish template parameter
    typedef typename ConcurrentQueue<T, Policy>::SizeType SizeType;
    typedef typename Policy::ValueType ValueType;

    explicit ExpirableQueue
    (
      SizeType maxsize=std::numeric_limits<SizeType>::max(),
      utils::Duration_t stalenessInterval = boost::posix_time::seconds(120),
      utils::TimePoint_t now = utils::getCurrentTime()
    );
    bool deqNowait(ValueType&);

    SizeType enqNowait
    (
      T const&,
      const utils::TimePoint_t& now = utils::getCurrentTime()
    );

    void setStalenessInterval(const utils::Duration_t&);

    utils::Duration_t stalenessInterval() const;    

    SizeType clear();

    bool empty() const;

    bool full() const;

    bool stale(const utils::TimePoint_t&) const;

    SizeType size() const;

    bool clearIfStale(const utils::TimePoint_t&, SizeType& clearedEvents);

  private:
    typedef ConcurrentQueue<T, Policy> queue_t;

    queue_t events_;
    utils::Duration_t stalenessInterval_;
    utils::TimePoint_t stalenessTime_;

    /*
      The following are not implemented, to prevent copying and
      assignment.
     */
    ExpirableQueue(ExpirableQueue&);
    ExpirableQueue& operator=(ExpirableQueue&);
  };

  
  template <class T, class Policy>
  ExpirableQueue<T, Policy>::ExpirableQueue
  (
    SizeType maxsize,
    utils::Duration_t stalenessInterval,
    utils::TimePoint_t now
  ) :
    events_(maxsize),
    stalenessInterval_(stalenessInterval),
    stalenessTime_(now+stalenessInterval_)
  {
  }

  template <class T, class Policy>
  bool
  ExpirableQueue<T, Policy>::deqNowait(ValueType& event)
  {
    stalenessTime_ = utils::getCurrentTime() + stalenessInterval_;
    return events_.deqNowait(event);
  }

  template <class T, class Policy>
  typename ExpirableQueue<T, Policy>::SizeType
  ExpirableQueue<T, Policy>::enqNowait(T const& event, const utils::TimePoint_t& now)
  {
    if ( stale(now) )
    {
      events_.addExternallyDroppedEvents(1);
      return 1;
    }
    return events_.enqNowait(event);
  }  

  template <class T, class Policy>
  inline
  void
  ExpirableQueue<T, Policy>::setStalenessInterval(const utils::Duration_t& t)
  {
    stalenessInterval_ = t;
  }

  template <class T, class Policy>
  inline
  utils::Duration_t
  ExpirableQueue<T, Policy>::stalenessInterval() const
  {
    return stalenessInterval_;
  }

  template <class T, class Policy>
  inline
  typename ExpirableQueue<T, Policy>::SizeType
  ExpirableQueue<T, Policy>::clear()
  {
    return events_.clear();
  }  

  template <class T, class Policy>
  inline
  bool
  ExpirableQueue<T, Policy>::empty() const
  {
    return events_.empty();
  }

  template <class T, class Policy>
  inline
  typename ExpirableQueue<T, Policy>::SizeType
  ExpirableQueue<T, Policy>::size() const
  {
    return events_.size();
  }

  template <class T, class Policy>
  inline
  bool
  ExpirableQueue<T, Policy>::full() const
  {
    return events_.full();
  }

  template <class T, class Policy>
  inline
  bool
  ExpirableQueue<T, Policy>::stale(const utils::TimePoint_t& now) const
  {
    return (stalenessTime_ < now);
  }

  template <class T, class Policy>
  inline
  bool
  ExpirableQueue<T, Policy>::clearIfStale
  (
    const utils::TimePoint_t& now,
    SizeType& clearedEvents
  )
  {
    if (stalenessTime_ < now)
    {
      clearedEvents = clear();
      return true;
    }
    else
    {
      clearedEvents = 0;
      return false;
    }
  }

} // namespace stor
  

#endif // EventFilter_StorageManager_ExpirableQueue_h