Message Queue

Das CXI-Meldungssystem behält sich vor, asynchrone Meldungen sofort synchron zu verarbeiten; bei einigen Systemen (DOS, ANSI) ist dies sogar notwendig, da keine andere Möglichkeit besteht.

Dieses Vorgehen führt zu ernsthaften Problemen (Rekursionen, Deadlocks), wenn Meldungen in geschlossenen Kreisläufen verschickt werden.

Müssen Meldungen aus diesen Gründen zwingend ASYNCHRON verarbeitet werden, so kann das nun vorgestellte Konzept der globalen Message Queue eingesetzt werden.

Das Konzept sieht eine globale Warteschlange vor, die zunächst alle Meldungen entgegennimmt und speichert. Erst wenn explizit die Funktion GlobalWaitState aufgerufen wird, entnimmt das System eine Meldung aus dieser Queue und sendet sie an den eigentlichen Empfänger.

Vorteile:

Meldungen sind asynchron, es erfolgt kein synchroner Aufruf des Handlers, auch nicht unter DOS.

Rekursive Aufrufe der Handler-Funktionen sind ausgeschlossen.

Bei zyklischen Meldungen treten keine Deadlocks auf.

Nachteile:

Da jede Meldung zweimal verschickt wird, treten Performance-Verluste auf.

Die globale Queue benötigt Speicher für jede abgelegte Meldung.

Beim Verschicken einer Meldung wird nicht erkannt, ob der eigentliche Empfänger existiert. Ein Fehlen des Handlers wird erst beim Weiterleiten durch die Queue bemerkt, in diesem Fall erhält der Default-Handler (sofern vorhanden) die Meldung.

Funktionen:
GlobalWaitState - Globale Queue abarbeiten (mit Beispiel)
PostMessageTreeOptions - Globale Queue aktivieren

Portabilität:
Die globale Queue ist ab Version 2.04 definiert und in allen unterstützten Systemen verfügbar, bisher jedoch nur für Borland-Compiler. Um die globale Queue benutzen zu können, muss CXIMSGQ.CPP in das Projekt aufgenommen werden. Jeder Prozess besitzt eine eigene globale Queue.

Siehe auch:
Post, PostMessageTree - Verschicken einer Meldung
Pass - Weiterleiten einer empfangenen Meldung

CXIMSGQ.CPP:

/*#+++ CXIMSGQ.CPP +++#*/ 
/****************************************************************/
/*  CXI - Common Data Storage & Exchange Interface - Version 2  */
/*  Copyright (c) 1996 by TeraCon Verteilte Datensysteme GmbH   */
/****************************************************************/ 
#include <stdio.h>
#include <stdlib.h> 
#include "cxib.h" 
extern const char cximsgqversionstr [] =
  "@$VERSION=CXIMSGQ 2.08 00: BC++, "
  __DATE__ ", " __TIME__;  
static class CxiMsgQ : public CxiMessageHandlerWithQueue, public CxiSysAlloc
{
public:
  CxiMsgQ () : CxiMessageHandlerWithQueue (".@QUEUE", 0,
    cxiOptHandleQueueNoDelay | cxiOptMsgDelayInit)
    { LockQueue (); DisableAutoDispatch (); }
  ~CxiMsgQ ()
    {}
protected:
  long Handler2 (CxiTree& msg, CxiNode& ext);
}
*pCxiMsgQ;  
long CxiMsgQ::Handler2 (CxiTree& msg, CxiNode& ext)
{
  // msg.DumpVarsMessage ("CXIMSGQ.TRC", "Queue::Handler2");
  return Pass (msg, ext.GetItem ("MsgA.To"), cxiOptPostMsgNoForceQueue);
}  
extern "C"
CxiError CXIGLOBALFUNCNX cxiGlobalWaitState (CxiInt all)
{
  if (!pCxiMsgQ)
    return cxiErrorNullPointer;
  cxiCallSystemIdle (0);
  if (all & cxiOptWaitStateAll)
    pCxiMsgQ->ExecuteQueueUntilEmpty ();
  else
    pCxiMsgQ->PollQueue ();
  cxiCallSystemIdle (0);
  return cxiNoError;
}  
static void beginqueue (void)
{
  pCxiMsgQ = new CxiMsgQ;
  if (pCxiMsgQ)
    pCxiMsgQ->EndOfInit ();
}  
static void endqueue (void)
{
  if (pCxiMsgQ)
    pCxiMsgQ->InvalidateObject ();
  delete pCxiMsgQ;
  pCxiMsgQ = 0;
}  
#pragma startup  beginqueue   29
#pragma exit     endqueue     29  
/*#--- CXIMSGQ.CPP ---#*/