implemented json stream connection

CMakeLists.txt

@@ -41,12 +41,14 @@ option( UNITY_BUILD OFF )
 
 include_directories( vendor/boost/process/include )
 include_directories( ${Boost_INCLUDE_DIR} )
+include_directories( ${OPENSSL_INCLUDE_DIR} )
 include_directories( include )
 
 set( sources 
   src/process.cpp
   src/http_connection.cpp
   src/json_rpc_connection.cpp
+  src/json_rpc_stream_connection.cpp
   src/value.cpp
   src/lexical_cast.cpp
   src/spin_lock.cpp 
@@ -82,7 +84,7 @@ set( sources
 )
 setup_library( fc SOURCES ${sources} )
 
-setup_executable( json_rpc_test SOURCES tests/json_rpc_test.cpp LIBRARIES fc ${Boost_THREAD_LIBRARY} ${Boost_CONTEXT_LIBRARY} ${Boost_SYSTEM_LIBRARY} ${Boost_CHRONO_LIBRARY} ${Boost_FILESYSTEM_LIBRARY} )
+setup_executable( json_rpc_test SOURCES tests/json_rpc_test.cpp LIBRARIES fc ${pthread_library} ${rt_library} ${Boost_THREAD_LIBRARY} ${Boost_CONTEXT_LIBRARY} ${Boost_SYSTEM_LIBRARY} ${Boost_CHRONO_LIBRARY} ${Boost_FILESYSTEM_LIBRARY} ${rt_library} )
 
 #add_executable( test_vec tests/vector_test.cpp )
 #target_link_libraries( test_vec fc ${Boost_SYSTEM_LIBRARY} ${Boost_CHRONO_LIBRARY} ${Boost_THREAD_LIBRARY} ${Boost_CONTEXT_LIBRARY} )

include/fc/function.hpp

@@ -28,6 +28,8 @@ class function {
     template<typename... Args2>
     R operator()( Args2... args2)const { return func->call(fc::forward<Args2>(args2)...); }
 
+    bool operator!()const { return !func; }
+
   protected:
 
     struct impl_base : public fc::retainable {

include/fc/iostream.hpp

@@ -76,12 +76,4 @@ namespace fc {
       v = fc::lexical_cast<T>(str); 
       return o;
   }
-
-  template<typename T>
-  cin_t& operator>>( cin_t& o, T& v ) {
-      fc::string str;
-      getline( o, str, ' ' );
-      v = fc::lexical_cast<T>(str); 
-      return o;
-  }
 }

include/fc/json_rpc_connection.hpp

@@ -2,8 +2,16 @@
 #include <fc/json.hpp>
 #include <fc/future.hpp>
 #include <fc/function.hpp>
+#include <fc/ptr.hpp>
 
 namespace fc {  namespace json {
+  class rpc_connection;
+
+  struct rpc_server_method : public fc::retainable {
+    typedef fc::shared_ptr<rpc_server_method> ptr;
+    virtual value call( const value& v ) = 0;
+  };
+
   namespace detail {
     struct pending_result : virtual public promise_base {
       typedef shared_ptr<pending_result> ptr;
@@ -31,20 +39,28 @@ namespace fc {  namespace json {
         ~pending_result_impl(){}
     };
 
-    struct server_method : public fc::retainable {
-       typedef fc::shared_ptr<server_method> ptr;
-       virtual value call( const value& param ) = 0;
-    };
+
     template<typename R, typename Args>
-    struct server_method_impl : public server_method {
-       server_method_impl( const fc::function<R,Args>& a ):func(a){};
-       virtual value call( const value& param ) {
-          return value( func( value_cast<Args>(param) ) );
-       }
-       fc::function<R,Args> func;
+    struct rpc_server_method_impl : public rpc_server_method {
+      rpc_server_method_impl( const fc::function<R,Args>& f ):func(f){} 
+      virtual value call( const value& v ) {
+        return value( func( fc::value_cast<Args>( v ) ) ); 
+      }
+      fc::function<R,Args> func;
     };
 
-  } // namespace detail 
+    template<typename InterfaceType>
+    struct add_method_visitor {
+      public:
+        add_method_visitor( const fc::ptr<InterfaceType>& p, fc::json::rpc_connection& c ):_ptr(p){}
+
+        template<typename R, typename Args, typename Type>
+        void operator()( const char* name, fc::function<R,Args>& meth, Type );
+
+        const fc::ptr<InterfaceType>& _ptr;
+        fc::json::rpc_connection&     _con;
+    };
+  }
 
   /**
    *  This is the base JSON RPC connection that handles the protocol
@@ -75,22 +91,40 @@ namespace fc {  namespace json {
 
       template<typename R, typename Args >
       void add_method( const fc::string& name, const fc::function<R,Args>& a ) {
-         this->add_method( name, detail::server_method::ptr(new detail::server_method_impl<R,Args>(a) ) );
+         this->add_method( name, rpc_server_method::ptr(new detail::rpc_server_method_impl<R,Args>(a) ) );
       }
 
+      template<typename InterfaceType>
+      void add_interface( const fc::ptr<InterfaceType>& it ) {
+        it->template visit<InterfaceType>( detail::add_method_visitor<InterfaceType>( it, *this ) );
+      }
+
+      void add_method( const fc::string& name, const fc::json::rpc_server_method::ptr& func );
+
     protected:
       void         handle_message( const value& m );
-      virtual void send_invoke( uint64_t id, const fc::string& m, value&& param );
-      virtual void send_error( uint64_t id, int64_t code, const fc::string& msg );
-      virtual void send_result( uint64_t id, value&& r );
+      virtual void send_invoke( uint64_t id, const fc::string& m, value&& param ) = 0;
+      virtual void send_error( uint64_t id, int64_t code, const fc::string& msg ) = 0;
+      virtual void send_result( uint64_t id, value&& r ) = 0;
+
 
     private:
       void invoke( detail::pending_result::ptr&& p, const fc::string& m, value&& param );
-      void add_method( const fc::string& name, detail::server_method::ptr&& m );
+      void add_method( const fc::string& name, rpc_server_method::ptr&& m );
 
       class impl;
       fc::shared_ptr<class impl> my;
   };
 
+  namespace detail {
+
+    template<typename InterfaceType>
+    template<typename R, typename Args, typename Type>
+    void add_method_visitor<InterfaceType>::operator()( const char* name, fc::function<R,Args>& meth, Type ) {
+        _con.add_method( name, rpc_server_method::ptr( new rpc_server_method_impl<R,Args>(meth) ) );
+    }
+
+  } // namespace detail
+
 } } // fc::json
 

include/fc/reflect.hpp

@@ -9,20 +9,12 @@
 #define _FC_REFLECT_HPP_
 
 #include <fc/utility.hpp>
-#include <boost/static_assert.hpp>
-//#include <boost/preprocessor/seq/for_each_i.hpp>
-#include <boost/preprocessor/seq/for_each.hpp>
 #include <boost/preprocessor/seq/enum.hpp>
 #include <boost/preprocessor/seq/seq.hpp>
 #include <boost/preprocessor/stringize.hpp>
-//#include <boost/preprocessor/tuple/elem.hpp>
-//#include <boost/preprocessor/facilities/empty.hpp>
 #include <stdint.h>
 
 
-//#include <mace/void.hpp>
-//#include <mace/reflect/typeinfo.hpp>
-
 namespace fc { 
 
 /**

include/fc/sha1.hpp

@@ -34,7 +34,7 @@ namespace fc {
 
         private:
           struct      impl;
-          fwd<impl,89> my;
+          fwd<impl,96> my;
       };
 
       template<typename T>

src/json_rpc_connection.cpp

@@ -23,8 +23,8 @@ namespace fc { namespace json {
     public:
       impl():_next_req_id(0){ }
       ~impl(){ cancel_pending_requests();  }
-      int64_t                                                    _next_req_id;
-      std::unordered_map<std::string,detail::server_method::ptr> _methods;
+      int64_t                                                _next_req_id;
+      std::unordered_map<std::string,rpc_server_method::ptr> _methods;
 
       detail::pending_result::ptr _pr_head;
       detail::pending_result::ptr _pr_tail;

src/json_rpc_stream_connection.cpp

@@ -1,45 +1,78 @@
+#include <fc/json_rpc_stream_connection.hpp>
+#include <fc/iostream.hpp>
+#include <fc/thread.hpp>
 
-      /** note the life of i and o must be longer than rpc_connection's life */
-      rpc_connection( istream& i, ostream& o );
+namespace fc { namespace json {
 
-      /** note the life of i and o must be longer than rpc_connection's life */
-      void init( istream& i, ostream& o );
+  class rpc_stream_connection::impl : public fc::retainable {
+    public:
+      fc::istream&           in;
+      fc::ostream&           out;
+      rpc_stream_connection& self;
+      fc::function<void>     on_close;
 
-      istream* _in;
-      ostream* _out;
+      impl( fc::istream& i, fc::ostream& o, rpc_stream_connection& s )
+      :in(i),out(o),self(s){
+        _read_loop_complete = fc::async( [=](){ read_loop(); } ); 
+      }
+      
+      ~impl() {
+        try {
+            self.cancel_pending_requests();
+            _read_loop_complete.cancel();
+            _read_loop_complete.wait();
+        } catch ( ... ) {}
+      }
 
       fc::future<void> _read_loop_complete;
       void read_loop() {
         fc::string line;
-        fc::getline( *_in, line );
-        while( !_in->eof() ) {
+        fc::getline( in, line );
+        while( !in.eof() ) {
             try {
                 fc::value v= fc::json::from_string( line );
 
             } catch (...) {
               wlog( "%s", fc::except_str().c_str() );
             }
-            fc::getline( *_in, line );
+            fc::getline( in, line );
         }
-        slog( "Exit Read Loop, canceling waiting tasks!" );
-
-        auto cur = _pr_head;
-        while( cur ) {
-          cur->handle_error( "Connection Closed" );
-          cur = cur->next;
-        }
-
-        _pr_head.reset();
-        _pr_tail.reset();
+        self.cancel_pending_requests();
+        if( !!on_close ) on_close();
       }
+  };
 
-  rpc_connection::rpc_connection( istream& i, ostream& o )
-  :my( new impl() )
-  {
-    init( i, o );
+  rpc_stream_connection::rpc_stream_connection( fc::istream& i, fc::ostream& o )
+  :my( new impl(i,o,*this) ){
   }
-  void rpc_connection::init( istream& i, ostream& o ) {
-    my->_in  = &i;
-    my->_out = &o;
-    my->_read_loop_complete = fc::async( [=](){ my->read_loop(); } );
+
+  // the life of the streams must exceed the life of all copies
+  // of this rpc_stream_connection
+  void rpc_stream_connection::open( fc::istream& i, fc::ostream& o) {
+    my.reset( new impl(i,o,*this) );
   }
+
+  // cancels all pending requests, closes the ostream
+  // results on_close() being called if the stream is not already closed.
+  void rpc_stream_connection::close() {
+    my->out.close();
+  }
+
+  /**
+   *  When the connection is closed, call the given method
+   */
+  void rpc_stream_connection::on_close( const fc::function<void>& oc ) {
+    my->on_close = oc;
+  }
+
+  void rpc_stream_connection::send_invoke( uint64_t id, const fc::string& m, value&& param ) {
+  }
+  void rpc_stream_connection::send_error( uint64_t id, int64_t code, const fc::string& msg ) {
+  }
+  void rpc_stream_connection::send_result( uint64_t id, value&& r ) {
+  }
+
+} } // fc::json
+
+
+

src/value.cpp

@@ -270,6 +270,10 @@ value& value::operator=( const value& v ){
   gh(v.holder)->copy_helper(holder);
   return *this;
 }
+bool value::is_null()const {
+    return strcmp(gh(holder)->type(), "null") == 0;
+}
+
 
 value::object::const_iterator value::find( const char* key )const {
   if( strcmp(gh(holder)->type(), "object") == 0) {