path: root/src/qq_file_global.c

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <errno.h>
#include <assert.h>
#include <iconv.h>
#include <pthread.h>

#include <MESA/stream.h>
#include <MESA/DocumentAnalyze.h>
#include <MESA/MESA_handle_logger.h>

#include "qq_file_entry.h"
#include "qq_file_global.h"
#include "connector.h"

#define QQ_OPT_SERVICE "SERVICE_IMQQ"
#define QQ_OPT_FILESET "opt_file_set"
#define QQ_OPT_FILENAME "opt_file_name"
#define QQ_OPT_FILECONT "opt_file_content"
#define QQ_OPT_NUMFROM "opt_im_from"
#define QQ_OPT_NUMTO "opt_im_to"
#define QQ_OPT_ACCOUNT "opt_im_account"

extern qq_global_info_t g_qq_global_info;
const unsigned char gzip_feature[4]={0x1f, 0x8b, 0x08, 0x00};

int write_file_cont2disk(char *buffer, int filelen, char *filename, int seq, int random)
{
	FILE *fp;
	int writelen, offset=0;
	char filenamestr[512];

	sprintf(filenamestr, "%s/%d_%u_%s", QQ_DEBUG_FILEPATH, seq, random, filename);

	fp = fopen(filenamestr, "ab+");
	if(fp == NULL)
	{
		printf("fopen file %s failed.\n", filenamestr);
		return -1;
	}
	
	while(offset < filelen)
	{
		writelen = fwrite(buffer + offset, 1, filelen - offset, fp);
		if(writelen < 0)
		{
			printf("read error: %s\n", strerror(errno));
			return -1;
		}

		offset += writelen;
	}
	fclose(fp);
	return 0;
}

int buf_cache_check_size(trans_file_info_t *pdstBuf, int wantedlen, int thread_seq)
{
	int ret=0, needlen, block_num=0;
		
	if(pdstBuf->max_len > wantedlen + pdstBuf->len)
		return 0;

	needlen = wantedlen + pdstBuf->len - pdstBuf->max_len;
	block_num = needlen/REALLOC_BLOCK_SIZE;
	if(needlen%REALLOC_BLOCK_SIZE)
		block_num += 1;

	pdstBuf->max_len += block_num*REALLOC_BLOCK_SIZE;

	if(pdstBuf->max_len > MAX_MALLOC_SIZE)
	{
		pdstBuf->max_len -= block_num*REALLOC_BLOCK_SIZE;
		return -1;
	}
		
	if(pdstBuf->buf == NULL)
	{
		pdstBuf->buf = (char *)dictator_malloc(thread_seq, pdstBuf->max_len);
	}
	else
	{
		pdstBuf->buf = (char *)dictator_realloc(thread_seq, pdstBuf->buf, pdstBuf->max_len);
		ret = 1;
	}

	return ret;
}

int save_initial_tcp_data(trans_file_info_t *pdstBuf, char *data, int datalen, int thread_seq)
{
	if(buf_cache_check_size(pdstBuf, datalen, thread_seq) < 0)
	{
		return -1;
	}
	memcpy(pdstBuf->buf+pdstBuf->len, data, datalen);
	pdstBuf->len += datalen;

	return 0;
}


int convert_from_unicode2utf8(char *in, int in_len, char *out, int *out_len)
{
	iconv_t cd;
	size_t inlen = in_len, outlen=*out_len, ret;
	
	cd = iconv_open("utf-8", "unicode");
	if(cd == (iconv_t)-1)
	{
		return -1;
	}
	ret = iconv(cd, &in, &inlen, &out, &outlen);
	if(ret == (size_t)-1)
	{
		iconv_close(cd);
		return -1;
	}
	iconv_close(cd);
	*out_len = outlen;

	return 0;
}

int sendback_qq_data(qq_proto_pme_t *qq_pme, struct streaminfo *a_udp, char *filecont, int contlen)
{
	int nest_opt_num = 1;
	void *nested_opt_handle, *nested_opt_unit_handle, *simple_opt=NULL;
	
	if(qq_pme->connector_hdl == NULL)
	{
		appd_basic_t basic;
		ulong64 streamid[1];
		char qqfrom[32], qq_to[32];

		streamid[0] = project_req_get_ulong(a_udp, g_qq_global_info.project_id);
		
		basic.addr = (struct ipaddr*)&a_udp->addr;
		basic.addr_num = 1;
		basic.appd_type = APPD_SENDBACK;
		basic.destination = DEST_APPD;
		basic.service_name = QQ_OPT_SERVICE;
		basic.streamid =  streamid;
		basic.streamid_num = 1;
		
		qq_pme->connector_hdl = connector_start(&basic, a_udp);
		if(qq_pme->connector_hdl == NULL)
			return APP_STATE_DROPME;

		if(qq_pme->filename_len > 0)
		{
			nest_opt_num = 2;
		}
		sprintf(qqfrom, "%u", qq_pme->qqnum_from);
		sprintf(qq_to, "%u", qq_pme->qqnum_to);

		simple_opt = connector_create_simple_opt(3);
		connector_append_simple_opt(simple_opt, QQ_OPT_NUMFROM,qqfrom, strlen(qqfrom));
		connector_append_simple_opt(simple_opt, QQ_OPT_ACCOUNT,qqfrom, strlen(qqfrom));
		connector_append_simple_opt(simple_opt, QQ_OPT_NUMTO,qq_to, strlen(qq_to));
		connector(qq_pme->connector_hdl, COMPLETE, simple_opt, NULL);
		connector_free_simple_opt(&simple_opt);
	}

	nested_opt_handle = connector_create_nested_opt(1);
	nested_opt_unit_handle = connector_create_nested_opt_unit(QQ_OPT_FILESET, nest_opt_num);
	if(nest_opt_num == 2)
	{
		connector_append_nested_opt_unit(nested_opt_unit_handle, QQ_OPT_FILENAME,qq_pme->filename, qq_pme->filename_len);
	}
	connector_append_nested_opt_unit(nested_opt_unit_handle, QQ_OPT_FILECONT,filecont, contlen);
	connector_append_nested_opt(nested_opt_handle,nested_opt_unit_handle);

	if(qq_pme->end_flag)
	{
		connector(qq_pme->connector_hdl, FRAG_END, NULL, nested_opt_handle);
		connector_finish(qq_pme->connector_hdl);
		qq_pme->connector_hdl = NULL;
	}
	else
	{
		connector(qq_pme->connector_hdl, FRAG, NULL, nested_opt_handle);
	}
	connector_free_nested_opt_unit(&nested_opt_unit_handle);	
	connector_free_nested_opt(&nested_opt_handle);
	
	return APP_STATE_GIVEME;
}

int protocol_process(qq_proto_pme_t *qq_pme, long long prot_flag, char *buf_in, int inlen, struct streaminfo *a_udp, const void *raw_pkt)
{
	int rec = APP_STATE_GIVEME;
	char buffer[512]={0}, *buf=buf_in;
	int outlen=512, ret, buflen=inlen, k;
	result_array_t result_array;
	
	switch(prot_flag)
	{
		case QQ_FILENAME:
			if(qq_pme->pending_flag==0)
				break;
			else
			{
				qq_pme->fileinfo.block_seq = -1; //���ݿ����¼������
				qq_pme->end_flag = 0;
				qq_pme->pending_flag = 0;
				qq_pme->fileinfo.file_seq += 1;

				if(qq_pme->qqnum_from == 0)
				{
					qq_pme->qqnum_to = qq_pme->qqnum[qq_pme->file_dir];
					if(qq_pme->qqnum_to == qq_pme->qqnum[0])
						qq_pme->qqnum_from = qq_pme->qqnum[1];
					else
						qq_pme->qqnum_from = qq_pme->qqnum[0];
				}

				if(g_qq_global_info.debug_sw)
				{
					qq_pme->random = rand();
				}
			}
			if(!convert_from_unicode2utf8(buf_in, inlen, buffer, &outlen))
			{
				snprintf(qq_pme->filename, sizeof(qq_pme->filename), "%s", buffer);
				qq_pme->filename_len = strlen(qq_pme->filename);
				MESA_HANDLE_RUNTIME_LOGV2(g_qq_global_info.runtime_log,RLOG_LV_DEBUG, MODULE_NAME, "DETECT_QQ_FILE %s, QQ_FROM: %u, QQ_TO: %u, FILE_LEN: %u, FILENAME: %s", \
						qq_pme->tcp_flag?"TCP":"UDP", qq_pme->qqnum_from, qq_pme->qqnum_to, qq_pme->total_file_len,qq_pme->filename);
			}
			else
			{
				MESA_HANDLE_RUNTIME_LOGV2(g_qq_global_info.runtime_log,RLOG_LV_INFO, MODULE_NAME, "convert_from_unicode2utf8 error: %s\n", strerror(errno));
				return APP_STATE_DROPME;
			}
			break;

		case QQ_FILECONT_END:
		case QQ_FILECONT:
			if(qq_pme->filename_len == 0)
				return APP_STATE_DROPME;
			
			if(prot_flag==QQ_FILECONT_END)
			{
				if(qq_pme->end_flag || qq_pme->pending_flag)
					break;
				else
				{
					qq_pme->end_flag = 1;
					qq_pme->pending_flag = 1;
				}
			}
			else if(qq_pme->pending_flag)
			{
				qq_pme->end_flag = 0;
				qq_pme->pending_flag = 0;
			}
			
			if(inlen >= 4 && !memcmp(buf_in, gzip_feature, 4) && qq_pme->compress_flag)
			{
				docanalyze_streamparam_t doc_hanlde = docanalyze_startstream(DOC_GZIP_TYPE, g_qq_global_info.instanse, qq_pme->thread_id);
				if(doc_hanlde ==NULL)
				{
					MESA_HANDLE_RUNTIME_LOGV2(g_qq_global_info.runtime_log,RLOG_LV_DEBUG, MODULE_NAME, "docanalyze_startstream() error.");
					return APP_STATE_GIVEME;
				}
				memset(&result_array, 0, sizeof(result_array_t));
				ret = docanalyze_parsestream(doc_hanlde, buf_in, inlen, &result_array);
				if(ret == DOC_PRO_OK)
				{
					for (k=0; k < result_array.result_num; k++)
					{
						buf = result_array.result_buff[k].presult;
						buflen = result_array.result_buff[k].size;
					}
				}
				qq_pme->curr_file_len += buflen;

				if(g_qq_global_info.debug_sw)
				{
					write_file_cont2disk(buf, buflen, qq_pme->filename, qq_pme->fileinfo.file_seq, qq_pme->random);
				}
				rec = sendback_qq_data(qq_pme, a_udp, buf, buflen);
				
				docanalyze_freeresult(&result_array);
				docanalyze_endstream(doc_hanlde);
			}
			else
			{
				qq_pme->curr_file_len += buflen;

				if(g_qq_global_info.debug_sw)
				{
					write_file_cont2disk(buf, buflen, qq_pme->filename, qq_pme->fileinfo.file_seq, qq_pme->random);
				}
				rec = sendback_qq_data(qq_pme, a_udp, buf, buflen);
			}
			break;

		default: break;
	}

	return rec;
}