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.
#include <vlutil/vlProcessControl.h>
#include <vlutil/vlInetTcpSocket.h>
#include <vlutil/vlExceptions.h>
#include <vlutil/vlMiniDumper.h>
#include <vlutil/vlPrintUtil.h>
#include <vlutil/vlPrint.h>
#include <cmdLine/cmdLine.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);
DtInetTcpSocket* portalSocket = 0;
using namespace MAKVrExchange;
char keybrdTick();
void printBytes( const char* buffer, unsigned int size );
void processCmdLine( int argc, char* argv[] );
bool isPortalOk();
int main( int argc, char* argv[] )
{
DtINIT_MINIDUMPER( "MessageDump" );
DtNotifyLevel = DtNlVerbose;
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";
}
DtTime timeout = clk.absRealTime() + 3.0;
while ( portalSocket && timeout > clk.absRealTime() && ! portalSocket->isOk() )
{
DtSleep( 0.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";
}
if ( ! raw )
{
}
DtInfo << "Opened queues " << controlQueueName.getValue() << ", " << interactionQueueName.getValue()
<< ", and " << objectQueueName.getValue() << " for reading.\n";
int ticksSincePortalTest= 0;
int maxTicksBeforeTest = 3;
sendMessage( controlQ, msg );
bool shutdownLoop = false;
while ( ! shutdownLoop )
{
if ( keybrdTick() == 'q' )
{
msg.setBrokerId( brokerId );
sendMessage( controlQ, msg );
break;
}
if ( ticksSincePortalTest++ > maxTicksBeforeTest )
{
if ( ! isPortalOk() )
{
DtInfo << "Lost Connection to Portal, perhaps it was shutdown. Exiting\n";
msg.setBrokerId( brokerId );
sendMessage( controlQ, msg );
break;
}
else
{
ticksSincePortalTest = 0;
}
}
while ( ! controlQ.isEmpty() )
{
unsigned int size = 0;
const char* message = (const char*) controlQ.message( &size );
ptr = fact.
create( message, size );
DtInfo << "MESSAGE: " << size << " bytes\n"
<< "---------------------\n";
if ( ptr )
{
DtInfo << *ptr << std::endl;
{
DtInfo << "Portal is being shutdown. Exiting" << std::endl;
msg.setBrokerId( brokerId );
sendMessage( controlQ, msg );
shutdownLoop = true;
}
DtDELETE( ptr );
}
else
{
printBytes( message, size );
}
}
while ( ! interactionQ.isEmpty() )
{
unsigned int size = 0;
const char* message = (const char*) interactionQ.message( &size );
ptr = fact.
create( message, size );
DtInfo << "MESSAGE: " << size << " bytes\n"
<< "---------------------\n";
if ( ptr )
{
DtInfo << *ptr << std::endl;
DtDELETE( ptr );
}
else
{
printBytes( message, size );
}
}
while ( ! objectQ.isEmpty() )
{
unsigned int size = 0;
const char* message = (const char*) objectQ.message( &size );
ptr = fact.
create( message, size );
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[] )
{
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() )
{
int receiveSize = 0;
while ( receiveSize > 0 )
{
DtU32 netPacket;
receiveSize = portalSocket->recv( (char*) &netPacket, 4 );
}
DtNetU32 length = 2*sizeof(DtNetU32);
DtNetU32 tstPkt[2];
tstPkt[0] = length;
tstPkt[1] = brokerId.getValue();
portalSocket->send( (char*) tstPkt, length );
return true;
}
return false;
}
return true;
}
{
unsigned int msgSize = msg.
netSize();
std::vector<char> buffer;
buffer.resize( msgSize );
try
{
}
{
DtWarn << "Error: " << ex.what() << std::endl;
}
}
