- 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 <algorithm>
    00030 #include <vector>
    00031 #include <string>
    00032 #include <sys/types.h>
    00033 #include <cstring>
    00034 #include <arpa/inet.h>
    00035 #include "StaticString.h"
    00036
    00037 /**
    00038  * This file provides a bunch of classes for reading and writing messages in the
    00039  * MessageChannel format. Unlike MessageChannel, whose operations take control over
    00040  * the I/O handle and may block, these classes are like parsers and data generators.
    00041  * Reader classes require the user to feed data to them. Writer classes generate a
    00042  * bunch of bytes that the user can send out. These classes will never block, making
    00043  * them ideal for use in evented servers.
    00044  *
    00045  * <h2>Reading messages</h2>
    00046  * To read a single message, one must instantiate a reader object and feed network
    00047  * data to it with the feed() method. This method returns the number of bytes
    00048  * actually processed by the reader (i.e. the number of bytes that it has recognized
    00049  * as part of the message).
    00050  *
    00051  * When the reader has either successfully parsed the data or encountered an error,
    00052  * it will indicate so via the done() method. With hasError() one can check whether
    00053  * an error was encountered or whether the reader succeeded, and with errorCode()
    00054  * one can obtain the exact error reason. Not all readers support hasError() and
    00055  * errorCode() because some readers can never encounter errors and some readers
    00056  * only have a single reason to fail.
    00057  *
    00058  * When successful, the parsed message can be obtained with value(). This method
    00059  * may only be called when done() is true and there is no error, otherwise the
    00060  * return value is undefined.
    00061  *
    00062  * At this point, the reader object cannot process any more data and feed() will
    00063  * always return 0. To reuse the object for processing another message, one must
    00064  * reset its state by calling reset().
    00065  *
    00066  * The following example demonstrates how to read a continuous stream of 32-bit
    00067  * integers:
    00068  * @code
    00069    Uint32Reader intReader;
    00070    while (true) {
    00071        // Read a bunch of network data...
    00072        char buf[1024];
    00073        ssize_t size = recv(fd, buf, sizeof(buf));
    00074        size_t consumed = 0;
    00075
    00076        // ...and process it all. We only feed data to the reader that
    00077        // hasn't already been fed.
    00078        while (consumed < size) {
    00079            consumed += intReader.feed(buf + consumed, size - consumed);
    00080            if (intReader.done()) {
    00081                printf("Integer: %d\n", (int) intReader.value());
    00082                // The state must be reset before the reader can be reused.
    00083                intReader.reset();
    00084            }
    00085        }
    00086    }
    00087  * @endcode
    00088  *
    00089  * Some readers return non-primitive values in their value() methods, such as
    00090  * ArrayReader and ScalarReader which return <tt>const vector<StaticString> &</tt>
    00091  * and <tt>StaticString</tt>, respectively. These values are only valid until either
    00092  * of the following things occur:
    00093  *
    00094  * - The buffer containing last the fed data has been destroyed or modified.
    00095  * - The reader itself has been destroyed.
    00096  *
    00097  * This is because the readers try to apply copy-zero optimizations whenever
    00098  * possible. For example, in case of ScalarReader, it'll check whether the data
    00099  * that has been fed in the first feed() call already contains a full scalar message.
    00100  * If so then it'll just return a StaticString that points to the scalar message
    00101  * in the fed data; it will not copy the fed data. In this case it is important
    00102  * that the buffer containing the fed data is not modified or destroyed while the
    00103  * StaticString is in use.
    00104  * If the first feed() call does not supply a full scalar message then it will
    00105  * buffer all fed data until the buffer contains a full scalar message, and the
    00106  * result will point to this buffer. Because the buffer is owned by the reader,
    00107  * the result will be invalidated as soon as the reader is destroyed.
    00108  */
    00109
    00110 namespace Passenger {
    00111
    00112 using namespace std;
    00113
    00114 /**
    00115  * Class for reading a 16-bit big-endian integer.
    00116  */
    00117 class Uint16Reader {
    00118 private:
    00119         uint16_t val;
    00120         uint8_t  consumed;
    00121
    00122 public:
    00123         Uint16Reader() {
    00124                 consumed = 0;
    00125         }
    00126
    00127         void reset() {
    00128                 consumed = 0;
    00129         }
    00130
    00131         size_t feed(const char *data, size_t size) {
    00132                 size_t locallyConsumed;
    00133
    00134                 locallyConsumed = std::min(size, sizeof(uint16_t) - consumed);
    00135                 memcpy(&val + consumed, data, locallyConsumed);
    00136                 consumed += locallyConsumed;
    00137                 if (done()) {
    00138                         val = ntohs(val);
    00139                 }
    00140                 return locallyConsumed;
    00141         }
    00142
    00143         bool done() const {
    00144                 return consumed == sizeof(uint16_t);
    00145         }
    00146
    00147         uint16_t value() const {
    00148                 return val;
    00149         }
    00150
    00151         void generate(void *buf, uint16_t val) const {
    00152                 val = htons(val);
    00153                 memcpy(buf, &val, sizeof(val));
    00154         }
    00155 };
    00156
    00157 /**
    00158  * Class for reading a 32-bit big-endian integer.
    00159  */
    00160 class Uint32Reader {
    00161 private:
    00162         uint32_t val;
    00163         uint8_t  consumed;
    00164
    00165 public:
    00166         Uint32Reader() {
    00167                 consumed = 0;
    00168         }
    00169
    00170         void reset() {
    00171                 consumed = 0;
    00172         }
    00173
    00174         size_t feed(const char *data, size_t size) {
    00175                 size_t locallyConsumed;
    00176
    00177                 locallyConsumed = std::min(size, sizeof(uint32_t) - consumed);
    00178                 memcpy(&val + consumed, data, locallyConsumed);
    00179                 consumed += locallyConsumed;
    00180                 if (done()) {
    00181                         val = ntohl(val);
    00182                 }
    00183                 return locallyConsumed;
    00184         }
    00185
    00186         bool done() const {
    00187                 return consumed == sizeof(uint32_t);
    00188         }
    00189
    00190         uint32_t value() const {
    00191                 return val;
    00192         }
    00193 };
    00194
    00195 /**
    00196  * Class for reading a an array message.
    00197  */
    00198 class ArrayReader {
    00199 public:
    00200         enum Error {
    00201                 TOO_LARGE
    00202         };
    00203
    00204 private:
    00205         enum State {
    00206                 READING_HEADER,
    00207                 READING_BODY,
    00208                 DONE,
    00209                 ERROR
    00210         };
    00211
    00212         uint16_t toReserve;
    00213         uint16_t maxSize;
    00214         Uint16Reader headerReader;
    00215         uint8_t state;
    00216         uint8_t error;
    00217         string buffer;
    00218         vector<StaticString> result;
    00219
    00220         void parseBody(const char *data, size_t size) {
    00221                 const char *start = data;
    00222                 const char *terminator;
    00223                 size_t rest = size;
    00224
    00225                 while ((terminator = (const char *) memchr(start, '\0', rest)) != NULL) {
    00226                         size_t len = terminator - start;
    00227                         result.push_back(StaticString(start, len));
    00228                         start = terminator + 1;
    00229                         rest = size - (start - data);
    00230                 }
    00231         }
    00232
    00233 public:
    00234         ArrayReader() {
    00235                 state = READING_HEADER;
    00236                 toReserve = 0;
    00237                 maxSize = 0;
    00238         }
    00239
    00240         void reserve(uint16_t size) {
    00241                 toReserve = size;
    00242                 result.reserve(size);
    00243         }
    00244
    00245         void setMaxSize(uint16_t size) {
    00246                 maxSize = size;
    00247         }
    00248
    00249         void reset() {
    00250                 state = READING_HEADER;
    00251                 headerReader.reset();
    00252                 buffer.clear();
    00253                 result.clear();
    00254                 if (toReserve > 0) {
    00255                         result.reserve(toReserve);
    00256                 }
    00257         }
    00258
    00259         size_t feed(const char *data, size_t size) {
    00260                 size_t consumed = 0;
    00261
    00262                 while (consumed < size && !done()) {
    00263                         const char *current = data + consumed;
    00264                         size_t rest = size - consumed;
    00265
    00266                         switch (state) {
    00267                         case READING_HEADER:
    00268                                 consumed += headerReader.feed(current, rest);
    00269                                 if (headerReader.done()) {
    00270                                         if (maxSize > 0 && headerReader.value() > maxSize) {
    00271                                                 state = ERROR;
    00272                                                 error = TOO_LARGE;
    00273                                         } else {
    00274                                                 state = READING_BODY;
    00275                                         }
    00276                                 }
    00277                                 break;
    00278                         case READING_BODY:
    00279                                 if (buffer.empty() && rest >= headerReader.value()) {
    00280                                         parseBody(current, headerReader.value());
    00281                                         state = DONE;
    00282                                         consumed += headerReader.value();
    00283                                 } else {
    00284                                         size_t toConsume = std::min(rest,
    00285                                                 headerReader.value() - buffer.size());
    00286                                         if (buffer.capacity() < headerReader.value()) {
    00287                                                 buffer.reserve(headerReader.value());
    00288                                         }
    00289                                         buffer.append(current, toConsume);
    00290                                         consumed += toConsume;
    00291                                         if (buffer.size() == headerReader.value()) {
    00292                                                 parseBody(buffer.data(), buffer.size());
    00293                                                 state = DONE;
    00294                                         }
    00295                                 }
    00296                                 break;
    00297                         default:
    00298                                 // Never reached.
    00299                                 abort();
    00300                         }
    00301                 }
    00302                 return consumed;
    00303         }
    00304
    00305         bool done() const {
    00306                 return state == DONE || state == ERROR;
    00307         }
    00308
    00309         bool hasError() const {
    00310                 return state == ERROR;
    00311         }
    00312
    00313         Error errorCode() const {
    00314                 return (Error) error;
    00315         }
    00316
    00317         const vector<StaticString> &value() const {
    00318                 return result;
    00319         }
    00320 };
    00321
    00322 /**
    00323  * Class for reading a scalar message.
    00324  */
    00325 class ScalarReader {
    00326 public:
    00327         enum Error {
    00328                 TOO_LARGE
    00329         };
    00330
    00331 private:
    00332         enum State {
    00333                 READING_HEADER,
    00334                 READING_BODY,
    00335                 DONE,
    00336                 ERROR
    00337         };
    00338
    00339         uint8_t state;
    00340         uint8_t error;
    00341         uint32_t maxSize;
    00342         Uint32Reader headerReader;
    00343         string buffer;
    00344         StaticString result;
    00345
    00346 public:
    00347         ScalarReader(uint32_t maxSize = 0) {
    00348                 state = READING_HEADER;
    00349                 this->maxSize = maxSize;
    00350         }
    00351
    00352         void reset() {
    00353                 state = READING_HEADER;
    00354                 headerReader.reset();
    00355                 buffer.clear();
    00356         }
    00357
    00358         size_t feed(const char *data, size_t size) {
    00359                 size_t consumed = 0;
    00360
    00361                 while (consumed < size && !done()) {
    00362                         const char *current = data + consumed;
    00363                         size_t rest = size - consumed;
    00364
    00365                         switch (state) {
    00366                         case READING_HEADER:
    00367                                 consumed += headerReader.feed(current, rest);
    00368                                 if (headerReader.done()) {
    00369                                         if (maxSize > 0 && headerReader.value() > maxSize) {
    00370                                                 state = ERROR;
    00371                                                 error = TOO_LARGE;
    00372                                         } else {
    00373                                                 state = READING_BODY;
    00374                                         }
    00375                                 }
    00376                                 break;
    00377                         case READING_BODY:
    00378                                 if (buffer.empty() && rest >= headerReader.value()) {
    00379                                         result = StaticString(current, headerReader.value());
    00380                                         state = DONE;
    00381                                         consumed += headerReader.value();
    00382                                 } else {
    00383                                         size_t toConsume = std::min(rest,
    00384                                                 headerReader.value() - buffer.size());
    00385                                         if (buffer.capacity() < headerReader.value()) {
    00386                                                 buffer.reserve(headerReader.value());
    00387                                         }
    00388                                         buffer.append(current, toConsume);
    00389                                         consumed += toConsume;
    00390                                         if (buffer.size() == headerReader.value()) {
    00391                                                 result = StaticString(buffer);
    00392                                                 state = DONE;
    00393                                         }
    00394                                 }
    00395                                 break;
    00396                         default:
    00397                                 // Never reached.
    00398                                 abort();
    00399                         };
    00400                 }
    00401                 return consumed;
    00402         }
    00403
    00404         bool done() const {
    00405                 return state == DONE || state == ERROR;
    00406         }
    00407
    00408         bool hasError() const {
    00409                 return state == ERROR;
    00410         }
    00411
    00412         Error errorCode() const {
    00413                 return (Error) error;
    00414         }
    00415
    00416         StaticString value() const {
    00417                 return result;
    00418         }
    00419 };
    00420
    00421 } // namespace Passenger
    00422
    00423 #endif /* _PASSENGER_MESSAGE_READERS_WRITERS_H_ */