hxnetapi.cpp

linux下的一款播放器

		m_pInterruptSafeMacWriteQueue->AddTail(pBuffer); // AddRef called inside
#else
	pBuffer->AddRef();
	m_PendingWriteBuffers.AddTail((void*) pBuffer);
    /* Transfer pending buffers to TCP send queue */
    TransferBuffers();
#endif

	m_pMutex->Lock();
    theErr = DoWrite();
    m_pMutex->Unlock();
    lResult = ConvertNetworkError(theErr);

    return lResult;
}

STDMETHODIMP HXTCPSocket::WantWrite()
{
    if (mSendTCP->GetQueuedItemCount() == 0)
    {
	m_pTCPResponse->WriteReady(HXR_OK);
    }
    else
    {
	m_bWantWritePending = TRUE;
    }

    return HXR_OK;
}

STDMETHODIMP HXTCPSocket::GetLocalAddress(ULONG32& lAddress)
{
    return HXR_NOTIMPL;
}

STDMETHODIMP HXTCPSocket::GetForeignAddress(ULONG32& lAddress)
{
    if(m_bConnected && m_lForeignAddress)
    {
	lAddress = m_lForeignAddress;
	return HXR_OK;
    }
    return HXR_FAIL;
}

STDMETHODIMP HXTCPSocket::GetLocalPort(UINT16& port)
{
    return HXR_NOTIMPL;
}

STDMETHODIMP HXTCPSocket::GetForeignPort(UINT16& port)
{
    if(m_bConnected)
    {
	port = m_nForeignPort;
	return HXR_OK;
    }
    return HXR_FAIL;
}

// the tcp socket will still need to be inited
STDMETHODIMP
HXTCPSocket::AcceptConnection(conn* pNewConn)
{
    HX_ASSERT(!m_bConnected);
    HX_ASSERT(!m_bInitComplete);
    HX_ASSERT(m_pCtrl == NULL);
    m_pCtrl = pNewConn;
    m_pCtrl->AddRef();
    m_pCtrl->nonblocking();

    if ( m_pCallback )
    {
	HX_DELETE(m_pCallback);
    }
    m_pCallback = new TCPSocketCallback;
    if ( !m_pCallback )
    {
	return HXR_OUTOFMEMORY;
    }
    m_pCallback->m_pContext = this;
    m_pCtrl->set_callback(m_pCallback);
    m_lForeignAddress = DwToHost(m_pCtrl->get_addr());

    m_bInitComplete = TRUE;
    m_bConnected = TRUE;

    return HXR_OK;
}

HX_RESULT
HXTCPSocket::DoRead()
{
#ifdef _MACINTOSH
	if (m_bInDoRead)
	{
		return HXR_OK; // whatever needs to be done will be done by the caller that's already here.

		// xxxbobclark the m_bInDoRead flag is hacked around calling ReadDone(), because
		//             ReadDone() may call Read() which in turn calls us here, and we do
		//             not want to bail out in that instance. (Otherwise we only remove
		//             one packet at a time, which, given the scheduler granularity and
		//             high bit rates, implies that our bandwidth would be too low.)
	}
#endif
	m_bInDoRead = TRUE;

    HX_RESULT theErr = HXR_OK;
    // check how much room we have in TCP receive queue
    UINT16 count = mReceiveTCP->GetMaxAvailableElements();

    if (count > 0)
    {
#if !defined(THREADS_SUPPORTED) && !defined(_UNIX_THREADED_NETWORK_IO)
	UINT32 ulBytesToRead = conn::bytes_to_preparetcpread(m_pCtrl);

	if (ulBytesToRead > 0)
	{
	    if ((UINT32)count > ulBytesToRead)
	    {
		count = (UINT16)ulBytesToRead;
	    }

	    // attempt to read data from TCP link
	    theErr = m_pCtrl->read(m_pBuffer, &count);
	    if (!theErr && count > 0)
	    {
		conn::bytes_to_actualtcpread(m_pCtrl, (UINT32)count);
		mReceiveTCP->EnQueue(m_pBuffer, count);
	    }
	    else if (theErr)
	    {
		theErr = ConvertNetworkError(theErr);
	    }
	}
#elif defined(_UNIX_THREADED_NETWORK_IO)
        //XXXgfw duplicated code. Clean this up...
        if( ReadNetworkThreadingPref((IUnknown*)m_pContext) )
        {
            // in THREADS_SUPPORTED mode, this will be taken care by the thrdconn.cpp
            // attempt to read data from TCP link
            theErr = m_pCtrl->read(m_pBuffer, &count);
            if (!theErr && count > 0)
            {
                mReceiveTCP->EnQueue(m_pBuffer, count);
            }
            else if (theErr)
            {
                theErr = ConvertNetworkError(theErr);
            }
        }
        else
        {
            UINT32 ulBytesToRead = conn::bytes_to_preparetcpread(m_pCtrl);

            if (ulBytesToRead > 0)
            {
                if ((UINT32)count > ulBytesToRead)
                {
                    count = (UINT16)ulBytesToRead;
                }

                // attempt to read data from TCP link
                theErr = m_pCtrl->read(m_pBuffer, &count);
                if (!theErr && count > 0)
                {
                    conn::bytes_to_actualtcpread(m_pCtrl, (UINT32)count);
                    mReceiveTCP->EnQueue(m_pBuffer, count);
                }
                else if (theErr)
                {
                    theErr = ConvertNetworkError(theErr);
                }
            }

        }
#else
	// in THREADS_SUPPORTED mode, this will be taken care by the thrdconn.cpp
	// attempt to read data from TCP link
	theErr = m_pCtrl->read(m_pBuffer, &count);
	if (!theErr && count > 0)
	{
	    mReceiveTCP->EnQueue(m_pBuffer, count);
	}
	else if (theErr)
	{
	    theErr = ConvertNetworkError(theErr);
	}
#endif /* !THREADS_SUPPORTED */
    }

    count = mReceiveTCP->GetQueuedItemCount();
    if (m_bReadPending && count > 0)
    {
	/* If we are at interrupt time and the response object is not interrupt safe,
	 * schedule a callback to return back the data
	 */
	if (!IsSafe())
	{
		m_bInDoRead = FALSE;
	    return HXR_OK;
	}

	m_bReadPending = FALSE;
	if (m_nRequired < count)
	{
	    // XXXAAK -- UINT32 down to UINT16 - possible truncation???
	    count = (UINT16)m_nRequired;
	}

	CHXBuffer* pBuffer = new CHXBuffer;
	pBuffer->AddRef();
	mReceiveTCP->DeQueue(m_pBuffer, count);
	pBuffer->Set((UCHAR*) m_pBuffer, count);

	m_bInDoRead = FALSE;
	theErr = m_pTCPResponse->ReadDone(HXR_OK, pBuffer);
	m_bInDoRead = TRUE;

	pBuffer->Release();

	/* mask any kind of errors */
        // Huh??? Don't mask OUTOFMEMORY errors!
        if( theErr != HXR_OUTOFMEMORY )
        {
	    theErr = HXR_OK;
        }
    }

    if (theErr && m_bReadPending)
    {
	/* If we are at interrupt time and the response object is not interrupt safe,
	 * schedule a callback to return back the data
	 */
	if (!IsSafe())
	{
		m_bInDoRead = FALSE;
	    return HXR_OK;
	}

#ifdef _MACINTOSH
	if (m_pMacCommandCallback && m_pMacCommandCallback->ScheduleCallback(TCP_READ_DONE_COMMAND, m_pScheduler, 0, theErr))
	{
	    m_bReadPending = FALSE;
    	    m_bInDoRead = FALSE;
	    return HXR_OK;
	}
	else
	{
	    // failed to schedule a callback, notify the responser with error directly
	    m_bReadPending = FALSE;
	    m_pTCPResponse->ReadDone(theErr, NULL);
	}
#else
	m_bReadPending = FALSE;
	m_pTCPResponse->ReadDone(theErr, NULL);
#endif
    }

    if (!theErr		    &&
	m_bReadPending	    &&
	m_pSchedulerReadCallback)
	{
	    m_pSchedulerReadCallback->ScheduleCallback(TCP_READ_COMMAND, m_pScheduler, SCHED_GRANULARITY);
    }

 	m_bInDoRead = FALSE;
    return theErr;
}

HX_RESULT
HXTCPSocket::DoWrite()
{
    HX_RESULT theErr = HXR_OK;

    if (m_bInWrite) return HXR_OK;
    m_bInWrite = TRUE;

#ifdef _MACINTOSH
	if (m_pInterruptSafeMacWriteQueue)
		m_pInterruptSafeMacWriteQueue->TransferToSimpleList(m_PendingWriteBuffers);
	TransferBuffers(); // PENDING_BUFFERS_ARE_EMPTIED_AT_START_OF_DO_WRITE
#endif

    // check how data we have in TCP send queue
    UINT16 count    = mSendTCP->GetQueuedItemCount();
    UINT16 actual   = count;

    if(count > 0)
    {
	mSendTCP->DeQueue(m_pBuffer,count);
	theErr = m_pCtrl->write(m_pBuffer, &actual);
    }

    switch(theErr)
    {
	case HXR_AT_INTERRUPT:
	case HXR_WOULD_BLOCK:
	case HXR_OK:
	    // enqueue the data that was not sent
	    if(actual != count)
	    {
		mSendTCP->EnQueue(m_pBuffer + actual, count - actual);
	    }

	    // mask out these errors
	    theErr = HXR_OK;
	    break;
	default:
	    theErr = ConvertNetworkError(theErr);
	    break;
    }

    if (!theErr && m_bWantWritePending && mSendTCP->GetQueuedItemCount() == 0)
    {
	m_bWantWritePending = FALSE;
	m_pTCPResponse->WriteReady(HXR_OK);
    }

#ifndef _MACINTOSH
	// m_PendingWriteBuffers will always be empty due to the full buffer transfer at the top of this routine.
	// see PENDING_BUFFERS_ARE_EMPTIED_AT_START_OF_DO_WRITE
    if (!theErr && m_PendingWriteBuffers.GetCount()  > 0)
    {
	TransferBuffers();
    }
#endif

    if (!theErr &&
	((mSendTCP && mSendTCP->GetQueuedItemCount() > 0) ||
	m_PendingWriteBuffers.GetCount() > 0)) // see PENDING_BUFFERS_ARE_EMPTIED_AT_START_OF_DO_WRITE
    {
	if (m_pSchedulerWriteCallback)
	{
	    m_pSchedulerWriteCallback->ScheduleCallback(TCP_WRITE_COMMAND, m_pScheduler, SCHED_GRANULARITY);
	}
    }

    if (m_bWriteFlushPending &&
	((mSendTCP->GetQueuedItemCount() == 0 &&
	 m_PendingWriteBuffers.GetCount() == 0)	||
	 theErr))
    {
	m_bWriteFlushPending	= FALSE;
	Release();
    }
    else if (!theErr && !m_bWriteFlushPending &&
	(mSendTCP->GetQueuedItemCount() > 0 || m_PendingWriteBuffers.GetCount() > 0))
    {
	m_bWriteFlushPending	= TRUE;
	AddRef();
    }

    m_bInWrite = FALSE;
    return theErr;
}

/* If we are at interrupt time and the response object is not interrupt safe,
 * schedule a callback to return back the data
 */
BOOL
HXTCPSocket::IsSafe()
{
    if (m_pInterruptState && m_pInterruptState->AtInterruptTime() &&
	(!m_pResponseInterruptSafe ||
	 !m_pResponseInterruptSafe->IsInterruptSafe()))
    {
	if (m_pNonInterruptReadCallback)
	{
	    m_pNonInterruptReadCallback->ScheduleCallback(TCP_READ_COMMAND, m_pScheduler, 0);
	}

	return FALSE;
    }

    return TRUE;
}


void
HXTCPSocket::ConnectDone(BOOL bResult)
{
    AddRef();
    if (bResult == TRUE)
    {
	m_bConnected = TRUE;
	//XXX need to set m_lForeignAddr here
	//XXXJR hack!
	m_lForeignAddress = DwToHost(m_pCtrl->get_addr());
	m_pTCPResponse->ConnectDone(HXR_OK);
    }
    else
    {
#ifdef _MACINTOSH
	if (!(m_pMacCommandCallback && m_pMacCommandCallback->ScheduleCallback(TCP_CONNECT_DONE_COMMAND, m_pScheduler, 0, HXR_NET_CONNECT)))
	{
		//note: only happens when there's a problem (e.g. macleod 1/2 server problem)
		m_pTCPResponse->ConnectDone(HXR_NET_CONNECT); // couldn't use the delayed callback... take our chances.
	}
#else
	m_pTCPResponse->ConnectDone(HXR_NET_CONNECT);
#endif
    }
    Release();
}

void
HXTCPSocket::CloseDone()
{
    m_pTCPResponse->Closed(HXR_OK);
}

void
HXTCPSocket::DNSDone(BOOL bSuccess)
{
    AddRef();
    if (!bSuccess)
    {
	m_pTCPResponse->ConnectDone(HXR_DNR);
    }
    Release();
}

void
HXTCPSocket::TransferBuffers()
{
    IHXBuffer* pBuffer = 0;

    while (m_PendingWriteBuffers.GetCount() > 0)
    {
	pBuffer = (IHXBuffer*) m_PendingWriteBuffers.GetHead();
	if ((UINT16) pBuffer->GetSize() < mSendTCP->GetMaxAvailableElements())
	{
	    mSendTCP->EnQueue(	pBuffer->GetBuffer(),
				(UINT16) pBuffer->GetSize());
	    pBuffer->Release();
	    m_PendingWriteBuffers.RemoveHead();
	}
	else
	{
	    break;
	}
    }
}

STDMETHODIMP
HXTCPSocket::SetOption(HX_SOCKET_OPTION option, UINT32 ulValue)
{
    HX_RESULT res = HXR_OK;

    switch(option)
    {
    case HX_SOCKOPT_REUSE_ADDR:
	m_bReuseAddr = (BOOL)ulValue;
    break;
    case HX_SOCKOPT_REUSE_PORT:
	m_bReusePort = (BOOL)ulValue;
    break;
    case HX_SOCKOPT_MULTICAST_IF:
	res = HXR_UNEXPECTED;
    break;
    default:
	HX_ASSERT(!"I don't know this option");
	res = HXR_FAIL;
    }

    return res;
}

STDMETHODIMP
HXTCPSocket::SetSecure(BOOL bSecure)
{
    HX_RESULT res = HXR_OK;

    m_bSecureSocket = bSecure;

    return res;
}

STDMETHODIMP HXTCPSocket::HandleCallback(INT32 theCommand, HX_RESULT theError)
{
    HX_RESULT theErr = HXR_OK;
    if (!m_bInDestructor)
    {
	AddRef();
	m_pMutex->Lock();
	if (!m_bInDestructor)
	{
		switch(theCommand)
		{
			case TCP_READ_COMMAND:
			    theErr = DoRead();
			    break;

			case TCP_WRITE_COMMAND:
			    DoWrite(); // protected from re-entry by m_bInWrite
			    break;

			case TCP_READ_DONE_COMMAND:
		    	m_bReadPending = FALSE;
				m_pTCPResponse->ReadDone(theError, NULL);
			    break;

			case TCP_CONNECT_DONE_COMMAND:
				m_pTCPResponse->ConnectDone(theError);
				break;

			case TCP_BIND_COMMAND:
			default:
			    theErr = DoRead();
			    DoWrite();
			    break;
	   	}
	}
	m_pMutex->Unlock();

        // we want out of memory errors to be reported immediately
        // because fiddling around waiting for the error to propagate
        // normally will just make the situation worse; mask out all
        // other errors, as they will eventually get dealt with in
        // ReadDone() or similar functions.
        
        if (theErr != HXR_OUTOFMEMORY)
        {
            theErr = HXR_OK;
        }
        
	if( theErr )
	{
	    IHXErrorMessages * pErrorNotifier = NULL;
	    IUnknown * pPlayer = NULL;
	    IHXClientEngine* pEngine = NULL;
	    UINT32 nNumPlayers = 0;

	    m_pContext->QueryInterface(IID_IHXClientEngine, (void**)&pEngine);
	    if( pEngine )
	    {
		nNumPlayers = pEngine->GetPlayerCount();
		for( int ii=0; ii<nNumPlayers; ii++ )
		{
		    pEngine->GetPlayer(ii,pPlayer);
		    if( pPlayer )
		    {
			pPlayer->QueryInterface( IID_IHXErrorMessages, (void**)&pErrorNotifier );
		    }
		    if( pErrorNotifier )
		    {
			pErrorNotifier->Report( HXLOG_ERR, theErr, 0, NULL, NULL );
			pErrorNotifier->Release();
		    }
		    HX_RELEASE( pPlayer );
		}
	    }