main.cpp

通达信股票软件数据接收程序 能够将连接服务器

#include "config.h"

#include <string>
#include <vector>
#include <set>
#include <map>
#include <queue>
#include <iostream>
#include <fstream>

#include <loki/Factory.h>

#include <boost/thread/detail/config.hpp>
#include <boost/thread/condition.hpp>
#include <boost/thread/thread.hpp>
#include <boost/thread/recursive_mutex.hpp>
// #include <boost/smart_assert/assert_impl.hpp>
#include <boost/date_time/posix_time/posix_time_types.hpp>

#include "response.h"
#include "socketHandler.h"
#include "ClientSocket.h"

#include "stock.h"
#include "request.h"
#include "commands.h"
#include "cli.hpp"
#include "util.hpp"
#include "StockIo.h"

// using namespace smart_assert;
using namespace StockMarket;
using namespace boost;

static condition initialize_c;
static recursive_mutex initialize_m;

static void send_queued_request(ClientSocket& cs)
{
	static bool initial_send = true;
//	cout << "send_queued_request enter\r\n";
	recursive_mutex::scoped_lock guard(req_queue_mutex);
	CmdQueue_t &q = CmdQueue::Instance();
	if(!q.empty())
	{
		CmdData_t &d = q.front();			// retreive a command
		if(! initial_send)
		{
			d->next();
		}
		else
		{
			initial_send = false;
		}
		if(*d)
		{
			d->send(cs);
		}
		else
		{
			initial_send = true;
			q.pop();					// if no request, then destroy the command
		}
	}
//	cout << "send_queued_request exit\r\n";
}

// this is the thread to request data
void communication(int running_mode_deal, gregorian::date load_tr_date_from, gregorian::date load_tr_date_to)
{
	SocketHandler h;
	ClientSocket cs(h);
	cs.Open("218.18.103.38",7709);
//	cs.Open("210.22.9.235",7709);
//	cs.Open("61.141.193.172",7709);

 	// Add after Open
	h.Add(&cs);
	//// To do: if disconnect then reconnect

	//// use to count time
	posix_time::ptime now(posix_time::second_clock::local_time());
	gregorian::date today_date = now.date();
	uint today = date_to_uint(today_date);

	posix_time::ptime t1 = now;
	posix_time::time_duration reqInterval(0,2,0);
	posix_time::ptime td_start(now.date(), posix_time::time_duration(9,30,0));
	posix_time::ptime td_end(now.date(), posix_time::time_duration(15,0,0) + reqInterval);
	bool initial_stocksets_finished = false;
	bool initial_gbbq_info_started = false;
	bool initial_gbbq_info_finished = false;
	bool initial_all_finished = false;

	// calculate the initial transaction period
	bool get_first_instant_transact = true;
	
	if(running_mode_deal)
	{
		if(now > td_end)		
			load_tr_date_to = gregorian::day_clock::local_day();
		else
			load_tr_date_to = gregorian::day_clock::local_day() - gregorian::date_duration(1);
	}
	
	gregorian::date tr_date(load_tr_date_from);
	
	// just to calculate the total stocks in each market
	AddStockListRequest(false);
	hist_transact.clear();

	h.Select(1,0);
	while (true)
	{
		h.GetCount();
		h.Select(1,0);	//		wait for ready first before sending request and handle response if any

		recursive_mutex::scoped_lock guard(req_queue_mutex);
		CmdQueue_t &q = CmdQueue::Instance();
		if(q.empty())		// Hmm...All initial request has been sent
		{
			if(initial_all_finished)
			{
				if(!running_mode_deal) break;
					
				now = posix_time::second_clock::local_time();
				if(now - t1 >= reqInterval && now < td_end && now >= td_start)
				{
					t1 = now;
					if(get_first_instant_transact)
					{
						AddInstantTransRequest(1000);
						get_first_instant_transact = false;
					}
					else
					{
						AddInstantTransRequest(100);
					}
					AddStockListRequest(true);
				}
				else
				{
					break;
				}
			}
			else if(initial_gbbq_info_finished)
			{
				if(!hist_transact.empty())
				{
					uint d = date_to_uint(tr_date);
					save_day_transaction(d, hist_transact[d]);
					tr_date += gregorian::date_duration(1);
				}
				
				hist_transact.clear();

				bool bNewRequestAdded = false;
				while(tr_date <= load_tr_date_to)
				{
					if(!transact_data_exsist(tr_date))
					{
						cout << "AddHistTransRequest start " << tr_date << endl;
						AddHistTransRequest(tr_date, tr_date);
						bNewRequestAdded = true;
						break;
					}
					tr_date += gregorian::date_duration(1);
				}
				if(!bNewRequestAdded)
				{
					initial_all_finished = true;
				}
			}
			else if(initial_gbbq_info_started)
			{
				cout << "initial_gbbq_info_finished" << endl;
				// clear the unused data at initial stage
				stock_basic_info::Instance().save_gbbq_info(date_to_uint(today_date));
				initial_gbbq_info_finished = true;
			}
			else if( initial_stocksets_finished)
			{
				stock_basic_info::Instance().save_stock_set();
				
				if(!stock_basic_info::Instance().is_gbbq_file_exist(today))
				{
					cout << "AddGBBQRequest start" << endl;
					stock_basic_info::Instance().clear_gbbq_info();
					AddGBBQRequest();
				}
				else
				{
					initial_gbbq_info_finished = true;
				}
				initial_gbbq_info_started = true;
			}
			else
			{
				cout << "AddStockListRequest start" << endl;
				// to get all stocks list
				AddStockListRequest(true);
				initial_stocksets_finished = true;
			}

			if(q.empty())
			{
				CmdData_t r(new StockHeartBeat());
				q.push(r);
			}

		}
 		// if yes, then check time-interval, if more than 1 minite..then add new request to it
 		// 	当收到Response 的时候才设置Ready, 这些避免发送重复的序列号
 		if(Request::ready())	
		{
			send_queued_request(cs);
		}
	}
	cout << "Communitcation exit" << endl;

}

void cli()
{
	recursive_mutex::scoped_lock lk(initialize_m);
	initialize_c.wait(lk);

	cout << "/////////////////////////////////////////////////////////\n\n";
	cout << "\t\t Stock share analyse system...\n\n";
	cout << "/////////////////////////////////////////////////////////\n\n";
	cout << "Type anything or [q or Q] to quit\n\n";

	cli_grammar g;

	string str;
	while ( getline( cin, str ) )
	{
		if ( str[0] == 'q' || str[0] == 'Q' )
			break;

		if ( parse( str.c_str(), g).full )		// No skip
		{
			cout << "Success \n";
		}
		else
		{
			cout << "ERROR: bad command\n";
		}
	}

	cout << "Bye... :-) \n\n";

}

// The main program to receive commands from user and show information
int main()
{
	{
	// init response handler
	recursive_mutex::scoped_lock guard(res_fact_mutex);
	ResponseFactory::Instance().Register(CMD_STOCK_LIST, CreateStockListRes);
	ResponseFactory::Instance().Register(CMD_STOCKHOLD_CHANGE, CreateStockHoldChgRes);
	ResponseFactory::Instance().Register(CMD_HEART_BEAT, CreateStockHeartBeatRes);
	ResponseFactory::Instance().Register(CMD_INSTANT_TRANS, CreateInstantTransRes);
	ResponseFactory::Instance().Register(CMD_HIS_TRANS, CreateHisTransRes);
	}

//	if(!stock_basic_info::Instance().load_basic_info()) return 1;
//	stock_basic_info::Instance().save_basic_info();
//	if(!stock_basic_info::Instance().load_block_info()) return 1;
//	stock_basic_info::Instance().save_block_info();
//	stock_basic_info::Instance().load_gbbq_info_mat(gregorian::day_clock::local_day(), 20);

	int running_mode_deal = 0; gregorian::date load_tr_date_from, load_tr_date_to;
//	load_tr_date_from = gregorian::day_clock::local_day() - gregorian::date_duration(5);
//	load_tr_date_to = gregorian::day_clock::local_day() - gregorian::date_duration(1);
	load_tr_date_from = gregorian::date(2007, 9, 22);
	load_tr_date_to = gregorian::date(2007, 9, 22);
//	load_tr_date_to = gregorian::day_clock::local_day();

	communication(running_mode_deal, load_tr_date_from, load_tr_date_to);

//	system("D:\\temp\\shutdown.bat");
	return 0;


}