/* Copyright (c) 2007-2013 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_ENCODER_HPP_INCLUDED__ #define __ZMQ_ENCODER_HPP_INCLUDED__ #if defined(_MSC_VER) #ifndef NOMINMAX #define NOMINMAX #endif #endif #include #include #include #include #include "err.hpp" #include "msg.hpp" #include "i_encoder.hpp" namespace zmq { // Helper base class for encoders. It implements the state machine that // fills the outgoing buffer. Derived classes should implement individual // state machine actions. template class encoder_base_t : public i_encoder { public: inline encoder_base_t (size_t bufsize_) : bufsize (bufsize_), in_progress (NULL) { buf = (unsigned char*) malloc (bufsize_); alloc_assert (buf); } // The destructor doesn't have to be virtual. It is made virtual // just to keep ICC and code checking tools from complaining. inline virtual ~encoder_base_t () { free (buf); } // The function returns a batch of binary data. The data // are filled to a supplied buffer. If no buffer is supplied (data_ // points to NULL) decoder object will provide buffer of its own. inline size_t encode (unsigned char **data_, size_t size_) { unsigned char *buffer = !*data_ ? buf : *data_; size_t buffersize = !*data_ ? bufsize : size_; if (in_progress == NULL) return 0; size_t pos = 0; while (pos < buffersize) { // If there are no more data to return, run the state machine. // If there are still no data, return what we already have // in the buffer. if (!to_write) { if (new_msg_flag) { int rc = in_progress->close (); errno_assert (rc == 0); rc = in_progress->init (); errno_assert (rc == 0); in_progress = NULL; break; } (static_cast (this)->*next) (); } // If there are no data in the buffer yet and we are able to // fill whole buffer in a single go, let's use zero-copy. // There's no disadvantage to it as we cannot stuck multiple // messages into the buffer anyway. Note that subsequent // write(s) are non-blocking, thus each single write writes // at most SO_SNDBUF bytes at once not depending on how large // is the chunk returned from here. // As a consequence, large messages being sent won't block // other engines running in the same I/O thread for excessive // amounts of time. if (!pos && !*data_ && to_write >= buffersize) { *data_ = write_pos; pos = to_write; write_pos = NULL; to_write = 0; return pos; } // Copy data to the buffer. If the buffer is full, return. size_t to_copy = std::min (to_write, buffersize - pos); memcpy (buffer + pos, write_pos, to_copy); pos += to_copy; write_pos += to_copy; to_write -= to_copy; } *data_ = buffer; return pos; } void load_msg (msg_t *msg_) { zmq_assert (in_progress == NULL); in_progress = msg_; (static_cast (this)->*next) (); } inline bool has_data () { return to_write > 0; } protected: // Prototype of state machine action. typedef void (T::*step_t) (); // This function should be called from derived class to write the data // to the buffer and schedule next state machine action. inline void next_step (void *write_pos_, size_t to_write_, step_t next_, bool new_msg_flag_) { write_pos = (unsigned char*) write_pos_; to_write = to_write_; next = next_; new_msg_flag = new_msg_flag_; } private: // Where to get the data to write from. unsigned char *write_pos; // How much data to write before next step should be executed. size_t to_write; // Next step. If set to NULL, it means that associated data stream // is dead. step_t next; bool new_msg_flag; // The buffer for encoded data. size_t bufsize; unsigned char *buf; encoder_base_t (const encoder_base_t&); void operator = (const encoder_base_t&); protected: msg_t *in_progress; }; } #endif