stream.hpp

Go to the documentation of this file.

//  Copyright (c) 2005-2007 Andre Merzky (andre@merzky.net)
//  Copyright (c) 2005-2009 Hartmut Kaiser
//  Copyright (c) 2007 Ole Weidner (oweidner@cct.lsu.edu)
// 
//  Use, modification and distribution is subject to the Boost Software
//  License, Version 1.0. (See accompanying file LICENSE_1_0.txt or copy at
//  http://www.boost.org/LICENSE_1_0.txt)

#ifndef SAGA_COMM_STREAM_STREAM_HPP
#define SAGA_COMM_STREAM_STREAM_HPP

// include stl
#include <string>
#include <vector>

// include dependent spec sections
#include <saga/saga/util.hpp>
#include <saga/saga/call.hpp>
#include <saga/saga/base.hpp>
#include <saga/saga/session.hpp>
#include <saga/saga/task.hpp>
#include <saga/saga/buffer.hpp>
#include <saga/saga/url.hpp>
#include <saga/saga/context.hpp>

#include <saga/saga/detail/attribute.hpp>
#include <saga/saga/detail/monitorable.hpp>

// include package config
#include <saga/saga/packages/stream/config.hpp>

// suppress warnings about dependent classes not being exported from the dll
#if defined(BOOST_MSVC)
#pragma warning(push)
#pragma warning(disable: 4251 4231 4660)
#endif

namespace saga
{
    namespace stream {
    
    namespace attributes 
    {
        //  read write attributes
        char const* const stream_bufsize = "Bufsize";
        char const* const stream_timeout = "Timeout";
        char const* const stream_blocking = "Blocking";
        char const* const stream_compression = "Compression";
        char const* const stream_nodelay = "Nodelay";
        char const* const stream_reliable = "Reliable";
    }
    
    namespace metrics 
    {
        char const* const stream_state = "stream.State";
        char const* const stream_read = "stream.Read";
        char const* const stream_write = "stream.Write";
        char const* const stream_exception = "stream.Exception";
        char const* const stream_dropped = "stream.Dropped";
    }

    namespace attributes 
    {
        // string based representation of stream states
        char const* const stream_state_unknown = "Unknown";
        char const* const stream_state_new = "New";
        char const* const stream_state_open = "open";
        char const* const stream_state_closed = "Closed";
        char const* const stream_state_dropped = "Dropped";
        char const* const stream_state_error = "Error";
    }

    enum state
    {
        Unknown      = -1,
        New          =  1,
        Open         =  2,
        Closed       =  3,
        Dropped      =  4,
        Error        =  5
    };

    enum activity
    {
        Read         =  1,
        Write        =  2,
        Exception    =  4
    };

    class SAGA_STREAM_PACKAGE_EXPORT stream 
        : public saga::object,
          public saga::detail::attribute<stream>,
          public saga::detail::monitorable<stream>
    {

        friend struct saga::detail::attribute<stream>;    // needs to access get_impl()
        friend struct saga::detail::monitorable<stream>;  // needs to access get_impl()

        typedef saga::detail::attribute<stream> attribute_base_type;
        typedef saga::detail::monitorable<stream> monitorable_base_type;

    public:

      //typedef stream_base::state state; 

    protected:

        TR1::shared_ptr <saga::impl::stream> get_impl_sp(void) const;
        saga::impl::stream* get_impl (void) const;
        friend class saga::impl::stream;
        friend struct saga::detail::create_default<stream>;

        explicit stream (saga::impl::stream *);
        explicit stream (int);    // dummy constructor to create empty facade

    private:
        // factory
        SAGA_CALL_CREATE_PRIV_2(session const&, saga::url)

        // stream inspection methods
        SAGA_CALL_CONST_PRIV_0(get_url)
        SAGA_CALL_CONST_PRIV_0(get_context)

        // stream methods
        SAGA_CALL_PRIV_1(connect, double)
        SAGA_CALL_PRIV_2(wait, saga::stream::activity, double)
        SAGA_CALL_PRIV_1(close, double)
        SAGA_CALL_PRIV_2(read, saga::mutable_buffer, saga::ssize_t)
        SAGA_CALL_PRIV_2(write, saga::const_buffer, saga::ssize_t)

        void init_attributes();
        void init_metrics();

    public:
        explicit stream (session const & s, saga::url url = saga::url());

        explicit stream (saga::url url);

        stream ();

        explicit stream (saga::object const& o);

        ~stream (void);

        static stream create(session const& s, saga::url name = saga::url())
        {
            return stream(s, name);
        }
        SAGA_CALL_CREATE_2_DEF_1(session const&, saga::url, saga::url())

        static stream create(saga::url name = saga::url())
        {
            return stream(name);
        }
        template <typename Tag>
        static saga::task create(saga::url name = saga::url())
        {
            return create<Tag>(detail::get_the_session(), name);
        }

        stream &operator= (saga::object const& o);

        saga::url get_url() const
        {
            saga::task t = get_urlpriv(saga::task_base::Sync()); 
            return t.get_result<saga::url>();
        }
        SAGA_CALL_CONST_PUB_0_DEF_0(get_url)

        
        saga::context get_context() const
        {
            saga::task t = get_contextpriv(saga::task_base::Sync()); 
            return t.get_result<saga::context>();
        }
        SAGA_CALL_CONST_PUB_0_DEF_0(get_context)

        
        void connect(double timeout  = -1.0) 
        {
            saga::task t = connectpriv(timeout, saga::task_base::Sync()); 
            return t.get_result();
        }
        SAGA_CALL_PUB_1_DEF_1(connect, double, -1.0)

        
        saga::stream::activity
        wait(saga::stream::activity what, double timeout = -1.0) 
        {
            saga::task t = waitpriv(what, timeout, saga::task_base::Sync()); 
            return t.get_result<saga::stream::activity>();
        }
        SAGA_CALL_PUB_2_DEF_1(wait, saga::stream::activity, double, -1.0)

        
        void close(double timeout = 0.0) 
        {
          saga::task t = closepriv(timeout, saga::task_base::Sync()); 
          t.get_result ();
        }
        SAGA_CALL_PUB_1_DEF_1(close, double, 0.0)

        
        saga::ssize_t read(saga::mutable_buffer buffer, saga::ssize_t length = 0) 
        {
            saga::task t = readpriv(buffer, length, saga::task_base::Sync()); 
            return t.get_result<saga::ssize_t>();
        }
        SAGA_CALL_PUB_2_DEF_1(read, saga::mutable_buffer, saga::ssize_t, 0)

        
        saga::ssize_t write(saga::const_buffer buffer, saga::ssize_t length = 0) 
        {
            saga::task t = writepriv(buffer, length, saga::task_base::Sync()); 
            return t.get_result<saga::ssize_t>();
        }
        SAGA_CALL_PUB_2_DEF_1(write, saga::const_buffer, saga::ssize_t, 0)
    };

  }

  namespace detail
  {
      // we need to specialize the create_default template (used in hold_any) 
      // to be able to create a truly default constructed stream (empty 
      // facade)
      template<>
      struct create_default<saga::stream::stream>
      {
          static saga::stream::stream* call()
          {
              return new saga::stream::stream(1);    // use dummy constructor
          }
          template <typename T_> static void call(T_* obj) 
          { 
              new (obj) saga::stream::stream(1); 
          }
      };
  }

    namespace adaptors
    {
        saga::stream::state SAGA_STREAM_PACKAGE_EXPORT 
            stream_state_value_to_enum(std::string const& val);
        std::string SAGA_STREAM_PACKAGE_EXPORT 
            stream_state_enum_to_value(int s);
    }
} 

// re-enable warnings about dependent classes not being exported from the dll
#if defined(BOOST_MSVC)
#pragma warning(pop)
#endif

#endif // SAGA_COMM_STREAM_STREAM_HPP