hxnetapi.cpp

linux下的一款播放器

	}
    }
    if (m_bReusePort)
    {
	if (m_pData->reuse_port(m_bReusePort) != HXR_OK)
	{
	    // err...what do we need to do?
	    HX_ASSERT(!"reuse_port() failed");
	}
    }

#ifdef _UNIX
    m_pData->SetAsyncDNSPref( ReadAsyncDNSPref((IUnknown*)m_pContext) );
#endif


    // XXXST -- local addr binding stuff, removed dependency to m_nLocalPort
    // 0 for local port will make the system choose a free port
    theErr = m_pData->init(ulLocalAddr, nPort);

    if (theErr)
    {
	theErr = ConvertNetworkError(theErr);
	if (theErr)
	{
	    m_pData->done();
	    m_pData->Release();
	    m_pData = 0;
	    return theErr;
	}
    }

    m_pData->nonblocking();

    m_pData->set_receive_buf_size(DESIRED_RCV_BUFSIZE);
    if (!m_pCallback)
    {
	m_pCallback = new UDPSocketCallback;
	m_pCallback->m_pContext = this;
    }
    m_pData->set_callback(m_pCallback);

#ifdef _WINDOWS
#if defined (_WIN32)
    ulPlatformData = (UINT32)GetModuleHandle(NULL);
#elif defined (_WIN16)
    ulPlatformData = (UINT32)(int)g_hInstance;
#endif
    m_pData->SetWindowHandle(ulPlatformData);
#endif /* defined (_WINDOWS) */

    if (m_pSchedulerReadCallback)
    {
	    m_pSchedulerReadCallback->ScheduleCallback(UDP_BIND_COMMAND, m_pScheduler, SCHED_GRANULARITY);
    }

    m_bInitComplete = TRUE;
    return theErr;
}

STDMETHODIMP HXUDPSocket::Read(UINT16 nBytes)
{
    if (!m_bInitComplete)
    {
	HX_RESULT ret = Bind(HXR_INADDR_ANY, 0);
	if (HXR_OK != ret)
	    return HXR_UNEXPECTED;
    }

    HX_RESULT theErr = HXR_OK;

    if(m_bReadPending)
    {
	return HXR_UNEXPECTED;
    }

    m_bReadPending = TRUE;

    if (m_bInRead)
    {
	return HXR_OK;
    }

    m_bInRead = TRUE;

    m_pMutex->Lock();
    UINT16 uNumIterations = 0;
    do
    {
	theErr = DoRead();
	uNumIterations++;
    } while (m_bReadPending && !theErr && m_ReadBuffers.GetCount() > 0 && uNumIterations < MAX_ITERATION_COUNT);

    m_pMutex->Unlock();
    theErr = ConvertNetworkError(theErr);

    if (m_bReadPending && m_pSchedulerReadCallback)
	    m_pSchedulerReadCallback->ScheduleCallback(UDP_READ_COMMAND, m_pScheduler, SCHED_GRANULARITY);

    m_bInRead = FALSE;

    return theErr;
}

STDMETHODIMP HXUDPSocket::Write(IHXBuffer* pBuffer)
{
    if (!m_bInitComplete)
    {
	HX_RESULT ret = Bind(HXR_INADDR_ANY, 0);
	if (HXR_OK != ret)
	    return HXR_UNEXPECTED;
    }
    HX_RESULT theErr = HXR_OK;
    HX_RESULT lResult = HXR_OK;

#if 0
    struct in_addr in;
    in.s_addr = m_sockAddr.sin_addr.s_addr;
    char* address = inet_ntoa(in);
    printf("address = %s:", address);
    UINT32 port = ntohl(m_sockAddr.sin_port);
    printf("%d\n", port);
#endif /* 0 */

#ifdef _MACINTOSH
	if (m_pInterruptSafeMacWriteQueue)
		m_pInterruptSafeMacWriteQueue->AddTail(pBuffer); // AddRef called inside
#else
	pBuffer->AddRef();
	m_WriteBuffers.AddTail((void*) pBuffer);
#endif


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

    lResult = ConvertNetworkError(theErr);

    return lResult;
}

STDMETHODIMP HXUDPSocket::WriteTo(ULONG32 ulAddr,
    UINT16 nPort, IHXBuffer* pBuffer)
{
    if (!m_bInitComplete)
    {
	HX_RESULT ret = Bind(HXR_INADDR_ANY, 0);
	if (HXR_OK != ret)
	    return HXR_UNEXPECTED;
    }

    m_sockAddr.sin_family	= AF_INET;
    m_sockAddr.sin_addr.s_addr	= DwToNet(ulAddr);   //*(long*)&ulAddr;
    m_sockAddr.sin_port		= WToNet(nPort);

    return (Write(pBuffer));
}

STDMETHODIMP HXUDPSocket::GetLocalPort(UINT16& nPort)
{
    // Get the local port from the socket info
    nPort = m_pData->get_local_port();

    return (INT16)nPort < 0 ? HXR_OK : HXR_FAIL;
}

STDMETHODIMP HXUDPSocket::JoinMulticastGroup(
    ULONG32	    ulMulticastAddr,
    ULONG32	    ulInterfaceAddr)
{
#if defined(HELIX_FEATURE_TRANSPORT_MULTICAST)
    HX_RESULT theErr = HXR_OK;
    HX_RESULT lResult = HXR_OK;

    m_pMutex->Lock();
    theErr = m_pData->join_multicast_group(ulMulticastAddr, ulInterfaceAddr);
    lResult = ConvertNetworkError(theErr);
    m_pMutex->Unlock();

    return lResult;
#else
    return HXR_NOTIMPL;
#endif /* HELIX_FEATURE_TRANSPORT_MULTICAST */
}

STDMETHODIMP HXUDPSocket::LeaveMulticastGroup(
    ULONG32	    ulMulticastAddr,
    ULONG32	    ulInterfaceAddr)
{
#if defined(HELIX_FEATURE_TRANSPORT_MULTICAST)
    HX_RESULT theErr = HXR_OK;
    HX_RESULT lResult = HXR_OK;

    m_pMutex->Lock();
    theErr = m_pData->leave_multicast_group(ulMulticastAddr, ulInterfaceAddr);
    while (!m_ReadBuffers.IsEmpty())
    {
	UDP_PACKET* pPacket = (UDP_PACKET*)m_ReadBuffers.RemoveHead();

	HX_RELEASE(pPacket->pBuffer);
	HX_DELETE(pPacket);
    }
    lResult = ConvertNetworkError(theErr);
    m_pMutex->Unlock();

    return lResult;
#else
    return HXR_NOTIMPL;
#endif /* HELIX_FEATURE_TRANSPORT_MULTICAST */
}

HX_RESULT
HXUDPSocket::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;
    IHXBuffer* pBuffer = 0;
    UINT32  ulAddress = 0;
    UINT16  ulPort = 0;
    UINT16  count = 0;


    do
    {
	/*
	 * Must reset count before every read
	 */
	count = TCP_BUF_SIZE;

	theErr = m_pData->readfrom(pBuffer, ulAddress, ulPort);

	if (!theErr && pBuffer)
	{
	    UDP_PACKET* pPacket = new UDP_PACKET;

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

	    m_ReadBuffers.AddTail((void*)pPacket);
	}
	else
	{
	    count = 0;
	}
    } while (!theErr && count > 0);

    if (m_bReadPending && m_ReadBuffers.GetCount() > 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_AT_INTERRUPT;
	}

	m_bReadPending = FALSE;

	UDP_PACKET* pPacket = (UDP_PACKET*)m_ReadBuffers.RemoveHead();
	pBuffer = pPacket->pBuffer;
	ulAddress = pPacket->ulAddress;
	ulPort = pPacket->ulPort;

	AddRef();


	m_bInDoRead = FALSE;
	m_pUDPResponse->ReadDone(HXR_OK, pBuffer, ulAddress, ulPort);
	m_bInDoRead = TRUE;

	HX_RELEASE(pBuffer);
	HX_DELETE(pPacket);

	Release();

	m_bInDoRead = FALSE;
	return HXR_OK;
    }

    /* if we called from within Read(), we will schedule a callback there, if necessary */
    if (!m_bInRead && m_pSchedulerReadCallback)
	    m_pSchedulerReadCallback->ScheduleCallback(UDP_READ_COMMAND, m_pScheduler, SCHED_GRANULARITY);

	m_bInDoRead = FALSE;
	return theErr;
}

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

    if (m_bInWrite) return HXR_OK;
    m_bInWrite = TRUE;

#ifdef _MACINTOSH
	if (m_pInterruptSafeMacWriteQueue)
		m_pInterruptSafeMacWriteQueue->TransferToSimpleList(m_WriteBuffers);
#endif

    while (!theErr && m_WriteBuffers.GetCount() > 0)
    {
	IHXBuffer* pBuffer = (IHXBuffer*) m_WriteBuffers.GetHead();
	UINT16 uLength = (UINT16) pBuffer->GetSize();
	theErr = m_pData->writeto(  pBuffer->GetBuffer(),   // sendto
				    &uLength,
				    (UINT32) m_sockAddr.sin_addr.s_addr,
				    WToHost(m_sockAddr.sin_port));
	if (!theErr)
	{
	    pBuffer->Release();
	    m_WriteBuffers.RemoveHead();
	}
    }

    if (m_pSchedulerWriteCallback &&
	m_WriteBuffers.GetCount() > 0)
    {
	    m_pSchedulerWriteCallback->ScheduleCallback(UDP_WRITE_COMMAND, m_pScheduler, SCHED_GRANULARITY);
    }

    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
HXUDPSocket::IsSafe()
{
    if (m_pInterruptState && m_pInterruptState->AtInterruptTime() &&
	(!m_pResponseInterruptSafe ||
	 !m_pResponseInterruptSafe->IsInterruptSafe()))
    {
	if (m_pNonInterruptReadCallback){
		m_pNonInterruptReadCallback->ScheduleCallback(UDP_READ_COMMAND, m_pScheduler, 0);
	}

	return FALSE;
    }

    return TRUE;
}

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

			case UDP_WRITE_COMMAND:
			    theErr = DoWrite();  // protected from re-entry by m_bInWrite
			    break;

			default:
			    theErr = DoRead();
                            if( theErr == HXR_OK )
                            {
			        theErr = DoWrite();
                            }
			    break;
	   	}
	}
	m_pMutex->Unlock();
	Release();
    }

    return theErr;
}


HX_RESULT HXUDPSocket::UDPSocketCallback::Func(NotificationType Type,
					       BOOL bSuccess, conn* pConn)
{
    if(m_pContext)
    {
	switch (Type)
	{
	case READ_NOTIFICATION:
#if defined(_UNIX_THREADED_NETWORK_IO)
            if( !ReadNetworkThreadingPref((IUnknown*)(m_pContext->m_pContext)) )
            {
                m_pContext->AddRef();
                m_pContext->m_pMutex->Lock();
                m_pContext->DoRead();
                m_pContext->m_pMutex->Unlock();
                m_pContext->Release();
            }
#elif !defined (THREADS_SUPPORTED) && !defined(_MACINTOSH)
	    m_pContext->AddRef();
	    m_pContext->m_pMutex->Lock();
	    m_pContext->DoRead();
	    m_pContext->m_pMutex->Unlock();
	    m_pContext->Release();
#endif
	    break;
	case WRITE_NOTIFICATION:
#if defined(_UNIX_THREADED_NETWORK_IO)
            if( !ReadNetworkThreadingPref((IUnknown*)(m_pContext->m_pContext)) )
            {
                m_pContext->AddRef();
                m_pContext->m_pMutex->Lock();
                m_pContext->DoWrite();
                m_pContext->m_pMutex->Unlock();
                m_pContext->Release();
            }
#elif !defined (THREADS_SUPPORTED) && !defined(_MACINTOSH)
	    m_pContext->AddRef();
	    m_pContext->m_pMutex->Lock();
	    m_pContext->DoWrite();
	    m_pContext->m_pMutex->Unlock();
	    m_pContext->Release();
#endif
	    break;
	case CONNECT_NOTIFICATION:
	default:
	    break;
	}
    }
    return HXR_OK;
}


STDMETHODIMP
HXUDPSocket::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_BROADCAST:
	{
	HX_RESULT theErr = HXR_OK;
	if(m_pData)
	{
		m_pMutex->Lock();
		theErr = m_pData->set_broadcast(ulValue);
		res = ConvertNetworkError(theErr);
		m_pMutex->Unlock();
	}
	}
	break;
    case HX_SOCKOPT_MULTICAST_IF:
	if (m_pData)
	{
	    res = m_pData->set_multicast_if(ulValue);
	}
	break;
    case HX_SOCKOPT_SET_SENDBUF_SIZE:
    {
      if (m_pData)
      {
          HX_RESULT theErr = HXR_OK;

           m_pMutex->Lock();
           theErr = m_pData->set_send_size(ulValue);
           res = ConvertNetworkError(theErr);
           m_pMutex->Unlock();
      }
      break;
    }
    default:
	HX_ASSERT(!"I don't know this option");
	res = HXR_FAIL;
    }

    return res;
}

STDMETHODIMP
HXUDPSocket::SetOption(HX_PRIVATE_SOCKET_OPTION option, UINT32 ulValue)
{
    HX_RESULT res = HXR_OK;

    switch(option)
    {
    case HX_PRIVATE_SOCKOPT_IGNORE_WSAECONNRESET:
    {
	if (m_pData)
	{
	    m_pMutex->Lock();
	    m_pData->IgnoreWSAECONNRESET((BOOL)ulValue);
	    m_pMutex->Unlock();
	}
	break;
    }
    default:
	HX_ASSERT(!"I don't know this option");
	res = HXR_FAIL;
    }
    return HXR_OK;
}

#ifdef _MACINTOSH

/////////////////////////////////////////////////////////////////////////
//
//	InterruptSafeMacQueue
//
//	For passing data between an interrupt and anything else (mac only).
//
InterruptSafeMacQueue::InterruptSafeMacQueue()
{
 	mQueueHeader.qFlags=0;
	mQueueHeader.qHead=0;
	mQueueHeader.qTail=0;
	mDestructing = FALSE; // just a safety check
}

/////////////////////////////////////////////////////////////////////////
//
HX_RESULT InterruptSafeMacQueue::AddTail(IUnknown* pObject)
{
	if (pObject && !mDestructing)
	{
	    IhxQueueElement * theElement = new IhxQueueElement();

	    if (theElement)
	    {
		    theElement->mNextElementInQueue = NULL;
		    theElement->mObject = pObject;