pms_server.cpp.svn-base

sigmadesign smp8623 gui source code ,bingo

		pthread_mutex_unlock(&ctxt.php_lock);

		ctxt.delete_php(RoomNo);

	END_DB()
}

void posting_answer(std::string& req)
{
	std::string RNo;
	std::string Answer = "0";
	std::string Date;
	std::string Time;

	DECODE_BEGIN(req)
		EXTRACT("RN(\\w+)", 				RNo)
		EXTRACT("AS([\\x20-\\x7B\\x7D-\\xFF]+)",	Answer)
		EXTRACT("DA([\\w]+)", 	Date)
		EXTRACT("TI([\\w]+)", 	Time)
	DECODE_END()

	BEGIN_DB()
		int RoomNo = get_room_id(qr, RNo);

		pthread_mutex_lock(&ctxt.php_lock);
		std::map<int, linked_ptr<php_status_t> >::iterator i = ctxt.php.find(RoomNo);
		if(i != ctxt.php.end())
		{
			if(i->second->type == 1)
			{
				if(Answer == "OK")
				{
					std::string admno = gen_inet(ctxt.php[RoomNo]->inettype, RoomNo, 0, "", 0);

					if(admno != "")
					{
						std::ostringstream msg;
						msg << "PA|RN"<<RoomNo<<"|NMroom"<<RoomNo<<"|PD"<<admno<<"|ASOK|\r\n";
						send(ctxt.php[RoomNo]->handle, msg.str().c_str(), msg.str().length(), 0);
					}
					else {
						std::ostringstream msg;
						msg << "PA|RN"<<RoomNo<<"|NM|PD|ASNO|\r\n";
						send(ctxt.php[RoomNo]->handle, msg.str().c_str(), msg.str().length(), 0);
					}
				}
			}
			else if(i->second->type == 0)
			{
				std::ostringstream msg;
				msg << "PA|RN"<<RoomNo<<"|AS"<<Answer<<"|\r\n";
				send(i->second->handle, msg.str().c_str(), msg.str().length(), 0);
			}
		}

		pthread_mutex_unlock(&ctxt.php_lock);

		ctxt.delete_php(RoomNo);

	END_DB()
}

void empty_room(std::string& req)
{
	std::string RNo = "0";

	std::vector<int> TZones;
	std::vector<int> TargetT;
	std::vector<int> GuestT_EmptyRoom;
	std::vector<int> Mode;

	DECODE_BEGIN(req)
		EXTRACT("RN(\\w+)", 				RNo)
	DECODE_END()

	BEGIN_DB()
		int RoomNo = get_room_id(qr, RNo);

		if(RoomNo == 0)
		{
			//std::cout<<"room 0\n";
			std::ostringstream msg;
			msg << "Неуспешна заявка за напускане на гост в стая " << RoomNo;
			write_event(4, msg);
			END_DB_RETURN()
			return;
		}

		TRANSACTION_BEGIN(*conn)
			{
				std::ostringstream update_q;
				update_q << "DELETE FROM RoomGuest WHERE RoomID=" << RoomNo;
				DB_EXECUTE(update_q.str().c_str());
				//std::cout<<update_q.str()<<std::endl;
			}

			std::ostringstream roomtemp_q;
			roomtemp_q << "SELECT TZoneID, DefaultFreeT, Mode FROM RoomTZone WHERE RoomID="<<RoomNo;
			mysqlpp::StoreQueryResult roomtemp_res = DB_STORE(roomtemp_q.str().c_str());

			{
				std::ostringstream update_q;
				update_q << "UPDATE Room SET GuestNo=0, GuestName='', LangID=0, LastChange='"<<get_date_time<<"', PIN='', WakeUp='0000-00-00 00:00:00' WHERE RoomID=" << RoomNo;
				DB_EXECUTE(update_q.str().c_str());
				//std::cout<<update_q.str()<<std::endl;
			}

			{
				std::ostringstream update_q;
				update_q << "UPDATE RoomTZone SET TargetT=DefaultFreeT WHERE RoomID=" << RoomNo;
				DB_EXECUTE(update_q.str().c_str());
			//	std::cout<<update_q.str()<<std::endl;
			}

			{
				std::ostringstream update_q;
				update_q << "UPDATE RoomStatus SET GreenIndic=0, RedIndic=0 WHERE RoomID=" << RoomNo;
				DB_EXECUTE(update_q.str().c_str());
				//std::cout<<update_q.str()<<std::endl;
			}

			{
				std::ostringstream update_q;
				update_q << "DELETE FROM Message WHERE RoomID=" << RoomNo;
				DB_EXECUTE(update_q.str().c_str());
			//	std::cout<<update_q.str()<<std::endl;
			}

			{
				std::ostringstream update_q;
				update_q << "UPDATE MoviePay SET IsActive='FALSE' WHERE RoomID=" << RoomNo;
				DB_EXECUTE(update_q.str().c_str());
				//std::cout<<update_q.str()<<std::endl;
			}

			{
				std::ostringstream update_q;
				update_q << "UPDATE InternetPay SET IsActive='FALSE' WHERE RoomID=" << RoomNo;
				DB_EXECUTE(update_q.str().c_str());
//				std::cout<<update_q.str()<<std::endl;
			}

			for(uint16 i = 0; i < roomtemp_res.num_rows(); i ++)
			{
	//			std::cout<<i<<std::endl;
				TZones.push_back((int) roomtemp_res[i][0]);
				TargetT.push_back((int) roomtemp_res[i][1]);
				Mode.push_back((int) roomtemp_res[i][2]);
			}

		TRANSACTION_END()

		if(bOK)
		{
		//	std::cout<<"delete keys\n";
			bms_delete_keys(RoomNo);

		}
		else
		{
			std::ostringstream msg;
			msg << "Неуспешна заявка за напускане на гост в стая "<<RoomNo;

			write_event(4, msg);
		}

		if(ctxt.conf->using_rc)
		{
			int Season = get_season(qr);
			std::string ips = get_rc_ip(qr, RoomNo);

			for(uint16 i = 0; i < TZones.size(); i++)
			{
				std::ostringstream clmt;
				clmt.fill('0');
				clmt << "CLMT:" << std::setw(2) << TZones[i] - 1 << Season << std::setw(2) << TargetT[i] << std::setw(2) << TargetT[i] << Mode[i] << "*";

				send_to_rc(ips, clmt);
			}

			std::ostringstream indc;
			indc << "INDC:00*";

			send_to_rc(ips, indc);
		}

	END_DB()
}

void pms_msg_handler(void* msg)
{
	msg_t* _msg = static_cast<msg_t*> (msg);

	mysqlpp::Connection::thread_start();

	pcrecpp::RE("(\\x0D)").GlobalReplace("", &_msg->command);
	pcrecpp::RE("(\\x0A)").GlobalReplace("", &_msg->command);
	pcrecpp::StringPiece input_buffer(_msg->command);
//	for(int i = 0; i < _msg->command.length(); i ++)
//		std::cout << std::hex << (int)_msg->command[i] << " ";
//	std::cout<<"\n";
	std::vector<std::string> msg_commands;
	std::string commands;

	while(pcrecpp::RE("\\x02([\\x20-\\xFF]+)\\x03").Consume(&input_buffer, &commands))
	{
		msg_commands.push_back(commands);
	}

//	pcrecpp::RE("(\3)").Replace("", &_msg->command);

//	pcrecpp::RE("(\2)").GlobalReplace("", &_msg->command);
//	pcrecpp::RE("(\3)").GlobalReplace("", &_msg->command);
//	pcrecpp::RE("(\\s{2})").GlobalReplace("", &_msg->command);

//	std::vector<std::string> msg_decomposed;
//	pcrecpp::StringPiece input(_msg->command);
//	std::string command;

	for(uint16 c = 0; c < msg_commands.size(); c++)
	{
		std::vector<std::string> msg_decomposed;
		pcrecpp::StringPiece input(msg_commands[c]);
		std::string command;
	while(pcrecpp::RE("([\\x20-\\x7B\\x7D-\\xFF]+)\\|").Consume(&input, &command))
	{
		msg_decomposed.push_back(command);

	}

	if(ctxt.pmsanswered == 0)
	{
		if(msg_decomposed[0] == "LS")
		{
			pthread_mutex_lock(&ctxt.pms_lock);
			link_start(msg_commands[c]);
			pthread_mutex_unlock(&ctxt.pms_lock);
		} else if(msg_decomposed[0] == "LA")
		{
			pthread_mutex_lock(&ctxt.pms_lock);
			link_alive(msg_commands[c]);
			pthread_mutex_unlock(&ctxt.pms_lock);
		}
		else
		{
			logger( ERROR, "Ignorring command %s. Incorrect state.", msg_decomposed[0].c_str() );
		}
	} else if(ctxt.pmsanswered == 1)
	{
		if(msg_decomposed[0] == "LA")
		{
			pthread_mutex_lock(&ctxt.pms_lock);
			link_alive(msg_commands[c]);
			pthread_mutex_unlock(&ctxt.pms_lock);
		}
		else
		{
			logger( ERROR, "Ignorring command %s. Incorrect state.", msg_decomposed[0].c_str() );
		}

	} else {

		if(msg_decomposed[0] == "KR")
		{
			key_request(msg_commands[c]);


		} else if(msg_decomposed[0] == "KD")
		{
			key_delete(msg_commands[c]);

		} else if(msg_decomposed[0] == "DS")
		{
			ctxt.SF_RoomNo = 0;
			//sync_db(msg_commands[c], _msg->sd);

		} else if(msg_decomposed[0] == "DE")
		{
			init_synch_room_data(0, 0, "", 0);
			//sync_db(msg_commands[c], _msg->sd);
		}else if(msg_decomposed[0] == "GI")
		{
			check_in(msg_commands[c]);

		} else if(msg_decomposed[0] == "GO")
		{
			bool empty = false;

			for(uint16 i = 0; i < msg_decomposed.size(); i ++)
			{
				if(msg_decomposed[i] == "SF")
					empty = true;
			}

			if(empty)
				empty_room(msg_commands[c]);
			else
				check_out(msg_commands[c]);

		} else if(msg_decomposed[0] == "GC")
		{
			guest_change(msg_commands[c]);

		} else if(msg_decomposed[0] == "XI")
		{
			bill_info(msg_commands[c]);

		} else if(msg_decomposed[0] == "XB")
		{
			bill_total(msg_commands[c]);

		} else if(msg_decomposed[0] == "PA")
		{
			posting_answer(msg_commands[c]);

		} else if((msg_decomposed[0] == "XL") || (msg_decomposed[0] == "WC") || (msg_decomposed[0] == "WR") || (msg_decomposed[0] == "XD") )
		{
			process_command(msg_commands[c]);
		}

	}
	}

	if(_msg)
	{
		delete _msg;
		_msg = NULL;
	}

	mysqlpp::Connection::thread_end();
}

void*
pms_server_func(void*, int& pmssock, int& pms_port, std::string& pms_host, int& pmsanswered, void (*msg_handler)(void*))
{
	pmssock = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
	if(pmssock == -1){
		logger(ERROR, "Can't create socket for pms server: %s\n", strerror(errno));
		ctxt.shutdown = 1;

		return NULL;
	}

	struct sockaddr_in isa;

	isa.sin_family = AF_INET;
	isa.sin_port = htons(pms_port);
	inet_aton(pms_host.c_str(), &isa.sin_addr);

	logger(INFO, "Successfully started pms server\n");

	try{
		do{
			pollfd pfd;
			pfd.fd = pmssock;
			pfd.events = POLLOUT;

			//int ready = poll(&pfd, 1, 2000);
			int ready = 1;
			if(ready != -1)
			{
		//		printf("%.2x\n", pfd.revents);
				//if((pfd.revents & POLLHUP) == POLLHUP)
				{
					int ret = connect(pmssock, (sockaddr*) &isa, sizeof(isa));
					if(ret == -1)
					{
						logger(ERROR, "Can not connet to IP address %s: %s\n", pms_host.c_str(), strerror(errno));
						sleep(5);
					} else {
						logger(INFO, "Connect to pms %s: %s\n", pms_host.c_str(), strerror(errno));
						std::ostringstream init_msg;
						init_msg << "LS|DA" << get_date << "|TI"<<get_time<<"|";
						send_to_pms(init_msg, pmssock);

						do{
							pfd.events = POLLIN;

							int ready = poll(&pfd, 1, 2000);
							if(ready != -1)
							{

								if((pfd.revents & POLLHUP) == POLLHUP)
								{
									logger(ERROR, "Remote site closed connection for pms\n");

									throw errno;
								}

								if((pfd.revents & POLLIN) == POLLIN)
								{
									char buffer[TCP_MSG_LENGTH];

									ssize_t received = recv(pmssock, buffer, TCP_MSG_LENGTH, 0);

									if(received > 0)
									{
										buffer[received] = '\0';
										msg_t* new_msg = new msg_t(buffer);

										logger(RECV, "msg %s from pms length: %d\n", buffer, received);
										ctxt.pool_tcp->tpool_add_work(msg_handler, new_msg);
									}
									else
									{
										throw errno;
									}
								}
							} else {

								throw errno;
							}
						}while(ctxt.shutdown != 1);
					}
				}
				continue;
			} else {

				throw errno;
			}

		} while (ctxt.shutdown != 1);
	} catch(int& n)
	{
		logger(FATAL, "Error in pms server on IP address %s: %s\n", pms_host.c_str(), strerror(n));
		//ctxt.shutdown = 1;
	}

	pmsanswered = 0;
	shutdown(pmssock, SHUT_RDWR);
	close(pmssock);
	logger(INFO, "Closing pms server\n");

	return NULL;
}

void*
pms_server(void* context)
{
    sigset_t mask;

    sigfillset(&mask); /* Mask all allowed signals */
    pthread_sigmask(SIG_BLOCK, &mask, NULL);

    std::cout<<"starting PMS 1\n";
	while(ctxt.shutdown != 1)
	{
		pms_server_func(context, ctxt.pmssock, ctxt.conf->pms_port, ctxt.conf->pms_host, ctxt.pmsanswered, pms_msg_handler);
		sleep(10);
	}

	return NULL;
}

void pms_msg_handler_2(void* msg)
{
	msg_t* _msg = static_cast<msg_t*> (msg);

	mysqlpp::Connection::thread_start();

	pcrecpp::StringPiece input_buffer(_msg->command);
		std::vector<std::string> msg_commands;
		std::string commands;

		while(pcrecpp::RE("\\x02([\\x20-\\xFF]+)\\x03").Consume(&input_buffer, &commands))
		{
			msg_commands.push_back(commands);
		}

	//	pcrecpp::RE("(\3)").Replace("", &_msg->command);

	//	pcrecpp::RE("(\2)").GlobalReplace("", &_msg->command);
	//	pcrecpp::RE("(\3)").GlobalReplace("", &_msg->command);
	//	pcrecpp::RE("(\\s{2})").GlobalReplace("", &_msg->command);

	//	std::vector<std::string> msg_decomposed;
	//	pcrecpp::StringPiece input(_msg->command);
	//	std::string command;

		for(uint16 c = 0; c < msg_commands.size(); c++)
		{
			std::cout<<msg_commands[c]<<std::endl;
			std::vector<std::string> msg_decomposed;
			pcrecpp::StringPiece input(msg_commands[c]);
			std::string command;
	while(pcrecpp::RE("([\\x20-\\x7B\\x7D-\\xFF]+)\\|").Consume(&input, &command))
	{
		msg_decomposed.push_back(command);

	}

	if(ctxt.pmsanswered2 == 0)
	{
		if(msg_decomposed[0] == "LS")
		{
			std::cout<<"state "<<ctxt.pmsanswered2<<std::endl;
			pthread_mutex_lock(&ctxt.pms_lock2);
			link_start_2(msg_commands[c]);
			pthread_mutex_unlock(&ctxt.pms_lock2);
			std::cout<<"enter\n";
		} else if(msg_decomposed[0] == "LA")
		{
			pthread_mutex_lock(&ctxt.pms_lock2);
			link_alive_2(msg_commands[c]);
			pthread_mutex_unlock(&ctxt.pms_lock2);
		}
		else
		{
			logger( ERROR, "Ignorring command %s. Incorrect state.", msg_decomposed[0].c_str() );
		}
	} else if(ctxt.pmsanswered2 == 1)
	{
		if(msg_decomposed[0] == "LA")
		{
			pthread_mutex_lock(&ctxt.pms_lock2);
			link_alive_2(msg_commands[c]);
			pthread_mutex_unlock(&ctxt.pms_lock2);
		}
		else
		{
			logger( ERROR, "Ignorring command %s. Incorrect state.", msg_decomposed[0].c_str() );
		}

	} else {

		if(msg_decomposed[0] == "XI")
		{
			bill_info(msg_commands[c]);

		} else if(msg_decomposed[0] == "XB")
		{
			bill_total(msg_commands[c]);

		} else if(msg_decomposed[0] == "PA")
		{
			posting_answer(msg_commands[c]);

		} else if((msg_decomposed[0] == "XL") || (msg_decomposed[0] == "WC") || (msg_decomposed[0] == "WR") || (msg_decomposed[0] == "XD") )
		{
			process_command(msg_commands[c]);
		}

	}
		}

	if(_msg)
	{
		delete _msg;
		_msg = NULL;
	}

	mysqlpp::Connection::thread_end();
}

void*
pms_server_2(void* context)
{
    sigset_t mask;

    sigfillset(&mask); /* Mask all allowed signals */
    pthread_sigmask(SIG_BLOCK, &mask, NULL);

	while(ctxt.shutdown != 1)
	{
		pms_server_func(context, ctxt.pmssock2, ctxt.conf->pms_port_2, ctxt.conf->pms_host_2, ctxt.pmsanswered2, pms_msg_handler_2);
		sleep(10);
	}

	return NULL;
}