q.cpp

Windows CE 6.0 Server 源码

// Notes:
//  Synchronously peek at a message.  
//  Execution is blocked until either a matching message arrives 
//   or ReceiveTimeout expires.
//
HRESULT CMSMQQueue::Peek(
    VARIANT *wantDestQueue,
    VARIANT *wantBody,
    VARIANT *lReceiveTimeout,
    IMSMQMessage FAR* FAR* ppmsg)
{
    return InternalReceive(MQ_ACTION_PEEK_CURRENT, 
                           0,             
                           NULL,          // no transaction
                           wantDestQueue,
                           wantBody,
                           lReceiveTimeout,
                           ppmsg);
}


//=--------------------------------------------------------------------------=
// CMSMQQueue::PeekCurrent
//=--------------------------------------------------------------------------=
// Synchronously peeks at a message with cursor.
//
// Parameters:
//  ppmsg     [out] pointer to pointer to peeked message.
//
// Output:
//  Returns NULL in *ppmsg if no received msg.
//
// Notes:
//  Synchronously peek at a message.  
//  Execution is blocked until either a matching message arrives 
//   or ReceiveTimeout expires.
//
HRESULT CMSMQQueue::PeekCurrent(
    VARIANT *wantDestQueue,
    VARIANT *wantBody,
    VARIANT *lReceiveTimeout,
    IMSMQMessage FAR* FAR* ppmsg)
{
    return InternalReceive(MQ_ACTION_PEEK_CURRENT, 
                           m_hCursor,     
                           NULL,          // no transaction
                           wantDestQueue,
                           wantBody,
                           lReceiveTimeout,
                           ppmsg);
}


//=--------------------------------------------------------------------------=
// CMSMQQueue::ReceiveCurrent
//=--------------------------------------------------------------------------=
// Synchronously receive next matching message.
//
// Parameters:
//  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::ReceiveCurrent(
    VARIANT *ptransaction,
    VARIANT *wantDestQueue,
    VARIANT *wantBody,
    VARIANT *lReceiveTimeout,
    IMSMQMessage FAR* FAR* ppmsg)
{
    return InternalReceive(MQ_ACTION_RECEIVE, 
                           m_hCursor,
                           ptransaction,
                           wantDestQueue,
                           wantBody,
                           lReceiveTimeout,
                           ppmsg);
}


//=--------------------------------------------------------------------------=
// CMSMQQueue::PeekNext
//=--------------------------------------------------------------------------=
// Synchronously peek at next matching message.
//
// Parameters:
//  ppmsg     [out] pointer to pointer to peeked message.
//
// Output:
//  Returns NULL in *ppmsg if no peekable msg.
//
// Notes:
//  Synchronously peek at a message.  
//  Execution is blocked until either a matching message arrives 
//   or ReceiveTimeout expires.
//
HRESULT CMSMQQueue::PeekNext(
    VARIANT *wantDestQueue,
    VARIANT *wantBody,
    VARIANT *lReceiveTimeout,
    IMSMQMessage FAR* FAR* ppmsg)
{
    return InternalReceive(MQ_ACTION_PEEK_NEXT, 
                           m_hCursor,
                           NULL,   // no transaction
                           wantDestQueue,
                           wantBody,
                           lReceiveTimeout,
                           ppmsg);
}


//=--------------------------------------------------------------------------=
// CMSMQQueue::EnableNotification
//=--------------------------------------------------------------------------=
// Enables async message arrival notification
//
// Parameters:
//  pqvent          [in]  queue's event handler
//  msgcursor       [in]  indicates whether they want
//                        to wait on first, current or next.
//                        Default: MQMSG_FIRST
//
// Output:
//
// Notes:
//
HRESULT CMSMQQueue::EnableNotification(
    IMSMQEvent *pqevent,
    VARIANT *pvarMsgCursor,
    VARIANT *pvarReceiveTimeout)
{
    // Callback params -- ownership is transferred to
    //  callback.
    // UNDONE: what if callback isn't invoked? mem will leak.
    //  
    MQMSGPROPS *pmsgprops = NULL;
    MQMSGCURSOR msgcursor = MQMSG_FIRST;
    long lReceiveTimeout = INFINITE;
    static BOOL fWeirdLoadLibraryWorkaround = FALSE;
    HRESULT hresult;
    //
    // 1884: error if queue still has outstanding event
    //  handler. 
    // UNDONE: need specific error
    //
    if (m_hasActiveEventHandler) {
      return E_INVALIDARG;
    }
    if (pqevent == NULL) {
      return E_INVALIDARG;
    }
    if (pvarMsgCursor) {
      if (V_VT(pvarMsgCursor) != VT_ERROR) {
        IfFailRet(VariantChangeType(pvarMsgCursor, 
                                    pvarMsgCursor, 
                                    0, 
                                    VT_I4));
        msgcursor = (MQMSGCURSOR)(V_I4(pvarMsgCursor));
        if ((msgcursor != MQMSG_FIRST) &&
            (msgcursor != MQMSG_CURRENT) &&
            (msgcursor != MQMSG_NEXT)) {
          return E_INVALIDARG;
        }
      }
    }
    IfFailRet(GetOptionalReceiveTimeout(
                pvarReceiveTimeout,
                &lReceiveTimeout));
    IfNullRet(pmsgprops = new MQMSGPROPS); 
    InitMessageProps(pmsgprops);

	// sync to make sure the reference counts are accurate
    EnterCriticalSection(&m_CSlocal);
    	RELEASE(m_pqevent);
    	m_pqevent = pqevent;
    	ADDREF(m_pqevent);
	LeaveCriticalSection(&m_CSlocal);
	
	IfFailGo(CMSMQMessage::CreateAsyncReceiveMessageProps(pmsgprops));
	CMSMQMessage::FreeMessageProps(pmsgprops);
	IfFailGo(CMSMQMessage::CreateAsyncReceiveMessageProps(pmsgprops));
	
    // UNDONE: need to load mqoa.dll one extra time to workaround
    //  case in which Falcon rt calls to ActiveX
    //  ReceiveCallback after its dl, i.e. this one,
    //  has been unloaded.  This might happen if it wants
    //  to report that the callback has been canceled!
    // Note: this means that once you've made an async
    //  receive request then this dll will remain loaded.
    //
    if (!fWeirdLoadLibraryWorkaround) {
      LoadLibrary(L"mqoa.dll");
      fWeirdLoadLibraryWorkaround = TRUE;
    }
    //
    // register callback for the async notification.
    //
    // 2016: workaround VC5.0 codegen: operator ?: doesn't
    /// work correctly?
    //
    DWORD dwAction;
    PMQRECEIVECALLBACK fnReceiveCallback;
    HANDLE hCursor;
    
    switch (msgcursor) {
    case MQMSG_FIRST:
      dwAction = MQ_ACTION_PEEK_CURRENT;
      fnReceiveCallback = ReceiveCallback;
      hCursor = 0;
      break;
    case MQMSG_CURRENT:
      dwAction = MQ_ACTION_PEEK_CURRENT;
      fnReceiveCallback = ReceiveCallbackCurrent;
      hCursor = m_hCursor;
      break;
    case MQMSG_NEXT:
      dwAction = MQ_ACTION_PEEK_NEXT;
      fnReceiveCallback = ReceiveCallbackNext;
      hCursor = m_hCursor;
      break;
    default:
      ASSERT(0, L"bad msgcursor!");
      break;
    } // switch

    hresult = MQReceiveMessage(
                m_lHandle, 
                lReceiveTimeout,
                dwAction,
                pmsgprops,
                0,                            // overlapped
                fnReceiveCallback,
                hCursor,
                NULL               // no transaction
              );
              
    if (SUCCEEDED(hresult)) {
      //
      // 1884: queue now has active event handler,
      //  this is consumed by the callback.
      //
      m_hasActiveEventHandler = TRUE;
    }
    //
    // 1212: ignore BUFFER_OVERFLOW errors -- they will be 
    //  handled later by InternalReceive.
    //
    if (hresult == MQ_ERROR_BUFFER_OVERFLOW) {
      return NOERROR;
    }
Error:
    if (FAILED(hresult)) {
      CMSMQMessage::FreeMessageProps(pmsgprops);
    }
    return CreateErrorHelper(hresult, m_ObjectType);
}

//=--------------------------------------------------------------------------=
// CMSMQQueue::Reset
//=--------------------------------------------------------------------------=
// Resets message queue
//
// Parameters:
//
// Output:
//
// Notes:
//
HRESULT CMSMQQueue::Reset()
{
    HRESULT hresult = NOERROR;

    if (m_hCursor) {
      hresult = MQCloseCursor(m_hCursor);
    }
    m_hCursor = 0;

    if ((m_lAccess & (MQ_RECEIVE_ACCESS | MQ_PEEK_ACCESS)) && SUCCEEDED(hresult)) {
      hresult = MQCreateCursor(m_lHandle, &m_hCursor);
    }

    return CreateErrorHelper(hresult, m_ObjectType);

}


//=--------------------------------------------------------------------------=
// CMSMQQueue::Init
//=--------------------------------------------------------------------------=
// Inits new instance with handle and creating MSMQQueueInfo instance.
//
// Parameters:
//    pqinfo       [in]  
//    lHandle     [in] 
//    lAccess     [in]
//    lShareMode  [in]
//
// Output:
//    HRESULT       - S_OK, E_NOINTERFACE
//
// Notes:
//  Dtor must release.
//
HRESULT CMSMQQueue::Init(
    IMSMQQueueInfo *pqinfo, 
    QUEUEHANDLE lHandle,
    long lAccess,
    long lShareMode)
{
    BSTR bstrFormatName = NULL;
    HRESULT hresult = NOERROR;

	// Syncronization: This method is called only when this object is 
	// initialized by it's parent. There should not be more than one caller
	
    m_lHandle = lHandle;
    m_lAccess = lAccess;
    m_lShareMode = lShareMode;

    // Need to copy incoming pqinfo since we need
    //  to snapshot it otherwise it might change
    //  under our feet.
    // We do this by creating a new qinfo and initing
    //  it with the formatname of the incoming qinfo
    //  and then finally refreshing it.
    //
    // m_pqinfo released in dtor
    //
    IfNullRet(m_pqinfo = new CMSMQQueueInfo(NULL));     
    IfFailGo(pqinfo->get_FormatName(&bstrFormatName));
    
    //
    // ownership is not transferred -- the format name is copied.
    // note: cast is safe cos we just created the object
    //  and we know its true type.
    //
    IfFailGoTo(((CMSMQQueueInfo *)m_pqinfo)->Init(bstrFormatName), Error2); 
    SysFreeString(bstrFormatName);      // UNDONE: needs to be freed in a single location
    
    //
    // 2536: only attempt to use the DS when
    //  the first prop is accessed... or Refresh
    //  is called explicitly.
    //
#if 0
    // Might fail if no DS or if DIRECT so we ignore
    //  the error...
    //
    hresult = m_pqinfo->Refresh();
#endif // 0

    IfFailGo(AddQueue(this));              // add to global list
    return Reset();

Error2:
    SysFreeString(bstrFormatName);
    // fall through...

Error:
    RELEASE(m_pqinfo);
    return hresult;
}