rmsource.cpp

神龙卡 SDK_84xx_DShow_145_02.zip 这个是 windows 上二个是linux

//////////////////////////////////////////////////////////////////////////
// IRMSourceStreamEx methods
HRESULT STDMETHODCALLTYPE CRMSourceStream::put_ServerIp(char* szServerIp)
{
	strcpy(m_pConnParams->rmprops.serverIp, szServerIp);
	return S_OK;
}

//////////////////////////////////////////////////////////////////////////

HRESULT STDMETHODCALLTYPE CRMSourceStream::put_ServerType(int iServerType)
{
	m_pConnParams->rmprops.servertype = iServerType;
	return S_OK;
}

//////////////////////////////////////////////////////////////////////////

HRESULT STDMETHODCALLTYPE CRMSourceStream::put_PlayFileName(char* szPlayFileName)
{
	strcpy(m_pConnParams->rmprops.browseinfo, szPlayFileName);
	return S_OK;
}

//////////////////////////////////////////////////////////////////////////

HRESULT STDMETHODCALLTYPE CRMSourceStream::put_ConnectionParams(char* szServerIp, 
	int iServerType, char* szPlayFileName)
{
	strcpy(m_pConnParams->rmprops.serverIp, szServerIp);
	m_pConnParams->rmprops.servertype = iServerType;
	strcpy(m_pConnParams->rmprops.browseinfo, szPlayFileName);
	return S_OK;
}

//////////////////////////////////////////////////////////////////////////

HRESULT STDMETHODCALLTYPE CRMSourceStream::StartRtsp()
{
	//if no session is active, initialize the rtsp
	if(m_pConnParams->sSessionId[0] == '\0')
	{
		PHOSTENT g_temphost;
		char g_sHost[1024];
		struct sockaddr_in   g_dest;	
		gethostname(g_sHost, 1024);
		g_temphost = gethostbyname(g_sHost);
		memcpy(&(g_dest.sin_addr), g_temphost->h_addr_list[0], g_temphost->h_length);
		int		iUserPort = 5004;
		int		iRtpPort  = 5006;
		// initialize the connection parameters
		IRtsp_Initialize(pIRtsp, m_pConnParams, inet_ntoa(g_dest.sin_addr), iUserPort, iRtpPort);
		
		return SetUpThreadProc();
	}
	else return S_FALSE;
}
// End of IRMSourceStreamEx methods.
//////////////////////////////////////////////////////////////////////////

/*****************************************************************************************
 *	CRMReceive -- receives and stores the file data										 * 
 *	output: BOOL - FALSE if failure												   		 * 
 *****************************************************************************************/
BOOL CRMSourceStream::CRMReceive(struct tagIRtsp* This, SOCKET local_sock, int iPort, BYTE *pData, LONG lDataLen, LONG *lBytesRead, CONN_DATA *pConnData)
{
	int			iAddrLen;	// receives the length of the server address structure
	SOCKADDR	ServerAddr;	// server address structure
	int			value;
	DWORD		nonblocking, starttime, curtime, lastping;
	Command		com;

	WSASetLastError(0);
	iAddrLen = sizeof(ServerAddr);


	//prime the loop, receive response from anyone who sends data to this port
	value = ioctlsocket ( local_sock, FIONREAD , &nonblocking);//check socket for data
	starttime = timeGetTime();
	lastping = starttime;
	while( nonblocking < 20 )      //loop while there is no data to read
	{
		curtime = timeGetTime();
		//find milliseconds since starting 
		// timeout is 60 seconds before it sends EC_COMPLETE message if server doesn't send
		if((int)(curtime - starttime) > 60000)
		{
			DEBUG_PRINT(("\nTimed out"));
			return FALSE; //if we timeout abort the receive 
		}
		else if(pConnData->finished)
		{
			DEBUG_PRINT(("\nEOF"));
			return FALSE; //if we reach end of file abort the receive 
		}
		else if(CheckRequest(&com))	// Check state: init, pause, run, stop, or exit
			if(com == CMD_STOP) return TRUE;
		if( (int)(curtime - lastping) > 200 )
		{
			DEBUG_PRINT (("PING!\n"));
			IRtsp_Ping(This, pConnData);	// Ping the server to keep connection alive
			lastping = curtime;
		}
		Sleep(4);
		ioctlsocket ( local_sock, FIONREAD , &nonblocking);//check socket for data
	}

	//receive response from anyone who sends data to this port
	*lBytesRead = (LONG) (recvfrom(local_sock, (char *)pData, lDataLen, 0, &ServerAddr, &iAddrLen));
	if(*lBytesRead == SOCKET_ERROR){    //if there was an error in the receive
		ISock_SocketError(pConnData->pISock, WSAGetLastError());
		return FALSE;
	}
	else if (*lBytesRead < 20)			//if there is too little data
	{
		ISock_SocketError(pConnData->pISock, WSAGetLastError());
		return FALSE;
	}	
//	CloseHandle(local_timeout);
	return TRUE;
}


// FillBuffer
// Reads data from file into the supplied video buffer
//////////////////////////////////////////////////////////////////////////
HRESULT CRMSourceStream::FillBuffer(IMediaSample *pms)
{
    BYTE *pData;
    long lDataLen;
	long lBytesRead;
	BYTE ptTemp[10000];
	BOOL local_exit = TRUE;
	
	if(m_readytogo != 1) return S_FALSE;

    pms->GetPointer(&pData);
    lDataLen = pms->GetSize();

	if (m_bInit)
	{
		// Set the discontinuity for the first sample
		pms->SetDiscontinuity(TRUE);
		m_bInit = FALSE;
		// if SGI server, write the file header we got from the setup response
		if (m_pConnParams->rmprops.servertype == SGI)
		{
			memcpy(	(LPVOID) pData, m_pConnParams->sHeader, m_pConnParams->iHeaderLen);
			lDataLen = m_pConnParams->iHeaderLen;
			lBytesRead = (LONG) lDataLen;
		}		
	}

	local_exit = CRMReceive(pIRtsp, m_RTPsock, 5006, ptTemp, sizeof(ptTemp), &lBytesRead, m_pConnParams);
	if (lBytesRead < 20 || local_exit == FALSE)
	{
		return S_FALSE;
	}
	//remove and store OVS header for MPEG1
	if (m_pConnParams->rmprops.servertype == OVS && m_pConnParams->cont_type == MPEG1)  
	{
		memcpy(	(LPVOID)&m_OGFhead, (LPVOID) ptTemp, 20);
		lBytesRead -= 20;
		memcpy(pData, ptTemp + 20, lBytesRead);
	}
	else memcpy(pData, ptTemp, lBytesRead);
	pms->SetActualDataLength(lBytesRead);

//	fwrite(pData, lBytesRead, 1, dataout);

	DEBUG_PRINT(("\nFillbuffer got %lu bytes",lBytesRead));
	if(m_pConnParams->cont_type == MPEG1)
	{
		DEBUG_PRINT(("\nFillbuffer should have %lu bytes for packet %u",fix4endian(m_OGFhead.size),fix2endian(m_OGFhead.fseq)));
		DEBUG_PRINT(("\nHeader size %u and cseq %u and cnum %u",m_OGFhead.DP,fix2endian(m_OGFhead.cseq),fix2endian(m_OGFhead.cnum)));
	}
	return NOERROR;

} // FillBuffer


// GetMediaType
//////////////////////////////////////////////////////////////////////////
HRESULT CRMSourceStream::GetMediaType(CMediaType *pmt)
{
    CAutoLock cAutoLock(m_pFilter->pStateLock());

    pmt->SetType(&MEDIATYPE_Stream);
	pmt->SetSubtype(&MEDIASUBTYPE_MPEG1System);

    return NOERROR;

} // GetMediaType


// DecideBufferSize
//////////////////////////////////////////////////////////////////////////
HRESULT CRMSourceStream::DecideBufferSize(IMemAllocator *pAlloc,ALLOCATOR_PROPERTIES *pProperties)
{
    CAutoLock cAutoLock(m_pFilter->pStateLock());
    ASSERT(pAlloc);
    ASSERT(pProperties);
    HRESULT hr = NOERROR;

	m_pAlloc = pAlloc;
   	m_pAlloc->AddRef();

	// Allocate 120 buffers (8K - 20 each need to be exact data size for problem in splitter)
    pProperties->cBuffers = 120;
    if(m_pConnParams->cont_type == MPEG1) pProperties->cbBuffer = 8172; // 10240;
	else pProperties->cbBuffer = 8192;
    ASSERT(pProperties->cbBuffer);

    // Ask the allocator to reserve us some sample memory, NOTE the function
    // can succeed (that is return NOERROR) but still not have allocated the
    // memory that we requested, so we must check we got whatever we wanted

    ALLOCATOR_PROPERTIES Actual;
    hr = pAlloc->SetProperties(pProperties,&Actual);
    if (FAILED(hr)) {
        return hr;
    }

    // Is this allocator unsuitable

    if (Actual.cbBuffer < pProperties->cbBuffer) {
        return E_FAIL;
    }

    return NOERROR;

} // DecideBufferSize


// SetMediaType
// Called when a media type is agreed between filters
//////////////////////////////////////////////////////////////////////////
HRESULT CRMSourceStream::SetMediaType(const CMediaType *pMediaType)
{
    CAutoLock cAutoLock(m_pFilter->pStateLock());

    // Pass the call up to my base class

    HRESULT hr = CSourceStream::SetMediaType(pMediaType);
    if (SUCCEEDED(hr)) {
         return NOERROR;
    } else {
        return hr;
    }

} // SetMediaType


// NonDelegatingQueryInterface
// Override this to say what interfaces we support where
//////////////////////////////////////////////////////////////////////////
STDMETHODIMP CRMSource::NonDelegatingQueryInterface(REFIID riid, void ** ppv)
{
    CheckPointer(ppv,E_POINTER);

    if (riid == IID_ISpecifyPropertyPages)
        return GetInterface((ISpecifyPropertyPages *) this, ppv);
	else if (riid == IID_IRMSourceStream)
	{
		ASSERT (m_paStreams[0]);
        return m_paStreams[0]->QueryInterface (riid, ppv);
	}
	else if (riid == IID_IRMSourceStreamEx)
	{
		ASSERT (m_paStreams[0]);
		return m_paStreams[0]->QueryInterface (riid, ppv);
	}
    else if (riid == IID_IPersistStream)
        return GetInterface((IPersistStream *) this, ppv);
    else if (riid == IID_IPersistPropertyBag)
        return GetInterface((IPersistPropertyBag *) this, ppv);
    else if (riid == IID_IPropertyBag)
        return GetInterface((IPropertyBag *) this, ppv);
    else if (riid == IID_IRMSource)
        return GetInterface((IRMSource *) this, ppv);
    else return CSource::NonDelegatingQueryInterface(riid, ppv);
}

  
//////////////////////////////////////////////////////////////////////////
STDMETHODIMP CRMSourceStream::NonDelegatingQueryInterface(REFIID riid, void ** ppv)
{
	 if (riid == IID_IRMSourceStream)
		 return	GetInterface((IRMSourceStream *) this, ppv);
	 else if (riid == IID_IRMSourceStreamEx)
		 return GetInterface((IRMSourceStreamEx *) this, ppv);
	 else return CSourceStream::NonDelegatingQueryInterface(riid, ppv);
}

// SetUpThreadProc
//////////////////////////////////////////////////////////////////////////
HRESULT STDMETHODCALLTYPE CRMSourceStream::SetUpThreadProc()
{
	DWORD	dwThread;
	STATUS	status;			// store the STATUS return value of RTSP functions
	char	*sURL = "test";	// presentation URL

	
	if (m_pConnParams->rmprops.servertype == OVS)
	{
		status = IRtsp_Open(pIRtsp, m_pConnParams, m_pConnParams->rmprops.browseinfo);
	}
	else if (m_pConnParams->rmprops.servertype == SGI)
	{
		strcpy(m_pConnParams->sName, "Lair.mpg");
		strcpy(sURL, "rtsp://172.30.9.2/Lair");
		status = IRtsp_Open(pIRtsp, m_pConnParams, sURL);
	}
	
	if(m_pConnParams->cont_type != MPEG1)
	{
		IFilterGraph* l_pGraph = m_pRMSource->GetFilterGraph();
		IRtsp_Close(pIRtsp, m_pConnParams);	
		if(l_pGraph)l_pGraph->Reconnect((IPin*)(m_pRMSource->GetPin(0)));
		status = IRtsp_Open(pIRtsp, m_pConnParams, m_pConnParams->rmprops.browseinfo);
	}
	// if SETUP is succesful, start the machine.
	if (status == 200)
	{					
		m_bInit = TRUE;
		if (m_pConnParams->rmprops.servertype == OVS)
		{
			m_Event = CreateEvent (0, 0, 0,0);
			strcpy(m_pConnParams->sName, "lair.mpg");
			// if OVS, we have to ping the server regularly to keep the session alive
			m_hPingThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE) OSKeepAlive , (void *)this, 0, &dwThread);
			if(	m_hPingThread == NULL){
				DEBUG_PRINT(("\nPinging thread failure value"));
			}
		}
	}
	else
	{
		DEBUG_PRINT(("\nIRtsp_Open failed..."));
		return status;
	}
	return S_OK;
}


// OnThreadCreate
// As we go active reset the stream time to zero
//////////////////////////////////////////////////////////////////////////
HRESULT CRMSourceStream::OnThreadCreate()
{
/*	PHOSTENT g_temphost;
	char g_sHost[1024];
	struct sockaddr_in   g_dest;

	gethostname(g_sHost, 1024);
	g_temphost = gethostbyname(g_sHost);
	memcpy(&(g_dest.sin_addr), g_temphost->h_addr_list[0], g_temphost->h_length);
	
	// initialize the connection parameters
	int		iUserPort = 5004;
	int		iRtpPort  = 5006;
	m_RTPsock = NULL;

	IRtsp_Initialize(pIRtsp, m_pConnParams, inet_ntoa(g_dest.sin_addr), iUserPort, iRtpPort);
*/
	DEBUG_PRINT(("\nReturning from ThreadCreate"));
    return NOERROR;

} // OnThreadCreate


//////////////////////////////////////////////////////////////////////////
HRESULT CRMSourceStream::OnThreadDestroy()
{
	ISock_closethesocket(m_pConnParams->pISock, &m_RTPsock);

	return NOERROR;
}


//////////////////////////////////////////////////////////////////////////
HRESULT CRMSourceStream::OnThreadStartPlay(void) 
{
	m_pConnParams->finished = 0;
	m_readytogo = 1;
	if(IRtsp_Play(pIRtsp, m_pConnParams) != 200) m_readytogo = -1;
	if(ISock_openthesocket(m_pConnParams->pISock, &m_RTPsock, m_pConnParams)<0) 
		m_readytogo = -1;
	return NOERROR;
}


// ThreadProc
// When this returns the thread exits
// Return codes > 0 indicate an error occured
//////////////////////////////////////////////////////////////////////////
DWORD CRMSourceStream::ThreadProc(void) {

    HRESULT hr;  // the return code from calls
    Command com;

    do {
	com = GetRequest();
	if (com != CMD_INIT) {
	    DbgLog((LOG_ERROR, 1, TEXT("Thread expected init command")));
	    Reply((DWORD) E_UNEXPECTED);
	}
    } while (com != CMD_INIT);

    DbgLog((LOG_TRACE, 1, TEXT("CSourceStream worker thread initializing")));

    hr = OnThreadCreate(); // perform set up tasks
    if (FAILED(hr)) {
        DbgLog((LOG_ERROR, 1, TEXT("CSourceStream::OnThreadCreate failed. Aborting thread.")));
        OnThreadDestroy();
		Reply(hr);	// send failed return code from OnThreadCreate
        return 1;
    }

    // Initialisation suceeded
    Reply(NOERROR);

    Command cmd;
    do {
	cmd = GetRequest();

	switch (cmd) {

	case CMD_EXIT:
		if(m_pConnParams->finished == 0) IRtsp_Stop(pIRtsp, m_pConnParams);
	    Reply(NOERROR);
	    break;

	case CMD_RUN:
	    DbgLog((LOG_ERROR, 1, TEXT("CMD_RUN received before a CMD_PAUSE???")));
	    // !!! fall through???
	
	case CMD_PAUSE:
	    DoBufferProcessingLoop();
		Reply(NOERROR);
		break;

	case CMD_STOP:
		IRtsp_Stop(pIRtsp, m_pConnParams);
	    Reply(NOERROR);
	    break;

	default:
	    DbgLog((LOG_ERROR, 1, TEXT("Unknown command %d received!"), cmd));
	    Reply((DWORD) E_NOTIMPL);
	    break;
	}
    } while (cmd != CMD_EXIT);

    hr = OnThreadDestroy();	// tidy up.
    if (FAILED(hr)) {
        DbgLog((LOG_ERROR, 1, TEXT("CSourceStream::OnThreadDestroy failed. Exiting thread.")));
        return 1;
    }

    DbgLog((LOG_TRACE, 1, TEXT("CSourceStream worker thread exiting")));
    return 0;
}