2
Copyright (c) 2009-2011 250bpm s.r.o.
3
Copyright (c) 2007-2009 iMatix Corporation
4
Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
6
This file is part of 0MQ.
8
0MQ is free software; you can redistribute it and/or modify it under
9
the terms of the GNU Lesser General Public License as published by
10
the Free Software Foundation; either version 3 of the License, or
11
(at your option) any later version.
13
0MQ is distributed in the hope that it will be useful,
14
but WITHOUT ANY WARRANTY; without even the implied warranty of
15
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16
GNU Lesser General Public License for more details.
18
You should have received a copy of the GNU Lesser General Public License
19
along with this program. If not, see <http://www.gnu.org/licenses/>.
25
#include "decoder.hpp"
26
#include "session_base.hpp"
31
zmq::decoder_t::decoder_t (size_t bufsize_, int64_t maxmsgsize_) :
32
decoder_base_t <decoder_t> (bufsize_),
34
maxmsgsize (maxmsgsize_)
36
int rc = in_progress.init ();
37
errno_assert (rc == 0);
39
// At the beginning, read one byte and go to one_byte_size_ready state.
40
next_step (tmpbuf, 1, &decoder_t::one_byte_size_ready);
43
zmq::decoder_t::~decoder_t ()
45
int rc = in_progress.close ();
46
errno_assert (rc == 0);
49
void zmq::decoder_t::set_session (session_base_t *session_)
54
bool zmq::decoder_t::one_byte_size_ready ()
56
// First byte of size is read. If it is 0xff read 8-byte size.
57
// Otherwise allocate the buffer for message data and read the
58
// message data into it.
60
next_step (tmpbuf, 8, &decoder_t::eight_byte_size_ready);
63
// There has to be at least one byte (the flags) in the message).
69
// in_progress is initialised at this point so in theory we should
70
// close it before calling zmq_msg_init_size, however, it's a 0-byte
71
// message and thus we can treat it as uninitialised...
73
if (maxmsgsize >= 0 && (int64_t) (*tmpbuf - 1) > maxmsgsize) {
78
rc = in_progress.init_size (*tmpbuf - 1);
79
if (rc != 0 && errno == ENOMEM) {
80
rc = in_progress.init ();
81
errno_assert (rc == 0);
85
errno_assert (rc == 0);
87
next_step (tmpbuf, 1, &decoder_t::flags_ready);
92
bool zmq::decoder_t::eight_byte_size_ready ()
94
// 8-byte size is read. Allocate the buffer for message body and
95
// read the message data into it.
96
size_t size = (size_t) get_uint64 (tmpbuf);
98
// There has to be at least one byte (the flags) in the message).
104
// in_progress is initialised at this point so in theory we should
105
// close it before calling zmq_msg_init_size, however, it's a 0-byte
106
// message and thus we can treat it as uninitialised...
108
if (maxmsgsize >= 0 && (int64_t) (size - 1) > maxmsgsize) {
113
rc = in_progress.init_size (size - 1);
114
if (rc != 0 && errno == ENOMEM) {
115
rc = in_progress.init ();
116
errno_assert (rc == 0);
120
errno_assert (rc == 0);
122
next_step (tmpbuf, 1, &decoder_t::flags_ready);
126
bool zmq::decoder_t::flags_ready ()
128
// Store the flags from the wire into the message structure.
129
in_progress.set_flags (tmpbuf [0]);
131
next_step (in_progress.data (), in_progress.size (),
132
&decoder_t::message_ready);
137
bool zmq::decoder_t::message_ready ()
139
// Message is completely read. Push it further and start reading
140
// new message. (in_progress is a 0-byte message after this point.)
141
if (unlikely (!session))
143
int rc = session->write (&in_progress);
144
if (unlikely (rc != 0)) {
150
next_step (tmpbuf, 1, &decoder_t::one_byte_size_ready);