Adding Overloads for Options on connection::write

deanberris · deanberris · commit 3bad001441be · 2010-10-30T21:54:22.000+08:00
This commit enables the user to call 'write' with a special callback
function. Instead of just passing in a range, the user can then provide
a callback function once the writes are done. This solution allows for
writing handlers that by themselves can chunk up responses to conserve
memory on their own.

An example streaming server using this approach should be doable.

This implementation is largely untested, but builds.

Also, setting the headers will by default send out the headers. In case
any errors are encountered at the time of the sending out of the
headers, that will be saved and then thrown in subsequent calls to
'write'.

This commit also contains a change to the macro that determines the
buffer size for connections, and for the chunk sizes of the linearized
buffers to reflect the correct type.
diff --git a/boost/network/protocol/http/server/async_connection.hpp b/boost/network/protocol/http/server/async_connection.hpp
@@ -10,11 +10,16 @@
 #include <boost/scope_exit.hpp>
 #include <boost/network/protocol/http/algorithms/linearize.hpp>
 #include <boost/network/utils/thread_pool.hpp>
+#include <boost/range/adaptor/sliced.hpp>
 #include <boost/range/algorithm/transform.hpp>
+#include <boost/range/algorithm/copy.hpp>
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/asio/streambuf.hpp>
 #include <boost/asio/strand.hpp>
 #include <boost/asio/buffer.hpp>
+#include <boost/make_shared.hpp>
+#include <list>
+#include <vector>
 #include <iterator>
 
 #ifndef BOOST_NETWORK_HTTP_SERVER_CONNECTION_HEADER_BUFFER_MAX_SIZE
@@ -53,6 +58,8 @@ namespace boost { namespace network { namespace http {
             , service_unavailable = 503
         };
 
+        typedef typename string<Tag>::type string_type;
+
         async_connection(
             asio::io_service & io_service
             , Handler & handler
@@ -76,7 +83,7 @@ namespace boost { namespace network { namespace http {
          *  then sent as soon as the first call to `write` or `flush` commences.
          */
         template <class Range>
-        void set_headers(Range headers) {
+        void set_headers(Range headers, bool immediate = true) {
             if (headers_already_sent)
                 boost::throw_exception(std::logic_error("Headers have already been sent."));
 
@@ -99,6 +106,8 @@ namespace boost { namespace network { namespace http {
                 stream << consts::crlf();
             }
             stream << consts::crlf();
+            if (immediate) write_headers_only();
+
             commit = true;
         }
 
@@ -107,19 +116,34 @@ namespace boost { namespace network { namespace http {
         }
 
         template <class Range>
-        void write(Range) {
-            if (!headers_already_sent) {
-                // TODO write out the headers that are already
-                // linearized to the headers_buffer.
-            }
-            // linearize the range into a shared array
-            // schedule a stranded asynchronous write
+        void write(Range const & range) {
+            write_impl(
+                boost::make_iterator_range(range)
+                , boost::bind(
+                    &async_connection<Tag,Handler>::default_error
+                    , async_connection<Tag,Handler>::shared_from_this()
+                    , _1
+                    )
+                );
+        }
+
+        template <class Range, class Callback>
+        void write(Range const & range, Callback const & callback) {
+            write_impl(
+                boost::make_iterator_range(range)
+                , callback
+                );
         }
 
         asio::ip::tcp::socket & socket()    { return socket_;               }
         utils::thread_pool & thread_pool()  { return thread_pool_;          }
 
     private:
+
+        void default_error(boost::system::error_code const & ec) {
+            // TODO implement a sane default here, for now ignore the error
+        }
+
         asio::ip::tcp::socket socket_;
         asio::io_service::strand strand;
         Handler & handler;
@@ -129,6 +153,11 @@ namespace boost { namespace network { namespace http {
 
         typedef boost::array<char, BOOST_NETWORK_HTTP_SERVER_CONNECTION_BUFFER_SIZE> 
             buffer_type;
+        typedef boost::array<char, BOOST_NETWORK_HTTP_SERVER_CONNECTION_BUFFER_SIZE> 
+            array;
+        typedef std::list<shared_ptr<array> > array_list;
+        typedef boost::shared_ptr<array_list> shared_array_list;
+        typedef boost::shared_ptr<std::vector<asio::const_buffer> > shared_buffers;
         buffer_type read_buffer_;
         boost::uint16_t status;
 
@@ -157,6 +186,106 @@ namespace boost { namespace network { namespace http {
             // FIXME -- damn all that work got wiped out because Jeni tripped on the power. :(
         }
 
+        template <class Range, class Callback>
+        void write_headers(Range range, Callback callback) {
+            // TODO send out the headers, then once that's done 
+            // call the write again on the range and callback
+        }
+
+        void write_headers_only() {
+        }
+
+        void handle_write_headers(boost::system::error_code const & ec) {
+            if (ec) {
+                // TODO signal somehow that there was an error so that subsequent
+                // calls to write would throw an exception
+                return;
+            }
+            headers_already_sent = true;
+        }
+
+        void handle_write(
+            boost::function<void(boost::system::error_code const &)> callback
+            , shared_array_list temporaries
+            , shared_buffers buffers
+            , boost::system::error_code const & ec
+        ) {
+            // we want to forget the temporaries and buffers
+            thread_pool().post(boost::bind(callback, ec));
+        }
+
+        template <class Range, class Callback>
+        void write_impl(Range range, Callback callback) {
+            if (!headers_already_sent) {
+                write_headers(range, callback);
+                return;
+            }
+
+            // linearize the whole range into a vector
+            // of fixed-sized buffers, then schedule an asynchronous
+            // write of these buffers -- make sure they are live
+            // by making these linearized buffers shared and made
+            // part of the completion handler.
+            //
+            // once the range has been linearized and sent, schedule
+            // a wrapper to be called in the io_service's thread, that
+            // will re-schedule the given callback into the thread pool
+            // referred to here so that the io_service's thread can concentrate
+            // on doing I/O.
+            //
+
+            shared_array_list temporaries = 
+                boost::make_shared<array_list>();
+            shared_buffers buffers = 
+                boost::make_shared<std::vector<asio::const_buffer> >(0);
+
+            std::size_t range_size = boost::distance(range);
+            buffers->resize(
+                (range_size / BOOST_NETWORK_HTTP_SERVER_CONNECTION_BUFFER_SIZE)
+                + (range_size % BOOST_NETWORK_HTTP_SERVER_CONNECTION_BUFFER_SIZE)
+                );
+            std::size_t slice_size = 
+                std::min(
+                    range_size,
+                    BOOST_NETWORK_HTTP_SERVER_CONNECTION_BUFFER_SIZE
+                    );
+            typename boost::range_iterator<Range>::type
+                start = boost::begin(range)
+                , end  = boost::end(range);
+            while (slice_size != 0) {
+                using boost::adaptors::sliced;
+                shared_ptr<array> new_array = make_shared<array>();
+                boost::copy(
+                    range | sliced(0,slice_size)
+                    , new_array->begin()
+                    );
+                temporaries->push_back(new_array);
+                buffers->push_back(asio::buffer(new_array->data(), slice_size));
+                std::advance(start, slice_size);
+                range = boost::make_iterator_range(start, end);
+                range_size = boost::distance(range);
+                slice_size = std::min(range_size, BOOST_NETWORK_HTTP_SERVER_CONNECTION_BUFFER_SIZE);
+            }
+
+            if (!buffers->empty()) {
+                boost::function<void(boost::system::error_code const &)> f = callback;
+                asio::async_write(
+                    socket_
+                    , *buffers
+                    , strand.wrap(
+                        boost::bind(
+                            &async_connection<Tag,Handler>::handle_write
+                            , async_connection<Tag,Handler>::shared_from_this()
+                            , f
+                            , temporaries
+                            , buffers // keep these alive until the handler is called!
+                            , boost::asio::placeholders::error
+                            )
+                        )
+                    );
+            }
+        }
+
     };
     
 } /* http */
diff --git a/boost/network/protocol/http/server/sync_connection.hpp b/boost/network/protocol/http/server/sync_connection.hpp
@@ -9,7 +9,7 @@
 #define BOOST_NETWORK_HTTP_SERVER_SYNC_CONNECTION_HPP_
 
 #ifndef BOOST_NETWORK_HTTP_SERVER_CONNECTION_BUFFER_SIZE
-#define BOOST_NETWORK_HTTP_SERVER_CONNECTION_BUFFER_SIZE 1024
+#define BOOST_NETWORK_HTTP_SERVER_CONNECTION_BUFFER_SIZE 1024uL
 #endif
 
 #include <boost/enable_shared_from_this.hpp>
