1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230
|
#pragma once
#include <string>
#include "libipc/imp/export.h"
#include "libipc/def.h"
#include "libipc/buffer.h"
#include "libipc/shm.h"
namespace ipc {
using handle_t = void*;
using buff_t = buffer;
enum : unsigned {
sender,
receiver
};
template <typename Flag>
struct LIBIPC_EXPORT chan_impl {
static ipc::handle_t init_first();
static bool connect (ipc::handle_t * ph, char const * name, unsigned mode);
static bool connect (ipc::handle_t * ph, prefix, char const * name, unsigned mode);
static bool reconnect (ipc::handle_t * ph, unsigned mode);
static void disconnect(ipc::handle_t h);
static void destroy (ipc::handle_t h);
static char const * name(ipc::handle_t h);
// Release memory without waiting for the connection to disconnect.
static void release(ipc::handle_t h) noexcept;
// Force cleanup of all shared memory storage that handles depend on.
static void clear(ipc::handle_t h) noexcept;
static void clear_storage(char const * name) noexcept;
static void clear_storage(prefix, char const * name) noexcept;
static std::size_t recv_count (ipc::handle_t h);
static bool wait_for_recv(ipc::handle_t h, std::size_t r_count, std::uint64_t tm);
static bool send(ipc::handle_t h, void const * data, std::size_t size, std::uint64_t tm);
static buff_t recv(ipc::handle_t h, std::uint64_t tm);
static bool try_send(ipc::handle_t h, void const * data, std::size_t size, std::uint64_t tm);
static buff_t try_recv(ipc::handle_t h);
};
template <typename Flag>
class chan_wrapper {
private:
using detail_t = chan_impl<Flag>;
ipc::handle_t h_ = detail_t::init_first();
unsigned mode_ = ipc::sender;
bool connected_ = false;
public:
chan_wrapper() noexcept = default;
explicit chan_wrapper(char const * name, unsigned mode = ipc::sender)
: connected_{this->connect(name, mode)} {
}
chan_wrapper(prefix pref, char const * name, unsigned mode = ipc::sender)
: connected_{this->connect(pref, name, mode)} {
}
chan_wrapper(chan_wrapper&& rhs) noexcept
: chan_wrapper{} {
swap(rhs);
}
~chan_wrapper() {
detail_t::destroy(h_);
}
void swap(chan_wrapper& rhs) noexcept {
std::swap(h_ , rhs.h_);
std::swap(mode_ , rhs.mode_);
std::swap(connected_, rhs.connected_);
}
chan_wrapper& operator=(chan_wrapper rhs) noexcept {
swap(rhs);
return *this;
}
char const * name() const noexcept {
return detail_t::name(h_);
}
// Release memory without waiting for the connection to disconnect.
void release() noexcept {
detail_t::release(h_);
h_ = nullptr;
}
// Clear shared memory files under opened handle.
void clear() noexcept {
detail_t::clear(h_);
h_ = nullptr;
}
// Clear shared memory files under a specific name.
static void clear_storage(char const * name) noexcept {
detail_t::clear_storage(name);
}
// Clear shared memory files under a specific name with a prefix.
static void clear_storage(prefix pref, char const * name) noexcept {
detail_t::clear_storage(pref, name);
}
ipc::handle_t handle() const noexcept {
return h_;
}
bool valid() const noexcept {
return (handle() != nullptr);
}
unsigned mode() const noexcept {
return mode_;
}
chan_wrapper clone() const {
return chan_wrapper { name(), mode_ };
}
/**
* Building handle, then try connecting with name & mode flags.
*/
bool connect(char const * name, unsigned mode = ipc::sender | ipc::receiver) {
if (name == nullptr || name[0] == '\0') return false;
detail_t::disconnect(h_); // clear old connection
return connected_ = detail_t::connect(&h_, name, mode_ = mode);
}
bool connect(prefix pref, char const * name, unsigned mode = ipc::sender | ipc::receiver) {
if (name == nullptr || name[0] == '\0') return false;
detail_t::disconnect(h_); // clear old connection
return connected_ = detail_t::connect(&h_, pref, name, mode_ = mode);
}
/**
* Try connecting with new mode flags.
*/
bool reconnect(unsigned mode) {
if (!valid()) return false;
if (connected_ && (mode_ == mode)) return true;
return connected_ = detail_t::reconnect(&h_, mode_ = mode);
}
void disconnect() {
if (!valid()) return;
detail_t::disconnect(h_);
connected_ = false;
}
std::size_t recv_count() const {
return detail_t::recv_count(h_);
}
bool wait_for_recv(std::size_t r_count, std::uint64_t tm = invalid_value) const {
return detail_t::wait_for_recv(h_, r_count, tm);
}
static bool wait_for_recv(char const * name, std::size_t r_count, std::uint64_t tm = invalid_value) {
return chan_wrapper(name).wait_for_recv(r_count, tm);
}
/**
* If timeout, this function would call 'force_push' to send the data forcibly.
*/
bool send(void const * data, std::size_t size, std::uint64_t tm = default_timeout) {
return detail_t::send(h_, data, size, tm);
}
bool send(buff_t const & buff, std::uint64_t tm = default_timeout) {
return this->send(buff.data(), buff.size(), tm);
}
bool send(std::string const & str, std::uint64_t tm = default_timeout) {
return this->send(str.c_str(), str.size() + 1, tm);
}
/**
* If timeout, this function would just return false.
*/
bool try_send(void const * data, std::size_t size, std::uint64_t tm = default_timeout) {
return detail_t::try_send(h_, data, size, tm);
}
bool try_send(buff_t const & buff, std::uint64_t tm = default_timeout) {
return this->try_send(buff.data(), buff.size(), tm);
}
bool try_send(std::string const & str, std::uint64_t tm = default_timeout) {
return this->try_send(str.c_str(), str.size() + 1, tm);
}
buff_t recv(std::uint64_t tm = invalid_value) {
return detail_t::recv(h_, tm);
}
buff_t try_recv() {
return detail_t::try_recv(h_);
}
};
template <relat Rp, relat Rc, trans Ts>
using chan = chan_wrapper<ipc::wr<Rp, Rc, Ts>>;
/**
* \class route
*
* \note You could use one producer/server/sender for sending messages to a route,
* then all the consumers/clients/receivers which are receiving with this route,
* would receive your sent messages.
* A route could only be used in 1 to N (one producer/writer to multi consumers/readers).
*/
using route = chan<relat::single, relat::multi, trans::broadcast>;
/**
* \class channel
*
* \note You could use multi producers/writers for sending messages to a channel,
* then all the consumers/readers which are receiving with this channel,
* would receive your sent messages.
*/
using channel = chan<relat::multi, relat::multi, trans::broadcast>;
} // namespace ipc
|