14 * You should have received a copy of the GNU General Public License |
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/>. |
15 * along with this program. If not, see <http://www.gnu.org/licenses/>. |
16 */ |
16 */ |
17 #pragma once |
17 #pragma once |
18 |
18 |
19 #include <mqueue.h> |
|
20 #include <string> |
19 #include <string> |
|
20 #include <cstring> |
|
21 #include <unistd.h> |
21 #include <stdexcept> |
22 #include <stdexcept> |
22 #include <cstring> |
23 #include <arpa/inet.h> |
|
24 #include <sys/types.h> |
|
25 #include <sys/socket.h> |
|
26 #include <netinet/in.h> |
23 |
27 |
24 namespace relpipe { |
28 namespace relpipe { |
25 namespace out { |
29 namespace out { |
26 namespace socket { |
30 namespace socket { |
27 |
31 |
28 class Socket { |
32 class Socket { |
29 private: |
33 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 Socket(std::string queueName, mqd_t handle, bool unlinkOnClose) : queueName(queueName), handle(handle), unlinkOnClose(unlinkOnClose) { |
|
36 } |
|
37 |
|
38 public: |
34 public: |
39 |
35 |
40 virtual ~Socket() { |
36 virtual ~Socket() { |
41 if (handle >= 0) mq_close(handle); |
|
42 if (unlinkOnClose) mq_unlink(queueName.c_str()); |
|
43 } |
37 } |
44 |
38 |
45 static Socket* open(std::string queueName, bool unlinkOnClose = false) { |
39 void send(const std::string& message) { |
46 mqd_t handle = mq_open(queueName.c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP); |
40 |
47 if (handle >= 0) return new Socket(queueName, handle, unlinkOnClose); |
41 struct sockaddr_in a; |
48 else throw std::logic_error("Unable to open Socket: " + queueName + " error: " + strerror(errno)); |
42 memset((char *) &a, 0, sizeof (a)); |
|
43 a.sin_family = AF_INET; |
|
44 a.sin_addr.s_addr = inet_addr("127.0.0.1"); // TODO: use getaddrinfo() instead (because of error -1 = 255.255.255.255) |
|
45 a.sin_port = htons(1234); |
|
46 |
|
47 int s = ::socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); |
|
48 sendto(s, message.c_str(), message.size(), 0, (sockaddr*) & a, sizeof (a)); |
|
49 |
|
50 close(s); |
|
51 |
|
52 // TODO: send message |
49 } |
53 } |
50 |
54 |
51 void send(std::string message) { |
55 // virtual const std::string receive(); |
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 Socket message size."); |
|
62 else if (msgSize >= 0) return std::string(buffer, msgSize); |
|
63 else throw std::logic_error("Unable to receive Socket message from " + queueName + " error: " + strerror(errno)); |
|
64 } |
|
65 |
56 |
66 }; |
57 }; |
67 |
58 |
68 } |
59 } |
69 } |
60 } |