/***************************************************
Receive
    receives data to the datasource from the current
    connection.
Params
    name			- the name of the file to receive to
    fileType		- UTM_OM_WRITING:create/overwrite  
					  UTM_OM_APPEND:create/append
    bytesReceived	- number of bytes received in total
    [timeOut]		- timeOut in seconds
Return
    UTE_SUCCESS				- success
    UTE_FILE_TYPE_ERROR		- invalid file type
	UTE_DS_OPEN_FAILED		- unable to open specified data source
    UTE_SOCK_TIMEOUT		- timeout
    UTE_SOCK_RECEIVE_ERROR	- receive socket error
    UTE_DS_WRITE_FAILED		- data source write error
    UTE_ABORTED				- aborted
****************************************************/
int CUT_WSThread::Receive(CUT_DataSource & dest, long* bytesReceived, OpenMsgType type, int timeOut) 
{
    char    data[WSS_BUFFER_SIZE];
    int     count;
    int     error = UTE_SUCCESS;

    *bytesReceived = 0;

    // Check data source open type 
    if(type != UTM_OM_APPEND && type != UTM_OM_WRITING)
        return OnError(UTE_FILE_TYPE_ERROR);

	// Open data source
	if(dest.Open(type) == -1)
		return OnError(UTE_DS_OPEN_FAILED);

    // Start reading in the data
    while(error == UTE_SUCCESS ) {
        if(timeOut > 0) 
		{
            if(WaitForReceive(timeOut, 0) != UTE_SUCCESS) 
			{
                error = UTE_SOCK_TIMEOUT;
                break;
			}
		}

        count = Receive(data,sizeof(data));

        // Check to see if the connection is closed
        if(count == 0)
            break;

        // If an error then break
        if(count == SOCKET_ERROR) {
            error = UTE_SOCK_RECEIVE_ERROR;
            break;
			}

        // Write the the data source
		if(dest.Write(data, count) != count) {
            error = UTE_DS_WRITE_FAILED;
            break;
			}

        // Count the bytes copied
        *bytesReceived += count;

        // Send notify
        if(ReceiveFileStatus(*bytesReceived) == FALSE) {
            error = UTE_ABORTED;
            break;
			}
		}

	// Close data source
	dest.Close();

    return OnError(error);
}

bool TcpClient::Connect_()
{
    int fd;
    sockaddr_in sock_addr;

    fd = socket(AF_INET, SOCK_STREAM, 0);
    sock_addr.sin_family = PF_INET;
    sock_addr.sin_port = htons(server_addr_.port_);
    inet_aton(server_addr_.ip_.c_str(), &sock_addr.sin_addr);
    int reuseaddr = 1;
    if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &reuseaddr, sizeof(reuseaddr)) == -1) {
        OnError(errno, strerror(errno));
        close(fd);
        return false;
    }

    if (connect(fd, (sockaddr*)&sock_addr, sizeof(sockaddr_in)) == -1) {
        OnError(errno, strerror(errno));
        close(fd);
        return false;
    }
    IPAddress local_addr;
    SocketAddrToIPAddress(sock_addr, local_addr);
    OnConnected(fd, local_addr);

    return true;
}

bool NET_BaseSocket::SendData(const void* data, DWORD data_size)
{
//	MutexWrap socket_access(m_socket_mutex);
	if (!m_created)
		return false;

	if (data_size > m_out_buf_size)
	{
		CString str;
		str.Format("NET_BaseSocket::SendData, Размер пакета: %d, размер буфера: %d", data_size, m_out_buf_size);
		OnError(str);
		return false;
	}
	int free_size = m_out_data->SpaceFor(RingBuffer_WRITE);
	if ((int)data_size > free_size)
	{
		OnError("NET_BaseSocket::SendData, data_size > free_size");
		return false;
	}
	{
		MutexWrap out_write_access(m_out_data->m_w_mutex);
		if (!m_out_data->Write((const char*)data, data_size))
		{
			OnError("NET_BaseSocket::SendData, !m_out_data->Write");
			return false;
		}
	}
	SignalOnPacketWriten();
	return true;
}

/********************************
GetMsgSize()
    Retrieve the size of the message specified by the index.
    When this function is invoked on a valid connection and the message 
    specified by the Index is not marked for deletion, The POP3 server 
    issues a positive response with a line containing 
    information for the message.
PARAM
    int index - the message number 
    long *size - pointer to the message size
RETURN
    UTE_NOCONNECTION        - NOT connected
    UTE_SVR_NO_RESPONSE     - no response from Server
    UTE_LIST_FAILED         - LIST command was not reponded to positivly
    UTE_INVALID_RESPONSE    - Malformed response
*********************************/
int CUT_POP3Client::GetMsgSize(int index,long *size){

    char param[MAX_PATH+1];

    //check to see if the connection is made
    if(!IsConnected())
        return OnError(UTE_NOCONNECTION);

    //send the LIST command
    _snprintf(m_szBuf,sizeof(m_szBuf)-1,"LIST %d\r\n",index);
    Send(m_szBuf);

    //get the response
    if(ReceiveLine(m_szBuf,sizeof(m_szBuf),m_nPOP3TimeOut) <= 0)
        return OnError(UTE_SVR_NO_RESPONSE);

    if(m_szBuf[0] != '+')
        return OnError(UTE_LIST_FAILED);

    CUT_StrMethods::RemoveCRLF(m_szBuf);
    
    // a server response to a LIST command that includes an argument
    // should have +OK[SPACE]messageNumber[SPACE]size
    // Hence we will be looking for the third octet of the server response
    if(CUT_StrMethods::ParseString(m_szBuf," ()",2,param,sizeof(param)) != UTE_SUCCESS)
        return OnError(UTE_INVALID_RESPONSE);

    *size = atol(param);

    return OnError(UTE_SUCCESS);
}

/********************************************************************
GetNumMsgs
    The objective of this function is to retrieve the number 
    of messages stored in the server for the user maildrop. 

NOTE: Server should not include any messages marked for deletion in the response

PARAM:
    int *number - pionter to the number of available messages
RETURN:
    UTE_NOCONNECTION        - NOT connected
    UTE_SVR_NO_RESPONSE     - no response from Server
    UTE_STAT_FAILED         - STAT command was not reponded to positivly
    UTE_INVALID_RESPONSE    - Malformed response
*********************************************************************/
int CUT_POP3Client::GetNumMsgs(int *number){

    char param[MAX_PATH+1];

    //check to see if the connection is made
    if(!IsConnected())
        return OnError(UTE_NOCONNECTION);

    //send the STAT command
    Send("STAT\r\n");

    //get the response
    if(ReceiveLine(m_szBuf,sizeof(m_szBuf),m_nPOP3TimeOut) <= 0)
        return OnError(UTE_SVR_NO_RESPONSE);

    if(m_szBuf[0] != '+')
        return OnError(UTE_STAT_FAILED);

    CUT_StrMethods::RemoveCRLF(m_szBuf);

    if(CUT_StrMethods::ParseString(m_szBuf," ",1,param,sizeof(param)) != UTE_SUCCESS)
        return OnError(UTE_INVALID_RESPONSE);

    *number = atoi(param);

    return OnError(UTE_SUCCESS);
}

/********************************
CloseTop()
    Close the currently open message by reading all
    availabe lines of the Top response if any.
PARAM
    NONE
RETURN
    UTE_SUCCESS         - Success
    UTE_NOCONNECTION    - NOT connected
*********************************/
int CUT_POP3Client::CloseTop(){
    int     error = UTE_SUCCESS;

    if(m_bTopOpen == FALSE)
        return OnError(UTE_SUCCESS);

    //check to see if the connection is made
    if(!IsConnected())
        return OnError(UTE_NOCONNECTION);
    
    int len;

    while(!m_bReadTopFinished) {
        if(IsAborted()) {       // Test for abortion flag
            error = UTE_ABORTED;
            break;
            }

        //read in a line
        len = ReceiveLine(m_szBuf,sizeof(m_szBuf),m_nPOP3TimeOut);
        
        if( len  <= 0 )
            break;
        
        if(m_szBuf[0] == '.' && len <= 3)
            break;
        }

    m_bTopOpen = FALSE;
    m_bReadTopFinished = FALSE;

    return OnError(error);
}

//-----------------------------------------------------------------------------------------------------//
STDMETHODIMP CMarketDataProvider::_OnError(ErrorNumberEnum enErrorNumber, BSTR bsDescription,
										   RequestsTypeEnum enReqType, VARIANT varRequest)
{

	CString sDesc = (LPCTSTR)_bstr_t(bsDescription);
	
	if(enErrorNumber == enNotConnected || enErrorNumber == enConnectionWasDisconnected)
	{
		OnDisconnect(enErrorNumber, bsDescription);
		return S_OK;
	}

	switch(enReqType)
	{
		case enRequestLastQuote:
		{
				OnError(enErrorNumber, bsDescription);
				break;
		}
		case enSubscribeQuote:
		{
				OnError(enErrorNumber, bsDescription);
				break;
		}
		default:
		{
				OnError(enErrorNumber, bsDescription);
		}
	}
	return S_OK;
}

/*********************************************
SplitAddress
    Internal function
    Splits the given address into its two parts,
    domain name and user name
Params
    name    - user name buffer
    domain  - domain name buffer
    address - address to split apart
Return
    UTE_SUCCESS - success
    UTE_ERROR   - error bad address format
**********************************************/
int CUT_FingerClient::SplitAddress(LPSTR name,LPSTR domain,LPCSTR address){

    int t;
    int len         = (int)strlen(address);
    int position    = -1;

    //search for the @ symbol
    for(t=0; t<len; t++) {
        if(address[t] == '@'){
            position = t;
            break;
        }
    }
    //if no @ was found or no domain then return an error
    if(position == -1 || position == (len-1))
        return OnError(UTE_ERROR);

    //copy the name
    strcpy(name, address);
    name[position] = 0;

    //copy the domain
    strcpy(domain, &address[position+1]);

    return OnError(UTE_SUCCESS);
}

/**
 @brief 큐 버퍼에서 새로운 패킷을 뽑아낸다.. 패킷이 도착하지 않았으면 FALSE를 리턴한다
*/
BOOL XENetworkConnection::PumpPacket( XPacket *pOutPacket, xtError* pOut )
{
	// 패킷펌핑하고 있을때 다른 스레드가 이 커넥션을 사용하지 못하게 막는다.
	//
	BOOL bRet = FALSE;
	{
		// 최초 읽은 4바이트는 패킷의 총 길이다		
		if( m_qBuffer.size() >= sizeof(DWORD) ) {		// 큐버퍼에 최소4바이트가 모여 패킷길이를 알수 있게 되었다
			if( m_sizePacket == 0 ) {
				m_sizePacket = m_qBuffer.PopDWORD();	// 큐에서 4바이트를 꺼낸다.
				if( XBREAK( m_sizePacket > 0xffff /*|| m_sizePacket <= 4*/) ) {
					m_sizePacket = 0;
					// 정상적이지 않은 패킷이 날아왔음. 접속끊어야 함.
					m_qBuffer.clear();
					DoDisconnect();		// 즉시 접속을 끊어라.
					// 에러 핸들러 호출
					*pOut = xERR_PACKET_SIZE_TO_LARGE;
					OnError( xERR_PACKET_SIZE_TO_LARGE, m_sizePacket );
					return FALSE;
				}
			}
		}
		if( XBREAK( m_sizePacket > 0xffff ) ) {		// 뭔가 잘못된 경우다
			m_sizePacket = 0;
			*pOut = xERR_PACKET_SIZE_TO_LARGE;
			return FALSE;
		}
		// 패킷 하나가 완전히 도착하면 패킷 아카이브를 만든다
		if( m_sizePacket > 0 && (DWORD)m_qBuffer.Size() >= m_sizePacket )	// 필요한 패킷양 이상이 큐에 쌓였다
		{
			if( pOutPacket->IsOverBuffer( m_sizePacket ) ) {
				pOutPacket->ReAllocBuffer( m_sizePacket + 16 );	// 안전을 위해 16바이트 더 늘임.
			}
			int popSize = m_qBuffer.PopBlock( pOutPacket->GetBufferPacket()
																			, pOutPacket->GetMaxBufferSize()
																			, m_sizePacket );	// 큐에서 sizePacket만큼 pop시킨다
			if( XBREAK( popSize == 0 ) )
			{
				// 패킷에 담는데 에러가 남 접속끊어야 함.
				m_qBuffer.clear();
				//				SetbDestroy( TRUE );
				DoDisconnect();
				*pOut = xERR_ERROR;
				OnError( xERR_PACKET_SIZE_TO_LARGE, m_sizePacket );
				return FALSE;
			}
			pOutPacket->SetPacketLength( m_sizePacket );	// 완성된 패킷이 준비되었다
#ifdef _XCRYPT_PACKET
			pOutPacket->IsCryptThenSetFlag();	// 암호화 된 메모리인지 플래그를 세팅한다.
#endif
			m_sizePacket = 0;
			bRet = TRUE;
		}
	}
	//
	if( bRet ) {
		*pOut = xERR_NOERROR;
	}
	return bRet;
}

/***************************************************
WaitForSend
    Waits up til the specified time to see if the
    connection is ready to send data if it is then
    it will return UTE_SUCCESS
Params
    secs  - max. seconds to wait
    uSecs - max. micro seconds to wait
Return
    UTE_ERROR	- error
    UTE_SUCCESS	- success
****************************************************/
int CUT_WSThread::WaitForSend(long secs,long uSecs){
    
    fd_set writeSet;
    struct timeval tv;

    tv.tv_sec = secs;
    tv.tv_usec =uSecs;

    FD_ZERO(&writeSet);
	
	//This will generate a warning 
	// this is internal to winsock (warning C4127: conditional expression is constant)
	// maybe you want to use 
	// #pragma warning (disable : 4127) 
	// but on the other hand maybe you don't
	
	#pragma warning( disable : 4127 )
	FD_SET(m_clientSocket,&writeSet);
	#pragma warning( default : 4127 )
	

    //wait up to the specified time to see if data is avail
    if( select(-1,NULL,&writeSet,NULL,&tv)!= 1)
        return OnError(UTE_ERROR);
    
    return OnError(UTE_SUCCESS);
}

void EventLoop :: OneLoop(const int iTimeoutMs)
{
    int n = epoll_wait(m_iEpollFd, m_EpollEvents, MAX_EVENTS, 1);
    if (n == -1)
    {
        if (errno != EINTR)
        {
            PLErr("epoll_wait fail, errno %d", errno);
            return;
        }
    }

    for (int i = 0; i < n; i++)
    {
        int iFd = m_EpollEvents[i].data.fd;
        auto it = m_mapEvent.find(iFd);
        if (it == end(m_mapEvent))
        {
            continue;
        }

        int iEvents = m_EpollEvents[i].events;
        Event * poEvent = it->second.m_poEvent;

        int ret = 0;
        if (iEvents & EPOLLERR)
        {
            OnError(iEvents, poEvent);
            continue;
        }
        
        try
        {
            if (iEvents & EPOLLIN)
            {
                ret = poEvent->OnRead();
            }

            if (iEvents & EPOLLOUT)
            {
                ret = poEvent->OnWrite();
            }
        }
        catch (...)
        {
            ret = -1;
        }

        if (ret != 0)
        {
            OnError(iEvents, poEvent);
        }
    }
}

int CUT_HTTPServer::SetPath(LPCSTR path) {

    if(path == NULL)
        return OnError(UTE_PARAMETER_INVALID_VALUE);    
    
    strncpy(m_szPath, path, MAX_PATH);

    if(m_szPath[strlen(m_szPath) - 1] != '\\')
        strcat(m_szPath,"\\");
    
    return OnError(UTE_SUCCESS);
}

	bool CServerSocket::ActiveStart()
	{
		HANDLE hIOCPHandle;
		sockaddr_in addr;
		m_hIOCP = CreateIoCompletionPort(INVALID_HANDLE_VALUE,0,0,0);
		if(m_hIOCP!=0)
		{
			m_IOCPSOCKET.socket = WSASocket(PF_INET,SOCK_STREAM,IPPROTO_IP,NULL,0,WSA_FLAG_OVERLAPPED);
			if(m_IOCPSOCKET.socket != INVALID_SOCKET)
			{
				memset(&addr,0,sizeof(sockaddr_in));
				addr.sin_family = PF_INET;
				if(strcmp(m_IOCPSOCKET.ip,"0.0.0.0")==0)
				{
					addr.sin_addr.S_un.S_addr = INADDR_ANY;

				}
				else
				{
					addr.sin_addr.S_un.S_addr = inet_addr(m_IOCPSOCKET.ip);
				}
				WSAHtons(m_IOCPSOCKET.socket,m_IOCPSOCKET.port,&addr.sin_port);
				//°ó¶¨Socket
				if(bind(m_IOCPSOCKET.socket,(sockaddr*)&addr,sizeof(sockaddr_in))==0)
				{
					m_AcceptThread = new CAcceptThread(this);
					m_AcceptThread->Resume();
					for (int i = 0 ;i<= m_iWorkThreadCount ;i++)
					{
						m_WorkThread[i] = new CWorkThread(this,false);
					}
					return true;

				}
				else
				{
					this->OnError(NULL,GetLastError());
					closesocket(m_IOCPSOCKET.socket);
					m_IOCPSOCKET.socket = INVALID_SOCKET;
				}
			}//endif
			else
			{
				OnError(NULL,WSAGetLastError());			
				hIOCPHandle = m_hIOCP;
				m_hIOCP = 0 ;
				CloseHandle(m_hIOCP);
			}
		}
		else
			OnError(NULL,GetLastError());
			return false;
	}

/***************************************************
GetClientAddress
    Returns the addess of the connected client
Params
    address - buffer for the address
    maxlen - length of the buffer
Return
    UTE_SUCCESS				- success
    UTE_BUFFER_TOO_SHORT	- buffer too short
    UTE_NAME_LOOKUP_FAILED	- failure to resolve name
****************************************************/
int CUT_WSThread::GetClientAddress(LPSTR address,int maxlen){

    char *buf = inet_ntoa(m_clientSocketAddr.sin_addr);
    
    if(buf == NULL)
        return OnError(UTE_NAME_LOOKUP_FAILED);

    if(strlen(buf) > (unsigned int) (maxlen-1) )
        return OnError(UTE_BUFFER_TOO_SHORT);

    strcpy(address, buf);

    return OnError(UTE_SUCCESS);
}

bool_t SyncSession::OnRead(EventResult *evr)
{
	bool_t ret = FALSE;

	if(sock_.Status() != Socket::STATUS_CONNECTED)
	{
		ECILA_ERROR(("[SyncSession::OnWrite] "ECILA_FI64U"\n", UIDX())); return FALSE;
	}

	char_t *buffer = recv_buffer_->GetInPtr();
	int32_t remain = recv_buffer_->Remain();

	int32_t nbytes = sock_.Recv((char_t*)buffer, remain);
	if(nbytes == Socket::ERR_TRANSFER)
	{
		return OnError(evr);
	}
	
	ret = recv_buffer_->ForwardInPtr(nbytes);
	if(!ret) { ECILA_TRACE(); return FALSE; }

	ret = pkt_handler_->HandleEvents(this->UIDX(), this, EVENT_READ);
	if(!ret) { ECILA_TRACE(); return FALSE; }

	if(recv_buffer_->Length() == 0) { recv_buffer_->Reset(); }

	return TRUE;
}

// 다수의 워커스레드에서 해당 커넥션(소켓)으로부터 데이타를 받으면 호출됨
bool XEWinConnectionInServer::tRecvData( DWORD readbytes )
{
	bool bOk = true;
	do {
// 		if( IsDisconnected() == TRUE )		// 이런경우가 있음(언젠진 모름)
// 			break;
// 		if( IsDestroy() )
// 			break;
		// 이전에 m_Buffer에 받았던데이타가 아직 큐에 밀어넣지 못한채로 또 불렸다.
		// 받은 데이터를 큐에 밀어넣음.
		int bytePushed = XENetworkConnection::PushBlock( m_Buffer, readbytes );	// 최적화를 위해 만듬. 큐에 하나씩 보내지않고 뭉탱이로 보냄
		const int sizeQ = GetSizeQueue();
		if( sizeQ > GetsizeMaxQ() )
			SetsizeMaxQ( sizeQ );
		// 만약 큐가 꽉차서 더이상 못밀어 넣었다면?
		// 큐는 넉넉하게 잡아놔야겠지만 그것마저 꽉 채울정도라면 비정상으로 보고 끊어야 하지 않을까 싶다.
		// 일단은 로그라도 남겨야 한다.
		// 밀어넣은데이타와 실제 푸쉬된 데이타의 크기가 다르면 에러.
// 		XBREAKF( bytePushed != readbytes, "bytePushed(%d) != readBytes(%d): ip=%s", 
// 														bytePushed, readbytes, m_szIP );
		if( bytePushed != readbytes )	{
			OnError( XENetworkConnection::xERR_Q_IS_FULL, sizeQ );
			DoDisconnect();
			bOk = false;
		}
// 		m_bFlush = TRUE;
	} while(0);
	return bOk;
}

void ASyncTCP::write_end(const boost::system::error_code& error)
{
	if (!mIsClosing)
	{
		if (error)
		{
			// let listeners know
			OnError(error);

			mIsConnected = false;

			if (!mDoReconnect)
			{
				OnDisconnect();
				return;
			}
			if (!mIsReconnecting)
			{
				mIsReconnecting = true;
				_log.Log(LOG_STATUS, "TCP: Reconnecting in %d seconds...", RECONNECT_TIME);
				// schedule a timer to reconnect after 30 seconds
				mReconnectTimer.expires_from_now(boost::posix_time::seconds(RECONNECT_TIME));
				mReconnectTimer.async_wait(boost::bind(&ASyncTCP::do_reconnect, this, boost::asio::placeholders::error));
			}
		}
	}
}

ssize_t
DaemonSocketPDU::Send(int aFd)
{
  struct iovec iv;
  memset(&iv, 0, sizeof(iv));
  iv.iov_base = GetData(GetLeadingSpace());
  iv.iov_len = GetSize();

  struct msghdr msg;
  memset(&msg, 0, sizeof(msg));
  msg.msg_iov = &iv;
  msg.msg_iovlen = 1;
  msg.msg_control = nullptr;
  msg.msg_controllen = 0;

  ssize_t res = TEMP_FAILURE_RETRY(sendmsg(aFd, &msg, 0));
  if (res < 0) {
    MOZ_ASSERT(errno != EBADF); /* internal error */
    OnError("sendmsg", errno);
    return -1;
  }

  Consume(res);

  if (mConsumer) {
    // We successfully sent a PDU, now store the
    // result runnable in the consumer.
    mConsumer->StoreUserData(*this);
  }

  return res;
}

OP_STATUS
OpScopeTPReader::SetError(Error err)
{
    error = err;
    OnError(err);
    return OpStatus::ERR;
}

void ConnectSocketChannel::Connect()
{
	RETURN_IF_EQUAL(mState, ConnectionState::Connected);
	mState = ConnectionState::Connecting;
	mLoop->AssertInLoopThread();
	mSocket.CreateSocket();
	mSocket.EnableAsync(true);
	mSocket.EnableTcpNoDelay(true);

	ChannelEventResult connectResult = mSocket.Connect();	//connect timeout
	switch (connectResult)
	{
	case ChannelEventResult::Success:
		mState = ConnectionState::Connected;
		return;
	case ChannelEventResult::InProgress:
		mState = ConnectionState::Connecting;
		BeginWrite();
		SubmitToLoop();
		break;
	case ChannelEventResult::Retry:
		Retry();
		break;
	case ChannelEventResult::Fault:
	default:
		OnError();
		break;
	}

}