myvtm/libtoolkit/log_udpdaemon.c

#include "precompile.h"

#include "log_udpdaemon.h"
#include "log_hdr.h"
#include "log_base.h"
#include "log.h"
#include <winpr/winsock.h>
#include <winpr/synch.h>
#include <winpr/thread.h>
#include <winpr/string.h>

typedef struct udplogdaemonfactory_t udplogdaemonfactory_t;

typedef struct udplogdaemon_t {
	logbase_t base;
	char *ip;
	unsigned short port;
	struct sockaddr_in listen_addr;
	SOCKET listen_handle;
	volatile LONG stop;
	HANDLE run_thread;
}udplogdaemon_t;

struct udplogdaemonfactory_t
{
	logfactory_t base;
};

static unsigned int __stdcall udplogdaemon_run(void* param)
{
	udplogdaemon_t *log = (udplogdaemon_t *)param;

	for (;;) {
		struct sockaddr_in addr;
		int addrlen = sizeof(addr);
		char buf[MAX_LOG_LEN];
		int len;

		len = recvfrom(log->listen_handle, buf, sizeof(buf), 0, (struct sockaddr*)&addr, &addrlen);

		if (log->stop)
			break;

		if (len > 0 && len < MAX_LOG_LEN) {
			struct log_hdr *hdr;
			struct log_hdr_str *inst;
			struct log_hdr_str *msg;
			int msg_len;
			int inst_len;
			int n = 0;
			hdr = (struct log_hdr *)&buf[n];
			n += sizeof(struct log_hdr);
			inst = (struct log_hdr_str *)&buf[n];
			inst_len = ntohs(inst->len);
			n += 2 + inst_len;
			msg = (struct log_hdr_str*)&buf[n];
			msg_len = ntohs(msg->len);
			if (inst_len && msg_len) {
				msg->data[msg_len] = 0;
				inst->data[inst_len] = 0;
				xlog_log(inst->data, hdr->level, msg->data);
			}
		} else {
			DWORD dwError = WSAGetLastError();
			dwError = dwError + 0;
			//...
		}
	}

	return 0;
}

static void udplogdaemonfactory_destroy(void *self)
{
	udplogdaemonfactory_t *fac = (udplogdaemonfactory_t *)self;
}

static void* udplogdaemonfactory_create_log(void *self, const char* inst)
{
	udplogdaemonfactory_t *fac = (udplogdaemonfactory_t *)self;
	udplogdaemon_t *log = MALLOC_T(udplogdaemon_t);
	log->base.level = XLOG_LEVEL_ALL;
	log->base.factory = self;
	log->base.inst_name = _strdup(inst);
	log->ip = 0;
	log->port = 0;
	log->listen_handle = INVALID_SOCKET;
	log->run_thread = NULL;
	return log;
}

static int udplogdaemonfactory_set_log_param(void *self, 
											 void *log, 
											 const char *key, 
											 const char *value)
{
	udplogdaemonfactory_t *fac = (udplogdaemonfactory_t *)self;
	udplogdaemon_t *slog = (udplogdaemon_t*)log;
	if (_stricmp(key, "ip") == 0) {
		slog->ip = _strdup(value);
	} else if (_stricmp(key, "port") == 0) {
		int nport = atoi(value);
		if (nport <= 65535 && nport > 0)
			slog->port = (unsigned short)nport;
	} else if (_stricmp(key, "level") == 0) {
		if (_stricmp(value, "all") == 0) {
			slog->base.level = 0;
		} else if (_stricmp(value, "trace") == 0) {
			slog->base.level = 1;
		} else if (_stricmp(value, "debug") == 0) {
			slog->base.level = 2;
		} else if (_stricmp(value, "info") == 0) {
			slog->base.level = 3;
		} else if (_stricmp(value, "warn") == 0) {
			slog->base.level = 4;
		} else if (_stricmp(value, "error") == 0) {
			slog->base.level = 5;
		} else if (_stricmp(value, "fatal") == 0) {
			slog->base.level = 6;
		} else if (_stricmp(value, "none") == 0) {
			slog->base.level = 7;
		} else {
			return -1;
		}
	} else {
		return -1;
	}
	return 0;
}

static int udplogdaemonfactory_init_log(void *self, void *log)
{
	udplogdaemonfactory_t *fac = (udplogdaemonfactory_t *)self;
	udplogdaemon_t *slog = (udplogdaemon_t*)log;

	memset(&slog->listen_addr, 0, sizeof(slog->listen_addr));
	slog->listen_addr.sin_addr.s_addr = inet_addr(slog->ip);
	slog->listen_addr.sin_port = htons(slog->port);
	slog->listen_addr.sin_family = AF_INET;

	slog->listen_handle = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
	if (slog->listen_handle == INVALID_SOCKET)
		return -1;

	if (bind(slog->listen_handle, (struct sockaddr*)&slog->listen_addr, sizeof(slog->listen_addr)) != 0) {
		closesocket(slog->listen_handle);
		slog->listen_handle = INVALID_SOCKET;
		return -1;
	}

	{
		int buf_len = 1 << 20;
		setsockopt(slog->listen_handle, SOL_SOCKET, SO_RCVBUF, (char*)&buf_len, sizeof(buf_len));
	}

	slog->stop = 0;

	slog->run_thread = (HANDLE)_beginthreadex(NULL, 0, &udplogdaemon_run, slog, 0, NULL);
	if (slog->run_thread == NULL) {
		closesocket(slog->listen_handle);
		slog->listen_handle = INVALID_SOCKET;
		return -1;
	}

	return 0;
}

static int udplogdaemonfactory_term_log(void *self, void *log)
{
	udplogdaemonfactory_t *fac = (udplogdaemonfactory_t *)self;
	udplogdaemon_t *slog = (udplogdaemon_t*)log;
	if (slog->run_thread) {
		char buf[1];
		int buf_len = 1;
		int i, tries = 3;
		SOCKET sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
		InterlockedExchange(&slog->stop, 1);
		for (i = 0; i < tries; ++i)
			sendto(sock, buf, buf_len, 0, (struct sockaddr*)&slog->listen_addr, sizeof(slog->listen_addr));
		closesocket(sock);
		WaitForSingleObject(slog->run_thread, INFINITE);
		CloseHandle(slog->run_thread);
		slog->run_thread = 0;
		closesocket(slog->listen_handle);
		slog->listen_handle = INVALID_SOCKET;
	}
	if (slog->ip) {
		free(slog->ip);
		slog->ip = NULL;;
	}
	return 0;
}

static void udplogdaemonfactory_destroy_log(void *self, void *log)
{
	udplogdaemon_t *slog = (udplogdaemon_t*)log;
	if (slog) {
		if (slog->base.inst_name)
			free(slog->base.inst_name);
		free(slog);
	}
}

static int udplogdaemonfactory_record_log(void *self, 
										  void *log, 
										  int level, 
										  unsigned long ts_low, 
										  unsigned long ts_high, 
										  const char *s, 
										  int sn)
{
	return -1;
}

int udplogdaemonfactory_create(logfactory_t **p_fac)
{
	udplogdaemonfactory_t *fac;

	if (!p_fac)
		return -1;

	fac = MALLOC_T(udplogdaemonfactory_t);
	fac->base.name = _strdup("udpdaemon");
	fac->base.record_log = &udplogdaemonfactory_record_log;
	fac->base.init_log = &udplogdaemonfactory_init_log;
	fac->base.set_log_param = &udplogdaemonfactory_set_log_param;
	fac->base.term_log = &udplogdaemonfactory_term_log;
	fac->base.create_log = &udplogdaemonfactory_create_log;
	fac->base.destroy_log = &udplogdaemonfactory_destroy_log;
	fac->base.destroy = &udplogdaemonfactory_destroy;

	*p_fac = &fac->base;

	return 0;
}