- Main Page
- Modules
- Namespaces
- Classes
- Files

- File List

# MessageReadersWriters.h

    00001 /*
    00002  *  Phusion Passenger - http://www.modrails.com/
    00003  *  Copyright (c) 2010 Phusion
    00004  *
    00005  *  "Phusion Passenger" is a trademark of Hongli Lai & Ninh Bui.
    00006  *
    00007  *  Permission is hereby granted, free of charge, to any person obtaining a copy
    00008  *  of this software and associated documentation files (the "Software"), to deal
    00009  *  in the Software without restriction, including without limitation the rights
    00010  *  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
    00011  *  copies of the Software, and to permit persons to whom the Software is
    00012  *  furnished to do so, subject to the following conditions:
    00013  *
    00014  *  The above copyright notice and this permission notice shall be included in
    00015  *  all copies or substantial portions of the Software.
    00016  *
    00017  *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
    00018  *  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
    00019  *  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
    00020  *  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
    00021  *  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
    00022  *  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
    00023  *  THE SOFTWARE.
    00024  */
    00025 #ifndef _PASSENGER_MESSAGE_READERS_WRITERS_H_
    00026 #define _PASSENGER_MESSAGE_READERS_WRITERS_H_
    00027
    00028 #include <boost/cstdint.hpp>
    00029 #include <oxt/macros.hpp>
    00030 #include <algorithm>
    00031 #include <vector>
    00032 #include <string>
    00033 #include <sys/types.h>
    00034 #include <cstring>
    00035 #include <arpa/inet.h>
    00036 #include "StaticString.h"
    00037 #include "Exceptions.h"
    00038 #include "Utils/MemZeroGuard.h"
    00039
    00040 /**
    00041  * This file provides a bunch of classes for reading and writing messages in the
    00042  * MessageChannel format. Unlike MessageChannel, whose operations take control over
    00043  * the I/O handle and may block, these classes act like parsers and data generators.
    00044  * To read messages one must feed data to them. To write messages one must instruct
    00045  * the classes to generate a bunch of data. These classes will never block, making
    00046  * them ideal for use in evented servers.
    00047  *
    00048  * <h2>Reading messages</h2>
    00049  * To read a single message, one must instantiate a message object and feed network
    00050  * data to it with the feed() method. This method returns the number of bytes
    00051  * actually processed by the message object (i.e. the number of bytes that it has
    00052  * recognized as part of the message).
    00053  *
    00054  * When the message object has either successfully parsed the data or encountered
    00055  * an error, it will indicate so via the done() method. With hasError() one can
    00056  * check whether an error was encountered or whether the reader succeeded, and
    00057  * with errorCode() one can obtain the exact error reason. Not all message objects
    00058  * support hasError() and errorCode() because some of them can never encounter
    00059  * errors and some of them can only fail for a single reason.
    00060  *
    00061  * When successful, the parsed message can be obtained with value(). This method
    00062  * may only be called when done() is true and there is no error, otherwise the
    00063  * return value is undefined.
    00064  *
    00065  * At this point, the message object cannot process any more data and feed() will
    00066  * always return 0. To reuse the object for processing another message, one must
    00067  * reset its state by calling reset().
    00068  *
    00069  * The following example demonstrates how to read a continuous stream of 32-bit
    00070  * integers:
    00071  * @code
    00072    Uint32Message intMessage;
    00073    while (true) {
    00074        // Read a bunch of network data...
    00075        char buf[1024];
    00076        ssize_t size = recv(fd, buf, sizeof(buf));
    00077        size_t consumed = 0;
    00078
    00079        // ...and process it all. We only feed data to the message object
    00080        // that hasn't already been fed.
    00081        while (consumed < size) {
    00082            consumed += intReader.feed(buf + consumed, size - consumed);
    00083            if (intMessage.done()) {
    00084                printf("Integer: %d\n", (int) intMessage.value());
    00085                // The state must be reset before the reader can be reused.
    00086                intMessage.reset();
    00087            }
    00088        }
    00089    }
    00090  * @endcode
    00091  *
    00092  * Some message objects return non-primitive values in their value() methods, such as
    00093  * ArrayMessage and ScalarMessage which return <tt>const vector<StaticString> &</tt>
    00094  * and <tt>StaticString</tt>, respectively. These values are only valid until either
    00095  * of the following things occur:
    00096  *
    00097  * - The buffer containing last the fed data has been destroyed or modified.
    00098  * - The message object itself has been destroyed.
    00099  *
    00100  * This is because the message objects try to apply copy-zero optimizations whenever
    00101  * possible. For example, in case of ScalarMessage, it'll check whether the data
    00102  * that has been fed in the first feed() call already contains a full scalar message.
    00103  * If so then it'll just return a StaticString that points to the scalar message
    00104  * in the fed data; it will not copy the fed data. In this case it is important
    00105  * that the buffer containing the fed data is not modified or destroyed while the
    00106  * StaticString is in use.
    00107  * If the first feed() call does not supply a full scalar message then it will
    00108  * buffer all fed data until the buffer contains a full scalar message, and the
    00109  * result will point to this buffer. Because the buffer is owned by the message
    00110  * object, the result will be invalidated as soon as the message object is destroyed.
    00111  */
    00112
    00113 namespace Passenger {
    00114
    00115 using namespace std;
    00116
    00117 /**
    00118  * For 16-bit big-endian integers.
    00119  */
    00120 class Uint16Message {
    00121 private:
    00122         uint16_t val;
    00123         uint8_t  consumed;
    00124
    00125 public:
    00126         Uint16Message() {
    00127                 consumed = 0;
    00128         }
    00129
    00130         void reset() {
    00131                 consumed = 0;
    00132         }
    00133
    00134         size_t feed(const char *data, size_t size) {
    00135                 size_t locallyConsumed;
    00136
    00137                 locallyConsumed = std::min(size, sizeof(uint16_t) - consumed);
    00138                 memcpy((char *) &val + consumed, data, locallyConsumed);
    00139                 consumed += locallyConsumed;
    00140                 if (locallyConsumed > 0 && done()) {
    00141                         val = ntohs(val);
    00142                 }
    00143                 return locallyConsumed;
    00144         }
    00145
    00146         bool done() const {
    00147                 return consumed == sizeof(uint16_t);
    00148         }
    00149
    00150         uint16_t value() const {
    00151                 return val;
    00152         }
    00153
    00154         static void generate(void *buf, uint16_t val) {
    00155                 val = htons(val);
    00156                 memcpy(buf, &val, sizeof(val));
    00157         }
    00158 };
    00159
    00160 /**
    00161  * For 32-bit big-endian integers.
    00162  */
    00163 class Uint32Message {
    00164 private:
    00165         uint32_t val;
    00166         uint8_t  consumed;
    00167
    00168 public:
    00169         Uint32Message() {
    00170                 consumed = 0;
    00171         }
    00172
    00173         void reset() {
    00174                 consumed = 0;
    00175         }
    00176
    00177         size_t feed(const char *data, size_t size) {
    00178                 size_t locallyConsumed;
    00179
    00180                 locallyConsumed = std::min(size, sizeof(uint32_t) - consumed);
    00181                 memcpy((char *) &val + consumed, data, locallyConsumed);
    00182                 consumed += locallyConsumed;
    00183                 if (locallyConsumed > 0 && done()) {
    00184                         val = ntohl(val);
    00185                 }
    00186                 return locallyConsumed;
    00187         }
    00188
    00189         bool done() const {
    00190                 return consumed == sizeof(uint32_t);
    00191         }
    00192
    00193         uint32_t value() const {
    00194                 return val;
    00195         }
    00196
    00197         static void generate(void *buf, uint32_t val) {
    00198                 val = htonl(val);
    00199                 memcpy(buf, &val, sizeof(val));
    00200         }
    00201 };
    00202
    00203 /**
    00204  * For array messages.
    00205  */
    00206 class ArrayMessage {
    00207 public:
    00208         enum Error {
    00209                 TOO_LARGE
    00210         };
    00211
    00212 private:
    00213         enum State {
    00214                 READING_HEADER,
    00215                 READING_BODY,
    00216                 DONE,
    00217                 ERROR
    00218         };
    00219
    00220         uint16_t toReserve;
    00221         uint16_t maxSize;
    00222         Uint16Message headerReader;
    00223         uint8_t state;
    00224         uint8_t error;
    00225         string buffer;
    00226         vector<StaticString> result;
    00227
    00228         void parseBody(const char *data, size_t size) {
    00229                 const char *start = data;
    00230                 const char *terminator;
    00231                 size_t rest = size;
    00232
    00233                 while ((terminator = (const char *) memchr(start, '\0', rest)) != NULL) {
    00234                         size_t len = terminator - start;
    00235                         result.push_back(StaticString(start, len));
    00236                         start = terminator + 1;
    00237                         rest = size - (start - data);
    00238                 }
    00239         }
    00240
    00241 public:
    00242         ArrayMessage() {
    00243                 state = READING_HEADER;
    00244                 toReserve = 0;
    00245                 maxSize = 0;
    00246         }
    00247
    00248         void reserve(uint16_t size) {
    00249                 toReserve = size;
    00250                 result.reserve(size);
    00251         }
    00252
    00253         void setMaxSize(uint16_t size) {
    00254                 maxSize = size;
    00255         }
    00256
    00257         void reset() {
    00258                 state = READING_HEADER;
    00259                 headerReader.reset();
    00260                 buffer.clear();
    00261                 result.clear();
    00262                 if (toReserve > 0) {
    00263                         result.reserve(toReserve);
    00264                 }
    00265         }
    00266
    00267         size_t feed(const char *data, size_t size) {
    00268                 size_t consumed = 0;
    00269
    00270                 while (consumed < size && !done()) {
    00271                         const char *current = data + consumed;
    00272                         size_t rest = size - consumed;
    00273
    00274                         switch (state) {
    00275                         case READING_HEADER:
    00276                                 consumed += headerReader.feed(current, rest);
    00277                                 if (headerReader.done()) {
    00278                                         if (maxSize > 0 && headerReader.value() > maxSize) {
    00279                                                 state = ERROR;
    00280                                                 error = TOO_LARGE;
    00281                                         } else if (headerReader.value() == 0) {
    00282                                                 state = DONE;
    00283                                         } else {
    00284                                                 state = READING_BODY;
    00285                                         }
    00286                                 }
    00287                                 break;
    00288                         case READING_BODY:
    00289                                 if (buffer.empty() && rest >= headerReader.value()) {
    00290                                         parseBody(current, headerReader.value());
    00291                                         state = DONE;
    00292                                         consumed += headerReader.value();
    00293                                 } else {
    00294                                         size_t toConsume = std::min(rest,
    00295                                                 headerReader.value() - buffer.size());
    00296                                         if (buffer.capacity() < headerReader.value()) {
    00297                                                 buffer.reserve(headerReader.value());
    00298                                         }
    00299                                         buffer.append(current, toConsume);
    00300                                         consumed += toConsume;
    00301                                         if (buffer.size() == headerReader.value()) {
    00302                                                 parseBody(buffer.data(), buffer.size());
    00303                                                 state = DONE;
    00304                                         }
    00305                                 }
    00306                                 break;
    00307                         default:
    00308                                 // Never reached.
    00309                                 abort();
    00310                         }
    00311                 }
    00312                 return consumed;
    00313         }
    00314
    00315         bool done() const {
    00316                 return state == DONE || state == ERROR;
    00317         }
    00318
    00319         bool hasError() const {
    00320                 return state == ERROR;
    00321         }
    00322
    00323         Error errorCode() const {
    00324                 return (Error) error;
    00325         }
    00326
    00327         const vector<StaticString> &value() const {
    00328                 return result;
    00329         }
    00330
    00331         /**
    00332          * Given a bunch of array items, generates an array message. The message is
    00333          * generated in the form of an array of StaticStrings which must all be written
    00334          * out (e.g. with writev()) in the given order. These StaticStrings point
    00335          * to the buffers pointed to by <em>args</em> as well as <em>headerBuf</em>,
    00336          * so <em>args</em> and <em>headerBuf</em> must stay valid until the message
    00337          * has been written out or copied.
    00338          *
    00339          * @param args An array of array items to be included in the array message.
    00340          * @param argsCount The number of items in <em>args</em>.
    00341          * @param headerBuf A pointer to a buffer in which the array message header
    00342          *                  is to be stored.
    00343          * @param out A pointer to a StaticString array in which the generated array
    00344          *            message data will be stored. Exactly <tt>outputSize(argsCount)</tt>
    00345          *            items will be stored in this array.
    00346          * @param outCount The number of items in <em>out</em>.
    00347          */
    00348         static void generate(StaticString args[], unsigned int argsCount,
    00349                 char headerBuf[sizeof(uint16_t)], StaticString *out, unsigned int outCount)
    00350         {
    00351                 if (OXT_UNLIKELY(outCount < outputSize(argsCount))) {
    00352                         throw ArgumentException("outCount too small.");
    00353                 }
    00354
    00355                 unsigned int size = 0;
    00356                 unsigned int i;
    00357
    00358                 for (i = 0; i < argsCount; i++) {
    00359                         size += args[i].size() + 1;
    00360                 }
    00361                 if (OXT_UNLIKELY(size > 0xFFFF)) {
    00362                         throw ArgumentException("Data size exceeds maximum size for array messages.");
    00363                 }
    00364
    00365                 Uint16Message::generate(headerBuf, size);
    00366                 out[0] = StaticString(headerBuf, sizeof(uint16_t));
    00367                 for (i = 0; i < argsCount; i++) {
    00368                         out[1 + 2 * i] = args[i];
    00369                         out[1 + 2 * i + 1] = StaticString("\0", 1);
    00370                 }
    00371         }
    00372
    00373         static unsigned int outputSize(unsigned int argsCount) {
    00374                 return argsCount * 2 + 1;
    00375         }
    00376 };
    00377
    00378 /**
    00379  * Class for reading a scalar message.
    00380  */
    00381 class ScalarMessage {
    00382 public:
    00383         enum Error {
    00384                 TOO_LARGE
    00385         };
    00386
    00387 private:
    00388         enum State {
    00389                 READING_HEADER,
    00390                 READING_BODY,
    00391                 DONE,
    00392                 ERROR
    00393         };
    00394
    00395         uint8_t state;
    00396         uint8_t error;
    00397         uint32_t maxSize;
    00398         Uint32Message headerReader;
    00399         string buffer;
    00400         StaticString result;
    00401
    00402 public:
    00403         ScalarMessage(uint32_t maxSize = 0) {
    00404                 state = READING_HEADER;
    00405                 this->maxSize = maxSize;
    00406         }
    00407
    00408         void setMaxSize(uint32_t maxSize) {
    00409                 this->maxSize = maxSize;
    00410         }
    00411
    00412         /**
    00413          * Resets the internal state so that this object can be used for processing
    00414          * another scalar message.
    00415          *
    00416          * If <em>zeroBuffer</em> is true, then the contents of the internal buffer
    00417          * will be securely filled with zeroes. This is useful if e.g. the buffer
    00418          * might contain sensitive password data.
    00419          */
    00420         void reset(bool zeroBuffer = false) {
    00421                 state = READING_HEADER;
    00422                 if (zeroBuffer) {
    00423                         MemZeroGuard guard(buffer);
    00424                 }
    00425                 headerReader.reset();
    00426                 buffer.clear();
    00427         }
    00428
    00429         size_t feed(const char *data, size_t size) {
    00430                 size_t consumed = 0;
    00431
    00432                 while (consumed < size && !done()) {
    00433                         const char *current = data + consumed;
    00434                         size_t rest = size - consumed;
    00435
    00436                         switch (state) {
    00437                         case READING_HEADER:
    00438                                 consumed += headerReader.feed(current, rest);
    00439                                 if (headerReader.done()) {
    00440                                         if (maxSize > 0 && headerReader.value() > maxSize) {
    00441                                                 state = ERROR;
    00442                                                 error = TOO_LARGE;
    00443                                         } else if (headerReader.value() == 0) {
    00444                                                 state = DONE;
    00445                                         } else {
    00446                                                 state = READING_BODY;
    00447                                         }
    00448                                 }
    00449                                 break;
    00450                         case READING_BODY:
    00451                                 if (buffer.empty() && rest >= headerReader.value()) {
    00452                                         result = StaticString(current, headerReader.value());
    00453                                         state = DONE;
    00454                                         consumed += headerReader.value();
    00455                                 } else {
    00456                                         size_t toConsume = std::min(rest,
    00457                                                 headerReader.value() - buffer.size());
    00458                                         if (buffer.capacity() < headerReader.value()) {
    00459                                                 buffer.reserve(headerReader.value());
    00460                                         }
    00461                                         buffer.append(current, toConsume);
    00462                                         consumed += toConsume;
    00463                                         if (buffer.size() == headerReader.value()) {
    00464                                                 result = StaticString(buffer);
    00465                                                 state = DONE;
    00466                                         }
    00467                                 }
    00468                                 break;
    00469                         default:
    00470                                 // Never reached.
    00471                                 abort();
    00472                         };
    00473                 }
    00474                 return consumed;
    00475         }
    00476
    00477         bool done() const {
    00478                 return state == DONE || state == ERROR;
    00479         }
    00480
    00481         bool hasError() const {
    00482                 return state == ERROR;
    00483         }
    00484
    00485         Error errorCode() const {
    00486                 return (Error) error;
    00487         }
    00488
    00489         const StaticString &value() const {
    00490                 return result;
    00491         }
    00492
    00493         static void generate(const StaticString &data, char headerBuf[sizeof(uint32_t)],
    00494                 StaticString output[2])
    00495         {
    00496                 if (OXT_UNLIKELY(data.size() > 0xFFFFFFFF)) {
    00497                         throw ArgumentException("Data size exceeds maximum size for scalar messages.");
    00498                 }
    00499
    00500                 Uint32Message::generate(headerBuf, data.size());
    00501                 output[0] = StaticString(headerBuf, sizeof(uint32_t));
    00502                 output[1] = data;
    00503         }
    00504 };
    00505
    00506 } // namespace Passenger
    00507
    00508 #endif /* _PASSENGER_MESSAGE_READERS_WRITERS_H_ */