macasyncfile.cpp

linux下的一款播放器

void CMacAsyncFile::ReadAsyncData(BOOL ASYNC)
{
    m_pMutex->Lock();

    OSErr e = noErr;

#if defined(_DEBUG) && defined (_LOG_DATA)
    char str[255]; /* Flawfinder: ignore */
   ::sprintf(str, "ReadAsyncData: %d;g", (int) ASYNC); /* Flawfinder: ignore */
    DEBUGSTR(c2pstr(str));
#endif
 
    // mOutStandingCallbacks: if there are any, then we bail right
    // now 'cause we don't want scads of pending callbacks... the math
    // will work out in the end because subsequent calls will have us
    // read more bytes.
    
    if(!mRefNum || m_bFileDone || mOutStandingCallbacks > 0)
    {
        m_pMutex->Unlock();
	return;
    }

    ULONG32 emptysize = mReadQueue->GetAvailableElements();

    if(emptysize < kNeedToReadThreshhold)
    {
        m_pMutex->Unlock();
        DEBUGSTR("\p ReadAsyncData: emptysize < kNeedToReadThreshhold ;g");
 	return;
    }
    
    if (m_bCheckFromPQ)
    {
        BOOL bFilled = FillBufferFromPQ(m_ulReadPositionInFile);
        if (bFilled)
        {
            emptysize = mReadQueue->GetAvailableElements();
            if(emptysize < kNeedToReadThreshhold)
            {
                m_pMutex->Unlock();
                return;
            }
        }
    }
    
    if (emptysize > kMacMaxChunkToRead)
    {
	emptysize = kMacMaxChunkToRead;
    }
    
    if (m_bInternalSeekNeeded)
    {
        m_bInternalSeekNeeded = FALSE;
        PerformInternalSeek(ASYNC);
        // more stuff will be done once we get async callback for internal seek
        if (ASYNC)
        {
            m_pMutex->Unlock();
            return;
        }
    }

    HXParamBlockRec *pb = new HXParamBlockRec;

    memset(pb, 0, sizeof(HXParamBlockRec));

    pb->param = this;
    pb->io.ioParam.ioCompletion = zmIOCallbackUPP;
    pb->io.ioParam.ioRefNum = mRefNum;

    pb->entry = new HXAsyncQueueBuffer;
    memset(pb->entry, 0, sizeof(HXAsyncQueueBuffer));
    pb->entry->size = emptysize;
    pb->entry->current_size = emptysize;
    
    pb->entry->position_in_file = m_ulReadPositionInFile;

    char* tempbuffer = new char[emptysize];

    pb->entry->buffer 		= tempbuffer;
    pb->entry->current_buf_ptr  = tempbuffer;
    pb->entry->state = AB_STATE_EMPTY;

    pb->io.ioParam.ioBuffer = tempbuffer;
    pb->io.ioParam.ioReqCount = emptysize;

    HX_ASSERT(mSeekFromWhere == 0);

    e = ReadData(pb, ASYNC);


    if (!ASYNC)
    {
         if (e == noErr)
         {
	     pb->entry->state = AB_STATE_FULL;
	     pb->entry->size = pb->io.ioParam.ioActCount;
	     pb->entry->current_size = pb->io.ioParam.ioActCount;
	     m_ulReadPositionInFile += pb->entry->size;
	     mAsyncQueue->AddTail(pb->entry);
	 }
	 else
	 {
	     HX_DELETE(pb->entry);
	 }
	     	 
	 HX_DELETE(pb);
    }
    else
    {
        if (e != noErr)
        {
	    HX_DELETE(pb->entry);
	    HX_DELETE(pb);
        }
    }    

    DEBUGSTR("\p ReadAsyncData EXIT; g");

    m_pMutex->Unlock();

    return;
}

ULONG32 CMacAsyncFile::WriteAsyncData(IHXBuffer* pBuffer, BOOL ASYNC)
{
    ULONG32 outWrittenSize = 0;
    OSErr e = noErr;

#if defined(_DEBUG) && defined (_LOG_DATA)
    char str[255]; /* Flawfinder: ignore */
   ::sprintf(str, "WriteAsyncData: %d;g", (int) ASYNC); /* Flawfinder: ignore */
    DEBUGSTR(c2pstr(str));
#endif
 
    if(!mRefNum)
    {
	return outWrittenSize;
    }

    pBuffer->AddRef();

    HXParamBlockRec *pb = new HXParamBlockRec;

    memset(pb, 0, sizeof(HXParamBlockRec));

    pb->param = this;
    pb->io.ioParam.ioCompletion = zmIOCallbackUPP;
    pb->io.ioParam.ioRefNum = mRefNum;

    pb->entry = new HXAsyncQueueBuffer;
    memset(pb->entry, 0, sizeof(HXAsyncQueueBuffer));
    
    outWrittenSize = pBuffer->GetSize();
    pb->entry->size = outWrittenSize;
    pb->entry->current_size = outWrittenSize;
    pb->entry->state = AB_STATE_WRITE;

    char* tempbuffer = NULL;
    if (ASYNC)
    {
    	tempbuffer = new char[outWrittenSize];
    	BlockMoveData(pBuffer->GetBuffer(), tempbuffer, outWrittenSize);
    	pb->entry->buffer = tempbuffer;
    	pb->entry->current_buf_ptr = tempbuffer;
	pb->io.ioParam.ioBuffer = tempbuffer;
    }
    else
    {
	pb->io.ioParam.ioBuffer = ( char* ) pBuffer->GetBuffer();
    }

    pb->io.ioParam.ioReqCount = outWrittenSize;

    HX_ASSERT(mSeekFromWhere == 0);

    e = WriteData(pb, ASYNC);


    if (!ASYNC)
    {
	 HX_DELETE(pb->entry);
	 HX_DELETE(pb);
    }
    else
    {
        if (e != noErr)
        {
	    HX_DELETE(pb->entry);
	    HX_DELETE(pb);
        }
    }    

    DEBUGSTR("\p WriteAsyncData EXIT; g");
    pBuffer->Release();
    return outWrittenSize;
}

void CMacAsyncFile::UpdateFilePos(BOOL bASync, BOOL bInternal)
{
    AddRef();
    m_pMutex->Lock();
     DEBUGSTR("\p UpdateFilePos;g");
   OSErr e = noErr;

    if(!mRefNum)
    {
	m_pMutex->Unlock();
	Release();
	return;
    }

    HXParamBlockRec *pb = new HXParamBlockRec;

    memset(pb, 0, sizeof(HXParamBlockRec));

    pb->param = this;
    pb->io.ioParam.ioRefNum = mRefNum;
    pb->io.ioParam.ioReqCount = 0;
    pb->io.ioParam.ioCompletion = zmIOCallbackUPP;

    pb->entry = new HXAsyncQueueBuffer;
    memset(pb->entry, 0, sizeof(HXAsyncQueueBuffer));

    HX_ASSERT(mSeekFromWhere > 0);

    if(mSeekFromWhere)
    {
	pb->io.ioParam.ioPosMode = mSeekFromWhere;
	pb->io.ioParam.ioPosOffset = mSeekPos;

	mSeekFromWhere = 0;
	mSeekPos = 0;
	if (bInternal)
	{
            pb->entry->state = AB_STATE_ASYNC_INTERNAL_SEEK;
        }
        else
        {
            pb->entry->state = AB_STATE_ASYNC_SEEK;
        }
    }

    e = SeekData(pb, bASync);

    if (!bASync)
    {
         if (e == noErr)
         {
	     m_ulReadPositionInFile = pb->io.ioParam.ioPosOffset;
             if (!bInternal)
             {
	         mFilePos = pb->io.ioParam.ioPosOffset;
	     }
	 }
	     
	 HX_DELETE(pb->entry);
	 HX_DELETE(pb);
    }
    else
    {
        if (e != noErr)
        {
	    HX_DELETE(pb->entry);
	    HX_DELETE(pb);
        }
    }    

    if (!bASync && !bInternal)
    {
	PendingAsyncSeekDone();
    }

    m_pMutex->Unlock();
    Release();
    return;

}

OSErr
CMacAsyncFile::ReadData(HXParamBlockRec* pb, BOOL ASYNC)
{
    m_pMutex->Lock();

    OSErr e = noErr;

    if(ASYNC)
    {
	mAsyncQueue->AddTail(pb->entry);
	mOutStandingCallbacks++;
    }
    
    // xxxbobclark large PBRead's on SMB-mounted volumes are failing here.
    // That's why we reduced the constants in the .h file for carbon builds.

    e = PBRead((ParmBlkPtr) pb, ASYNC);

    if (e == eofErr)
    {
	    m_bFileDone = TRUE;
	    
        DEBUGSTR("\pFIle Done in ReadData;g");
	    /* Mask this error if this is the last read */
	    /* If either we are in aysnc mode OR we actually read some data */
	    if (ASYNC ||  pb->io.ioParam.ioActCount > 0)
	    {
	    	e = noErr;
	    }
    }

    if(e != noErr && ASYNC && mOutStandingCallbacks > 0)
    {
        mAsyncQueue->RemoveTail();
        mOutStandingCallbacks--;
    }

    m_pMutex->Unlock();

    return e;
}

OSErr
CMacAsyncFile::WriteData(HXParamBlockRec* pb, BOOL ASYNC)
{
    m_pMutex->Lock();

    OSErr e = noErr;

    if(ASYNC)
    {
	mAsyncQueue->AddTail(pb->entry);
	mOutStandingCallbacks++;
    }

    e = PBWrite((ParmBlkPtr) pb, ASYNC);

    if(e != noErr && ASYNC && mOutStandingCallbacks > 0)
    {
        mAsyncQueue->RemoveTail();
        mOutStandingCallbacks--;
    }

    m_pMutex->Unlock();

    return e;
}

OSErr
CMacAsyncFile::SeekData(HXParamBlockRec* pb, BOOL ASYNC)
{
    m_pMutex->Lock();

    OSErr e = noErr;

    if(ASYNC)
    {
	    mAsyncQueue->AddTail(pb->entry);
	    mOutStandingCallbacks++;
    }

    e = PBSetFPos((ParmBlkPtr) pb, ASYNC);

    if (e == eofErr)
    {
	DEBUGSTR("\p File done in SeekData");
	m_bFileDone = TRUE;
	e = noErr;
    }

    if(e != noErr && ASYNC && mOutStandingCallbacks > 0)
    {
        mAsyncQueue->RemoveTail();
	mOutStandingCallbacks--;
    }

    m_pMutex->Unlock();

    return e;
}

void
CMacAsyncFile::PendingAsyncSeekDone()
{
    AddRef();
    m_pMutex->Lock();
    HX_ASSERT(m_bSeekPending == TRUE);

    m_bSeekPending = FALSE;
     DEBUGSTR("\p PendingAsyncSeekDone;g");
   // ? should check status
    m_pResponse->AsyncSeekDone(HXR_OK);
    m_pMutex->Unlock();
    Release();
}

void
CMacAsyncFile::AllPendingCallbacksDone()
{   
    DEBUGSTR("\p AllPendingCallbacksDone;g");

    m_pMutex->Lock();
    
    HX_ASSERT(m_pPendingCallbackList->GetCount() == 0);
     	      
    if(m_bReadPending)
    {
        EnqueueAsyncBuffers();
	ProcessPendingRead();
    }

    /* If there is still a pending read and no outstanding reads
     * issue one more read
     */
    if(m_bReadPending && mOutStandingCallbacks == 0)
    {
        DEBUGSTR("\pm_bReadPending && mOutStandingCallbacks == 0;g");
	ReadAsyncData(TRUE);
    }
    m_pMutex->Unlock();
}

void
CMacAsyncFile::ProcessPendingRead()
{
    AddRef();
    m_pMutex->Lock();
    
    ULONG32 totalinqueue = mReadQueue->GetQueuedItemCount();
    ULONG32 numberInPendingCallbackList = m_pPendingCallbackList->GetCount();
    if(m_bReadPending &&
       ((m_bFileDone && numberInPendingCallbackList == 0) ||
       (totalinqueue >= m_ulPendingReadCount)))
    {
	m_bReadPending = FALSE;

	if (totalinqueue == 0)
	{
       DEBUGSTR("\p AsyncReadDone FAIL;g");
 	    m_pResponse->AsyncReadDone(HXR_FAIL, NULL);
	}
	else
	{
   	    UINT32 count = m_ulPendingReadCount;
	
	    if (count > totalinqueue)
	    {
		count = totalinqueue;
	    }

	    IHXBuffer *pBuffer = new CHXBuffer;
	    pBuffer->AddRef();
	    pBuffer->SetSize(count);
	    char *buf = (char*) pBuffer->GetBuffer();

		    mFilePos += mReadQueue->DeQueue(buf, count);
		    
#if defined(_DEBUG) && defined (_LOG_DATA)
       char str[255]; /* Flawfinder: ignore */
       ::sprintf(str, "AsyncReadDone: HXR_OK: %d;g", (int) count); /* Flawfinder: ignore */
        DEBUGSTR(c2pstr(str));
#endif	    
	    m_pResponse->AsyncReadDone(HXR_OK, pBuffer);
	    pBuffer->Release();
	}
    }
    m_pMutex->Unlock();
    Release();
}

HX_RESULT
CMacAsyncFile::SetAsyncResponse(CMacAsyncFileResponse * pResponse)
{
    if(!pResponse)
    {
	return HXR_UNEXPECTED;
    }

    m_pMutex->Lock();
    m_pResponse = pResponse;
    m_pMutex->Unlock();
    return HXR_OK;
}