Support UDP server.

src/kernel/Communicator.h

@@ -31,9 +31,8 @@
 
 class CommConnection
 {
-protected:
+public:
 	virtual ~CommConnection() { }
-	friend class Communicator;
 };
 
 class CommTarget
@@ -158,7 +157,8 @@ class CommSession
 private:
 	struct timespec begin_time;
 	int timeout;
-	int passive;
+	short passive;
+	short reliable;
 
 public:
 	CommSession() { this->passive = 0; }
@@ -298,16 +298,6 @@ class Communicator
 
 	int create_handler_threads(size_t handler_threads);
 
-	int nonblock_connect(CommTarget *target);
-	int nonblock_listen(CommService *service);
-
-	struct CommConnEntry *launch_conn(CommSession *session,
-									  CommTarget *target);
-	struct CommConnEntry *accept_conn(class CommServiceTarget *target,
-									  CommService *service);
-
-	void release_conn(struct CommConnEntry *entry);
-
 	void shutdown_service(CommService *service);
 
 	void shutdown_io_service(IOService *service);
@@ -319,10 +309,14 @@ class Communicator
 
 	int send_message(struct CommConnEntry *entry);
 
+	int request_new_conn(CommSession *session, CommTarget *target);
+
 	int request_idle_conn(CommSession *session, CommTarget *target);
 	int reply_idle_conn(CommSession *session, CommTarget *target);
 
-	int request_new_conn(CommSession *session, CommTarget *target);
+	int reply_message_unreliable(struct CommConnEntry *entry);
+
+	int reply_unreliable(CommSession *session, CommTarget *target);
 
 	void handle_incoming_request(struct poller_result *res);
 	void handle_incoming_reply(struct poller_result *res);
@@ -336,6 +330,8 @@ class Communicator
 	void handle_connect_result(struct poller_result *res);
 	void handle_listen_result(struct poller_result *res);
 
+	void handle_recvfrom_result(struct poller_result *res);
+
 	void handle_ssl_accept_result(struct poller_result *res);
 
 	void handle_sleep_result(struct poller_result *res);
@@ -344,6 +340,14 @@ class Communicator
 
 	static void handler_thread_routine(void *context);
 
+	static int nonblock_connect(CommTarget *target);
+	static int nonblock_listen(CommService *service, int *reliable);
+
+	static struct CommConnEntry *launch_conn(CommSession *session,
+											 CommTarget *target);
+	static struct CommConnEntry *accept_conn(class CommServiceTarget *target,
+											 CommService *service);
+
 	static int first_timeout(CommSession *session);
 	static int next_timeout(CommSession *session);
 
@@ -358,11 +362,17 @@ class Communicator
 	static poller_message_t *create_request(void *context);
 	static poller_message_t *create_reply(void *context);
 
+	static int recv_request(const void *buf, size_t size,
+							struct CommConnEntry *entry);
+
 	static int partial_written(size_t n, void *context);
 
 	static void *accept(const struct sockaddr *addr, socklen_t addrlen,
 						int sockfd, void *context);
 
+	static void *recvfrom(const struct sockaddr *addr, socklen_t addrlen,
+						  const void *buf, size_t size, void *context);
+
 	static void callback(struct poller_result *res, void *context);
 
 public:

src/server/WFDnsServer.h

@@ -29,6 +29,7 @@ using WFDnsServer = WFServer<protocol::DnsRequest,
 
 static constexpr struct WFServerParams DNS_SERVER_PARAMS_DEFAULT =
 {
+	.transport_type			=	TT_UDP,
 	.max_connections		=	2000,
 	.peer_response_timeout	=	10 * 1000,
 	.receive_timeout		=	-1,

src/server/WFHttpServer.h

@@ -30,6 +30,7 @@ using WFHttpServer = WFServer<protocol::HttpRequest,
 
 static constexpr struct WFServerParams HTTP_SERVER_PARAMS_DEFAULT =
 {
+	.transport_type			=	TT_TCP,
 	.max_connections		=	2000,
 	.peer_response_timeout	=	10 * 1000,
 	.receive_timeout		=	-1,

src/server/WFMySQLServer.h

@@ -30,6 +30,7 @@ class MySQLServer;
 
 static constexpr struct WFServerParams MYSQL_SERVER_PARAMS_DEFAULT =
 {
+	.transport_type			=	TT_TCP,
 	.max_connections		=	2000,
 	.peer_response_timeout	=	10 * 1000,
 	.receive_timeout		=	-1,

src/server/WFRedisServer.h

@@ -29,6 +29,7 @@ using WFRedisServer = WFServer<protocol::RedisRequest,
 
 static constexpr struct WFServerParams REDIS_SERVER_PARAMS_DEFAULT =
 {
+	.transport_type			=	TT_TCP,
 	.max_connections		=	2000,
 	.peer_response_timeout	=	10 * 1000,
 	.receive_timeout		=	-1,

src/server/WFServer.cc

@@ -27,6 +27,7 @@
 #include <condition_variable>
 #include <openssl/ssl.h>
 #include "CommScheduler.h"
+#include "EndpointParams.h"
 #include "WFConnection.h"
 #include "WFGlobal.h"
 #include "WFServer.h"
@@ -121,10 +122,34 @@ int WFServerBase::create_listen_fd()
 	{
 		const struct sockaddr *bind_addr;
 		socklen_t addrlen;
+		int type, protocol;
 		int reuse = 1;
 
+		switch (this->params.transport_type)
+		{
+		case TT_TCP:
+		case TT_TCP_SSL:
+			type = SOCK_STREAM;
+			protocol = 0;
+			break;
+		case TT_UDP:
+			type = SOCK_DGRAM;
+			protocol = 0;
+			break;
+#ifdef IPPROTO_SCTP
+		case TT_SCTP:
+		case TT_SCTP_SSL:
+			type = SOCK_STREAM;
+			protocol = IPPROTO_SCTP;
+			break;
+#endif
+		default:
+			errno = EPROTONOSUPPORT;
+			return -1;
+		}
+
 		this->get_addr(&bind_addr, &addrlen);
-		this->listen_fd = socket(bind_addr->sa_family, SOCK_STREAM, 0);
+		this->listen_fd = socket(bind_addr->sa_family, type, protocol);
 		if (this->listen_fd >= 0)
 		{
 			setsockopt(this->listen_fd, SOL_SOCKET, SO_REUSEADDR,

src/server/WFServer.h

@@ -28,10 +28,12 @@
 #include <mutex>
 #include <condition_variable>
 #include <openssl/ssl.h>
+#include "EndpointParams.h"
 #include "WFTaskFactory.h"
 
 struct WFServerParams
 {
+	enum TransportType transport_type;
 	size_t max_connections;
 	int peer_response_timeout;	/* timeout of each read or write operation */
 	int receive_timeout;	/* timeout of receiving the whole message */
@@ -42,6 +44,7 @@ struct WFServerParams
 
 static constexpr struct WFServerParams SERVER_PARAMS_DEFAULT =
 {
+	.transport_type			=	TT_TCP,
 	.max_connections		=	2000,
 	.peer_response_timeout	=	10 * 1000,
 	.receive_timeout		=	-1,