1 /*
2 Copyright (c) 2016 Contributors as noted in the AUTHORS file
3
4 This file is part of libzmq, the ZeroMQ core engine in C++.
5
6 libzmq is free software; you can redistribute it and/or modify it under
7 the terms of the GNU Lesser General Public License (LGPL) as published
8 by the Free Software Foundation; either version 3 of the License, or
9 (at your option) any later version.
10
11 As a special exception, the Contributors give you permission to link
12 this library with independent modules to produce an executable,
13 regardless of the license terms of these independent modules, and to
14 copy and distribute the resulting executable under terms of your choice,
15 provided that you also meet, for each linked independent module, the
16 terms and conditions of the license of that module. An independent
17 module is a module which is not derived from or based on this library.
18 If you modify this library, you must extend this exception to your
19 version of the library.
20
21 libzmq is distributed in the hope that it will be useful, but WITHOUT
22 ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
23 FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
24 License for more details.
25
26 You should have received a copy of the GNU Lesser General Public License
27 along with this program. If not, see <http://www.gnu.org/licenses/>.
28 */
29
30 #include "precompiled.hpp"
31 #include "macros.hpp"
32 #include "dgram.hpp"
33 #include "pipe.hpp"
34 #include "wire.hpp"
35 #include "random.hpp"
36 #include "likely.hpp"
37 #include "err.hpp"
38
dgram_t(class ctx_t * parent_,uint32_t tid_,int sid_)39 zmq::dgram_t::dgram_t (class ctx_t *parent_, uint32_t tid_, int sid_) :
40 socket_base_t (parent_, tid_, sid_),
41 _pipe (NULL),
42 _last_in (NULL),
43 _more_out (false)
44 {
45 options.type = ZMQ_DGRAM;
46 options.raw_socket = true;
47 }
48
~dgram_t()49 zmq::dgram_t::~dgram_t ()
50 {
51 zmq_assert (!_pipe);
52 }
53
xattach_pipe(pipe_t * pipe_,bool subscribe_to_all_,bool locally_initiated_)54 void zmq::dgram_t::xattach_pipe (pipe_t *pipe_,
55 bool subscribe_to_all_,
56 bool locally_initiated_)
57 {
58 LIBZMQ_UNUSED (subscribe_to_all_);
59 LIBZMQ_UNUSED (locally_initiated_);
60
61 zmq_assert (pipe_);
62
63 // ZMQ_DGRAM socket can only be connected to a single peer.
64 // The socket rejects any further connection requests.
65 if (_pipe == NULL)
66 _pipe = pipe_;
67 else
68 pipe_->terminate (false);
69 }
70
xpipe_terminated(pipe_t * pipe_)71 void zmq::dgram_t::xpipe_terminated (pipe_t *pipe_)
72 {
73 if (pipe_ == _pipe) {
74 if (_last_in == _pipe) {
75 _last_in = NULL;
76 }
77 _pipe = NULL;
78 }
79 }
80
xread_activated(pipe_t *)81 void zmq::dgram_t::xread_activated (pipe_t *)
82 {
83 // There's just one pipe. No lists of active and inactive pipes.
84 // There's nothing to do here.
85 }
86
xwrite_activated(pipe_t *)87 void zmq::dgram_t::xwrite_activated (pipe_t *)
88 {
89 // There's just one pipe. No lists of active and inactive pipes.
90 // There's nothing to do here.
91 }
92
xsend(msg_t * msg_)93 int zmq::dgram_t::xsend (msg_t *msg_)
94 {
95 // If there's no out pipe, just drop it.
96 if (!_pipe) {
97 const int rc = msg_->close ();
98 errno_assert (rc == 0);
99 return -1;
100 }
101
102 // If this is the first part of the message it's the ID of the
103 // peer to send the message to.
104 if (!_more_out) {
105 if (!(msg_->flags () & msg_t::more)) {
106 errno = EINVAL;
107 return -1;
108 }
109 } else {
110 // dgram messages are two part only, reject part if more is set
111 if (msg_->flags () & msg_t::more) {
112 errno = EINVAL;
113 return -1;
114 }
115 }
116
117 // Push the message into the pipe.
118 if (!_pipe->write (msg_)) {
119 errno = EAGAIN;
120 return -1;
121 }
122
123 if (!(msg_->flags () & msg_t::more))
124 _pipe->flush ();
125
126 // flip the more flag
127 _more_out = !_more_out;
128
129 // Detach the message from the data buffer.
130 const int rc = msg_->init ();
131 errno_assert (rc == 0);
132
133 return 0;
134 }
135
xrecv(msg_t * msg_)136 int zmq::dgram_t::xrecv (msg_t *msg_)
137 {
138 // Deallocate old content of the message.
139 int rc = msg_->close ();
140 errno_assert (rc == 0);
141
142 if (!_pipe || !_pipe->read (msg_)) {
143 // Initialise the output parameter to be a 0-byte message.
144 rc = msg_->init ();
145 errno_assert (rc == 0);
146
147 errno = EAGAIN;
148 return -1;
149 }
150 _last_in = _pipe;
151
152 return 0;
153 }
154
xhas_in()155 bool zmq::dgram_t::xhas_in ()
156 {
157 if (!_pipe)
158 return false;
159
160 return _pipe->check_read ();
161 }
162
xhas_out()163 bool zmq::dgram_t::xhas_out ()
164 {
165 if (!_pipe)
166 return false;
167
168 return _pipe->check_write ();
169 }
170