Message Dump Example

MessageDump is a useful application that spies on the internal shared memory queue of VR-Exchange and reports what it finds.

It works both as an example and as a tool for debugging custom brokers.

---

/*********************************************************************
 ** Copyright (c) 2007 VT MAK
 *********************************************************************/

#include <portal/smSubscribableMessageQueue.h>

#include <portal/pMessageFactory.h>
#include <portal/memoryManager.h>

#include <portal/pControlMessage.h>
#include <portal/pConnection.h>
#include <portal/pEntity.h>
#include <portal/pGenericInter.h>
#include <portal/pGenericObject.h>
#include <portal/pAggregate.h>
#include <portal/pFireInter.h>
#include <portal/pDetonationInter.h>
#include <portal/brokerErrorMessage.h>
#include <portal/pObject.h>
#include <portal/pAcknowledgeInter.h>         //DtPortalAcknowledgeInteraction
#include <portal/pCollisionInter.h>           //DtPortalCollisionInteraction
#include <portal/pCommentInter.h>             //DtPortalCommentInteraction
#include <portal/pStartInter.h>               //DtPortalStartInteraction
#include <portal/pStopInter.h>                //DtPortalStopInteraction
#include <portal/pRadioSignalInter.h>         //DtPortalRadioSignalInteraction
#include <portal/pRadioTransmitterObject.h>   //DtPortalRadioTransmitterObject
#include <portal/pRadioReceiverObject.h>      //DtPortalRadioReceiverObject
#include <portal/pIffObject.h>                //DtPortalIffObject
#include <portal/pEmitterSystemObject.h>      //DtPortalEmitterSystemObject
#include <portal/pEnvironmentProcessObject.h> //DtPortalEnvironmentProcessObject
#include <portal/pDataInter.h>                //DtPortalDataInteraction
#include <portal/pSetDataInter.h>             //DtPortalSetDataInteraction
#include <portal/pDesignatorObject.h>         //DtPortalDesignatorObject
#include <portal/pCreateEntityInter.h>        //DtPortalCreateEntityInteraction
#include <portal/pRemoveEntityInter.h>        //DtPortalRemoveEntityInteraction
#include <portal/pEventReportInter.h>         //DtPortalEventReportInteraction
#include <portal/pDataQueryInter.h>           //DtPortalDataQueryInteraction
#include <portal/pActionRequestInter.h>       //DtPortalActionRequestInteraction
#include <portal/pActionResponseInter.h>      //DtPortalActionResponseInteraction
#include <portal/pViewControlInter.h>         //DtPortalViewControlInteraction
#include <portal/pGriddedDataObject.h>         //DtPortalGriddedDataObject

#include <vlutil/vlProcessControl.h>
#include <vlutil/vlPrint.h>
#include <vlutil/vlPrintUtil.h>
#include <vlutil/vlExceptions.h>
#include <cmdLine/cmdLine.h>
#include <vlutil/vlInetTcpSocket.h>
#include <namespace.h>

#include <iostream>
#include <assert.h>

DtCmdLine::ValueArg<int>       brokerId("a", "BrokerId", "Broker Id", false, 4, "integer");
DtCmdLine::ValueArg<int>       port("p", "tcpPort", "TCP Port", false, 5111, "integer");
DtCmdLine::ValueArg<DtString>  controlQueueName("", "cqn", "Control Queue Name", false, "CQ", "string");
DtCmdLine::ValueArg<DtString>  interactionQueueName("", "iqn", "Interaction Queue Name", false, "IQ", "string");
DtCmdLine::ValueArg<DtString>  objectQueueName("", "oqn", "Object Queue Name", false, "OQ", "string");
DtCmdLine::SwitchArg           raw("r", "raw", "Display Raw Output", false);

// Used to maintain connectivity
DtInetTcpSocket *portalSocket = 0;

char keybrdTick();
void printBytes(const char *buffer, unsigned int size);
void processCmdLine( int argc, char** argv);
bool isPortalOk();

using namespace MAKVrExchange;

// This is a fairly complex example of a broker which reads all messages
// off the given message queue and prints them out to standard out. 
//
// The example does not use a DtPortalConnection as real brokers should, 
// rather it looks at raw messages by pulling them directly from the queue. 
//
// This example also manages it's end of the TCP connection instead of relying 
// on the more robust code found in DtPortalConnection. 
int main(int argc, char *argv[])
{
   try
   {

      DtInfo << "VR-Exchange Message Dump Utility!" << std::endl;
      processCmdLine(argc, argv);

      DtInfo << "Subscribing as Broker " << brokerId.getValue() << std::endl;

      if (port.getValue() > 0)
      {
         DtClock clk;
         DtInfo << "Attempting to connect to queueManager on port " << port.getValue() 
            << std::endl;
         portalSocket = new DtInetTcpSocket(DtInetEndpoint("127.0.0.1", port, DtInetProtocol_TCP), 0, 
                                            0, DtDefaultSockOpts | DtSockOptNonBlocking);
         portalSocket->setTcpNoDelayOption(true);

         if (!portalSocket->isOk())
         {
            DtInfo 
               << "Broker(" << brokerId.getValue() << ") was unable to estabilish a TCP connection. \n"
               << "This may indicate that the VRX application is not running, or that VRX is \n"
               << "configured to run with out the TCP Connection Manager. Changing the TCP port to\n"
               << "0 will disable this check.\n";
         }

         // Check for up to a second if the TCP Socket is estabilished.
         DtTime timeout = clk.absRealTime() + 3.0;
         while( portalSocket &&
                timeout > clk.absRealTime() &&
                !portalSocket->isOk())
         {
            DtSleep(.01);
         };


         if (!portalSocket || !portalSocket->isOk())
         {

            DtInfo << "Broker(" << brokerId.getValue() << ") was unable to estabilish a TCP connection "
               << "in three seconds. This may indicate a network problem. \n";

            throw DtException("Broker TCP Socket didn't connect in time");
         }
         DtInfo << "Connected to queueManager.\n";
      }
      else
      {
         DtInfo << "Broker(" << brokerId.getValue()
            << ") DtPortalConnection not using TCP socket to maintain connectivity.\n";
      }


      DtPortalMessageFactory fact;
      if (!raw)
      {
         // If we want to print out the actual values for each type of
         // message we will need to register creators
         fact.registerCreator(DtPortalBrokerShutdownRequestMessage());
         fact.registerCreator(DtBrokerExitingMessage());
         fact.registerCreator(DtBrokerStartedMessage());
         fact.registerCreator(DtAbnormalBrokerExitMessage());
         fact.registerCreator(DtBrokerLaunchGuiRequest());            
         fact.registerCreator(DtPortalEntity());
         fact.registerCreator(DtPortalGenericObject());
         fact.registerCreator(DtPortalAggregate());
         fact.registerCreator(DtPortalFireInteraction());
         fact.registerCreator(DtPortalDetonationInteraction());
         fact.registerCreator(DtBrokerErrorMessage());
         fact.registerCreator(DtPortalGenericInteraction());
         fact.registerCreator(DtPortalAcknowledgeInteraction());
         fact.registerCreator(DtPortalCollisionInteraction());
         fact.registerCreator(DtPortalCommentInteraction());
         fact.registerCreator(DtPortalStartInteraction());
         fact.registerCreator(DtPortalStopInteraction());
         fact.registerCreator(DtPortalRadioTransmitterObject());
         fact.registerCreator(DtPortalRadioReceiverObject());
         fact.registerCreator(DtPortalRadioSignalInteraction());
         fact.registerCreator(DtPortalIffObject());
         fact.registerCreator(DtPortalEmitterSystemObject());
         fact.registerCreator(DtPortalEnvironmentProcessObject());         
         fact.registerCreator(DtPortalDataInteraction());
         fact.registerCreator(DtPortalSetDataInteraction());
         fact.registerCreator(DtPortalDesignatorObject());
         fact.registerCreator(DtPortalCreateEntityInteraction());
         fact.registerCreator(DtPortalRemoveEntityInteraction());
         fact.registerCreator(DtPortalEventReportInteraction());
         fact.registerCreator(DtPortalDataQueryInteraction());
         fact.registerCreator(DtPortalActionRequestInteraction());
         fact.registerCreator(DtPortalActionResponseInteraction());
         fact.registerCreator(DtPortalViewControlInteraction());
         fact.registerCreator(DtPortalGriddedDataObject());
      }

      DtSubscribableMessageQueue controlQ(controlQueueName.getValue(), brokerId);
      DtSubscribableMessageQueue interactionQ(interactionQueueName.getValue(), brokerId);
      DtSubscribableMessageQueue objectQ(objectQueueName.getValue(), brokerId);

      DtInfo << "Opened queues " << controlQueueName.getValue() << ", " << interactionQueueName.getValue()
         << ", and " << objectQueueName.getValue() << " for reading.\n";

      int ticksSincePortalTest= 0;
      int maxTicksBeforeTest = 3;

      DtBrokerStartedMessage msg;
      std::vector<char> buffer;
      buffer.resize(msg.netSize());

      msg.writeSelfToBuffer(&buffer[0]);
      controlQ.sendMessage(&buffer[0], msg.netSize());

      bool shutdownLoop =false;
      while(!shutdownLoop)
      {
         if (keybrdTick() == 'q') 
         {
            break;
         }
         if (ticksSincePortalTest++ > maxTicksBeforeTest) 
         {
            if (!isPortalOk())
            {
               DtInfo << "Lost Connection to Portal, perhaps it was shutdown. Exiting\n";
               break;
            }
            else
            {
               ticksSincePortalTest =0;
            }
         }

         DtPortalMessage *ptr;
         // process messages from the portal
         while (!controlQ.isEmpty())
         {
            unsigned int size = 0;
            const char *message = (const char *) controlQ.message(&size);
            ptr = fact.create(message);
            DtInfo << "MESSAGE: " << size << " bytes\n" 
               << "---------------------\n";

            if (ptr)
            {
               DtInfo << *ptr << std::endl;

               if(ptr->kind() == DtBrokerShutdownRequestKind && 
                  ((DtPortalControlMessage*)ptr)->brokerId() < 5)
               {
                  // I'm gonna need to shut down to avoid locking the portal..
                  DtInfo << "Portal is being shutdown. Exiting\n";
                  shutdownLoop = true;
               }

               DtDELETE( ptr );
            }
            else
            {
               printBytes(message, size);
            }
         }

         // Process general interactions
         while (!interactionQ.isEmpty())
         {
            unsigned int size = 0;
            const char *message = (const char *) interactionQ.message(&size);
            ptr = fact.create(message);
            DtInfo << "MESSAGE: " << size << " bytes\n" 
               << "---------------------\n";
            if (ptr)
            {
               DtInfo << *ptr << std::endl;
               DtDELETE( ptr );
            }
            else
            {
               printBytes(message, size);
            }
         }

         // Process general interactions
         while (!objectQ.isEmpty())
         {
            unsigned int size = 0;
            const char *message = (const char *) objectQ.message(&size);
            ptr = fact.create(message);
            DtInfo << "MESSAGE: " << size << " bytes\n" 
               << "---------------------\n";
            if (ptr)
            {
               DtInfo << *ptr << std::endl;
               DtDELETE( ptr );
            }
            else
            {
               printBytes(message, size);
            }
         }
         DtSleep(0.01);
      }

   }
   catch (const DtException &ex)
   {
      DtInfo << "Message Dump received exception: " << ex << std::endl;
   }
   DtDELETE( portalSocket );


   DtInfo << "Message Dump has been shutdown. Press any key to exit. "  << std::endl;
   DtGetCh();
   return 0;
}

void processCmdLine( int argc, char** argv)
{
   DtCmdLine::CmdLine cmd("VR-Exchange Message Dump", ' ', DtPortalConnection::version().c_str());
   cmd.add(raw);
   cmd.add(brokerId);
   cmd.add(port);
   cmd.add(controlQueueName);
   cmd.add(interactionQueueName);
   cmd.add(objectQueueName);
   cmd.parse(argc, argv);
}

void printBytes(const char *buffer, unsigned int size)
{
   for (unsigned int i = 0; i < size; ++i)
   {
      DtInfo("%02x ", (unsigned char)buffer[i]);
      if (i %6 == 0)
      {
         DtInfo ("\n");
      }
   }
   DtInfo("\n");

}

char keybrdTick()
{
   char *keyPtr = DtPollBlockingInputLine();
   if (keyPtr ) return *keyPtr;
   else return 0;
}

bool isPortalOk()
{
   if (portalSocket )
   {
      if ( portalSocket->isOk())
      {
         // Clear our socket out, since the queue manager will be sending "ping" 
         // packets from time to time. If this fills up it will make the queuemanager 
         // thing we have died.
         int receiveSize = 0;
         while(receiveSize > 0)
         {
            DtU32 netPacket;
            receiveSize = portalSocket->recv((char*)(&netPacket), 4);
         }

         // Now we ping the queue manager to make sure its alive. We will do that
         // with our broker ID so that the queue manager knows who we are and
         // can watch to make sure if we die our stuff gets cleaned up. 
         DtNetU32 length = sizeof(DtNetU32) * 2;
         DtNetU32 tstPkt[2];
         tstPkt[0] = length;
         tstPkt[1] = brokerId.getValue();
         portalSocket->send((char *)tstPkt, length);
         return true;
      }
      return false;
   }
   return true;
}