rtt: corba_mqueue_test.cpp Source File

00001 /***************************************************************************
00002   tag: The SourceWorks  Tue Sep 7 00:54:57 CEST 2010  corba_mqueue_test.cpp
00003 
00004                         corba_mqueue_test.cpp -  description
00005                            -------------------
00006     begin                : Tue September 07 2010
00007     copyright            : (C) 2010 The SourceWorks
00008     email                : peter@thesourceworks.com
00009 
00010  ***************************************************************************
00011  *                                                                         *
00012  *   This program is free software; you can redistribute it and/or modify  *
00013  *   it under the terms of the GNU General Public License as published by  *
00014  *   the Free Software Foundation; either version 2 of the License, or     *
00015  *   (at your option) any later version.                                   *
00016  *                                                                         *
00017  ***************************************************************************/
00018 
00019 #include "unit.hpp"
00020 
00021 #include "corba_mqueue_test.hpp"
00022 
00023 #include <iostream>
00024 
00025 #include <transports/corba/DataFlowI.h>
00026 #include <rtt/transports/corba/RemotePorts.hpp>
00027 #include <rtt/transports/mqueue/MQLib.hpp>
00028 #include <rtt/transports/corba/CorbaConnPolicy.hpp>
00029 
00030 using namespace std;
00031 using corba::TaskContextProxy;
00032 
00033 void
00034 CorbaMQueueTest::setUp()
00035 {
00036     // connect DataPorts
00037     mr1 = new InputPort<double>("mr");
00038     mw1 = new OutputPort<double>("mw");
00039 
00040     mr2 = new InputPort<double>("mr");
00041     mw2 = new OutputPort<double>("mw");
00042 
00043     tc =  new TaskContext( "localroot" );
00044     tc->ports()->addPort( *mr1 );
00045     tc->ports()->addPort( *mw1 );
00046 
00047     t2 = new TaskContext("localother");
00048     t2->ports()->addPort( *mr2 );
00049     t2->ports()->addPort( *mw2 );
00050 
00051     ts2 = ts = 0;
00052     tp2 = tp = 0;
00053 }
00054 
00055 
00056 void
00057 CorbaMQueueTest::tearDown()
00058 {
00059     delete tp;
00060     delete ts;
00061     delete tp2;
00062     delete ts2;
00063     delete tc;
00064     delete t2;
00065 
00066     delete mr1;
00067     delete mw1;
00068     delete mr2;
00069     delete mw2;
00070 }
00071 
00072 void CorbaMQueueTest::new_data_listener(base::PortInterface* port)
00073 {
00074     signalled_port = port;
00075 }
00076 
00077 
00078 #define ASSERT_PORT_SIGNALLING(code, read_port) \
00079     signalled_port = 0; \
00080     code; \
00081     usleep(100000); \
00082     BOOST_CHECK( read_port == signalled_port );
00083 
00084 void CorbaMQueueTest::testPortDataConnection()
00085 {
00086     // This test assumes that there is a data connection mw1 => mr2
00087     // Check if connection succeeded both ways:
00088     BOOST_CHECK( mw1->connected() );
00089     BOOST_CHECK( mr2->connected() );
00090 
00091     double value = 0;
00092 
00093     // Check if no-data works
00094     BOOST_CHECK( !mr2->read(value) );
00095 
00096     // Check if writing works (including signalling)
00097     ASSERT_PORT_SIGNALLING(mw1->write(1.0), mr2)
00098     BOOST_CHECK( mr2->read(value) );
00099     BOOST_CHECK_EQUAL( 1.0, value );
00100     ASSERT_PORT_SIGNALLING(mw1->write(2.0), mr2);
00101     BOOST_CHECK( mr2->read(value) );
00102     BOOST_CHECK_EQUAL( 2.0, value );
00103 }
00104 
00105 void CorbaMQueueTest::testPortBufferConnection()
00106 {
00107     // This test assumes that there is a buffer connection mw1 => mr2 of size 3
00108     // Check if connection succeeded both ways:
00109     BOOST_CHECK( mw1->connected() );
00110     BOOST_CHECK( mr2->connected() );
00111 
00112     double value = 0;
00113 
00114     // Check if no-data works
00115     BOOST_CHECK( !mr2->read(value) );
00116 
00117     // Check if writing works
00118     ASSERT_PORT_SIGNALLING(mw1->write(1.0), mr2);
00119     ASSERT_PORT_SIGNALLING(mw1->write(2.0), mr2);
00120     ASSERT_PORT_SIGNALLING(mw1->write(3.0), mr2);
00121     // it will be emptied too fast by mqueue.
00122     //ASSERT_PORT_SIGNALLING(mw1->write(4.0), 0);
00123     BOOST_CHECK( mr2->read(value) );
00124     BOOST_CHECK_EQUAL( 1.0, value );
00125     BOOST_CHECK( mr2->read(value) );
00126     BOOST_CHECK_EQUAL( 2.0, value );
00127     BOOST_CHECK( mr2->read(value) );
00128     BOOST_CHECK_EQUAL( 3.0, value );
00129     BOOST_CHECK_EQUAL( mr2->read(value), OldData );
00130 }
00131 
00132 void CorbaMQueueTest::testPortDisconnected()
00133 {
00134     BOOST_CHECK( !mw1->connected() );
00135     BOOST_CHECK( !mr2->connected() );
00136 }
00137 
00138 
00139 // Registers the fixture into the 'registry'
00140 BOOST_FIXTURE_TEST_SUITE(  CorbaMQueueTestSuite,  CorbaMQueueTest )
00141 
00142 
00143 BOOST_AUTO_TEST_CASE( testPortConnections )
00144 {
00145     // This test tests the differen port-to-port connections.
00146     ts  = corba::TaskContextServer::Create( tc, false ); //no-naming
00147     ts2 = corba::TaskContextServer::Create( t2, false ); //no-naming
00148 
00149     // Create a default CORBA policy specification
00150     RTT::corba::CConnPolicy policy = toCORBA( RTT::ConnPolicy() );
00151     policy.type = RTT::corba::CData;
00152     policy.init = false;
00153     policy.lock_policy = RTT::corba::CLockFree;
00154     policy.size = 0;
00155     policy.data_size = 0;
00156 
00157     // Set up an event handler to check if signalling works properly as well
00158     Handle hl( mr2->getNewDataOnPortEvent()->setup(
00159                 boost::bind(&CorbaMQueueTest::new_data_listener, this, _1) ) );
00160     hl.connect();
00161 
00162     corba::CDataFlowInterface_var ports  = ts->server()->ports();
00163     corba::CDataFlowInterface_var ports2 = ts2->server()->ports();
00164 
00165     // WARNING: in the following, there is four configuration tested. There is
00166     // also three different ways to disconnect. We need to test those three
00167     // "disconnection methods", so beware when you change something ...
00168 
00169     policy.type = RTT::corba::CData;
00170     policy.pull = false;
00171     policy.transport = ORO_MQUEUE_PROTOCOL_ID;
00172     BOOST_CHECK( ports->createConnection("mw", ports2, "mr", policy) );
00173     usleep(100000); // gives dispatcher time to catch up.
00174     testPortDataConnection();
00175     ports->disconnectPort("mw");
00176     testPortDisconnected();
00177 
00178     policy.type = RTT::corba::CData;
00179     policy.pull = true;
00180     policy.transport = ORO_MQUEUE_PROTOCOL_ID;
00181     BOOST_CHECK( ports->createConnection("mw", ports2, "mr", policy) );
00182     usleep(100000);
00183     testPortDataConnection();
00184     ports2->disconnectPort("mr");
00185     testPortDisconnected();
00186 
00187 #if 1
00188 
00189     policy.type = RTT::corba::CBuffer;
00190     policy.pull = false;
00191     policy.size = 3;
00192     policy.transport = ORO_MQUEUE_PROTOCOL_ID;
00193     BOOST_CHECK( ports->createConnection("mw", ports2, "mr", policy) );
00194     testPortBufferConnection();
00195     ports->disconnectPort("mw");
00196     testPortDisconnected();
00197 
00198     policy.type = RTT::corba::CBuffer;
00199     policy.pull = true;
00200     policy.size = 3;
00201     policy.transport = ORO_MQUEUE_PROTOCOL_ID;
00202     BOOST_CHECK( ports->createConnection("mw", ports2, "mr", policy) );
00203     testPortBufferConnection();
00204     ports->disconnectPort("mw");
00205     testPortDisconnected();
00206 #endif
00207 }
00208 
00209 BOOST_AUTO_TEST_SUITE_END()
00210