thrdconn.cpp

linux下的一款播放器

        /* Signal the main thread that intialization is complete */
        m_pInitEvent->SignalEvent();
        m_pMutex->Unlock();
    }

    return theErr;
}

HX_RESULT
ThreadedConn::listen(ULONG32 ulAddr,  UINT16 port, UINT16 backlog,
                     UINT16 blocking, ULONG32 ulPlatform)
{
    HX_RESULT theErr = HXR_OK;

#if (defined (_WIN32) || defined (WIN32)) && !defined(WIN32_PLATFORM_PSPC)
    m_ulUserHandle  = ulPlatform;

    if (!m_pNotifier)
    {
        m_pNotifier =
            CAsyncNetThread::GetCAsyncNetThreadNotifier((HINSTANCE)ulPlatform,
                                                        TRUE);
    }
    if (m_pNotifier)
    {
        m_pInternalWindowHandle = (void*) m_pNotifier->GetWindowHandle();
        m_pNotifier->AttachSocket(this);
    }
    else
    {
        theErr = HXR_OUTOFMEMORY;
    }
#endif /*defined (_WIN32) || defined (WIN32)*/

    HX_DELETE(m_pIncommingConnections);
    m_pIncommingConnections = new CHXSimpleList();

    if (!theErr)
    {
        ThrdConnGenericCallback* pCallback      =
                new ThrdConnGenericCallback(this, LISTEN_CALLBACK_TYPE);
        pCallback->m_ulLocalAddr = ulAddr;
        pCallback->m_uPort      = port;
        pCallback->m_uBacklog   = backlog;
        pCallback->m_bBlocking  = (BOOL) blocking;
        pCallback->m_ulHandle   = ulPlatform;
        /* Will be released by the thread engine */
        pCallback->AddRef();
        HXThreadMessage msg(HXMSG_ASYNC_CALLBACK, this, pCallback);
        theErr = m_pNetworkThread->PostMessage(&msg);

        if ( SUCCEEDED(theErr) )
        {
            /*
             * Wait for the actual Listen to complete.
             */
	    // listen is called from the network thread, so this will wait forever
#ifdef HELIX_FEATURE_NETWORK_USE_SELECT
	    m_bListenning = TRUE;
#else
            m_pListenEvent->Wait();
#endif

            if ( m_bListenning )
            {
                theErr = HXR_OK;
            }
            else
            {
                theErr = HXR_FAIL;
            }
        }
    }
#if defined(HELIX_FEATURE_NETWORK_USE_SELECT)
    set_callback(m_pNetCallback); // for accept msg
#endif //HELIX_FEATURE_NETWORK_USE_SELECT
    return theErr;
}

HX_RESULT
ThreadedConn::ActualListen( ULONG32     ulAddr,
                            UINT16      port,
                            UINT16      backlog,
                            UINT16      blocking,
                            ULONG32     ulPlatform)
{
    HX_RESULT err = HXR_UNEXPECTED;
    HX_ASSERT(m_pActualConn);
    if ( m_pActualConn )
    {
        m_pMutex->Lock();
        err = m_pActualConn->listen(ulAddr, port, backlog, blocking, ulPlatform);

        if ( SUCCEEDED(err) )
        {
            m_bListenning = TRUE;
        }
        else
        {
            m_bListenning = FALSE;
        }

        m_pListenEvent->SignalEvent();
        m_pMutex->Unlock();
    }
    return err;
}

#if     defined (_WINDOWS) || defined (_WIN32)
// we need it for dns_find_ip_addr since async stuff needs a window handle...
HX_RESULT
ThreadedConn::SetWindowHandle(ULONG32 handle)
{
    HX_RESULT theErr = HXR_OK;

    m_pMutex->Lock();
    m_ulUserHandle  = handle;
#if (defined (_WIN32) || defined (WIN32)) && !defined(WIN32_PLATFORM_PSPC)
    if (!m_pNotifier)
    {
        m_pNotifier =
            CAsyncNetThread::GetCAsyncNetThreadNotifier((HINSTANCE)handle,
                                                        TRUE);
    }
    if (m_pNotifier)
    {
        m_pInternalWindowHandle = (void*)m_pNotifier->GetWindowHandle();
        m_pNotifier->AttachSocket(this);
    }
    else
    {
        theErr = HXR_OUTOFMEMORY;
    }
#endif /*defined (_WIN32) || defined (WIN32)*/

    if (!theErr)
    {
        ThrdConnGenericCallback* pCallback      =
                new ThrdConnGenericCallback(this, SETWINDOWHANDLE_CALLBACK_TYPE);
        pCallback->m_ulHandle   = handle;
        /* Will be released by the thread engine */
        pCallback->AddRef();
        HXThreadMessage msg(HXMSG_ASYNC_CALLBACK, this, pCallback);
        theErr = m_pNetworkThread->PostMessage(&msg);
    }

    m_pMutex->Unlock();

    return theErr;
}


HX_RESULT
ThreadedConn::ActuallSetWindowHandle(ULONG32 handle)
{
    HX_RESULT theErr = HXR_UNEXPECTED;

    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pMutex->Lock();
        theErr = m_pActualConn->SetWindowHandle(handle);
        if (!theErr)
        {
        }
        m_pMutex->Unlock();
    }

    return theErr;
}

#endif /* defined (_WINDOWS) || defined (_WIN32)*/

HX_RESULT
ThreadedConn::connect(const char*   host,
                     UINT16         port,
                     UINT16         blocking,
                     ULONG32        ulPlatform)
{
    HX_RESULT theErr = HXR_OK;

#if (defined (_WIN32) || defined (WIN32)) && !defined(WIN32_PLATFORM_PSPC)
    m_ulUserHandle  = ulPlatform;

    if (!m_pNotifier)
    {
        m_pNotifier =
            CAsyncNetThread::GetCAsyncNetThreadNotifier((HINSTANCE)ulPlatform,
                                                        TRUE);
    }
    if (m_pNotifier)
    {
        m_pInternalWindowHandle = (void*) m_pNotifier->GetWindowHandle();
        m_pNotifier->AttachSocket(this);
    }
    else
    {
        theErr = HXR_OUTOFMEMORY;
    }
#endif /*defined (_WIN32) || defined (WIN32)*/

    if (!theErr)
    {
        ThrdConnGenericCallback* pCallback      =
                new ThrdConnGenericCallback(this, CONNECT_CALLBACK_TYPE);
        pCallback->m_HostName   = host;
        pCallback->m_uPort              = port;
#if defined(HELIX_FEATURE_NETWORK_USE_SELECT)
	blocking = 1; // we don't get fd_connect when doing loopback
#endif //defined(HELIX_FEATURE_NETWORK_USE_SELECT)
        pCallback->m_bBlocking  = (BOOL) blocking;
        pCallback->m_ulHandle   = ulPlatform;
        /* Will be released by the thread engine */
        pCallback->AddRef();
        HXThreadMessage msg(HXMSG_ASYNC_CALLBACK, this, pCallback);
        theErr = m_pNetworkThread->PostMessage(&msg);
    }

    return theErr;
}

HX_RESULT
ThreadedConn::ActualConnect(const char* host,
                            UINT16      port,
                            UINT16      blocking,
                            ULONG32     ulPlatform)
{
    HX_RESULT theErr = HXR_UNEXPECTED;
    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pMutex->Lock();
        theErr = m_pActualConn->connect(host, port, blocking, ulPlatform);
        m_pMutex->Unlock();
    }

    return theErr;
}

#if defined(HELIX_FEATURE_NETWORK_USE_SELECT)
HX_RESULT
ThreadedConn::accept (ULONG32 ulAddr)
{
    HX_RESULT theErr = HXR_OK;

#if (defined (_WIN32) || defined (WIN32)) && !defined(WIN32_PLATFORM_PSPC)
    m_ulUserHandle  = NULL; //ulPlatform;

    if (!m_pNotifier)
    {
        m_pNotifier =
            CAsyncNetThread::GetCAsyncNetThreadNotifier((HINSTANCE)m_ulUserHandle,
                                                        TRUE);
    }
    if (m_pNotifier)
    {
        m_pInternalWindowHandle = (void*) m_pNotifier->GetWindowHandle();
        m_pNotifier->AttachSocket(this);
    }
    else
    {
        theErr = HXR_OUTOFMEMORY;
    }
#endif /*defined (_WIN32) || defined (WIN32)*/

    if (!theErr)
    {
        ThrdConnGenericCallback* pCallback      =
                new ThrdConnGenericCallback(this, ACCEPT_CALLBACK_TYPE);
        pCallback->m_ulHandle   = m_ulUserHandle;
	pCallback->m_ulLocalAddr = ulAddr;
        /* Will be released by the thread engine */
        pCallback->AddRef();
        HXThreadMessage msg(HXMSG_ASYNC_CALLBACK, this, pCallback);
        theErr = m_pNetworkThread->PostMessage(&msg);
    }

    return theErr;
}

HX_RESULT
ThreadedConn::ActualAccept(ULONG32 ulAddr,
                            ULONG32     ulPlatform)
{
    HX_RESULT theErr = HXR_UNEXPECTED;
    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pMutex->Lock();
        theErr = m_pActualConn->CheckForConnection();
        m_pMutex->Unlock();
	if (theErr == HXR_WOULD_BLOCK)
	{
	    ThrdConnGenericCallback* pCallback      =
		    new ThrdConnGenericCallback(this, ACCEPT_CALLBACK_TYPE);
	    pCallback->m_ulHandle   = ulPlatform;
	    pCallback->m_ulLocalAddr = ulAddr;
	    /* Will be released by the thread engine */
	    pCallback->AddRef();
	    HXThreadMessage msg(HXMSG_ASYNC_CALLBACK, this, pCallback);
	    theErr = m_pNetworkThread->PostMessage(&msg);
	}
    }

    return theErr;
}
#endif //defined(HELIX_FEATURE_NETWORK_USE_SELECT)

HX_RESULT
ThreadedConn::blocking(void)
{
    ThrdConnGenericCallback* pCallback  = new ThrdConnGenericCallback(this, BLOCKING_CALLBACK_TYPE);
    /* Will be released by the thread engine */
    pCallback->AddRef();
    HXThreadMessage msg(HXMSG_ASYNC_CALLBACK, this, pCallback);
    return m_pNetworkThread->PostMessage(&msg);
}

HX_RESULT
ThreadedConn::ActualBlocking(void)
{
    HX_RESULT theErr = HXR_UNEXPECTED;
    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pMutex->Lock();
        theErr = m_pActualConn->blocking();
        m_pMutex->Unlock();
    }

    return theErr;
}

HX_RESULT
ThreadedConn::nonblocking(void)
{
    ThrdConnGenericCallback* pCallback  = new ThrdConnGenericCallback(this, NONBLOCKING_CALLBACK_TYPE);
    /* Will be released by the thread engine */
    pCallback->AddRef();
    HXThreadMessage msg(HXMSG_ASYNC_CALLBACK, this, pCallback);
    return m_pNetworkThread->PostMessage(&msg);
}

HX_RESULT
ThreadedConn::ActualNonBlocking(void)
{
    HX_RESULT theErr = HXR_UNEXPECTED;
    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pMutex->Lock();
        theErr = m_pActualConn->nonblocking();
        m_pMutex->Unlock();
    }

    return theErr;
}


HX_RESULT
ThreadedConn::read (void* buf, UINT16* size)
{
    HX_RESULT theErr = HXR_OK;
    UINT16 uCount = 0;

    m_pMutex->Lock();
    m_bOutstandingReadNotification = FALSE;

    if (m_uSocketType != HX_TCP_SOCKET)
    {
        theErr = HXR_NET_SOCKET_INVALID;
        goto cleanup;
    }

    uCount = m_pReceiveTCP->GetQueuedItemCount();
    if (uCount > 0)
    {
        uCount = (uCount <= *size ? uCount : *size);
        m_pReceiveTCP->DeQueue(buf, uCount);
        *size = uCount;
    }
    else
    {
        *size = 0;
        if (!mLastError && m_pActualConn)
        {
            theErr = HXR_WOULD_BLOCK;
        }
        else
        {
            theErr = mLastError;
        }
    }

cleanup:

    if (!mLastError && !m_bNetworkIOPending)
    {
        theErr = PostIOMessage();
    }

    m_pMutex->Unlock();

    return theErr;
}

HX_RESULT
ThreadedConn::readfrom (REF(IHXBuffer*)    pBuffer,
                        REF(UINT32)         ulAddress,
                        REF(UINT16)         ulPort)
{
    HX_RESULT theErr = HXR_OK;
    UDP_PACKET* pPacket = NULL;

    m_pMutex->Lock();
    m_bOutstandingReadNotification = FALSE;

    pBuffer = NULL;
    ulAddress = 0;
    ulPort = 0;

    if (m_uSocketType != HX_UDP_SOCKET)
    {
        theErr = HXR_NET_SOCKET_INVALID;
        goto cleanup;
    }

    if (m_ReadUDPBuffers.GetCount() > 0)
    {
        pPacket = (UDP_PACKET*)m_ReadUDPBuffers.RemoveHead();;

        pBuffer = pPacket->pBuffer;
        ulAddress = pPacket->ulAddress;
        ulPort = pPacket->ulPort;

        HX_DELETE(pPacket);
    }
    else
    {
        if (!mLastError)
        {
            theErr = HXR_WOULD_BLOCK;
        }
        else
        {
            theErr = mLastError;
        }
    }

cleanup:

    if (!m_bReadPostPendingWouldBlock && !mLastError && !m_bNetworkIOPending)
    {
        theErr = PostIOMessage();
    }

    m_pMutex->Unlock();

    if( mLastError == HXR_OUTOFMEMORY )
    {
        theErr = mLastError;
    }
    return theErr;
}

HX_RESULT
ThreadedConn::write (void* buf, UINT16* size)
{
    HX_RESULT theErr = HXR_OK;
    HX_ASSERT(m_pActualConn && m_uSocketType == HX_TCP_SOCKET);

    m_pMutex->Lock();
    m_bOutstandingWriteNotification = FALSE;
    UINT16 uCount = m_pSendTCP->GetMaxAvailableElements();
    if (uCount > 0)
    {
        uCount = (uCount <= *size ? uCount : *size);
        m_pSendTCP->EnQueue(buf, uCount);
        *size = uCount;
    }
    else
    {
        *size = 0;
        if (!mLastError)
        {
            theErr = HXR_WOULD_BLOCK;
        }
        else
        {
            theErr = mLastError;
        }