From d026ee76c093fe93afe15053d099e6ecc68a64ff Mon Sep 17 00:00:00 2001 From: Martin Ritchie Date: Tue, 24 Apr 2007 09:01:58 +0000 Subject: Merged revisions 1-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-531832 via svnmerge from https://svn.apache.org/repos/asf/incubator/qpid/branches/M2 ........ r530838 | ritchiem | 2007-04-20 17:03:10 +0100 (Fri, 20 Apr 2007) | 2 lines Added addition log when broker is ready to process. Updated HeapExhaustion - to use system.out.println so you can set amqj.logging.level=warn to speed up test. ........ r531437 | rgreig | 2007-04-23 12:23:39 +0100 (Mon, 23 Apr 2007) | 1 line QPID-394. Removed revision tags. ........ r531456 | bhupendrab | 2007-04-23 13:58:24 +0100 (Mon, 23 Apr 2007) | 1 line QPID-470 : Catching the security exception for users, which have no access to management console ........ r531458 | bhupendrab | 2007-04-23 14:13:28 +0100 (Mon, 23 Apr 2007) | 1 line QPID-445 : Fixed the ClassCastException ........ r531512 | ritchiem | 2007-04-23 16:52:43 +0100 (Mon, 23 Apr 2007) | 3 lines QPID-472 - Creation of TemporaryQueues will not guarantee unique queue names if created rapidly. Updated TemporaryQueueTest.java so that it checks Headers/Queue/Topic for unroutable/mandatory messages beig returned. ........ r531513 | ritchiem | 2007-04-23 16:54:15 +0100 (Mon, 23 Apr 2007) | 1 line QPID-436 - topic exchange doesn't obey the mandatory flag ........ r531515 | ritchiem | 2007-04-23 16:58:04 +0100 (Mon, 23 Apr 2007) | 2 lines Update to system test so that the run as part of the build process as they were not running. Change to AMQMessage to ensure that the TxAckTest passes. Was failing as the reference count was being changed out of the increment/decrementReference methods ........ r531517 | ritchiem | 2007-04-23 16:59:59 +0100 (Mon, 23 Apr 2007) | 1 line Comment updates on the origin of the tests ........ r531518 | ritchiem | 2007-04-23 17:02:41 +0100 (Mon, 23 Apr 2007) | 1 line White Space changes ........ r531524 | bhupendrab | 2007-04-23 17:28:00 +0100 (Mon, 23 Apr 2007) | 1 line QPID-445 ........ r531526 | ritchiem | 2007-04-23 17:38:24 +0100 (Mon, 23 Apr 2007) | 3 lines QPID-290 - Java broker does not honor maximum number of channels threshold Applied patch from Nuno Santos ........ r531527 | ritchiem | 2007-04-23 17:38:44 +0100 (Mon, 23 Apr 2007) | 1 line ResetMessageListenerTest - needs to have IMMEDIATE_PREFETCH = true. ........ git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@531842 13f79535-47bb-0310-9956-ffa450edef68 --- java/systests/pom.xml | 4 +- .../java/org/apache/qpid/server/ack/TxAckTest.java | 22 +- .../qpid/server/channel/MaxChannelsTest.java | 75 ++++++ .../ReturnUnroutableMandatoryMessageTest.java | 140 ++++++++++- .../apache/qpid/server/failure/HeapExhaustion.java | 6 +- .../apache/qpid/server/queue/ConcurrencyTest.java | 265 -------------------- .../qpid/server/queue/ConcurrencyTestDisabled.java | 265 ++++++++++++++++++++ .../qpid/server/queue/MessageTestHelper.java | 3 +- .../apache/qpid/server/queue/PersistentTest.java | 276 --------------------- .../qpid/server/queue/PersistentTestManual.java | 276 +++++++++++++++++++++ .../main/java/org/apache/qpid/test/VMTestCase.java | 5 + .../apache/qpid/test/client/QueueBrowserTest.java | 6 +- 12 files changed, 789 insertions(+), 554 deletions(-) create mode 100644 java/systests/src/main/java/org/apache/qpid/server/channel/MaxChannelsTest.java delete mode 100644 java/systests/src/main/java/org/apache/qpid/server/queue/ConcurrencyTest.java create mode 100644 java/systests/src/main/java/org/apache/qpid/server/queue/ConcurrencyTestDisabled.java delete mode 100644 java/systests/src/main/java/org/apache/qpid/server/queue/PersistentTest.java create mode 100644 java/systests/src/main/java/org/apache/qpid/server/queue/PersistentTestManual.java (limited to 'java/systests') diff --git a/java/systests/pom.xml b/java/systests/pom.xml index 614166754c..d9d07ed6f9 100644 --- a/java/systests/pom.xml +++ b/java/systests/pom.xml @@ -62,7 +62,8 @@ org.apache.maven.plugins maven-surefire-plugin - true + ${basedir}/src/main + target/classes @@ -89,3 +90,4 @@ + diff --git a/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java b/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java index 9fcd88b1a8..3ee8277eba 100644 --- a/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java +++ b/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java @@ -24,6 +24,8 @@ import junit.framework.TestCase; import org.apache.qpid.AMQException; import org.apache.qpid.framing.BasicPublishBody; import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.framing.ContentHeaderBody; +import org.apache.qpid.framing.AMQFrameDecodingException; import org.apache.qpid.framing.abstraction.MessagePublishInfo; import org.apache.qpid.server.RequiredDeliveryException; import org.apache.qpid.server.queue.AMQMessage; @@ -102,7 +104,7 @@ public class TxAckTest extends TestCase _storeContext, null, new LinkedList(), new HashSet()); - for(int i = 0; i < messageCount; i++) + for (int i = 0; i < messageCount; i++) { long deliveryTag = i + 1; @@ -144,7 +146,7 @@ public class TxAckTest extends TestCase private void assertCount(List tags, int expected) { - for(long tag : tags) + for (long tag : tags) { UnacknowledgedMessage u = _map.get(tag); assertTrue("Message not found for tag " + tag, u != null); @@ -161,6 +163,7 @@ public class TxAckTest extends TestCase assertCount(_unacked, 0); } + void undoPrepare() { _op.consolidate(); @@ -175,7 +178,6 @@ public class TxAckTest extends TestCase _op.consolidate(); _op.commit(_storeContext); - //check acked messages are removed from map Set keys = new HashSet(_map.getDeliveryTags()); keys.retainAll(_acked); @@ -195,6 +197,20 @@ public class TxAckTest extends TestCase TestMessage(long tag, long messageId, MessagePublishInfo publishBody, TransactionalContext txnContext) { super(messageId, publishBody, txnContext); + try + { + setContentHeaderBody(new ContentHeaderBody() + { + public int getSize() + { + return 1; + } + }); + } + catch (AMQException e) + { + // won't happen + } _tag = tag; } diff --git a/java/systests/src/main/java/org/apache/qpid/server/channel/MaxChannelsTest.java b/java/systests/src/main/java/org/apache/qpid/server/channel/MaxChannelsTest.java new file mode 100644 index 0000000000..5ddccb8a7b --- /dev/null +++ b/java/systests/src/main/java/org/apache/qpid/server/channel/MaxChannelsTest.java @@ -0,0 +1,75 @@ +/* + * + * Copyright (c) 2006 The Apache Software Foundation + * + * Licensed 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.server.protocol; + +import junit.framework.TestCase; +import org.apache.mina.common.IoSession; +import org.apache.qpid.codec.AMQCodecFactory; +import org.apache.qpid.server.AMQChannel; +import org.apache.qpid.server.virtualhost.VirtualHost; +import org.apache.qpid.server.registry.IApplicationRegistry; +import org.apache.qpid.server.registry.ApplicationRegistry; +import org.apache.qpid.server.exchange.ExchangeRegistry; +import org.apache.qpid.server.queue.QueueRegistry; +import org.apache.qpid.server.queue.AMQQueue; +import org.apache.qpid.server.store.MessageStore; +import org.apache.qpid.server.store.SkeletonMessageStore; +import org.apache.qpid.AMQException; +import org.apache.qpid.protocol.AMQConstant; +import org.apache.qpid.framing.AMQShortString; + +import javax.management.JMException; + +/** Test class to test MBean operations for AMQMinaProtocolSession. */ +public class MaxChannelsTest extends TestCase +{ +// private MessageStore _messageStore = new SkeletonMessageStore(); + + public void testChannels() throws Exception + { + IApplicationRegistry appRegistry = ApplicationRegistry.getInstance(); + AMQMinaProtocolSession _protocolSession = new AMQMinaProtocolSession(new MockIoSession(), + appRegistry.getVirtualHostRegistry(), + new AMQCodecFactory(true), + null); + _protocolSession.setVirtualHost(appRegistry.getVirtualHostRegistry().getVirtualHost("test")); + + // check the channel count is correct + int channelCount = _protocolSession.getChannels().size(); + assertEquals("Initial channel count wrong", 0, channelCount); + + long maxChannels = 10L; + _protocolSession.setMaximumNumberOfChannels(maxChannels); + assertEquals("Number of channels not correctly set.", new Long(maxChannels), _protocolSession.getMaximumNumberOfChannels()); + + + try + { + for (long currentChannel = 0L; currentChannel < maxChannels; currentChannel++) + { + _protocolSession.addChannel(new AMQChannel(_protocolSession, (int) currentChannel, null, null)); + } + } + catch (AMQException e) + { + assertEquals("Wrong exception recevied.", e.getErrorCode(), AMQConstant.NOT_ALLOWED); + } + assertEquals("Maximum number of channels not set.", new Long(maxChannels), new Long(_protocolSession.getChannels().size())); + } + +} diff --git a/java/systests/src/main/java/org/apache/qpid/server/exchange/ReturnUnroutableMandatoryMessageTest.java b/java/systests/src/main/java/org/apache/qpid/server/exchange/ReturnUnroutableMandatoryMessageTest.java index 87491ed3d3..ca352b2fd7 100644 --- a/java/systests/src/main/java/org/apache/qpid/server/exchange/ReturnUnroutableMandatoryMessageTest.java +++ b/java/systests/src/main/java/org/apache/qpid/server/exchange/ReturnUnroutableMandatoryMessageTest.java @@ -22,6 +22,8 @@ public class ReturnUnroutableMandatoryMessageTest extends TestCase implements Ex private static final Logger _logger = Logger.getLogger(ReturnUnroutableMandatoryMessageTest.class); private final List _bouncedMessageList = Collections.synchronizedList(new ArrayList()); + private static final String VIRTUALHOST = "test"; + private static final String BROKER = "vm://:1"; static { @@ -53,10 +55,10 @@ public class ReturnUnroutableMandatoryMessageTest extends TestCase implements Ex * * @throws Exception */ - public void testReturnUnroutableMandatoryMessage() throws Exception + public void testReturnUnroutableMandatoryMessage_HEADERS() throws Exception { _bouncedMessageList.clear(); - Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "/test"); + Connection con = new AMQConnection(BROKER, "guest", "guest", "consumer1", VIRTUALHOST); AMQSession consumerSession = (AMQSession) con.createSession(false, Session.CLIENT_ACKNOWLEDGE); @@ -70,7 +72,7 @@ public class ReturnUnroutableMandatoryMessageTest extends TestCase implements Ex //((AMQSession) consumerSession).declareExchangeSynch(ExchangeDefaults.HEADERS_EXCHANGE_NAME, ExchangeDefaults.HEADERS_EXCHANGE_CLASS); // This is the default now - Connection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "/test"); + Connection con2 = new AMQConnection(BROKER, "guest", "guest", "producer1", VIRTUALHOST); con2.setExceptionListener(this); AMQSession producerSession = (AMQSession) con2.createSession(false, Session.CLIENT_ACKNOWLEDGE); @@ -127,6 +129,138 @@ public class ReturnUnroutableMandatoryMessageTest extends TestCase implements Ex } + public void testReturnUnroutableMandatoryMessage_QUEUE() throws Exception + { + _bouncedMessageList.clear(); + Connection con = new AMQConnection(BROKER, "guest", "guest", "consumer1", VIRTUALHOST); + + + AMQSession consumerSession = (AMQSession) con.createSession(false, Session.CLIENT_ACKNOWLEDGE); + + AMQQueue valid_queue = new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_CLASS, "testReturnUnroutableMandatoryMessage_QUEUE"); + AMQQueue invalid_queue = new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_CLASS, "testReturnUnroutableMandatoryMessage_QUEUE_INVALID"); + MessageConsumer consumer = consumerSession.createConsumer(valid_queue); + + //force synch to ensure the consumer has resulted in a bound queue + //((AMQSession) consumerSession).declareExchangeSynch(ExchangeDefaults.HEADERS_EXCHANGE_NAME, ExchangeDefaults.HEADERS_EXCHANGE_CLASS); + // This is the default now + + Connection con2 = new AMQConnection(BROKER, "guest", "guest", "producer1", VIRTUALHOST); + + con2.setExceptionListener(this); + AMQSession producerSession = (AMQSession) con2.createSession(false, Session.CLIENT_ACKNOWLEDGE); + + // Need to start the "producer" connection in order to receive bounced messages + _logger.info("Starting producer connection"); + con2.start(); + + + MessageProducer nonMandatoryProducer = producerSession.createProducer(valid_queue, false, false); + MessageProducer mandatoryProducer = producerSession.createProducer(invalid_queue); + + // First test - should be routed + _logger.info("Sending non-mandatory message"); + TextMessage msg1 = producerSession.createTextMessage("msg1"); + nonMandatoryProducer.send(msg1); + + // Second test - should be bounced + _logger.info("Sending non-routable mandatory message"); + TextMessage msg2 = producerSession.createTextMessage("msg2"); + mandatoryProducer.send(msg2); + + + _logger.info("Starting consumer connection"); + con.start(); + TextMessage tm = (TextMessage) consumer.receive(1000L); + + assertTrue("No message routed to receiver", tm != null); + assertTrue("Wrong message routed to receiver: " + tm.getText(), "msg1".equals(tm.getText())); + + try + { + Thread.sleep(1000L); + } + catch (InterruptedException e) + { + ; + } + + assertTrue("Wrong number of messages bounced (expect 1): " + _bouncedMessageList.size(), _bouncedMessageList.size() == 1); + Message m = _bouncedMessageList.get(0); + assertTrue("Wrong message bounced: " + m.toString(), m.toString().contains("msg2")); + + + con.close(); + con2.close(); + } + + + public void testReturnUnroutableMandatoryMessage_TOPIC() throws Exception + { + _bouncedMessageList.clear(); + Connection con = new AMQConnection(BROKER, "guest", "guest", "consumer1", VIRTUALHOST); + + + AMQSession consumerSession = (AMQSession) con.createSession(false, Session.CLIENT_ACKNOWLEDGE); + + AMQTopic valid_topic = new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_CLASS, "test.Return.Unroutable.Mandatory.Message.TOPIC"); + AMQTopic invalid_topic = new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_CLASS, "test.Return.Unroutable.Mandatory.Message.TOPIC.invalid"); + MessageConsumer consumer = consumerSession.createConsumer(valid_topic); + + //force synch to ensure the consumer has resulted in a bound queue + //((AMQSession) consumerSession).declareExchangeSynch(ExchangeDefaults.HEADERS_EXCHANGE_NAME, ExchangeDefaults.HEADERS_EXCHANGE_CLASS); + // This is the default now + + Connection con2 = new AMQConnection(BROKER, "guest", "guest", "producer1", VIRTUALHOST); + + con2.setExceptionListener(this); + AMQSession producerSession = (AMQSession) con2.createSession(false, Session.CLIENT_ACKNOWLEDGE); + + // Need to start the "producer" connection in order to receive bounced messages + _logger.info("Starting producer connection"); + con2.start(); + + + MessageProducer nonMandatoryProducer = producerSession.createProducer(valid_topic, false, false); + MessageProducer mandatoryProducer = producerSession.createProducer(invalid_topic); + + // First test - should be routed + _logger.info("Sending non-mandatory message"); + TextMessage msg1 = producerSession.createTextMessage("msg1"); + nonMandatoryProducer.send(msg1); + + // Second test - should be bounced + _logger.info("Sending non-routable mandatory message"); + TextMessage msg2 = producerSession.createTextMessage("msg2"); + mandatoryProducer.send(msg2); + + + _logger.info("Starting consumer connection"); + con.start(); + TextMessage tm = (TextMessage) consumer.receive(1000L); + + assertTrue("No message routed to receiver", tm != null); + assertTrue("Wrong message routed to receiver: " + tm.getText(), "msg1".equals(tm.getText())); + + try + { + Thread.sleep(1000L); + } + catch (InterruptedException e) + { + ; + } + + assertTrue("Wrong number of messages bounced (expect 1): " + _bouncedMessageList.size(), _bouncedMessageList.size() == 1); + Message m = _bouncedMessageList.get(0); + assertTrue("Wrong message bounced: " + m.toString(), m.toString().contains("msg2")); + + + con.close(); + con2.close(); + } + + public static junit.framework.Test suite() { return new junit.framework.TestSuite(ReturnUnroutableMandatoryMessageTest.class); diff --git a/java/systests/src/main/java/org/apache/qpid/server/failure/HeapExhaustion.java b/java/systests/src/main/java/org/apache/qpid/server/failure/HeapExhaustion.java index 52eb5414ff..228bd1ec6f 100644 --- a/java/systests/src/main/java/org/apache/qpid/server/failure/HeapExhaustion.java +++ b/java/systests/src/main/java/org/apache/qpid/server/failure/HeapExhaustion.java @@ -16,7 +16,7 @@ public class HeapExhaustion extends TestCase { private static final Logger _logger = Logger.getLogger(HeapExhaustion.class); - protected QpidClientConnection conn; + protected QpidClientConnection conn; protected final String BROKER = "localhost"; protected final String vhost = "/test"; protected final String queue = "direct://amq.direct//queue"; @@ -65,7 +65,7 @@ public class HeapExhaustion extends TestCase conn.put(queue, payload, 1); copies++; total += size; - _logger.info("put copy " + copies + " OK for total bytes: " + total); + System.out.println("put copy " + copies + " OK for total bytes: " + total); } } @@ -85,7 +85,7 @@ public class HeapExhaustion extends TestCase conn.put(queue, payload, 1); copies++; total += size; - _logger.debug("put copy " + copies + " OK for total bytes: " + total); + System.out.println("put copy " + copies + " OK for total bytes: " + total); Thread.sleep(200); } } diff --git a/java/systests/src/main/java/org/apache/qpid/server/queue/ConcurrencyTest.java b/java/systests/src/main/java/org/apache/qpid/server/queue/ConcurrencyTest.java deleted file mode 100644 index 4971db2d28..0000000000 --- a/java/systests/src/main/java/org/apache/qpid/server/queue/ConcurrencyTest.java +++ /dev/null @@ -1,265 +0,0 @@ -/* - * - * 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.server.queue; - -import org.apache.qpid.AMQException; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.server.handler.OnCurrentThreadExecutor; -import org.apache.qpid.server.registry.IApplicationRegistry; -import org.apache.qpid.server.registry.ApplicationRegistry; -import org.apache.qpid.server.virtualhost.VirtualHost; - -import java.util.*; -import java.util.concurrent.Executor; - -/** - * Tests delivery in the face of concurrent incoming _messages, subscription alterations - * and attempts to asynchronously process queued _messages. - */ -public class ConcurrencyTest extends MessageTestHelper -{ - private final Random random = new Random(); - - private final int numMessages = 1000; - - private final List _subscribers = new ArrayList(); - private final Set _active = new HashSet(); - private final List _messages = new ArrayList(); - private int next = 0;//index to next message to send - private final List _received = Collections.synchronizedList(new ArrayList()); - private final Executor _executor = new OnCurrentThreadExecutor(); - private final List _threads = new ArrayList(); - - private final SubscriptionSet _subscriptionMgr = new SubscriptionSet(); - private final DeliveryManager _deliveryMgr; - - private boolean isComplete; - private boolean failed; - private VirtualHost _virtualHost; - - public ConcurrencyTest() throws Exception - { - - IApplicationRegistry applicationRegistry = ApplicationRegistry.getInstance(); - _virtualHost = applicationRegistry.getVirtualHostRegistry().getVirtualHost("test"); - _deliveryMgr = new ConcurrentSelectorDeliveryManager(_subscriptionMgr, new AMQQueue(new AMQShortString("myQ"), false, new AMQShortString("guest"), false, - _virtualHost)); - } - - public void testConcurrent1() throws InterruptedException, AMQException - { - initSubscriptions(10); - initMessages(numMessages); - initThreads(1, 4, 4, 4); - doRun(); - check(); - } - - public void testConcurrent2() throws InterruptedException, AMQException - { - initSubscriptions(10); - initMessages(numMessages); - initThreads(4, 2, 2, 2); - doRun(); - check(); - } - - void check() - { - assertFalse("Failed", failed); - - _deliveryMgr.processAsync(_executor); - - assertEquals("Did not recieve the correct number of messages", _messages.size(), _received.size()); - for(int i = 0; i < _messages.size(); i++) - { - assertEquals("Wrong message at " + i, _messages.get(i), _received.get(i)); - } - } - - void initSubscriptions(int subscriptions) - { - for(int i = 0; i < subscriptions; i++) - { - _subscribers.add(new SubscriptionTestHelper("Subscriber" + i, _received)); - } - } - - void initMessages(int messages) throws AMQException - { - for(int i = 0; i < messages; i++) - { - _messages.add(message()); - } - } - - void initThreads(int senders, int subscribers, int suspenders, int processors) - { - addThreads(senders, senders == 1 ? new Sender() : new OrderedSender()); - addThreads(subscribers, new Subscriber()); - addThreads(suspenders, new Suspender()); - addThreads(processors, new Processor()); - } - - void addThreads(int count, Runnable runner) - { - for(int i = 0; i < count; i++) - { - _threads.add(new Thread(runner, runner.toString())); - } - } - - void doRun() throws InterruptedException - { - for(Thread t : _threads) - { - t.start(); - } - - for(Thread t : _threads) - { - t.join(); - } - } - - private void toggle(Subscription s) - { - synchronized (_active) - { - if (_active.contains(s)) - { - _active.remove(s); - Subscription result = _subscriptionMgr.removeSubscriber(s); - assertTrue("Removed subscription " + result + " but trying to remove subscription " + s, - result != null && result.equals(s)); - } - else - { - _active.add(s); - _subscriptionMgr.addSubscriber(s); - } - } - } - - private AMQMessage nextMessage() - { - synchronized (_messages) - { - if (next < _messages.size()) - { - return _messages.get(next++); - } - else - { - if (!_deliveryMgr.hasQueuedMessages()) { - isComplete = true; - } - return null; - } - } - } - - private boolean randomBoolean() - { - return random.nextBoolean(); - } - - private SubscriptionTestHelper randomSubscriber() - { - return _subscribers.get(random.nextInt(_subscribers.size())); - } - - private class Sender extends Runner - { - void doRun() throws Throwable - { - AMQMessage msg = nextMessage(); - if (msg != null) - { - _deliveryMgr.deliver(null, new AMQShortString(toString()), msg, false); - } - } - } - - private class OrderedSender extends Sender - { - synchronized void doRun() throws Throwable - { - super.doRun(); - } - } - - private class Suspender extends Runner - { - void doRun() throws Throwable - { - randomSubscriber().setSuspended(randomBoolean()); - } - } - - private class Subscriber extends Runner - { - void doRun() throws Throwable - { - toggle(randomSubscriber()); - } - } - - private class Processor extends Runner - { - void doRun() throws Throwable - { - _deliveryMgr.processAsync(_executor); - } - } - - private abstract class Runner implements Runnable - { - public void run() - { - try - { - while (!stop()) - { - doRun(); - } - } - catch (Throwable t) - { - failed = true; - t.printStackTrace(); - } - } - - abstract void doRun() throws Throwable; - - boolean stop() - { - return isComplete || failed; - } - } - - public static junit.framework.Test suite() - { - return new junit.framework.TestSuite(ConcurrencyTest.class); - } - -} diff --git a/java/systests/src/main/java/org/apache/qpid/server/queue/ConcurrencyTestDisabled.java b/java/systests/src/main/java/org/apache/qpid/server/queue/ConcurrencyTestDisabled.java new file mode 100644 index 0000000000..068f37574d --- /dev/null +++ b/java/systests/src/main/java/org/apache/qpid/server/queue/ConcurrencyTestDisabled.java @@ -0,0 +1,265 @@ +/* + * + * 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.server.queue; + +import org.apache.qpid.AMQException; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.server.handler.OnCurrentThreadExecutor; +import org.apache.qpid.server.registry.IApplicationRegistry; +import org.apache.qpid.server.registry.ApplicationRegistry; +import org.apache.qpid.server.virtualhost.VirtualHost; + +import java.util.*; +import java.util.concurrent.Executor; + +/** + * Tests delivery in the face of concurrent incoming _messages, subscription alterations + * and attempts to asynchronously process queued _messages. + */ +public class ConcurrencyTestDisabled extends MessageTestHelper +{ + private final Random random = new Random(); + + private final int numMessages = 1000; + + private final List _subscribers = new ArrayList(); + private final Set _active = new HashSet(); + private final List _messages = new ArrayList(); + private int next = 0;//index to next message to send + private final List _received = Collections.synchronizedList(new ArrayList()); + private final Executor _executor = new OnCurrentThreadExecutor(); + private final List _threads = new ArrayList(); + + private final SubscriptionSet _subscriptionMgr = new SubscriptionSet(); + private final DeliveryManager _deliveryMgr; + + private boolean isComplete; + private boolean failed; + private VirtualHost _virtualHost; + + public ConcurrencyTestDisabled() throws Exception + { + + IApplicationRegistry applicationRegistry = ApplicationRegistry.getInstance(); + _virtualHost = applicationRegistry.getVirtualHostRegistry().getVirtualHost("test"); + _deliveryMgr = new ConcurrentSelectorDeliveryManager(_subscriptionMgr, new AMQQueue(new AMQShortString("myQ"), false, new AMQShortString("guest"), false, + _virtualHost)); + } + + public void testConcurrent1() throws InterruptedException, AMQException + { + initSubscriptions(10); + initMessages(numMessages); + initThreads(1, 4, 4, 4); + doRun(); + check(); + } + + public void testConcurrent2() throws InterruptedException, AMQException + { + initSubscriptions(10); + initMessages(numMessages); + initThreads(4, 2, 2, 2); + doRun(); + check(); + } + + void check() + { + assertFalse("Failed", failed); + + _deliveryMgr.processAsync(_executor); + + assertEquals("Did not recieve the correct number of messages", _messages.size(), _received.size()); + for(int i = 0; i < _messages.size(); i++) + { + assertEquals("Wrong message at " + i, _messages.get(i), _received.get(i)); + } + } + + void initSubscriptions(int subscriptions) + { + for(int i = 0; i < subscriptions; i++) + { + _subscribers.add(new SubscriptionTestHelper("Subscriber" + i, _received)); + } + } + + void initMessages(int messages) throws AMQException + { + for(int i = 0; i < messages; i++) + { + _messages.add(message()); + } + } + + void initThreads(int senders, int subscribers, int suspenders, int processors) + { + addThreads(senders, senders == 1 ? new Sender() : new OrderedSender()); + addThreads(subscribers, new Subscriber()); + addThreads(suspenders, new Suspender()); + addThreads(processors, new Processor()); + } + + void addThreads(int count, Runnable runner) + { + for(int i = 0; i < count; i++) + { + _threads.add(new Thread(runner, runner.toString())); + } + } + + void doRun() throws InterruptedException + { + for(Thread t : _threads) + { + t.start(); + } + + for(Thread t : _threads) + { + t.join(); + } + } + + private void toggle(Subscription s) + { + synchronized (_active) + { + if (_active.contains(s)) + { + _active.remove(s); + Subscription result = _subscriptionMgr.removeSubscriber(s); + assertTrue("Removed subscription " + result + " but trying to remove subscription " + s, + result != null && result.equals(s)); + } + else + { + _active.add(s); + _subscriptionMgr.addSubscriber(s); + } + } + } + + private AMQMessage nextMessage() + { + synchronized (_messages) + { + if (next < _messages.size()) + { + return _messages.get(next++); + } + else + { + if (!_deliveryMgr.hasQueuedMessages()) { + isComplete = true; + } + return null; + } + } + } + + private boolean randomBoolean() + { + return random.nextBoolean(); + } + + private SubscriptionTestHelper randomSubscriber() + { + return _subscribers.get(random.nextInt(_subscribers.size())); + } + + private class Sender extends Runner + { + void doRun() throws Throwable + { + AMQMessage msg = nextMessage(); + if (msg != null) + { + _deliveryMgr.deliver(null, new AMQShortString(toString()), msg, false); + } + } + } + + private class OrderedSender extends Sender + { + synchronized void doRun() throws Throwable + { + super.doRun(); + } + } + + private class Suspender extends Runner + { + void doRun() throws Throwable + { + randomSubscriber().setSuspended(randomBoolean()); + } + } + + private class Subscriber extends Runner + { + void doRun() throws Throwable + { + toggle(randomSubscriber()); + } + } + + private class Processor extends Runner + { + void doRun() throws Throwable + { + _deliveryMgr.processAsync(_executor); + } + } + + private abstract class Runner implements Runnable + { + public void run() + { + try + { + while (!stop()) + { + doRun(); + } + } + catch (Throwable t) + { + failed = true; + t.printStackTrace(); + } + } + + abstract void doRun() throws Throwable; + + boolean stop() + { + return isComplete || failed; + } + } + + public static junit.framework.Test suite() + { + return new junit.framework.TestSuite(ConcurrencyTestDisabled.class); + } + +} diff --git a/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java b/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java index 03a56df487..88272023e8 100644 --- a/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java +++ b/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java @@ -29,6 +29,7 @@ import org.apache.qpid.server.store.SkeletonMessageStore; import org.apache.qpid.server.store.StoreContext; import org.apache.qpid.server.registry.ApplicationRegistry; import org.apache.qpid.server.util.TestApplicationRegistry; +import org.apache.qpid.server.util.NullApplicationRegistry; import org.apache.qpid.server.txn.TransactionalContext; import org.apache.qpid.server.txn.NonTransactionalContext; import org.apache.qpid.server.RequiredDeliveryException; @@ -51,7 +52,7 @@ class MessageTestHelper extends TestCase MessageTestHelper() throws Exception { - ApplicationRegistry.initialise(new TestApplicationRegistry()); + ApplicationRegistry.initialise(new NullApplicationRegistry()); } AMQMessage message() throws AMQException diff --git a/java/systests/src/main/java/org/apache/qpid/server/queue/PersistentTest.java b/java/systests/src/main/java/org/apache/qpid/server/queue/PersistentTest.java deleted file mode 100644 index 4ad10b68ff..0000000000 --- a/java/systests/src/main/java/org/apache/qpid/server/queue/PersistentTest.java +++ /dev/null @@ -1,276 +0,0 @@ -/* - * 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.server.queue; - -import org.apache.qpid.client.AMQConnection; -import org.apache.qpid.client.AMQSession; -import org.apache.qpid.AMQException; -import org.apache.qpid.AMQChannelClosedException; -import org.apache.qpid.AMQConnectionClosedException; -import org.apache.qpid.util.CommandLineParser; -import org.apache.qpid.url.URLSyntaxException; -import org.apache.log4j.Logger; - -import javax.jms.Session; -import javax.jms.JMSException; -import javax.jms.Queue; -import javax.jms.MessageProducer; -import javax.jms.TextMessage; -import java.io.IOException; -import java.util.Properties; - -public class PersistentTest -{ - private static final Logger _logger = Logger.getLogger(PersistentTest.class); - - - private static final String QUEUE = "direct://amq.direct//PersistentTest-Queue2?durable='true',exclusive='true'"; - - protected AMQConnection _connection; - - protected Session _session; - - protected Queue _queue; - private Properties properties; - - private String _brokerDetails; - private String _username; - private String _password; - private String _virtualpath; - - public PersistentTest(Properties overrides) - { - properties = new Properties(defaults); - properties.putAll(overrides); - - _brokerDetails = properties.getProperty(BROKER_PROPNAME); - _username = properties.getProperty(USERNAME_PROPNAME); - _password = properties.getProperty(PASSWORD_PROPNAME); - _virtualpath = properties.getProperty(VIRTUAL_HOST_PROPNAME); - - createConnection(); - } - - protected void createConnection() - { - try - { - _connection = new AMQConnection(_brokerDetails, _username, _password, "PersistentTest", _virtualpath); - - _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE); - - _connection.start(); - } - catch (Exception e) - { - _logger.error("Unable to create test class due to:" + e.getMessage(), e); - System.exit(0); - } - } - - public void test() throws AMQException, URLSyntaxException - { - - //Create the Durable Queue - try - { - _session.createConsumer(_session.createQueue(QUEUE)).close(); - } - catch (JMSException e) - { - _logger.error("Unable to create Queue due to:" + e.getMessage(), e); - System.exit(0); - } - - try - { - if (testQueue()) - { - // close connection - _connection.close(); - // wait - System.out.println("Restart Broker Now"); - try - { - System.in.read(); - } - catch (IOException e) - { - // - } - finally - { - System.out.println("Continuing...."); - } - - //Test queue is still there. - AMQConnection connection = new AMQConnection(_brokerDetails, _username, _password, "DifferentClientID", _virtualpath); - - AMQSession session = (AMQSession) connection.createSession(false, Session.AUTO_ACKNOWLEDGE); - - try - { - session.createConsumer(session.createQueue(QUEUE)); - _logger.error("Create consumer succeeded." + - " This shouldn't be allowed as this means the queue didn't exist when it should"); - - connection.close(); - - exit(); - } - catch (JMSException e) - { - try - { - connection.close(); - } - catch (JMSException cce) - { - if (cce.getLinkedException() instanceof AMQConnectionClosedException) - { - _logger.error("Channel Close Bug still present QPID-432, should see an 'Error closing session'"); - } - else - { - exit(cce); - } - } - - if (e.getLinkedException() instanceof AMQChannelClosedException) - { - _logger.info("AMQChannelClosedException received as expected"); - } - else - { - exit(e); - } - } - } - } - catch (JMSException e) - { - _logger.error("Unable to test Queue due to:" + e.getMessage(), e); - System.exit(0); - } - } - - private void exit(JMSException e) - { - _logger.error("JMSException received:" + e.getMessage()); - e.printStackTrace(); - exit(); - } - - private void exit() - { - try - { - _connection.close(); - } - catch (JMSException e) - { - // - } - System.exit(0); - } - - private boolean testQueue() throws JMSException - { - String TEST_TEXT = "init"; - - //Create a new session to send producer - Session session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE); - - Queue q = session.createQueue(QUEUE); - MessageProducer producer = session.createProducer(q); - - producer.send(session.createTextMessage(TEST_TEXT)); - - //create a new consumer on the original session - TextMessage m = (TextMessage) _session.createConsumer(q).receive(); - - - if ((m != null) && m.getText().equals(TEST_TEXT)) - { - return true; - } - else - { - _logger.error("Incorrect values returned from Queue Test:" + m); - System.exit(0); - return false; - } - } - - /** Holds the name of the property to get the test broker url from. */ - public static final String BROKER_PROPNAME = "broker"; - - /** Holds the default broker url for the test. */ - public static final String BROKER_DEFAULT = "tcp://localhost:5672"; - - /** Holds the name of the property to get the test broker virtual path. */ - public static final String VIRTUAL_HOST_PROPNAME = "virtualHost"; - - /** Holds the default virtual path for the test. */ - public static final String VIRTUAL_HOST_DEFAULT = ""; - - /** Holds the name of the property to get the broker access username from. */ - public static final String USERNAME_PROPNAME = "username"; - - /** Holds the default broker log on username. */ - public static final String USERNAME_DEFAULT = "guest"; - - /** Holds the name of the property to get the broker access password from. */ - public static final String PASSWORD_PROPNAME = "password"; - - /** Holds the default broker log on password. */ - public static final String PASSWORD_DEFAULT = "guest"; - - /** Holds the default configuration properties. */ - public static Properties defaults = new Properties(); - - static - { - defaults.setProperty(BROKER_PROPNAME, BROKER_DEFAULT); - defaults.setProperty(USERNAME_PROPNAME, USERNAME_DEFAULT); - defaults.setProperty(PASSWORD_PROPNAME, PASSWORD_DEFAULT); - defaults.setProperty(VIRTUAL_HOST_PROPNAME, VIRTUAL_HOST_DEFAULT); - } - - public static void main(String[] args) - { - PersistentTest test; - - Properties options = CommandLineParser.processCommandLine(args, new CommandLineParser(new String[][]{})); - - - test = new PersistentTest(options); - try - { - test.test(); - System.out.println("Test was successfull."); - } - catch (Exception e) - { - _logger.error("Unable to test due to:" + e.getMessage(), e); - } - } -} diff --git a/java/systests/src/main/java/org/apache/qpid/server/queue/PersistentTestManual.java b/java/systests/src/main/java/org/apache/qpid/server/queue/PersistentTestManual.java new file mode 100644 index 0000000000..5abbbd2aae --- /dev/null +++ b/java/systests/src/main/java/org/apache/qpid/server/queue/PersistentTestManual.java @@ -0,0 +1,276 @@ +/* + * 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.server.queue; + +import org.apache.qpid.client.AMQConnection; +import org.apache.qpid.client.AMQSession; +import org.apache.qpid.AMQException; +import org.apache.qpid.AMQChannelClosedException; +import org.apache.qpid.AMQConnectionClosedException; +import org.apache.qpid.util.CommandLineParser; +import org.apache.qpid.url.URLSyntaxException; +import org.apache.log4j.Logger; + +import javax.jms.Session; +import javax.jms.JMSException; +import javax.jms.Queue; +import javax.jms.MessageProducer; +import javax.jms.TextMessage; +import java.io.IOException; +import java.util.Properties; + +public class PersistentTestManual +{ + private static final Logger _logger = Logger.getLogger(PersistentTestManual.class); + + + private static final String QUEUE = "direct://amq.direct//PersistentTest-Queue2?durable='true',exclusive='true'"; + + protected AMQConnection _connection; + + protected Session _session; + + protected Queue _queue; + private Properties properties; + + private String _brokerDetails; + private String _username; + private String _password; + private String _virtualpath; + + public PersistentTestManual(Properties overrides) + { + properties = new Properties(defaults); + properties.putAll(overrides); + + _brokerDetails = properties.getProperty(BROKER_PROPNAME); + _username = properties.getProperty(USERNAME_PROPNAME); + _password = properties.getProperty(PASSWORD_PROPNAME); + _virtualpath = properties.getProperty(VIRTUAL_HOST_PROPNAME); + + createConnection(); + } + + protected void createConnection() + { + try + { + _connection = new AMQConnection(_brokerDetails, _username, _password, "PersistentTest", _virtualpath); + + _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + + _connection.start(); + } + catch (Exception e) + { + _logger.error("Unable to create test class due to:" + e.getMessage(), e); + System.exit(0); + } + } + + public void test() throws AMQException, URLSyntaxException + { + + //Create the Durable Queue + try + { + _session.createConsumer(_session.createQueue(QUEUE)).close(); + } + catch (JMSException e) + { + _logger.error("Unable to create Queue due to:" + e.getMessage(), e); + System.exit(0); + } + + try + { + if (testQueue()) + { + // close connection + _connection.close(); + // wait + System.out.println("Restart Broker Now"); + try + { + System.in.read(); + } + catch (IOException e) + { + // + } + finally + { + System.out.println("Continuing...."); + } + + //Test queue is still there. + AMQConnection connection = new AMQConnection(_brokerDetails, _username, _password, "DifferentClientID", _virtualpath); + + AMQSession session = (AMQSession) connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + + try + { + session.createConsumer(session.createQueue(QUEUE)); + _logger.error("Create consumer succeeded." + + " This shouldn't be allowed as this means the queue didn't exist when it should"); + + connection.close(); + + exit(); + } + catch (JMSException e) + { + try + { + connection.close(); + } + catch (JMSException cce) + { + if (cce.getLinkedException() instanceof AMQConnectionClosedException) + { + _logger.error("Channel Close Bug still present QPID-432, should see an 'Error closing session'"); + } + else + { + exit(cce); + } + } + + if (e.getLinkedException() instanceof AMQChannelClosedException) + { + _logger.info("AMQChannelClosedException received as expected"); + } + else + { + exit(e); + } + } + } + } + catch (JMSException e) + { + _logger.error("Unable to test Queue due to:" + e.getMessage(), e); + System.exit(0); + } + } + + private void exit(JMSException e) + { + _logger.error("JMSException received:" + e.getMessage()); + e.printStackTrace(); + exit(); + } + + private void exit() + { + try + { + _connection.close(); + } + catch (JMSException e) + { + // + } + System.exit(0); + } + + private boolean testQueue() throws JMSException + { + String TEST_TEXT = "init"; + + //Create a new session to send producer + Session session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + + Queue q = session.createQueue(QUEUE); + MessageProducer producer = session.createProducer(q); + + producer.send(session.createTextMessage(TEST_TEXT)); + + //create a new consumer on the original session + TextMessage m = (TextMessage) _session.createConsumer(q).receive(); + + + if ((m != null) && m.getText().equals(TEST_TEXT)) + { + return true; + } + else + { + _logger.error("Incorrect values returned from Queue Test:" + m); + System.exit(0); + return false; + } + } + + /** Holds the name of the property to get the test broker url from. */ + public static final String BROKER_PROPNAME = "broker"; + + /** Holds the default broker url for the test. */ + public static final String BROKER_DEFAULT = "tcp://localhost:5672"; + + /** Holds the name of the property to get the test broker virtual path. */ + public static final String VIRTUAL_HOST_PROPNAME = "virtualHost"; + + /** Holds the default virtual path for the test. */ + public static final String VIRTUAL_HOST_DEFAULT = ""; + + /** Holds the name of the property to get the broker access username from. */ + public static final String USERNAME_PROPNAME = "username"; + + /** Holds the default broker log on username. */ + public static final String USERNAME_DEFAULT = "guest"; + + /** Holds the name of the property to get the broker access password from. */ + public static final String PASSWORD_PROPNAME = "password"; + + /** Holds the default broker log on password. */ + public static final String PASSWORD_DEFAULT = "guest"; + + /** Holds the default configuration properties. */ + public static Properties defaults = new Properties(); + + static + { + defaults.setProperty(BROKER_PROPNAME, BROKER_DEFAULT); + defaults.setProperty(USERNAME_PROPNAME, USERNAME_DEFAULT); + defaults.setProperty(PASSWORD_PROPNAME, PASSWORD_DEFAULT); + defaults.setProperty(VIRTUAL_HOST_PROPNAME, VIRTUAL_HOST_DEFAULT); + } + + public static void main(String[] args) + { + PersistentTestManual test; + + Properties options = CommandLineParser.processCommandLine(args, new CommandLineParser(new String[][]{})); + + + test = new PersistentTestManual(options); + try + { + test.test(); + System.out.println("Test was successfull."); + } + catch (Exception e) + { + _logger.error("Unable to test due to:" + e.getMessage(), e); + } + } +} diff --git a/java/systests/src/main/java/org/apache/qpid/test/VMTestCase.java b/java/systests/src/main/java/org/apache/qpid/test/VMTestCase.java index 31fd77691d..540c91ddaf 100644 --- a/java/systests/src/main/java/org/apache/qpid/test/VMTestCase.java +++ b/java/systests/src/main/java/org/apache/qpid/test/VMTestCase.java @@ -114,4 +114,9 @@ public class VMTestCase extends TestCase TransportConnection.killVMBroker(1); super.tearDown(); } + + public void testDummyinVMTestCase() + { + // keep maven happy + } } diff --git a/java/systests/src/main/java/org/apache/qpid/test/client/QueueBrowserTest.java b/java/systests/src/main/java/org/apache/qpid/test/client/QueueBrowserTest.java index ac65eec979..ec9df8f1b3 100644 --- a/java/systests/src/main/java/org/apache/qpid/test/client/QueueBrowserTest.java +++ b/java/systests/src/main/java/org/apache/qpid/test/client/QueueBrowserTest.java @@ -20,8 +20,8 @@ */ package org.apache.qpid.test.client; -import org.apache.qpid.test.VMTestCase; import org.apache.log4j.Logger; +import org.apache.qpid.test.VMTestCase; import javax.jms.Queue; import javax.jms.ConnectionFactory; @@ -36,6 +36,8 @@ import javax.jms.QueueReceiver; import javax.jms.Message; import java.util.Enumeration; +import junit.framework.TestCase; + public class QueueBrowserTest extends VMTestCase { private static final Logger _logger = Logger.getLogger(QueueBrowserTest.class); @@ -87,7 +89,7 @@ public class QueueBrowserTest extends VMTestCase * */ - public void queueBrowserMsgsRemainOnQueueTest() throws JMSException + public void testQueueBrowserMsgsRemainOnQueue() throws JMSException { // create QueueBrowser -- cgit v1.2.1