rssmanager.cpp

beereader source code

// RSSManager.cpp: implementation of the CRSSManager class.
//
//////////////////////////////////////////////////////////////////////

#include "stdafx.h"
#include "RSSManager.h"

#ifdef _DEBUG
#undef THIS_FILE
static char THIS_FILE[]=__FILE__;
#define new DEBUG_NEW
#endif

//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////

CRSSManager::CRSSManager(CRSSReader* pReader,CRSSWriter* pWriter)
{
	CoInitialize( NULL );

	if( pReader && pWriter )
	{
		m_Reader = pReader;
		m_Writer = pWriter;
	}
	else
	{
		m_ReaderXML = new CRSSReaderXML();
		m_WriterXML = new CRSSWriterXML();
		m_Reader = (CRSSReader*)m_ReaderXML;
		m_Writer = (CRSSWriter*)m_WriterXML;
	}

	ReaderQueue.RemoveAll();
	WriterQueue.RemoveAll();
	hReaderThread = hWriterThread = NULL;
	bReaderRunning = bWriterRunning = FALSE;
	mutexReader = CreateMutex(NULL,FALSE,_T("Reader Mutex"));
	mutexWriter = CreateMutex(NULL,FALSE,_T("Writer Mutex"));

	m_UpdateTimer = 60; //缺省为60分钟更新一次.

	m_hSuspend_Writer_Thread = NULL;
	m_ReaderingChannel = m_WriteringChannel = NULL;
}

CRSSManager::~CRSSManager()
{
	if( m_ReaderXML ) delete( m_ReaderXML );
	if( m_WriterXML ) delete( m_WriterXML );
    
	if( WaitForSingleObject(mutexReader,1000L) == WAIT_OBJECT_0 )
	{
	   if( bReaderRunning )
		   TerminateThread(hReaderThread,0);
	   ReleaseMutex( mutexReader );
	}
	
	CRSSChannel* pChannel;
	int i;
	for( i=0;i<ReaderQueue.GetSize();i++)
	{
		pChannel = (CRSSChannel*)ReaderQueue.GetAt(i);
		ASSERT(pChannel);
		delete( pChannel );
	}
	ReaderQueue.RemoveAll();

	if( WaitForSingleObject(mutexWriter,1000L) == WAIT_OBJECT_0 )
	{
	   if( bWriterRunning )
		   TerminateThread(hWriterThread,0);
	   ReleaseMutex( mutexWriter );
	}
	
	for( i=0;i<WriterQueue.GetSize();i++)
	{
		pChannel = (CRSSChannel*)WriterQueue.GetAt(i);
		ASSERT(pChannel);
		delete( pChannel );
	}
	WriterQueue.RemoveAll();

	CoUninitialize();
}
/** 
  * 添加一个频道,一般用于添加一个新的频道。
  *
  * szURL - 频道的URI。
  *
 **/
CRSSChannel* CRSSManager::AddChannel(CRSSChannel*& pChannel,CString szURL)
{	
//	CRSSChannel* pChannel;
	if( WaitForSingleObject(mutexReader,1000L) == WAIT_OBJECT_0 )
	{
	   if( !pChannel )
		  pChannel = new CRSSChannel; 
	   pChannel->SetURL( szURL );

	   m_Channels.Add( pChannel );
	   ReaderQueue.InsertAt( 0,pChannel);
	   ReleaseMutex(mutexReader);
	   BeginReaderThread();
	   return pChannel;
	}
	return NULL;
}
/** 
  * 添加一个频道。一般用于加载一个缓冲中的频道。
  *
  * szGUID - 频道缓冲的唯一标识，可以是一个文件名，也可以是一个数据库表，这要根据CRSSWriter实现类而定。
  * bRefresh - 加载频道缓冲后是否立即进行频道刷新工作。
 **/
CRSSChannel* CRSSManager::AddChannel(CRSSChannel*& pChannel,CString szGUID, BOOL bRefresh)
{
	if( m_Reader )
	{
		if( !m_Reader->LoadLocal( pChannel,_T(""),szGUID ) ) return NULL;
		if( bRefresh )
		{
			if( WaitForSingleObject(mutexReader,1000L) == WAIT_OBJECT_0 )
			{	           
	           m_Channels.Add(pChannel);
			   ReaderQueue.InsertAt(0,pChannel);
	           ReleaseMutex(mutexReader);
	           BeginReaderThread();
			}
		}
		return pChannel;
	}
	return NULL;
}

BOOL  CRSSManager::RemoveChannel( CRSSChannel* pChannel)
{
    if( (WaitForSingleObject(mutexReader,1000L) == WAIT_OBJECT_0)  && (WaitForSingleObject(mutexWriter,1000L) == WAIT_OBJECT_0) )
	{	           
	    if( (m_ReaderingChannel != pChannel) && (m_WriteringChannel != pChannel) )
		{
			int i;
			CRSSChannel* pch = NULL;
			for( i=0; i<ReaderQueue.GetSize(); i++ )
			{
			    pch = (CRSSChannel*)ReaderQueue.GetAt(i);
				if( pch == pChannel )
				{
					ReaderQueue.RemoveAt(i);
					break;
				}
			}
            
			ReleaseMutex(mutexReader);

			for( i=0; i<WriterQueue.GetSize(); i++ )
			{
			    pch = (CRSSChannel*)WriterQueue.GetAt(i);
				if( pch == pChannel )
				{
					WriterQueue.RemoveAt(i);
					break;
				}
			}

  		   ReleaseMutex(mutexWriter);
           
		   delete(pChannel);
		}
		else
		{
		    pChannel->SetRemoveTag(TRUE);
			ReleaseMutex(mutexReader);
            ReleaseMutex(mutexWriter);
		} //end if.
	}

	return TRUE;
}
BOOL CRSSManager::BeginReaderThread()
{
	DWORD dw;
	if( WaitForSingleObject(mutexReader,1000L) == WAIT_OBJECT_0 )
	{
	    if( bReaderRunning == FALSE && ReaderQueue.GetSize() > 0 )
		{
		  if( !hReaderThread )
			  hReaderThread = ::CreateThread( NULL,0,ReaderThread,(LPVOID)this,0,&dw );
	      if( !hReaderThread )
		  {
	          ReleaseMutex(mutexReader);
			  return FALSE;
		  }
		  else  bReaderRunning = TRUE;
		}
	    ReleaseMutex(mutexReader);
	}	
	return TRUE;
}

DWORD WINAPI ReaderThread(LPVOID pParam)
{
     CRSSManager* pMgr = (CRSSManager*)pParam;
	 ASSERT(pMgr);

	 if( !pMgr->GetReader() ) return 0;

	 CRSSChannel* pChannel = NULL;

	 CoInitialize( NULL );

	 while(1)
	 {
	    if( WaitForSingleObject(pMgr->mutexReader,1000L) == WAIT_OBJECT_0 )
		{			
			if( pMgr->ReaderQueue.GetSize() > 0 ) //读队列非空.
			{
				pChannel = (CRSSChannel*)pMgr->ReaderQueue.GetAt(0);
				while( pChannel->GetRemoveTag() )
				{
				   if( WaitForSingleObject(pMgr->mutexWriter,1000L) == WAIT_OBJECT_0 )
				   {
				       if( pMgr->m_WriteringChannel != pChannel )
					   {
					      	int i;
			                CRSSChannel* pch = NULL;
			                for( i=0; i<pMgr->ReaderQueue.GetSize(); i++ )
							{
			                    pch = (CRSSChannel*)pMgr->ReaderQueue.GetAt(i);
				                if( pch == pChannel )
								{
					               pMgr->ReaderQueue.RemoveAt(i);
					               break;
								}
							}
             			    for( i=0; i<pMgr->WriterQueue.GetSize(); i++ )
							{
			                    pch = (CRSSChannel*)pMgr->WriterQueue.GetAt(i);
				                if( pch == pChannel )
								{
					                pMgr->WriterQueue.RemoveAt(i);
					                break;
								}
							}  		                    
                            delete(pChannel);
							if( pMgr->ReaderQueue.GetSize() > 0 )
							    pChannel = (CRSSChannel*)pMgr->ReaderQueue.GetAt(0);
							else 
							{
								pChannel = NULL;
								break;
							}
					   } //end if
                       ReleaseMutex(pMgr->mutexWriter);
				   }// end if
				}// while end.
				if( pChannel )
				{
				   pMgr->m_ReaderingChannel = pChannel;
				   ReleaseMutex(pMgr->mutexReader);
				}
				else
				{
				   pMgr->bReaderRunning = FALSE;
				   ReleaseMutex(pMgr->mutexReader);
				   CoUninitialize();
				   pMgr->hReaderThread = NULL;
				   return 1;
				}
			}
			else
			{
				pMgr->bReaderRunning = FALSE;
				ReleaseMutex(pMgr->mutexReader);
				CoUninitialize();
				pMgr->hReaderThread = NULL;
				return 1;
			}
		}

		pMgr->GetReader()->Read(pChannel,pChannel->GetURL());

        if( WaitForSingleObject(pMgr->mutexReader,1000L) == WAIT_OBJECT_0 )
		{
			if( pMgr->ReaderQueue.GetSize() > 0 )
			{				
				if( pChannel == (CRSSChannel*)pMgr->ReaderQueue.GetAt(0) )
				{
					pMgr->ReaderQueue.RemoveAt(0); 
				}
			}
			ReleaseMutex(pMgr->mutexReader);
		}

	 }
	CoUninitialize();
	pMgr->hReaderThread = NULL;
	return 0;
}

///////////////////////////////////////////////////////////////////////////////
// Write Process...
BOOL  CRSSManager::CreateRSSFeed( CRSSChannel* pChannel , CString szFileName )
{
	if( !pChannel || szFileName.IsEmpty() || !m_Writer )  return FALSE;

	return m_Writer->Write( pChannel,szFileName );

}

BOOL  CRSSManager::SaveLocalCache( CRSSChannel* pChannel )
{
    if( !pChannel ) return FALSE; 	
	if( WaitForSingleObject(mutexWriter,1000L) == WAIT_OBJECT_0 )
	{
	    int i;
		for( i=0;i<WriterQueue.GetSize(); i++ )
		   if( (CRSSChannel*)WriterQueue.GetAt(i) == pChannel )
			   break;
	   if( i == WriterQueue.GetSize() )
	   {
	       pChannel->SetSaveTag(TRUE);
		   WriterQueue.Add(pChannel);
	   }
	   ReleaseMutex(mutexWriter);
	   return  BeginWriterThread();	   
	}
	
	return FALSE;
}

BOOL CRSSManager::BeginWriterThread()
{
	DWORD dw;
	if( WaitForSingleObject(mutexWriter,1000L) == WAIT_OBJECT_0 )
	{
	    if( bWriterRunning == FALSE && WriterQueue.GetSize() > 0 )
		{
		  if( !hWriterThread )
			  hWriterThread = ::CreateThread( NULL,0,WriterThread,(LPVOID)this,0,&dw );
	      else
			  ::ResumeThread( hWriterThread );
		  if( !hWriterThread )
		  {
	          ReleaseMutex(mutexWriter);
			  return FALSE;
		  }
		  else  bWriterRunning = TRUE;
		}
	    ReleaseMutex(mutexWriter);
	}	
	return TRUE;
}

DWORD WINAPI WriterThread(LPVOID pParam)
{
     CRSSManager* pMgr = (CRSSManager*)pParam;
	 ASSERT(pMgr);

	 if( !pMgr->GetWriter() ) return 0;

	 CRSSChannel* pChannel = NULL;

	 CoInitialize( NULL );

	 ::SendMessage(pMgr->GetSuspend_Writer_Thread(),WM_SUSPEND_WRITER,0,0); 

	 while(1)
	 {
	    if( WaitForSingleObject(pMgr->mutexWriter,1000L) == WAIT_OBJECT_0 )
		{
			if( pMgr->WriterQueue.GetSize() > 0 )
			{
				pChannel = (CRSSChannel*)pMgr->WriterQueue.GetAt(0);
				while( pChannel->GetRemoveTag() )
				{
				   if( WaitForSingleObject(pMgr->mutexReader,1000L) == WAIT_OBJECT_0 )
				   {
				       if( pMgr->m_ReaderingChannel != pChannel )
					   {
					      	int i;
			                CRSSChannel* pch = NULL;
			                for( i=0; i<pMgr->ReaderQueue.GetSize(); i++ )
							{
			                    pch = (CRSSChannel*)pMgr->ReaderQueue.GetAt(i);
				                if( pch == pChannel )
								{
					               pMgr->ReaderQueue.RemoveAt(i);
					               break;
								}
							}
             			    for( i=0; i<pMgr->WriterQueue.GetSize(); i++ )
							{
			                    pch = (CRSSChannel*)pMgr->WriterQueue.GetAt(i);
				                if( pch == pChannel )
								{
					                pMgr->WriterQueue.RemoveAt(i);
					                break;
								}
							}  		                    
                            delete(pChannel);
							if( pMgr->WriterQueue.GetSize() > 0 )
							    pChannel = (CRSSChannel*)pMgr->WriterQueue.GetAt(0);
							else 
							{
								pChannel = NULL;
								break;
							}
					   } //end if
                       ReleaseMutex(pMgr->mutexReader);
				   }// end if
				}// while end.
				if( pChannel )
				{
				   pMgr->m_WriteringChannel = pChannel;
				   ReleaseMutex(pMgr->mutexWriter);
				}
				else
				{
				     pMgr->bWriterRunning = FALSE;
				     ReleaseMutex(pMgr->mutexWriter);
                     ::SendMessage(pMgr->GetSuspend_Writer_Thread(),WM_SUSPEND_WRITER,3,0); 
				     if( ::SuspendThread(pMgr->hWriterThread) == -1 ) //处于挂起状态.
					 {
				        CoUninitialize();
				        pMgr->hWriterThread = NULL;
				        return 1;
					 }
				     ::SendMessage(pMgr->GetSuspend_Writer_Thread(),WM_SUSPEND_WRITER,1,0); 			   
				}
			}
			else
			{
				pMgr->bWriterRunning = FALSE;
				ReleaseMutex(pMgr->mutexWriter);
                ::SendMessage(pMgr->GetSuspend_Writer_Thread(),WM_SUSPEND_WRITER,3,0); 
				if( ::SuspendThread(pMgr->hWriterThread) == -1 ) //处于挂起状态.
				{
				   CoUninitialize();
				   pMgr->hWriterThread = NULL;
				   return 1;
				}
				::SendMessage(pMgr->GetSuspend_Writer_Thread(),WM_SUSPEND_WRITER,1,0); 			
			}
		}

		pMgr->GetWriter()->Save( pChannel,pChannel->GetCacheGUID() );  
       
		::SendMessage(pMgr->GetSuspend_Writer_Thread(),WM_SUSPEND_WRITER,2,(LPARAM)pChannel); 			
			
        if( WaitForSingleObject(pMgr->mutexWriter,1000L) == WAIT_OBJECT_0 )
		{
			if( pMgr->WriterQueue.GetSize() > 0 )
			{				
				if( pChannel == (CRSSChannel*)pMgr->WriterQueue.GetAt(0) )
				{
					pMgr->WriterQueue.RemoveAt(0); 
				}
			}
			ReleaseMutex(pMgr->mutexWriter);
		}

	 }
	CoUninitialize();
	pMgr->hWriterThread = NULL;
	return 0;
}

////////////////////////////////////////////////////////////////////////////
// Update...
int  CRSSManager::Update(CRSSChannel* pChannel,BOOL bAuto )
{
	int iRet = 0;
	if( WaitForSingleObject(mutexReader,1000L) == WAIT_OBJECT_0 )
	{
	   if( pChannel )	   
	   {
	      if( (bAuto && CheckUpdate(pChannel)) || !bAuto )
		  {
			  ReaderQueue.Add(pChannel);		  
		      iRet = 1;
		  }
	   }
	  
  	   ReleaseMutex(mutexReader);
	   BeginReaderThread();
	}

	return iRet;

}

BOOL  CRSSManager::CheckUpdate(CRSSChannel* pChannel)
{
	CTime ct = CTime::GetCurrentTime();
    CTimeSpan ts = ct-pChannel->GetLastUpdateTime();

	if( pChannel->m_UpdateTimer <= 0 )  return FALSE;

	int iTimer = pChannel->m_UpdateTimer>pChannel->GetTTL()?pChannel->GetTTL():pChannel->m_UpdateTimer;
	if( ts.GetTotalMinutes() > iTimer ) //达到更新时间.
	{
		//检测是否满足SkipHours条件.
		if( pChannel->GetSkipHours(ct.GetHour()) )
			return FALSE;
		//检测是否满足SkipDays条件.
		if( pChannel->GetSkipDays(ct.GetDayOfWeek()-1) )
			return FALSE;
		pChannel->SetLastUpdateTime();
		return TRUE;
	}
	
	return FALSE;
}