/*
|
* =====================================================================================
|
*
|
* Filename: sendq.h
|
*
|
* Description:
|
*
|
* Version: 1.0
|
* Created: 2021年04月14日 09时22分59秒
|
* Revision: none
|
* Compiler: gcc
|
*
|
* Author: Li Chao (), lichao@aiotlink.com
|
* Organization:
|
*
|
* =====================================================================================
|
*/
|
#ifndef SENDQ_IWKMSK7M
|
#define SENDQ_IWKMSK7M
|
|
#include "defs.h"
|
#include "msg.h"
|
#include "timed_queue.h"
|
#include <deque>
|
#include <functional>
|
#include <list>
|
#include <mutex>
|
#include <string>
|
#include <unordered_map>
|
|
namespace bhome_shm
|
{
|
class ShmMsgQueue;
|
} // namespace bhome_shm
|
|
class SendQ
|
{
|
public:
|
typedef std::string Remote;
|
typedef bhome_msg::MsgI MsgI;
|
typedef std::function<void(const MsgI &msg)> OnMsgEvent;
|
struct MsgInfo {
|
MsgI msg_;
|
OnMsgEvent on_expire_;
|
// OnMsgEvent on_send_;
|
};
|
typedef TimedData<MsgInfo> TimedMsg;
|
typedef TimedMsg::TimePoint TimePoint;
|
typedef TimedMsg::Duration Duration;
|
|
void Append(const MQId &id, const MsgI &msg, OnMsgEvent onExpire = OnMsgEvent())
|
{
|
Append(std::string((const char *) &id, sizeof(id)), msg, onExpire);
|
}
|
void Append(const Remote &addr, const MsgI &msg, OnMsgEvent onExpire = OnMsgEvent())
|
{
|
using namespace std::chrono_literals;
|
Append(addr, msg, Now() + 60s, onExpire);
|
}
|
bool TrySend(bhome_shm::ShmMsgQueue &mq);
|
// bool empty() const { return store_.empty(); }
|
|
private:
|
static TimePoint Now() { return TimedMsg::Clock::now(); }
|
void Append(const Remote &addr, const MsgI &msg, const TimePoint &expire, OnMsgEvent onExpire)
|
{
|
//TODO simple queue, organize later ?
|
|
msg.AddRef();
|
TimedMsg tmp(expire, MsgInfo{msg, onExpire});
|
std::unique_lock<std::mutex> lock(mutex_in_);
|
auto &al = in_[addr];
|
if (!al.empty()) {
|
al.front().emplace_back(std::move(tmp));
|
} else {
|
al.insert(al.begin(), Array())->emplace_back(std::move(tmp));
|
}
|
}
|
typedef std::deque<TimedMsg> Array;
|
typedef std::list<Array> ArrayList;
|
typedef std::unordered_map<Remote, ArrayList> Store;
|
|
int DoSend1Remote(bhome_shm::ShmMsgQueue &mq, const Remote &remote, Array &arr);
|
int DoSend1Remote(bhome_shm::ShmMsgQueue &mq, const Remote &remote, ArrayList &arr);
|
|
class MsgIter
|
{
|
Array::iterator iter_;
|
|
public:
|
MsgIter(Array::iterator iter) :
|
iter_(iter) {}
|
MsgIter &operator++() { return ++iter_, *this; }
|
bool operator==(const MsgIter &a) { return iter_ == a.iter_; }
|
MsgI &operator*() { return iter_->data().msg_; }
|
};
|
|
std::mutex mutex_in_;
|
std::mutex mutex_out_;
|
Store in_;
|
Store out_;
|
};
|
|
#endif // end of include guard: SENDQ_IWKMSK7M
|
