Add write_handler to iostream as an ostream replacement

changelog.md

@@ -9,6 +9,9 @@ HEAD
   iostream transport. This provides a clean interface for triggering the shut 
   down of external sockets and other cleanup without hooking in to higher level
   WebSocket handlers.
+- Feature: Adds the ability to register a write handler when using the iostream
+  transport. This handler can be used to handle transport output in place of
+  registering an ostream to write to.
 - Improvement: Message payload logging now prints text for text messages rather
   than binary.
 - Documentation: Add Sending & Receiving Messages step to chapter one of the

test/transport/iostream/connection.cpp

@@ -170,11 +170,11 @@ config::elog_type elogger;
 BOOST_AUTO_TEST_CASE( const_methods ) {
     iostream_con::ptr con(new iostream_con(true,alogger,elogger));
 
-    BOOST_CHECK( con->is_secure() == false );
-    BOOST_CHECK( con->get_remote_endpoint() == "iostream transport" );
+    BOOST_CHECK( !con->is_secure() );
+    BOOST_CHECK_EQUAL( con->get_remote_endpoint(), "iostream transport" );
 }
 
-BOOST_AUTO_TEST_CASE( write_before_ostream_set ) {
+BOOST_AUTO_TEST_CASE( write_before_output_method_set ) {
     stub_con::ptr con(new stub_con(true,alogger,elogger));
 
     con->write("foo");
@@ -182,10 +182,10 @@ BOOST_AUTO_TEST_CASE( write_before_ostream_set ) {
 
     std::vector<websocketpp::transport::buffer> bufs;
     con->write(bufs);
-    BOOST_CHECK( con->ec == make_error_code(websocketpp::transport::iostream::error::output_stream_required) );
+    BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::transport::iostream::error::output_stream_required) );
 }
 
-BOOST_AUTO_TEST_CASE( async_write ) {
+BOOST_AUTO_TEST_CASE( async_write_ostream ) {
     stub_con::ptr con(new stub_con(true,alogger,elogger));
 
     std::stringstream output;
@@ -195,10 +195,43 @@ BOOST_AUTO_TEST_CASE( async_write ) {
     con->write("foo");
 
     BOOST_CHECK( !con->ec );
-    BOOST_CHECK( output.str() == "foo" );
+    BOOST_CHECK_EQUAL( output.str(), "foo" );
 }
 
-BOOST_AUTO_TEST_CASE( async_write_vector_0 ) {
+websocketpp::lib::error_code write_handler(std::string & o, websocketpp::connection_hdl, char const * buf, size_t len) {
+    o += std::string(buf,len);
+    return websocketpp::lib::error_code();
+}
+
+websocketpp::lib::error_code write_handler_error(websocketpp::connection_hdl, char const *, size_t) {
+    return make_error_code(websocketpp::transport::error::general);
+}
+
+BOOST_AUTO_TEST_CASE( async_write_handler ) {
+    stub_con::ptr con(new stub_con(true,alogger,elogger));
+    std::string output;
+
+    con->set_write_handler(websocketpp::lib::bind(
+        &write_handler,
+        websocketpp::lib::ref(output),
+        websocketpp::lib::placeholders::_1,
+        websocketpp::lib::placeholders::_2,
+        websocketpp::lib::placeholders::_3
+    ));
+    con->write("foo");
+    BOOST_CHECK( !con->ec );
+    BOOST_CHECK_EQUAL(output, "foo");
+}
+
+BOOST_AUTO_TEST_CASE( async_write_handler_error ) {
+    stub_con::ptr con(new stub_con(true,alogger,elogger));
+
+    con->set_write_handler(&write_handler_error);
+    con->write("foo");
+    BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::transport::error::general) );
+}
+
+BOOST_AUTO_TEST_CASE( async_write_vector_0_ostream ) {
     std::stringstream output;
 
     stub_con::ptr con(new stub_con(true,alogger,elogger));
@@ -209,10 +242,31 @@ BOOST_AUTO_TEST_CASE( async_write_vector_0 ) {
     con->write(bufs);
 
     BOOST_CHECK( !con->ec );
-    BOOST_CHECK( output.str() == "" );
+    BOOST_CHECK_EQUAL( output.str(), "" );
 }
 
-BOOST_AUTO_TEST_CASE( async_write_vector_1 ) {
+BOOST_AUTO_TEST_CASE( async_write_vector_0_write_handler ) {
+    std::string output;
+
+    stub_con::ptr con(new stub_con(true,alogger,elogger));
+    
+    con->set_write_handler(websocketpp::lib::bind(
+        &write_handler,
+        websocketpp::lib::ref(output),
+        websocketpp::lib::placeholders::_1,
+        websocketpp::lib::placeholders::_2,
+        websocketpp::lib::placeholders::_3
+    ));
+
+    std::vector<websocketpp::transport::buffer> bufs;
+
+    con->write(bufs);
+
+    BOOST_CHECK( !con->ec );
+    BOOST_CHECK_EQUAL( output, "" );
+}
+
+BOOST_AUTO_TEST_CASE( async_write_vector_1_ostream ) {
     std::stringstream output;
 
     stub_con::ptr con(new stub_con(true,alogger,elogger));
@@ -227,10 +281,34 @@ BOOST_AUTO_TEST_CASE( async_write_vector_1 ) {
     con->write(bufs);
 
     BOOST_CHECK( !con->ec );
-    BOOST_CHECK( output.str() == "foo" );
+    BOOST_CHECK_EQUAL( output.str(), "foo" );
 }
 
-BOOST_AUTO_TEST_CASE( async_write_vector_2 ) {
+BOOST_AUTO_TEST_CASE( async_write_vector_1_write_handler ) {
+    std::string output;
+
+    stub_con::ptr con(new stub_con(true,alogger,elogger));
+    con->set_write_handler(websocketpp::lib::bind(
+        &write_handler,
+        websocketpp::lib::ref(output),
+        websocketpp::lib::placeholders::_1,
+        websocketpp::lib::placeholders::_2,
+        websocketpp::lib::placeholders::_3
+    ));
+
+    std::vector<websocketpp::transport::buffer> bufs;
+
+    std::string foo = "foo";
+
+    bufs.push_back(websocketpp::transport::buffer(foo.data(),foo.size()));
+
+    con->write(bufs);
+
+    BOOST_CHECK( !con->ec );
+    BOOST_CHECK_EQUAL( output, "foo" );
+}
+
+BOOST_AUTO_TEST_CASE( async_write_vector_2_ostream ) {
     std::stringstream output;
 
     stub_con::ptr con(new stub_con(true,alogger,elogger));
@@ -247,7 +325,33 @@ BOOST_AUTO_TEST_CASE( async_write_vector_2 ) {
     con->write(bufs);
 
     BOOST_CHECK( !con->ec );
-    BOOST_CHECK( output.str() == "foobar" );
+    BOOST_CHECK_EQUAL( output.str(), "foobar" );
+}
+
+BOOST_AUTO_TEST_CASE( async_write_vector_2_write_handler ) {
+    std::string output;
+
+    stub_con::ptr con(new stub_con(true,alogger,elogger));
+    con->set_write_handler(websocketpp::lib::bind(
+        &write_handler,
+        websocketpp::lib::ref(output),
+        websocketpp::lib::placeholders::_1,
+        websocketpp::lib::placeholders::_2,
+        websocketpp::lib::placeholders::_3
+    ));
+
+    std::vector<websocketpp::transport::buffer> bufs;
+
+    std::string foo = "foo";
+    std::string bar = "bar";
+
+    bufs.push_back(websocketpp::transport::buffer(foo.data(),foo.size()));
+    bufs.push_back(websocketpp::transport::buffer(bar.data(),bar.size()));
+
+    con->write(bufs);
+
+    BOOST_CHECK( !con->ec );
+    BOOST_CHECK_EQUAL( output, "foobar" );
 }
 
 BOOST_AUTO_TEST_CASE( async_read_at_least_too_much ) {
@@ -256,7 +360,7 @@ BOOST_AUTO_TEST_CASE( async_read_at_least_too_much ) {
     char buf[10];
 
     con->async_read_at_least(11,buf,10);
-    BOOST_CHECK( con->ec == make_error_code(websocketpp::transport::iostream::error::invalid_num_bytes) );
+    BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::transport::iostream::error::invalid_num_bytes) );
 }
 
 BOOST_AUTO_TEST_CASE( async_read_at_least_double_read ) {
@@ -266,7 +370,7 @@ BOOST_AUTO_TEST_CASE( async_read_at_least_double_read ) {
 
     con->async_read_at_least(5,buf,10);
     con->async_read_at_least(5,buf,10);
-    BOOST_CHECK( con->ec == make_error_code(websocketpp::transport::iostream::error::double_read) );
+    BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::transport::iostream::error::double_read) );
 }
 
 BOOST_AUTO_TEST_CASE( async_read_at_least ) {
@@ -277,32 +381,32 @@ BOOST_AUTO_TEST_CASE( async_read_at_least ) {
     memset(buf,'x',10);
 
     con->async_read_at_least(5,buf,10);
-    BOOST_CHECK( con->ec == make_error_code(websocketpp::error::test) );
+    BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::error::test) );
 
     std::stringstream channel;
     channel << "abcd";
     channel >> *con;
-    BOOST_CHECK( channel.tellg() == -1 );
-    BOOST_CHECK( con->ec == make_error_code(websocketpp::error::test) );
+    BOOST_CHECK_EQUAL( channel.tellg(), -1 );
+    BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::error::test) );
 
     std::stringstream channel2;
     channel2 << "e";
     channel2 >> *con;
-    BOOST_CHECK( channel2.tellg() == -1 );
+    BOOST_CHECK_EQUAL( channel2.tellg(), -1 );
     BOOST_CHECK( !con->ec );
-    BOOST_CHECK( std::string(buf,10) == "abcdexxxxx" );
+    BOOST_CHECK_EQUAL( std::string(buf,10), "abcdexxxxx" );
 
     std::stringstream channel3;
     channel3 << "f";
     channel3 >> *con;
-    BOOST_CHECK( channel3.tellg() == 0 );
+    BOOST_CHECK_EQUAL( channel3.tellg(), 0 );
     BOOST_CHECK( !con->ec );
-    BOOST_CHECK( std::string(buf,10) == "abcdexxxxx" );
+    BOOST_CHECK_EQUAL( std::string(buf,10), "abcdexxxxx" );
     con->async_read_at_least(1,buf+5,5);
     channel3 >> *con;
-    BOOST_CHECK( channel3.tellg() == -1 );
+    BOOST_CHECK_EQUAL( channel3.tellg(), -1 );
     BOOST_CHECK( !con->ec );
-    BOOST_CHECK( std::string(buf,10) == "abcdefxxxx" );
+    BOOST_CHECK_EQUAL( std::string(buf,10), "abcdefxxxx" );
 }
 
 BOOST_AUTO_TEST_CASE( async_read_at_least2 ) {
@@ -313,20 +417,20 @@ BOOST_AUTO_TEST_CASE( async_read_at_least2 ) {
     memset(buf,'x',10);
 
     con->async_read_at_least(5,buf,5);
-    BOOST_CHECK( con->ec == make_error_code(websocketpp::error::test) );
+    BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::error::test) );
 
     std::stringstream channel;
     channel << "abcdefg";
     channel >> *con;
-    BOOST_CHECK( channel.tellg() == 5 );
+    BOOST_CHECK_EQUAL( channel.tellg(), 5 );
     BOOST_CHECK( !con->ec );
-    BOOST_CHECK( std::string(buf,10) == "abcdexxxxx" );
+    BOOST_CHECK_EQUAL( std::string(buf,10), "abcdexxxxx" );
 
     con->async_read_at_least(1,buf+5,5);
     channel >> *con;
-    BOOST_CHECK( channel.tellg() == -1 );
+    BOOST_CHECK_EQUAL( channel.tellg(), -1 );
     BOOST_CHECK( !con->ec );
-    BOOST_CHECK( std::string(buf,10) == "abcdefgxxx" );
+    BOOST_CHECK_EQUAL( std::string(buf,10), "abcdefgxxx" );
 }
 
 void timer_callback_stub(websocketpp::lib::error_code const &) {}
@@ -346,7 +450,7 @@ BOOST_AUTO_TEST_CASE( async_read_at_least_read_some ) {
     memset(buf,'x',10);
 
     con->async_read_at_least(5,buf,5);
-    BOOST_CHECK( con->ec == make_error_code(websocketpp::error::test) );
+    BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::error::test) );
 
     char input[10] = "abcdefg";
     BOOST_CHECK_EQUAL(con->read_some(input,5), 5);
@@ -370,7 +474,7 @@ BOOST_AUTO_TEST_CASE( async_read_at_least_read_some_indef ) {
     memset(buf,'x',20);
 
     con->async_read_indef(5,buf,5);
-    BOOST_CHECK( con->ec == make_error_code(websocketpp::error::test) );
+    BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::error::test) );
 
     // here we expect to return early from read some because the outstanding
     // read was for 5 bytes and we were called with 10.
@@ -395,7 +499,7 @@ BOOST_AUTO_TEST_CASE( async_read_at_least_read_all ) {
     memset(buf,'x',20);
 
     con->async_read_indef(5,buf,5);
-    BOOST_CHECK( con->ec == make_error_code(websocketpp::error::test) );
+    BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::error::test) );
 
     char input[11] = "aaaaabbbbb";
     BOOST_CHECK_EQUAL(con->read_all(input,10), 10);
@@ -408,7 +512,7 @@ BOOST_AUTO_TEST_CASE( eof_flag ) {
     stub_con::ptr con(new stub_con(true,alogger,elogger));
     char buf[10];
     con->async_read_at_least(5,buf,5);
-    BOOST_CHECK( con->ec == make_error_code(websocketpp::error::test) );
+    BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::error::test) );
     con->eof();
     BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::transport::error::eof) );
 }
@@ -417,14 +521,14 @@ BOOST_AUTO_TEST_CASE( fatal_error_flag ) {
     stub_con::ptr con(new stub_con(true,alogger,elogger));
     char buf[10];
     con->async_read_at_least(5,buf,5);
-    BOOST_CHECK( con->ec == make_error_code(websocketpp::error::test) );
+    BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::error::test) );
     con->fatal_error();
     BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::transport::error::pass_through) );
 }
 
 BOOST_AUTO_TEST_CASE( shutdown ) {
     stub_con::ptr con(new stub_con(true,alogger,elogger));
-    BOOST_CHECK( con->ec == make_error_code(websocketpp::error::test) );
+    BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::error::test) );
     con->shutdown();
     BOOST_CHECK_EQUAL( con->ec, websocketpp::lib::error_code() );
 }
@@ -437,7 +541,7 @@ BOOST_AUTO_TEST_CASE( shutdown_handler ) {
     stub_con::ptr con(new stub_con(true,alogger,elogger));
     
     con->set_shutdown_handler(&sd_handler);
-    BOOST_CHECK( con->ec == make_error_code(websocketpp::error::test) );
+    BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::error::test) );
     con->shutdown();
     BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::transport::error::general) );
 }
@@ -450,7 +554,7 @@ BOOST_AUTO_TEST_CASE( shared_pointer_memory_cleanup ) {
     char buf[10];
     memset(buf,'x',10);
     con->async_read_at_least(5,buf,5);
-    BOOST_CHECK( con->ec == make_error_code(websocketpp::error::test) );
+    BOOST_CHECK_EQUAL( con->ec, make_error_code(websocketpp::error::test) );
     BOOST_CHECK_EQUAL(con.use_count(), 2);
 
     char input[10] = "foo";

websocketpp/common/functional.hpp

@@ -72,6 +72,7 @@ namespace lib {
         /// \todo this feels hacky, is there a better way?
         using ::_1;
         using ::_2;
+        using ::_3;
     }
     
     // See above definition for more details on what this is and why it exists

websocketpp/transport/iostream/base.hpp

@@ -40,10 +40,12 @@ namespace transport {
 /// Transport policy that uses STL iostream for I/O and does not support timers
 namespace iostream {
 
-/// The type and signature of the callback passed to the accept method
-typedef lib::function<void(connection_hdl, char const *, size_t)> write_handler;
+/// The type and signature of the callback used by iostream transport to write
+typedef lib::function<lib::error_code(connection_hdl, char const *, size_t)> 
+    write_handler;
 
-/// The type and signature of the callback passed to the accept method
+/// The type and signature of the callback used by iostream transport to signal 
+/// a transport shutdown.
 typedef lib::function<lib::error_code(connection_hdl)> shutdown_handler;
 
 /// iostream transport errors

websocketpp/transport/iostream/connection.hpp

@@ -313,6 +313,24 @@ public:
         return timer_ptr();
     }
     
+    /// Sets the write handler
+    /**
+     * The write handler is called when the iostream transport receives data
+     * that needs to be written to the appropriate output location. This handler
+     * can be used in place of registering an ostream for output.
+     *
+     * The signature of the handler is 
+     * `lib::error_code (connection_hdl, char const *, size_t)` The
+     * code returned will be reported and logged by the core library.
+     *
+     * @since 0.5.0
+     *
+     * @param h The handler to call on connection shutdown.
+     */
+    void set_write_handler(write_handler h) {
+        m_write_handler = h;
+    }
+    
     /// Sets the shutdown handler
     /**
      * The shutdown handler is called when the iostream transport receives a
@@ -322,7 +340,7 @@ public:
      * If you are using iostream transport with another socket library, this is
      * a good time to close/shutdown the socket for this connection.
      *
-     * The signature of the handler is lib::error_code (connection_hdl). The
+     * The signature of the handler is `lib::error_code (connection_hdl)`. The
      * code returned will be reported and logged by the core library.
      *
      * @since 0.5.0
@@ -400,7 +418,7 @@ protected:
 
     /// Asyncronous Transport Write
     /**
-     * Write len bytes in buf to the output stream. Call handler to report
+     * Write len bytes in buf to the output method. Call handler to report
      * success or failure. handler may or may not be called during async_write,
      * but it must be safe for this to happen.
      *
@@ -408,6 +426,10 @@ protected:
      * output_stream_required: No output stream was registered to write to
      * bad_stream: a ostream pass through error
      *
+     * This method will attempt to write to the registered ostream first. If an
+     * ostream is not registered it will use the write handler. If neither are
+     * registered then an error is passed up to the connection.
+     *
      * @param buf buffer to read bytes from
      * @param len number of bytes to write
      * @param handler Callback to invoke with operation status.
@@ -418,23 +440,26 @@ protected:
         m_alog.write(log::alevel::devel,"iostream_con async_write");
         // TODO: lock transport state?
 
-        if (!m_output_stream) {
-            handler(make_error_code(error::output_stream_required));
-            return;
-        }
+        lib::error_code ec;
 
-        m_output_stream->write(buf,len);
-
-        if (m_output_stream->bad()) {
-            handler(make_error_code(error::bad_stream));
+        if (m_output_stream) {
+            m_output_stream->write(buf,len);
+            
+            if (m_output_stream->bad()) {
+                ec = make_error_code(error::bad_stream);
+            }
+        } else if (m_write_handler) {
+            ec = m_write_handler(m_connection_hdl, buf, len);
         } else {
-            handler(lib::error_code());
+            ec = make_error_code(error::output_stream_required);
         }
+
+        handler(ec);
     }
 
     /// Asyncronous Transport Write (scatter-gather)
     /**
-     * Write a sequence of buffers to the output stream. Call handler to report
+     * Write a sequence of buffers to the output method. Call handler to report
      * success or failure. handler may or may not be called during async_write,
      * but it must be safe for this to happen.
      *
@@ -442,6 +467,10 @@ protected:
      * output_stream_required: No output stream was registered to write to
      * bad_stream: a ostream pass through error
      *
+     * This method will attempt to write to the registered ostream first. If an
+     * ostream is not registered it will use the write handler. If neither are
+     * registered then an error is passed up to the connection.
+     *
      * @param bufs vector of buffers to write
      * @param handler Callback to invoke with operation status.
      */
@@ -451,21 +480,30 @@ protected:
         m_alog.write(log::alevel::devel,"iostream_con async_write buffer list");
         // TODO: lock transport state?
 
-        if (!m_output_stream) {
-            handler(make_error_code(error::output_stream_required));
-            return;
-        }
+        lib::error_code ec;
 
-        std::vector<buffer>::const_iterator it;
-        for (it = bufs.begin(); it != bufs.end(); it++) {
-            m_output_stream->write((*it).buf,(*it).len);
+        if (m_output_stream) {
+            std::vector<buffer>::const_iterator it;
+            for (it = bufs.begin(); it != bufs.end(); it++) {
+                m_output_stream->write((*it).buf,(*it).len);
 
-            if (m_output_stream->bad()) {
-                handler(make_error_code(error::bad_stream));
+                if (m_output_stream->bad()) {
+                    ec = make_error_code(error::bad_stream);
+                    break;
+                }
             }
+        } else if (m_write_handler) {
+            std::vector<buffer>::const_iterator it;
+            for (it = bufs.begin(); it != bufs.end(); it++) {
+                ec = m_write_handler(m_connection_hdl, (*it).buf, (*it).len);
+                if (ec) {break;}
+            }
+            
+        } else {
+            ec = make_error_code(error::output_stream_required);
         }
-
-        handler(lib::error_code());
+        
+        handler(ec);
     }
 
     /// Set Connection Handle

websocketpp/transport/iostream/endpoint.hpp

@@ -117,6 +117,24 @@ public:
         return m_is_secure;
     }
     
+    /// Sets the write handler
+    /**
+     * The write handler is called when the iostream transport receives data
+     * that needs to be written to the appropriate output location. This handler
+     * can be used in place of registering an ostream for output.
+     *
+     * The signature of the handler is 
+     * `lib::error_code (connection_hdl, char const *, size_t)` The
+     * code returned will be reported and logged by the core library.
+     *
+     * @since 0.5.0
+     *
+     * @param h The handler to call on connection shutdown.
+     */
+    void set_write_handler(write_handler h) {
+        m_write_handler = h;
+    }
+    
     /// Sets the shutdown handler
     /**
      * The shutdown handler is called when the iostream transport receives a
@@ -181,11 +199,15 @@ protected:
         if (m_shutdown_handler) {
             tcon->set_shutdown_handler(m_shutdown_handler);
         }
+        if (m_write_handler) {
+            tcon->set_write_handler(m_write_handler);
+        }
         return lib::error_code();
     }
 private:
     std::ostream *  m_output_stream;
     shutdown_handler m_shutdown_handler;
+    write_handler   m_write_handler;
     
     elog_type *     m_elog;
     alog_type *     m_alog;