init
This commit is contained in:
qing committed
commit
70d467d052
97 files changed
+10386
No files matched your search
@@ -0,0 +1,93 @@
|
||||
/*
|
||||
* 目前是Message Queue 的发送端（不涉及创建消息队列）
|
||||
*
|
||||
* FIXME: 集成发送、接收和创建
|
||||
*/
|
||||
|
||||
#pragma once
|
||||
#include <iostream>
|
||||
#include <cstring>
|
||||
#include <memory>
|
||||
#include <boost/interprocess/ipc/message_queue.hpp>
|
||||
|
||||
namespace qing {
|
||||
namespace bip = boost::interprocess;
|
||||
using sf_t = std::function<void(const char*)>; /* 通信成功的回调函数 success */
|
||||
using ef_t = std::function<void(void)>; /* 通信失败的回调函数 error */
|
||||
class Mq {
|
||||
public:
|
||||
enum {CREATOR=0, USER=1};
|
||||
Mq(const std::string& name, int msglen, size_t msgcnt, int mode = USER) {
|
||||
this->name = std::string("/") + name;
|
||||
this->msglen = msglen;
|
||||
this->msgcnt = msgcnt;
|
||||
this->mode = mode;
|
||||
switch (mode) {
|
||||
case USER:
|
||||
mqp = std::make_unique<bip::message_queue>(bip::open_only, this->name.c_str());
|
||||
break;
|
||||
case CREATOR: /* 由创建者创建消息队列 */
|
||||
bip::message_queue::remove(this->name.c_str());
|
||||
mqp = std::make_unique<bip::message_queue>(
|
||||
bip::create_only, this->name.c_str(),
|
||||
msgcnt, msglen * sizeof(char)
|
||||
);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
/* 创建者负责清理 */
|
||||
~Mq() {
|
||||
if (mode == CREATOR)
|
||||
bip::message_queue::remove(name.c_str());
|
||||
}
|
||||
|
||||
/* */
|
||||
Mq(const Mq&) = delete;
|
||||
Mq& operator=(const Mq&) = delete;
|
||||
|
||||
|
||||
/* 发送消息 */
|
||||
void send(std::string &msg)
|
||||
{
|
||||
if (msg.size() + 1 > msglen) /* over the len */
|
||||
throw std::length_error("Message queue Send Length over");
|
||||
|
||||
mqp->send(msg.c_str(), msg.size() + 1, 0); /* include \0 */
|
||||
}
|
||||
|
||||
/* 尝试接收消息 */
|
||||
void recv(sf_t success, ef_t notyet) {
|
||||
|
||||
auto data = std::make_unique<char[]>(msglen); /* 接收缓冲区 */
|
||||
unsigned int priority; /* 优先级 */
|
||||
bip::message_queue::size_type recvd_size;
|
||||
|
||||
try { /* try_receive()返回bool类型 */
|
||||
auto r = mqp->try_receive(
|
||||
data.get(), msglen*sizeof(char),
|
||||
recvd_size, priority
|
||||
);
|
||||
|
||||
if (r) /* 成功接收到数据 */
|
||||
success(data.get());
|
||||
|
||||
else /* 没有收到数据 */
|
||||
notyet();
|
||||
}
|
||||
|
||||
catch (std::exception &exp) {
|
||||
std::cerr << "exception type: " << typeid(exp).name() << std::endl;
|
||||
std::cerr << "exception message: " << exp.what() << std::endl;
|
||||
}
|
||||
}
|
||||
|
||||
private:
|
||||
|
||||
/* */
|
||||
std::unique_ptr<bip::message_queue> mqp;
|
||||
std::string name;
|
||||
size_t msglen=0, msgcnt=0;
|
||||
int mode;
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,157 @@
|
||||
#pragma once
|
||||
#include <string>
|
||||
#include <iostream>
|
||||
#include <unistd.h>
|
||||
#include <fcntl.h> /* open() */
|
||||
#include <sys/stat.h> /* mkfifo() */
|
||||
#include <sys/select.h>
|
||||
|
||||
namespace qing {
|
||||
|
||||
/* 20260309(qing): 该类不会抛出异常 */
|
||||
class Pipe{
|
||||
public:
|
||||
static const int CREATOR = 1;
|
||||
static const int USER =2;
|
||||
static const ssize_t PIPE_MAX_LEN = 2048;
|
||||
Pipe(const std::string& path, int mode)
|
||||
: fd(-1), isCreator(mode == CREATOR) {
|
||||
std::string real_path = "/tmp/pst-";
|
||||
real_path += path;
|
||||
fifo_path = real_path;
|
||||
if (isCreator && mkfifo(fifo_path.c_str(), 0666) == -1 && errno != EEXIST) /* wr_wr_wr */
|
||||
perror("mkfifo"); /* 文件已存在是正常情况，其他错误才需要处理 */
|
||||
}
|
||||
|
||||
~Pipe() {
|
||||
/* 如果打开了文件，则关闭文件 */
|
||||
if (fd != -1) close(fd);
|
||||
/**********************/
|
||||
/* 如果管道由该对象创建，就要负责清理 */
|
||||
if (isCreator) unlink(fifo_path.c_str()); /* 创建者负责清理 */
|
||||
}
|
||||
Pipe(const Pipe&) = delete;
|
||||
Pipe& operator=(const Pipe&) = delete;
|
||||
|
||||
/* 这里才是打开文件的指令 - 以读模式打开，可以非阻塞 */
|
||||
bool openForRead(bool nonblock = false) {
|
||||
|
||||
/* 设置是否以非阻塞方式打开文件 */
|
||||
int flags = O_RDONLY;
|
||||
if (nonblock) {
|
||||
flags |= O_NONBLOCK;
|
||||
}
|
||||
|
||||
fd = open(fifo_path.c_str(), flags);
|
||||
|
||||
if (fd == -1) {
|
||||
perror("open for read");
|
||||
return false;
|
||||
}
|
||||
|
||||
else {
|
||||
FD_ZERO(&read_fds);
|
||||
FD_SET(fd, &read_fds);
|
||||
if (fd > max_fd) max_fd = fd;
|
||||
return true;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/* 以写模式打开文件 */
|
||||
bool openForWrite(/*bool nonblock = falsei*/) {
|
||||
|
||||
/* 写入管道是不需要阻塞/非阻塞的？ */
|
||||
int flags = O_WRONLY;
|
||||
//if (nonblock) {
|
||||
// flags |= O_NONBLOCK;
|
||||
//}
|
||||
fd = open(fifo_path.c_str(), flags);
|
||||
if (fd == -1) {
|
||||
perror("open for write");
|
||||
return false;
|
||||
}
|
||||
|
||||
else {
|
||||
FD_ZERO(&write_fds);
|
||||
FD_SET(fd, &write_fds);
|
||||
if (fd > max_fd) max_fd = fd;
|
||||
return true;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/* 写入管道，如果失败返回-1。成功返回写入的字符数。 */
|
||||
ssize_t writePipe(const std::string &data) {
|
||||
if (fd == -1) {
|
||||
std::cerr << "Pipe not opened for writing" << std::endl;
|
||||
return -1;
|
||||
}
|
||||
else {
|
||||
return write(fd, data.c_str(), data.size());
|
||||
}
|
||||
}
|
||||
|
||||
/* 将探测与读取解耦，使用者可以自行决定要不要阻塞 */
|
||||
bool waitForRead(int timeout_sec = 0, int timeout_usec = 0) {
|
||||
|
||||
/* 如果文件没打开，等待是失败的 */
|
||||
if (fd == -1) return false;
|
||||
|
||||
fd_set temp_fds = read_fds;
|
||||
struct timeval timeout = {timeout_sec, timeout_usec};
|
||||
|
||||
int ret = select(max_fd + 1, &temp_fds, nullptr, nullptr, &timeout);
|
||||
|
||||
if (ret == -1) {
|
||||
perror("select");
|
||||
return false;
|
||||
}
|
||||
|
||||
else {
|
||||
return ret > 0 && FD_ISSET(fd, &temp_fds);
|
||||
}
|
||||
}
|
||||
|
||||
/* 读取数据到字符串，返回实际读取的字节数 */
|
||||
ssize_t readPipe(std::string *output) {
|
||||
|
||||
if (fd == -1) {
|
||||
std::cerr << "Pipe not opened for reading" << std::endl;
|
||||
return -1;
|
||||
}
|
||||
|
||||
else
|
||||
|
||||
{
|
||||
|
||||
char buffer[PIPE_MAX_LEN];
|
||||
ssize_t n = read(fd, buffer, sizeof(buffer) - 1);
|
||||
if (n >0) {
|
||||
buffer[n] = '\0';
|
||||
*output = buffer;
|
||||
}
|
||||
|
||||
else if (n == 0) {
|
||||
/* 管道写入端关闭 */
|
||||
*output = "";
|
||||
}
|
||||
|
||||
else {
|
||||
perror("read");
|
||||
}
|
||||
|
||||
return n;
|
||||
|
||||
}
|
||||
}
|
||||
private:
|
||||
std::string fifo_path;
|
||||
int fd;
|
||||
bool isCreator;
|
||||
fd_set read_fds;
|
||||
fd_set write_fds;
|
||||
fd_set except_fds;
|
||||
int max_fd;
|
||||
};
|
||||
}
|
||||
Reference in new issue
Block a user