src/ZeroMQ.h
branchv_0
changeset 0 452d06d24ac2
equal deleted inserted replaced
-1:000000000000 0:452d06d24ac2
       
     1 /**
       
     2  * Relational pipes
       
     3  * Copyright © 2022 František Kučera (Frantovo.cz, GlobalCode.info)
       
     4  *
       
     5  * This program is free software: you can redistribute it and/or modify
       
     6  * it under the terms of the GNU General Public License as published by
       
     7  * the Free Software Foundation, version 3 of the License.
       
     8  *
       
     9  * This program is distributed in the hope that it will be useful,
       
    10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
       
    11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
       
    12  * GNU General Public License for more details.
       
    13  *
       
    14  * You should have received a copy of the GNU General Public License
       
    15  * along with this program. If not, see <http://www.gnu.org/licenses/>.
       
    16  */
       
    17 #pragma once
       
    18 
       
    19 #include <mqueue.h>
       
    20 #include <string>
       
    21 #include <stdexcept>
       
    22 #include <cstring>
       
    23 
       
    24 namespace relpipe {
       
    25 namespace out {
       
    26 namespace zeromq {
       
    27 
       
    28 class ZeroMQ {
       
    29 private:
       
    30 	const static size_t MSG_SIZE = 8192; // TODO: configurable/dynamic
       
    31 	std::string queueName;
       
    32 	mqd_t handle = -2;
       
    33 	bool unlinkOnClose = false;
       
    34 
       
    35 	ZeroMQ(std::string queueName, mqd_t handle, bool unlinkOnClose) : queueName(queueName), handle(handle), unlinkOnClose(unlinkOnClose) {
       
    36 	}
       
    37 
       
    38 public:
       
    39 
       
    40 	virtual ~ZeroMQ() {
       
    41 		if (handle >= 0) mq_close(handle);
       
    42 		if (unlinkOnClose) mq_unlink(queueName.c_str());
       
    43 	}
       
    44 
       
    45 	static ZeroMQ* open(std::string queueName, bool unlinkOnClose = false) {
       
    46 		mqd_t handle = mq_open(queueName.c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
       
    47 		if (handle >= 0) return new ZeroMQ(queueName, handle, unlinkOnClose);
       
    48 		else throw std::logic_error("Unable to open ZeroMQ: " + queueName + " error: " + strerror(errno));
       
    49 	}
       
    50 
       
    51 	void send(std::string message) {
       
    52 		int result = mq_send(handle, message.c_str(), message.size(), 0);
       
    53 		if (result) throw std::logic_error("Unable to send message to" + queueName + " error: " + strerror(errno));
       
    54 	}
       
    55 
       
    56 	std::string receive() {
       
    57 		char buffer[MSG_SIZE + 1];
       
    58 		memset(buffer, 0, MSG_SIZE + 1);
       
    59 		ssize_t msgSize = mq_receive(handle, buffer, MSG_SIZE, nullptr);
       
    60 
       
    61 		if (msgSize > sizeof (buffer))throw std::logic_error("Invalid ZeroMQ message size.");
       
    62 		else if (msgSize >= 0) return std::string(buffer, msgSize);
       
    63 		else throw std::logic_error("Unable to receive ZeroMQ message from " + queueName + " error: " + strerror(errno));
       
    64 	}
       
    65 
       
    66 };
       
    67 
       
    68 }
       
    69 }
       
    70 }