decoder.hpp

Go to the documentation of this file.

/* SPDX-License-Identifier: MPL-2.0 */
 
#ifndef __ZMQ_DECODER_HPP_INCLUDED__
#define __ZMQ_DECODER_HPP_INCLUDED__
 
#include <algorithm>
#include <cstddef>
#include <cstring>
 
#include "decoder_allocators.hpp"
#include "err.hpp"
#include "i_decoder.hpp"
#include "stdint.hpp"
 
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, which 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.
//
//  Buffer management is done by an allocator policy.
template <typename T, typename A = c_single_allocator>
class decoder_base_t : public i_decoder
{
  public:
    explicit decoder_base_t (const size_t buf_size_) :
        _next (NULL), _read_pos (NULL), _to_read (0), _allocator (buf_size_)
    {
        _buf = _allocator.allocate ();
    }
 
    ~decoder_base_t () ZMQ_OVERRIDE { _allocator.deallocate (); }
 
    //  Returns a buffer to be filled with binary data.
    void get_buffer (unsigned char **data_, std::size_t *size_) ZMQ_FINAL
    {
        _buf = _allocator.allocate ();
 
        //  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 >= _allocator.size ()) {
            *data_ = _read_pos;
            *size_ = _to_read;
            return;
        }
 
        *data_ = _buf;
        *size_ = _allocator.size ();
    }
 
    //  Processes the data in the buffer previously allocated using
    //  get_buffer function. size_ argument specifies number of bytes
    //  actually filled into the buffer. Function returns 1 when the
    //  whole message was decoded or 0 when more data is required.
    //  On error, -1 is returned and errno set accordingly.
    //  Number of bytes processed is returned in bytes_used_.
    int decode (const unsigned char *data_,
                std::size_t size_,
                std::size_t &bytes_used_) ZMQ_FINAL
    {
        bytes_used_ = 0;
 
        //  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) {
            zmq_assert (size_ <= _to_read);
            _read_pos += size_;
            _to_read -= size_;
            bytes_used_ = size_;
 
            while (!_to_read) {
                const int rc =
                  (static_cast<T *> (this)->*_next) (data_ + bytes_used_);
                if (rc != 0)
                    return rc;
            }
            return 0;
        }
 
        while (bytes_used_ < size_) {
            //  Copy the data from buffer to the message.
            const size_t to_copy = std::min (_to_read, size_ - bytes_used_);
            // Only copy when destination address is different from the
            // current address in the buffer.
            if (_read_pos != data_ + bytes_used_) {
                memcpy (_read_pos, data_ + bytes_used_, to_copy);
            }
 
            _read_pos += to_copy;
            _to_read -= to_copy;
            bytes_used_ += to_copy;
            //  Try to get more space in the message to fill in.
            //  If none is available, return.
            while (_to_read == 0) {
                // pass current address in the buffer
                const int rc =
                  (static_cast<T *> (this)->*_next) (data_ + bytes_used_);
                if (rc != 0)
                    return rc;
            }
        }
 
        return 0;
    }
 
    void resize_buffer (std::size_t new_size_) ZMQ_FINAL
    {
        _allocator.resize (new_size_);
    }
 
  protected:
    //  Prototype of state machine action. Action should return false if
    //  it is unable to push the data to the system.
    typedef int (T::*step_t) (unsigned char const *);
 
    //  This function should be called from derived class to read data
    //  from the buffer and schedule next state machine action.
    void next_step (void *read_pos_, std::size_t to_read_, step_t next_)
    {
        _read_pos = static_cast<unsigned char *> (read_pos_);
        _to_read = to_read_;
        _next = next_;
    }
 
    A &get_allocator () { return _allocator; }
 
  private:
    //  Next step. If set to NULL, it means that associated data stream
    //  is dead. Note that there can be still data in the process in such
    //  case.
    step_t _next;
 
    //  Where to store the read data.
    unsigned char *_read_pos;
 
    //  How much data to read before taking next step.
    std::size_t _to_read;
 
    //  The duffer for data to decode.
    A _allocator;
    unsigned char *_buf;
 
    ZMQ_NON_COPYABLE_NOR_MOVABLE (decoder_base_t)
};
}
 
#endif