本文共 8687 字，大约阅读时间需要 28 分钟。

 

转载请标明是引用于

 例子代码:(包括客户端与服务端)(编译工具：VS2005)

 

参考书籍：<<DirectShow开发指南>>

1、设计思路.

1、1需要写一个能够接收服务器发来的数据，并提供给Filter Graph中其他解码的Source Filter，如图所示

0001的source Filter用于接收网络上发来的MPEG数据，且工作在拉模式下。(可参考 ).

DirectShow提供了MPEG1解码的所有filter，包括MPEG1-1 Stream Splitter、MPEG Vieo Decoder、MPEG Audio Decoder，因为MPEG-1 Stream Splitter（流分离器）只能工作在拉模式下，这样不得不把Source Filter也设计成拉模式。（不过如果有第三方的可以工作的推模式下的Splitter FIlter，则完全可以把Source Filter设计成推模式，即接到网络上的数据就主动传给Splitter Filter，而不是被动地等待Splitter去读取）

1、2 一各种双缓冲队列技术.

必须让Source Filter接收一定数量的数据，并缓存起来。因为在Source Filter和Splitter连接过程中，会从Source Filter中读取一部分数据，以获得数据的格式,否则Filter Graph无法完成连接。而当完整的Filter Graph构建完成，且处于运行状态后，Source Filter必须动态地接收数据，并持续把新的数据给Splitter。

这里使用双缓冲技术。

工作原理：

建立两个队列，一个是空闲的缓冲队列PoolList,用以接收存放数据，另一是尚未处理的数据缓冲队列DataList,等待Splitter读取。

当Source Filter使用socket接收数据后，从PoolList队列的头上取出一个缓冲块，存放数据，然后将这个缓冲块加入到DataList尾部，等待Splitter读取。

而在Splitter要求读取时，Source Filter会从DataList 队列的头上取出一个缓冲块，读取数据，再将读完的缓冲块加到PoolList尾部，回收再利用。

2、源码分析。

主要分析两个类CMemStream CMemReader 及CFilterGraph

//// MemFilter.h//#ifndef  __MemFilter_h__#define  __MemFilter_h__#include 
   
    #include 
    
     #include "asyncio.h"#include "asyncrdr.h"#include "CMediaSocketClient.h"#include "CDataAdmin.h"#include "MyDef.h"////  Define an internal filter that wraps the base CBaseReader stuff ////class CMemStream : public CAsyncStream{private:    CCritSec       m_csLock;    // 数据操作的同步对象	CDataAdmin *   m_pDataList; // 数据接收队列管理器    ULONG          m_ulPositionInPack;//  当前缓冲块内的偏移量    // Total length available	LONGLONG       m_llLength;  // 当前可以获得的数据总长度	// Bytes totally read    DWORD          m_dwTimeStart;public:    CMemStream(CDataAdmin* inBuffer)    {		// 成员变量初始化		m_pDataList         = inBuffer;		m_ulPositionInPack  = 0;		m_llLength          = 0;       }    HRESULT SetPointer(LONGLONG llPos)    {		return S_OK;    }	// 当Splitter要求数据时，必须在这个函数中提供    HRESULT Read(PBYTE pbBuffer,        // 装载读出数据的缓存地址                 DWORD dwBytesToRead,   // 要求读取的字节数                 BOOL bAlign,           // 数据对齐方式                 LPDWORD pdwBytesRead)  // 实际读取的字节数    {		if (m_pDataList == NULL)			return S_FALSE;		CAutoLock     lck(&m_csLock);		DWORD         dwHaveRead = 0;		PMPEG1_PACK   pPack  = NULL;		while (dwHaveRead < dwBytesToRead)		{			if (dwBytesToRead - dwHaveRead >= MPEG1_PACK_SIZE - m_ulPositionInPack)			{				// Just copy the whole pack data				pPack = m_pDataList->GetDataBuffer();				if (pPack != NULL)				{					CopyMemory((PVOID)(pbBuffer + dwHaveRead), 						(PVOID)((PBYTE)(pPack) + m_ulPositionInPack), (SIZE_T)MPEG1_PACK_SIZE - m_ulPositionInPack);					m_pDataList->ReleaseDataBuffer(pPack);					dwHaveRead += MPEG1_PACK_SIZE - m_ulPositionInPack;					m_ulPositionInPack = 0;				}				else if (m_pDataList->IsFlushing())				{				//	FILE * fp = fopen("c:\\log.txt","a+");				//	fprintf(fp,"failed!\n");				//	fclose(fp); 					return E_FAIL;				}				else 				{				//	FILE * fp = fopen("c:\\log.txt","a+");				//	fprintf(fp,"SLEEP!\n");				//	fclose(fp); 					Sleep(10);				}			}			else			{				// Copy part of the pack data				pPack = m_pDataList->PointToDataHead();				if (pPack != NULL)				{					m_ulPositionInPack = dwBytesToRead - dwHaveRead;					CopyMemory((PVOID)(pbBuffer + dwHaveRead), 						(PVOID)(pPack), (SIZE_T)m_ulPositionInPack);					dwHaveRead += m_ulPositionInPack;				}				else if (m_pDataList->IsFlushing())				{				//	FILE * fp = fopen("c:\\log.txt","a+");				//	fprintf(fp,"failed!\n");				//	fclose(fp); 					return E_FAIL;				}				else 				{				//	FILE * fp = fopen("c:\\log.txt","a+");				//	fprintf(fp,"SLEEP!\n");				//	fclose(fp);					Sleep(10);				}			}		}		*pdwBytesRead = dwBytesToRead;		return S_OK;    }    LONGLONG Size(LONGLONG *pSizeAvailable)    {        *pSizeAvailable = m_llLength;        return 0x7fffffffff;    }    DWORD Alignment()    {        return 1;    }    void Lock()    {        m_csLock.Lock();    }    void Unlock()    {        m_csLock.Unlock();    }	void AddAvailableLength(LONGLONG inLength)	{		m_llLength += inLength;	}};class CMemReader : public CAsyncReader{public:    //  We're not going to be CoCreate'd so we don't need registration    //  stuff etc    STDMETHODIMP Register()    {        return S_OK;    }    STDMETHODIMP Unregister()    {        return S_OK;    }    CMemReader(CMemStream *pStream, CMediaType *pmt, HRESULT *phr) :        CAsyncReader(NAME("Mem Reader"), NULL, pStream, phr)    {        m_mt = *pmt;    }};#endif  // __MemFilter_h__
    
   

 

class CMemReader : public CAsyncReader{public:    //  We're not going to be CoCreate'd so we don't need registration    //  stuff etc    STDMETHODIMP Register()    {        return S_OK;    }    STDMETHODIMP Unregister()    {        return S_OK;    }    CMemReader(CMemStream *pStream, CMediaType *pmt, HRESULT *phr) :        CAsyncReader(NAME("Mem Reader"), NULL, pStream, phr)    {        m_mt = *pmt;    }};

 

下面来看Filter Graph的构建过程。（参见CFiltreGraph::BuildeGraph的实现）。可以看到BuildGraph函数首先创建一个Filter Graph组件，然后获得一系统控制接口，

接下来创建source filter,值得注意的是，因为这是一个应用程序实现的Filter,不能通过CoCreateInstance函数去创建它，而要直接成成:

// construct source filter	// Media type	if (m_bInit)	{		CMediaType   mt;      		mt.majortype = MEDIATYPE_Stream;		mt.subtype   = MEDIASUBTYPE_MPEG1System;		m_pSourceStream  = new CMemStream(m_pDataList);		m_pSourceReader  = new CMemReader(m_pSourceStream, &mt, &hr);		m_pSourceReader->AddRef();  // 增加一个引用计数,防止COM自动删除		//  Add our filter		hr = m_pGB->AddFilter(m_pSourceReader, NULL);		if (FAILED(hr)) 		{			m_bInit = false;		}  	}

BuildGraph函数最后并没有将Source Filter的输出Pin连出去，因为Source Filter在没有缓存  一定数据的数据之前，与Filter连接是不会成功的，解决方法是，启动一个等待线程， 当Source Filter预先接收到足够的数据后，再构建完整的Filter Graph。

if (m_bInit)	{		// Waiting for event to start filter graph		::AfxBeginThread((AFX_THREADPROC)CFilterGraph::WaitingThrd, this);	}

// When data buffer enough, start the filter graphUINT CFilterGraph::WaitingThrd(void * pParam){	CFilterGraph * pGraph = (CFilterGraph *) pParam;	if (pGraph != NULL && pGraph->m_pDataList != NULL)	{		// 询问数据接收队列管理器，是否已经接受到了足够的数据		::WaitForSingleObject(pGraph->m_pDataList->m_hBufEnough, INFINITE);		if (!pGraph->IsRunning())		{			// 开始构建完成的Filter Graph			pGraph->StartGraph();		}	}	return 1;}

// 构建完成的Filter Graph,并运行它bool CFilterGraph::StartGraph(void){	if (m_bInit)	{		m_bRun = true;		// 使用智能连接，将Source Filter 的输出Pin连出来		HRESULT hr  = m_pGB->Render(m_pSourceReader->GetPin(0));		if (FAILED(hr))		{			m_bRun = false;			return false;		}        // 设置视窗的一些属性		hr = m_pVW->put_Visible(OAFALSE);		hr = m_pVW->put_Owner((OAHWND)m_hOwner);		hr = m_pVW->put_WindowStyle(WS_CHILD | WS_CLIPSIBLINGS | WS_CLIPCHILDREN);		hr = m_pVW->SetWindowPosition(m_nLeft, m_nTop, m_nWidth, m_nHeight);		// Enable the parent window to get the mouse and keyboard event in the video window		hr = m_pVW->put_MessageDrain((OAHWND)m_hOwner);		hr = m_pVW->put_Visible(OATRUE);				// run the filter graph		hr = m_pMC->Run();		return true;	}		return false;}

从CMediaSocketClient::ReceivingLoop函数的实现中，查看客户端的整个数据接收过程。可以看到，ReceivingLoop函数使用了一个while循环不断地接收数据，

并把数据交给队列管理；同时对接收到的数据包进行计数。

CMemReader(CMemStream *pStream, CMediaType *pmt, HRESULT *phr) :        CAsyncReader(NAME("Mem Reader"), NULL, pStream, phr)    {        m_mt = *pmt;    }

 

在CMemStream中有Read函数，它就是对收到的数据取出来.

// 当Splitter要求数据时，必须在这个函数中提供    HRESULT Read(PBYTE pbBuffer,        // 装载读出数据的缓存地址                 DWORD dwBytesToRead,   // 要求读取的字节数                 BOOL bAlign,           // 数据对齐方式                 LPDWORD pdwBytesRead)  // 实际读取的字节数    {		if (m_pDataList == NULL)			return S_FALSE;		CAutoLock     lck(&m_csLock);		DWORD         dwHaveRead = 0;		PMPEG1_PACK   pPack  = NULL;		while (dwHaveRead < dwBytesToRead)		{			if (dwBytesToRead - dwHaveRead >= MPEG1_PACK_SIZE - m_ulPositionInPack)			{				// Just copy the whole pack data				pPack = m_pDataList->GetDataBuffer();				if (pPack != NULL)				{					CopyMemory((PVOID)(pbBuffer + dwHaveRead), 						(PVOID)((PBYTE)(pPack) + m_ulPositionInPack), (SIZE_T)MPEG1_PACK_SIZE - m_ulPositionInPack);					m_pDataList->ReleaseDataBuffer(pPack);					dwHaveRead += MPEG1_PACK_SIZE - m_ulPositionInPack;					m_ulPositionInPack = 0;				}				else if (m_pDataList->IsFlushing())				{				//	FILE * fp = fopen("c:\\log.txt","a+");				//	fprintf(fp,"failed!\n");				//	fclose(fp); 					return E_FAIL;				}				else 				{				//	FILE * fp = fopen("c:\\log.txt","a+");				//	fprintf(fp,"SLEEP!\n");				//	fclose(fp); 					Sleep(10);				}			}			else			{				// Copy part of the pack data				pPack = m_pDataList->PointToDataHead();				if (pPack != NULL)				{					m_ulPositionInPack = dwBytesToRead - dwHaveRead;					CopyMemory((PVOID)(pbBuffer + dwHaveRead), 						(PVOID)(pPack), (SIZE_T)m_ulPositionInPack);					dwHaveRead += m_ulPositionInPack;				}				else if (m_pDataList->IsFlushing())

客户端的设计可以用下面的框图表示: