thrdconn.cpp

linux下的一款播放器

    }

    if (!m_bWriteFlushPending && m_pSendTCP->GetQueuedItemCount() > 0 && m_bConnected)
    {
        m_bWriteFlushPending    = TRUE;
        AddRef();
    }

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

    m_pMutex->Unlock();

    return theErr;
}

HX_RESULT
ThreadedConn::writeto(void* buf,UINT16* len, ULONG32 addr, UINT16 port)
{
    HX_RESULT theErr = HXR_OK;
    HX_ASSERT(m_pActualConn && m_uSocketType == HX_UDP_SOCKET);

    m_pMutex->Lock();
    m_bOutstandingWriteNotification = FALSE;
    if (!mLastError)
    {
        UDPPacketInfo* pPacket  = new UDPPacketInfo;
        pPacket->m_pBuffer  = new CHXBuffer;
        pPacket->m_pBuffer->AddRef();
        pPacket->m_pBuffer->Set((UCHAR*) buf, (ULONG32) *len);
        pPacket->m_ulAddr   = addr;
        pPacket->m_uPort    = port;
        m_WriteUDPBuffers.AddTail((void*) pPacket);
    }
    else
    {
        theErr = mLastError;
    }

    if (!theErr && !m_bWriteFlushPending && m_WriteUDPBuffers.GetCount() > 0 && m_bConnected)
    {
        m_bWriteFlushPending    = TRUE;
        AddRef();
    }

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

    m_pMutex->Unlock();

    return theErr;
}

ULONG32
ThreadedConn::get_addr(void)
{
    ULONG32 ulAddr = 0;
    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pMutex->Lock();
        ulAddr = m_pActualConn->get_addr();
        m_pMutex->Unlock();
    }

    return ulAddr;
}

UINT16
ThreadedConn::get_local_port(void)
{
    UINT16 nPort = 0;
    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pMutex->Lock();
        nPort = m_pActualConn->get_local_port();
        m_pMutex->Unlock();
    }

    return nPort;
}


/* join_multicast_group() has this socket join a multicast group */
HX_RESULT
ThreadedConn::join_multicast_group(ULONG32 addr, ULONG32 if_addr)
{
    HX_RESULT theErr = HXR_UNEXPECTED;
    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pMutex->Lock();
        theErr = m_pActualConn->join_multicast_group(addr, if_addr);
        m_pMutex->Unlock();
    }

    return theErr;
}

HX_RESULT
ThreadedConn::leave_multicast_group(ULONG32 addr, ULONG32 if_addr)
{
    HX_RESULT theErr = HXR_UNEXPECTED;
    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pMutex->Lock();
        theErr = m_pActualConn->leave_multicast_group(addr, if_addr);
        while (!m_ReadUDPBuffers.IsEmpty())
        {
            UDP_PACKET* pPacket = (UDP_PACKET*)m_ReadUDPBuffers.RemoveHead();

            HX_RELEASE(pPacket->pBuffer);
            HX_DELETE(pPacket);
        }
        m_pMutex->Unlock();
    }

    return theErr;
}

HX_RESULT
ThreadedConn::reuse_addr(BOOL enabled)
{
    HX_RESULT theErr = HXR_UNEXPECTED;
    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pMutex->Lock();
        theErr = m_pActualConn->reuse_addr(enabled);
        m_pMutex->Unlock();
    }
    return theErr;
}

HX_RESULT
ThreadedConn::reuse_port(BOOL enabled)
{
    HX_RESULT theErr = HXR_UNEXPECTED;
    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pMutex->Lock();
        theErr = m_pActualConn->reuse_port(enabled);
        m_pMutex->Unlock();
    }
    return theErr;
}

HX_RESULT
ThreadedConn::set_broadcast(BOOL enable)
{
        HX_RESULT theErr = HXR_UNEXPECTED;
        HX_ASSERT(m_pActualConn);
        if(m_pActualConn)
        {
        m_pMutex->Lock();
        theErr = m_pActualConn->set_broadcast(enable);
        m_pMutex->Unlock();
        }
        return theErr;
}

HX_RESULT
ThreadedConn::set_multicast_if(ULONG32 ulInterface)
{
    HX_RESULT theErr = HXR_UNEXPECTED;
    HX_ASSERT(m_pActualConn);
    if(m_pActualConn)
    {
        m_pMutex->Lock();
        theErr = m_pActualConn->set_multicast_if(ulInterface);
        m_pMutex->Unlock();
    }
    return theErr;
}

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

HX_RESULT
ThreadedConn::last_error(void)
{
    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pMutex->Lock();
        mLastError = m_pActualConn->last_error();
        m_pMutex->Unlock();
    }

    return mLastError;
}

void
ThreadedConn::set_callback(HXAsyncNetCallback* pCallback)
{
    m_pMutex->Lock();
    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        /* Set our callback as the callback */
        m_pActualConn->set_callback(m_pNetCallback);
    }

    mCallBack = pCallback;
    m_pMutex->Unlock();
}


UINT16
ThreadedConn::connection_open(void)
{
    UINT16 uConnOpen = 0;
    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pMutex->Lock();
        uConnOpen = m_pActualConn->connection_open();
        m_pMutex->Unlock();
    }

    return uConnOpen;
}

UINT16
ThreadedConn::connection_really_open(void)
{
    UINT16 uConnOpen = 0;
    //HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pMutex->Lock();
        uConnOpen = m_pActualConn->connection_really_open();
        m_pMutex->Unlock();
    }

    return uConnOpen;
}

int
ThreadedConn::get_sock(void)
{
    int iSockNum = -1;
    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pMutex->Lock();
        iSockNum =  m_pActualConn->get_sock();
        m_pMutex->Unlock();
    }
    return iSockNum;
}

void
ThreadedConn::set_sock(int theSock)
{
    m_pMutex->Lock();
    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pActualConn->set_sock(theSock);
    }

    mSock = theSock;
    m_pMutex->Unlock();
}

BOOL
ThreadedConn::set_receive_buf_size(int DesiredSize)
{
    ThrdConnGenericCallback* pCallback  = new ThrdConnGenericCallback(this, SET_BUFFER_SIZE_CALLBACK_TYPE);
    pCallback->m_ulBufferSize = (UINT32) DesiredSize;
    /* Will be released by the thread engine */
    pCallback->AddRef();
    HXThreadMessage msg(HXMSG_ASYNC_CALLBACK, this, pCallback);
    m_pNetworkThread->PostMessage(&msg);
    return TRUE;
}


BOOL
ThreadedConn::ActualSetReceiveBufSize(UINT32 ulBufferSize)
{
    BOOL bResult = FALSE;
    HX_ASSERT(m_pActualConn);
    if (m_pActualConn)
    {
        m_pMutex->Lock();
        bResult = m_pActualConn->set_receive_buf_size((int) ulBufferSize);
        m_pMutex->Unlock();
    }

    return bResult;
}

void
ThreadedConn::OnAsyncDNS(BOOL bResult)
{
    if (mCallBack)
    {
        mCallBack->Func(DNS_NOTIFICATION, bResult);
    }
}

void
ThreadedConn::OnReadNotification(void)
{
    if (mCallBack)
    {
        mCallBack->Func(READ_NOTIFICATION);
    }
}

void
ThreadedConn::OnWriteNotification(void)
{
    if (mCallBack)
    {
        mCallBack->Func(WRITE_NOTIFICATION);
    }
}

void
ThreadedConn::OnConnect(BOOL bResult)
{
    if (mCallBack)
    {
        mCallBack->Func(CONNECT_NOTIFICATION, bResult);
    }
}

void
ThreadedConn::OnAcceptNotification()
{
    if( m_pIncommingConnections )
    {
        conn* pConn = (conn*)m_pIncommingConnections->RemoveHead();
	// accept_notification is posted in win_net:checkforconnection
	// callback->func(accept_notification..) calls this function
#ifndef HELIX_FEATURE_NETWORK_USE_SELECT
        if (mCallBack)
        {
            mCallBack->Func(ACCEPT_NOTIFICATION, TRUE, pConn);
        }
        if (pConn)
        {
            pConn->Release();
            pConn = NULL;
        }
#else
	// at this point we need to set pengine->reader = pconn
        HXThreadMessage msg(HXMSG_ASYNC_SETREADER_CONNECTION, this, pConn);
        m_pNetworkThread->PostMessage(&msg);
#endif

    }
}

void
ThreadedConn::HandleDNSNotification(BOOL bSuccess)
{
#if !defined( WIN32_PLATFORM_PSPC ) && !defined(HELIX_FEATURE_NETWORK_USE_SELECT)
    HXThreadMessage msg(HXMSG_ASYNC_DNS, this, (void*) bSuccess);
    m_pMainAppThread->PostMessage(&msg, m_pInternalWindowHandle);
#else
    OnAsyncDNS(bSuccess);
#endif
}

void
ThreadedConn::HandleConnectNotification(BOOL bSuccess)
{
    if (bSuccess)
    {
        m_bConnected = TRUE;
    }

#if !defined( WIN32_PLATFORM_PSPC ) && !defined(HELIX_FEATURE_NETWORK_USE_SELECT)
    HXThreadMessage msg(HXMSG_ASYNC_CONNECT, this, (void*) bSuccess);
    m_pMainAppThread->PostMessage(&msg, m_pInternalWindowHandle);
#else // No notifier on CE
    OnConnect(bSuccess);
#endif
}

void
ThreadedConn::HandleAcceptNotification(conn* pConn)
{
    ThreadedConn* pTConn = (ThreadedConn*)conn::new_socket(HX_TCP_SOCKET);
    pTConn->SetActualConn(pConn, m_ulUserHandle);
    m_pIncommingConnections->AddHead((conn*)pTConn);

#if !defined( WIN32_PLATFORM_PSPC ) && !defined(HELIX_FEATURE_NETWORK_USE_SELECT)
    HXThreadMessage msg(HXMSG_ASYNC_ACCEPT, this, NULL);
    m_pMainAppThread->PostMessage(&msg, m_pInternalWindowHandle);
#else // No notifier on CE
    OnAcceptNotification();
#endif
}

void
ThreadedConn::HandleCloseNotification()
{
    /* make one more read call to get back the actual error */
    m_bReadPostPendingWouldBlock = FALSE;
}

HX_RESULT
ThreadedConn::SetActualConn(conn* pConn, ULONG32 ulPlatform)
{
    if ( m_pActualConn )
    {
        m_pActualConn->done();
        m_pActualConn->Release();
    }
    m_pActualConn = pConn;
    m_pActualConn->AddRef();
    m_ulUserHandle = ulPlatform;
    m_bConnected = TRUE;
    return HXR_OK;
}

HX_RESULT
ThreadedConn::DoRead(BOOL bFromReadNotification)
{
#ifdef _CARBON
    AddRef(); // ensure that this object doesn't encounter its dtor before routine completes
#endif
    HX_RESULT theErr = HXR_OK;
    m_pMutex->Lock();

    /* Reset reading heuristing if we just got data. */
    if( bFromReadNotification )
    {
        m_bReadNowPending = m_bReadPostPendingWouldBlock = FALSE;
    }

    /* If the socket done has already been called, do not attempt to read
     * any more data
     */
    if (m_bIsDone)
    {
        goto exit;
    }

#ifdef THREADS_SUPPORTED
#ifdef HELIX_FEATURE_ADD_NETWORK_THREAD_SLEEP
    Sleep( m_ulNetworkThreadSleep );	// gives other threads a quote to run
#endif //HELIX_FEATURE_ADD_NETWORK_THREAD_SLEEP
#endif //THREADS_SUPPORTED

    if (m_uSocketType == HX_TCP_SOCKET)
    {
        UINT16 uCount = m_pReceiveTCP->GetMaxAvailableElements();

        if (uCount > 0)
        {
            UINT32 ulBytesToRead = conn::bytes_to_preparetcpread(this);

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

                if ( m_bReadPostPendingWouldBlock )
                {
                    /* fake a call return */
                    theErr = HXR_WOULD_BLOCK;
                }
                else
                {
                    /* call read and do heuristinc bookkeeping */
                    theErr = m_pActualConn->read(m_pTempBuffer, &uCount);