Merged revisions 439476-447993,447995-448007,448009-448141,448143-448157,448161-448194,448196-448210,448212-448218,448220-448223,448225-448233,448235,448237-448241,448243-448596,448598-448623,448625-448850,448852-448880,448882-448982,448984-449635,449637-449639,449641-449642,449644-449645,449647-449674,449676-449719,449721-449749,449751-449762,449764-449933,449935-449941,449943-450383,450385,450387-450400,450402-450433,450435-450503,450505-450555,450557-450860,450862-451024,451026-451149,451151-451316,451318-451931,451933-452139,452141-452162,452164-452320,452322,452324-452325,452327-452333,452335-452429,452431-452528,452530-452545,452547-453192,453194-453195,453197-453536,453538,453540-453656,453658-454676,454678-454735,454737,454739-454781,454783-462728,462730-462819,462821-462833,462835-462839,462841-463071,463073-463178,463180-463308,463310-463362,463364-463375,463377-463396,463398-463402,463404-463409,463411-463661,463663-463670,463672-463673,463675-464493,464495-464502,464504-464576,464578-464613,464615-464628,464630,464632-464866,464868-464899,464901-464942,464944-464949,464951-465004,465006-465016,465018-465053,465055-465165,465167-465321,465323-465406,465408-465427,465429-465431,465433-465548,465550-466044,466047-466075,466077,466079-466081,466083-466099,466101-466112,466114-466126,466128-466240,466242-466971,466973-466978,466980-467309,467311-467312,467316-467328,467330-467485,467487-467588,467590-467604,467606-467699,467701-467706,467708-467749,467751-468069,468071-468537,468539-469241,469244-469246,469248-469318,469320-469421,469423,469425-469429,469431-469435,469437-469462,469464-469469,469472-469477,469479-469490,469492-469503,469505-469529,469531-469598,469600-469624,469626-469737,469739-469752,469754-469806,469808-469928,469930-469953,469955-470011,470013-470109,470111-470335,470338-470339,470341-470379,470381,470383-470399,470401-470446,470448-470741,470743-470758,470760-470809,470811-470817,470819-470993,470995-471001,471003-471788,471790-471792,471794-472028,472030-472032,472034-472036,472038,472040,472043,472045-472059,472061,472063,472065-472066,472068,472070-472072,472074-472080,472082,472084-472092,472094-472107,472109-472123,472125-472158,472160-472165,472167-472172,472174-472457,472459-472460,472462-472464,472466-472470,472472-472483,472486-472491,472493-472494,472496-472497,472499,472501-472503,472505-472512,472514-472544,472546-472556,472558-472560,472562-472572,472574-472587,472589-472591,472593-472605,472607,472609-472731,472733-472786,472788-472843,472845-472849,472851-472859,472861-472878,472880-472903,472905,472907-472988,472990-472991,472993-473071,473073-473086,473088-473090,473093,473095-473096,473098-473106,473108-473110,473112-473185,473187-473260,473262,473268-473270,473275-473279,473281,473284-473287,473289-473295,473297-473306,473308-473330,473332-473335,473337,473339-473344,473346-473351,473353-473355,473357-473358,473361-473471,473473-473497,473499-473535,473537-473567,473569-473888,473890-474451,474454-474492,474494-474563,474565-474843,474845-474865,474867-474932,474934-475035,475037-475144,475146-475180,475182-475265,475267-475285,475287,475289-475293,475295-475296,475298-475302,475304-475631,475633-475649,475651-475748,475750-475752,475754-476107,476109-476302,476304-476413,476415-476430,476432-476700,476702-476868,476870-477147,477149-477213,477215-477263,477265-477340,477342-477635,477637-477789,477791-477825,477827-477841,477843,477846-477852,477854,477856,477858-477865,477867-477894,477896-478022,478024-478182,478184-478211,478213-478233,478235-478236,478238-478241,478243-478252,478254-478259,478261-478263,478265,478267-478269,478271-478286,478288-478342,478344-478379,478381-478412,478414-478443,478445-478636,478639-478658,478660-478821,478823-478853,478855-478922,478924-478962,478965-478974,478976-479029,479031-479049,479051-479210,479212-479214,479216-479407,479409-479415,479417-479425,479427-479559,479561-479639,479641-479676,479678-479685,479687-480030,480033-480086,480091-480093,480095-480118,480120-480139,480141,480143-480148,480150-480156,480158-480163,480165-480177,480179-480189,480191-480193,480195-480198,480200-480220,480222-480282,480284-480292,480294-480308,480310-480317,480320-480422,480424,480426-480581,480583-480656,480658-480692,480695-480702,480704,480706-480710,480712-480910,480913-480933,480935-480945,480947-480972,480974-480993,480995-481034,481036-481158,481161-481174,481176-481220,481222-481234,481236-481260,481263-481264,481266-481296,481298-481304,481306-481311,481313-481332,481334,481336-481380,481382-481441,481443-482144,482146-482180,482182-482193,482195-482232,482234-482236,482239,482241-482242,482244-482247,482250-482251,482253,482256-482261,482264-482288,482290-482364,482366,482368,482370-482554,482556,482558-482569,482572-482636,482638,482640-482696,482698-482722,482724-482732,482734-482771,482774-482957,482959-483045,483047-483105,483108,483110-483115,483117,483119-483127,483130-483134,483136-483148,483150-483158,483160-483164,483166-483178,483180-483391,483393-483400,483402-483403,483405-483418,483420-483421,483425-483436,483438-483470,483472-483502,483504-483558,483560-483599,483601-483637,483639-483644,483646-483659,483661-483670,483672-483878,483880-483910,483912-483915,483917-483940,483942,483944-483968,483970-483972,483974-483976,483978,483980-484612,484614-484657,484659-484693,484695-484718,484720-484842,484844-484847,484849-484986,484988-485019,485021-485489,485491-485544,485546-485591,485593,485595-485697,485699-485729,485731-485734,485736-485779,485781-485787,485789-485851,485853,485855-486007,486009,486011-486020,486022-486083,486085-486097,486099-486117,486120-486131,486133-486148,486150-486161,486163-486164,486166-486197,486199-486205,486208-486247,486249-486253,486256-486427,486429-486431,486433-486554,486556-486573,486575-486593,486595,486597-486609,486611-486619,486622,486625,486627-486641,486643-486645,486649-486687,486689-486721,486723-486730,486732-486746,486748-486759,486761,486763-486777,486779-486782,486784-486788,486790,486792,486794-486796,486798-487175,487178,487180-487213,487215,487217-487267,487269-487284,487286-487298,487300-487358,487360-487367,487369-487382,487384-487434,487436-487480,487482-487547,487549-487561,487563-487565,487567-487578,487580-487615,487617-487622,487624,487626,487628,487630-487635,487637-487703,487705-487777,487780-487781,487783-487800,487802-487803,487805-487820,487822-487848,487850-487902,487904-488103,488105-488133,488135-488158,488160-488163,488165-488187,488189-488216,488218-488248,488250-488278,488280,488282-488303,488305-488313,488315-488342,488344-488351,488353-488376,488378-488449,488451-488593,488595,488597-488623,488625-488700,488702-488704,488706-488710,488714,488716-488725,488727-488744,488746-488770,488772-488798,488800,488802-488807,488809,488811-488829,488831-488843,488845-488851,488853-489069,489071-489077,489079-489081,489084-489102,489104-489105,489107-489109,489111-489112,489114-489139,489141-489178,489181-489203,489205-489211,489213,489216-489329,489332-489402,489404-489417,489419-489421,489423-489643,489645-489690,489692-489703,489705-489714,489716-489747,489749-489753,489755-489803,489805-489904,489906-490372,490374-490504,490506-490604,490606-490707,490710-490733,490735-490871,490873-490984,490986-491028,491030,491032-491071,491073-491119,491121-491576,491578-491672,491674-491800,491802-491838,491840-491878,491880-492183,492185-492279,492281-492317,492319-492513,492515-492584,492586-492587,492589-492601,492603-492635,492637-492640,492642-492717,492719-492723,492725-492729,492731-492755,492757-492901,492903-492955,492957-492962,492964-492997,492999-493002,493004-493041,493043-493059,493062-493063,493065-493086,493088-493125,493127-493139,493141-493150,493152-493871,493873-494017,494019-494030,494032-494041,494043-494091,494093-494120,494122-494354,494356-494436,494438-494539,494541-494552,494554-494586,494588-494649,494651,494653-494654,494656-494657,494659-494764,494766-494768,494770-494796,494798-494799,494802,494804-494860,494862-494903,494905-494906,494908-495019,495021-495160,495162-495168,495171-495188,495190-495229,495231-495254,495256-495303,495305-495313,495315-495336,495338-495372,495374-495379,495381-495454,495457-495459,495462-495516,495518-495524,495526-495531,495533-495548,495551-495553,495555,495557-495558,495560,495562-495573,495575-495583,495585-495594,495596-495628,495630-495638,495640-495651,495653-495660,495662-495753,495755-496259,496261-496262,496264-496269,496271-496275,496277-496301,496303-496316,496318-496383,496385-496413,496415-496495,496497-496625,496627-496636,496638-496640,496642-496647,496650-496657,496659-496660,496663-496664,496666-496677,496679-496681,496683-496730,496732-496750,496752,496754-496784,496786-496832,496834-496840,496842-496990,496992-496995,496997-497340,497343-497351,497353-497403,497405-497424,497426-497438,497440-497481,497483-497497,497499-497765,497767-497769,497771-497775,497777-497778,497780,497782-497783,497785,497787-497812,497814-497871,497873-497877,497879-498573,498575-498588,498590,498592,498594-498636,498638-498669,498671-498686,498688-498689,498691-498719,498721-498964,498966-498969,498971-498973,498975-498982,498985-499035,499037-499040,499042,499044-499048,499050-499082,499084-499086,499088-499164,499167-499169,499171-499355,499357-499370,499372-499373,499375-499391,499393,499395-499425,499428,499430-499445,499447-499455,499457-499460,499462-499465,499467,499469-499489,499491-499492,499494-499531,499533-499562,499566-499627,499629-499715,499717-499732,499734-499755,499758-499763,499765-499780,499782-499795,499797-499802,499804-499844,499846,499848-499850,499852-499863,499865-499873,499875-499974,499976-499978,499980-500263,500265-500283,500285-500309,500311-501000,501002,501012-501057,501059-501095,501097-501390,501392-501410,501413-501447,501449-501454,501456,501458-501464,501466-501471,501473-501803,501805-501913,501915-501916,501918-501919,501921-501944,501946-502171,502173-502177,502181,502183-502247,502250-502252,502254-502260,502262-502267,502270,502272,502274-502575,502577-502609,502611-502619,502621-502626,502628-502654,502656-503592,503594-503603,503605-503608,503610-503636,503638-503645,503647-503705,503707-503789,503791-504024,504026-504111,504113-504506,504508-504735,504737-504863,504865-504867,504869-504914,504916-505241,505243-505254,505257-505267,505269-505354,505356-505891,505893-505971,505973-506400,506402-506404,506407-506438,506440-506516,506518-506541,506543-506966,506968-506971,506973-507095,507097-507108,507111-507454,507456,507459-507471,507473-507556,507558,507560-507581,507585-507594,507597,507599-507608,507610-507728,507730-507893,507895-507937,507940-508234,508236-508350,508352-508365,508367-508380,508383,508386-508415,508417-508648,508650-508941,508943-509146,509148-509171,509173-509175,509179-509201,509203-509207,509209-509215,509217-509222,509224-509477,509480-509627,509629-509634,509636-509641,509643-509736,509738-509931,509933-510059,510061-510075,510077-510158,510161-510896,510898-510938,510940-511388,511390-511922,511924-512287,512289-512698,512702-512813,512815-512817,512819-513359,513361-513370,513372-514702,514704-514886,514888-514902,514904-515126,515129-515141,515143-515516,515518-515534,515536-515538,515540-515648,515650-515651,515653-516070,516072-516411,516413-516448,516450,516452-517637,517639-517647,517649-517659,517661-517663,517665-517677,517679-517682,517684-517744,517746-518085,518087-518175,518177-518558,518560-518568,518571-518666,518668,518670-518699,518701-518987,518990-518992,518994-519908,519910-519932,519934-520414,520416-520842,520844-520937,520939-521362,521364-521681,521683-521704,521706-521709,521711-521714,521716-521781,521783-521792,521794-522462,522464-522527,522529-522534,522536-522566,522568-522958,522960,522962-522966,522968-522976,522978-522980,522982-522988,522992-522993,522995-523244,523246-523746,523748-524049,524051-524738,524741-524742,524744-524762,524764,524766,524768-525486,525488-525530,525532,525534,525537-525552,525554-525765,525767-525776,525778-525784,525789-525803,525805-525816,525818-525828,525830-525861,525863-525866,525868-526090,526092-526112,526114-526116,526119-526121,526123-526149,526151-526153,526155-526156,526160-526165,526167-526186,526188-526193,526196-526197,526200-526665,526667-526682,526686-526690,526693,526695-526708,526710-526713,526715-526775,526777-526802,526804-526806,526808-527048,527051-527052,527054-527181,527183-527486,527488-527492,527494-527498,527500-527508,527510-527517,527519-527536,527538-527555,527559-527802,527804-527842,527844-527847,527849-527875,527877-527940,527942-527958,527960-527971,527973-528002,528004,528006-528423,528425-529232,529234-529245,529247-529296,529298-529634,529636-529658,529660-529665,529667-529668,529670-530033,530035-530036,530038-530040,530045-530046,530050-530051,530053-530431,530433-530436,530439-530440,530443,530445-530446,530448,530450-530682,530684,530687-530696,530698-530733,530735-530776,530778-530795,530799,530801-530811,530813-530818,530820-530837,530839-531436,531438-531455,531457,531459-531511,531514,531516,531519-531523,531525,531528-531858,531860-531864,531866-531907,531909-531916,531918-531936,531938-531988,531990-532001,532003-532371,532373-532465,532467-532727,532729-532765,532767-532785,532788-532790,532792-532793,532795-533064,533066-533074,533076,533080-533130,533132-533139,533142-533703,533705-533720,533722-533763,533766-533818,533820-533839,533841-533859,533862-534035,534037-534112,534114-534116,534118-534472,534474-534477,534479-534762,534764-534896,534898-534902,534904-535253,535255-535308,535310-535808,535810-535873,535875-536007,536009-536140,536142-536162,536165-536242,536244-536252,536254-536278,536280-536338,536340-536448,536450-536479,536481-536482,536484-536485,536487-536495,536497,536500-536505,536507-536561,536563-536570,536572,536574-536583,536586-536823,536825-537014,537016-537018,537020-537025,537027-537028,537030-537160,537162-537170,537172-537672,537674-537781,537783-537833,537836-537840,537842-537844,537846-537953,537955-538034,538036-538078,538080-538083,538085-538097,538099-538108,538110-538239,538241-538881,538883-538906,538908-538911,538913-538921,538923-539177,539179-539190,539192-539475,539477-539500,539502-539593,539595-539782,539784-539787,539789-540106,540108-540168,540170-540510,540512-541919,541921-544507,544509-544865,544867-545145,545147-547177,547179-547627 via svnmerge from

https://svn.apache.org/repos/asf/incubator/qpid/branches/M2

........
  r539470 | ritchiem | 2007-05-18 14:50:59 +0100 (Fri, 18 May 2007) | 1 line
  
  QPID-401 Integrated python tests with maven tested on windows CMD.exe and linux FC5
........
  r539481 | ritchiem | 2007-05-18 15:30:06 +0100 (Fri, 18 May 2007) | 1 line
  
  QPID-401 Update to allow -Dskip-python-tests to disable python checking
........
  r539484 | ritchiem | 2007-05-18 15:35:13 +0100 (Fri, 18 May 2007) | 1 line
  
  QPID-401 Update to allow -Dskip-python-tests to disable python checking
........
  r541247 | rgodfrey | 2007-05-24 10:57:00 +0100 (Thu, 24 May 2007) | 1 line
  
  QPID-482 : Small performance tweaks
........
  r542484 | rupertlssmith | 2007-05-29 11:52:29 +0100 (Tue, 29 May 2007) | 1 line
  
  Can now pass property to skip python tests, set in settings.xml
........
  r542789 | ritchiem | 2007-05-30 11:09:28 +0100 (Wed, 30 May 2007) | 1 line
  
  Update to ensure fastinstall profile skips the broker python tests.
........
  r543496 | rupertlssmith | 2007-06-01 15:33:07 +0100 (Fri, 01 Jun 2007) | 1 line
  
  QPID-402: FailoverException falling through to client. All blocking operations now wrapped in failover support wrappers.
........
  r544109 | ritchiem | 2007-06-04 10:47:53 +0100 (Mon, 04 Jun 2007) | 7 lines
  
  Addition of a sustained test client. This is currently setup for running a pub/sub test. 
  
  The test allows for multiple clients to connect and participate in testing the broker throughput.
  
  A single producer continually sends messages to a topic which the clients then send batched results back about.
  
  The producer uses the timings in the reports to update the rate at which it sends messages. Ideally reaching a steady state where all messages produced are received by everyone within a specified time frame.
........
  r544422 | ritchiem | 2007-06-05 09:50:54 +0100 (Tue, 05 Jun 2007) | 1 line
  
  Added documentation on how to run the sustained tests.
........
  r546096 | rupertlssmith | 2007-06-11 12:23:08 +0100 (Mon, 11 Jun 2007) | 1 line
  
  Set up top dir location and path to parent pom. Needed in preparation for deploy and release plugins.
........
  r546190 | rupertlssmith | 2007-06-11 17:43:57 +0100 (Mon, 11 Jun 2007) | 1 line
  
  Removed log4j dependency from client. Using slf4j instead, end-user to supply logging implementation as desired. Log4j used for tests.
........
  r546441 | rupertlssmith | 2007-06-12 10:52:29 +0100 (Tue, 12 Jun 2007) | 1 line
  
  QPID-465, now throws UnsupportedOperationException when sending to a null queue in QueueSender.
........
  r546458 | ritchiem | 2007-06-12 12:41:17 +0100 (Tue, 12 Jun 2007) | 1 line
  
  Added repository info for running mvn rat:check
........
  r547627 | ritchiem | 2007-06-15 12:21:40 +0100 (Fri, 15 Jun 2007) | 1 line
  
  QPID-511 adjusted to use the ReadWriteThreadModel rather than setting editing the filterChain directly which could cause problems when using an InVM transport due to the way the InVM transport alters the filter chain during a connect call.
........


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@547730 13f79535-47bb-0310-9956-ffa450edef68

13 files changed, 1405 insertions, 89 deletions

diff --git a/java/integrationtests/docs/RunningSustainedTests.txt b/java/integrationtests/docs/RunningSustainedTests.txt
new file mode 100644
index 0000000000..2b37f4c5a7
--- /dev/null
+++ b/java/integrationtests/docs/RunningSustainedTests.txt
@@ -0,0 +1,15 @@
+In addition to the integration tests the framework provided by this package also allows for
+sustained tests to be run. Currently avaible tests:
+- org.apache.qpid.sustained.SustainedTestClient : Pub Sub test to determine steady state throughput.
+
+
+Running Tests.
+
+Run the tests as per the integration tests.
+ - Start a broker
+ - Start at least one Client [java org.apache.qpid.sustained.TestClient], ensuring unique naming
+ - Start Test Controller [java org.apache.qpid.sustained.TestCoordinator]
+ - Additional Test clients can be started:
+    [java org.apache.qpid.sustained.TestClient -j org.apache.qpid.sustained.SustainedTestClient]
+
+
diff --git a/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/CoordinatingTestCase.java b/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/CoordinatingTestCase.java
index 3003c00ca5..31de84e630 100644
--- a/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/CoordinatingTestCase.java
+++ b/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/CoordinatingTestCase.java
@@ -21,10 +21,7 @@
  */

 package org.apache.qpid.interop.coordinator;

 

-import java.util.Collection;

 import java.util.Map;

-import java.util.Properties;

-import java.util.concurrent.LinkedBlockingQueue;

 

 import javax.jms.*;

 

@@ -74,13 +71,13 @@ public abstract class CoordinatingTestCase extends TestCase
     private static final Logger log = Logger.getLogger(CoordinatingTestCase.class);

 

     /** Holds the contact details for the sending test client. */

-    TestClientDetails sender;

+    protected TestClientDetails sender;

 

     /** Holds the contact details for the receving test client. */

-    TestClientDetails receiver;

+    protected TestClientDetails receiver;

 

     /** Holds the conversation factory over which to coordinate the test. */

-    ConversationFactory conversationFactory;

+    protected ConversationFactory conversationFactory;

 

     /**

      * Creates a new coordinating test case with the specified name.

diff --git a/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/Coordinator.java b/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/Coordinator.java
index 7c5079327e..6eec20769a 100644
--- a/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/Coordinator.java
+++ b/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/Coordinator.java
@@ -60,28 +60,28 @@ public class Coordinator extends TKTestRunner
     public static final String DEFAULT_CONNECTION_PROPS_RESOURCE = "org/apache/qpid/interop/connection.properties";

 

     /** Holds the URL of the broker to coordinate the tests on. */

-    String brokerUrl;

+    protected String brokerUrl;

 

     /** Holds the virtual host to coordinate the tests on. If <tt>null</tt>, then the default virtual host is used. */

-    String virtualHost;

+    protected String virtualHost;

 

     /** Holds the list of all clients that enlisted, when the compulsory invite was issued. */

-    Set<TestClientDetails> enlistedClients = new HashSet<TestClientDetails>();

+    protected Set<TestClientDetails> enlistedClients = new HashSet<TestClientDetails>();

 

     /** Holds the conversation helper for the control conversation. */

-    private ConversationFactory conversationFactory;

+    protected ConversationFactory conversationFactory;

 

     /** Holds the connection that the coordinating messages are sent over. */

-    private Connection connection;

+    protected Connection connection;

 

     /**

      * Holds the name of the class of the test currently being run. Ideally passed into the {@link #createTestResult}

      * method, but as the signature is already fixed for this, the current value gets pushed here as a member variable.

      */

-    private String currentTestClassName;

+    protected String currentTestClassName;

 

     /** Holds the path of the directory to output test results too, if one is defined. */

-    private static String reportDir;

+    protected static String _reportDir;

 

     /**

      * Creates an interop test coordinator on the specified broker and virtual host.

@@ -89,7 +89,7 @@ public class Coordinator extends TKTestRunner
      * @param brokerUrl   The URL of the broker to connect to.

      * @param virtualHost The virtual host to run all tests on. Optional, may be <tt>null</tt>.

      */

-    Coordinator(String brokerUrl, String virtualHost)

+    public Coordinator(String brokerUrl, String virtualHost)

     {

         log.debug("Coordinator(String brokerUrl = " + brokerUrl + ", String virtualHost = " + virtualHost + "): called");

 

@@ -116,38 +116,36 @@ public class Coordinator extends TKTestRunner
             // Use the command line parser to evaluate the command line with standard handling behaviour (print errors

             // and usage then exit if there are errors).

             Properties options =

-                CommandLineParser.processCommandLine(args,

-                    new CommandLineParser(

-                        new String[][]

-                        {

-                            { "b", "The broker URL.", "broker", "false" },

-                            { "h", "The virtual host to use.", "virtual host", "false" },

-                            { "o", "The name of the directory to output test timings to.", "dir", "false" }

-                        }));

+                    CommandLineParser.processCommandLine(args,

+                                                         new CommandLineParser(

+                                                                 new String[][]

+                                                                         {

+                                                                                 {"b", "The broker URL.", "broker", "false"},

+                                                                                 {"h", "The virtual host to use.", "virtual host", "false"},

+                                                                                 {"o", "The name of the directory to output test timings to.", "dir", "false"}

+                                                                         }));

 

             // Extract the command line options.

             String brokerUrl = options.getProperty("b");

             String virtualHost = options.getProperty("h");

-            reportDir = options.getProperty("o");

-            reportDir = (reportDir == null) ? "." : reportDir;

+            _reportDir = options.getProperty("o");

+            _reportDir = (_reportDir == null) ? "." : _reportDir;

 

             // Scan for available test cases using a classpath scanner.

             Collection<Class<? extends CoordinatingTestCase>> testCaseClasses =

-                new ArrayList<Class<? extends CoordinatingTestCase>>();

+                    new ArrayList<Class<? extends CoordinatingTestCase>>();

             // ClasspathScanner.getMatches(CoordinatingTestCase.class, "^Test.*", true);

             // Hard code the test classes till the classpath scanner is fixed.

             Collections.addAll(testCaseClasses,

-                new Class[]

-                {

-                    CoordinatingTestCase1DummyRun.class, CoordinatingTestCase2BasicP2P.class,

-                    CoordinatingTestCase3BasicPubSub.class

-                });

+                               CoordinatingTestCase1DummyRun.class,

+                               CoordinatingTestCase2BasicP2P.class,

+                               CoordinatingTestCase3BasicPubSub.class);

 

             // Check that some test classes were actually found.

-            if ((testCaseClasses == null) || testCaseClasses.isEmpty())

+            if (testCaseClasses.isEmpty())

             {

                 throw new RuntimeException(

-                    "No test classes implementing CoordinatingTestCase were found on the class path.");

+                        "No test classes implementing CoordinatingTestCase were found on the class path.");

             }

 

             int i = 0;

@@ -194,7 +192,7 @@ public class Coordinator extends TKTestRunner
     public TestResult start(String[] testClassNames) throws Exception

     {

         log.debug("public TestResult start(String[] testClassNames = " + PrettyPrintingUtils.printArray(testClassNames)

-            + ": called");

+                  + ": called");

 

         // Connect to the broker.

         connection = TestClient.createConnection(DEFAULT_CONNECTION_PROPS_RESOURCE, brokerUrl, virtualHost);

@@ -228,7 +226,7 @@ public class Coordinator extends TKTestRunner
             // Record the current test class, so that the test results can be output to a file incorporating this name.

             this.currentTestClassName = testClassName;

 

-            result = super.start(new String[] { testClassName });

+            result = super.start(new String[]{testClassName});

         }

 

         // At this point in time, all tests have completed. Broadcast the shutdown message.

@@ -252,7 +250,7 @@ public class Coordinator extends TKTestRunner
     public static Set<TestClientDetails> extractEnlists(Collection<Message> enlists) throws JMSException

     {

         log.debug("public static Set<TestClientDetails> extractEnlists(Collection<Message> enlists = " + enlists

-            + "): called");

+                  + "): called");

 

         Set<TestClientDetails> enlistedClients = new HashSet<TestClientDetails>();

 

@@ -310,9 +308,9 @@ public class Coordinator extends TKTestRunner
             targetTest = new WrappedSuiteTestDecorator(suite);

             log.debug("Wrapped with a WrappedSuiteTestDecorator.");

         }

-

         // Wrap the tests in an inviting test decorator, to perform the invite/test cycle.

-        targetTest = new InvitingTestDecorator(targetTest, enlistedClients, conversationFactory, connection);

+

+        targetTest = newTestDecorator(targetTest, enlistedClients, conversationFactory, connection);

 

         TestSuite suite = new TestSuite();

         suite.addTest(targetTest);

@@ -323,6 +321,11 @@ public class Coordinator extends TKTestRunner
         return super.doRun(suite, wait);

     }

 

+    protected WrappedSuiteTestDecorator newTestDecorator(WrappedSuiteTestDecorator targetTest, Set<TestClientDetails> enlistedClients, ConversationFactory conversationFactory, Connection connection)

+    {

+        return new InvitingTestDecorator(targetTest, enlistedClients, conversationFactory, connection);

+    }

+

     /**

      * Creates the TestResult object to be used for test runs.

      *

@@ -335,10 +338,10 @@ public class Coordinator extends TKTestRunner
         TKTestResult result = new TKTestResult(fPrinter.getWriter(), delay, verbose, testCaseName);

 

         // Check if a directory to output reports to has been specified and attach test listeners if so.

-        if (reportDir != null)

+        if (_reportDir != null)

         {

             // Create the report directory if it does not already exist.

-            File reportDirFile = new File(reportDir);

+            File reportDirFile = new File(_reportDir);

 

             if (!reportDirFile.exists())

             {

@@ -377,4 +380,9 @@ public class Coordinator extends TKTestRunner
 

         return result;

     }

+

+    public void setReportDir(String reportDir)

+    {

+        _reportDir = reportDir;

+    }

 }

diff --git a/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/InvitingTestDecorator.java b/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/InvitingTestDecorator.java
index 858ed1a589..8695f7f66f 100644
--- a/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/InvitingTestDecorator.java
+++ b/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/InvitingTestDecorator.java
@@ -21,7 +21,6 @@
 package org.apache.qpid.interop.coordinator;

 

 import java.util.*;

-import java.util.concurrent.LinkedBlockingQueue;

 

 import javax.jms.Connection;

 import javax.jms.Destination;

@@ -30,7 +29,6 @@ import javax.jms.Message;
 

 import junit.framework.Test;

 import junit.framework.TestResult;

-import junit.framework.TestSuite;

 

 import org.apache.log4j.Logger;

 

@@ -107,7 +105,7 @@ public class InvitingTestDecorator extends WrappedSuiteTestDecorator
             CoordinatingTestCase coordTest = (CoordinatingTestCase) test;

 

             // Broadcast the invitation to find out what clients are available to test.

-            Set<TestClientDetails> enlists = null;

+            Set<TestClientDetails> enlists;

             try

             {

                 Message invite = conversationFactory.getSession().createMessage();

diff --git a/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/ListeningCoordinatorTest.java b/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/ListeningCoordinatorTest.java
new file mode 100644
index 0000000000..1b4461f8c2
--- /dev/null
+++ b/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/ListeningCoordinatorTest.java
@@ -0,0 +1,28 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one
+ *  or more contributor license agreements.  See the NOTICE file
+ *  distributed with this work for additional information
+ *  regarding copyright ownership.  The ASF licenses this file
+ *  to you under the Apache License, Version 2.0 (the
+ *  "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing,
+ *  software distributed under the License is distributed on an
+ *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ *  KIND, either express or implied.  See the License for the
+ *  specific language governing permissions and limitations
+ *  under the License.    
+ *
+ * 
+ */
+package org.apache.qpid.interop.coordinator;
+
+import javax.jms.Message;
+
+public interface ListeningCoordinatorTest
+{
+    public void latejoin(Message message);
+}
diff --git a/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/ListeningTestDecorator.java b/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/ListeningTestDecorator.java
new file mode 100644
index 0000000000..4312dfbcc6
--- /dev/null
+++ b/java/integrationtests/src/main/java/org/apache/qpid/interop/coordinator/ListeningTestDecorator.java
@@ -0,0 +1,200 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.interop.coordinator;
+
+import junit.framework.Test;
+import junit.framework.TestResult;
+import org.apache.log4j.Logger;
+import org.apache.qpid.util.ConversationFactory;
+import uk.co.thebadgerset.junit.extensions.WrappedSuiteTestDecorator;
+
+import javax.jms.Connection;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageListener;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Set;
+
+/**
+ * <p><table id="crc"><caption>CRC Card</caption> <tr><th> Responsibilities <th> Collaborations <tr><td> Broadcast test
+ * invitations and collect enlists. <td> {@link ConversationFactory}. <tr><td> Output test failures for clients
+ * unwilling to run the test case. <td> {@link Coordinator} <tr><td> Execute coordinated test cases. <td> {@link
+ * CoordinatingTestCase} </table>
+ */
+public class ListeningTestDecorator extends WrappedSuiteTestDecorator implements MessageListener
+{
+    private static final Logger log = Logger.getLogger(ListeningTestDecorator.class);
+
+    /** Holds the contact information for all test clients that are available and that may take part in the test. */
+    Set<TestClientDetails> allClients;
+
+    /** Holds the conversation helper for the control level conversation for coordinating the test through. */
+    ConversationFactory conversationFactory;
+
+    /** Holds the connection that the control conversation is held over. */
+    Connection connection;
+
+    /** Holds the underlying {@link CoordinatingTestCase}s that this decorator wraps. */
+    WrappedSuiteTestDecorator testSuite;
+
+    /** Hold the current running test case. */
+    CoordinatingTestCase _currentTest = null;
+
+    /**
+     * Creates a wrapped suite test decorator from another one.
+     *
+     * @param suite               The test suite.
+     * @param availableClients    The list of all clients that responded to the compulsory invite.
+     * @param controlConversation The conversation helper for the control level, test coordination conversation.
+     * @param controlConnection   The connection that the coordination messages are sent over.
+     */
+    public ListeningTestDecorator(WrappedSuiteTestDecorator suite, Set<TestClientDetails> availableClients,
+                                  ConversationFactory controlConversation, Connection controlConnection)
+    {
+        super(suite);
+
+        log.debug("public InvitingTestDecorator(WrappedSuiteTestDecorator suite, Set<TestClientDetails> allClients = "
+                  + availableClients + ", ConversationHelper controlConversation = " + controlConversation + "): called");
+
+        testSuite = suite;
+        allClients = availableClients;
+        conversationFactory = controlConversation;
+        connection = controlConnection;
+    }
+
+    /**
+     * Broadcasts a test invitation and accetps enlisting from participating clients. The wrapped test case is then
+     * repeated for every combination of test clients (provided the wrapped test case extends {@link
+     * CoordinatingTestCase}.
+     *
+     * <p/>Any JMSExceptions during the invite/enlist conversation will be allowed to fall through as runtime
+     * exceptions, resulting in the non-completion of the test run.
+     *
+     * @param testResult The the results object to monitor the test results with.
+     *
+     * @todo Better error recovery for failure of the invite/enlist conversation could be added.
+     */
+    public void run(TestResult testResult)
+    {
+        log.debug("public void run(TestResult testResult): called");
+
+        Collection<Test> tests = testSuite.getAllUnderlyingTests();
+
+        for (Test test : tests)
+        {
+            CoordinatingTestCase coordTest = (CoordinatingTestCase) test;
+
+            Set<TestClientDetails> enlists = signupClients(coordTest);
+
+            if (enlists.size() == 0)
+            {
+                throw new RuntimeException("No clients to test with");
+            }
+
+            Iterator<TestClientDetails> clients = enlists.iterator();
+            coordTest.setSender(clients.next());
+
+            while (clients.hasNext())
+            {
+                // Set the sending and receiving client details on the test case.
+                coordTest.setReceiver(clients.next());
+            }
+
+            // Pass down the connection to hold the coordination conversation over.
+            coordTest.setConversationFactory(conversationFactory);
+
+
+            if (coordTest instanceof ListeningCoordinatorTest)
+            {
+                _currentTest = coordTest;
+            }
+            // Execute the test case.
+            coordTest.run(testResult);
+
+            _currentTest = null;
+        }
+    }
+
+    private Set<TestClientDetails> signupClients(CoordinatingTestCase coordTest)
+    {
+        // Broadcast the invitation to find out what clients are available to test.
+        Set<TestClientDetails> enlists;
+        try
+        {
+            Message invite = conversationFactory.getSession().createMessage();
+            Destination controlTopic = conversationFactory.getSession().createTopic("iop.control");
+            ConversationFactory.Conversation conversation = conversationFactory.startConversation();
+
+            invite.setStringProperty("CONTROL_TYPE", "INVITE");
+            invite.setStringProperty("TEST_NAME", coordTest.getTestCaseNameForTestMethod(coordTest.getName()));
+
+            conversation.send(controlTopic, invite);
+
+            // Wait for a short time, to give test clients an opportunity to reply to the invitation.
+            Collection<Message> replies = conversation.receiveAll(allClients.size(), 5000);
+
+            log.debug("Received " + replies.size() + " enlist replies");
+            
+            enlists = Coordinator.extractEnlists(replies);
+
+            //Create topic to listen on for latejoiners
+            Destination listenTopic = conversationFactory.getSession().createTopic("iop.control.test." + coordTest.getTestCaseNameForTestMethod(coordTest.getName()));
+
+            //Listen for joiners
+            conversationFactory.getSession().createConsumer(listenTopic).setMessageListener(this);
+            log.debug("Created consumer on :" + listenTopic);
+        }
+        catch (JMSException e)
+        {
+            throw new RuntimeException("There was a JMSException during the invite/enlist conversation.", e);
+        }
+
+        return enlists;
+    }
+
+    /**
+     * Prints a string summarizing this test decorator, mainly for debugging purposes.
+     *
+     * @return String representation for debugging purposes.
+     */
+    public String toString()
+    {
+        return "ListeningTestDecorator: [ testSuite = " + testSuite + " ]";
+    }
+
+
+    public void onMessage(Message message)
+    {
+        try
+        {
+            if (message.getStringProperty("CONTROL_TYPE").equals("LATEJOIN"))
+            {
+                ((ListeningCoordinatorTest) _currentTest).latejoin(message);
+            }
+        }
+        catch (JMSException e)
+        {
+            log.debug("Unable to process message:" + message);
+        }
+    }
+}
diff --git a/java/integrationtests/src/main/java/org/apache/qpid/interop/testclient/TestClient.java b/java/integrationtests/src/main/java/org/apache/qpid/interop/testclient/TestClient.java
index 0b9c72e1b6..6cca23446f 100644
--- a/java/integrationtests/src/main/java/org/apache/qpid/interop/testclient/TestClient.java
+++ b/java/integrationtests/src/main/java/org/apache/qpid/interop/testclient/TestClient.java
@@ -75,16 +75,17 @@ public class TestClient implements MessageListener
     /** Holds all the test cases loaded from the classpath. */

     Map<String, InteropClientTestCase> testCases = new HashMap<String, InteropClientTestCase>();

 

-    InteropClientTestCase currentTestCase;

+    protected InteropClientTestCase currentTestCase;

 

-    private MessageProducer producer;

-    private Session session;

+    protected Connection _connection;

+    protected MessageProducer producer;

+    protected Session session;

 

-    private String clientName = CLIENT_NAME;

+    protected String clientName = CLIENT_NAME;

 

     /**

-     * Creates a new interop test client, listenting to the specified broker and virtual host, with the specified

-     * client identifying name.

+     * Creates a new interop test client, listenting to the specified broker and virtual host, with the specified client

+     * identifying name.

      *

      * @param brokerUrl   The url of the broker to connect to.

      * @param virtualHost The virtual host to conect to.

@@ -93,7 +94,7 @@ public class TestClient implements MessageListener
     public TestClient(String brokerUrl, String virtualHost, String clientName)

     {

         log.debug("public TestClient(String brokerUrl = " + brokerUrl + ", String virtualHost = " + virtualHost

-            + ", String clientName = " + clientName + "): called");

+                  + ", String clientName = " + clientName + "): called");

 

         // Retain the connection parameters.

         this.brokerUrl = brokerUrl;

@@ -117,13 +118,13 @@ public class TestClient implements MessageListener
     {

         // Use the command line parser to evaluate the command line.

         CommandLineParser commandLine =

-            new CommandLineParser(

-                new String[][]

-                {

-                    { "b", "The broker URL.", "broker", "false" },

-                    { "h", "The virtual host to use.", "virtual host", "false" },

-                    { "n", "The test client name.", "name", "false" }

-                });

+                new CommandLineParser(

+                        new String[][]

+                                {

+                                        {"b", "The broker URL.", "broker", "false"},

+                                        {"h", "The virtual host to use.", "virtual host", "false"},

+                                        {"n", "The test client name.", "name", "false"}

+                                });

 

         // Capture the command line arguments or display errors and correct usage and then exit.

         Properties options = null;

@@ -151,9 +152,17 @@ public class TestClient implements MessageListener
         // Create a test client and start it running.

         TestClient client = new TestClient(brokerUrl, virtualHost, (clientName == null) ? CLIENT_NAME : clientName);

 

+        // Use a class path scanner to find all the interop test case implementations.

+        Collection<Class<? extends InteropClientTestCase>> testCaseClasses =

+                new ArrayList<Class<? extends InteropClientTestCase>>();

+        // ClasspathScanner.getMatches(InteropClientTestCase.class, "^TestCase.*", true);

+        // Hard code the test classes till the classpath scanner is fixed.

+        Collections.addAll(testCaseClasses,

+                           new Class[]{TestCase1DummyRun.class, TestCase2BasicP2P.class, TestClient.class});

+

         try

         {

-            client.start();

+            client.start(testCaseClasses);

         }

         catch (Exception e)

         {

@@ -165,20 +174,12 @@ public class TestClient implements MessageListener
     /**

      * Starts the interop test client running. This causes it to start listening for incoming test invites.

      *

-     * @throws JMSException Any underlying JMSExceptions are allowed to fall through.

+     * @throws JMSException Any underlying JMSExceptions are allowed to fall through. @param testCaseClasses

      */

-    private void start() throws JMSException

+    protected void start(Collection<Class<? extends InteropClientTestCase>> testCaseClasses) throws JMSException

     {

         log.debug("private void start(): called");

 

-        // Use a class path scanner to find all the interop test case implementations.

-        Collection<Class<? extends InteropClientTestCase>> testCaseClasses =

-            new ArrayList<Class<? extends InteropClientTestCase>>();

-        // ClasspathScanner.getMatches(InteropClientTestCase.class, "^TestCase.*", true);

-        // Hard code the test classes till the classpath scanner is fixed.

-        Collections.addAll(testCaseClasses,

-            new Class[] { TestCase1DummyRun.class, TestCase2BasicP2P.class, TestCase3BasicPubSub.class });

-

         // Create all the test case implementations and index them by the test names.

         for (Class<? extends InteropClientTestCase> nextClass : testCaseClasses)

         {

@@ -200,9 +201,9 @@ public class TestClient implements MessageListener
         }

 

         // Open a connection to communicate with the coordinator on.

-        Connection connection = createConnection(DEFAULT_CONNECTION_PROPS_RESOURCE, brokerUrl, virtualHost);

+        _connection = createConnection(DEFAULT_CONNECTION_PROPS_RESOURCE, brokerUrl, virtualHost);

 

-        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

+        session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

 

         // Set this up to listen for control messages.

         MessageConsumer consumer = session.createConsumer(session.createTopic("iop.control." + clientName));

@@ -215,7 +216,7 @@ public class TestClient implements MessageListener
         producer = session.createProducer(null);

 

         // Start listening for incoming control messages.

-        connection.start();

+        _connection.start();

     }

 

     /**

@@ -232,22 +233,25 @@ public class TestClient implements MessageListener
      * @param virtualHost             The virtual host to connectio to, <tt>null</tt> to use the default.

      *

      * @return A JMS conneciton.

+     *

+     * @todo Make username/password configurable. Allow multiple urls for fail over. Once it feels right, move it to a

+     * Utils library class.

      */

     public static Connection createConnection(String connectionPropsResource, String brokerUrl, String virtualHost)

     {

         log.debug("public static Connection createConnection(String connectionPropsResource = " + connectionPropsResource

-            + ", String brokerUrl = " + brokerUrl + ", String virtualHost = " + virtualHost + "): called");

+                  + ", String brokerUrl = " + brokerUrl + ", String virtualHost = " + virtualHost + "): called");

 

         try

         {

             Properties connectionProps =

-                PropertiesUtils.getProperties(TestClient.class.getClassLoader().getResourceAsStream(

-                        connectionPropsResource));

+                    PropertiesUtils.getProperties(TestClient.class.getClassLoader().getResourceAsStream(

+                            connectionPropsResource));

 

             if (brokerUrl != null)

             {

                 String connectionString =

-                    "amqp://guest:guest/" + ((virtualHost != null) ? virtualHost : "") + "?brokerlist='" + brokerUrl + "'";

+                        "amqp://guest:guest/" + ((virtualHost != null) ? virtualHost : "") + "?brokerlist='" + brokerUrl + "'";

                 connectionProps.setProperty(CONNECTION_PROPERTY, connectionString);

             }

 

@@ -286,21 +290,21 @@ public class TestClient implements MessageListener
             String controlType = message.getStringProperty("CONTROL_TYPE");

             String testName = message.getStringProperty("TEST_NAME");

 

+            log.info("onMessage(Message message = " + message + "): for '" + controlType + "' to '" + testName + "'");

+

             // Check if the message is a test invite.

             if ("INVITE".equals(controlType))

             {

-                String testCaseName = message.getStringProperty("TEST_NAME");

-

                 // Flag used to indicate that an enlist should be sent. Only enlist to compulsory invites or invites

                 // for which test cases exist.

                 boolean enlist = false;

 

-                if (testCaseName != null)

+                if (testName != null)

                 {

-                    log.debug("Got an invite to test: " + testCaseName);

+                    log.debug("Got an invite to test: " + testName);

 

                     // Check if the requested test case is available.

-                    InteropClientTestCase testCase = testCases.get(testCaseName);

+                    InteropClientTestCase testCase = testCases.get(testName);

 

                     if (testCase != null)

                     {

@@ -308,6 +312,10 @@ public class TestClient implements MessageListener
                         currentTestCase = testCase;

                         enlist = true;

                     }

+                    else

+                    {

+                        log.warn("'" + testName + "' not part of this clients tests.");

+                    }

                 }

                 else

                 {

@@ -325,6 +333,8 @@ public class TestClient implements MessageListener
                     enlistMessage.setStringProperty("CLIENT_PRIVATE_CONTROL_KEY", "iop.control." + clientName);

                     enlistMessage.setJMSCorrelationID(message.getJMSCorrelationID());

 

+                    log.info("Sending Message '" + enlistMessage + "'. to " + message.getJMSReplyTo());

+

                     producer.send(message.getJMSReplyTo(), enlistMessage);

                 }

             }

@@ -369,9 +379,10 @@ public class TestClient implements MessageListener
             }

             else if ("TERMINATE".equals(controlType))

             {

-                System.out.println("Received termination instruction from coordinator.");

+                log.info("Received termination instruction from coordinator.");

 

                 // Is a cleaner shutdown needed?

+                _connection.close();

                 System.exit(0);

             }

             else

diff --git a/java/integrationtests/src/main/java/org/apache/qpid/interop/testclient/testcases/TestCase1DummyRun.java b/java/integrationtests/src/main/java/org/apache/qpid/interop/testclient/testcases/TestCase1DummyRun.java
index 874f86daa9..85b89172bb 100644
--- a/java/integrationtests/src/main/java/org/apache/qpid/interop/testclient/testcases/TestCase1DummyRun.java
+++ b/java/integrationtests/src/main/java/org/apache/qpid/interop/testclient/testcases/TestCase1DummyRun.java
@@ -20,14 +20,14 @@
  */

 package org.apache.qpid.interop.testclient.testcases;

 

-import javax.jms.JMSException;

-import javax.jms.Message;

-import javax.jms.Session;

-

 import org.apache.log4j.Logger;

 

 import org.apache.qpid.interop.testclient.InteropClientTestCase;

 

+import javax.jms.JMSException;

+import javax.jms.Message;

+import javax.jms.Session;

+

 /**

  * Implements tet case 1, dummy run. This test case sends no test messages, it exists to confirm that the test harness

  * is interacting with the coordinator correctly.

diff --git a/java/integrationtests/src/main/java/org/apache/qpid/interop/testclient/testcases/TestCase3BasicPubSub.java b/java/integrationtests/src/main/java/org/apache/qpid/interop/testclient/testcases/TestCase3BasicPubSub.java
index 2773cad3f3..ec8c72afa9 100644
--- a/java/integrationtests/src/main/java/org/apache/qpid/interop/testclient/testcases/TestCase3BasicPubSub.java
+++ b/java/integrationtests/src/main/java/org/apache/qpid/interop/testclient/testcases/TestCase3BasicPubSub.java
@@ -5,7 +5,6 @@ import javax.jms.*;
 import org.apache.log4j.Logger;

 

 import org.apache.qpid.interop.testclient.InteropClientTestCase;

-import org.apache.qpid.interop.testclient.TestClient;

 

 /**

  * Implements test case 3, basic pub/sub. Sends/received a specified number of messages to a specified route on the

@@ -120,8 +119,8 @@ public class TestCase3BasicPubSub implements InteropClientTestCase
             session = new Session[1];

 

             connection[0] =

-                TestClient.createConnection(TestClient.DEFAULT_CONNECTION_PROPS_RESOURCE, TestClient.brokerUrl,

-                    TestClient.virtualHost);

+                org.apache.qpid.interop.testclient.TestClient.createConnection(org.apache.qpid.interop.testclient.TestClient.DEFAULT_CONNECTION_PROPS_RESOURCE, org.apache.qpid.interop.testclient.TestClient.brokerUrl,

+                    org.apache.qpid.interop.testclient.TestClient.virtualHost);

             session[0] = connection[0].createSession(false, Session.AUTO_ACKNOWLEDGE);

 

             // Extract and retain the test parameters.

@@ -140,8 +139,8 @@ public class TestCase3BasicPubSub implements InteropClientTestCase
             for (int i = 0; i < numReceivers; i++)

             {

                 connection[i] =

-                    TestClient.createConnection(TestClient.DEFAULT_CONNECTION_PROPS_RESOURCE, TestClient.brokerUrl,

-                        TestClient.virtualHost);

+                    org.apache.qpid.interop.testclient.TestClient.createConnection(org.apache.qpid.interop.testclient.TestClient.DEFAULT_CONNECTION_PROPS_RESOURCE, org.apache.qpid.interop.testclient.TestClient.brokerUrl,

+                        org.apache.qpid.interop.testclient.TestClient.virtualHost);

                 session[i] = connection[i].createSession(false, Session.AUTO_ACKNOWLEDGE);

 

                 sendDestination = session[i].createTopic(sendKey);

diff --git a/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedTestClient.java b/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedTestClient.java
new file mode 100644
index 0000000000..3a0d587d44
--- /dev/null
+++ b/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedTestClient.java
@@ -0,0 +1,567 @@
+package org.apache.qpid.sustained;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.client.AMQNoConsumersException;
+import org.apache.qpid.client.AMQNoRouteException;
+import org.apache.qpid.interop.testclient.testcases.TestCase3BasicPubSub;
+
+import javax.jms.Connection;
+import javax.jms.Destination;
+import javax.jms.ExceptionListener;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Implements test case 3, basic pub/sub. Sends/received a specified number of messages to a specified route on the
+ * default topic exchange, using the specified number of receiver connections. Produces reports on the actual number of
+ * messages sent/received.
+ *
+ * <p><table id="crc"><caption>CRC Card</caption> <tr><th> Responsibilities <th> Collaborations <tr><td> Supply the name
+ * of the test case that this implements. <tr><td> Accept/Reject invites based on test parameters. <tr><td> Adapt to
+ * assigned roles. <tr><td> Send required number of test messages using pub/sub. <tr><td> Generate test reports.
+ * </table>
+ */
+public class SustainedTestClient extends TestCase3BasicPubSub implements ExceptionListener
+{
+    /** Used for debugging. */
+    private static final Logger log = Logger.getLogger(SustainedTestClient.class);
+
+    /** The role to be played by the test. */
+    private Roles role;
+
+    /** The number of test messages to send. */
+//    private int numMessages;
+
+    /** The number of receiver connection to use. */
+    private int numReceivers;
+
+    /** The routing key to send them to on the default direct exchange. */
+    private Destination sendDestination;
+
+    /** The routing key to send updates to on the default direct exchange. */
+    private Destination sendUpdateDestination;
+
+
+    /** The connections to send/receive the test messages on. */
+    private Connection[] connection;
+
+    /** The sessions to send/receive the test messages on. */
+    private Session[] session;
+
+    /** The producer to send the test messages with. */
+    MessageProducer producer;
+
+    /** Adapter that adjusts the send rate based on the updates from clients. */
+    SustainedRateAdapter _rateAdapter;
+
+    /**  */
+    int updateInterval;
+
+    private boolean _running = true;
+
+    /**
+     * Should provide the name of the test case that this class implements. The exact names are defined in the interop
+     * testing spec.
+     *
+     * @return The name of the test case that this implements.
+     */
+    public String getName()
+    {
+        log.debug("public String getName(): called");
+
+        return "Perf_SustainedPubSub";
+    }
+
+    /**
+     * Assigns the role to be played by this test case. The test parameters are fully specified in the assignment
+     * message. When this method return the test case will be ready to execute.
+     *
+     * @param role              The role to be played; sender or receiver.
+     * @param assignRoleMessage The role assingment message, contains the full test parameters.
+     *
+     * @throws JMSException Any JMSException resulting from reading the message are allowed to fall through.
+     */
+    public void assignRole(Roles role, Message assignRoleMessage) throws JMSException
+    {
+        log.debug("public void assignRole(Roles role = " + role + ", Message assignRoleMessage = " + assignRoleMessage
+                  + "): called");
+
+        // Take note of the role to be played.
+        this.role = role;
+
+        // Extract and retain the test parameters.
+        numReceivers = assignRoleMessage.getIntProperty("SUSTAINED_NUM_RECEIVERS");
+        updateInterval = assignRoleMessage.getIntProperty("SUSTAINED_UPDATE_INTERVAL");
+        String sendKey = assignRoleMessage.getStringProperty("SUSTAINED_KEY");
+        String sendUpdateKey = assignRoleMessage.getStringProperty("SUSTAINED_UPDATE_KEY");
+        int ackMode = assignRoleMessage.getIntProperty("ACKNOWLEDGE_MODE");
+
+        log.debug("numReceivers = " + numReceivers);
+        log.debug("updateInterval = " + updateInterval);
+        log.debug("ackMode = " + ackMode);
+        log.debug("sendKey = " + sendKey);
+        log.debug("sendUpdateKey = " + sendUpdateKey);
+        log.debug("role = " + role);
+
+        switch (role)
+        {
+            // Check if the sender role is being assigned, and set up a single message producer if so.
+            case SENDER:
+                log.info("*********** Creating SENDER");
+                // Create a new connection to pass the test messages on.
+                connection = new Connection[1];
+                session = new Session[1];
+
+                connection[0] =
+                        org.apache.qpid.interop.testclient.TestClient.createConnection(org.apache.qpid.interop.testclient.TestClient.DEFAULT_CONNECTION_PROPS_RESOURCE, org.apache.qpid.interop.testclient.TestClient.brokerUrl,
+                                                                                       org.apache.qpid.interop.testclient.TestClient.virtualHost);
+                session[0] = connection[0].createSession(false, ackMode);
+
+                // Extract and retain the test parameters.
+                sendDestination = session[0].createTopic(sendKey);
+
+                connection[0].setExceptionListener(this);
+
+                producer = session[0].createProducer(sendDestination);
+
+                sendUpdateDestination = session[0].createTopic(sendUpdateKey);
+                MessageConsumer updateConsumer = session[0].createConsumer(sendUpdateDestination);
+
+                _rateAdapter = new SustainedRateAdapter(this);
+                updateConsumer.setMessageListener(_rateAdapter);
+
+
+                break;
+
+                // Otherwise the receiver role is being assigned, so set this up to listen for messages on the required number
+                // of receiver connections.
+            case RECEIVER:
+                log.info("*********** Creating RECEIVER");
+                // Create the required number of receiver connections.
+                connection = new Connection[numReceivers];
+                session = new Session[numReceivers];
+
+                for (int i = 0; i < numReceivers; i++)
+                {
+                    connection[i] =
+                            org.apache.qpid.interop.testclient.TestClient.createConnection(org.apache.qpid.interop.testclient.TestClient.DEFAULT_CONNECTION_PROPS_RESOURCE,
+                                                                                           org.apache.qpid.interop.testclient.TestClient.brokerUrl,
+                                                                                           org.apache.qpid.interop.testclient.TestClient.virtualHost);
+                    session[i] = connection[i].createSession(false, ackMode);
+
+                    sendDestination = session[i].createTopic(sendKey);
+
+                    sendUpdateDestination = session[i].createTopic(sendUpdateKey);
+
+                    MessageConsumer consumer = session[i].createConsumer(sendDestination);
+
+                    consumer.setMessageListener(new SustainedListener(TestClient.CLIENT_NAME + "-" + i, updateInterval, session[i], sendUpdateDestination));
+                }
+
+                break;
+        }
+
+        // Start all the connection dispatcher threads running.
+        for (int i = 0; i < connection.length; i++)
+        {
+            connection[i].start();
+        }
+    }
+
+    /** Performs the test case actions. */
+    public void start() throws JMSException
+    {
+        log.debug("public void start(): called");
+
+        // Check that the sender role is being performed.
+        switch (role)
+        {
+            // Check if the sender role is being assigned, and set up a single message producer if so.
+            case SENDER:
+                Message testMessage = session[0].createTextMessage("test");
+
+//            for (int i = 0; i < numMessages; i++)
+                while (_running)
+                {
+                    producer.send(testMessage);
+
+                    _rateAdapter.sentMessage();
+                }
+                break;
+            case RECEIVER:
+
+        }
+    }
+
+    /**
+     * Gets a report on the actions performed by the test case in its assigned role.
+     *
+     * @param session The session to create the report message in.
+     *
+     * @return The report message.
+     *
+     * @throws JMSException Any JMSExceptions resulting from creating the report are allowed to fall through.
+     */
+    public Message getReport(Session session) throws JMSException
+    {
+        log.debug("public Message getReport(Session session): called");
+
+        // Close the test connections.
+        for (int i = 0; i < connection.length; i++)
+        {
+            connection[i].close();
+        }
+
+        Message report = session.createMessage();
+        report.setStringProperty("CONTROL_TYPE", "REPORT");
+
+        return report;
+    }
+
+    public void onException(JMSException jmsException)
+    {
+        Exception linked = jmsException.getLinkedException();
+
+        if (linked != null)
+        {
+            if (linked instanceof AMQNoRouteException)
+            {
+                log.warn("No route .");
+            }
+            else if (linked instanceof AMQNoConsumersException)
+            {
+                log.warn("No clients currently available for message:" + ((AMQNoConsumersException) linked).getUndeliveredMessage());
+            }
+            else
+            {
+
+                log.warn("LinkedException:" + linked);
+            }
+
+            _rateAdapter.NO_CLIENTS = true;
+        }
+        else
+        {
+            log.warn("Exception:" + linked);
+        }
+    }
+
+    class SustainedListener implements MessageListener
+    {
+        private int _received = 0;
+        private int _updateInterval = 0;
+        private Long _time;
+        MessageProducer _updater;
+        Session _session;
+        String _client;
+
+
+        public SustainedListener(String clientname, int updateInterval, Session session, Destination sendDestination) throws JMSException
+        {
+            _updateInterval = updateInterval;
+            _client = clientname;
+            _session = session;
+            _updater = session.createProducer(sendDestination);
+        }
+
+        public void setReportInterval(int reportInterval)
+        {
+            _updateInterval = reportInterval;
+            _received = 0;
+        }
+
+        public void onMessage(Message message)
+        {
+            if (log.isDebugEnabled())
+            {
+                log.debug("Message " + _received + "received in listener");
+            }
+
+            if (message instanceof TextMessage)
+            {
+
+                try
+                {
+                    if (((TextMessage) message).getText().equals("test"))
+                    {
+                        if (_received == 0)
+                        {
+                            _time = System.nanoTime();
+                            sendStatus(0, _received);
+                        }
+
+                        _received++;
+
+                        if (_received % _updateInterval == 0)
+                        {
+                            Long currentTime = System.nanoTime();
+
+                            try
+                            {
+                                sendStatus(currentTime - _time, _received);
+                                _time = currentTime;
+                            }
+                            catch (JMSException e)
+                            {
+                                log.error("Unable to send update.");
+                            }
+                        }
+
+                    }
+                }
+                catch (JMSException e)
+                {
+                    //ignore error
+                }
+            }
+        }
+
+        private void sendStatus(long time, int received) throws JMSException
+        {
+            Message updateMessage = _session.createTextMessage("update");
+            updateMessage.setStringProperty("CLIENT_ID", _client);
+            updateMessage.setStringProperty("CONTROL_TYPE", "UPDATE");
+            updateMessage.setLongProperty("RECEIVED", received);
+            updateMessage.setLongProperty("DURATION", time);
+
+            log.info("**** SENDING **** CLIENT_ID:" + _client + " RECEIVED:" + received + " DURATION:" + time);
+
+            _updater.send(updateMessage);
+        }
+
+    }
+
+    class SustainedRateAdapter implements MessageListener
+    {
+        private SustainedTestClient _client;
+        private long _variance = 250; //no. messages to allow drifting
+        private volatile long _delay;   //in nanos
+        private long _sent;
+        private Map<String, Long> _slowClients = new HashMap<String, Long>();
+        private static final long PAUSE_SLEEP = 10; // 10 ms
+        private static final long NO_CLIENT_SLEEP = 1000; // 1s 
+        private static final long MAX_MESSAGE_DRIFT = 1000; // no messages drifted from producer
+        private volatile boolean NO_CLIENTS = true;
+        private int _delayShifting;
+        private static final int REPORTS_WITHOUT_CHANGE = 10;
+        private static final double MAXIMUM_DELAY_SHIFT = .02; //2% 
+
+        SustainedRateAdapter(SustainedTestClient client)
+        {
+            _client = client;
+        }
+
+        public void onMessage(Message message)
+        {
+            if (log.isDebugEnabled())
+            {
+                log.debug("SustainedRateAdapter onMessage(Message message = " + message + "): called");
+            }
+
+            try
+            {
+                String controlType = message.getStringProperty("CONTROL_TYPE");
+
+                // Check if the message is a test invite.
+                if ("UPDATE".equals(controlType))
+                {
+                    NO_CLIENTS = false;
+                    long duration = message.getLongProperty("DURATION");
+                    long received = message.getLongProperty("RECEIVED");
+                    String client = message.getStringProperty("CLIENT_ID");
+
+                    log.info("**** SENDING **** CLIENT_ID:" + client + " RECEIVED:" + received + " DURATION:" + duration);
+
+
+                    recordSlow(client, received);
+
+                    adjustDelay(client, received, duration);
+                }
+            }
+            catch (JMSException e)
+            {
+                //
+            }
+        }
+
+        class Pair<X, Y>
+        {
+            X item1;
+            Y item2;
+
+            Pair(X i1, Y i2)
+            {
+                item1 = i1;
+                item2 = i2;
+            }
+
+            X getItem1()
+            {
+                return item1;
+            }
+
+            Y getItem2()
+            {
+                return item2;
+            }
+        }
+
+        Map<String, Pair<Long, Long>> delays = new HashMap<String, Pair<Long, Long>>();
+        Long totalReceived = 0L;
+        Long totalDuration = 0L;
+
+        private void adjustDelay(String client, long received, long duration)
+        {
+            Pair<Long, Long> current = delays.get(client);
+
+            if (current == null)
+            {
+                delays.put(client, new Pair<Long, Long>(received, duration));
+            }
+            else
+            {
+                //reduce totals
+                totalReceived -= current.getItem1();
+                totalDuration -= current.getItem2();
+            }
+
+            totalReceived += received;
+            totalDuration += duration;
+
+            long averageDuration = totalDuration / delays.size();
+
+            long diff = Math.abs(_delay - averageDuration);
+
+            //if the averageDuration differs from the current by more than the specified variane then adjust delay.
+            if (diff > _variance)
+            {
+                if (averageDuration > _delay)
+                {
+                    // we can go faster
+                    _delay -= diff;
+                    if (_delay < 0)
+                    {
+                        _delay = 0;
+                    }
+                }
+                else
+                {
+                    // we need to slow down
+                    _delay += diff;
+                }
+                delayChanged();
+            }
+            else
+            {
+                delayStable();
+            }
+
+        }
+
+        private void delayChanged()
+        {
+            _delayShifting = REPORTS_WITHOUT_CHANGE;
+        }
+
+        private void delayStable()
+        {
+            _delayShifting--;
+
+            if (_delayShifting < 0)
+            {
+                _delayShifting = 0;
+                log.info("Delay stabilised:" + _delay);
+            }
+        }
+
+        // Record Slow clients
+        private void recordSlow(String client, long received)
+        {
+            if (received < (_sent - _variance))
+            {
+                _slowClients.put(client, received);
+            }
+            else
+            {
+                _slowClients.remove(client);
+            }
+        }
+
+        public void sentMessage()
+        {
+            if (_sent % updateInterval == 0)
+            {
+
+                // Cause test to pause when we have slow
+                if (!_slowClients.isEmpty() || NO_CLIENTS)
+                {
+                    log.info("Pausing for slow clients");
+
+                    //_delay <<= 1;
+
+                    while (!_slowClients.isEmpty())
+                    {
+                        sleep(PAUSE_SLEEP);
+                    }
+
+                    if (NO_CLIENTS)
+                    {
+                        sleep(NO_CLIENT_SLEEP);
+                    }
+
+                    log.debug("Continuing");
+                    return;
+                }
+                else
+                {
+                    log.info("Delay:" + _delay);
+                }
+            }
+
+            _sent++;
+
+            if (_delay > 0)
+            {
+                // less than 10ms sleep doesn't work.
+                // _delay is in nano seconds
+                if (_delay < 1000000)
+                {
+                    sleep(0, (int) _delay);
+                }
+                else
+                {
+                    if (_delay < 30000000000L)
+                    {
+                        sleep(_delay / 1000000, (int) (_delay % 1000000));
+                    }
+                }
+            }
+        }
+
+        private void sleep(long sleep)
+        {
+            sleep(sleep, 0);
+        }
+
+        private void sleep(long milli, int nano)
+        {
+            try
+            {
+                log.debug("Sleep:" + milli + ":" + nano);
+                Thread.sleep(milli, nano);
+            }
+            catch (InterruptedException e)
+            {
+                //
+            }
+        }
+    }
+
+}
diff --git a/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedTestCoordinator.java b/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedTestCoordinator.java
new file mode 100644
index 0000000000..4081d87192
--- /dev/null
+++ b/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedTestCoordinator.java
@@ -0,0 +1,219 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one
+ *  or more contributor license agreements.  See the NOTICE file
+ *  distributed with this work for additional information
+ *  regarding copyright ownership.  The ASF licenses this file
+ *  to you under the Apache License, Version 2.0 (the
+ *  "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing,
+ *  software distributed under the License is distributed on an
+ *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ *  KIND, either express or implied.  See the License for the
+ *  specific language governing permissions and limitations
+ *  under the License.    
+ *
+ * 
+ */
+package org.apache.qpid.sustained;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.interop.coordinator.ListeningCoordinatorTest;
+import org.apache.qpid.interop.coordinator.TestClientDetails;
+import org.apache.qpid.interop.coordinator.testcases.CoordinatingTestCase3BasicPubSub;
+import org.apache.qpid.util.ConversationFactory;
+
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.Session;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+public class SustainedTestCoordinator extends CoordinatingTestCase3BasicPubSub implements ListeningCoordinatorTest
+{
+    /** Used for debugging. */
+    private static final Logger log = Logger.getLogger(SustainedTestCoordinator.class);
+    private List<TestClientDetails> _receivers;
+    private static final String SUSTAINED_KEY = "Perf_SustainedPubSub";
+    Map<String, Object> _testProperties;
+
+    /**
+     * Creates a new coordinating test case with the specified name.
+     *
+     * @param name The test case name.
+     */
+    public SustainedTestCoordinator(String name)
+    {
+        super(name);
+        _receivers = new LinkedList();
+    }
+
+    /**
+     * Adds a receiver to this test.
+     *
+     * @param receiver The contact details of the sending client in the test.
+     */
+    public void setReceiver(TestClientDetails receiver)
+    {
+        _receivers.add(receiver);
+    }
+
+
+    /**
+     * Performs the a single test run
+     *
+     * @throws Exception if there was a problem running the test.
+     */
+    public void testBasicPubSub() throws Exception
+    {
+        log.debug("public void testSinglePubSubCycle(): called");
+
+        Map<String, Object> testConfig = new HashMap<String, Object>();
+        testConfig.put("TEST_NAME", "Perf_SustainedPubSub");
+        testConfig.put("SUSTAINED_KEY", SUSTAINED_KEY);
+        //testConfig.put("SUSTAINED_MSG_RATE", 10);
+        testConfig.put("SUSTAINED_NUM_RECEIVERS", 2);
+        testConfig.put("SUSTAINED_UPDATE_INTERVAL", 25);
+        testConfig.put("SUSTAINED_UPDATE_KEY", SUSTAINED_KEY + ".UPDATE");
+        testConfig.put("ACKNOWLEDGE_MODE", AMQSession.NO_ACKNOWLEDGE);
+
+        sequenceTest(testConfig);
+    }
+
+    /**
+     * Holds a test coordinating conversation with the test clients. This is the basic implementation of the inner loop
+     * of Use Case 5. It consists of assigning the test roles, begining the test and gathering the test reports from the
+     * participants.
+     *
+     * @param testProperties The test case definition.
+     *
+     * @return The test results from the senders and receivers.
+     *
+     * @throws javax.jms.JMSException All underlying JMSExceptions are allowed to fall through.
+     */
+    protected Message[] sequenceTest(Map<String, Object> testProperties) throws JMSException
+    {
+        log.debug("protected Message[] sequenceTest(Object... testProperties = " + testProperties + "): called");
+
+        Session session = conversationFactory.getSession();
+        Destination senderControlTopic = session.createTopic(sender.privateControlKey);
+
+        ConversationFactory.Conversation senderConversation = conversationFactory.startConversation();
+
+        // Assign the sender role to the sending test client.
+        Message assignSender = conversationFactory.getSession().createMessage();
+        setPropertiesOnMessage(assignSender, testProperties);
+        assignSender.setStringProperty("CONTROL_TYPE", "ASSIGN_ROLE");
+        assignSender.setStringProperty("ROLE", "SENDER");
+
+        senderConversation.send(senderControlTopic, assignSender);
+
+        //Assign and wait for the receiver ckuebts to be ready.
+        _testProperties = testProperties;
+
+        // Wait for the senders to confirm their roles.
+        senderConversation.receive();
+
+        assignReceivers();
+
+        // Start the test.
+        Message start = session.createMessage();
+        start.setStringProperty("CONTROL_TYPE", "START");
+
+        senderConversation.send(senderControlTopic, start);
+
+        // Wait for the test sender to return its report.
+        Message senderReport = senderConversation.receive();
+
+        try
+        {
+            Thread.sleep(500);
+        }
+        catch (InterruptedException e)
+        {
+        }
+
+        // Ask the receiver for its report.
+        Message statusRequest = session.createMessage();
+        statusRequest.setStringProperty("CONTROL_TYPE", "STATUS_REQUEST");
+
+
+        return new Message[]{senderReport};
+    }
+
+    private void assignReceivers()
+    {
+        for (TestClientDetails receiver : _receivers)
+        {
+            registerReceiver(receiver);
+        }
+    }
+
+    private void registerReceiver(TestClientDetails receiver)
+    {
+        log.info("registerReceiver called for receiver:" + receiver);
+        try
+        {
+            Session session = conversationFactory.getSession();
+            Destination receiverControlTopic = session.createTopic(receiver.privateControlKey);
+            ConversationFactory.Conversation receiverConversation = conversationFactory.startConversation();
+            // Assign the receiver role the receiving client.
+            Message assignReceiver = session.createMessage();
+            setPropertiesOnMessage(assignReceiver, _testProperties);
+            assignReceiver.setStringProperty("CONTROL_TYPE", "ASSIGN_ROLE");
+            assignReceiver.setStringProperty("ROLE", "RECEIVER");
+
+            receiverConversation.send(receiverControlTopic, assignReceiver);
+
+            //Don't wait for receiver to be ready.... we can't this is being done in
+            // the dispatcher thread, and most likely the acceptance message we
+            // want is sitting in the Dispatcher._queue waiting its turn for being
+            // dispatched so if we block here we won't can't get the message.             
+            // So assume consumer is ready for action.
+            //receiverConversation.receive();
+        }
+        catch (JMSException e)
+        {
+            log.warn("Unable to assign receiver:" + receiver + ". Due to:" + e.getMessage());
+        }
+    }
+
+    public void latejoin(Message message)
+    {
+        try
+        {
+
+            TestClientDetails clientDetails = new TestClientDetails();
+            clientDetails.clientName = message.getStringProperty("CLIENT_NAME");
+            clientDetails.privateControlKey = message.getStringProperty("CLIENT_PRIVATE_CONTROL_KEY");
+
+
+            registerReceiver(clientDetails);
+        }
+        catch (JMSException e)
+        {
+            //swallow
+        }
+    }
+
+    /**
+     * Should provide a translation from the junit method name of a test to its test case name as defined in the interop
+     * testing specification. For example the method "testP2P" might map onto the interop test case name
+     * "TC2_BasicP2P".
+     *
+     * @param methodName The name of the JUnit test method.
+     *
+     * @return The name of the corresponding interop test case.
+     */
+    public String getTestCaseNameForTestMethod(String methodName)
+    {
+        return "Perf_SustainedPubSub";
+    }
+}
diff --git a/java/integrationtests/src/main/java/org/apache/qpid/sustained/TestClient.java b/java/integrationtests/src/main/java/org/apache/qpid/sustained/TestClient.java
new file mode 100644
index 0000000000..44fc090410
--- /dev/null
+++ b/java/integrationtests/src/main/java/org/apache/qpid/sustained/TestClient.java
@@ -0,0 +1,157 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one
+ *  or more contributor license agreements.  See the NOTICE file
+ *  distributed with this work for additional information
+ *  regarding copyright ownership.  The ASF licenses this file
+ *  to you under the Apache License, Version 2.0 (the
+ *  "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing,
+ *  software distributed under the License is distributed on an
+ *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ *  KIND, either express or implied.  See the License for the
+ *  specific language governing permissions and limitations
+ *  under the License.    
+ *
+ * 
+ */
+package org.apache.qpid.sustained;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.interop.testclient.InteropClientTestCase;
+import org.apache.qpid.util.CommandLineParser;
+
+import javax.jms.JMSException;
+import javax.jms.Message;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Properties;
+
+public class TestClient extends org.apache.qpid.interop.testclient.TestClient
+{
+    private static Logger log = Logger.getLogger(TestClient.class);
+
+    /**
+     * Creates a new interop test client, listenting to the specified broker and virtual host, with the specified client
+     * identifying name.
+     *
+     * @param brokerUrl   The url of the broker to connect to.
+     * @param virtualHost The virtual host to conect to.
+     * @param clientName  The client name to use.
+     */
+    public TestClient(String brokerUrl, String virtualHost, String clientName)
+    {
+        super(brokerUrl, virtualHost, clientName);
+    }
+
+    /**
+     * The entry point for the interop test coordinator. This client accepts the following command line arguments:
+     *
+     * <p/><table> <tr><td> -b         <td> The broker URL.       <td> Optional. <tr><td> -h         <td> The virtual
+     * host.     <td> Optional. <tr><td> -n         <td> The test client name. <td> Optional. <tr><td> name=value <td>
+     * Trailing argument define name/value pairs. Added to system properties. <td> Optional. </table>
+     *
+     * @param args The command line arguments.
+     */
+    public static void main(String[] args)
+    {
+        // Use the command line parser to evaluate the command line.
+        CommandLineParser commandLine =
+                new CommandLineParser(
+                        new String[][]
+                                {
+                                        {"b", "The broker URL.", "broker", "false"},
+                                        {"h", "The virtual host to use.", "virtual host", "false"},
+                                        {"n", "The test client name.", "name", "false"},
+                                        {"j", "Join this test client to running test.", "join", ""}
+                                });
+
+        // Capture the command line arguments or display errors and correct usage and then exit.
+        Properties options = null;
+
+        try
+        {
+            options = commandLine.parseCommandLine(args);
+        }
+        catch (IllegalArgumentException e)
+        {
+            System.out.println(commandLine.getErrors());
+            System.out.println(commandLine.getUsage());
+            System.exit(1);
+        }
+
+        // Extract the command line options.
+        String brokerUrl = options.getProperty("b");
+        String virtualHost = options.getProperty("h");
+        String clientName = options.getProperty("n");
+        String join = options.getProperty("j");
+
+        // Add all the trailing command line options (name=value pairs) to system properties. Tests may pick up
+        // overridden values from there.
+        commandLine.addCommandLineToSysProperties();
+
+        // Create a test client and start it running.
+        TestClient client = new TestClient(brokerUrl, virtualHost, (clientName == null) ? CLIENT_NAME : clientName);
+
+        // Use a class path scanner to find all the interop test case implementations.
+        Collection<Class<? extends InteropClientTestCase>> testCaseClasses =
+                new ArrayList<Class<? extends InteropClientTestCase>>();
+        // ClasspathScanner.getMatches(InteropClientTestCase.class, "^TestCase.*", true);
+        // Hard code the test classes till the classpath scanner is fixed.
+        Collections.addAll(testCaseClasses,
+                           SustainedTestClient.class);
+
+
+        try
+        {
+            client.start(testCaseClasses, join);
+        }
+        catch (Exception e)
+        {
+            log.error("The test client was unable to start.", e);
+            System.exit(1);
+        }
+    }
+
+    protected void start(Collection<Class<? extends InteropClientTestCase>> testCaseClasses, String join) throws JMSException, ClassNotFoundException
+    {
+        super.start(testCaseClasses);
+        log.debug("private void start(): called");
+
+        if (join != null && !join.equals(""))
+        {
+            Message latejoin = session.createMessage();
+
+            try
+            {
+                Object test = Class.forName(join).newInstance();
+                if (test instanceof InteropClientTestCase)
+                {
+                    currentTestCase = (InteropClientTestCase) test;
+                }
+                else
+                {
+                    throw new RuntimeException("Requested to join class '" + join + "' but this is not a InteropClientTestCase.");
+                }
+
+                latejoin.setStringProperty("CONTROL_TYPE", "LATEJOIN");
+                latejoin.setStringProperty("CLIENT_NAME", clientName);
+                latejoin.setStringProperty("CLIENT_PRIVATE_CONTROL_KEY", "iop.control." + clientName);
+                producer.send(session.createTopic("iop.control.test." + currentTestCase.getName()), latejoin);
+            }
+            catch (InstantiationException e)
+            {
+                log.warn("Unable to request latejoining of test:" + currentTestCase);
+            }
+            catch (IllegalAccessException e)
+            {
+                log.warn("Unable to request latejoining of test:" + currentTestCase);
+            }
+        }
+    }
+
+}
diff --git a/java/integrationtests/src/main/java/org/apache/qpid/sustained/TestCoordinator.java b/java/integrationtests/src/main/java/org/apache/qpid/sustained/TestCoordinator.java
new file mode 100644
index 0000000000..7e12fe39fb
--- /dev/null
+++ b/java/integrationtests/src/main/java/org/apache/qpid/sustained/TestCoordinator.java
@@ -0,0 +1,117 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one
+ *  or more contributor license agreements.  See the NOTICE file
+ *  distributed with this work for additional information
+ *  regarding copyright ownership.  The ASF licenses this file
+ *  to you under the Apache License, Version 2.0 (the
+ *  "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing,
+ *  software distributed under the License is distributed on an
+ *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ *  KIND, either express or implied.  See the License for the
+ *  specific language governing permissions and limitations
+ *  under the License.    
+ *
+ * 
+ */
+package org.apache.qpid.sustained;
+
+import org.apache.qpid.interop.coordinator.Coordinator;
+import org.apache.qpid.interop.coordinator.ListeningTestDecorator;
+import org.apache.qpid.interop.coordinator.TestClientDetails;
+import org.apache.qpid.util.CommandLineParser;
+import org.apache.qpid.util.ConversationFactory;
+import org.apache.log4j.Logger;
+
+import java.util.Properties;
+import java.util.Set;
+
+import junit.framework.TestResult;
+import uk.co.thebadgerset.junit.extensions.WrappedSuiteTestDecorator;
+
+import javax.jms.Connection;
+
+public class TestCoordinator extends Coordinator
+{
+
+    private static final Logger log = Logger.getLogger(TestCoordinator.class);
+
+    /**
+     * Creates an interop test coordinator on the specified broker and virtual host.
+     *
+     * @param brokerUrl   The URL of the broker to connect to.
+     * @param virtualHost The virtual host to run all tests on. Optional, may be <tt>null</tt>.
+     */
+    TestCoordinator(String brokerUrl, String virtualHost)
+    {
+        super(brokerUrl, virtualHost);
+    }
+
+    protected WrappedSuiteTestDecorator newTestDecorator(WrappedSuiteTestDecorator targetTest, Set<TestClientDetails> enlistedClients, ConversationFactory conversationFactory, Connection connection)
+    {
+        return  new ListeningTestDecorator(targetTest, enlistedClients, conversationFactory, connection);
+    }
+
+
+    /**
+     * The entry point for the interop test coordinator. This client accepts the following command line arguments:
+     *
+     * <p/><table> <tr><td> -b         <td> The broker URL.   <td> Mandatory. <tr><td> -h         <td> The virtual host.
+     * <td> Optional. <tr><td> name=value <td> Trailing argument define name/value pairs. Added to system properties.
+     * <td> Optional. </table>
+     *
+     * @param args The command line arguments.
+     */
+    public static void main(String[] args)
+    {
+        try
+        {
+            // Use the command line parser to evaluate the command line with standard handling behaviour (print errors
+            // and usage then exist if there are errors).
+            Properties options =
+                    CommandLineParser.processCommandLine(args,
+                                                         new CommandLineParser(
+                                                                 new String[][]
+                                                                         {
+                                                                                 {"b", "The broker URL.", "broker", "false"},
+                                                                                 {"h", "The virtual host to use.", "virtual host", "false"},
+                                                                                 {"o", "The name of the directory to output test timings to.", "dir", "false"}
+                                                                         }));
+
+            // Extract the command line options.
+            String brokerUrl = options.getProperty("b");
+            String virtualHost = options.getProperty("h");
+            String reportDir = options.getProperty("o");
+            reportDir = (reportDir == null) ? "." : reportDir;
+
+
+            String[] testClassNames = {SustainedTestCoordinator.class.getName()};
+
+            // Create a coordinator and begin its test procedure.
+            Coordinator coordinator = new TestCoordinator(brokerUrl, virtualHost);
+
+            coordinator.setReportDir(reportDir);
+
+            TestResult testResult = coordinator.start(testClassNames);
+
+            if (testResult.failureCount() > 0)
+            {
+                System.exit(FAILURE_EXIT);
+            }
+            else
+            {
+                System.exit(SUCCESS_EXIT);
+            }
+        }
+        catch (Exception e)
+        {
+            System.err.println(e.getMessage());
+            log.error("Top level handler caught execption.", e);
+            System.exit(EXCEPTION_EXIT);
+        }
+    }
+}