Support UDP

client.c

@@ -9,6 +9,7 @@
 #define TIMEOUT			30
 #define HTTP_PREFIX		"GET /"
 #define MNH_PREFIX		"mnhv1 "
+#define MNH_HEARTBEAT	"heartbeat\n"
 
 int str_cmp_head(const char *str, const char *head) {
 	int ca, cb;
@@ -19,18 +20,31 @@ int str_cmp_head(const char *str, const char *head) {
 	return cb ? (ca - cb) : 0;
 }
 
+int client_send(NODE *node, const void *buf, int len) {
+	if(node->udp)
+		return sendto(node->sock, buf, len, 0, (struct sockaddr*)&node->address, sizeof(node->address));
+	else
+		return send(node->sock, buf, len, 0);
+}
+
 void client_close(int epoll_fd, NODE *node) {
 	int fd = node->sock;
-	log_i("clean %d(%s)", fd, node->id);
+	log_i("clean %d(%s:%d)(%s)", 
+		fd, 
+		inet_ntoa(node->address.sin_addr), ntohs(node->address.sin_port),
+		node->id);
 
-	if(epoll_del(epoll_fd, fd) < 0)
-		log_e("delete %d from epoll error %d: %s", fd, errno, strerror(errno));
+	if(node->udp == false) {
+		if(epoll_del(epoll_fd, fd) < 0)
+			log_e("delete %d from epoll error %d: %s", fd, errno, strerror(errno));
+
+		close(fd);
+	}
 
-	close(fd);
 	node_free(node);
 }
 
-int client_http_sendResponse(int fd, int code, const char *body) {
+int client_http_sendResponse(NODE *node, int code, const char *body) {
 	char *codeMsg;
 	switch(code) {
 		case 200: codeMsg = "OK"; break;
@@ -48,7 +62,7 @@ int client_http_sendResponse(int fd, int code, const char *body) {
 		strlen(body),
 		body
 	);
-	return send(fd, buffer, strlen(buffer), 0);
+	return client_send(node, buffer, strlen(buffer));
 }
 
 void client_http_handleHandshake(int epoll_fd, NODE *node, void *buf, int len) {
@@ -61,15 +75,15 @@ void client_http_handleHandshake(int epoll_fd, NODE *node, void *buf, int len) {
 		if(n != NULL) {
 			char body[1024];
 			sprintf(body, "%s:%d", inet_ntoa(n->address.sin_addr), ntohs(n->address.sin_port));
-			client_http_sendResponse(node->sock, 200, body);
+			client_http_sendResponse(node, 200, body);
 			log_i("%s:%d query \"%s\": %s", 
 				inet_ntoa(node->address.sin_addr), ntohs(node->address.sin_port),
 				p,
 				body
 				);
 		} else {
 			// not found
-			client_http_sendResponse(node->sock, 404, "Not found");
+			client_http_sendResponse(node, 404, "Not found");
 			log_i("%s:%d query \"%s\": Not found", 
 				inet_ntoa(node->address.sin_addr), ntohs(node->address.sin_port),
 				p
@@ -101,27 +115,34 @@ void client_mnh_handleHandshake(int epoll_fd, NODE *node, void *buf, int len) {
 
 	char buffer[1024];
 	sprintf(buffer, "%s:%d", inet_ntoa(node->address.sin_addr), ntohs(node->address.sin_port));
-	send(node->sock, buffer, strlen(buffer), 0);
+	client_send(node, buffer, strlen(buffer));
 
 	node->state = STATE_MNHV1_HEARTBEAT;
 }
 
-void client_handle(int epoll_fd, NODE *node, void *buf, int len) {
+void client_mnh_handleHeartbeat(int epoll_fd, NODE *node, void *buf, int len) {
+	if(str_cmp_head(buf, MNH_HEARTBEAT) != 0) {
+		log_i("%d(%s) sent wrong heartbeat",node->sock, node->id);
+		return;
+	}
+
 	node_update(node);
+	client_send(node, ".", 1);
+}
+
+void client_handle(int epoll_fd, NODE *node, void *buf, int len) {
+	((char *) buf)[len] = 0;
 
 	if(node->state == STATE_HANDSHAKE) {
-		((char *) buf)[len] = 0;
+		node_update(node);
 		if(str_cmp_head(buf, HTTP_PREFIX) == 0)
 			client_http_handleHandshake(epoll_fd, node, buf, len);
 		else if(str_cmp_head(buf, MNH_PREFIX) == 0)
 			client_mnh_handleHandshake(epoll_fd, node, buf, len);
 		else
 			client_close(epoll_fd, node);
 	} else {
-		//log_d("%d(%s) recv %dB: %s", node->sock, node->id, len, buf);
-		log_d("%d(%s) recv %dB: %s", node->sock, node->id, len, "");
-
-		send(node->sock, ".", 1, 0);
+		client_mnh_handleHeartbeat(epoll_fd, node, buf, len);
 	}
 }
 

main.c

@@ -2,7 +2,8 @@
 #include "epoll.h"
 #include "node.h"
 #include "client.h"
-#include "server.h"
+#include "tcpServer.h"
+#include "udpServer.h"
 
 #include <stdlib.h>
 #include <stdbool.h>
@@ -33,13 +34,23 @@ int main(int argc, char const *argv[])
 		return 1;
 	}
 
-	int server_fd = server_create(port);
-	if(server_fd < 0) {
-		log_e("server_create error %d: %s", errno, strerror(errno));
+	int tcpServer_fd = tcpServer_create(port);
+	if(tcpServer_fd < 0) {
+		log_e("tcpServer_create error %d: %s", errno, strerror(errno));
 		return 1;
 	}
-	if(epoll_add(epoll_fd, server_fd, 0) < 0) {
-		log_e("add server to epoll error %d: %s", errno, strerror(errno));
+	if(epoll_add(epoll_fd, tcpServer_fd, &tcpServer_fd) < 0) {
+		log_e("add tcp server to epoll error %d: %s", errno, strerror(errno));
+		return 1;
+	}
+
+	int udpServer_fd = udpServer_create(port);
+	if(udpServer_fd < 0) {
+		log_e("udpServer_create error %d: %s", errno, strerror(errno));
+		return 1;
+	}
+	if(epoll_add(epoll_fd, udpServer_fd, &udpServer_fd) < 0) {
+		log_e("add udp server to epoll error %d: %s", errno, strerror(errno));
 		return 1;
 	}
 
@@ -59,8 +70,10 @@ int main(int argc, char const *argv[])
 		}
 
 		for(int i = 0; i < ready; i++) {
-			if(evs[i].data.ptr == NULL)
-				server_handle(epoll_fd, server_fd);
+			if(evs[i].data.ptr == &tcpServer_fd)
+				tcpServer_handle(epoll_fd, tcpServer_fd);
+			else if(evs[i].data.ptr == &udpServer_fd)
+				udpServer_handle(epoll_fd, udpServer_fd);
 			else {
 				NODE *node = evs[i].data.ptr;
 

node.c

@@ -36,11 +36,12 @@ void node_delete(NODE *node) {
 		end = prev;
 }
 
-NODE *node_malloc(int sock, struct sockaddr_in address) {
+NODE *node_malloc(bool udp, int sock, struct sockaddr_in address) {
 	NODE *node = malloc(sizeof(NODE));
 	if(node == NULL)
 		return NULL;
 
+	node->udp = udp;
 	node->sock = sock;
 	node->address = address;
 
@@ -69,6 +70,15 @@ NODE *node_queryById(const char *id) {
 	return NULL;
 }
 
+NODE *node_queryByAddress(bool udp, struct sockaddr_in address) {
+	for(NODE *node = head; node; node = node->next) {
+		if((node->udp == udp) && (memcmp(&node->address, &address, sizeof(struct sockaddr_in)) == 0))
+			return node;
+	}
+
+	return NULL;
+}
+
 void node_update(NODE *node) {
 	node_delete(node);
 	node_append(node);

node.h

@@ -1,3 +1,4 @@
+#include <stdbool.h>
 #include <arpa/inet.h>
 
 #ifndef __node_h__
@@ -10,6 +11,7 @@ typedef enum {
 } STATE;
 
 typedef struct NODE {
+	bool udp;
 	int sock;
 	struct sockaddr_in address;
 
@@ -21,11 +23,12 @@ typedef struct NODE {
 	struct NODE *next;
 } NODE;
 
-NODE *node_malloc(int sock, struct sockaddr_in address);
+NODE *node_malloc(bool udp, int sock, struct sockaddr_in address);
 void node_free(NODE *node);
 
 NODE *node_getFirst();
 NODE *node_queryById(const char *id);
+NODE *node_queryByAddress(bool udp, struct sockaddr_in address);
 
 void node_update(NODE *node);
 time_t node_getPast(NODE *node);

server.c → tcpServer.c

@@ -1,4 +1,4 @@
-#include "server.h"
+#include "tcpServer.h"
 #include "epoll.h"
 #include "log.h"
 #include "node.h"
@@ -7,7 +7,7 @@
 #include <string.h>
 #include <errno.h>
 
-int server_create(int port) {
+int tcpServer_create(int port) {
 	int fd = socket(AF_INET, SOCK_STREAM, 0);
 	if(fd < 0)
 		return -1;
@@ -35,7 +35,7 @@ int server_create(int port) {
 	return fd;
 }
 
-void server_handle(int epoll_fd, int server_fd) {
+void tcpServer_handle(int epoll_fd, int server_fd) {
 	struct sockaddr_in address;
 	int addrlen = sizeof(address);
 
@@ -46,7 +46,7 @@ void server_handle(int epoll_fd, int server_fd) {
 	}
 	log_i("accepted %d from %s:%d", new_fd, inet_ntoa(address.sin_addr), ntohs(address.sin_port));
 
-	NODE *node = node_malloc(new_fd, address);
+	NODE *node = node_malloc(false, new_fd, address);
 	if(node == NULL) {
 		log_e("node_create error: %d(%s)", errno, strerror(errno));
 		close(new_fd);

tcpServer.h

@@ -0,0 +1,7 @@
+#ifndef __tcpServer_h__
+#define __tcpServer_h__
+
+int tcpServer_create(int port);
+void tcpServer_handle(int epoll_fd, int server_fd);
+
+#endif

udpServer.c

@@ -0,0 +1,59 @@
+#include "udpServer.h"
+#include "epoll.h"
+#include "log.h"
+#include "node.h"
+#include "client.h"
+
+#include <unistd.h>
+#include <string.h>
+#include <errno.h>
+
+int udpServer_create(int port) {
+	int fd = socket(AF_INET, SOCK_DGRAM, 0);
+	if(fd < 0)
+		return -1;
+
+	int opt = 1;
+	setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
+	setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt));
+
+	struct sockaddr_in address;
+	int addrlen = sizeof(address);
+	address.sin_family = AF_INET;
+	address.sin_addr.s_addr = INADDR_ANY; 
+	address.sin_port = htons(port);
+
+	if(bind(fd, (struct sockaddr *)&address, sizeof(address)) < 0) {
+		close(fd);
+		return -1;
+	}
+
+	return fd;
+}
+
+void udpServer_handle(int epoll_fd, int server_fd) {
+	struct sockaddr_in address;
+	int addrlen = sizeof(address);
+	char buf[9001];
+
+	int len = recvfrom(server_fd, buf, 9000, 0, (struct sockaddr*)&address, &addrlen);
+	if(len < 0) {
+		log_e("recvfrom error %d: %s", errno, strerror(errno));
+		return;
+	}
+	log_d("recv %dB from %s:%d", len, inet_ntoa(address.sin_addr), ntohs(address.sin_port));
+
+	NODE *node = node_queryByAddress(true, address);
+	if(node == NULL) {
+		// New connection
+		log_d("new connection from %s:%d", inet_ntoa(address.sin_addr), ntohs(address.sin_port));
+
+		node = node_malloc(true, server_fd, address);
+		if(node == NULL) {
+			log_e("node_create error: %d(%s)", errno, strerror(errno));
+			return;
+		}
+	}
+
+	client_handle(epoll_fd, node, buf, len);
+}

udpServer.h

@@ -0,0 +1,7 @@
+#ifndef __udpServer_h__
+#define __udpServer_h__
+
+int udpServer_create(int port);
+void udpServer_handle(int epoll_fd, int server_fd);
+
+#endif