Merge pull request #1006 from Barenboim/master

Add WFRepeaterTask

docs/en/tutorial-09-http_file_server.md

@@ -7,7 +7,7 @@
 # About http\_file\_server
 
 http\_file\_server is a web server. You can start a web server after specifying the startup port and the root path (the default setting is the current path).   
-You can also specify a certificate file and a key file in PEM format to start an HTTPS web server.   
+You can also specify a certificate file and a key file in PEM format to start an HTTPS web server. User may access the server through command line, the request will be sent to IP address 127.0.0.1.  
 The program mainly demonstrates how to use disk IO tasks. In the Linux system, we use the aio interface in the kernel of Linux, and the file reading is completely asynchronous.
 
 # Starting a server
@@ -149,6 +149,50 @@ The memory of the buf domain is managed by ourselves. You can use  **append\_out
 After the reply is completed, we will **free()** this block of memory with this line in the process:   
 server\_task->set\_callback(\[](WFHttpTask \*t){ free(t->user\_data); });
 
+# Interact with the server through command line
+
+After the server is started, users may access it through command line. Simply input the file name that you want to get, or input Ctrl-D to end the program. The repeating process is implemnted by using WFRepeaterTask, which can be created by this factory function:
+~~~cpp
+using repeated_create_t = std::function<SubTask *(WFRepeaterTask *)>;
+using repeater_callback_t = std::function<void (WFRpeaterTask *)>;
+
+class WFTaskFactory
+{
+    WFRpeaterTask *create_repeater_task(repeated_create_t create, repeater_callback_t callback);
+};
+~~~
+As above, a repeater task is created with a task creator function. The repeater calls the task creator repeatedly and run the task until the creator return a NULL pointer. When the using's input is not empty, our creator will create an HTTP task on IP 127.0.0.1 to access the server.  
+~~~cpp
+{
+	auto&& create = [&scheme, port](WFRepeaterTask *)->SubTask *{
+		...
+		scanf("%1023s", buf);
+		if (*buf == '\0')
+			return NULL;
+
+		std::string url = scheme + "127.0.0.1:" + std::to_string(port) + "/" + buf;
+		WFHttpTask *task = WFTaskFactory::create_http_task(url, 0, 0,
+									[](WFHttpTask *task) {
+			...
+		});
+
+		return task;
+	};
+	
+	WFFacilities::WaitGroup wg(1);
+	WFRepeaterTask *repeater;
+	repeater = WFTaskFactory::create_repeater_task(create, [&wg](WFRepeaterTask *) {
+		wg.done();
+	});
+
+	repeater->start();
+	wg.wait();
+
+	server.stop();
+}
+~~~
+Finally, when the creator returned NULL, the repeater's callback is called and the program will be ended.
+
 # About the implementation of the file IO
 
 Linux operating system supports a set of asynchronous IO system calls with high efficiency and very little CPU occupation. If you use our framework in a Linux system, this set of interfaces are used by default.   

docs/en/tutorial-10-user_defined_protocol.md

@@ -202,8 +202,7 @@ Please see [server.cc](/tutorial/tutorial-10-user_defined_protocol/server.cc) fo
 
 # client
 
-The logic of the client is to receive the user input from standard IO, construct a request, send it to the server and get the results.   
-For simplicity, the process of reading standard input is completed in the callback, so we will send an empty request first. Also, for the sake of security, we limit the packet size of the server reply to 4KB.   
+The logic of the client is to receive the user input from standard IO, construct a request, send it to the server and get the results. Here we use WFRepeaterTask to implement the repeating process, terminates if the user's input is empty. For the sake of security, we limit the packet size of the server reply to 4KB.   
 The only thing that a client needs to know is how to generate a client task on a user-defined protocol. There are three interface options in [WFTaskFactory.h](/src/factory/WFTaskFactory.h):
 
 ~~~cpp

docs/tutorial-09-http_file_server.md

@@ -7,6 +7,7 @@
 
 http_file_server是一个web服务器，用户指定启动端口，根路径（默认为程序当路程），就可以启动一个web server。  
 用户还可以指定一个PEM格式的certificate file和key file，启动一个https web server。  
+程序在启动server之后，可以从命令行接受用户输入，并通过127.0.0.1地址来访问这个server。  
 程序主要展示了磁盘IO任务的用法。在Linux系统下，我们利用了Linux底层的aio接口，文件读取完全异步。
 
 # 启动server
@@ -141,6 +142,53 @@ buf域的内存我们是自己管理的，可以通过append_output_body_nocopy(
 在回复完成后，我们会free()这块内存，这个语句在process里：  
 server_task->set_callback([](WFHttpTask *t){ free(t->user_data); });
 
+# 命令行交互
+
+启动server后，用户可以在控制台输入文件名来访问server。当输入文件名为空（Ctrl-D），关闭server并结束程序。  
+这里，我们使用了WFRepeaterTask来实现这个循环接受输入的过程。WFRepeaterTask是一种循环任务，产生的接口如下：
+~~~cpp
+using repeated_create_t = std::function<SubTask *(WFRepeaterTask *)>;
+using repeater_callback_t = std::function<void (WFRpeaterTask *)>;
+
+class WFTaskFactory
+{
+    WFRpeaterTask *create_repeater_task(repeated_create_t create, repeater_callback_t callback);
+};
+~~~
+通过create函数，可以创建一个repeater任务。repeater内部会反复调用create，产生一个任务并运行，直到create返回空指针。  
+在我们的这个示例里，create函数内部调用scanf。当用户输入为空时，create返回NULL，整个循环过程结束。  
+当用户输入不为空（文件名），产生一个访问127.0.0.1地址的http任务来访问我们开启的server。
+~~~cpp
+{
+	auto&& create = [&scheme, port](WFRepeaterTask *)->SubTask *{
+		...
+		scanf("%1023s", buf);
+		if (*buf == '\0')
+			return NULL;
+
+		std::string url = scheme + "127.0.0.1:" + std::to_string(port) + "/" + buf;
+		WFHttpTask *task = WFTaskFactory::create_http_task(url, 0, 0,
+									[](WFHttpTask *task) {
+			...
+		});
+
+		return task;
+	};
+	
+	WFFacilities::WaitGroup wg(1);
+	WFRepeaterTask *repeater;
+	repeater = WFTaskFactory::create_repeater_task(create, [&wg](WFRepeaterTask *) {
+		wg.done();
+	});
+
+	repeater->start();
+	wg.wait();
+
+	server.stop();
+}
+~~~
+最后，当create返回NULL，repeater被callback。我们关闭server并结束程序。  
+
 # 关于文件异步IO的实现
 
 Linux操作系统支持一套效率很高，CPU占用非常少的异步IO系统调用。在Linux系统下使用我们的框架将默认使用这套接口。  

docs/tutorial-10-user_defined_protocol.md

@@ -191,8 +191,8 @@ server与普通的http server没有什么区别。我们优先IPv6启动，这
 
 # client端
 
-client端的逻辑是从标准IO接收用户输入，构造出请求发往server并得到结果。  
-为了简单，读取标准输入的过程都在callback里完成，因此我们会先发出一条空请求。同样为了安全我们限制server回复包不超4KB。  
+client端的逻辑是从标准IO接收用户输入，构造出请求发往server并得到结果。这里我们使用了WFRepeaterTask来实现这个重复过程，直到用户的输入为空。
+此外，为了安全我们限制server回复包不超4KB。  
 client端唯一需要了解的就是怎么产生一个自定义协议的client任务，在[WFTaskFactory.h](../src/factory/WFTaskFactory.h)有三个接口可以选择：
 ~~~cpp
 template<class REQ, class RESP>

src/factory/WFTask.h

@@ -45,7 +45,7 @@ enum
 	WFT_STATE_SSL_ERROR = 65,
 	WFT_STATE_DNS_ERROR = 66,					/* for client task only */
 	WFT_STATE_TASK_ERROR = 67,
-	WFT_STATE_ABORTED = CS_STATE_STOPPED		/* main process terminated */
+	WFT_STATE_ABORTED = CS_STATE_STOPPED
 };
 
 template<class INPUT, class OUTPUT>
@@ -721,6 +721,66 @@ protected:
 	virtual ~WFGoTask() { }
 };
 
+class WFRepeaterTask : public WFGenericTask
+{
+public:
+	void set_create(std::function<SubTask *(WFRepeaterTask *)> create)
+	{
+		this->create = std::move(create);
+	}
+
+	void set_callback(std::function<void (WFRepeaterTask *)> callback)
+	{
+		this->callback = std::move(callback);
+	}
+
+protected:
+	virtual void dispatch()
+	{
+		SubTask *task = this->create(this);
+
+		if (task)
+		{
+			series_of(this)->push_front(this);
+			series_of(this)->push_front(task);
+		}
+		else
+			this->state = WFT_STATE_SUCCESS;
+
+		this->subtask_done();
+	}
+
+	virtual SubTask *done()
+	{
+		SeriesWork *series = series_of(this);
+
+		if (this->state != WFT_STATE_UNDEFINED)
+		{
+			if (this->callback)
+				this->callback(this);
+
+			delete this;
+		}
+
+		return series->pop();
+	}
+
+protected:
+	std::function<SubTask *(WFRepeaterTask *)> create;
+	std::function<void (WFRepeaterTask *)> callback;
+
+public:
+	WFRepeaterTask(std::function<SubTask *(WFRepeaterTask *)>&& create,
+				   std::function<void (WFRepeaterTask *)>&& cb) :
+		create(std::move(create)),
+		callback(std::move(cb))
+	{
+	}
+
+protected:
+	virtual ~WFRepeaterTask() { }
+};
+
 class WFModuleTask : public ParallelTask, protected SeriesWork
 {
 public:

src/factory/WFTaskFactory.h

@@ -101,6 +101,9 @@ using WFEmptyTask = WFGenericTask;
 using WFDynamicTask = WFGenericTask;
 using dynamic_create_t = std::function<SubTask *(WFDynamicTask *)>;
 
+using repeated_create_t = std::function<SubTask *(WFRepeaterTask *)>;
+using repeater_callback_t = std::function<void (WFRepeaterTask *)>;
+
 using module_callback_t = std::function<void (const WFModuleTask *)>;
 
 class WFTaskFactory
@@ -314,6 +317,12 @@ public:
 
 	static WFDynamicTask *create_dynamic_task(dynamic_create_t create);
 
+	static WFRepeaterTask *create_repeater_task(repeated_create_t create,
+												repeater_callback_t callback)
+	{
+		return new WFRepeaterTask(std::move(create), std::move(callback));
+	}
+
 public:
 	static WFModuleTask *create_module_task(SubTask *first,
 											module_callback_t callback)

src/factory/WFTaskFactory.inl

@@ -145,8 +145,8 @@ protected:
 	std::function<SubTask *(WFDynamicTask *)> create;
 
 public:
-	__WFDynamicTask(std::function<SubTask *(WFDynamicTask *)>&& func) :
-		create(std::move(func))
+	__WFDynamicTask(std::function<SubTask *(WFDynamicTask *)>&& create) :
+		create(std::move(create))
 	{
 	}
 };

tutorial/tutorial-09-http_file_server.cc

@@ -132,9 +132,7 @@ int main(int argc, char *argv[])
 	}
 
 	/* Test the server. */
-	while (1)
-	{
-		WFFacilities::WaitGroup wg(1);
+	auto&& create = [&scheme, port](WFRepeaterTask *)->SubTask *{
 		char buf[1024];
 		*buf = '\0';
 		printf("Input file name: (Ctrl-D to exit): ");
@@ -142,12 +140,12 @@ int main(int argc, char *argv[])
 		if (*buf == '\0')
 		{
 			printf("\n");
-			break;
+			return NULL;
 		}
 
 		std::string url = scheme + "127.0.0.1:" + std::to_string(port) + "/" + buf;
 		WFHttpTask *task = WFTaskFactory::create_http_task(url, 0, 0,
-									[&wg](WFHttpTask *task) {
+									[](WFHttpTask *task) {
 			auto *resp = task->get_resp();
 			if (strcmp(resp->get_status_code(), "200") == 0)
 			{
@@ -159,13 +157,19 @@ int main(int argc, char *argv[])
 			{
 				printf("%s %s\n", resp->get_status_code(), resp->get_reason_phrase());
 			}
-
-			wg.done();
 		});
 
-		task->start();
-		wg.wait();
-	}
+		return task;
+	};
+
+	WFFacilities::WaitGroup wg(1);
+	WFRepeaterTask *repeater;
+	repeater = WFTaskFactory::create_repeater_task(create, [&wg](WFRepeaterTask *) {
+		wg.done();
+	});
+
+	repeater->start();
+	wg.wait();
 
 	server.stop();
 	return 0;

tutorial/tutorial-10-user_defined_protocol/client.cc

@@ -59,51 +59,49 @@ int main(int argc, char *argv[])
 
 	host = argv[1];
 	port = atoi(argv[2]);
-	std::function<void (WFTutorialTask *task)> callback =
-		[&host, port, &callback](WFTutorialTask *task) {
-		int state = task->get_state();
-		int error = task->get_error();
-		TutorialResponse *resp = task->get_resp();
+
+	auto&& create = [host, port](WFRepeaterTask *)->SubTask *{
 		char buf[1024];
-		void *body;
-		size_t body_size;
-
-		if (state != WFT_STATE_SUCCESS)
-		{
-			if (state == WFT_STATE_SYS_ERROR)
-				fprintf(stderr, "SYS error: %s\n", strerror(error));
-			else if (state == WFT_STATE_DNS_ERROR)
-				fprintf(stderr, "DNS error: %s\n", gai_strerror(error));
-			else
-				fprintf(stderr, "other error.\n");
-			return;
-		}
-
-		resp->get_message_body_nocopy(&body, &body_size);
-		if (body_size != 0)
-			printf("Server Response: %.*s\n", (int)body_size, (char *)body);
-
 		printf("Input next request string (Ctrl-D to exit): ");
 		*buf = '\0';
 		scanf("%1023s", buf);
-		body_size = strlen(buf);
-		if (body_size > 0)
+		size_t body_size = strlen(buf);
+		if (body_size == 0)
 		{
-			WFTutorialTask *next;
-			next = MyFactory::create_tutorial_task(host, port, 0, callback);
-			next->get_req()->set_message_body(buf, body_size);
-			next->get_resp()->set_size_limit(4 * 1024);
-			**task << next; /* equal to: series_of(task)->push_back(next) */
-		}
-		else
 			printf("\n");
+			return NULL;
+		}
+
+		WFTutorialTask *task = MyFactory::create_tutorial_task(host, port, 0,
+											[](WFTutorialTask *task) {
+			int state = task->get_state();
+			int error = task->get_error();
+			TutorialResponse *resp = task->get_resp();
+			void *body;
+			size_t body_size;
+
+			if (state == WFT_STATE_SUCCESS)
+			{
+				resp->get_message_body_nocopy(&body, &body_size);
+				printf("Server Response: %.*s\n", (int)body_size, (char *)body);
+			}
+			else
+			{
+				const char *str = WFGlobal::get_error_string(state, error);
+				fprintf(stderr, "Error: %s\n", str);
+			}
+		});
+
+		task->get_req()->set_message_body(buf, body_size);
+		task->get_resp()->set_size_limit(4 * 1024);
+		return task;
 	};
 
-	/* First request is emtpy. We will ignore the server response. */
 	WFFacilities::WaitGroup wait_group(1);
-	WFTutorialTask *task = MyFactory::create_tutorial_task(host, port, 0, callback);
-	task->get_resp()->set_size_limit(4 * 1024);
-	Workflow::start_series_work(task, [&wait_group](const SeriesWork *) {
+
+	WFRepeaterTask *repeater;
+	repeater = WFTaskFactory::create_repeater_task(std::move(create), nullptr);
+	Workflow::start_series_work(repeater, [&wait_group](const SeriesWork *) {
 		wait_group.done();
 	});