platform/codecs/JSONCodec.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_JSONCODEC_HEADER__
#define __PION_JSONCODEC_HEADER__

#include <vector>
#include <map>
#include <queue>
#include <pion/PionConfig.hpp>
#include <pion/platform/Codec.hpp>
#include <yajl/yajl_gen.h>


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


struct Context;

class JSONCodec :
    public pion::platform::Codec
{
public:

    // TODO: EmptyFieldException and EmptyTermException should probably be moved
    // to class Codec, and removed from LogCodec.

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

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


    JSONCodec(void)
        : pion::platform::Codec(),
        m_flush_after_write(false), m_yajl_generator(NULL), m_yajl_handle(NULL),
        m_no_events_written(true), m_first_read_attempt(true)
    {}
    
    virtual ~JSONCodec() {
        if (m_yajl_generator)
            yajl_gen_free(m_yajl_generator);
        if (m_yajl_handle)
            yajl_free(m_yajl_handle);
    }
    
    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();
    }

    struct JSONField {
        JSONField(const std::string& f, const pion::platform::Vocabulary::Term& t)
            : field_name(f), term(t)
        {}

        std::string                         field_name;

        pion::platform::Vocabulary::Term    term;

        PionTimeFacet                       time_facet;
    };

    typedef boost::shared_ptr<JSONField>    JSONFieldPtr;
    
    typedef PION_HASH_MAP<std::string,
        JSONFieldPtr,
        PION_HASH_STRING>                   FieldMap;

    typedef std::vector<JSONFieldPtr>       CurrentFormat;

    typedef PION_HASH_MULTIMAP<pion::platform::Vocabulary::TermRef, std::string>
                                            JSONObject;

    typedef boost::shared_ptr<JSONObject>   JSONObjectPtr;

    typedef std::queue<JSONObjectPtr>       JSONObjectQueue;

private:

    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);


    static const std::string        CONTENT_TYPE;

    static const std::string        FIELD_ELEMENT_NAME;

    static const std::string        TERM_ATTRIBUTE_NAME;

    static const unsigned int       READ_BUFFER_SIZE;

    static std::string              INDENT_STRING;


    FieldMap                        m_field_map;

    std::map<pion::platform::Vocabulary::TermRef, JSONFieldPtr>
                                    m_JSON_field_ptr_map;

    CurrentFormat                   m_format;

    bool                            m_flush_after_write;

    yajl_gen                        m_yajl_generator;

    yajl_handle                     m_yajl_handle;

    JSONObjectQueue                 m_json_object_queue;

    boost::shared_ptr<Context>      m_context;

    bool                            m_no_events_written;

    bool                            m_first_read_attempt;
};


// inline member functions for JSONCodec

inline void JSONCodec::mapFieldToTerm(const std::string& field,
                                      const pion::platform::Vocabulary::Term& term)
{
    if (m_field_map[field])
        throw PionException("Duplicate Field Name");

    // prepare a new JSON field object
    JSONFieldPtr field_ptr(new JSONField(field, term));
    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->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 format
    m_format.push_back(field_ptr);
}

// TODO: put this somewhere
struct Context {
    Context(const JSONCodec::FieldMap& field_map, JSONCodec::JSONObjectQueue& json_object_queue)
        : field_map(field_map), json_object_queue(json_object_queue)
        , m_array_started(false), m_array_ended(false)
    {}

    const JSONCodec::FieldMap& field_map;
    JSONCodec::JSONObjectQueue& json_object_queue;
    JSONCodec::JSONObjectPtr json_object_ptr;
    pion::platform::Vocabulary::TermRef term_ref;
    bool m_array_started;
    bool m_array_ended;
};

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

#endif

---