platform/codecs/LogCodec.hpp

// ------------------------------------------------------------------------
// Pion is a development platform for building Reactors that process Events
// ------------------------------------------------------------------------
// Copyright (C) 2007-2008 Atomic Labs, Inc.  (http://www.atomiclabs.com)
//
// Pion is free software: you can redistribute it and/or modify it under the
// terms of the GNU Affero General Public License as published by the Free
// Software Foundation, either version 3 of the License, or (at your option)
// any later version.
//
// Pion 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 Affero General Public License for
// more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with Pion.  If not, see <http://www.gnu.org/licenses/>.
//

#ifndef __PION_LOGCODEC_HEADER__
#define __PION_LOGCODEC_HEADER__

#include <vector>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/scoped_array.hpp>
#include <boost/lexical_cast.hpp>
#include <pion/PionConfig.hpp>
#include <pion/PionException.hpp>
#include <pion/PionHashMap.hpp>
#include <pion/PionDateTime.hpp>
#include <pion/PionAlgorithms.hpp>
#include <pion/platform/Codec.hpp>
#include <pion/platform/Vocabulary.hpp>

#ifdef PION_WIN32
#define OSEOL "\r\n"
#else
#define OSEOL "\n"
#endif

namespace pion {        // begin namespace pion
namespace plugins {     // begin namespace plugins


class LogCodec :
    public pion::platform::Codec
{
public:

    class EmptyFieldException : public PionException {
    public:
        EmptyFieldException(const std::string& codec_id)
            : PionException("LogCodec configuration includes an empty field name: ", codec_id) {}
    };

    class EmptyTermException : public PionException {
    public:
        EmptyTermException(const std::string& codec_id)
            : PionException("LogCodec configuration is missing a term identifier: ", codec_id) {}
    };

    class UnknownTermException : public PionException {
    public:
        UnknownTermException(const std::string& term_id)
            : PionException("LogCodec configuration maps field to an unknown term: ", term_id) {}
    };

    class BadFormatException : public PionException {
    public:
        BadFormatException(const std::string& term_id)
            : PionException("LogCodec format contains an unknown term: ", term_id) {}
    };


    LogCodec(void)
        : pion::platform::Codec(), m_read_buf(new char[READ_BUFFER_SIZE+1]),
        m_read_end(m_read_buf.get() + READ_BUFFER_SIZE),
        m_flush_after_write(false), m_handle_elf_headers(false), m_wrote_elf_headers(false),
        m_time_offset(0),
        m_event_split(EVENT_SPLIT_SET), m_event_join(EVENT_JOIN_STRING), m_comment_chars(COMMENT_CHAR_SET),
        m_field_split(FIELD_SPLIT_SET), m_field_join(FIELD_JOIN_STRING), m_consume_delims(true)
    {}

    virtual ~LogCodec() {}

    virtual const std::string& getContentType(void) const { return CONTENT_TYPE; }

    virtual pion::platform::CodecPtr clone(void) const;

    virtual void write(std::ostream& out, const pion::platform::Event& e);

    virtual void finish(std::ostream& out) {};

    virtual bool read(std::istream& in, pion::platform::Event& e);

    virtual void setConfig(const pion::platform::Vocabulary& v, const xmlNodePtr config_ptr);

    virtual void updateVocabulary(const pion::platform::Vocabulary& v);

    inline void reset(void) {
        m_field_map.clear();
        m_format.clear();
        m_wrote_elf_headers = false;
    }


private:

    struct LogField {
        LogField(const std::string& field, const pion::platform::Vocabulary::Term& term, char delim_start,
                 char delim_end, bool opt_delims, bool urlencode, char escape_char, const std::string& empty_val,
                 bool do_time_offset, const PionDateTime::time_duration_type& time_offset)
            : log_field(field), log_term(term), log_delim_start(delim_start), log_delim_end(delim_end),
              log_opt_delims(opt_delims), log_urlencode(urlencode), log_escape_char(escape_char),
              log_empty_val(empty_val), log_do_time_offset(do_time_offset), log_time_offset(time_offset)
        {}

        LogField(const LogField& f)
            : log_field(f.log_field), log_term(f.log_term), log_delim_start(f.log_delim_start),
              log_delim_end(f.log_delim_end), log_opt_delims(f.log_opt_delims),
              log_urlencode(f.log_urlencode), log_escape_char(f.log_escape_char),
              log_empty_val(f.log_empty_val), log_do_time_offset(f.log_do_time_offset), log_time_offset(f.log_time_offset)
        {}

        inline LogField& operator=(const LogField& f) {
            log_field = f.log_field;
            log_term = f.log_term;
            log_delim_start = f.log_delim_start;
            log_delim_end = f.log_delim_end;
            log_opt_delims = f.log_opt_delims;
            log_urlencode = f.log_urlencode;
            log_escape_char = f.log_escape_char;
            log_empty_val = f.log_empty_val;
            log_do_time_offset = f.log_do_time_offset;
            log_time_offset = f.log_time_offset;
            return *this;
        }

        inline void writeEmptyValue(std::ostream& out) const {
            if (log_delim_start != '\0')
                out << log_delim_start;
            out << log_empty_val;
            if (log_delim_end != '\0')
                out << log_delim_end;
        }

        inline void write(std::ostream& out, const pion::platform::Event::ParameterValue& value);

        inline void read(const char *buf, pion::platform::Event& e);

        std::string                         log_field;
        pion::platform::Vocabulary::Term    log_term;
        PionTimeFacet                       log_time_facet;
        char                                log_delim_start;
        char                                log_delim_end;
        bool                                log_opt_delims;
        bool                                log_urlencode;
        char                                log_escape_char;
        std::string                         log_empty_val;
        bool                                log_do_time_offset;
        PionDateTime::time_duration_type    log_time_offset;
    };

    typedef boost::shared_ptr<LogField>     LogFieldPtr;

    typedef PION_HASH_MAP<std::string,
        LogFieldPtr, PION_HASH_STRING>      FieldMap;

    typedef std::vector<LogFieldPtr>        CurrentFormat;

    typedef std::istream::traits_type       traits_type;

    typedef std::basic_streambuf<std::istream::char_type,
        std::istream::traits_type>          streambuf_type;

    typedef std::istream::int_type          int_type;


    inline void mapFieldToTerm(const std::string& field, const pion::platform::Vocabulary::Term& term,
                               char delim_start, char delim_end, bool opt_delims,
                               bool urlencode, char escape_char, const std::string& empty_val,
                               bool do_time_offset, const PionDateTime::time_duration_type& time_offset);

    inline char * cstyle(char *cstring);

    inline void changeELFFormat(char *fmt);

    inline void writeELFHeaders(std::ostream& out) const;

    inline int_type consumeVoidsAndComments(streambuf_type *buf_ptr);


    static const std::string        CONTENT_TYPE;

    static const std::string        FLUSH_ELEMENT_NAME;

    static const std::string        HEADERS_ELEMENT_NAME;

    static const std::string        TIME_OFFSET_ELEMENT_NAME;

    static const std::string        FIELD_ELEMENT_NAME;

    static const std::string        TERM_ATTRIBUTE_NAME;

    static const std::string        START_ATTRIBUTE_NAME;

    static const std::string        END_ATTRIBUTE_NAME;

    static const std::string        OPTIONAL_ATTRIBUTE_NAME;

    static const std::string        URLENCODE_ATTRIBUTE_NAME;

    static const std::string        ESCAPE_ATTRIBUTE_NAME;

    static const std::string        EMPTY_ATTRIBUTE_NAME;

    static const std::string        EVENTS_ELEMENT_NAME;

    static const std::string        FIELDS_ELEMENT_NAME;

    static const std::string        SPLIT_ATTRIBUTE_NAME;

    static const std::string        JOIN_ATTRIBUTE_NAME;

    static const std::string        COMMENT_ATTRIBUTE_NAME;

    static const std::string        CONSUME_ATTRIBUTE_NAME;

    static const unsigned int       READ_BUFFER_SIZE;


    static const std::string        EVENT_SPLIT_SET;
    static const std::string        EVENT_JOIN_STRING;
    static const std::string        COMMENT_CHAR_SET;
    static const std::string        FIELD_SPLIT_SET;
    static const std::string        FIELD_JOIN_STRING;
    static const bool               CONSUME_DELIMS_FLAG;


    static const std::string        VERSION_ELF_HEADER;
    static const std::string        DATE_ELF_HEADER;
    static const std::string        SOFTWARE_ELF_HEADER;
    static const std::string        FIELDS_ELF_HEADER;


    boost::scoped_array<char>       m_read_buf;

    const char * const              m_read_end;

    FieldMap                        m_field_map;

    CurrentFormat                   m_format;

    bool                            m_flush_after_write;

    bool                            m_handle_elf_headers;

    bool                            m_wrote_elf_headers;

    boost::int32_t                  m_time_offset;

    std::string                     m_event_split;

    std::string                     m_event_join;

    std::string                     m_comment_chars;

    std::string                     m_field_split;

    std::string                     m_field_join;

    bool                            m_consume_delims;
};


// inline member functions for LogCodec

inline void LogCodec::mapFieldToTerm(const std::string& field, const pion::platform::Vocabulary::Term& term,
                                     char delim_start, char delim_end, bool opt_delims,
                                     bool urlencode, char escape_char, const std::string& empty_val,
                                     bool do_time_offset, const PionDateTime::time_duration_type& time_offset)
{
    for (FieldMap::const_iterator i = m_field_map.begin(); i != m_field_map.end(); ++i) {
        if (i->second->log_term.term_ref == term.term_ref)
            throw PionException("Duplicate Field Term");
    }

    if (m_field_map.find(field) != m_field_map.end())
        throw PionException("Duplicate Field Name");

    // prepare a new Logfield object
    LogFieldPtr field_ptr(new LogField(field, term, delim_start, delim_end, opt_delims, urlencode, escape_char, empty_val, do_time_offset, time_offset));
    switch (term.term_type) {
        case pion::platform::Vocabulary::TYPE_DATE_TIME:
        case pion::platform::Vocabulary::TYPE_DATE:
        case pion::platform::Vocabulary::TYPE_TIME:
            field_ptr->log_time_facet.setFormat(term.term_format);
            break;
        default:
            break; // do nothing
    }
    // add it to the mapping of field names
    m_field_map[field] = field_ptr;
    // append the new field to the current (default) format
    m_format.push_back(field_ptr);
}

inline char * LogCodec::cstyle(char *cstring)
{
    char *ptr = cstring;
    size_t len = strlen(cstring);
    size_t num, nlen;

    while ( (ptr = strchr(ptr, '\\')) ) {
        nlen = 1;
        switch (ptr[1]) {
        case 'a': *ptr = '\a'; break;
        case 'b': *ptr = '\b'; break;
        case 'f': *ptr = '\f'; break;
        case 'n': *ptr = '\n'; break;
        case 'r': *ptr = '\r'; break;
        case 't': *ptr = '\t'; break;
        case 'v': *ptr = '\v'; break;
        case '_': *ptr = ' '; break;
        case '0': case '1': case '2': case '3':
        case '4': case '5': case '6': case '7':
            nlen = sscanf(ptr + 1, "%o", &num);
            *ptr = (char)num;
            break;
        case 'x':
            nlen = sscanf(ptr + 1, "%x", &num);
            *ptr = (char)num;
            break;
        }
        num = ptr - cstring + nlen;
        ptr++;
        memmove(ptr, ptr + nlen, len - num);
    }

    return cstring;
}

inline void LogCodec::changeELFFormat(char *fmt)
{
    m_format.clear();
    char *ptr;
    bool last_field = false;
    while (!last_field && *fmt != '\0' && m_event_split.find(*fmt) == std::string::npos) {
        // skip leading spaces
        while (*fmt == ' ') ++fmt;
        // find the end of the field name
        ptr = fmt;
        while (*ptr != '\0' && m_event_split.find(*ptr) == std::string::npos && *ptr != ' ') ++ptr;
        // set last_field if we're at the end
        if (*ptr == '\0' || m_event_split.find(*ptr) != std::string::npos) last_field = true;
        *ptr = '\0';
        FieldMap::const_iterator i = m_field_map.find(fmt);
        if (i == m_field_map.end())
            throw BadFormatException(fmt);
        m_format.push_back(i->second);
        fmt = ptr + 1;
    }
}

inline void LogCodec::writeELFHeaders(std::ostream& out) const
{
    PionDateTime time_now(boost::posix_time::second_clock::universal_time());
    out << VERSION_ELF_HEADER << " 1.0" << m_event_join;
    out << DATE_ELF_HEADER << ' ' << time_now << m_event_join;
    out << SOFTWARE_ELF_HEADER << " Pion v" << PION_VERSION << m_event_join;
    out << FIELDS_ELF_HEADER;
    CurrentFormat::const_iterator i = m_format.begin();
    while (i != m_format.end())
        out << ' ' << (*i++)->log_field;
    out << m_event_join;
}

inline LogCodec::int_type LogCodec::consumeVoidsAndComments(streambuf_type *buf_ptr)
{
    int_type c = buf_ptr->sgetc();
    char * const read_buf = m_read_buf.get();
    char * read_ptr;

    while (!traits_type::eq_int_type(c, traits_type::eof())) {
        if (m_field_split.find(c) != std::string::npos || m_event_split.find(c) != std::string::npos) {
            c = buf_ptr->snextc();
        } else if (m_comment_chars.find(c) != std::string::npos) {
            // ignore comment line (sorta...)
            read_ptr = read_buf;
            do {
                // check for end of line
                if (m_event_split.find(c) != std::string::npos)
                    break;
                // read in the comment in case it matters...
                if (read_ptr < m_read_end)
                    *(read_ptr++) = c;
                // get the next character
                c = buf_ptr->snextc();
            } while (!traits_type::eq_int_type(c, traits_type::eof()));
            *read_ptr = '\0';
            if (m_handle_elf_headers) {
                // check if it is an ELF format change
                read_buf[FIELDS_ELF_HEADER.size()] = '\0';
                if (FIELDS_ELF_HEADER == read_buf)
                    changeELFFormat(read_buf + FIELDS_ELF_HEADER.size() + 1);
            }
        } else {
            break;
        }
    }
    return c;
}


// inline member functions for LogCodec::LogField

inline void LogCodec::LogField::write(std::ostream& out, const pion::platform::Event::ParameterValue& value)
{
    std::ostringstream oss;

    switch(log_term.term_type) {
        case pion::platform::Vocabulary::TYPE_NULL:
        case pion::platform::Vocabulary::TYPE_OBJECT:
            break;  // ignore unsupported field
        case pion::platform::Vocabulary::TYPE_INT8:
        case pion::platform::Vocabulary::TYPE_INT16:
        case pion::platform::Vocabulary::TYPE_INT32:
            oss << boost::get<boost::int32_t>(value);
            break;
        case pion::platform::Vocabulary::TYPE_INT64:
            oss << boost::get<boost::int64_t>(value);
            break;
        case pion::platform::Vocabulary::TYPE_UINT8:
        case pion::platform::Vocabulary::TYPE_UINT16:
        case pion::platform::Vocabulary::TYPE_UINT32:
            oss << boost::get<boost::uint32_t>(value);
            break;
        case pion::platform::Vocabulary::TYPE_UINT64:
            oss << boost::get<boost::uint64_t>(value);
            break;
        case pion::platform::Vocabulary::TYPE_FLOAT:
            oss << boost::get<float>(value);
            break;
        case pion::platform::Vocabulary::TYPE_DOUBLE:
            // using boost::lexical_cast<std::string> ensures precision appropriate to type double
            oss << boost::lexical_cast<std::string>(boost::get<double>(value));
            break;
        case pion::platform::Vocabulary::TYPE_LONG_DOUBLE:
            // using boost::lexical_cast<std::string> ensures precision appropriate to type long double
            oss << boost::lexical_cast<std::string>(boost::get<long double>(value));
            break;
        case pion::platform::Vocabulary::TYPE_SHORT_STRING:
        case pion::platform::Vocabulary::TYPE_STRING:
        case pion::platform::Vocabulary::TYPE_LONG_STRING:
        case pion::platform::Vocabulary::TYPE_BLOB:
        case pion::platform::Vocabulary::TYPE_ZBLOB:
        {
            const pion::platform::Event::BlobType& ss = boost::get<const pion::platform::Event::BlobType&>(value);
            if (ss.size() > 0) {
                if (log_urlencode) {
                    std::string temp_str(ss.get());
                    oss << algo::url_encode(temp_str);
                } else {
                    oss.write(ss.get(), ss.size());
                }
            }
            break;
        }
        case pion::platform::Vocabulary::TYPE_CHAR:
        {
            const pion::platform::Event::BlobType& ss = boost::get<const pion::platform::Event::BlobType&>(value);
            if (ss.size() > 0) {
                if (log_urlencode) {
                    std::string temp_str(ss.get());
                    temp_str = algo::url_encode(temp_str);
                    if (temp_str.size() > log_term.term_size)
                        temp_str.resize(log_term.term_size);
                    oss << temp_str;
                } else {
                    oss.write(ss.get(), ss.size() < log_term.term_size ? ss.size() : log_term.term_size);
                }
            }
            break;
        }
        case pion::platform::Vocabulary::TYPE_DATE_TIME:
        case pion::platform::Vocabulary::TYPE_DATE:
        case pion::platform::Vocabulary::TYPE_TIME:
        {
            PionDateTime dt = boost::get<const PionDateTime&>(value);
            if (log_do_time_offset) {
                dt -= log_time_offset;
            }
            if (log_urlencode) {
                std::string temp_str;
                log_time_facet.toString(temp_str, dt);
                oss << algo::url_encode(temp_str);
            } else {
                log_time_facet.write(oss, dt);
            }
            break;
        }
    }

    if (log_delim_start != '\0')
        out << log_delim_start;
    if (oss.str().empty())
        out << log_empty_val;
    else {
        std::string src = oss.str();
        std::string dst;
        for (std::string::iterator i = src.begin(); i != src.end(); i++) {
            if (*i == log_delim_end)
                dst.push_back(log_escape_char);
            dst.push_back(*i);
        }
        out << dst;
    }
    if (log_delim_end != '\0')
        out << log_delim_end;
}

inline void LogCodec::LogField::read(const char *buf, pion::platform::Event& e)
{
    switch(log_term.term_type) {
        case pion::platform::Vocabulary::TYPE_NULL:
        case pion::platform::Vocabulary::TYPE_OBJECT:
            break;  // ignore unsupported field
        case pion::platform::Vocabulary::TYPE_INT8:
        case pion::platform::Vocabulary::TYPE_INT16:
        case pion::platform::Vocabulary::TYPE_INT32:
            e.setInt(log_term.term_ref, boost::lexical_cast<boost::int32_t>(buf));
            break;
        case pion::platform::Vocabulary::TYPE_INT64:
            e.setBigInt(log_term.term_ref, boost::lexical_cast<boost::int64_t>(buf));
            break;
        case pion::platform::Vocabulary::TYPE_UINT8:
        case pion::platform::Vocabulary::TYPE_UINT16:
        case pion::platform::Vocabulary::TYPE_UINT32:
            e.setUInt(log_term.term_ref, boost::lexical_cast<boost::uint32_t>(buf));
            break;
        case pion::platform::Vocabulary::TYPE_UINT64:
            e.setUBigInt(log_term.term_ref, boost::lexical_cast<boost::uint64_t>(buf));
            break;
        case pion::platform::Vocabulary::TYPE_FLOAT:
            e.setFloat(log_term.term_ref, boost::lexical_cast<float>(buf));
            break;
        case pion::platform::Vocabulary::TYPE_DOUBLE:
            e.setDouble(log_term.term_ref, boost::lexical_cast<double>(buf));
            break;
        case pion::platform::Vocabulary::TYPE_LONG_DOUBLE:
            e.setLongDouble(log_term.term_ref, boost::lexical_cast<long double>(buf));
            break;
        case pion::platform::Vocabulary::TYPE_SHORT_STRING:
        case pion::platform::Vocabulary::TYPE_STRING:
        case pion::platform::Vocabulary::TYPE_LONG_STRING:
        case pion::platform::Vocabulary::TYPE_BLOB:
        case pion::platform::Vocabulary::TYPE_ZBLOB:
            if (log_urlencode) {
                std::string temp_str(algo::url_decode(buf));
                e.setString(log_term.term_ref, temp_str);
            } else {
                e.setString(log_term.term_ref, buf);
            }
            break;
        case pion::platform::Vocabulary::TYPE_CHAR:
            if (log_urlencode) {
                std::string temp_str(algo::url_decode(buf));
                if (temp_str.size() > log_term.term_size)
                    temp_str.resize(log_term.term_size);
                e.setString(log_term.term_ref, temp_str);
            } else if (strlen(buf) > log_term.term_size) {
                e.setString(log_term.term_ref, std::string(buf, log_term.term_size));
            } else {
                e.setString(log_term.term_ref, buf);
            }
            break;
        case pion::platform::Vocabulary::TYPE_DATE_TIME:
        case pion::platform::Vocabulary::TYPE_DATE:
        case pion::platform::Vocabulary::TYPE_TIME:
        {
            PionDateTime dt;
            if (log_urlencode) {
                std::string temp_str(algo::url_decode(buf));
                log_time_facet.fromString(temp_str, dt);
            } else {
                log_time_facet.fromString(buf, dt);
            }
            if (log_do_time_offset) {
                dt += log_time_offset;
            }
            e.setDateTime(log_term.term_ref, dt);
            break;
        }
    }
}


}   // end namespace plugins
}   // end namespace pion

#endif

---