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 in { |
29 namespace in { |
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 |
34 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 |
35 |
38 public: |
36 public: |
39 |
37 |
40 virtual ~Socket() { |
38 virtual ~Socket() { |
41 if (handle >= 0) mq_close(handle); |
|
42 if (unlinkOnClose) mq_unlink(queueName.c_str()); |
|
43 } |
|
44 |
|
45 static Socket* open(std::string queueName, bool unlinkOnClose = false) { |
|
46 mqd_t handle = mq_open(queueName.c_str(), O_RDONLY | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP); |
|
47 if (handle >= 0) return new Socket(queueName, handle, unlinkOnClose); |
|
48 else throw std::logic_error("Unable to open Socket: " + 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 } |
39 } |
55 |
40 |
56 std::string receive() { |
41 std::string receive() { |
57 char buffer[MSG_SIZE + 1]; |
42 char buffer[MSG_SIZE + 1]; |
58 memset(buffer, 0, MSG_SIZE + 1); |
43 memset(buffer, 0, MSG_SIZE + 1); |
59 ssize_t msgSize = mq_receive(handle, buffer, MSG_SIZE, nullptr); |
44 |
|
45 int s = ::socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); |
|
46 |
|
47 struct sockaddr_in a; |
|
48 memset((char *) &a, 0, sizeof (a)); |
|
49 a.sin_family = AF_INET; |
|
50 a.sin_addr.s_addr = inet_addr("127.0.0.1"); // TODO: use getaddrinfo() instead (because of error -1 = 255.255.255.255) |
|
51 a.sin_port = htons(1234); |
|
52 ::bind(s, (sockaddr*) & a, sizeof (a)); |
|
53 |
|
54 struct sockaddr_in remoteAddress; |
|
55 memset((char *) &remoteAddress, 0, sizeof (remoteAddress)); |
|
56 socklen_t remoteAddressSize = sizeof(remoteAddress); |
|
57 |
|
58 ssize_t msgSize = recvfrom(s, buffer, sizeof (buffer), 0, (sockaddr*) & remoteAddress, &remoteAddressSize); |
|
59 |
|
60 close(s); |
60 |
61 |
61 if (msgSize > sizeof (buffer))throw std::logic_error("Invalid Socket message size."); |
62 if (msgSize > sizeof (buffer))throw std::logic_error("Invalid Socket message size."); |
62 else if (msgSize >= 0) return std::string(buffer, msgSize); |
63 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 else throw std::logic_error("Unable to receive Socket message the socket; error: " + std::string(strerror(errno))); |
64 } |
65 } |
65 |
66 |
66 }; |
67 }; |
67 |
68 |
68 } |
69 } |