/* 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 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 GNU Lesser General Public License for more details. You should have received a copy of the GNU Lesser General Public License along with this program. If not, see . */ #ifndef __ZMQ_DECODER_HPP_INCLUDED__ #define __ZMQ_DECODER_HPP_INCLUDED__ #include #include #include #include #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. 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. // // This class implements the state machine that parses the incoming buffer. // Derived class should implement individual state machine actions. template class decoder_base_t { public: inline decoder_base_t (size_t bufsize_) : read_pos (NULL), to_read (0), next (NULL), bufsize (bufsize_) { buf = (unsigned char*) malloc (bufsize_); 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_base_t () { free (buf); } // Returns a buffer to be filled with binary data. inline void get_buffer (unsigned char **data_, size_t *size_) { // If we are expected to read large message, we'll opt for zero- // copy, i.e. we'll ask caller to fill the data directly to the // message. Note that subsequent read(s) are non-blocking, thus // each single read reads at most SO_RCVBUF bytes at once not // depending on how large is the chunk returned from here. // As a consequence, large messages being received won't block // other engines running in the same I/O thread for excessive // amounts of time. if (to_read >= bufsize) { *data_ = read_pos; *size_ = to_read; return; } *data_ = buf; *size_ = bufsize; } // Processes the data in the buffer previously allocated using // get_buffer function. size_ argument specifies nemuber of bytes // actually filled into the buffer. Function returns number of // 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 (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. if (data_ == read_pos) { read_pos += size_; to_read -= size_; while (!to_read) { if (!(static_cast (this)->*next) ()) { if (unlikely (!(static_cast (this)->next))) return (size_t) -1; return size_; } } return size_; } size_t pos = 0; while (true) { // Try to get more space in the message to fill in. // If none is available, return. while (!to_read) { if (!(static_cast (this)->*next) ()) { if (unlikely (!(static_cast (this)->next))) return (size_t) -1; return pos; } } // If there are no more data in the buffer, return. if (pos == size_) return pos; // Copy the data from buffer to the message. size_t to_copy = std::min (to_read, size_ - pos); memcpy (read_pos, data_ + pos, to_copy); read_pos += to_copy; pos += to_copy; to_read -= to_copy; } } protected: // Prototype of state machine action. Action should return false if // it is unable to push the data to the system. typedef bool (T::*step_t) (); // This function should be called from derived class to read data // from the buffer and schedule next state machine action. inline void next_step (void *read_pos_, size_t to_read_, step_t next_) { read_pos = (unsigned char*) read_pos_; to_read = to_read_; 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; size_t to_read; step_t next; 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 { 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&); }; } #endif