myvtm/Other/libbizchan/bizchan.cpp

#include "bizchan.h"
#include "quicklz.h"

#ifdef RVC_OS_WIN
#define WIN32_LEAN_AND_MEAN
#include <Windows.h>
#include <WinSock2.h>
#include <WinSock.h>
#include <process.h>
#else
#include <sys/socket.h>
#include <fcntl.h>
#include <errno.h>
#include <unistd.h>
#include <pthread.h>
#include <netinet/in.h>
#include <string.h>
#include <arpa/inet.h>
//#include "winpr/winsock.h"
//#include "winpr/file.h"
#endif

#include <stdlib.h>
#include <stdio.h>
#include <assert.h>

#include "chan_protocol.h"

//#include "acmstrdec.h"
//#include "acmstrenc.h"
#include "openssl/rc4.h"

#include "ListEntry.h"
#include <screencodec.h>
#include "jpeg2k.h"

#ifdef RVC_OS_WIN
#include <DbgHelp.h>
#pragma comment(lib, "dbghelp.lib")
#endif

#define PING_INTERVAL	10000 // 10s
#define MAX_TIMEOUT 60000     //超时时间60S,20170313修改，解决排队机闪呼的问题
#define DEFAULT_RX_BUF_SIZE	8192
#define KEY_LEN	16

#ifndef RVC_OS_WIN
#include "SpBase.h"

typedef int SOCKET;

#ifndef INVALID_SOCKET
#define INVALID_SOCKET (SOCKET)(~0)
#endif
#ifndef SOCKET_ERROR
#define SOCKET_ERROR (-1)
#endif

#define max(a,b) (((a) > (b)) ? (a) : (b))
#define min(a,b) (((a) < (b)) ? (a) : (b))

unsigned long GetTickCount()
{
	struct timespec ts;
	clock_gettime(CLOCK_MONOTONIC, &ts);
	return (ts.tv_sec * 1000 + ts.tv_nsec / 1000000);
}

#endif

typedef struct recv_info_t {
	int offset;
	char *buf;
	int buf_len;
	char *unzip_buf;
	int unzip_len;
	qlz_state_decompress decompress_state;
}recv_info_t;

typedef struct send_buf_node {
	LIST_ENTRY entry;
	char *buf;
	int left;
	int sended;
	int need_encrypt;
}send_buf_node;

typedef struct send_info_t {
	//send_buf_node *send_list;
	LIST_ENTRY send_list;
#ifdef RVC_OS_WIN
	CRITICAL_SECTION lock;
#else
	pthread_mutexattr_t attr;
	pthread_mutex_t lock;
#endif
	//qlz_state_compress compress_state;
}send_info_t;

typedef struct img_recv_t {
	char *data;
	int length;
	int offset;
	int id;
}img_recv_t;

struct bizchan_t
{
	bizchan_config_t config;
	OnRecvPacket winsync_on_recv_cb;
	OnMode mode_cb;
	void *winsync_user_data;
	bizchan_callback_t cb;
	void *tag;
#ifdef RVC_OS_WIN
	HANDLE work_thread;
	HANDLE evt;
#else
	pthread_t work_thread;
	int evt[2];//pipe
#endif
	volatile int stop_flag;
	recv_info_t recv_info;
	send_info_t send_info;
	int screen_img_id;
	int photo_img_id;
	int b_primary_server;
	int remote_video_rtp_port;
	int remote_video_desc;
	char remote_client_id[32];
	int connected;
	char local_pwd[KEY_LEN];
	char remote_pwd[KEY_LEN];
	RC4_KEY local_key;
	RC4_KEY remote_key;
	LARGE_INTEGER last_remote_active_time;
	LARGE_INTEGER last_local_active_time;
	screen_decoder_session_t *dec_session;
};

typedef int (*lpfn_cryptionfun)(unsigned char *  out ,int outLen,const unsigned char * in,int  inLen);

//typedef int (*lpfn_decodestring)(unsigned char *  out ,int outLen,const unsigned char * in,int  inLen);
//typedef int (*lpfn_encodestring)(unsigned char *  out ,int outLen,const unsigned char * in,int  inLen);
//
//typedef int (*lpfn_decodestring_mobile)(unsigned char *  out ,int outLen,const unsigned char * in,int  inLen);
//typedef int (*lpfn_encodestring_mobile)(unsigned char *  out ,int outLen,const unsigned char * in,int  inLen);

static lpfn_cryptionfun decodestring = NULL;
static lpfn_cryptionfun encodestring = NULL;
static lpfn_cryptionfun decodestring_mobile = NULL;
static lpfn_cryptionfun encodestring_mobile = NULL;

static __inline unsigned int hash32_buf(const void *bf, size_t len, unsigned int hash)
{
	const unsigned char *s = (const unsigned char*)bf;

	while (len-- != 0)			/* "nemesi": k=257, r=r*257 */
		hash = hash * 257 + *s++;
	return (hash * 257);
}

static void generate_rand_key(char *key, int size)
{
	int i;
	srand(GetTickCount()*33);
	for (i = 0; i < size; ++i) {
		key[i] = (char)(rand() & 0xff);
	}
#ifdef RVC_OS_WIN
	srand(GetCurrentProcessId() * 33);
#else
	srand(getpid() * 33);
#endif
	for (i = 0; i < size; ++i) {
		key[i] ^= (char)(rand() & 0xff);
	}
}

static __inline unsigned int hash_key(const char *key, int size)
{
	return hash32_buf(key, size, 0);
}

static __inline int check_hash(char *key, int size, int hash_code)
{
	return hash32_buf(key, size, 0) == hash_code;
}

static __inline void GetTick(LARGE_INTEGER *last, LARGE_INTEGER *lt)
{
#ifdef RVC_OS_WIN
	DWORD dwNow = GetTickCount();
	if (last->LowPart > dwNow) {
		lt->LowPart = dwNow;
		lt->HighPart = last->HighPart + 1;
	} else {
		lt->LowPart = dwNow;
		lt->HighPart = last->HighPart;
	}
#else
	DWORD dwNow = GetTickCount();
	if (last->u.LowPart > dwNow) {
		lt->u.LowPart = dwNow;
		lt->u.HighPart = last->u.HighPart + 1;
	}
	else {
		lt->u.LowPart = dwNow;
		lt->u.HighPart = last->u.HighPart;
	}
#endif
}

BIZCHAN_API(int) bizchan_lib_init()
{
#ifdef RVC_OS_WIN
	WSADATA wsaData;
	return WSAStartup(0x0202, &wsaData);
#else
	return 0;
#endif
}

BIZCHAN_API(int) bizchan_lib_term()
{
#ifdef RVC_OS_WIN
	return WSACleanup();
#else
	return 0;
#endif
}

static int config_copy(bizchan_t* chan, const bizchan_config_t *src, bizchan_config_t *dst)
{
	memcpy(dst, src, sizeof(bizchan_config_t));
	dst->proxy_server = strdup(src->proxy_server);

	if (src->bak_proxy_server != NULL) {
		dst->bak_proxy_server = strdup(src->bak_proxy_server);
	}

	if (src->session_id != NULL) {
		dst->session_id = strdup(src->session_id);
	}
	if (src->agent_id != NULL) {
		dst->agent_id = strdup(src->agent_id);
	}
	if (src->client_id != NULL) {
		dst->client_id = strdup(src->client_id);
	}
	return 0;
}

static int config_check(const bizchan_config_t *config)
{
	if (!config->proxy_server)
		return -1;
	if (config->proxy_server_port <= 0 || config->proxy_server_port >= 0xffff)
		return -1;
	if (config->bak_proxy_server) {
		if (config->bak_proxy_server_port <= 0 || config->bak_proxy_server_port >= 0xffff)
			return -1;
	}
	if (!config->session_id)
		return -1;
	if (!config->agent_id)
		return -1;
	return 0;
}

static void config_free(bizchan_config_t *config)
{
	free(config->proxy_server);
	free(config->bak_proxy_server);
	free(config->session_id);
	free(config->agent_id);
	free(config->client_id);
}

static int callback_check(const bizchan_callback_t *cb)
{
	if (!cb->on_close)
		return -1;
	if (!cb->on_connect)
		return -1;
	if (!cb->on_recv_pkt)
		return -1;
	return 0;
}

static void invoke_on_connect(bizchan_t *chan, int error)
{
	if (error == 0)
		chan->cb.on_connect(chan, 
			error, 
			chan->b_primary_server ? chan->config.proxy_server : chan->config.bak_proxy_server,
			chan->remote_video_rtp_port, 
			chan->remote_video_desc, 
			chan->remote_client_id,
			chan->cb.user_data);
	else
		chan->cb.on_connect(chan, 
			error, 
			NULL, 
			0, 
			0, 
			NULL,
			chan->cb.user_data);
}




//static FILE *rx_log_fp = NULL;

static void invoke_on_recv_pkt(bizchan_t *chan, int type, int sub_type, int id, const char *pkt, int pkt_size)
{
	/*if (rx_log_fp == NULL) {
		rx_log_fp = fopen("c:\\rxlog.txt", "wt");
		fprintf(rx_log_fp, "===================\n");
	}
	{
		SYSTEMTIME st;
		GetLocalTime(&st);
		fprintf(rx_log_fp, "[%02d:%02d:%02d.%03d] type = %d, sub_type = %d, id = %d, pkt_size = %d, hash = %d\n", 
			st.wHour, st.wMinute, st.wSecond, st.wMilliseconds,
			type, sub_type, id, pkt_size, hash32_buf(pkt, pkt_size, 0));
		fflush(rx_log_fp);
	}*/

	if (type == ACM_TYPE_SRN) {
		int cat = ACM_SRN_CAT(sub_type);
		if (cat == ACM_SRN_ANS) {
			if (id == chan->screen_img_id) {
				if (pkt_size == 4) 
				{
					int err;
					memcpy(&err, pkt, 4);
					chan->cb.on_recv_screen(chan, id, err, 0, 0, NULL, 0, chan->cb.user_data);
				} 
				else 
				{
					int size = 0;
					int width, height;
					int rc;
					char *dec_buf = NULL;
					rc = screen_decoder_session_decode(chan->dec_session, pkt, pkt_size, &width, &height, dec_buf, &size);
					if (rc == 0) {
						dec_buf = (char*)malloc(size);
						rc = screen_decoder_session_decode(chan->dec_session, pkt, pkt_size, &width, &height, dec_buf, &size);
					}
					if (rc == 0) {
						chan->cb.on_recv_screen(chan, id, 0, width, height, dec_buf, size, chan->cb.user_data);
					} else {
						chan->cb.on_recv_screen(chan, id, rc, 0, 0, NULL, 0, chan->cb.user_data);
					}
					if (dec_buf)
						free(dec_buf);
				}
			}
		} else 
		{
			chan->cb.on_recv_pkt(chan, type, sub_type, id, pkt, pkt_size, chan->cb.user_data);
		}
	} else if (type == ACM_TYPE_PHT) {
		int cat = ACM_PHT_CAT(sub_type);
		if (cat == ACM_PHT_ANS)
		{
			if (id == chan->photo_img_id) {
				if (pkt_size == 4) {
					int err;
					memcpy(&err, pkt, 4);
					chan->cb.on_recv_photo(chan, id, err, 0, 0, NULL, 0, chan->cb.user_data);
				} else {
					jpeg2k_raw_image raw_image = {0};
					jpeg2k_coded_image codec_image = {0};
					int rc;
					codec_image.data = (unsigned char*)pkt;
					codec_image.len	 = pkt_size;
					rc = jpeg2k_decode(&raw_image, &codec_image);
					if (rc == 0) {
						chan->cb.on_recv_photo(chan, id, 0, raw_image.width, raw_image.height, (const char*)&raw_image.data[0], raw_image.len, chan->cb.user_data);
						jpeg2k_decode_free(&raw_image);
					} else {
						chan->cb.on_recv_photo(chan, id, rc, 0, 0, NULL, 0, chan->cb.user_data);
					}
				}
			}
		} else {
			chan->cb.on_recv_pkt(chan, type, sub_type, id, pkt, pkt_size, chan->cb.user_data);
		}
	} else if (type == ACM_TYPE_SYNC) {
		if (chan->winsync_on_recv_cb) {
			chan->winsync_on_recv_cb(sub_type, pkt, pkt_size, chan->winsync_user_data);
		} else {
			chan->cb.on_recv_pkt(chan, type, sub_type, id, pkt, pkt_size, chan->cb.user_data);
		}
	} else if (type == ACM_TYPE_MODE) {
		if (chan->mode_cb) {
			chan->mode_cb(TRUE, chan->winsync_user_data);
		}
		chan->cb.on_recv_pkt(chan, type, sub_type, id, pkt, pkt_size, chan->cb.user_data);
	} else if (type == ACM_TYPE_PING) {
		//..... remote ping, nothing todo
	} else {
		chan->cb.on_recv_pkt(chan, type, sub_type, id, pkt, pkt_size, chan->cb.user_data);
	}
}

static void invoke_on_close(bizchan_t *chan)
{
	if (chan->cb.on_close) {
		chan->cb.on_close(chan, chan->cb.user_data);
	}
}

static void invoke_on_destroy(bizchan_t *chan)
{
	if (chan->cb.on_destroy) {
		chan->cb.on_destroy(chan, chan->cb.user_data);
	}
}

static FILE *fp = NULL;

static int on_recv2(bizchan_t *chan, SOCKET conn)
{
	int n;
	char buf[0x10000+sizeof(acm_hdr)];
	if (fp == NULL) {
#ifdef RVC_OS_WIN
		fp = fopen ("g:\\rx.dat", "wb");
#else
		fp = fopen("/home/rx.dat", "wb");
#endif
	}
	
	do {
		n = recv(conn, buf, sizeof(buf), 0);
		if (n > 0) {
			fwrite(buf, 1, n, fp);
			fflush(fp);
		}
	} while (n > 0);
#ifdef RVC_OS_WIN
	return (n == 0 || (GetLastError() == WSAEWOULDBLOCK)) ? 0 : -1;
#else
	return (n == 0 || (errno == EWOULDBLOCK) || (errno == EINTR)) ? 0 : -1;
#endif
}

static int on_recv(bizchan_t *chan, SOCKET conn)
{
	char buffer[128];
	recv_info_t *ri = &chan->recv_info;
	int n;
	int result = 0;

	do {
		n = recv(conn, ri->buf+ri->offset, ri->buf_len-ri->offset, 0);
		snprintf(buffer, 128, "on_recv n:%d, ri->offset:%d, ri->buf_len:%d", n, ri->offset, ri->buf_len);
		chan->cb.dbg(chan, 0, buffer, chan->cb.user_data);
		if (n > 0) {
			int i = 0;
			ri->offset += n;
			GetTick(&chan->last_remote_active_time, &chan->last_remote_active_time);
			while (ri->offset-i >= sizeof(acm_hdr)) {
				acm_hdr *hdr = (acm_hdr*)&ri->buf[i];
				if (hdr->length == 0) {
					printf("broken");
				}
				if (ri->offset-i >= hdr->length+sizeof(acm_hdr)) {
					if (hdr->encrypt) {
						char *dec_buf = (char*)malloc(hdr->length);
						RC4_set_key(&chan->remote_key, sizeof(chan->remote_pwd), (unsigned char*)chan->remote_pwd);
						RC4(&chan->remote_key, hdr->length, (const unsigned char*)(&hdr->data[0]), (unsigned char*)dec_buf);
						if (hdr->compress) {
							int len = (int)qlz_size_decompressed(dec_buf);
							char *unzip_buf = (char*)malloc(len);
							len = qlz_decompress(dec_buf, unzip_buf, &ri->decompress_state);
							if (check_hash(unzip_buf, len, hdr->hash)) {
								invoke_on_recv_pkt(chan, hdr->type, hdr->sub_type, hdr->id, unzip_buf, len);
							} else {
								OutputDebugStringA("pkt hash failed!\n");
								chan->cb.dbg(chan, 0, "pkt hash failed! 1", chan->cb.user_data);
							}
							free(unzip_buf);
						} else {
							if (check_hash(dec_buf, hdr->length, hdr->hash)) {
								invoke_on_recv_pkt(chan, hdr->type, hdr->sub_type, hdr->id, dec_buf, hdr->length);
							} else {
								OutputDebugStringA("pkt hash failed!\n");
								chan->cb.dbg(chan, 0, "pkt hash failed! 2", chan->cb.user_data);
							}
						}
						free(dec_buf);
					} else {
						if (hdr->compress) {
							int len = (int)qlz_size_decompressed((const char *)&hdr->data[0]);
							char *unzip_buf = (char*)malloc(len);
							len = qlz_decompress((const char*)&hdr->data[0], unzip_buf, &ri->decompress_state);
							if (check_hash(unzip_buf, len, hdr->hash)) {
								invoke_on_recv_pkt(chan, hdr->type, hdr->sub_type, hdr->id, unzip_buf, len);
							} else {
								OutputDebugStringA("pkt hash failed!\n");
								chan->cb.dbg(chan, 0, "pkt hash failed! 3", chan->cb.user_data);
							}
							free(unzip_buf);
						} else {
							if (check_hash((char*)&hdr->data[0], hdr->length, hdr->hash)) {
								invoke_on_recv_pkt(chan, hdr->type, hdr->sub_type, hdr->id, (const char*)&hdr->data[0], hdr->length);
							} else {
								OutputDebugStringA("pkt hash failed!\n");
								chan->cb.dbg(chan, 0, "pkt hash failed! 4", chan->cb.user_data);
							}
						}
					}
					i += hdr->length+sizeof(acm_hdr);
				} else {
					break;
				}
			}
			if (i != ri->offset) {
				memmove(&ri->buf[0], &ri->buf[i], ri->offset-i);
			}
			ri->offset -= i;
			if (ri->offset == ri->buf_len) { // double large
				ri->buf_len = 2 * ri->buf_len;
				ri->buf = (char *)realloc(ri->buf, ri->buf_len);
			}
		}
	} while (n > 0);
#ifdef RVC_OS_WIN
	return (n == 0 || (GetLastError() == WSAEWOULDBLOCK)) ? 0 : -1;
#else
	//recv ==0,对端断开
	if (n == 0) {
		result = -1;
	}
	else {
		result = (n < 0 || (errno == EWOULDBLOCK) || (errno == EINTR) || (errno == EAGAIN)) ? 0 : -1;
	}
	snprintf(buffer, 128, "on_recv n:%d, result:%d, errno:%d", n, result, errno);
	chan->cb.dbg(chan, 0, buffer, chan->cb.user_data);
	return result;
#endif
}

static int on_send(bizchan_t *chan, SOCKET conn)
{
	char buffer[128];
	send_info_t *si = &chan->send_info;
	int n = 1;//默认为1
	int result = -1; 
	
#ifdef RVC_OS_WIN
	EnterCriticalSection(&si->lock);
#else
	pthread_mutex_lock(&si->lock);
#endif
	if (!ListEntry_IsEmpty(&si->send_list)) {
		do {
			send_buf_node *t = CONTAINING_RECORD(ListEntry_GetHead(&si->send_list), send_buf_node, entry);
			if (t->need_encrypt) {
				acm_hdr *hdr = (acm_hdr*)t->buf;
				char *enc_buf = (char*)malloc(t->left+sizeof(acm_hdr));
				RC4_set_key(&chan->local_key, sizeof(chan->local_pwd), (unsigned char*)chan->local_pwd);
				RC4(&chan->local_key, t->left, (const unsigned char*)(&hdr->data[0]), (unsigned char*)(enc_buf+sizeof(acm_hdr)));
				memcpy(enc_buf, hdr, sizeof(acm_hdr));
				free(t->buf);
				t->buf = enc_buf;
				t->need_encrypt = 0;
			}
			n = send(conn, t->buf+t->sended, t->left, 0);
			snprintf(buffer, 128, "on_send n:%d, t->sended:%d, t->left:%d", n, t->sended, t->left);
			chan->cb.dbg(chan, 0, buffer, chan->cb.user_data);
			if (n > 0) {
				char tmp[32];
				sprintf(tmp, "send out %d bytes!\n", n);
				OutputDebugStringA(tmp);
				t->left -= n;
				t->sended += n;
				if (t->left == 0) {
					ListEntry_DeleteNode(&t->entry);
					free(t->buf);
					free(t);
				}
				GetTick(&chan->last_local_active_time, &chan->last_local_active_time);
			}
		} while (n > 0 && !ListEntry_IsEmpty(&si->send_list));
	}
#ifdef RVC_OS_WIN
	LeaveCriticalSection(&si->lock);
	return (n >= 0 || (GetLastError() == WSAEWOULDBLOCK)) ? 0 : -1;
#else
	pthread_mutex_unlock(&si->lock);
	if (n > 0) {
		result = 0;
	}
	else if (n < 0){
		if ((errno == EWOULDBLOCK) || (errno == EINTR) || (errno == EAGAIN)) {
			result = 0;
		}
	}
	return result;
#endif
}

static void on_close(bizchan_t *chan)
{
	invoke_on_close(chan);
}

#ifdef RVC_OS_WIN
static int prepare_socket(SOCKET s, HANDLE evt)
{
	BOOL opt = TRUE;
	int rc;

	rc = setsockopt(s, IPPROTO_TCP, TCP_NODELAY, (char*)&opt, sizeof(opt));
	opt = TRUE;
	if (rc == 0)
		rc = setsockopt(s, SOL_SOCKET, SO_DONTLINGER, (char*)&opt, sizeof(opt));
	if (rc == 0)
		rc = WSAEventSelect(s, evt, FD_CONNECT | FD_ACCEPT | FD_CLOSE | FD_READ | FD_WRITE);

	return rc;
}

static void dump_exception(PEXCEPTION_POINTERS ExceptionInfo)
{
	char tmp[MAX_PATH];
	HANDLE hDumpFile;
	sprintf(tmp, ".\\bizchan_%d.dmp", GetCurrentProcessId());
	hDumpFile = CreateFileA( tmp, GENERIC_READ | GENERIC_WRITE, 
		0, NULL, CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL );
	if( ( hDumpFile != NULL ) && ( hDumpFile != INVALID_HANDLE_VALUE ) ) 
	{
		MINIDUMP_EXCEPTION_INFORMATION mdei; 
		MINIDUMP_TYPE mdt;

		mdei.ThreadId           = GetCurrentThreadId(); 
		mdei.ExceptionPointers  = ExceptionInfo; 
		mdei.ClientPointers     = FALSE; 

		mdt       = MiniDumpWithFullMemory;

		MiniDumpWriteDump( GetCurrentProcess(), GetCurrentProcessId(), 
			hDumpFile, mdt, (ExceptionInfo != 0) ? &mdei : 0, 0, 0 ); 

		CloseHandle( hDumpFile ); 
	}
}
#endif

 #ifdef RVC_OS_WIN
static void process(bizchan_t *chan)
{
	SOCKET conn = INVALID_SOCKET;
	struct sockaddr_in addr = {0};
	int rc;
	HANDLE evts[2] = {chan->evt, NULL};

	evts[1] = WSACreateEvent();
	// try connect to primary proxy server

	addr.sin_family = AF_INET;
	addr.sin_port = htons(chan->config.proxy_server_port);
	addr.sin_addr.s_addr = inet_addr(chan->config.proxy_server);
	chan->b_primary_server = TRUE;

	conn = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	if (conn == INVALID_SOCKET)
		goto on_error;
	rc = prepare_socket(conn, evts[1]);
	if (rc != 0)
		goto on_error;

	rc = connect(conn, (struct sockaddr*)&addr, sizeof(addr));
	if (rc == -1 && WSAGetLastError() == WSAEWOULDBLOCK) {
		rc = 0;		
	}

	if (rc == -1 && chan->config.bak_proxy_server && strlen(chan->config.bak_proxy_server)) { // try connect to back proxy server
		closesocket(conn);
		conn = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
		if (conn == INVALID_SOCKET)
			goto on_error;
		WSAResetEvent(evts[1]);
		rc = prepare_socket(conn, evts[1]);
		if (rc != 0)
			goto on_error;
		addr.sin_port = htons(chan->config.bak_proxy_server_port);
		addr.sin_addr.s_addr = inet_addr(chan->config.bak_proxy_server);
		chan->b_primary_server = 0;
		rc = connect(conn, (struct sockaddr*)&addr, sizeof(addr));
		if (rc == -1 && WSAGetLastError() == WSAEWOULDBLOCK) {
			rc = 0;		
		}
	}

	if (rc == -1)
		goto on_error;

	if (rc == 0)
	{
		proxy_ack_hdr ack_hdr;
		int ack_hdr_recv_bytes = 0;
		lpfn_cryptionfun encodefun = encodestring;
		lpfn_cryptionfun decodefun = decodestring;
		if (1 == chan->config.crypt_type){
			encodefun = encodestring_mobile;
			decodefun = decodestring_mobile;
		}
		while (!chan->stop_flag && !chan->connected) 
		{ // wait until connected
			DWORD dwRet = WaitForMultipleObjects(2, evts, FALSE, MAX_TIMEOUT);
			if (dwRet == WAIT_OBJECT_0) 
			{
				WSAResetEvent(evts[0]);
			} else if (dwRet == WAIT_OBJECT_0+1) 
			{
				int error = 0;
				WSANETWORKEVENTS netevents;
				if (WSAEnumNetworkEvents(conn, evts[1], &netevents) != SOCKET_ERROR) 
				{
					if (netevents.lNetworkEvents & FD_CONNECT) 
					{
						if (netevents.iErrorCode[FD_CONNECT_BIT])
							error = netevents.iErrorCode[FD_CONNECT_BIT];
					}
					if (error == 0) 
					{
						if (netevents.lNetworkEvents & FD_WRITE) 
						{
							error = netevents.iErrorCode[FD_WRITE_BIT];
							if (error == 0) {
								proxy_hdr hdr = {0};
								hdr.tag[0] = 'A';
								hdr.tag[1] = 'C';
								hdr.tag[2] = 'M';
								hdr.version = ACM_PROTOCOL_VERSION;
								generate_rand_key(chan->local_pwd, sizeof(chan->local_pwd));
								encodefun((unsigned char*)&hdr.encrypt_key[0], sizeof(hdr.encrypt_key), (unsigned char*)&chan->local_pwd[0], sizeof(chan->local_pwd));
								hdr.encrypt_keyhash = hash_key(chan->local_pwd, sizeof(chan->local_pwd));
								memset(hdr.callee_id, ' ', sizeof(hdr.callee_id));
								memset(hdr.caller_id, ' ', sizeof(hdr.caller_id));
								memset(hdr.client_id, ' ', sizeof(hdr.client_id));
								strncpy(&hdr.callee_id[0], chan->config.agent_id, sizeof(hdr.callee_id)-1);
								strncpy(&hdr.caller_id[0], chan->config.session_id, sizeof(hdr.caller_id)-1);
								if (chan->config.client_id) {
									strncpy(&hdr.client_id[0], chan->config.client_id, sizeof(hdr.client_id)-1);
								} else {
									hdr.client_id[0] = 0;
								}
								hdr.rtp_port = chan->config.video.rtp_port;
								hdr.media_desc = chan->config.video.desc;
								if (send(conn, (char*)&hdr, sizeof(hdr), 0) != sizeof(hdr)) {
									error = -1;
								}
							}
						}
					}
					if (error == 0) 
					{
						if (netevents.lNetworkEvents & FD_CLOSE)
							error = -1;
					}
					if (error == 0) 
					{
						if (netevents.lNetworkEvents & FD_READ) 
						{
							error = netevents.iErrorCode[FD_READ_BIT];
							if (error == 0) 
							{
								int t;
								do 
								{
									t = recv(conn, (char*)&ack_hdr + ack_hdr_recv_bytes, sizeof(ack_hdr)-ack_hdr_recv_bytes, 0);
									if (t > 0) {
										ack_hdr_recv_bytes += t;
										if (ack_hdr_recv_bytes == sizeof(ack_hdr)) {
											decodefun((unsigned char*)&chan->remote_pwd[0], sizeof(chan->remote_pwd), (unsigned char*)&ack_hdr.encrypt_key[0], sizeof(ack_hdr.encrypt_key));
											if (check_hash(chan->remote_pwd, sizeof(chan->remote_pwd), ack_hdr.encrypt_keyhash)) {
												chan->remote_video_rtp_port = ack_hdr.rtp_port;
												chan->remote_video_desc = ack_hdr.media_desc;
												memcpy(chan->remote_client_id, ack_hdr.client_id, sizeof(chan->remote_client_id));
												chan->connected = TRUE;
												break;
											} else {
												error = -1;
											}
										}
									}
								} 
								while (t > 0 && error == 0);
								if (t <= 0 && (WSAGetLastError() != WSAEWOULDBLOCK))
									error = -1;
							}
						}
					}
					if (error != 0)
						goto on_error;
				}
			} else {
				goto on_error;
			}
		}
	}

	assert(conn != INVALID_SOCKET);

	if (chan->connected) 
	{
		invoke_on_connect(chan, 0);

		chan->recv_info.offset = 0;
		rc = on_recv(chan, conn);
		if (rc) 
		{
			on_close(chan);
			goto on_error;
		}

		while (!chan->stop_flag) 
		{
			DWORD dwRet = WaitForMultipleObjects(2, evts, FALSE, 1000);
			if (dwRet == WAIT_OBJECT_0) 
			{
				WSAResetEvent(chan->evt);
				rc = on_send(chan, conn);
				if (rc != 0)
					break;
			} else if (dwRet == WAIT_OBJECT_0+1) 
			{
				WSANETWORKEVENTS netevents;
				if (WSAEnumNetworkEvents(conn, evts[1], &netevents) != SOCKET_ERROR) 
				{
					if (netevents.lNetworkEvents & FD_READ) 
					{
						rc = on_recv(chan, conn);
					}
					if (netevents.lNetworkEvents & FD_WRITE) {
						rc = on_send(chan, conn);
					}
					if (rc || (netevents.lNetworkEvents & FD_CLOSE)) {
						break;
					}
				}
			} else if (dwRet == WAIT_TIMEOUT) {
				LARGE_INTEGER now;
				GetTick(&chan->last_remote_active_time, &now);
				if (now.QuadPart - chan->last_remote_active_time.QuadPart >= PING_INTERVAL) {
					GetTick(&chan->last_local_active_time, &now);
					if (now.QuadPart - chan->last_local_active_time.QuadPart >= PING_INTERVAL) {
						bizchan_post_pkt(chan, ACM_TYPE_PING, 0, 0, 0, 0, NULL, 0);
					}
				}
			} else {
				goto on_error;
			}
		}

		on_close(chan);
	}

on_error:
	OutputDebugStringA("on_error");
	if (!chan->connected) {
		invoke_on_connect(chan, -1); // connect failed!
	} else {
		//....
		OutputDebugStringA("connected, and error!");
	}

	if (conn != INVALID_SOCKET)
		closesocket(conn);

	WSACloseEvent(evts[1]);
}

#else

static void do_shake_send(bizchan_t* chan, SOCKET conn, int* error) {
	proxy_hdr hdr = { 0 };
	proxy_ack_hdr ack_hdr;
	int ack_hdr_recv_bytes = 0;
	lpfn_cryptionfun encodefun = encodestring;
	lpfn_cryptionfun decodefun = decodestring;

	if (1 == chan->config.crypt_type) {
		encodefun = encodestring_mobile;
		decodefun = decodestring_mobile;
	}

	hdr.tag[0] = 'A';
	hdr.tag[1] = 'C';
	hdr.tag[2] = 'M';
	hdr.version = ACM_PROTOCOL_VERSION;
	generate_rand_key(chan->local_pwd, sizeof(chan->local_pwd));
	encodefun((unsigned char*)&hdr.encrypt_key[0], sizeof(hdr.encrypt_key), (unsigned char*)&chan->local_pwd[0], sizeof(chan->local_pwd));

	hdr.encrypt_keyhash = hash_key(chan->local_pwd, sizeof(chan->local_pwd));
	memset(hdr.callee_id, ' ', sizeof(hdr.callee_id));
	memset(hdr.caller_id, ' ', sizeof(hdr.caller_id));
	memset(hdr.client_id, ' ', sizeof(hdr.client_id));
	strncpy(&hdr.callee_id[0], chan->config.agent_id, sizeof(hdr.callee_id) - 1);
	strncpy(&hdr.caller_id[0], chan->config.session_id, sizeof(hdr.caller_id) - 1);
	if (chan->config.client_id) {
		strncpy(&hdr.client_id[0], chan->config.client_id, sizeof(hdr.client_id) - 1);
	}
	else {
		hdr.client_id[0] = 0;
	}
	hdr.rtp_port = chan->config.video.rtp_port;
	hdr.media_desc = chan->config.video.desc;
	if (send(conn, (char*)&hdr, sizeof(hdr), 0) != sizeof(hdr)) {
		*error = -1;
	}
}

static void do_shake_recv(bizchan_t* chan, SOCKET conn, int* error) {
	int t;
	proxy_ack_hdr ack_hdr;
	int ack_hdr_recv_bytes = 0;
	lpfn_cryptionfun encodefun = encodestring;
	lpfn_cryptionfun decodefun = decodestring;
	char buffer[128];

	if (1 == chan->config.crypt_type) {
		encodefun = encodestring_mobile;
		decodefun = decodestring_mobile;
	}

	do
	{
		t = recv(conn, (char*)&ack_hdr + ack_hdr_recv_bytes, sizeof(ack_hdr) - ack_hdr_recv_bytes, 0);
		if (t > 0) {
			snprintf(buffer, 128, "do_shake_recv t:%d, ack_hdr_recv_bytes:%d, sizeof(ack_hdr):%d", t, ack_hdr_recv_bytes, sizeof(ack_hdr));
			chan->cb.dbg(chan, 0, buffer, chan->cb.user_data);
			ack_hdr_recv_bytes += t;
			if (ack_hdr_recv_bytes == sizeof(ack_hdr)) {
				decodefun((unsigned char*)&chan->remote_pwd[0], sizeof(chan->remote_pwd), (unsigned char*)&ack_hdr.encrypt_key[0], sizeof(ack_hdr.encrypt_key));
				chan->cb.dbg(chan, 0, "do_shake_recv remote_pwd:", chan->cb.user_data);
				chan->cb.dbg(chan, 0, chan->remote_pwd, chan->cb.user_data);
				if (check_hash(chan->remote_pwd, sizeof(chan->remote_pwd), ack_hdr.encrypt_keyhash)) {
					chan->remote_video_rtp_port = ack_hdr.rtp_port;
					chan->remote_video_desc = ack_hdr.media_desc;
					memcpy(chan->remote_client_id, ack_hdr.client_id, sizeof(chan->remote_client_id));
					chan->connected = TRUE;
					break;
				}
				else {
					*error = -1;
					chan->cb.dbg(chan, 0, "do_shake_recv check_hash failed", chan->cb.user_data);
				}
			}
		}
	} while (t > 0 && *error == 0);
#ifdef RVC_OS_WIN
	if (t <= 0 && (WSAGetLastError() != WSAEWOULDBLOCK))
		*error = -1;
#else
	if (t <= 0 && ((errno != EWOULDBLOCK) || (errno != EINTR) || (errno != EAGAIN))) {
		*error = -1;
		snprintf(buffer, 128, "do_shake_recv errno:%d", errno);
		chan->cb.dbg(chan, 0, buffer, chan->cb.user_data);
		chan->cb.dbg(chan, 0, "do_shake_recv t<=0 failed", chan->cb.user_data);
	}
#endif
}

static void  build_fd_sets(SOCKET conn, int evt, fd_set* read_fds, fd_set* write_fds, fd_set* except_fds)
{
	FD_ZERO(read_fds);
	FD_SET(conn, read_fds);
	FD_SET(evt, read_fds);
	FD_ZERO(write_fds);
	//FD_SET(conn, write_fds);
	FD_ZERO(except_fds);
	FD_SET(conn, except_fds);
	FD_SET(evt, except_fds);
}

static int set_socket_noblock(SOCKET socket) {
#ifdef RVC_OS_WIN
	{
		//windows将socket设置成非阻塞的方式  
		unsigned long on = 1;
		if (ioctlsocket(socket, FIONBIO, &on) < 0) {
			return -1;
		}
	}
#else
	{
		//linux将socket设置成非阻塞的方式  
		//将新socket设置为non-blocking  
		int oldflag = fcntl(socket, F_GETFL, 0);
		int newflag = oldflag | O_NONBLOCK;
		if (fcntl(socket, F_SETFL, newflag) == -1) {
			return -1;
		}
	}
#endif
	return 0;
}


SOCKET connect_server(bizchan_t* chan, char* server, int port, int timeout)
{
	SOCKET m_hSocket;
	struct sockaddr_in addrSrv = { 0 };
	//fd_set writeset;

	m_hSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	if (m_hSocket == INVALID_SOCKET)
		return INVALID_SOCKET;

	addrSrv.sin_family = AF_INET;
	addrSrv.sin_addr.s_addr = inet_addr(server);
	addrSrv.sin_port = htons(port);
	int ret = connect(m_hSocket, (struct sockaddr*)&addrSrv, sizeof(addrSrv));
	if (ret == 0) {
		return m_hSocket;
	}
	else {
		chan->cb.dbg(chan, 0, "connect_server failed", chan->cb.user_data);
#ifdef RVC_OS_WIN
		closesocket(m_hSocket);
#else
		close(m_hSocket);
#endif
		return INVALID_SOCKET;
	}

/*
#ifdef RVC_OS_WIN
	//windows下检测WSAEWOULDBLOCK  
	if (ret < 0 && WSAGetLastError() != WSAEWOULDBLOCK) {
		closesocket(m_hSocket);
		return INVALID_SOCKET;
	}
#else
	//linux下需要检测EINPROGRESS和EINTR  
	if (ret < 0 && (errno != EINPROGRESS || errno != EINTR)) {
		close(m_hSocket);
		return INVALID_SOCKET;
	}
#endif

	chan->cb.dbg(chan, 0, "processLinux connect_server 5", chan->cb.user_data);
	FD_ZERO(&writeset);
	FD_SET(m_hSocket, &writeset);
	struct timeval tv;
	tv.tv_sec = timeout;
	//可以利用tv_usec做更小精度的超时设置  
	tv.tv_usec = 0;
	if (select(m_hSocket + 1, NULL, &writeset, NULL, &tv) != 1) {
#ifdef RVC_OS_WIN
		closesocket(m_hSocket);
#else
		close(m_hSocket);
#endif
		chan->cb.dbg(chan, 0, "processLinux connect_server 6", chan->cb.user_data);
		return INVALID_SOCKET;
	}

	return m_hSocket;*/
}

static void process(bizchan_t* chan) {
	int error = 0;
	SOCKET conn = connect_server(chan, chan->config.proxy_server, chan->config.proxy_server_port, 5);
	if (conn == INVALID_SOCKET && chan->config.bak_proxy_server && strlen(chan->config.bak_proxy_server)) { // try connect to back proxy server
		conn = connect_server(chan, chan->config.bak_proxy_server, chan->config.bak_proxy_server_port, 5);
	}

	if (conn == INVALID_SOCKET) {
		error = -1;
	}
	else {
		do_shake_send(chan, conn, &error);
		if (error != 0) {
			chan->cb.dbg(chan, 0, "do_shake_send failed", chan->cb.user_data);
			goto on_error;
		}
		do_shake_recv(chan, conn, &error);
		if (error != 0) {
			chan->cb.dbg(chan, 0, "do_shake_recv failed", chan->cb.user_data);
			goto on_error;
		}
	}

	if (chan->connected)
	{
		char buffer[128];
		int rc;
		fd_set read_fds;
		fd_set write_fds;
		fd_set except_fds;
		int maxfd = max(conn, chan->evt[0]);
		rc = set_socket_noblock(conn);
		if (rc)
		{
			on_close(chan);
			chan->cb.dbg(chan, 2, "set_socket_noblock failed", chan->cb.user_data);
			goto on_error;
		}
		invoke_on_connect(chan, 0);
		chan->recv_info.offset = 0;
		/*rc = on_recv(chan, conn);
		if (rc)
		{
			on_close(chan);
			chan->cb.dbg(chan, 2, "on_recv failed", chan->cb.user_data);
			goto on_error;
		}*/
		while (!chan->stop_flag)
		{
			fd_set read_fds;
			fd_set write_fds;
			fd_set except_fds;
			int maxfd = max(conn, chan->evt[0]);
			struct timeval tv;
			tv.tv_sec = 1;
			//可以利用tv_usec做更小精度的超时设置  
			tv.tv_usec = 0;
			// Select() updates fd_set's, so we need to build fd_set's before each select()call. 
			build_fd_sets(conn, chan->evt[0], &read_fds, &write_fds, &except_fds);
			int activity = select(maxfd + 1, &read_fds, &write_fds, &except_fds, &tv);
			if (activity == -1) {
				perror("select()");
				chan->cb.dbg(chan, 0, "chan select() error.", chan->cb.user_data);
				goto on_error;
			}
			else if (activity == 0) {
				// timeout 
				printf("select() returns 0, timeout.\n");
				//chan->cb.dbg(chan, 0, "select() returns 0, timeout.", chan->cb.user_data);
				LARGE_INTEGER now;
				GetTick(&chan->last_remote_active_time, &now);
				if (now.QuadPart - chan->last_remote_active_time.QuadPart >= PING_INTERVAL) {
					GetTick(&chan->last_local_active_time, &now);
					if (now.QuadPart - chan->last_local_active_time.QuadPart >= PING_INTERVAL) {
						bizchan_post_pkt(chan, ACM_TYPE_PING, 0, 0, 0, 0, NULL, 0);
					}
				}
			}else {
				/* All fd_set's should be checked. */
				if (FD_ISSET(conn, &read_fds)) {
					rc = on_recv(chan, conn);
					if (rc != 0)
						break;
				}
				if (FD_ISSET(chan->evt[0], &read_fds)) {
					//读取管道数据并丢弃，管道主要用于唤醒select
					char pipe_read_buffer[4];
					int n = read(chan->evt[0], pipe_read_buffer, 4);
					if (n >= 0) {
						rc = on_send(chan, conn);
						if (rc != 0)
							break;
					}
					else {
						chan->cb.dbg(chan, 0, "pipe read < 0", chan->cb.user_data);
					}
				}
				if (FD_ISSET(conn, &except_fds)) {
					chan->cb.dbg(chan, 0, "select() except_fds set.", chan->cb.user_data);
					error = -1;
				}
				if (error != 0) {
					goto on_error;
				}
			}
		}
		on_close(chan);
	}

on_error:
	OutputDebugStringA("on_error");
	chan->cb.dbg(chan, 0, "on_error", chan->cb.user_data);

	if (!chan->connected) {
		invoke_on_connect(chan, -1); // connect failed!
	}
	else {
		//....
		OutputDebugStringA("connected, and error!");
		chan->cb.dbg(chan, 0, "connected, and error!", chan->cb.user_data);
	}

	if (conn != INVALID_SOCKET) {
#ifdef RVC_OS_WIN
		closesocket(conn);
#else
		close(conn);
#endif
	}
}
#endif


#ifdef RVC_OS_WIN
static unsigned int __stdcall work_proc(LPVOID arg)
#else
static void* work_proc(void* arg)
#endif
{
	bizchan_t *chan = (bizchan_t *)arg;
#ifdef RVC_OS_WIN
	__try {
		process(chan);
	} __except(dump_exception(GetExceptionInformation()), EXCEPTION_EXECUTE_HANDLER) {
		//....
	}
#else
	chan->cb.dbg(chan, 0, "work_proc start", chan->cb.user_data);
	process(chan);
#endif
	return 0;
}

#ifdef RVC_OS_WIN
static int init_decode_func(const bizchan_config_t* config, const bizchan_callback_t* cb)
{
	int ret = -1;
	if (!encodestring) {
		HMODULE hInst = LoadLibraryA("acmstrenc.dll");
		if (hInst) {
			encodestring = (lpfn_cryptionfun)GetProcAddress(hInst, "encodestring");
		}
		if (!encodestring)
			return ret;
	}
	if (!decodestring) {
		HMODULE hInst = LoadLibraryA("acmstrdec.dll");
		if (hInst) {
			decodestring = (lpfn_cryptionfun)GetProcAddress(hInst, "decodestring");
		}
		if (!decodestring)
			return ret;
	}

	if (!encodestring_mobile) {
		HMODULE hInst = LoadLibraryA("acmstrenc_mobile.dll");
		if (hInst) {
			encodestring_mobile = (lpfn_cryptionfun)GetProcAddress(hInst, "encodestring");
		}
		if (!encodestring_mobile)
			return ret;
	}
	if (!decodestring_mobile) {
		HMODULE hInst = LoadLibraryA("acmstrdec_mobile.dll");
		if (hInst) {
			decodestring_mobile = (lpfn_cryptionfun)GetProcAddress(hInst, "decodestring");
		}
		if (!decodestring_mobile)
			return ret;
	}

	if (encodestring && decodestring && encodestring_mobile && decodestring_mobile){
		ret = 0;
	}
	return ret;
}

#else
static int init_decode_func(const bizchan_config_t* config, const bizchan_callback_t* cb)
{
	int ret = -1;
	if (!encodestring) {
		HMODULE hInst = LoadLibraryA("libacmstrenc.so");
		if (hInst) {
			encodestring = (lpfn_cryptionfun)GetProcAddress(hInst, "encodestring");
		}
		else {
			cb->dbg(NULL, 2, "libacmstrenc LoadLibraryA failed", cb->user_data);
		}
		if (!encodestring) {
			cb->dbg(NULL, 2, "libacmstrenc GetProcAddress failed", cb->user_data);
			return ret;
		}

	}
	if (!decodestring) {
		HMODULE hInst = LoadLibraryA("libacmstrdec.so");
		if (hInst) {
			decodestring = (lpfn_cryptionfun)GetProcAddress(hInst, "decodestring");
		}
		if (!decodestring)
			return ret;
	}

	if (1 == config->crypt_type) {
		if (!encodestring_mobile) {
			HMODULE hInst = LoadLibraryA("libacmstrenc_mobile.so");
			if (hInst) {
				encodestring_mobile = (lpfn_cryptionfun)GetProcAddress(hInst, "encodestring");
			}
			if (!encodestring_mobile)
				return ret;
		}
		if (!decodestring_mobile) {
			HMODULE hInst = LoadLibraryA("libacmstrdec_mobile.so");
			if (hInst) {
				decodestring_mobile = (lpfn_cryptionfun)GetProcAddress(hInst, "decodestring");
			}
			if (!decodestring_mobile)
				return ret;
		}
	}

	if (encodestring && decodestring) {
		ret = 0;
	}

	return ret;
}
#endif

BIZCHAN_API(int) bizchan_create(const bizchan_config_t *config, const bizchan_callback_t *cb, bizchan_t **p_chan)
{
	bizchan_t *chan = NULL;

	if (-1 == init_decode_func(config, cb)){
		cb->dbg(chan, 2, "init_decode_func failed", cb->user_data);
		return -1;
	}

	if (!config || !p_chan) {
		cb->dbg(chan, 2, "config or chan is null, failed", cb->user_data);
		return -1;
	}

	if (config_check(config) != 0) {
		cb->dbg(chan, 2, "config_check failed", cb->user_data);
		return -1;
	}

	if (callback_check(cb) != 0) {
		cb->dbg(chan, 2, "callback_check failed", cb->user_data);
		return -1;
	}

	chan = (bizchan_t*)malloc(sizeof(bizchan_t));
	if (!chan)
		goto on_error;
	memset(chan, 0, sizeof(bizchan_t));
	chan->recv_info.buf_len = DEFAULT_RX_BUF_SIZE;
	chan->recv_info.buf = (char*)malloc(chan->recv_info.buf_len);
	if (!chan->recv_info.buf)
		goto on_error;
	memcpy(&chan->cb, cb, sizeof(bizchan_callback_t));
	if (config_copy(chan, config, &chan->config) != 0)
		goto on_error;
	ListEntry_InitHead(&chan->send_info.send_list);
#ifdef RVC_OS_WIN
	InitializeCriticalSection(&chan->send_info.lock);
#else
	pthread_mutexattr_init(&chan->send_info.attr);
	pthread_mutexattr_settype(&chan->send_info.attr, PTHREAD_MUTEX_RECURSIVE);
	pthread_mutex_init(&chan->send_info.lock, &chan->send_info.attr);
#endif
	screen_decoder_session_create(&chan->dec_session);
#ifdef RVC_OS_WIN
	chan->evt = WSACreateEvent();
	if (!chan->evt)
		goto on_error;
#else
	pipe(chan->evt);
#endif

	*p_chan = chan;
	cb->dbg(chan, 0, "bizchan_create success", cb->user_data);

	return 0;

on_error:
	cb->dbg(chan, 2, "bizchan_create on_error", cb->user_data);
	bizchan_destroy(chan);
	return -1;
}

BIZCHAN_API(int) bizchan_winsync_set_cb(bizchan_t *chan, OnRecvPacket pkt_cb, OnMode mode_cb, void *user_data)
{
	chan->winsync_on_recv_cb = pkt_cb;
	chan->mode_cb = mode_cb;
	chan->winsync_user_data = user_data;
	return 0;
}

BIZCHAN_API(void) bizchan_destroy(bizchan_t *chan)
{
	if (chan) {
#ifdef RVC_OS_WIN
		if (chan->evt) {
			WSACloseEvent(chan->evt);
			chan->evt = NULL;
		}
#else
		close(chan->evt[0]);
		close(chan->evt[1]);
#endif
		invoke_on_destroy(chan);
#ifdef RVC_OS_WIN
		assert(chan->work_thread == NULL);
#endif
		config_free(&chan->config);
		if (chan->recv_info.buf)
			free(chan->recv_info.buf);
		if (chan->recv_info.unzip_buf)
			free(chan->recv_info.unzip_buf);
		while (!ListEntry_IsEmpty(&chan->send_info.send_list)) {
			send_buf_node *p = CONTAINING_RECORD(ListEntry_RemoveListHead(&chan->send_info.send_list), send_buf_node, entry);
			free(p->buf);
			free(p);
		}
#ifdef RVC_OS_WIN
		DeleteCriticalSection(&chan->send_info.lock);
#else
		pthread_mutex_destroy(&chan->send_info.lock);
#endif
		if (chan->dec_session) {
			screen_decoder_session_destroy(chan->dec_session);
			chan->dec_session = NULL;
		}
		free(chan);
	}
}

BIZCHAN_API(void) bizchan_set_tag(bizchan_t *chan, void *tag)
{
	chan->tag = tag;
}

BIZCHAN_API(void*) bizchan_get_tag(bizchan_t *chan)
{
	return chan->tag;
}

BIZCHAN_API(int) bizchan_start_connect(bizchan_t *chan)
{
	DWORD threadId;
	if (!chan) {
		return -1;
	}

	if (chan->work_thread) {
		chan->cb.dbg(chan, 2, "work_thread exsit!!!", chan->cb.user_data);
		return -1;
	}

#ifdef RVC_OS_WIN
	WSAResetEvent(chan->evt);
#endif
	chan->cb.dbg(chan, 0, "bizchan_start_connect 2", chan->cb.user_data);

	chan->stop_flag = 0;
	chan->connected = 0;

#ifdef RVC_OS_WIN
	chan->work_thread = (HANDLE)_beginthreadex(NULL, 0, &work_proc, (LPVOID)chan, 0, (unsigned int*)&threadId);
	if (!chan->work_thread) {
		return -1;
	}
#else
	int err = pthread_create(&chan->work_thread, NULL, work_proc, chan);
	if (0 == err) {
		Dbg("create work thread success, %lu.", chan->work_thread);
	}
	else {
		Dbg("create work thread failed.");
	}
#endif // RVC_OS_WIN

	// we now return, when connected, on_connect will invoked in work_proc thread
	chan->cb.dbg(chan, 0, "bizchan_start_connect success", chan->cb.user_data);

	return 0;
}

BIZCHAN_API(int) bizchan_start_close(bizchan_t *chan)
{
	chan->cb.dbg(chan, 2, "bizchan_start_close", chan->cb.user_data);
	chan->stop_flag = 1;
#ifdef RVC_OS_WIN
	WSASetEvent(chan->evt);
#else
	write(chan->evt[1], "stop", sizeof("stop"));
#endif
	return 0;
}

BIZCHAN_API(int) bizchan_close(bizchan_t *chan)
{
#ifdef RVC_OS_WIN
	if (chan->work_thread) {
		WaitForSingleObject(chan->work_thread, INFINITE);
		CloseHandle(chan->work_thread);
		chan->work_thread = NULL;
	}
#else
	pthread_join(chan->work_thread, NULL);
	chan->work_thread = 0;
#endif // RVC_OS_WIN

	return 0;
}

//static FILE *tx_log_fp = NULL;

BIZCHAN_API(int) bizchan_post_pkt(bizchan_t *chan, int type, int compress, int encrypt, int sub_type, int id, const char *pkt, int pkt_size)
{
	send_buf_node *t;

	if (!chan->connected)
		return -1;
	
	t = (send_buf_node *)malloc(sizeof(send_buf_node));

	//if (!tx_log_fp) {
	//	tx_log_fp = fopen("c:\\txlog.txt", "wt");
	//	fprintf(tx_log_fp, "===================\n");
	//}
	//{
	//	SYSTEMTIME st;
	//	GetLocalTime(&st);
	//	fprintf(tx_log_fp, "[%02d:%02d:%02d.%03d] type = %d, compress = %d, sub_type = %d, id = %d, pkt_size = %d, hash = %d\n", 
	//		st.wHour, st.wMinute, st.wSecond, st.wMilliseconds,
	//		type, compress, sub_type, id, pkt_size, hash32_buf(pkt, pkt_size, 0));
	//	fflush(tx_log_fp);
	//}

	if (type == ACM_TYPE_SRN) {
		int cat = ACM_SRN_CAT(sub_type);
		if (cat == ACM_SRN_REQ) {
			chan->screen_img_id = id;
		}
	} else if (type == ACM_TYPE_PHT) {
		int cat = ACM_PHT_CAT(sub_type);
		if (cat == ACM_PHT_REQ) {
			chan->photo_img_id = id;
		}
	}

	if (!compress || pkt_size == 0) {
		acm_hdr *hdr;
		t->buf = (char*)malloc(pkt_size + sizeof(acm_hdr));
		t->left = pkt_size + sizeof(acm_hdr);
		t->sended = 0;
		hdr = (acm_hdr*)&t->buf[0];
		hdr->compress = 0;
		hdr->length = pkt_size;
		hdr->sub_type = sub_type;
		hdr->type = type;
		hdr->id = id;
		hdr->encrypt = !!encrypt;
		hdr->hash = hash_key(pkt, pkt_size);
		memcpy(&hdr->data[0], pkt, pkt_size);
		t->need_encrypt = encrypt;
	} else {
		qlz_state_compress state_compress;
		acm_hdr *hdr;
		int new_pkt_size;
		t->buf = (char*)malloc(2*pkt_size + sizeof(acm_hdr) + 16);
		hdr = (acm_hdr *)&t->buf[0];
		new_pkt_size = (int)qlz_compress(pkt, (char*)&hdr->data[0], pkt_size, &state_compress);
		if (new_pkt_size < pkt_size) {
			hdr->compress = 1;
			hdr->length = new_pkt_size;
			t->left = new_pkt_size + sizeof(acm_hdr);
		} else {
			hdr->compress = 0;
			hdr->length = pkt_size;
			memcpy(&hdr->data[0], pkt, pkt_size);
			t->left = pkt_size + sizeof(acm_hdr);
		}
		hdr->type = type;
		hdr->id = id;
		hdr->sub_type = sub_type;
		hdr->encrypt = !!encrypt;
		hdr->hash = hash_key(pkt, pkt_size);
		t->sended = 0;
		t->need_encrypt = encrypt;
	}

#ifdef RVC_OS_WIN
	EnterCriticalSection(&chan->send_info.lock);
	ListEntry_AddTail(&chan->send_info.send_list, &t->entry);
	LeaveCriticalSection(&chan->send_info.lock);
	WSASetEvent(chan->evt);
#else
	pthread_mutex_lock(&chan->send_info.lock);
	ListEntry_AddTail(&chan->send_info.send_list, &t->entry);
	pthread_mutex_unlock(&chan->send_info.lock);

	write(chan->evt[1], "post", sizeof("post"));
#endif



	return 0;
}

BIZCHAN_API(int) bizchan_winsync_send(bizchan_t *chan, int sub_type, const void *buf, int size)
{
	if (chan && chan->work_thread) {
		return bizchan_post_pkt(chan, ACM_TYPE_SYNC, 1, sub_type, 1, 0, (const char*)buf, size);
	} else {
		return -1;
	}
}