Progress in implementation of upload filter

cppcms/application.h

@@ -274,6 +274,17 @@ namespace cppcms {
 		void assign_context(booster::shared_ptr<http::context> conn);
 
 		///
+		/// Add context to applications such that context ownership isn't transferred
+		/// to the application
+		///
+		void add_context(http::context &conn);
+
+		///
+		/// Remove context added with add_context
+		///
+		void remove_context();
+
+		///
 		/// Returns true if current application was created as asynchronous application.
 		///
 		bool is_asynchronous();

cppcms/applications_pool.h

@@ -51,6 +51,7 @@ namespace cppcms {
 
 		static const int thread_specific= 0x0010; ///< Make synchronous application thread specific
 		static const int prepopulated	= 0x0020; ///< Make sure all applications are created from the beginning (ignored in thread_specific is set)
+		static const int content_filter = 0x0040; ///< Make this asynchronous application to handle content
 		/// \cond INTERNAL
 		static const int legacy		= 0x8000; ///< Use legacy handling of application life time when the application is created in the event loop and than dispatched as a job to a thread pool
 		/// \endcond

cppcms/http_context.h

@@ -33,6 +33,7 @@ namespace cppcms {
 	namespace http {
 		class request;
 		class response;
+		class file;
 
 		///
 		/// \brief context is a central class that holds all specific connection related information.
@@ -190,6 +191,12 @@ namespace cppcms {
 			///
 			void submit_to_asynchronous_application(booster::intrusive_ptr<application> app,std::string const &matched_url);
 		private:
+			friend class impl::cgi::connection;
+			bool has_file_filter();
+			int send_to_file_filter(file &f,int stage);
+			int on_headers_ready(bool has_content);
+			int translate_exception();
+			void make_error_message(std::exception const &e);
 			void on_request_ready(bool error);
 			void submit_to_pool_internal(booster::shared_ptr<application_specific_pool> pool,std::string const &matched,bool now);
 			static void dispatch(booster::shared_ptr<application_specific_pool> const &pool,booster::shared_ptr<context> const &self,std::string const &url);

cppcms/http_request.h

@@ -26,6 +26,8 @@ namespace http {
 	class cookie;
 	class file;
 	class content_limits;
+	class content_filter;
+	class basic_content_filter;
 
 	///
 	/// \brief This class represents all information related to the HTTP/CGI request.
@@ -295,13 +297,29 @@ namespace http {
 		///  Get content limits for incoming data processing
 		///
 		content_limits &limits();
+
+		///
+		/// Get installed content filter
+		///
+		basic_content_filter *content_filter();
+		
+		///
+		/// Install content filter, setting it to 0 would remove the filter
+		///
+		void content_filter(basic_content_filter *flt);
+
+		///
+		/// Returns true when full request content is ready
+		///
+		bool is_ready();
 	public:
 		/// \cond INTERNAL
 		request(impl::cgi::connection &);
 		~request();
 		/// \endcond
 	private:
-
+		void set_ready();
+		friend class context;
 		friend class impl::cgi::connection;
 
 		void set_post_data(std::vector<char> &post_data);

src/application.cpp

@@ -31,11 +31,13 @@ namespace cppcms {
 
 struct application::_data {
 	_data(cppcms::service *s):
-		service(s)
+		service(s),
+		temp_conn(0)
 	{
 	}
 	cppcms::service *service;
 	booster::shared_ptr<http::context> conn;
+	http::context *temp_conn;
 	url_dispatcher url;
 	booster::hold_ptr<url_mapper> url_map;
 	std::vector<application *> managed_children;
@@ -92,10 +94,25 @@ booster::shared_ptr<http::context> application::get_context()
 	return root()->d->conn;
 }
 
+void application::add_context(http::context &conn)
+{
+	if(root()->d->conn)
+		throw cppcms_error("Context already assigned");
+	root()->d->temp_conn = &conn;
+}
+
+void application::remove_context()
+{
+	root()->d->temp_conn = 0;
+}
+
 http::context &application::context()
 {
-	if(!root()->d->conn)
+	if(!root()->d->conn) {
+		if(root()->d->temp_conn)
+			return *root()->d->temp_conn;
 		throw cppcms_error("Access to unassigned context");
+	}
 	return *root()->d->conn;
 }
 
@@ -118,6 +135,7 @@ bool application::is_asynchronous()
 void application::assign_context(booster::shared_ptr<http::context> conn)
 {
 	root()->d->conn=conn;
+	root()->d->temp_conn = 0;
 }
 
 void application::set_pool(booster::weak_ptr<application_specific_pool> p)

src/cgi_api.cpp

@@ -307,6 +307,12 @@ void connection::load_content(booster::system::error_code const &e,http::context
 		return;
 	}
 	
+	int status = context->on_headers_ready(content_length > 0);
+	if(status != 0) {
+		handle_http_error(status,context,h);
+		return;
+	}
+
 	if(content_length > 0) {
 		if(content_type.is_multipart_form_data()) {
 			// 64 MB
@@ -364,8 +370,30 @@ void connection::on_some_multipart_read(booster::system::error_code const &e,siz
 	char const *end = begin + n;
 	multipart_parser::parsing_result_type r = multipart_parser::continue_input;
 	long long allowed=context->request().limits().content_length_limit();
+	bool has_filter = context->has_file_filter();
 	while(begin!=end) {
 		r = multipart_parser_->consume(begin,end);
+		if(has_filter) {
+			int status;
+			switch(r) { 
+			case multipart_parser::meta_ready:
+				status = context->send_to_file_filter(multipart_parser_->get_file(),0);
+				break;
+			case multipart_parser::content_partial:
+				status = context->send_to_file_filter(multipart_parser_->get_file(),1);
+				break;
+			case multipart_parser::content_ready:
+				status = context->send_to_file_filter(multipart_parser_->last_file(),2);
+				break;
+			default:
+				status = 0;
+			}
+			if(status != 0) {
+				handle_http_error(status,context,h);
+				return;
+			}
+		}
+
 		if(r==multipart_parser::content_ready || r==multipart_parser::content_partial) {
 			http::file &f= 	(r == multipart_parser::content_ready)
 					?	multipart_parser_->last_file() 

src/http_context.cpp

@@ -21,6 +21,7 @@
 #include <booster/log.h>
 #include <booster/backtrace.h>
 #include <booster/aio/io_service.h>
+#include <cppcms/http_content_filter.h>
 
 #include "cached_settings.h"
 
@@ -38,6 +39,9 @@ namespace http {
 		std::auto_ptr<http::response> response;
 		std::auto_ptr<cache_interface> cache;
 		std::auto_ptr<session_interface> session;
+		booster::shared_ptr<application_specific_pool> pool;
+		booster::intrusive_ptr<application> app;
+		std::string matched;
 		_data(context &cntx) :
 			locale(cntx.connection().service().locale()),
 			request(cntx.connection())
@@ -120,6 +124,20 @@ namespace {
 		}
 
 	};
+
+	class context_guard {
+	public:
+		context_guard(cppcms::application &app,cppcms::http::context &ctx) : app_(&app)
+		{
+			app_->add_context(ctx);
+		}
+		~context_guard() 
+		{
+			app_->remove_context();
+		}
+	private:	
+		cppcms::application *app_;
+	};
 }
 
 
@@ -167,10 +185,27 @@ void context::submit_to_pool_internal(booster::shared_ptr<application_specific_p
 	}
 }
 
-void context::on_request_ready(bool error)
+int context::translate_exception()
 {
-	if(error) return;
+	try {
+		throw;
+	}
+	catch(abort_upload const &e) {
+		return e.code();
+	}
+	catch(std::exception const &e) {
+		make_error_message(e);
+		return 500;
+	}
+	catch(...) {
+		BOOSTER_ERROR("cppcms") << "Unknown exception";
+		return 500;
+	}
+	return 0;
+}
 
+int context::on_headers_ready(bool has_content)
+{
 	char const *host = conn_->cgetenv("HTTP_HOST");
 	char const *path_info = conn_->cgetenv("PATH_INFO");
 	char const *script_name = conn_->cgetenv("SCRIPT_NAME");
@@ -183,15 +218,94 @@ void context::on_request_ready(bool error)
 			path_info,
 			matched
 			);
+	if(!pool)
+		return 404;
+
+	int flags;
+	if(!has_content || ((flags=pool->flags()) & app::op_mode_mask) == app::synchronous || (flags & app::content_filter)==0) {
+		d->pool.swap(pool);
+		d->matched.swap(matched);
+		return 0;
+	}
+
+	booster::intrusive_ptr<application> app = d->pool->get(service());
+	if(!app)
+		return 500;
+
+	try {
+		context_guard g(*app,*this);
+		app->main(matched);
+	}
+	catch(...) {
+		return translate_exception();
+	}
+	d->pool.swap(pool);
+	d->app.swap(app);
+	d->matched.swap(matched);
+	return 0;
+}
+
+bool context::has_file_filter()
+{
+	return dynamic_cast<multipart_filter *>(request().content_filter())!=0;
+}
+int context::send_to_file_filter(file &f,int stage)
+{
+	try {
+		context_guard g(*d->app,*this);
+		multipart_filter *filter=static_cast<multipart_filter *>(request().content_filter());
+		switch(stage) {
+		case 0: filter->on_new_file(f); break;
+		case 1: filter->on_upload_progress(f); break;
+		case 2: filter->on_data_ready(f); break;
+		}
+	}
+	catch(...) {
+		return translate_exception();
+	}
+	return 0;
+}
 
-	if(!pool) {
-		response().io_mode(http::response::asynchronous);
-		response().make_error_response(http::response::not_found);
-		async_complete_response();
+void context::on_request_ready(bool error)
+{
+	booster::shared_ptr<application_specific_pool> pool;
+	booster::intrusive_ptr<application> app;
+	pool.swap(d->pool);
+	app.swap(d->app);
+	basic_content_filter *filter = 0;
+
+	if(error && app && (filter=request().content_filter())!=0) {
+		context_guard g(*app,*this);
+		try {
+			filter->on_error();
+		}
+		catch(...) {}
 		return;
 	}
 
-	submit_to_pool_internal(pool,matched,true);
+	if(error)
+		return;
+
+	request().set_ready();
+
+	if(app && filter) {
+		context_guard g(*app,*this);
+		try {
+			filter->on_end_of_content();
+		}
+		catch(...) {
+			translate_exception();
+			return;
+		}
+	}
+
+	if(app) {
+		app->assign_context(self());
+		dispatch(app,d->matched,false);
+		return;
+	}
+
+	submit_to_pool_internal(pool,d->matched,true);
 }
 
 namespace {
@@ -226,6 +340,21 @@ void context::dispatch(booster::shared_ptr<application_specific_pool> const &poo
 	app->assign_context(self);
 	dispatch(app,url,true);
 }
+
+void context::make_error_message(std::exception const &e)
+{
+	BOOSTER_ERROR("cppcms") << "Caught exception ["<<e.what()<<"]\n" << booster::trace(e)  ;
+	if(!response().some_output_was_written()) {
+		if(service().cached_settings().security.display_error_message) {
+			std::ostringstream ss;
+			ss << e.what() << '\n';
+			ss << booster::trace(e);
+			response().make_error_response(http::response::internal_server_error,ss.str());
+		}
+		else
+			response().make_error_response(http::response::internal_server_error);
+	}
+}
 // static 
 void context::dispatch(booster::intrusive_ptr<application> const &app,std::string const &url,bool syncronous)
 {
@@ -248,20 +377,9 @@ void context::dispatch(booster::intrusive_ptr<application> const &app,std::strin
 			app->response().make_error_response(http::response::forbidden);
 		}
 	}
-	catch(std::exception const &e){
-		BOOSTER_ERROR("cppcms") << "Caught exception ["<<e.what()<<"]\n" << booster::trace(e)  ;
-		if(app->get_context()) {
-			if(!app->response().some_output_was_written()) {
-				if(app->service().cached_settings().security.display_error_message) {
-					std::ostringstream ss;
-					ss << e.what() << '\n';
-					ss << booster::trace(e);
-					app->response().make_error_response(http::response::internal_server_error,ss.str());
-				}
-				else
-					app->response().make_error_response(http::response::internal_server_error);
-			}
-		}
+	catch(...){
+		if(app->get_context())
+			app->context().translate_exception();
 	}
 	
 	if(app->get_context()) {

src/http_request.cpp

@@ -129,9 +129,30 @@ bool request::parse_cookies()
 struct request::_data {
 	std::vector<char> post_data;
 	content_limits limits;
-	_data(cppcms::service &srv) : limits(srv.cached_settings()) {}
+	basic_content_filter *filter;
+	bool ready;
+	_data(cppcms::service &srv) : limits(srv.cached_settings()),filter(0),ready(false) {}
 };
 
+basic_content_filter *request::content_filter()
+{
+	return d->filter;
+}
+
+void request::content_filter(basic_content_filter *filter)
+{
+	d->filter = filter;
+}
+
+bool request::is_ready()
+{
+	return d->ready;
+}
+void request::set_ready()
+{
+	d->ready = true;
+}
+
 void request::set_post_data(std::vector<char> &post_data)
 {
 	d->post_data.clear();

src/http_response.cpp

@@ -526,7 +526,7 @@ response::response(context &context) :
 	d(new _data(&context.connection())),
 	context_(context),
 	stream_(0),
-	io_mode_(normal),
+	io_mode_(asynchronous),
 	disable_compression_(0),
 	ostream_requested_(0),
 	copy_to_cache_(0),

tests/cache_frontend_test.cpp

@@ -54,6 +54,7 @@ public:
 		booster::shared_ptr<dummy_api> api(new dummy_api(srv_,env,output_));
 		booster::shared_ptr<cppcms::http::context> cnt(new cppcms::http::context(api));
 		assign_context(cnt);
+		response().io_mode(cppcms::http::response::normal);
 		output_.clear();
 	}
 

tests/http_timeouts_test.py

@@ -76,9 +76,9 @@ if write:
 else:
     print 'Read from client timeouts'
     test_unfinished_out('')
-    test_unfinished_out('GET /')
-    test_unfinished_out('POST / HTTP/1.0\r\nContent-Length:10000\r\n\r\nbla bla')
-    test_unfinished_out('POST / HTTP/1.0\r\nContent-Length:10000\r\n\r\n', ['ss','ss','ss','ss'])
+    test_unfinished_out('GET /sync/long')
+    test_unfinished_out('POST /sync/long HTTP/1.0\r\nContent-Length:10000\r\n\r\nbla bla')
+    test_unfinished_out('POST /sync/long HTTP/1.0\r\nContent-Length:10000\r\n\r\n', ['ss','ss','ss','ss'])
 
     print 'Disconnect the client timeout'
 

tests/response_test.cpp

@@ -90,6 +90,7 @@ public:
 		booster::shared_ptr<dummy_api> api(new dummy_api(srv_,env,output_,mark_chunks,mark_eof));
 		booster::shared_ptr<cppcms::http::context> cnt(new cppcms::http::context(api));
 		assign_context(cnt);
+		response().io_mode(cppcms::http::response::normal);
 		output_.clear();
 	}
 

tests/tc_test.cpp

@@ -98,6 +98,7 @@ public:
 		booster::shared_ptr<dummy_api> api(new dummy_api(srv_,env,output_));
 		booster::shared_ptr<cppcms::http::context> cnt(new cppcms::http::context(api));
 		assign_context(cnt);
+		response().io_mode(cppcms::http::response::normal);
 		response().out();
 		output_.clear();
 	}