platform/reactors/ScriptReactor.hpp

// ------------------------------------------------------------------------
// Pion is a development platform for building Reactors that process Events
// ------------------------------------------------------------------------
// Copyright (C) 2007-2009 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_SCRIPTREACTOR_HEADER__
#define __PION_SCRIPTREACTOR_HEADER__

#ifdef _MSC_VER
    #include <windows.h>
    #include <string.h>
#else
    #include <errno.h>
    #include <unistd.h>
    #include <sys/wait.h>
#endif
#include <iosfwd>
#include <string>
#include <vector>
#include <boost/scoped_ptr.hpp>
#include <boost/thread/thread.hpp>
#include <boost/iostreams/stream_buffer.hpp>
#include <boost/iostreams/device/file_descriptor.hpp>
#include <pion/PionConfig.hpp>
#include <pion/PionLogger.hpp>
#include <pion/PionException.hpp>
#include <pion/PionScheduler.hpp>
#include <pion/platform/Codec.hpp>
#include <pion/platform/Reactor.hpp>

#ifdef _MSC_VER
    #define FILE_DESC_TYPE          HANDLE
    #define INVALID_DESCRIPTOR      INVALID_HANDLE_VALUE
    #define CLOSE_DESCRIPTOR(x)     CloseHandle(x)
    #define PROCESS_INFO_TYPE       LPPROCESS_INFORMATION
    #define INVALID_PROCESS         NULL
#else
    #define FILE_DESC_TYPE          int
    #define INVALID_DESCRIPTOR      -1
    #define CLOSE_DESCRIPTOR(x)     ::close(x)
    #define PROCESS_INFO_TYPE       pid_t
    #define INVALID_PROCESS         -1
#endif


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


class ScriptReactor :
    public pion::platform::Reactor
{
public: 

    class EmptyInputCodecException : public PionException {
    public:
        EmptyInputCodecException(const std::string& reactor_id)
            : PionException("ScriptReactor configuration is missing a required InputCodec parameter: ", reactor_id) {}
    };

    class EmptyOutputCodecException : public PionException {
    public:
        EmptyOutputCodecException(const std::string& reactor_id)
            : PionException("ScriptReactor configuration is missing a required OutputCodec parameter: ", reactor_id) {}
    };

    class EmptyCommandException : public PionException {
    public:
        EmptyCommandException(const std::string& reactor_id)
            : PionException("ScriptReactor configuration is missing a required Command parameter: ", reactor_id) {}
    };

    class CommandParsingException : public PionException {
    public:
        CommandParsingException(const std::string& command)
            : PionException("ScriptReactor was unable to parse arguments out of command string: ", command) {}
    };

    class OpenPipeException : public PionException {
    public:
        OpenPipeException(const std::string& command)
            : PionException("ScriptReactor failed to open pipe: ", command) {}
    };

    class ReadFromPipeException : public PionException {
    public:
        ReadFromPipeException(const std::string& reactor_id)
            : PionException("ScriptReactor failed reading an event from pipe: ", reactor_id) {}
    };

    class WriteToPipeException : public PionException {
    public:
        WriteToPipeException(const std::string& reactor_id)
            : PionException("ScriptReactor failed writing an event to pipe: ", reactor_id) {}
    };


    ScriptReactor(void)
        : Reactor(TYPE_PROCESSING),
        m_logger(PION_GET_LOGGER("pion.ScriptReactor")),
        m_input_pipe(INVALID_DESCRIPTOR), m_output_pipe(INVALID_DESCRIPTOR), m_child(INVALID_PROCESS)
    {
    }

    virtual ~ScriptReactor() { stop(); }

    virtual void start(void);

    virtual void stop(void) { stopIfRunning(); }

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

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

    virtual void updateCodecs(void);

    virtual void process(const pion::platform::EventPtr& e);

    inline void setLogger(PionLogger log_ptr) { m_logger = log_ptr; }

    inline PionLogger getLogger(void) { return m_logger; }


private:

    bool stopIfRunning(void);

    void readEvents(void);

    void openPipe(void);

    void closePipe(void);

    void parseArguments(void);


#ifdef BOOST_IOSTREAMS_WINDOWS
    struct winpipe_handle_source : private boost::iostreams::file_descriptor {
        typedef char char_type;
        struct category : boost::iostreams::source_tag, boost::iostreams::closable_tag { };
        std::streamsize read(char_type* s, std::streamsize n) {
            DWORD result = 0, error = 0;
            if (!::ReadFile(handle(), s, n, &result, NULL) && (error = GetLastError()) != ERROR_BROKEN_PIPE)
                throw boost::iostreams::detail::bad_read();
            return error == ERROR_BROKEN_PIPE ? -1 : static_cast<std::streamsize>(result);
        }
        using boost::iostreams::file_descriptor::close;
        using boost::iostreams::file_descriptor::handle;
        template <typename FLAG_TYPE>
        winpipe_handle_source(HANDLE h, FLAG_TYPE f) : boost::iostreams::file_descriptor(h, f) { }
        winpipe_handle_source(const winpipe_handle_source& w) :
            boost::iostreams::file_descriptor(static_cast<const boost::iostreams::file_descriptor&>(w)) { }
    };

    struct winpipe_handle_sink : private boost::iostreams::file_descriptor {
        typedef char char_type;
        struct category : boost::iostreams::sink_tag, boost::iostreams::closable_tag { };
        std::streamsize write(const char_type* s, std::streamsize n) {
            DWORD ignore = 0;
            if (!::WriteFile(handle(), s, n, &ignore, NULL))
                throw boost::iostreams::detail::bad_write();
            return n;
        }
        using boost::iostreams::file_descriptor::close;
        using boost::iostreams::file_descriptor::handle;
        template <typename FLAG_TYPE>
        winpipe_handle_sink(HANDLE h, FLAG_TYPE f) : boost::iostreams::file_descriptor(h, f) { }
        winpipe_handle_sink(const winpipe_handle_sink& w) :
            boost::iostreams::file_descriptor(static_cast<const boost::iostreams::file_descriptor&>(w)) { }
    };

    typedef boost::iostreams::stream_buffer<winpipe_handle_source>  IStreamBuffer;
    typedef boost::iostreams::stream_buffer<winpipe_handle_sink>    OStreamBuffer;
#else
    typedef boost::iostreams::stream_buffer<boost::iostreams::file_descriptor_source>   IStreamBuffer;
    typedef boost::iostreams::stream_buffer<boost::iostreams::file_descriptor_sink>     OStreamBuffer;
#endif


    static const std::string            INPUT_CODEC_ELEMENT_NAME;

    static const std::string            OUTPUT_CODEC_ELEMENT_NAME;

    static const std::string            COMMAND_ELEMENT_NAME;


    PionLogger                          m_logger;

    pion::platform::CodecPtr            m_input_codec_ptr;

    pion::platform::CodecPtr            m_output_codec_ptr;

    std::string                         m_input_codec_id;

    std::string                         m_output_codec_id;

    std::string                         m_command;

    std::vector<std::string>            m_args;

    FILE_DESC_TYPE                      m_input_pipe;

    FILE_DESC_TYPE                      m_output_pipe;

    PROCESS_INFO_TYPE                   m_child;

    boost::scoped_ptr<OStreamBuffer>    m_input_streambuf_ptr;

    boost::scoped_ptr<IStreamBuffer>    m_output_streambuf_ptr;

    boost::scoped_ptr<std::ostream>     m_input_stream_ptr;

    boost::scoped_ptr<std::istream>     m_output_stream_ptr;

    boost::scoped_ptr<boost::thread>    m_thread_ptr;

    boost::mutex                        m_write_mutex;
};


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

#endif

---