Line | Count | Source |
1 | | /* SPDX-License-Identifier: MPL-2.0 */ |
2 | | |
3 | | #include "precompiled.hpp" |
4 | | #include "macros.hpp" |
5 | | #include "pair.hpp" |
6 | | #include "err.hpp" |
7 | | #include "pipe.hpp" |
8 | | #include "msg.hpp" |
9 | | |
10 | | zmq::pair_t::pair_t (class ctx_t *parent_, uint32_t tid_, int sid_) : |
11 | 0 | socket_base_t (parent_, tid_, sid_), _pipe (NULL) |
12 | 0 | { |
13 | 0 | options.type = ZMQ_PAIR; |
14 | 0 | } |
15 | | |
16 | | zmq::pair_t::~pair_t () |
17 | 0 | { |
18 | 0 | zmq_assert (!_pipe); |
19 | 0 | } |
20 | | |
21 | | void zmq::pair_t::xattach_pipe (pipe_t *pipe_, |
22 | | bool subscribe_to_all_, |
23 | | bool locally_initiated_) |
24 | 0 | { |
25 | 0 | LIBZMQ_UNUSED (subscribe_to_all_); |
26 | 0 | LIBZMQ_UNUSED (locally_initiated_); |
27 | |
|
28 | 0 | zmq_assert (pipe_ != NULL); |
29 | | |
30 | | // ZMQ_PAIR socket can only be connected to a single peer. |
31 | | // The socket rejects any further connection requests. |
32 | 0 | if (_pipe == NULL) |
33 | 0 | _pipe = pipe_; |
34 | 0 | else |
35 | 0 | pipe_->terminate (false); |
36 | 0 | } |
37 | | |
38 | | void zmq::pair_t::xpipe_terminated (pipe_t *pipe_) |
39 | 0 | { |
40 | 0 | if (pipe_ == _pipe) { |
41 | 0 | _pipe = NULL; |
42 | 0 | } |
43 | 0 | } |
44 | | |
45 | | void zmq::pair_t::xread_activated (pipe_t *) |
46 | 0 | { |
47 | | // There's just one pipe. No lists of active and inactive pipes. |
48 | | // There's nothing to do here. |
49 | 0 | } |
50 | | |
51 | | void zmq::pair_t::xwrite_activated (pipe_t *) |
52 | 0 | { |
53 | | // There's just one pipe. No lists of active and inactive pipes. |
54 | | // There's nothing to do here. |
55 | 0 | } |
56 | | |
57 | | int zmq::pair_t::xsend (msg_t *msg_) |
58 | 0 | { |
59 | 0 | if (!_pipe || !_pipe->write (msg_)) { |
60 | 0 | errno = EAGAIN; |
61 | 0 | return -1; |
62 | 0 | } |
63 | | |
64 | 0 | if (!(msg_->flags () & msg_t::more)) |
65 | 0 | _pipe->flush (); |
66 | | |
67 | | // Detach the original message from the data buffer. |
68 | 0 | const int rc = msg_->init (); |
69 | 0 | errno_assert (rc == 0); |
70 | |
|
71 | 0 | return 0; |
72 | 0 | } |
73 | | |
74 | | int zmq::pair_t::xrecv (msg_t *msg_) |
75 | 0 | { |
76 | | // Deallocate old content of the message. |
77 | 0 | int rc = msg_->close (); |
78 | 0 | errno_assert (rc == 0); |
79 | |
|
80 | 0 | if (!_pipe || !_pipe->read (msg_)) { |
81 | | // Initialise the output parameter to be a 0-byte message. |
82 | 0 | rc = msg_->init (); |
83 | 0 | errno_assert (rc == 0); |
84 | |
|
85 | 0 | errno = EAGAIN; |
86 | 0 | return -1; |
87 | 0 | } |
88 | 0 | return 0; |
89 | 0 | } |
90 | | |
91 | | bool zmq::pair_t::xhas_in () |
92 | 0 | { |
93 | 0 | if (!_pipe) |
94 | 0 | return false; |
95 | | |
96 | 0 | return _pipe->check_read (); |
97 | 0 | } |
98 | | |
99 | | bool zmq::pair_t::xhas_out () |
100 | 0 | { |
101 | 0 | if (!_pipe) |
102 | 0 | return false; |
103 | | |
104 | 0 | return _pipe->check_write (); |
105 | 0 | } |