#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h> 
#include <stdio.h> 
#include <stdlib.h>
#include <assert.h>
#include <errno.h>
#include <pthread.h>
#include <string.h>
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <sys/prctl.h>

#include "wy_singleflow_entry.h"


extern struct wysf_global_info g_wysf_global_info;

static int udp_broadcast_server_socket_init(void *log_runtime, u_int16_t port)
{
	struct sockaddr_in servaddr;
	int on = 1, socket_fd;
	
	if((socket_fd = socket(AF_INET,SOCK_DGRAM,0))<0)//AF_INET:IPV4 AF_INET6:IPV6
	{
		return -1;
	}
	if(setsockopt(socket_fd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on))<0)
	{
		MESA_RUNTIME_LOGV3(log_runtime, RLOG_LV_FATAL, "setsockopt for udp_port %u failed, because:%s!!", port, strerror(errno));
	}
	on = 5242880;
	if(setsockopt(socket_fd, SOL_SOCKET, SO_RCVBUF, &on, sizeof(on))<0)
	{
		MESA_RUNTIME_LOGV3(log_runtime, RLOG_LV_FATAL, "setsockopt for udp_port %u failed, because:%s!!", port, strerror(errno));
	}

	bzero(&servaddr,sizeof(servaddr));
	servaddr.sin_family = AF_INET;
	servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
	servaddr.sin_port = htons(port);
	
	if(bind(socket_fd,(struct sockaddr *)&servaddr,sizeof(servaddr))<0)
	{
		MESA_RUNTIME_LOGV3(log_runtime, RLOG_LV_FATAL, "bind udp socket port %u failed, because: %s!!", port, strerror(errno));
		close(socket_fd);
		return -2;
	}
	evutil_make_socket_nonblocking(socket_fd);
	return socket_fd;
}

static long broadcast_stream_tuple_htable_cb(void *data, const uchar *key, uint size, void *user_arg)
{
	int32_t clj_ip=*(int32_t *)user_arg;
	std::map<int32_t, struct judian_as_group*>::iterator iter;
	struct judian_as_group *group;

	if((iter = g_wysf_global_info.forwardip2judian->find(clj_ip)) == g_wysf_global_info.forwardip2judian->end())
	{
		char ipbuffer[32];
		inet_ntop(AF_INET, &clj_ip, ipbuffer, 32);
		MESA_RUNTIME_LOGV3(g_wysf_global_info.log_runtime, RLOG_LV_FATAL, "lookup group for clj_ip: %s not found!!", ipbuffer);
		return -1;
	}
	group = iter->second;

	if(data != NULL)
	{
		MESA_htable_del(g_wysf_global_info.stream_tuple_mapping, key, size, NULL);
	}
	MESA_htable_add(g_wysf_global_info.stream_tuple_mapping, key, size, group);
	return 0;
}

void udp_server_read_broadcast_cb(evutil_socket_t fd, short events, void *arg)
{
	struct udp_broadcast_msg *broadmsg;
	char buffer[4096];
	int from_len = 0, buf_len, clj_ip;
	struct sockaddr_in saddr;
	long cb_ret;

	from_len = sizeof (struct sockaddr);
	while((buf_len = recvfrom (fd, buffer, 4096, 0, (struct sockaddr*)&saddr, (socklen_t *)&from_len))>0)
	{
		broadmsg = (struct udp_broadcast_msg *)buffer;
		if((u_int32_t)buf_len!=sizeof(struct udp_broadcast_msg) || broadmsg->header.magic_number != WYSF_MAGIC_NUMBER ||
			broadmsg->header.total_len!=sizeof(struct broadcast_message))
		{
			continue;
		}

		clj_ip = broadmsg->msg.clj_ip;
		broadmsg->msg.clj_ip = 0;
		MESA_htable_search_cb(g_wysf_global_info.stream_tuple_mapping, (uchar*)&broadmsg->msg, sizeof(struct broadcast_message),
			broadcast_stream_tuple_htable_cb, &clj_ip, &cb_ret);

		from_len = sizeof (struct sockaddr);
	}
}

void *thread_fwdnode_breoadcast(void *arg)
{
	struct event_base *evbase;
	struct event udp_server_event;

	evbase = event_base_new();
	
	if((g_wysf_global_info.broadcast_udp_servert_sockfd = udp_broadcast_server_socket_init(g_wysf_global_info.log_runtime, g_wysf_global_info.broadcast_udp_server_port)) < 0)
	{
		assert(0);return NULL;
	}
	event_assign(&udp_server_event, evbase, g_wysf_global_info.broadcast_udp_servert_sockfd, 
		EV_READ|EV_PERSIST, udp_server_read_broadcast_cb, NULL);
	event_add(&udp_server_event, NULL);

	event_base_dispatch(evbase);
	printf("Libevent dispath error, should not run here.\n");
	MESA_RUNTIME_LOGV3(g_wysf_global_info.log_runtime, RLOG_LV_FATAL, "Libevent dispath error, should not run here.");
	assert(0);return NULL;
}