q.cpp

Windows CE 6.0 Server 源码

    CHECK_POINTER(ppvObjOut);

    // we support IMSMQQueue and ISupportErrorInfo
    //
    if (DO_GUIDS_MATCH(riid, IID_IMSMQQueue)) {
        *ppvObjOut = (void *)(IMSMQQueue *)this;
        AddRef();
        return S_OK;
    } else if (DO_GUIDS_MATCH(riid, IID_ISupportErrorInfo)) {
        *ppvObjOut = (void *)(ISupportErrorInfo *)this;
        AddRef();
        return S_OK;
    }

    // call the super-class version and see if it can oblige.
    //
    return CAutomationObject::InternalQueryInterface(riid, ppvObjOut);
}



// TODO: implement your interface methods and property exchange functions
//       here.


//=--------------------------------------------------------------------------=
// CMSMQQueue::get_Access
//=--------------------------------------------------------------------------=
// Gets access
//
// Parameters:
//    plAccess [out] 
//
// Output:
//
// Notes:
//
HRESULT CMSMQQueue::get_Access(long FAR* plAccess)
{
    *plAccess = m_lAccess;
    return NOERROR;
}


//=--------------------------------------------------------------------------=
// CMSMQQueue::get_ShareMode
//=--------------------------------------------------------------------------=
// Gets sharemode
//
// Parameters:
//    plShareMode [out] 
//
// Output:
//
// Notes:
//
HRESULT CMSMQQueue::get_ShareMode(long FAR* plShareMode)
{
    *plShareMode = m_lShareMode;
    return NOERROR;
}

//=--------------------------------------------------------------------------=
// CMSMQQueue::get_queueinfo
//=--------------------------------------------------------------------------=
// Gets defining queueinfo
//
// Parameters:
//    ppqinfo [out] 
//
// Output:
//
// Notes:
//
HRESULT CMSMQQueue::get_QueueInfo(IMSMQQueueInfo FAR* FAR* ppqinfo)
{
    ASSERT(m_pqinfo, L"null.");
    *ppqinfo = m_pqinfo;
    ADDREF(*ppqinfo);
    return NOERROR;
}


//=--------------------------------------------------------------------------=
// CMSMQQueue::get_Handle
//=--------------------------------------------------------------------------=
// Gets queue handle
//
// Parameters:
//    plHandle [out] 
//
// Output:
//
// Notes:
//
HRESULT CMSMQQueue::get_Handle(long FAR* plHandle)
{
    // could be -1 if closed.
    *plHandle = (long)m_lHandle;
    return NOERROR;
}


//=--------------------------------------------------------------------------=
// CMSMQQueue::get_IsOpen
//=--------------------------------------------------------------------------=
// Tests if queue is open, i.e. has a valid handle
//
// Parameters:
//    pisOpen [out] 
//
// Output:
//
// Notes:
//
HRESULT CMSMQQueue::get_IsOpen(VARIANT_BOOL FAR* pisOpen)
{
  *pisOpen = (m_lHandle != INVALID_HANDLE_VALUE);
  return NOERROR;
}


//=--------------------------------------------------------------------------=
// CMSMQQueue::Close
//=--------------------------------------------------------------------------=
// Closes queue if open.
//
// Parameters:
//
// Output:
//
// Notes:
//
HRESULT CMSMQQueue::Close()
{
    HRESULT hresult = NOERROR;

    EnterCriticalSection(&g_csCallback);
    if (m_hCursor) {
      hresult = MQCloseCursor(m_hCursor);
      m_hCursor = 0;      
    }
    if (m_lHandle != INVALID_HANDLE_VALUE) {
      hresult = MQCloseQueue(m_lHandle);
      m_lHandle = INVALID_HANDLE_VALUE;
    }
    // REVIEW: should it be an error to attempt to close
    //  an already closed queue?
    //
    LeaveCriticalSection(&g_csCallback);


    return CreateErrorHelper(hresult, m_ObjectType);
}                   


//=--------------------------------------------------------------------------=
// CMSMQQueue::InternalReceive
//=--------------------------------------------------------------------------=
// Synchronously receive or peek next, matching msg.
//
// Parameters:
//  dwAction      [in]
//  hCursor       [in]
//  ptransaction  [in]
//  wantDestQueue [in]  if missing -> FALSE
//  wantBody      [in]  if missing -> TRUE
//  lReceiveTimeout [in]
//  ppmsg         [out] pointer to pointer to received message.  
//                      NULL if don't want msg.
//
// Output:
//  Returns NULL in *ppmsg if no received msg.
//
// Notes:
//  Synchronously receive a message.  
//  Execution is blocked until either a matching message arrives 
//   or ReceiveTimeout expires.
//
HRESULT CMSMQQueue::InternalReceive(
    DWORD dwAction, 
    HANDLE hCursor,
    VARIANT *pvarTransaction,
    VARIANT *wantDestQueue,
    VARIANT *wantBody,
    VARIANT *pvarReceiveTimeout,
    IMSMQMessage FAR* FAR* ppmsg)
{
    MQMSGPROPS msgprops;
    CMSMQMessage *pmsg = NULL;
    DWORD dwBodySize;
    UINT iBody = (DWORD)-1, iBodySize = (DWORD)-1;

#if 0  // no transactions on CE
    ITransaction *ptransaction = NULL;
    BOOL isRealXact = FALSE;
#endif // 0

    BOOL fWantDestQueue = FALSE;
    BOOL fWantBody = TRUE;
    long lReceiveTimeout = INFINITE;
#if DEBUG
    UINT iLoop = 0;
#endif // DEBUG
    HRESULT hresult = NOERROR;


    if (ppmsg == NULL) {
      return E_INVALIDARG;
    }
    *ppmsg = NULL;     
    //
    // process optional params
    //
    if (V_VT(wantDestQueue) != VT_ERROR ) {
      fWantDestQueue = GetBool(wantDestQueue);
    }
    if (V_VT(wantBody) != VT_ERROR ) {
      fWantBody = GetBool(wantBody);
    }
    IfFailRet(GetOptionalReceiveTimeout(
                pvarReceiveTimeout,
                &lReceiveTimeout));
    IfNullRet(pmsg = new CMSMQMessage(NULL));
    IfFailGo(pmsg->CreateReceiveMessageProps(
              fWantDestQueue,
              fWantBody,
              &msgprops));
    //
    // get optional transaction...
    //
#if 0 // no transactions on CE
    IfFailGo(GetOptionalTransaction(
               pvarTransaction,
               &ptransaction,
               &isRealXact));
#endif

    do {
#if DEBUG
      //
      // we should only ever perform this loop at most twice
      //  per-message
      //
      ASSERT(iLoop < 2, L"possible infinite recursion?");
#endif // DEBUG
      //
      // 1694: need to special-case retrying PeekNext 
      //  after a buffer overflow by using vanilla
      //  Peek on retries since otherwise Falcon will
      //  advance the peek cursor unnecessarily.
      //
      if (hresult == MQ_ERROR_BUFFER_OVERFLOW) {
        if (dwAction == MQ_ACTION_PEEK_NEXT) {
          dwAction = MQ_ACTION_PEEK_CURRENT;
        }  
      }
     
      
      hresult = MQReceiveMessage(m_lHandle, 
                                 lReceiveTimeout,
                                 dwAction,
                                 (MQMSGPROPS *)&msgprops,
                                 0,   // dwAppDefined
                                 0,   // fnRcvClbk
                                 hCursor,
                                 NULL);  

      
      if (hresult == MQ_ERROR_BUFFER_OVERFLOW) {
        // grow buffer...
        iBodySize = OrdinalOfPropId(&msgprops, PROPID_M_BODY_SIZE);
        iBody = OrdinalOfPropId(&msgprops, PROPID_M_BODY);
        ASSERT((iBodySize != -1) && (iBody != -1),
               L"msgprops lookup failed.");
        dwBodySize = msgprops.aPropVar[iBodySize].lVal;
        IfNullGo(AllocateBodyBuffer(&msgprops, iBody, dwBodySize));
      } 
#if DEBUG
      iLoop++;
#endif // DEBUG

    } while (hresult == MQ_ERROR_BUFFER_OVERFLOW);
    if (SUCCEEDED(hresult)) {
      // set message props
      IfFailGo(pmsg->SetMessageProps(&msgprops));
      *ppmsg = pmsg;
    }
    //
    // fall through...
    //

Error:
    if (FAILED(hresult)) {
      delete pmsg;
      ASSERT(*ppmsg == NULL, L"msg should be NULL.");
      if (hresult == MQ_ERROR_IO_TIMEOUT) {
        //
        // map time-out error to NULL msg return
        //
        hresult = NOERROR;
      }
      	//
      	// 2209: only free msg buf if body was requested
      	//
      if (fWantBody) {
        //
        // 2099: need to free message buffer since it has
        //  no owner.
        //
        if (iBody == -1) {
          iBody = OrdinalOfPropId(&msgprops, PROPID_M_BODY);
          ASSERT(iBody != -1, L"msgprops lookup failed.");
        }
        FreeBodyBuffer(&msgprops, iBody);
      }
    }
    CMSMQMessage::FreeMessageProps(&msgprops);
    
#if 0 // no transactions on CE
    if (isRealXact) {
      RELEASE(ptransaction);
    }
#endif

    return CreateErrorHelper(hresult, m_ObjectType);
}


//=--------------------------------------------------------------------------=
// CMSMQQueue::Receive
//=--------------------------------------------------------------------------=
// Synchronously receives a message.
//
// Parameters:
//  ptransaction  [in, optional]
//  ppmsg         [out] pointer to pointer to received message.
//
// Output:
//  Returns NULL in *ppmsg if no received msg.
//
// Notes:
//  Synchronously receive a message.  
//  Execution is blocked until either a matching message arrives 
//   or ReceiveTimeout expires.
//
HRESULT CMSMQQueue::Receive(
    VARIANT *ptransaction,
    VARIANT *wantDestQueue,
    VARIANT *wantBody,
    VARIANT *lReceiveTimeout,
    IMSMQMessage FAR* FAR* ppmsg)
{
    return InternalReceive(MQ_ACTION_RECEIVE, 
                           0, // no cursor
                           NULL,
                           wantDestQueue,
                           wantBody,
                           lReceiveTimeout,
                           ppmsg);
}


//=--------------------------------------------------------------------------=
// CMSMQQueue::Peek
//=--------------------------------------------------------------------------=
// Synchronously peeks at a message.
//
// Parameters:
//  ppmsg     [out] pointer to pointer to peeked message.
//
// Output:
//  Returns NULL in *ppmsg if no received msg.
//