Status4ZxClient ZxClient::getStatus(){
    Status4ZxClient value = Close;
    if (pthread_mutex_lock(&m_statusMutex)) {
        LoggerPtr rootLogger = Logger::getRootLogger();  
        LOG4CXX_ERROR(rootLogger, "pthread_mutex_lock(&m_statusMutex) failed!");
        return value;
    }
    value =  m_status;
    pthread_mutex_unlock(&m_statusMutex);
    return value;
}

bool MemCacheServerHandler::GetPlatInfo(const string& platID, std::string& dist)
{
	if (g_pInst == NULL || !g_pInst->CanUse())
	{
		return false;
	}

// 	if (uid != 2928562105553271)
// 	{
// 		return false;
// 	}

	int     dbid    = g_pInst->GetServerId(platID);
	TCRDB*  pConn   = g_pInst->GetDB(dbid);
	if (pConn == NULL)
	{
		g_pInst->SetEnbale(false);
		LOG4CXX_ERROR(logger,"memcache_get_connect_null_error");
		return false;
	}

	
	int len  = 0;
	int klen = platID.length();
	char* buffer = (char*)tcrdbget(pConn, platID.c_str(), klen, &len);
	if (buffer == NULL)
	{
		int ecode = tcrdbecode(pConn);
		if (ecode != TTENOREC)
		{
			g_pInst->SetEnbale(false);
			LOG4CXX_ERROR(logger,"memcache_get_ecode_error");
		}
		return false;
	}
	std::string outbuf(buffer, len);
	dist = outbuf;
	free(buffer);	
	//g_pInst->GetCounter().increase("mem_get_user");
	return true;
}

bool getAvatar(const std::string &db_path, const std::string &name, std::string &photo) {
	bool ret = false;
	sqlite3 *db;
	LOG4CXX_INFO(logger, "Opening database " << db_path);
	if (sqlite3_open(db_path.c_str(), &db)) {
		sqlite3_close(db);
		LOG4CXX_ERROR(logger, "Can't open database");
	}
	else {
		sqlite3_stmt *stmt;
		PREP_STMT(stmt, "SELECT avatar_image FROM Contacts WHERE skypename=?");
		if (stmt) {
			BEGIN(stmt);
			BIND_STR(stmt, name);
			if(sqlite3_step(stmt) == SQLITE_ROW) {
				int size = sqlite3_column_bytes(stmt, 0);
				if (size > 0) {
				    const void *data = sqlite3_column_blob(stmt, 0);
				    photo = std::string((const char *)data + 1, size - 1);
				    ret = true;
				} else {
				    ret = false;
				}
			}
			else {
				LOG4CXX_ERROR(logger, (sqlite3_errmsg(db) == NULL ? "" : sqlite3_errmsg(db)));
			}

			int ret;
			while((ret = sqlite3_step(stmt)) == SQLITE_ROW) {
			}
			FINALIZE_STMT(stmt);
		}
		else {
			LOG4CXX_ERROR(logger, "Can't create prepared statement");
			LOG4CXX_ERROR(logger, (sqlite3_errmsg(db) == NULL ? "" : sqlite3_errmsg(db)));
		}
		sqlite3_close(db);
	}
	return ret;
}

/*
   Blocks until data can be written.
   If no data can be written in 2ms*100 the stream gets restarted.
   returns the amount of data that can be written.
*/
unsigned int PortAudio::getWriteAvailable()
{
    size_t writeAvailable = 0;
    int waitCount = 0;
    int failCount = 0;

    while( writeAvailable == 0 ) {
        writeAvailable = ringbuf.getWriteAvailable();

        if( writeAvailable == 0 ) {
            usleep(200000); //200 ms
            //LOG4CXX_DEBUG(narratorPaLog, "getWriteAvailable = 0, Pa_IsStreamActive = " << Pa_IsStreamActive(pStream));
        }

        if(waitCount++ > 10) {
            LOG4CXX_ERROR(narratorPaLog, "getWriteAvailable waittime exceeded, restarting stream");

            mError = Pa_AbortStream(pStream);
            // If abortstream fails, try reopening
            if(failCount++ > 0 || ( mError != paNoError && mError != paStreamIsStopped )) {
                LOG4CXX_ERROR(narratorPaLog, "Failed to abort stream, trying to close and reopen: " << Pa_GetErrorText(mError));

                mError = Pa_CloseStream(pStream);
                if(mError != paNoError)
                    LOG4CXX_ERROR(narratorPaLog, "Failed to close stream: " << Pa_GetErrorText(mError));

                isOpen = false;
                open(mRate, mChannels);
            }

            mError = Pa_StartStream(pStream);
            if(mError != paNoError) LOG4CXX_ERROR(narratorPaLog, "Failed to start stream: " << Pa_GetErrorText(mError));
            isStarted = true;

            waitCount = 0;
        }
    }

    //LOG4CXX_DEBUG(narratorPaLog, "getWriteAvailable = " << writeAvailable << ", Pa_IsStreamActive = " << Pa_IsStreamActive(pStream));
    return writeAvailable;
}

bool MyCacheServer::loadConfig(const std::string& cfg, const std::string& servernode)
{
	try 
	{
		config = YAML::LoadFile(cfg)[servernode];

		address_ = config["listen"].as<std::string>();
		if (address_.empty())
		{
			LOG4CXX_ERROR(logger, "loadConfig failed: " << cfg << ": no listen");
			return false;
		}

		workernum_ = config["workers"].as<int>();
		if (workernum_ == 0)
		{
			workernum_ = 1;
		}

		if (config["caches"].size() == 0 || config["databases"].size() == 0)
		{
			LOG4CXX_ERROR(logger, "loadConfig failed: " << cfg << ": no caches/databases");
			return false;
		}

		s_tablenum = config["tablenums"].as<int>();
		if (s_tablenum == 0)
			s_tablenum = 10;

		s_cacheonly = config["cacheonly"].as<int>();

		return true;

	}
	catch (const std::exception& err)
	{
		LOG4CXX_ERROR(logger, "loadConfig failed: " << cfg << ": "<< err.what());
	}

	return false;
}

std::string GetApplicationDataPath(const char* applicationName,const char* version)
{
  string versionSuffix;
  if(version && strlen(version) > 0)
    versionSuffix = string("\\")+string(version);
  TCHAR szPath[MAX_PATH];
  if(SUCCEEDED(SHGetFolderPath(NULL,CSIDL_APPDATA,NULL,0,szPath))) {
    return string(szPath)+string("\\")+string(applicationName)+versionSuffix;
  }
    LOG4CXX_ERROR(KrisLibrary::logger(),"GetApplicationDataPath: SHGetFolderPath failed? returning local folder\n");
  return string(applicationName)+versionSuffix;
}

void HTTPRequest::setProxy(std::string IP, std::string port, std::string username, std::string password) {
	if (curlhandle) {
		std::string proxyIpPort = IP + ":" + port;
		curl_easy_setopt(curlhandle, CURLOPT_PROXY, proxyIpPort.c_str());
		if (username.length() && password.length()) {
			std::string proxyUserPass = username + ":" + password;
			curl_easy_setopt(curlhandle, CURLOPT_PROXYUSERPWD, proxyUserPass.c_str());
		}
	} else {
		LOG4CXX_ERROR(logger, "Trying to set proxy while CURL isn't initialized");
	}
}

std::set<std::string> UserDB::getRegisteredUsers()
{
	std::string q = "select user from users";
	if(exe(q) != SQLITE_OK) {
		LOG4CXX_ERROR(logger, "Failed to registered users from database!");
		exit(0);
	}
	
	std::set<std::string> users;
	for(int i=0 ; i<data.size() ; i++)
		users.insert(data[i][0]);
	return users;
}

void SQLite3Backend::setUser(const UserInfo &user) {
	sqlite3_reset(m_setUser);
	sqlite3_bind_text(m_setUser, 1, user.jid.c_str(), -1, SQLITE_STATIC);
	sqlite3_bind_text(m_setUser, 2, user.uin.c_str(), -1, SQLITE_STATIC);
	sqlite3_bind_text(m_setUser, 3, user.password.c_str(), -1, SQLITE_STATIC);
	sqlite3_bind_text(m_setUser, 4, user.language.c_str(), -1, SQLITE_STATIC);
	sqlite3_bind_text(m_setUser, 5, user.encoding.c_str(), -1, SQLITE_STATIC);
	sqlite3_bind_int (m_setUser, 6, user.vip);

	if(sqlite3_step(m_setUser) != SQLITE_DONE) {
		LOG4CXX_ERROR(logger, "setUser query"<< (sqlite3_errmsg(m_db) == NULL ? "" : sqlite3_errmsg(m_db)));
	}
}

void client_socket_utils::session_connect(socket_session_ptr pSession)
{
	std::string& addr = pSession->get_remote_addr();
	try
	{
		//注册关闭回调函数
		pSession->installCloseCallBack(boost::bind(&client_socket_utils::close_callback, this, _1));
		//注册读到数据回调函数
		pSession->installReadDataCallBack(boost::bind(&client_socket_utils::read_data_callback, this, _1, _2, _3));

		std::vector<std::string> ip_port;
		boost::split(ip_port, addr, boost::is_any_of(":"));

		if (ip_port.size() < 2)
		{
			//throw std::runtime_error("ip 格式不正确！");
			LOG4CXX_ERROR(firebird_log, "[" << addr << "] ip 格式不正确!");
			return;
		}

		tcp::resolver resolver(pSession->socket().get_io_service());
		tcp::resolver::query query(ip_port[0], ip_port[1]);
		tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
		//pSession->set_begin_endpoint(endpoint_iterator);//设置起始地址，以便重连

		//由于客户端是不断重连的，即使还未连接也要保存该session
		m_manager.add_session(pSession);

		tcp::endpoint endpoint = *endpoint_iterator;
		pSession->socket().async_connect(endpoint,
				boost::bind(&client_socket_utils::handle_connect, this, boost::asio::placeholders::error, ++endpoint_iterator, pSession));
	} catch (std::exception& e)
	{
		LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << addr << "],socket异常:[" << e.what() << "]");
	} catch (...)
	{
		LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << addr << "],socket异常:[未知异常]");
	}
}

void Connection::cleanup()
{
    LOG4CXX_ERROR(logger, mInstanceName << " cleaning up");
    
    mReceiverThread->requestStop();
    mSenderThread->requestStop();
    
    // Wake up processor threads blocked in I/O operation
    try 
    {
        mSocket->shutdown();
    } 
    catch (aqlib::Exception ex) 
    {
        LOG4CXX_ERROR(logger, mInstanceName << ": " << ex.toString());
    }
    
    mReceiverThread->join();
    mSenderThread->join();
    
    LOG4CXX_ERROR(logger, mInstanceName << " cleaned up");
}

long PortAudio::stop()
{
    if(isStarted) {
        mError = Pa_StopStream(pStream);
        if(mError != paNoError)
            LOG4CXX_ERROR(narratorPaLog, "Failed to stop stream: " << Pa_GetErrorText(mError));

        ringbuf.flush();
        isStarted = false;
    }

    return mLatency;
}

bool PlatNetHandler::sendEventToAllGamed(Event *e)
{
	bool succ = true;
	for (int i=1; i<=serverConfig.gamedNum(); i++) {
		int fd = getFdFromId(i);
		if (fd<0) {
			LOG4CXX_ERROR(logger_, "Get Fd from ID failed!! gamed:"<<i<<" fd:"<<" cmd:"<<e->cmd());
			continue;
		}
		string text;
		e->SerializeToString(&text);
		text = "ev,"+text;
		//LOG4CXX_DEBUG(logger_, "event:"<<e->DebugString()<<"to fd:"<<fd<<" with size:"<<text.length());
		bool sendSucc = sendIntSizedFdString(fd, text);
		if (!sendSucc) {
			LOG4CXX_ERROR(logger_, "Write to cache failed!!");
		}
		//LOG4CXX_DEBUG(logger_, "sendEventToAllGamed, event: " << e->cmd() << ", size: " << text.size() << ", target fd: " << fd << ", write to cache: " << sendSucc);
		succ = succ && sendSucc;
	}
	return succ;
}

	int QFSFileHelper::Read(char *buffer, int length) {
		int bytes_read = qfshelper->kfsClient->Read(fd, buffer, length);
		if(bytes_read != length) {
			if(bytes_read < 0) {
				LOG4CXX_ERROR(logger, "Failure while reading file " << filename << " ERROR : " << KFS::ErrorCodeToStr(bytes_read));
				THROW_EXCEPTION(AppendStoreReadException, "Failed while reading file " + filename + " ERROR : " + KFS::ErrorCodeToStr(bytes_read));
			}
			else {
				LOG4CXX_ERROR(logger, "Less number of bytes read from file than specified")
			}
		}
		return bytes_read;	
	}

long MessageHandler::insertAudio(long translationid, const MessageAudio &ma)
{
    LOG4CXX_INFO(narratorMsgHlrLog, "Inserting new messageaudio for translation with id: " << translationid);
    LOG4CXX_DEBUG(narratorMsgHlrLog, "tagid: " << ma.getTagid() << ", uri: " << ma.getUri() << ", md5: " << ma.getMd5() << ", size: " << ma.getSize() << ", length: " << ma.getLength() << ", encoding: " << ma.getEncoding());

    int audioid = -1;

    if(ma.isAudioDataNil()) {
        LOG4CXX_ERROR(narratorMsgHlrLog, "No audio data has been set");
        return -1;
    }

    if(!db->prepare("INSERT INTO messageaudio (translation_id, tagid, text, length, encoding, data, size, md5) VALUES (?, ?, ?, ?, ?, ?, ?, ?)")) {
        LOG4CXX_ERROR(narratorMsgHlrLog, "Query failed '" << db->getLasterror() << "'");
        return -1;
    }

    if(!db->bind(1, translationid) ||
            !db->bind(2, ma.getTagid()) ||
            !db->bind(3, ma.getText().c_str()) ||
            !db->bind(4, ma.getLength()) ||
            !db->bind(5, ma.getEncoding().c_str()) ||
            !db->bind(6, ma.getAudioData(), ma.getSize()) ||
            !db->bind(7, (long)ma.getSize()) ||
            !db->bind(8, ma.getMd5(), 32, NULL)) {
        LOG4CXX_ERROR(narratorMsgHlrLog, "Bind failed '" << db->getLasterror() << "'");
        return -1;
    }

    narrator::DBResult result;
    if(!db->perform(&result)) {
        LOG4CXX_ERROR(narratorMsgHlrLog, "Query failed '" << db->getLasterror() << "'");
        return -1;
    }

    audioid = result.getInsertId();

    return audioid;
}

static void* _receiveThread(void* param) {
    MsgReceiver_t *receiver = (MsgReceiver_t *)param;
    amqp_envelope_t envelope;
    amqp_rpc_reply_t reply;
    amqp_frame_t *frame;
    time_t curtime;
    
    assert(receiver != NULL);
    
    LOG4CXX_DEBUG(logger, "_receiveThread: event receiver thread started");
    
    sendLeaseRequest(receiver);
    curtime = getCurrentTime();
    
    while(receiver->thread_run) {
        amqp_maybe_release_buffers(receiver->conn_state);
        
        pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
        
        if(getElapsedSecond(curtime, getCurrentTime()) > 60*5) {
            sendLeaseRequest(receiver);
            curtime = getCurrentTime();
        }
        
        reply = amqp_consume_message(receiver->conn_state, &envelope, NULL, 0);
        if(reply.reply_type == AMQP_RESPONSE_NORMAL) {
            // call handler
            MsgClientCallback_t callback = receiver->callback;

            LOG4CXX_INFO(logger, "_receiveThread: received a message");
            
            if(callback != NULL) {
                LOG4CXX_INFO(logger, "_receiveThread: received a message - calling a callback function");
                callback((const char*)envelope.routing_key.bytes, envelope.routing_key.len, (const char*)envelope.message.body.bytes, envelope.message.body.len);
            }
            
            int ack_result = amqp_basic_ack(receiver->conn_state, receiver->channel, envelope.delivery_tag, false);
            if(ack_result != 0) {
                LOG4CXX_ERROR(logger, "_receiveThread: ack failed " << ack_result);
            }

            amqp_destroy_envelope(&envelope);
        } else {
            LOG4CXX_INFO(logger, "_receiveThread: an exception occurred");
        }
        
        pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
    }
    
    receiver->thread_run = false;
}

void
Server::handleReceiveFromFront(boost::system::error_code a_errorCode, size_t a_receivedLength)
{
    if (0 == a_errorCode && a_receivedLength > 0) {
        RequestPtr requestPtr = std::make_shared<Request>();

        requestPtr->endpointFront = m_endpointFront;
        LOG4CXX_INFO(MiLog4cxx::logger, "FRONT received package from endpoint="<<
                requestPtr->endpointFront.address().to_string()<<":"<<requestPtr->endpointFront.port());

        int ret = requestPtr->pkg.ReadPkg(m_bufferFront, a_receivedLength);
        if (0 != ret) {
            LOG4CXX_ERROR(MiLog4cxx::logger, "FRONT failed to ReadPkg error="<<ret);
        }
        else {
            // dwSeqFrom 和 dwSeqConn 不一定全局唯一，因此需要生成一个local唯一的pkgId.
            requestPtr->localPkgId = getSeq();
            LOG4CXX_INFO(MiLog4cxx::logger, "FRONT received package cmd="<<requestPtr->pkg.m_stHead.wCommand<<
                                            " dwSeqFrom="<<requestPtr->pkg.m_stHead.dwSeqFrom<< " dwSeqConn="<<requestPtr->pkg.m_stHead.dwSeqConn
                                            <<" pkgId(new dwSeqConn)="<<requestPtr->localPkgId);
            // 把 request 放入 queue 中，并 notify_all workers处理
            m_queueMutex.lock();
            if (m_stop) {
                LOG4CXX_ERROR(MiLog4cxx::logger, "FRONT server is exiting, not allowed to process more request");
                m_queueMutex.unlock();
                return;
            }
            m_queue.emplace(requestPtr);
            m_queueMutex.unlock();
            m_condition.notify_all();
        }
    }
    else {
        LOG4CXX_ERROR(MiLog4cxx::logger, "FRONT receive failed with errorCode="<<a_errorCode);
    }

    doReceiveFromFront();
}

bool WorldNetHandler::sendEventToGamed(Event *e, int gid) {
	int fd = getGamedFdFromGamedId(gid>=0? gid: 1);
	//LOG4CXX_DEBUG(logger_, "look up gameid:"<<gid<<", get fd:"<<fd);
	if (fd<0)  {
		if(gid <= serverConfig.physicsGameNum())
		{//只输出一区
			LOG4CXX_ERROR(logger_, "Get Fd from ID failed!! gid:"<<gid<<" fd:"<<fd<<" cmd:"<<e->cmd());
		}
		return false;
	}
	e->set_worldsrvid(nid_);
	string text;
	e->SerializeToString(&text);
	text = "ev,"+text;
	//LOG4CXX_DEBUG(logger_, "event:"<<e->DebugString()<<"to fd:"<<fd<<" with size:"<<text.length());
	//LOG4CXX_INFO(logger_, "send event " << e->cmd() << " to gamed fd " << fd << ", size: " << text.size() << ", 0 at " << text.find((char)0, 0));
	bool succ = sendIntSizedFdString(fd, text);
	//LOG4CXX_DEBUG(logger_, "sendEventToGamed, event: " << e->cmd() << ", size: " << text.size() << ", target fd: " << fd << ", write to cache: " << succ);
	if (!succ) {
		LOG4CXX_ERROR(logger_, "Write to cache failed!!");
	}
	return succ;
}

int OutgoingSessionManager::createSession(const std::string& targetCompId) {
  ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
  OutgoingSessions::iterator it = outgoingSessions_.find(targetCompId);
  if(it == outgoingSessions_.end()) {
    OutgoingSessionPtr session(new OutgoingSession(targetCompId));
    if(session->open() == 0) {
      std::pair<OutgoingSessions::iterator, bool> result = outgoingSessions_.insert(OutgoingSessions::value_type(targetCompId, session));
      if(result.second) {
        return 0;
      }
      else {
        LOG4CXX_ERROR(logger_, "Error creating an outgoing session for " << targetCompId);
        return -1;
      }
    }
    else {
      LOG4CXX_ERROR(logger_, "Error creating an outgoing session for "<< targetCompId << ": Unable to activate a new thread.");
      return -1;
    }
  }
  else
    return 0;
}

bool ToSaMeshThreadSocket::setup() {
	fdClient = -1;
	struct sockaddr_un address;	
	if ((fdSocket = socket(AF_LOCAL, SOCK_STREAM, 0)) == -1) {
		LOG4CXX_ERROR(logger, "ToSaMeshThreadSocket::setup: error in socket()\n");
		return false;
	}
	fcntl(fdSocket, F_SETFL, O_NONBLOCK);
	unlink(TOSA_MESH_SOCKET_FILE);
	address.sun_family = AF_LOCAL;
	strcpy(address.sun_path, TOSA_MESH_SOCKET_FILE);
	if (bind(fdSocket, (struct sockaddr *) &address, sizeof (address)) == -1) {
		LOG4CXX_ERROR(logger, "ToSaMeshThreadSocket::setup: error in bind()\n");
		return false;
	}
	if (listen(fdSocket, TOSA_MESH_SOCKET_BACKLOG) == -1) {
		LOG4CXX_ERROR(logger, "ToSaMeshThreadSocket::setup: error in listen()\n");
		return false;
	}
	char mode[] = "0777";
	chmod(TOSA_MESH_SOCKET_FILE, strtol(mode, 0, 8));
	return true;
}