diff options
Diffstat (limited to 'src/decoder.hpp')
| -rw-r--r-- | src/decoder.hpp | 85 |
1 files changed, 67 insertions, 18 deletions
diff --git a/src/decoder.hpp b/src/decoder.hpp index f05f651..f554e46 100644 --- a/src/decoder.hpp +++ b/src/decoder.hpp @@ -1,19 +1,20 @@ /* - Copyright (c) 2007-2010 iMatix Corporation + Copyright (c) 2007-2011 iMatix Corporation + Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file This file is part of 0MQ. 0MQ is free software; you can redistribute it and/or modify it under - the terms of the Lesser GNU General Public License as published by + the terms of the GNU Lesser General Public License as published by the Free Software Foundation; either version 3 of the License, or (at your option) any later version. 0MQ is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - Lesser GNU General Public License for more details. + GNU Lesser General Public License for more details. - You should have received a copy of the Lesser GNU General Public License + You should have received a copy of the GNU Lesser General Public License along with this program. If not, see <http://www.gnu.org/licenses/>. */ @@ -27,37 +28,39 @@ #include "err.hpp" +#include "../include/zmq.h" + namespace zmq { // Helper base class for decoders that know the amount of data to read // in advance at any moment. Knowing the amount in advance is a property - // of the protocol used. Both AMQP and backend protocol are based on - // size-prefixed paradigm, therefore they are using decoder_t to parse - // the messages. On the other hand, XML-based transports (like XMPP or - // SOAP) don't allow for knowing the size of data to read in advance and - // should use different decoding algorithms. + // of the protocol used. 0MQ framing protocol is based size-prefixed + // paradigm, whixh qualifies it to be parsed by this class. + // On the other hand, XML-based transports (like XMPP or SOAP) don't allow + // for knowing the size of data to read in advance and should use different + // decoding algorithms. // - // Decoder implements the state machine that parses the incoming buffer. + // This class implements the state machine that parses the incoming buffer. // Derived class should implement individual state machine actions. - template <typename T> class decoder_t + template <typename T> class decoder_base_t { public: - inline decoder_t (size_t bufsize_) : + inline decoder_base_t (size_t bufsize_) : read_pos (NULL), to_read (0), next (NULL), bufsize (bufsize_) { buf = (unsigned char*) malloc (bufsize_); - zmq_assert (buf); + alloc_assert (buf); } // The destructor doesn't have to be virtual. It is mad virtual // just to keep ICC and code checking tools from complaining. - inline virtual ~decoder_t () + inline virtual ~decoder_base_t () { free (buf); } @@ -89,6 +92,10 @@ namespace zmq // bytes actually processed. inline size_t process_buffer (unsigned char *data_, size_t size_) { + // Check if we had an error in previous attempt. + if (unlikely (!(static_cast <T*> (this)->next))) + return (size_t) -1; + // In case of zero-copy simply adjust the pointers, no copying // is required. Also, run the state machine in case all the data // were processed. @@ -96,9 +103,13 @@ namespace zmq read_pos += size_; to_read -= size_; - while (!to_read) - if (!(static_cast <T*> (this)->*next) ()) + while (!to_read) { + if (!(static_cast <T*> (this)->*next) ()) { + if (unlikely (!(static_cast <T*> (this)->next))) + return (size_t) -1; return size_; + } + } return size_; } @@ -107,9 +118,13 @@ namespace zmq // Try to get more space in the message to fill in. // If none is available, return. - while (!to_read) - if (!(static_cast <T*> (this)->*next) ()) + while (!to_read) { + if (!(static_cast <T*> (this)->*next) ()) { + if (unlikely (!(static_cast <T*> (this)->next))) + return (size_t) -1; return pos; + } + } // If there are no more data in the buffer, return. if (pos == size_) @@ -140,6 +155,13 @@ namespace zmq next = next_; } + // This function should be called from the derived class to + // abort decoder state machine. + inline void decoding_error () + { + next = NULL; + } + private: unsigned char *read_pos; @@ -149,6 +171,32 @@ namespace zmq size_t bufsize; unsigned char *buf; + decoder_base_t (const decoder_base_t&); + const decoder_base_t &operator = (const decoder_base_t&); + }; + + // Decoder for 0MQ framing protocol. Converts data batches into messages. + + class decoder_t : public decoder_base_t <decoder_t> + { + public: + + decoder_t (size_t bufsize_); + ~decoder_t (); + + void set_inout (struct i_inout *destination_); + + private: + + bool one_byte_size_ready (); + bool eight_byte_size_ready (); + bool flags_ready (); + bool message_ready (); + + struct i_inout *destination; + unsigned char tmpbuf [8]; + ::zmq_msg_t in_progress; + decoder_t (const decoder_t&); void operator = (const decoder_t&); }; @@ -156,3 +204,4 @@ namespace zmq } #endif + |