HTTP handling and AMQP fixes

* Post AMQP events via strand to reduce threading issues
* HTTP updates including stall timeout and connection recycling

CMakeLists.txt

@@ -2,16 +2,19 @@ cmake_minimum_required(VERSION 2.8)
 set(CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR})
 
 PROJECT(asio-protocols)
+set (VERSION_MAJOR 1)
+set (VERSION_MINOR 0)
+set (VERSION_PATCH 4)
+set (PROJECT_VERSION "${VERSION_MAJOR}.${VERSION_MINOR}.${VERSION_PATCH}")
+set (PROJECT_VERSION_REVISION 1)
+
 option(USE_CLANG "build application with clang" OFF)
 if(USE_CLANG)
 	set(CMAKE_CXX_COMPILER "/usr/bin/clang++-3.6")
 endif()
+include(set_cxx_norm.cmake)
+set_cxx_norm(${CXX_NORM_CXX11})
 enable_testing()
-set (VERSION_MAJOR 1)
-set (VERSION_MINOR 0)
-set (VERSION_PATCH 3)
-set (PROJECT_VERSION "${VERSION_MAJOR}.${VERSION_MINOR}.${VERSION_PATCH}")
-set (PROJECT_VERSION_REVISION 2)
 
 set(Protocols_LIBRARY asio-protocols)
 
@@ -28,74 +31,18 @@ find_package(Future REQUIRED)
 find_package(AMQP REQUIRED)
 include_directories(${AMQP_INCLUDE_DIR})
 
+add_subdirectory(test)
+
 add_library(
 	${Protocols_LIBRARY}
 	src/buffer.cpp
 	src/source.cpp
 	src/sink.cpp
 	src/stream.cpp
 	src/amqp.cpp
-	src/http.cpp
-)
-
-add_executable(
-	stream_tests
-	tests/stream.cpp
+	#	src/http.cpp
 )
 
-target_link_libraries(
-	stream_tests
-	${Protocols_LIBRARY}
-	${Boost_LIBRARIES}
-	${OPENSSL_LIBRARIES}
-	${OPENSSL_CRYPTO_LIBRARY}
-	${OPENSSL_SSL_LIBRARY}
-	z
-)
-if(WIN32)
-	target_link_libraries(stream_tests wsock32 ws2_32)
-endif()
-set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
-
-add_executable(
-	amqp_tests
-	src/amqp_tests.cpp
-)
-
-target_link_libraries(
-	amqp_tests
-	${Protocols_LIBRARY}
-	${AMQP_LIBRARY}
-	${Boost_LIBRARIES}
-	${OPENSSL_LIBRARIES}
-	${OPENSSL_CRYPTO_LIBRARY}
-	${OPENSSL_SSL_LIBRARY}
-	z
-)
-if(WIN32)
-	target_link_libraries(amqp_tests wsock32 ws2_32)
-endif()
-set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
-
-add_executable(
-	http_tests
-	tests/http.cpp
-)
-
-target_link_libraries(
-	http_tests
-	${Protocols_LIBRARY}
-	${Boost_LIBRARIES}
-	${OPENSSL_LIBRARIES}
-	${OPENSSL_CRYPTO_LIBRARY}
-	${OPENSSL_SSL_LIBRARY}
-	z
-)
-if(WIN32)
-	target_link_libraries(http_tests wsock32 ws2_32)
-endif()
-set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
-
 install(
 	DIRECTORY include/
 	DESTINATION "include"
@@ -107,10 +54,6 @@ install(
 
 include (CTest)
 
-add_test (amqp_tests amqp_tests --log_level=all --log_format=XML --log_sink=amqp_tests.xml)
-add_test (http_tests http_tests --log_level=all --log_format=XML --log_sink=http_tests.xml)
-add_test (stream_tests stream_tests --log_level=all --log_format=XML --log_sink=stream_tests.xml)
-
 include (InstallRequiredSystemLibraries)
 set (CPACK_PACKAGE_NAME "libasio-protocols")
 set (CPACK_PACKAGE_VENDOR "")

docs/notes.md

@@ -0,0 +1,9 @@
+
+
+    line_stream("\n"), "\x0D\x0A"
+
+line_stream takes a char_stream. It collects characters until the linebreak, at which point it emits those characters.
+
+    auto ls = line_stream(cs);
+
+

docs/statsd.md

@@ -0,0 +1,19 @@
+# Protocol format
+
+Latest information seems to be [https://github.com/b/statsd_spec](here).
+
+This is a line-based protocol.
+
+    key:value|type
+
+Types are:
+
+* g - gauge, instantaneous value
+* c - counter, inc/dec, also supports rate (|@rate)
+* ms - timer
+* h - histogram, alias for timer
+* m - meter, rate
+
+A key with no other information should be treated as a meter with value of 1.
+
+

example/tcp-client.cpp

@@ -0,0 +1,23 @@
+void
+example()
+{
+	auto client = net::asio::tcp::client();
+	client->connect(
+		"localhost",
+		80
+	)->then(
+		[](std::shared_ptr<stream> s) {
+			s->sink([](const std::string &in) {
+				std::cout << in;
+			});
+			return s->write(
+				"GET / HTTP/1.1\x0D\x0A\x0D\x0A"
+			)->then(
+				[s](size_t written) {
+					return s->remote_eof();
+				}
+			);
+		}
+	);
+}
+

ht.cpp

@@ -0,0 +1,45 @@
+#include <net/asio/http.h>
+
+int
+main(int argc, const char *argv[])
+{
+	boost::asio::io_service service { };
+	auto client_ = std::make_shared<net::http::client>(service);
+	bool show_request = false;
+	bool show_headers = false;
+	for(int i = 1; i < argc; ++i) {
+		auto v = std::string { argv[i] };
+		if(v == "--headers") {
+			show_headers = true;
+		} else if(v == "--request") {
+			show_request = true;
+		} else {
+			auto req = net::http::request {
+				net::http::uri {
+					argv[i]
+				}
+			};
+			req << net::http::header("User-agent", "some-user-agent");
+			auto res = client_->GET(
+				std::move(req)
+			);
+			res->completion()->on_done([i, show_request, show_headers](const std::shared_ptr<net::http::response> &res) {
+				if(show_request) {
+					std::cout << res->request().bytes() << "\n";
+				}
+				if(show_headers) {
+					res->each_header([](const net::http::header &h) {
+						std::cout << h.key() << ": " << h.value() << "\n";
+					});
+					std::cout << "\n";
+				}
+				std::cout << res->body();
+			});
+		}
+	}
+
+	// std::cout << "run\n";
+	service.run();
+	// std::cout << "done\n";
+}
+

ht.md

@@ -0,0 +1,39 @@
+# HTTP
+
+Some basic implementation details:
+
+## GET /path => empty
+
+    return client_.get(uri)
+        ->expect_status(200, 201, 202, 204)
+        ->ignore_body()
+        ->completion();
+
+## GET /path => binary
+
+    return client_.get(uri)
+        ->expect_status(200, 201, 202, 204)
+        ->stream_body([ca](const std::string &in) {
+            return ca->append_body(in);
+        })
+        ->completion();
+
+## GET /path => JSON
+
+    return client_.get(uri)
+        ->expect_status(200, 201, 202, 204)
+        ->expect_content_type("application/javascript")
+        ->completion();
+    json::parse(resp->body());
+
+## JSON => POST /path => JSON
+
+    return client_.get(uri)
+        ->expect_status(200, 201, 202, 204)
+        ->expect_content_type("application/javascript")
+        ->body(json_in)
+        ->completion();
+    json::parse(resp->body());
+
+## PB => POST /path
+