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-521792,521794-522462,522464-522527,522529-522534,522536-522566,522568-522993,522995-523244,523246-525530,525532,525534,525537-526149,526151-526682,526686-526713,526715-530399 via svnmerge from

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

........
  r521682 | bhupendrab | 2007-03-23 11:50:55 +0000 (Fri, 23 Mar 2007) | 2 lines
  
  QPID-418 (merged from trunk)
  svn merge -r521336:521345 https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid/java .
........
  r521705 | rgreig | 2007-03-23 12:44:14 +0000 (Fri, 23 Mar 2007) | 1 line
  
  Updates to performance tests.
........
  r521710 | ritchiem | 2007-03-23 12:59:18 +0000 (Fri, 23 Mar 2007) | 2 lines
  
  QPID-419 Access Control QPID-423 Authentication per virtualhost
  Improved error handling when hostconfig is not specifed. Was NPE-ing
........
  r521715 | ritchiem | 2007-03-23 13:10:33 +0000 (Fri, 23 Mar 2007) | 2 lines
  
  QPID-423 Authentication per virtualhost
  Improved error handling when hostconfig is not specifed. Was NPE-ing
........
  r521782 | bhupendrab | 2007-03-23 16:02:51 +0000 (Fri, 23 Mar 2007) | 1 line
  
  QPID-420 (merged from trunk)  And r518998:518999 and r520846:520850
........
  r522959 | ritchiem | 2007-03-27 16:39:51 +0100 (Tue, 27 Mar 2007) | 2 lines
  
  Updated assembly/bin.xml to include transient_config.xml persistent_config.xml 
........
  r522961 | ritchiem | 2007-03-27 16:42:40 +0100 (Tue, 27 Mar 2007) | 1 line
  
  forgot to rename files after they had been copied
........
  r522967 | ritchiem | 2007-03-27 16:56:03 +0100 (Tue, 27 Mar 2007) | 1 line
  
  correctly renamed transient and persistent config files
........
  r522977 | ritchiem | 2007-03-27 17:06:15 +0100 (Tue, 27 Mar 2007) | 1 line
  
  updated config files
........
  r522981 | ritchiem | 2007-03-27 17:10:45 +0100 (Tue, 27 Mar 2007) | 1 line
  
  Added echo of progress and reduced timeout
........
  r522989 | ritchiem | 2007-03-27 17:33:04 +0100 (Tue, 27 Mar 2007) | 1 line
  
  fixed error where script wouldn't pickup running pids
........
  r522990 | ritchiem | 2007-03-27 17:36:34 +0100 (Tue, 27 Mar 2007) | 1 line
  
  Added additional logging and comments
........
  r522991 | ritchiem | 2007-03-27 17:37:17 +0100 (Tue, 27 Mar 2007) | 1 line
  
  Added additional  comments
........
  r523747 | rajith | 2007-03-29 16:32:56 +0100 (Thu, 29 Mar 2007) | 1 line
  
  Fix for setting the message id
........
  r524050 | rgreig | 2007-03-30 12:51:09 +0100 (Fri, 30 Mar 2007) | 1 line
  
  Removed excess logging to optimize performance.
........
  r524739 | ritchiem | 2007-04-02 08:29:06 +0100 (Mon, 02 Apr 2007) | 1 line
  
  Added BDB Test scripts and updated pom to contain same tests as were used in perftesting
........
  r524740 | ritchiem | 2007-04-02 08:47:29 +0100 (Mon, 02 Apr 2007) | 1 line
  
  Fixed error with passwordfile parameter
........
  r524743 | ritchiem | 2007-04-02 09:07:55 +0100 (Mon, 02 Apr 2007) | 1 line
  
  Added CTQ tests
........
  r524763 | ritchiem | 2007-04-02 11:50:06 +0100 (Mon, 02 Apr 2007) | 1 line
  
  Added verify password method to PrincipalDatabase
........
  r524765 | ritchiem | 2007-04-02 11:55:12 +0100 (Mon, 02 Apr 2007) | 1 line
  
  Moved broker details to a separate variable.
........
  r524767 | ritchiem | 2007-04-02 12:17:54 +0100 (Mon, 02 Apr 2007) | 1 line
  
  ignored idea files
........
  r525487 | ritchiem | 2007-04-04 11:42:59 +0100 (Wed, 04 Apr 2007) | 3 lines
  
  Added default timeout to AMQConnection.close();
........
  r525553 | ritchiem | 2007-04-04 17:34:35 +0100 (Wed, 04 Apr 2007) | 1 line
  
  Updated case of properties
........
  r525766 | ritchiem | 2007-04-05 09:51:55 +0100 (Thu, 05 Apr 2007) | 1 line
  
  QPID-308 Added test case to demonstrate heap exhaustion of broker. Can't be run InVM as it kills the broker.
........
  r525777 | ritchiem | 2007-04-05 10:29:22 +0100 (Thu, 05 Apr 2007) | 20 lines
  
  QPID-414 : Addition of CRAM-MD5-HASHED authentication. Same as CRAM-MD5 but the client uses the hash of the password rather than the original password. This allows the broker to store the hash not the original password.
  
  Added initial tool for generation passwords.
  
  Broker:
  Renamed MD5PasswordFilePrincipalDatabase.java to Base64MD5PasswordFilePrincipalDatabase.java as that more accurately represents the file contents. 
  PlainPasswordVhostFilePrincipalDatabase.java - import tidy up
  PrincipalDatabaseAuthenticationManager.java - Changed to add our SASL providers at the start of the SASL list.
  CRAMMD5Hashed* - New SASL mechanism that delegates to CRAM-MD5 but understands that the password to use is the hash of the users password.
  JCAProvider - Removed the addProvider() line as this is done after the construction in PrincipalDatabaseAuthenticationManager.
  PlainSaslServerFactory - White Space
  Passwd.java - New util stub for managing passwords ala htpasswd.
  
  Client
  Added CRAM-MD5-HASHED to CallbackHandlerRegistry 
  Added ClientFactory for CRAMMD5Hashed that returns the first CRAM-MD5 SaslClient.
  DynamicSaslRegistrar.java - Tidied imports added new JCAProviders at the start of the Sasl lists.
  DynamicSaslRegistrar.properties - Added CRAM-MD5-HASHED handler.
  JCAProvider.java - as with broker stopped JCAProvider.java adding itself as the DynamicSaslRegistrar.java does this on the client.
  UsernameHashedPasswordCallbackHandler.java - New callback handler that is used by CRAM-MD5-HASHED. It hashes the client's password and uses that in the CRAM-MD5 algorithm.
........
  r525785 | ritchiem | 2007-04-05 10:48:43 +0100 (Thu, 05 Apr 2007) | 1 line
  
  Old ant folder
........
  r525786 | ritchiem | 2007-04-05 10:57:33 +0100 (Thu, 05 Apr 2007) | 1 line
  
  QPID-440 - added comments in the code relating to this bug.
........
  r525787 | ritchiem | 2007-04-05 10:58:20 +0100 (Thu, 05 Apr 2007) | 2 lines
  
  QPID-308 removed closeConnection() that calls close(-1) and may result in a client hang. better to call closeConnection(long timeout) so forced this my removing closeConnection();
........
  r525788 | ritchiem | 2007-04-05 11:00:56 +0100 (Thu, 05 Apr 2007) | 1 line
  
  QPID-414 update to config.xml to give usage example.
........
  r525804 | ritchiem | 2007-04-05 13:19:31 +0100 (Thu, 05 Apr 2007) | 1 line
  
  QPID-308 Updated HeapExhaustion to be able to be run from command line
........
  r525817 | ritchiem | 2007-04-05 14:14:50 +0100 (Thu, 05 Apr 2007) | 1 line
  
  Update to qpid stop scripts to properly check for existing broker instances and promptly stop them.
........
  r525829 | ritchiem | 2007-04-05 14:50:56 +0100 (Thu, 05 Apr 2007) | 1 line
  
  Updated scripts to work correctly under solaris and bash 2.0
........
  r525862 | rgodfrey | 2007-04-05 17:37:40 +0100 (Thu, 05 Apr 2007) | 1 line
  
  QPID-443 : Fix to transactionality of message publishing
........
  r525867 | ritchiem | 2007-04-05 17:47:59 +0100 (Thu, 05 Apr 2007) | 2 lines
  
  QPID-416 Provided simple update to Access Control via FileAccessManager to allow access rights for a virtualhost to be stored in a separate file.
  Updated PrincipalDatabaseAccessManager to use the default AccessManager if the specified PrincipalDatabase is not an AccessManager.
........
  r526091 | ritchiem | 2007-04-06 09:21:01 +0100 (Fri, 06 Apr 2007) | 5 lines
  
  QPID-416 Update to Access control to allow simply read/write permissions per Virtual host.
  access - updated file to have examples of access control.
  AccessManager - Deprecated old isAuthorised method
  Implemented new isAuthorized method on all AccessManagers
........
  r526113 | ritchiem | 2007-04-06 11:28:43 +0100 (Fri, 06 Apr 2007) | 1 line
  
  Updated case of properties to be true cammelCase and updated tests to run for a duration of 10 minutes rather than set message count. To provide better results for graphing.
........
  r526117 | ritchiem | 2007-04-06 11:42:11 +0100 (Fri, 06 Apr 2007) | 9 lines
  
  QPID-416 Update to Access control to allow simply read/write permissions per Virtual host.
  access - updated file to have examples of access control.
  Changed AMQProtocolSession to record an authorized Principal not just a String.
   - Required 
  Added AccessRights files needed for VirtualHostAccess control.
  
  Updated ConnectionOpenMethodHandler to allow Principals with any access to connect not just read.
  
  UsernamePrincipal - Added a toString
........
  r526118 | rgodfrey | 2007-04-06 11:55:17 +0100 (Fri, 06 Apr 2007) | 1 line
........
  r526122 | ritchiem | 2007-04-06 12:26:06 +0100 (Fri, 06 Apr 2007) | 1 line
  
  removed pauses between batches
........
  r526154 | rgodfrey | 2007-04-06 14:24:46 +0100 (Fri, 06 Apr 2007) | 1 line
  
  QPID-443 : Fix to transactionality of message publishing
........
  r526157 | bhupendrab | 2007-04-06 14:32:56 +0100 (Fri, 06 Apr 2007) | 1 line
  
  QPID-444   : Enabling the Qpid to use SASL. jmxmp can be plugged into for SASL. Can be configured to use security. 
........
  r526158 | ritchiem | 2007-04-06 14:34:52 +0100 (Fri, 06 Apr 2007) | 1 line
  
  Duplicate of BDB-Qpid.sh
........
  r526159 | bhupendrab | 2007-04-06 14:37:47 +0100 (Fri, 06 Apr 2007) | 1 line
  
  QPID-444  : adding jmxport, which is used when out of the box JMXAgent is not used
........
  r526166 | ritchiem | 2007-04-06 14:51:41 +0100 (Fri, 06 Apr 2007) | 1 line
  
  QPID-414 - Initial script to run the passwd gen. 
........
  r526187 | bhupendrab | 2007-04-06 15:53:36 +0100 (Fri, 06 Apr 2007) | 2 lines
  
  QPID-444   : Enabling the SASL support. jmxmp can be plugged into for SASL.
........
  r526194 | rgreig | 2007-04-06 16:21:19 +0100 (Fri, 06 Apr 2007) | 1 line
  
  Added some ramping up performance tests.
........
  r526195 | marnie | 2007-04-06 16:21:33 +0100 (Fri, 06 Apr 2007) | 1 line
  
  QPID-381 Amended session constructor to be non-transactional and use client ack mode.
........
  r526198 | rgreig | 2007-04-06 16:26:02 +0100 (Fri, 06 Apr 2007) | 1 line
  
  Fixed message sizes.
........
  r526199 | rgreig | 2007-04-06 16:29:06 +0100 (Fri, 06 Apr 2007) | 1 line
  
  Fixed commit batch size.
........
  r526666 | ritchiem | 2007-04-09 08:47:14 +0100 (Mon, 09 Apr 2007) | 1 line
  
  Updated so the FileAppender includes time stamps by default.. ConversionPattern made the same as STDOUT and RollingFileAppender
........
  r526691 | ritchiem | 2007-04-09 10:39:47 +0100 (Mon, 09 Apr 2007) | 1 line
  
  Added $@ to allow pass through of command line args to each sub process
........
  r526692 | bhupendrab | 2007-04-09 10:45:06 +0100 (Mon, 09 Apr 2007) | 4 lines
  
  QPID-444   : added log statements and some config parameters.
  Removed the autoDelete parameter from createNewQueue method used from Management Console.
........
  r526694 | bhupendrab | 2007-04-09 10:51:46 +0100 (Mon, 09 Apr 2007) | 1 line
........
  r526709 | bhupendrab | 2007-04-09 12:02:08 +0100 (Mon, 09 Apr 2007) | 2 lines
  
  QPID-444  : updated the management console dependency configuration for sasl support
........
  r526776 | rgreig | 2007-04-09 16:26:04 +0100 (Mon, 09 Apr 2007) | 1 line
  
  Stopped throwing away exception causes.
........
  r526803 | rgreig | 2007-04-09 17:09:24 +0100 (Mon, 09 Apr 2007) | 1 line
  
  Got rid of some uses of System.out instead of log4j logging.
........
  r526807 | rgreig | 2007-04-09 17:12:49 +0100 (Mon, 09 Apr 2007) | 1 line
  
  Got rid of some uses of System.out instead of log4j logging.
........
  r527049 | ritchiem | 2007-04-10 08:58:26 +0100 (Tue, 10 Apr 2007) | 1 line
  
  Moved bdb tests to bdbstore package
........
  r527050 | ritchiem | 2007-04-10 09:00:42 +0100 (Tue, 10 Apr 2007) | 1 line
  
  QueueDeclareHandler.java - Added more detail to error messages. Such as returning the queue name that was attempted to be declared but failed.
........
  r527053 | ritchiem | 2007-04-10 09:03:15 +0100 (Tue, 10 Apr 2007) | 1 line
  
  Added a test to check that Persistent Queues do actually persist.
........
  r527182 | ritchiem | 2007-04-10 17:29:47 +0100 (Tue, 10 Apr 2007) | 1 line
  
  QPID-446 Initial MBean framework.
........
  r527487 | ritchiem | 2007-04-11 14:31:18 +0100 (Wed, 11 Apr 2007) | 5 lines
  
  QPID-446  AMQUserManagementMBean Initial implementation of user management in authentication file.
  UserManagement - Added annotations for MBeanOperations
  PrincipalDatabase - Added new methods to update,create,delete Principal.
   - Implemented method on all PrincipalDatabase implementations, most return false to say not complete except 
  Base64MD5PasswordFilePrincipalDatabase - which now stores in memory the password file and flushes any changes to disk.
........
  r527493 | ritchiem | 2007-04-11 14:50:40 +0100 (Wed, 11 Apr 2007) | 1 line
  
  QPID-446  Missed the commit of JMXManagedObjectRegistry change on verifyPassword char[] to String
........
  r527499 | bhupendrab | 2007-04-11 15:16:02 +0100 (Wed, 11 Apr 2007) | 1 line
  
  QPID-444  : added CRAM-MD5-HASHED mechanism for sasl
........
  r527509 | bhupendrab | 2007-04-11 15:47:22 +0100 (Wed, 11 Apr 2007) | 1 line
........
  r527518 | ritchiem | 2007-04-11 16:21:37 +0100 (Wed, 11 Apr 2007) | 14 lines
  
  QPID-446 
  JMXManagedObjectRegistry - Split instantiation from starting up. To all the setting of the Access file when loaded later in the startup sequence.
  ManagedObjectRegistry - Added Start method
  MBeanInvocationHandlerImpl - Updated to allow the setting of the access properties object from the AMQUserManagementMBean
  NoopManagedObjectRegistry - implemented no-op start
  ConfigurationFileApplicationRegistry - Adjusted to split creation of ManagedObjectRegistry from starting server to allow the setting of access rights.
  AMQUserManagementMBean - Implemented reading of access rights file.
  Base64MD5PasswordFilePrincipalDatabase - added comment for future Management.
  PrincipalDatabaseManager - added initialiseManagement method
  ConfigurationFilePrincipalDatabaseManager - implemented general Management initialisation.
  PropertiesPrincipalDatabaseManager - no-op implementation
........
  r527537 | ritchiem | 2007-04-11 16:47:30 +0100 (Wed, 11 Apr 2007) | 2 lines
  
  QPID-446 Update to contain jmx config settings.
........
  r527556 | bhupendrab | 2007-04-11 17:07:58 +0100 (Wed, 11 Apr 2007) | 1 line
  
  synchronized with hash mechanism used in Broker
........
  r527557 | ritchiem | 2007-04-11 17:08:54 +0100 (Wed, 11 Apr 2007) | 1 line
  
  Fixed Bug in convertPassword where data wasn't correctly updated PropertiesPrincipalDatabase, 
........
  r527558 | ritchiem | 2007-04-11 17:09:54 +0100 (Wed, 11 Apr 2007) | 1 line
  
  QpiQPID-446 Update to ensure qpid.password file is correctly written in savePasswordFile
........
  r527803 | ritchiem | 2007-04-12 08:16:54 +0100 (Thu, 12 Apr 2007) | 5 lines
  
  QPID-446 Update to write accessRights file and correctly write Base64 MD5 Hashed password to password file.
  MBeanInvocationHandlerImpl - made statics ADMIN,READONLY,READWRITE public so they can be used in writing the access file.
  AMQUserManagementMBean - Update to write the access File.
  PrincipalDatabase - create getUser(username) to retrieve a Principal from the database this is then implemented in all PDs. Used to check for existence of a user.
........
  r527843 | ritchiem | 2007-04-12 09:52:19 +0100 (Thu, 12 Apr 2007) | 10 lines
  
  QPID-446 Update to send userList to JMX Management console.
  Currently niave implementation just sending ALL users in one go. If a LDAPPrincipalDatabase was created this could be quite a lot of data a) to send but b) to create in broker Heap.
  PrincipalDatabase - javadoc'd and getUsers method, 
   -changed verifyPassword method to take String for username rather than Principal only the Managment Console uses this method and it the MC should be changed to use the Broker SASL modules directly rather than having very similar ones of its own.
   - Removed AccountNotFound exception from createPrincipal as it made no sence
  No-op implementation in PlainPasswordFilePrincipalDatabase and PropertiesPrincipalDatabase
  Base64MD5PasswordFilePrincipalDatabase changed local User class to implement Principal so current Map can be returned via getUsers
   - Added locking to ensure integrity of files in the face of multiple edits.
........
  r527848 | ritchiem | 2007-04-12 10:11:19 +0100 (Thu, 12 Apr 2007) | 1 line
  
  QPID-446 Removed hashing of presented password in Base64MD5PasswordFilePrincipalDatabase. 
........
  r527876 | rgodfrey | 2007-04-12 11:31:51 +0100 (Thu, 12 Apr 2007) | 3 lines
  
  QPID-451 Throw InvalidDestinationException on attempt to publish to a Queue which does not exist
  Changed QueueSenderAdapter to check if the routing key is bound to a queue on the given exchange.
  The checking can be turned off by setting the system property org.apache.qpid.client.verifyQueueBindingBeforePublish to anything but true
........
  r527941 | bhupendrab | 2007-04-12 14:49:10 +0100 (Thu, 12 Apr 2007) | 1 line
  
  not needed for management console
........
  r527959 | bhupendrab | 2007-04-12 15:40:36 +0100 (Thu, 12 Apr 2007) | 1 line
  
  refining the mbean operations
........
  r527972 | ritchiem | 2007-04-12 16:11:16 +0100 (Thu, 12 Apr 2007) | 3 lines
  
  QPID-446
  Updated sample configs to contain jmx security options.
........
  r528003 | marnie | 2007-04-12 17:15:48 +0100 (Thu, 12 Apr 2007) | 1 line
  
  QPID-352 Changes
........
  r528005 | marnie | 2007-04-12 17:16:34 +0100 (Thu, 12 Apr 2007) | 1 line
  
  QPID-352 Changes
........
  r528424 | rgreig | 2007-04-13 11:17:12 +0100 (Fri, 13 Apr 2007) | 1 line
  
  Created new ping client that sends messages only. Usefull for examaning known queue states in mgmnt console.
........
  r529233 | bhupendrab | 2007-04-16 14:25:58 +0100 (Mon, 16 Apr 2007) | 1 line
  
  added parameter for SASL
........
  r529246 | bhupendrab | 2007-04-16 14:48:31 +0100 (Mon, 16 Apr 2007) | 1 line
  
  removed default username as guest. Added hashing for new user password field. 
........
  r529297 | rgodfrey | 2007-04-16 16:53:45 +0100 (Mon, 16 Apr 2007) | 1 line
  
  QPID-453 : AMQShortString should implement Comparable
........
  r529635 | bhupendrab | 2007-04-17 16:07:06 +0100 (Tue, 17 Apr 2007) | 1 line
  
  QPID-422 : Combined all user configured notifications on one view.
........
  r529659 | ritchiem | 2007-04-17 17:08:00 +0100 (Tue, 17 Apr 2007) | 7 lines
  
  QPID-454 Message 'taken' notion is per message. But should be per message per queue
  AMQChannel - pass queue in on all take/release/getSubscriptionDelievered calls
  BasicRejectMethodHandler - pass queue in on getSubscriptionDelievered calls
  AMQMessage - Changes to require AMQQueue on all take/release/getSubscriptionDelievered calls
  ConcurrentSelectorDeliveryManager - pass queue in on take/release/getSubscriptionDelievered calls
  SubscriptionImpl - - pass queue in on release calls
........
  r529666 | ritchiem | 2007-04-17 17:19:59 +0100 (Tue, 17 Apr 2007) | 11 lines
  
  QPID-455 Prefetched messages can cause problems with client tools.
  AMQSession - suspend channel at startup until start() and recieve/setMessageListener are called.
  BasicMessageConsumer - mainly style sheet changes
  MessageListenerMultiConsumerTest - removed one test case as we cannot ensure round-robin effect at start up .. added test case for only c2 consuming when c1 does nothing.
  MessageListenerTest - added new test that can demonstrate a further bug of message 'loss' when a receive is called only once before a message listener is set. Prefetched message end up on _SynchronousQueue regression of QPID-293 as of r501004.
  MessageRequeueTest - Was missing a conn.start()
  DurableSubscriptionTest - Removed blocking receives() so we don't block on failure
  CommitRollbackTest - Text message was wrong on testGetThenDisconnect tests so adjusted
........
  r529669 | bhupendrab | 2007-04-17 17:43:53 +0100 (Tue, 17 Apr 2007) | 1 line
  
  QPID-417
........
  r530034 | bhupendrab | 2007-04-18 15:32:02 +0100 (Wed, 18 Apr 2007) | 2 lines
  
  AMQUserManagementMBean.java - calling relaod within viewUsers method.
  Creating user list on management console instead of typing the user name.
........
  r530037 | ritchiem | 2007-04-18 15:37:30 +0100 (Wed, 18 Apr 2007) | 1 line
  
  QPID-454 Message 'taken' notion is per message. REVERTED as it just wasn't right.. needs to be refactored.
........
  r530041 | ritchiem | 2007-04-18 15:40:47 +0100 (Wed, 18 Apr 2007) | 1 line
  
  QPID-457 Fixed rollback inTran problem with test case
........
  r530042 | ritchiem | 2007-04-18 15:42:16 +0100 (Wed, 18 Apr 2007) | 1 line
  
  QPID-457 Fixed rollback inTran problem with test case Missed the actual file fix.
........
  r530043 | ritchiem | 2007-04-18 15:46:36 +0100 (Wed, 18 Apr 2007) | 1 line
  
  QPID-458 Fix to make the CSDM check if a message is taken when deliverying to browser. Removing the message from the queue and continuing if that is the caee.
........
  r530044 | ritchiem | 2007-04-18 15:54:36 +0100 (Wed, 18 Apr 2007) | 1 line
  
  Removed e.printstacktrace that sneaked in with the other code style changes.
........
  r530047 | ritchiem | 2007-04-18 16:09:28 +0100 (Wed, 18 Apr 2007) | 1 line
  
  Fix for intermittent CRT expected <1> but was <2> errors
........
  r530048 | ritchiem | 2007-04-18 16:10:24 +0100 (Wed, 18 Apr 2007) | 3 lines
  
  ResetMessageListenerTest was using the wrong queue for running tests. This was causing problems during testing.
  
  Changed queue to use ResetMessageListenerTest queue
........
  r530049 | ritchiem | 2007-04-18 16:11:22 +0100 (Wed, 18 Apr 2007) | 2 lines
  
  QPID-455 Prefetched messages can cause problems with client tools.
  Removed the changes as this was causing problems. Guarded with a check for now but solution is till not correct.
........
  r530052 | ritchiem | 2007-04-18 16:12:45 +0100 (Wed, 18 Apr 2007) | 1 line
  
  QPID-455 - Guarded test with a check until a full solution is found
........


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

35 files changed, 1038 insertions, 479 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java b/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
index 413524b6d8..0e3d99eeba 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
@@ -7,9 +7,9 @@
  * 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
@@ -20,6 +20,29 @@
  */
 package org.apache.qpid.client;
 
+import java.io.IOException;
+import java.net.ConnectException;
+import java.nio.channels.UnresolvedAddressException;
+import java.text.MessageFormat;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.jms.*;
+import javax.jms.IllegalStateException;
+import javax.naming.NamingException;
+import javax.naming.Reference;
+import javax.naming.Referenceable;
+import javax.naming.StringRefAddr;
+
+import org.apache.log4j.Logger;
+
 import org.apache.qpid.AMQConnectionFailureException;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.AMQUndeliveredException;
@@ -44,28 +67,6 @@ import org.apache.qpid.jms.ConnectionURL;
 import org.apache.qpid.jms.FailoverPolicy;
 import org.apache.qpid.url.URLSyntaxException;
 
-import org.apache.log4j.Logger;
-
-import javax.jms.*;
-import javax.jms.IllegalStateException;
-import javax.naming.NamingException;
-import javax.naming.Reference;
-import javax.naming.Referenceable;
-import javax.naming.StringRefAddr;
-import java.io.IOException;
-import java.net.ConnectException;
-import java.nio.channels.UnresolvedAddressException;
-import java.text.MessageFormat;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.LinkedHashMap;
-import java.util.LinkedList;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
-
 public class AMQConnection extends Closeable implements Connection, QueueConnection, TopicConnection, Referenceable
 {
     private static final Logger _logger = Logger.getLogger(AMQConnection.class);
@@ -95,7 +96,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
     private AMQProtocolHandler _protocolHandler;
 
     /** Maps from session id (Integer) to AMQSession instance */
-    private final Map _sessions = new LinkedHashMap(); //fixme this is map is replicated in amqprotocolsession as _channelId2SessionMap    
+    private final Map _sessions = new LinkedHashMap(); // fixme this is map is replicated in amqprotocolsession as _channelId2SessionMap
 
     private String _clientName;
 
@@ -125,15 +126,14 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
 
     /*
      * _Connected should be refactored with a suitable wait object.
-    */
+     */
     private boolean _connected;
 
     /*
      * The last error code that occured on the connection. Used to return the correct exception to the client
-    */
+     */
     private AMQException _lastAMQException = null;
 
-
     /*
      * The connection meta data
      */
@@ -149,6 +149,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
 
     /** Thread Pool for executing connection level processes. Such as returning bounced messages. */
     private final ExecutorService _taskPool = Executors.newCachedThreadPool();
+    private static final long DEFAULT_TIMEOUT = 1000 * 30;
 
     /**
      * @param broker      brokerdetails
@@ -160,13 +161,13 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
      * @throws AMQException
      * @throws URLSyntaxException
      */
-    public AMQConnection(String broker, String username, String password,
-                         String clientName, String virtualHost) throws AMQException, URLSyntaxException
+    public AMQConnection(String broker, String username, String password, String clientName, String virtualHost)
+        throws AMQException, URLSyntaxException
     {
-        this(new AMQConnectionURL(ConnectionURL.AMQ_PROTOCOL + "://" +
-                                  username + ":" + password + "@" +
-                                  (clientName == null ? "" : clientName) + "/" +
-                                  virtualHost + "?brokerlist='" + AMQBrokerDetails.checkTransport(broker) + "'"), null);
+        this(new AMQConnectionURL(
+                ConnectionURL.AMQ_PROTOCOL + "://" + username + ":" + password + "@"
+                + ((clientName == null) ? "" : clientName) + "/" + virtualHost + "?brokerlist='"
+                + AMQBrokerDetails.checkTransport(broker) + "'"), null);
     }
 
     /**
@@ -179,44 +180,38 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
      * @throws AMQException
      * @throws URLSyntaxException
      */
-    public AMQConnection(String broker, String username, String password,
-                         String clientName, String virtualHost, SSLConfiguration sslConfig) throws AMQException, URLSyntaxException
+    public AMQConnection(String broker, String username, String password, String clientName, String virtualHost,
+        SSLConfiguration sslConfig) throws AMQException, URLSyntaxException
     {
-        this(new AMQConnectionURL(ConnectionURL.AMQ_PROTOCOL + "://" +
-                                  username + ":" + password + "@" +
-                                  (clientName == null ? "" : clientName) + "/" +
-                                  virtualHost + "?brokerlist='" + AMQBrokerDetails.checkTransport(broker) + "'"), sslConfig);
+        this(new AMQConnectionURL(
+                ConnectionURL.AMQ_PROTOCOL + "://" + username + ":" + password + "@"
+                + ((clientName == null) ? "" : clientName) + "/" + virtualHost + "?brokerlist='"
+                + AMQBrokerDetails.checkTransport(broker) + "'"), sslConfig);
     }
 
-
-    public AMQConnection(String host, int port, String username, String password,
-                         String clientName, String virtualHost) throws AMQException, URLSyntaxException
+    public AMQConnection(String host, int port, String username, String password, String clientName, String virtualHost)
+        throws AMQException, URLSyntaxException
     {
         this(host, port, false, username, password, clientName, virtualHost, null);
     }
 
-    public AMQConnection(String host, int port, String username, String password,
-                         String clientName, String virtualHost, SSLConfiguration sslConfig) throws AMQException, URLSyntaxException
+    public AMQConnection(String host, int port, String username, String password, String clientName, String virtualHost,
+        SSLConfiguration sslConfig) throws AMQException, URLSyntaxException
     {
         this(host, port, false, username, password, clientName, virtualHost, sslConfig);
     }
 
-
-    public AMQConnection(String host, int port, boolean useSSL, String username, String password,
-                         String clientName, String virtualHost, SSLConfiguration sslConfig) throws AMQException, URLSyntaxException
+    public AMQConnection(String host, int port, boolean useSSL, String username, String password, String clientName,
+        String virtualHost, SSLConfiguration sslConfig) throws AMQException, URLSyntaxException
     {
-        this(new AMQConnectionURL(useSSL ?
-                                  ConnectionURL.AMQ_PROTOCOL + "://" +
-                                  username + ":" + password + "@" +
-                                  (clientName == null ? "" : clientName) +
-                                  virtualHost + "?brokerlist='tcp://" + host + ":" + port + "'"
-                                  + "," + ConnectionURL.OPTIONS_SSL + "='true'" :
-                                                                                ConnectionURL.AMQ_PROTOCOL + "://" +
-                                                                                username + ":" + password + "@" +
-                                                                                (clientName == null ? "" : clientName) +
-                                                                                virtualHost + "?brokerlist='tcp://" + host + ":" + port + "'"
-                                                                                + "," + ConnectionURL.OPTIONS_SSL + "='false'"
-        ), sslConfig);
+        this(new AMQConnectionURL(
+                useSSL
+                ? (ConnectionURL.AMQ_PROTOCOL + "://" + username + ":" + password + "@"
+                    + ((clientName == null) ? "" : clientName) + virtualHost + "?brokerlist='tcp://" + host + ":" + port
+                    + "'" + "," + ConnectionURL.OPTIONS_SSL + "='true'")
+                : (ConnectionURL.AMQ_PROTOCOL + "://" + username + ":" + password + "@"
+                    + ((clientName == null) ? "" : clientName) + virtualHost + "?brokerlist='tcp://" + host + ":" + port
+                    + "'" + "," + ConnectionURL.OPTIONS_SSL + "='false'")), sslConfig);
     }
 
     public AMQConnection(String connection) throws AMQException, URLSyntaxException
@@ -229,13 +224,13 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         this(new AMQConnectionURL(connection), sslConfig);
     }
 
-
     public AMQConnection(ConnectionURL connectionURL, SSLConfiguration sslConfig) throws AMQException
     {
         if (_logger.isInfoEnabled())
         {
             _logger.info("Connection:" + connectionURL);
         }
+
         _sslConfiguration = sslConfig;
         if (connectionURL == null)
         {
@@ -249,7 +244,6 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         _password = connectionURL.getPassword();
         setVirtualHost(connectionURL.getVirtualHost());
 
-
         if (connectionURL.getDefaultQueueExchangeName() != null)
         {
             _defaultQueueExchangeName = connectionURL.getDefaultQueueExchangeName();
@@ -270,7 +264,6 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
             _temporaryTopicExchangeName = connectionURL.getTemporaryTopicExchangeName();
         }
 
-
         _failoverPolicy = new FailoverPolicy(connectionURL);
 
         _protocolHandler = new AMQProtocolHandler(this);
@@ -278,7 +271,6 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         // We are not currently connected
         _connected = false;
 
-
         Exception lastException = new Exception();
         lastException.initCause(new ConnectException());
 
@@ -296,7 +288,8 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
 
                 if (_logger.isInfoEnabled())
                 {
-                    _logger.info("Unable to connect to broker at " + _failoverPolicy.getCurrentBrokerDetails(), e.getCause());
+                    _logger.info("Unable to connect to broker at " + _failoverPolicy.getCurrentBrokerDetails(),
+                        e.getCause());
                 }
             }
         }
@@ -322,7 +315,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
                 }
             }
 
-            if (message == null || message.equals(""))
+            if ((message == null) || message.equals(""))
             {
                 message = "Unable to Connect";
             }
@@ -335,11 +328,13 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
                 {
                     e = new AMQUnresolvedAddressException(message, _failoverPolicy.getCurrentBrokerDetails().toString());
                 }
+
                 e.initCause(lastException);
             }
 
             throw e;
         }
+
         _connectionMetaData = new QpidConnectionMetaData(this);
     }
 
@@ -369,6 +364,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         {
             virtualHost = virtualHost.substring(1);
         }
+
         _virtualHost = virtualHost;
     }
 
@@ -382,7 +378,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
             _protocolHandler.attainState(AMQState.CONNECTION_OPEN);
             _failoverPolicy.attainedConnection();
 
-            //Again this should be changed to a suitable notify
+            // Again this should be changed to a suitable notify
             _connected = true;
         }
         catch (AMQException e)
@@ -401,6 +397,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         try
         {
             makeBrokerConnection(bd);
+
             return true;
         }
         catch (Exception e)
@@ -409,8 +406,10 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
             {
                 _logger.info("Unable to connect to broker at " + bd);
             }
+
             attemptReconnection();
         }
+
         return false;
     }
 
@@ -421,6 +420,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
             try
             {
                 makeBrokerConnection(_failoverPolicy.getNextBrokerDetails());
+
                 return true;
             }
             catch (Exception e)
@@ -436,13 +436,14 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
                 {
                     if (_logger.isInfoEnabled())
                     {
-                        _logger.info(e.getMessage() + ":Unable to connect to broker at " + _failoverPolicy.getCurrentBrokerDetails());
+                        _logger.info(e.getMessage() + ":Unable to connect to broker at "
+                            + _failoverPolicy.getCurrentBrokerDetails());
                     }
                 }
             }
         }
 
-        //connection unsuccessful
+        // connection unsuccessful
         return false;
     }
 
@@ -474,14 +475,14 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         return createSession(transacted, acknowledgeMode, AMQSession.DEFAULT_PREFETCH_HIGH_MARK);
     }
 
-    public org.apache.qpid.jms.Session createSession(final boolean transacted, final int acknowledgeMode,
-                                                     final int prefetch) throws JMSException
+    public org.apache.qpid.jms.Session createSession(final boolean transacted, final int acknowledgeMode, final int prefetch)
+        throws JMSException
     {
         return createSession(transacted, acknowledgeMode, prefetch, prefetch);
     }
 
     public org.apache.qpid.jms.Session createSession(final boolean transacted, final int acknowledgeMode,
-                                                     final int prefetchHigh, final int prefetchLow) throws JMSException
+        final int prefetchHigh, final int prefetchLow) throws JMSException
     {
         checkNotClosed();
         if (channelLimitReached())
@@ -491,85 +492,81 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         else
         {
             return (org.apache.qpid.jms.Session) new FailoverSupport()
-            {
-                public Object operation() throws JMSException
-                {
-                    int channelId = _idFactory.incrementAndGet();
-
-                    if (_logger.isDebugEnabled())
-                    {
-                        _logger.debug("Write channel open frame for channel id " + channelId);
-                    }
-
-                    // We must create the session and register it before actually sending the frame to the server to
-                    // open it, so that there is no window where we could receive data on the channel and not be set
-                    // up to handle it appropriately.
-                    AMQSession session = new AMQSession(AMQConnection.this, channelId, transacted, acknowledgeMode,
-                                                        prefetchHigh, prefetchLow);
-                    _protocolHandler.addSessionByChannel(channelId, session);
-                    registerSession(channelId, session);
-
-                    boolean success = false;
-                    try
-                    {
-                        createChannelOverWire(channelId, prefetchHigh, prefetchLow, transacted);
-                        success = true;
-                    }
-                    catch (AMQException e)
-                    {
-                        JMSException jmse = new JMSException("Error creating session: " + e);
-                        jmse.setLinkedException(e);
-                        throw jmse;
-                    }
-                    finally
-                    {
-                        if (!success)
-                        {
-                            _protocolHandler.removeSessionByChannel(channelId);
-                            deregisterSession(channelId);
-                        }
-                    }
-
-                    if (_started)
                     {
-                        try
-                        {
-                            session.start();
-                        }
-                        catch (AMQException e)
+                        public Object operation() throws JMSException
                         {
-                            throw new JMSAMQException(e);
+                            int channelId = _idFactory.incrementAndGet();
+
+                            if (_logger.isDebugEnabled())
+                            {
+                                _logger.debug("Write channel open frame for channel id " + channelId);
+                            }
+
+                            // We must create the session and register it before actually sending the frame to the server to
+                            // open it, so that there is no window where we could receive data on the channel and not be set
+                            // up to handle it appropriately.
+                            AMQSession session =
+                                new AMQSession(AMQConnection.this, channelId, transacted, acknowledgeMode, prefetchHigh,
+                                    prefetchLow);
+                            _protocolHandler.addSessionByChannel(channelId, session);
+                            registerSession(channelId, session);
+
+                            boolean success = false;
+                            try
+                            {
+                                createChannelOverWire(channelId, prefetchHigh, prefetchLow, transacted);
+                                success = true;
+                            }
+                            catch (AMQException e)
+                            {
+                                JMSException jmse = new JMSException("Error creating session: " + e);
+                                jmse.setLinkedException(e);
+                                throw jmse;
+                            }
+                            finally
+                            {
+                                if (!success)
+                                {
+                                    _protocolHandler.removeSessionByChannel(channelId);
+                                    deregisterSession(channelId);
+                                }
+                            }
+
+                            if (_started)
+                            {
+                                try
+                                {
+                                    session.start();
+                                }
+                                catch (AMQException e)
+                                {
+                                    throw new JMSAMQException(e);
+                                }
+                            }
+
+                            return session;
                         }
-                    }
-                    return session;
-                }
-            }.execute(this);
+                    }.execute(this);
         }
     }
 
     private void createChannelOverWire(int channelId, int prefetchHigh, int prefetchLow, boolean transacted)
-            throws AMQException
+        throws AMQException
     {
 
         // TODO: Be aware of possible changes to parameter order as versions change.
 
-        _protocolHandler.syncWrite(
-                ChannelOpenBody.createAMQFrame(channelId,
-                                               _protocolHandler.getProtocolMajorVersion(),
-                                               _protocolHandler.getProtocolMinorVersion(),
-                                               null),    // outOfBand
-                                                         ChannelOpenOkBody.class);
-
-        //todo send low water mark when protocol allows.
-        //todo Be aware of possible changes to parameter order as versions change.
-        _protocolHandler.syncWrite(
-                BasicQosBody.createAMQFrame(channelId,
-                                            _protocolHandler.getProtocolMajorVersion(),
-                                            _protocolHandler.getProtocolMinorVersion(),
-                                            false,    // global
-                                            prefetchHigh,    // prefetchCount
-                                            0),    // prefetchSize
-                                                   BasicQosOkBody.class);
+        _protocolHandler.syncWrite(ChannelOpenBody.createAMQFrame(channelId, _protocolHandler.getProtocolMajorVersion(),
+                _protocolHandler.getProtocolMinorVersion(), null), // outOfBand
+            ChannelOpenOkBody.class);
+
+        // todo send low water mark when protocol allows.
+        // todo Be aware of possible changes to parameter order as versions change.
+        _protocolHandler.syncWrite(BasicQosBody.createAMQFrame(channelId, _protocolHandler.getProtocolMajorVersion(),
+                _protocolHandler.getProtocolMinorVersion(), false, // global
+                prefetchHigh, // prefetchCount
+                0), // prefetchSize
+            BasicQosOkBody.class);
 
         if (transacted)
         {
@@ -579,10 +576,8 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
             }
 
             // TODO: Be aware of possible changes to parameter order as versions change.
-            _protocolHandler.syncWrite(TxSelectBody.createAMQFrame(channelId,
-                                                                   _protocolHandler.getProtocolMajorVersion(),
-                                                                   _protocolHandler.getProtocolMinorVersion()),
-                                       TxSelectOkBody.class);
+            _protocolHandler.syncWrite(TxSelectBody.createAMQFrame(channelId, _protocolHandler.getProtocolMajorVersion(),
+                    _protocolHandler.getProtocolMinorVersion()), TxSelectOkBody.class);
         }
     }
 
@@ -596,11 +591,10 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         {
             _protocolHandler.removeSessionByChannel(channelId);
             deregisterSession(channelId);
-            throw new AMQException("Error reopening channel " + channelId + " after failover: " + e);
+            throw new AMQException("Error reopening channel " + channelId + " after failover: " + e, e);
         }
     }
 
-
     public void setFailoverPolicy(FailoverPolicy policy)
     {
         _failoverPolicy = policy;
@@ -645,12 +639,13 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
 
     private boolean channelLimitReached()
     {
-        return _maximumChannelCount != 0 && _sessions.size() == _maximumChannelCount;
+        return (_maximumChannelCount != 0) && (_sessions.size() == _maximumChannelCount);
     }
 
     public String getClientID() throws JMSException
     {
         checkNotClosed();
+
         return _clientName;
     }
 
@@ -666,6 +661,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
     public ConnectionMetaData getMetaData() throws JMSException
     {
         checkNotClosed();
+
         return _connectionMetaData;
 
     }
@@ -673,6 +669,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
     public ExceptionListener getExceptionListener() throws JMSException
     {
         checkNotClosed();
+
         return _exceptionListener;
     }
 
@@ -706,6 +703,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
                     throw new JMSAMQException(e);
                 }
             }
+
             _started = true;
         }
     }
@@ -726,13 +724,14 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
                     throw new JMSAMQException(e);
                 }
             }
+
             _started = false;
         }
     }
 
     public void close() throws JMSException
     {
-        close(-1);
+        close(DEFAULT_TIMEOUT);
     }
 
     public void close(long timeout) throws JMSException
@@ -752,7 +751,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
                     {
                         try
                         {
-                            //adjust timeout
+                            // adjust timeout
                             long taskPoolTimeout = adjustTimeout(timeout, startCloseTime);
 
                             _taskPool.awaitTermination(taskPoolTimeout, TimeUnit.MILLISECONDS);
@@ -763,7 +762,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
                         }
                     }
 
-                    //adjust timeout
+                    // adjust timeout
                     timeout = adjustTimeout(timeout, startCloseTime);
 
                     _protocolHandler.closeConnection(timeout);
@@ -771,7 +770,9 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
                 }
                 catch (AMQException e)
                 {
-                    throw new JMSException("Error closing connection: " + e);
+                    JMSException jmse = new JMSException("Error closing connection: " + e);
+                    jmse.setLinkedException(e);
+                    throw jmse;
                 }
             }
         }
@@ -785,6 +786,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         {
             timeout = 0;
         }
+
         return timeout;
     }
 
@@ -803,6 +805,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
 
             session.markClosed();
         }
+
         _sessions.clear();
     }
 
@@ -842,6 +845,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
                 }
             }
         }
+
         _sessions.clear();
         if (sessionException != null)
         {
@@ -850,42 +854,42 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
     }
 
     public ConnectionConsumer createConnectionConsumer(Destination destination, String messageSelector,
-                                                       ServerSessionPool sessionPool,
-                                                       int maxMessages) throws JMSException
+        ServerSessionPool sessionPool, int maxMessages) throws JMSException
     {
         checkNotClosed();
+
         return null;
     }
 
-    public ConnectionConsumer createConnectionConsumer(Queue queue, String messageSelector,
-                                                       ServerSessionPool sessionPool,
-                                                       int maxMessages) throws JMSException
+    public ConnectionConsumer createConnectionConsumer(Queue queue, String messageSelector, ServerSessionPool sessionPool,
+        int maxMessages) throws JMSException
     {
         checkNotClosed();
+
         return null;
     }
 
-    public ConnectionConsumer createConnectionConsumer(Topic topic, String messageSelector,
-                                                       ServerSessionPool sessionPool,
-                                                       int maxMessages) throws JMSException
+    public ConnectionConsumer createConnectionConsumer(Topic topic, String messageSelector, ServerSessionPool sessionPool,
+        int maxMessages) throws JMSException
     {
         checkNotClosed();
+
         return null;
     }
 
-    public ConnectionConsumer createDurableConnectionConsumer(Topic topic, String subscriptionName,
-                                                              String messageSelector, ServerSessionPool sessionPool,
-                                                              int maxMessages)
-            throws JMSException
+    public ConnectionConsumer createDurableConnectionConsumer(Topic topic, String subscriptionName, String messageSelector,
+        ServerSessionPool sessionPool, int maxMessages) throws JMSException
     {
         // TODO Auto-generated method stub
         checkNotClosed();
+
         return null;
     }
 
     public long getMaximumChannelCount() throws JMSException
     {
         checkNotClosed();
+
         return _maximumChannelCount;
     }
 
@@ -974,6 +978,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         {
             proceed = _connectionListener.preFailover(redirect);
         }
+
         return proceed;
     }
 
@@ -994,6 +999,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
             {
                 markAllSessionsClosed();
             }
+
             return resubscribe;
         }
         else
@@ -1057,12 +1063,15 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         {
             if (cause instanceof AMQException)
             {
-                je = new JMSException(Integer.toString(((AMQException) cause).getErrorCode().getCode()), "Exception thrown against " + toString() + ": " + cause);
+                je =
+                    new JMSException(Integer.toString(((AMQException) cause).getErrorCode().getCode()),
+                        "Exception thrown against " + toString() + ": " + cause);
             }
             else
             {
                 je = new JMSException("Exception thrown against " + toString() + ": " + cause);
             }
+
             if (cause instanceof Exception)
             {
                 je.setLinkedException((Exception) cause);
@@ -1090,6 +1099,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
                 {
                     _logger.info("Closing AMQConnection due to :" + cause.getMessage());
                 }
+
                 _closed.set(true);
                 closeAllSessions(cause, -1, -1); // FIXME: when doing this end up with RejectedExecutionException from executor.
             }
@@ -1145,9 +1155,11 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
             buf.append("Host: ").append(String.valueOf(bd.getHost()));
             buf.append("\nPort: ").append(String.valueOf(bd.getPort()));
         }
+
         buf.append("\nVirtual Host: ").append(String.valueOf(_virtualHost));
         buf.append("\nClient ID: ").append(String.valueOf(_clientName));
-        buf.append("\nActive session count: ").append(_sessions == null ? 0 : _sessions.size());
+        buf.append("\nActive session count: ").append((_sessions == null) ? 0 : _sessions.size());
+
         return buf.toString();
     }
 
@@ -1158,11 +1170,8 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
 
     public Reference getReference() throws NamingException
     {
-        return new Reference(
-                AMQConnection.class.getName(),
-                new StringRefAddr(AMQConnection.class.getName(), toURL()),
-                AMQConnectionFactory.class.getName(),
-                null);          // factory location
+        return new Reference(AMQConnection.class.getName(), new StringRefAddr(AMQConnection.class.getName(), toURL()),
+                AMQConnectionFactory.class.getName(), null); // factory location
     }
 
     public SSLConfiguration getSSLConfiguration()
@@ -1175,19 +1184,16 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         return _defaultTopicExchangeName;
     }
 
-
     public void setDefaultTopicExchangeName(AMQShortString defaultTopicExchangeName)
     {
         _defaultTopicExchangeName = defaultTopicExchangeName;
     }
 
-
     public AMQShortString getDefaultQueueExchangeName()
     {
         return _defaultQueueExchangeName;
     }
 
-
     public void setDefaultQueueExchangeName(AMQShortString defaultQueueExchangeName)
     {
         _defaultQueueExchangeName = defaultQueueExchangeName;
@@ -1200,10 +1206,9 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
 
     public AMQShortString getTemporaryQueueExchangeName()
     {
-        return _temporaryQueueExchangeName;  //To change body of created methods use File | Settings | File Templates.
+        return _temporaryQueueExchangeName; // To change body of created methods use File | Settings | File Templates.
     }
 
-
     public void setTemporaryTopicExchangeName(AMQShortString temporaryTopicExchangeName)
     {
         _temporaryTopicExchangeName = temporaryTopicExchangeName;
diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQConnectionURL.java b/java/client/src/main/java/org/apache/qpid/client/AMQConnectionURL.java
index 0dcc544ea8..b3fbd1f510 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQConnectionURL.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQConnectionURL.java
@@ -7,9 +7,9 @@
  * 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
@@ -20,12 +20,6 @@
  */
 package org.apache.qpid.client;
 
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.jms.BrokerDetails;
-import org.apache.qpid.jms.ConnectionURL;
-import org.apache.qpid.url.URLHelper;
-import org.apache.qpid.url.URLSyntaxException;
-
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.HashMap;
@@ -33,6 +27,12 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.StringTokenizer;
 
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.jms.BrokerDetails;
+import org.apache.qpid.jms.ConnectionURL;
+import org.apache.qpid.url.URLHelper;
+import org.apache.qpid.url.URLSyntaxException;
+
 public class AMQConnectionURL implements ConnectionURL
 {
     private String _url;
@@ -49,7 +49,6 @@ public class AMQConnectionURL implements ConnectionURL
     private AMQShortString _temporaryTopicExchangeName;
     private AMQShortString _temporaryQueueExchangeName;
 
-
     public AMQConnectionURL(String fullURL) throws URLSyntaxException
     {
         _url = fullURL;
@@ -58,18 +57,18 @@ public class AMQConnectionURL implements ConnectionURL
         _failoverOptions = new HashMap<String, String>();
 
         // Connection URL format
-        //amqp://[user:pass@][clientid]/virtualhost?brokerlist='tcp://host:port?option=\'value\',option=\'value\';vm://:3/virtualpath?option=\'value\'',failover='method?option=\'value\',option='value''"
+        // amqp://[user:pass@][clientid]/virtualhost?brokerlist='tcp://host:port?option=\'value\',option=\'value\';vm://:3/virtualpath?option=\'value\'',failover='method?option=\'value\',option='value''"
         // Options are of course optional except for requiring a single broker in the broker list.
         try
         {
             URI connection = new URI(fullURL);
 
-            if (connection.getScheme() == null || !(connection.getScheme().equalsIgnoreCase(AMQ_PROTOCOL)))
+            if ((connection.getScheme() == null) || !(connection.getScheme().equalsIgnoreCase(AMQ_PROTOCOL)))
             {
                 throw new URISyntaxException(fullURL, "Not an AMQP URL");
             }
 
-            if (connection.getHost() == null || connection.getHost().equals(""))
+            if ((connection.getHost() == null) || connection.getHost().equals(""))
             {
                 String uid = AMQConnectionFactory.getUniqueClientID();
                 if (uid == null)
@@ -91,7 +90,7 @@ public class AMQConnectionURL implements ConnectionURL
 
             if (userInfo == null)
             {
-                //Fix for Java 1.5 which doesn't parse UserInfo for non http URIs
+                // Fix for Java 1.5 which doesn't parse UserInfo for non http URIs
                 userInfo = connection.getAuthority();
 
                 if (userInfo != null)
@@ -112,16 +111,16 @@ public class AMQConnectionURL implements ConnectionURL
 
             if (userInfo == null)
             {
-                throw URLHelper.parseError(AMQ_PROTOCOL.length() + 3,
-                                           "User information not found on url", fullURL);
+                throw URLHelper.parseError(AMQ_PROTOCOL.length() + 3, "User information not found on url", fullURL);
             }
             else
             {
                 parseUserInfo(userInfo);
             }
+
             String virtualHost = connection.getPath();
 
-            if (virtualHost != null && (!virtualHost.equals("")))
+            if ((virtualHost != null) && (!virtualHost.equals("")))
             {
                 setVirtualHost(virtualHost);
             }
@@ -130,7 +129,7 @@ public class AMQConnectionURL implements ConnectionURL
                 int authLength = connection.getAuthority().length();
                 int start = AMQ_PROTOCOL.length() + 3;
                 int testIndex = start + authLength;
-                if (testIndex < fullURL.length() && fullURL.charAt(testIndex) == '?')
+                if ((testIndex < fullURL.length()) && (fullURL.charAt(testIndex) == '?'))
                 {
                     throw URLHelper.parseError(start, testIndex - start, "Virtual host found", fullURL);
                 }
@@ -141,14 +140,9 @@ public class AMQConnectionURL implements ConnectionURL
 
             }
 
-
             URLHelper.parseOptions(_options, connection.getQuery());
 
             processOptions();
-
-            //Fragment is #string (not used)
-            //System.out.println(connection.getFragment());
-
         }
         catch (URISyntaxException uris)
         {
@@ -165,11 +159,10 @@ public class AMQConnectionURL implements ConnectionURL
             }
             else
             {
-                if (slash != 0 && fullURL.charAt(slash - 1) == ':')
+                if ((slash != 0) && (fullURL.charAt(slash - 1) == ':'))
                 {
                     throw URLHelper.parseError(slash - 2, fullURL.indexOf('?') - slash + 2,
-                                               "Virtual host looks like a windows path, forward slash not allowed in URL",
-                                               fullURL);
+                        "Virtual host looks like a windows path, forward slash not allowed in URL", fullURL);
                 }
                 else
                 {
@@ -182,14 +175,14 @@ public class AMQConnectionURL implements ConnectionURL
 
     private void parseUserInfo(String userinfo) throws URLSyntaxException
     {
-        //user info = user:pass
+        // user info = user:pass
 
         int colonIndex = userinfo.indexOf(':');
 
         if (colonIndex == -1)
         {
             throw URLHelper.parseError(AMQ_PROTOCOL.length() + 3, userinfo.length(),
-                                       "Null password in user information not allowed.", _url);
+                "Null password in user information not allowed.", _url);
         }
         else
         {
@@ -205,7 +198,7 @@ public class AMQConnectionURL implements ConnectionURL
         {
             String brokerlist = _options.get(OPTIONS_BROKERLIST);
 
-            //brokerlist tcp://host:port?option='value',option='value';vm://:3/virtualpath?option='value'
+            // brokerlist tcp://host:port?option='value',option='value';vm://:3/virtualpath?option='value'
             StringTokenizer st = new StringTokenizer(brokerlist, "" + URLHelper.BROKER_SEPARATOR);
 
             while (st.hasMoreTokens())
@@ -244,19 +237,16 @@ public class AMQConnectionURL implements ConnectionURL
             _defaultTopicExchangeName = new AMQShortString(_options.get(OPTIONS_DEFAULT_TOPIC_EXCHANGE));
         }
 
-
         if (_options.containsKey(OPTIONS_DEFAULT_QUEUE_EXCHANGE))
         {
             _defaultQueueExchangeName = new AMQShortString(_options.get(OPTIONS_DEFAULT_QUEUE_EXCHANGE));
         }
 
-
         if (_options.containsKey(OPTIONS_TEMPORARY_QUEUE_EXCHANGE))
         {
             _temporaryQueueExchangeName = new AMQShortString(_options.get(OPTIONS_TEMPORARY_QUEUE_EXCHANGE));
         }
 
-
         if (_options.containsKey(OPTIONS_TEMPORARY_TOPIC_EXCHANGE))
         {
             _temporaryTopicExchangeName = new AMQShortString(_options.get(OPTIONS_TEMPORARY_TOPIC_EXCHANGE));
@@ -439,12 +429,11 @@ public class AMQConnectionURL implements ConnectionURL
         return sb.toString();
     }
 
-
     public static void main(String[] args) throws URLSyntaxException
     {
-
-        String url2 = "amqp://ritchiem:bob@temp?brokerlist='tcp://localhost:5672;jcp://fancyserver:3000/',failover='roundrobin'";
-        //"amqp://user:pass@clientid/virtualhost?brokerlist='tcp://host:1?option1=\'value\',option2=\'value\';vm://:3?option1=\'value\'',failover='method?option1=\'value\',option2='value''";
+        String url2 =
+            "amqp://ritchiem:bob@temp?brokerlist='tcp://localhost:5672;jcp://fancyserver:3000/',failover='roundrobin'";
+        // "amqp://user:pass@clientid/virtualhost?brokerlist='tcp://host:1?option1=\'value\',option2=\'value\';vm://:3?option1=\'value\'',failover='method?option1=\'value\',option2='value''";
 
         ConnectionURL connectionurl2 = new AMQConnectionURL(url2);
 
@@ -452,5 +441,4 @@ public class AMQConnectionURL implements ConnectionURL
         System.out.println(connectionurl2);
 
     }
-
 }
diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java b/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java
index 661372845a..585991d905 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java
@@ -53,6 +53,8 @@ public abstract class AMQDestination implements Destination, Referenceable
     private String _url;
     private AMQShortString _urlAsShortString;
 
+    private boolean _validated;
+
     private byte[] _byteEncoding;
     private static final int IS_DURABLE_MASK = 0x1;
     private static final int IS_EXCLUSIVE_MASK = 0x2;
@@ -198,12 +200,16 @@ public abstract class AMQDestination implements Destination, Referenceable
     {
         return toURL();
 
-        /*
-        return "Destination: " + _destinationName + ", " +
-               "Queue Name: " + _queueName + ", Exchange: " + _exchangeName +
-               ", Exchange class: " + _exchangeClass + ", Exclusive: " + _isExclusive +
-               ", AutoDelete: " + _isAutoDelete + ", Routing  Key: " + getRoutingKey();
-         */
+    }
+
+    public boolean isValidated()
+    {
+        return _validated;
+    }
+
+    public void setValidated(boolean validated)
+    {
+        _validated = validated;
     }
 
     public String toURL()
@@ -348,15 +354,7 @@ public abstract class AMQDestination implements Destination, Referenceable
         {
             return false;
         }
-  /*      if (_isExclusive != that._isExclusive)
-        {
-            return false;
-        }
-        if (_isAutoDelete != that._isAutoDelete)
-        {
-            return false;
-        }
-        */
+
         return true;
     }
 
@@ -370,8 +368,7 @@ public abstract class AMQDestination implements Destination, Referenceable
         {
             result = 29 * result + _queueName.hashCode();
         }
-//        result = result * (_isExclusive ? 13 : 7);
-//        result = result * (_isAutoDelete ? 13 : 7);
+
         return result;
     }
 
diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
index 82f9a036d2..8bb5b622f7 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
@@ -202,6 +202,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
     /** Responsible for decoding a message fragment and passing it to the appropriate message consumer. */
 
     private static final Logger _dispatcherLogger = Logger.getLogger(Dispatcher.class);
+    private AtomicBoolean _firstDispatcher = new AtomicBoolean(true);
 
     private class Dispatcher extends Thread
     {
@@ -327,8 +328,11 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
                                                    ") is closed rejecting(requeue)...");
                         }
                     }
-
-                    rejectMessage(message, true);
+                    // Don't reject if we're already closing
+                    if (!_closed.get())
+                    {
+                        rejectMessage(message, true);
+                    }
                 }
                 else
                 {
@@ -995,42 +999,42 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
         throw new java.lang.UnsupportedOperationException();
     }
 
-    public MessageProducer createProducer(Destination destination, boolean mandatory,
-                                          boolean immediate, boolean waitUntilSent)
+    public BasicMessageProducer createProducer(Destination destination, boolean mandatory,
+                                               boolean immediate, boolean waitUntilSent)
             throws JMSException
     {
         return createProducerImpl(destination, mandatory, immediate, waitUntilSent);
     }
 
-    public MessageProducer createProducer(Destination destination, boolean mandatory, boolean immediate)
+    public BasicMessageProducer createProducer(Destination destination, boolean mandatory, boolean immediate)
             throws JMSException
     {
         return createProducerImpl(destination, mandatory, immediate);
     }
 
-    public MessageProducer createProducer(Destination destination, boolean immediate)
+    public BasicMessageProducer createProducer(Destination destination, boolean immediate)
             throws JMSException
     {
         return createProducerImpl(destination, DEFAULT_MANDATORY, immediate);
     }
 
-    public MessageProducer createProducer(Destination destination) throws JMSException
+    public BasicMessageProducer createProducer(Destination destination) throws JMSException
     {
         return createProducerImpl(destination, DEFAULT_MANDATORY, DEFAULT_IMMEDIATE);
     }
 
-    private org.apache.qpid.jms.MessageProducer createProducerImpl(Destination destination, boolean mandatory,
-                                                                   boolean immediate)
+    private BasicMessageProducer createProducerImpl(Destination destination, boolean mandatory,
+                                                    boolean immediate)
             throws JMSException
     {
         return createProducerImpl(destination, mandatory, immediate, false);
     }
 
-    private org.apache.qpid.jms.MessageProducer createProducerImpl(final Destination destination, final boolean mandatory,
-                                                                   final boolean immediate, final boolean waitUntilSent)
+    private BasicMessageProducer createProducerImpl(final Destination destination, final boolean mandatory,
+                                                    final boolean immediate, final boolean waitUntilSent)
             throws JMSException
     {
-        return (org.apache.qpid.jms.MessageProducer) new FailoverSupport()
+        return (BasicMessageProducer) new FailoverSupport()
         {
             public Object operation() throws JMSException
             {
@@ -1248,8 +1252,10 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
                 {
                     JMSException ex = new JMSException("Error registering consumer: " + e);
 
-                    //todo remove
-                    e.printStackTrace();
+                    if (_logger.isDebugEnabled())
+                    {
+                        e.printStackTrace();
+                    }
                     ex.setLinkedException(e);
                     throw ex;
                 }
@@ -1926,6 +1932,24 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
 
     synchronized void startDistpatcherIfNecessary()
     {
+        if (Boolean.parseBoolean(System.getProperties().getProperty("REGISTER_CONSUMERS_FLOWED", "false")))
+        {
+//            if (!connectionStopped)
+            {
+                if (isSuspended() && _firstDispatcher.getAndSet(false))
+                {
+                    try
+                    {
+                        suspendChannel(false);
+                    }
+                    catch (AMQException e)
+                    {
+                        _logger.info("Suspending channel threw an exception:" + e);
+                    }
+                }
+            }
+        }
+
         startDistpatcherIfNecessary(false);
     }
 
@@ -1974,6 +1998,27 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
 
         bindQueue(amqd, queueName, protocolHandler, consumer.getRawSelectorFieldTable());
 
+        // The dispatcher will be null if we have just created this session
+        // so suspend the channel before we register our consumer so that we don't
+        // start prefetching until a receive/mListener is set.
+        if (Boolean.parseBoolean(System.getProperties().getProperty("REGISTER_CONSUMERS_FLOWED", "false")))
+        {
+            if (_dispatcher == null)
+            {
+                if (!isSuspended())
+                {
+                    try
+                    {
+                        suspendChannel(true);
+                    }
+                    catch (AMQException e)
+                    {
+                        _logger.info("Suspending channel threw an exception:" + e);
+                    }
+                }
+            }
+        }
+
         try
         {
             consumeFromQueue(consumer, queueName, protocolHandler, nowait, consumer.getMessageSelector());
@@ -2089,7 +2134,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
         // Remove the consumer from the map
         BasicMessageConsumer consumer = (BasicMessageConsumer) _consumers.get(consumerTag);
         if (consumer != null)
-        {            
+        {
 //            fixme this isn't right.. needs to check if _queue contains data for this consumer
             if (consumer.isAutoClose())// && _queue.isEmpty())
             {
diff --git a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
index 73010ce517..1c3cdbcb65 100644
--- a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
+++ b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
@@ -20,9 +20,9 @@
  */
 package org.apache.qpid.client;
 
+import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
-import java.util.Arrays;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.TimeUnit;
@@ -34,6 +34,7 @@ import javax.jms.Message;
 import javax.jms.MessageListener;
 
 import org.apache.log4j.Logger;
+
 import org.apache.qpid.AMQException;
 import org.apache.qpid.client.message.AbstractJMSMessage;
 import org.apache.qpid.client.message.MessageFactoryRegistry;
@@ -138,10 +139,10 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
     private boolean _noConsume;
     private List<StackTraceElement> _closedStack = null;
 
-    protected BasicMessageConsumer(int channelId, AMQConnection connection, AMQDestination destination, String messageSelector,
-                                   boolean noLocal, MessageFactoryRegistry messageFactory, AMQSession session,
-                                   AMQProtocolHandler protocolHandler, FieldTable rawSelectorFieldTable,
-                                   int prefetchHigh, int prefetchLow, boolean exclusive, int acknowledgeMode, boolean noConsume, boolean autoClose)
+    protected BasicMessageConsumer(int channelId, AMQConnection connection, AMQDestination destination,
+                                   String messageSelector, boolean noLocal, MessageFactoryRegistry messageFactory, AMQSession session,
+                                   AMQProtocolHandler protocolHandler, FieldTable rawSelectorFieldTable, int prefetchHigh, int prefetchLow,
+                                   boolean exclusive, int acknowledgeMode, boolean noConsume, boolean autoClose)
     {
         _channelId = channelId;
         _connection = connection;
@@ -160,7 +161,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
         _autoClose = autoClose;
         _noConsume = noConsume;
 
-        //Force queue browsers not to use acknowledge modes.
+        // Force queue browsers not to use acknowledge modes.
         if (_noConsume)
         {
             _acknowledgeMode = Session.NO_ACKNOWLEDGE;
@@ -175,12 +176,14 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
     public String getMessageSelector() throws JMSException
     {
         checkPreConditions();
+
         return _messageSelector;
     }
 
     public MessageListener getMessageListener() throws JMSException
     {
         checkPreConditions();
+
         return _messageListener.get();
     }
 
@@ -198,14 +201,14 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
     {
         checkPreConditions();
 
-        //if the current listener is non-null and the session is not stopped, then
-        //it is an error to call this method.
+        // if the current listener is non-null and the session is not stopped, then
+        // it is an error to call this method.
 
-        //i.e. it is only valid to call this method if
+        // i.e. it is only valid to call this method if
         //
-        //    (a) the connection is stopped, in which case the dispatcher is not running
-        //    OR
-        //    (b) the listener is null AND we are not receiving synchronously at present
+        // (a) the connection is stopped, in which case the dispatcher is not running
+        // OR
+        // (b) the listener is null AND we are not receiving synchronously at present
         //
 
         if (!_session.getAMQConnection().started())
@@ -215,7 +218,8 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
 
             if (_logger.isDebugEnabled())
             {
-                _logger.debug("Session stopped : Message listener(" + messageListener + ") set for destination " + _destination);
+                _logger.debug("Session stopped : Message listener(" + messageListener + ") set for destination "
+                              + _destination);
             }
         }
         else
@@ -224,6 +228,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
             {
                 throw new javax.jms.IllegalStateException("Another thread is already receiving synchronously.");
             }
+
             if (!_messageListener.compareAndSet(null, messageListener))
             {
                 throw new javax.jms.IllegalStateException("Attempt to alter listener while session is started.");
@@ -233,7 +238,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
 
             if (messageListener != null)
             {
-                //handle case where connection has already been started, and the dispatcher has alreaded started
+                // handle case where connection has already been started, and the dispatcher has alreaded started
                 // putting values on the _synchronousQueue
 
                 synchronized (_session)
@@ -263,10 +268,12 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
         {
             throw new javax.jms.IllegalStateException("Another thread is already receiving.");
         }
+
         if (isMessageListenerSet())
         {
             throw new javax.jms.IllegalStateException("A listener has already been set.");
         }
+
         _receivingThread = Thread.currentThread();
     }
 
@@ -331,6 +338,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
             {
                 return null;
             }
+
             Object o = null;
             if (l > 0)
             {
@@ -340,6 +348,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
             {
                 o = _synchronousQueue.take();
             }
+
             final AbstractJMSMessage m = returnMessageOrThrow(o);
             if (m != null)
             {
@@ -352,6 +361,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
         catch (InterruptedException e)
         {
             _logger.warn("Interrupted: " + e);
+
             return null;
         }
         finally
@@ -365,6 +375,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
         if (isAutoClose() && _closeWhenNoMessages && _synchronousQueue.isEmpty())
         {
             close(false);
+
             return true;
         }
         else
@@ -387,6 +398,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
             {
                 return null;
             }
+
             Object o = _synchronousQueue.poll();
             final AbstractJMSMessage m = returnMessageOrThrow(o);
             if (m != null)
@@ -414,8 +426,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
      * @throws JMSException if the argument is a throwable. If it is a JMSException it is rethrown as is, but if not a
      *                      JMSException is created with the linked exception set appropriately
      */
-    private AbstractJMSMessage returnMessageOrThrow(Object o)
-            throws JMSException
+    private AbstractJMSMessage returnMessageOrThrow(Object o) throws JMSException
     {
         // errors are passed via the queue too since there is no way of interrupting the poll() via the API.
         if (o instanceof Throwable)
@@ -425,6 +436,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
             {
                 e.setLinkedException((Exception) o);
             }
+
             throw e;
         }
         else
@@ -433,7 +445,6 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
         }
     }
 
-
     public void close() throws JMSException
     {
         close(true);
@@ -441,7 +452,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
 
     public void close(boolean sendClose) throws JMSException
     {
-        //synchronized (_closed)
+        // synchronized (_closed)
 
         if (_logger.isInfoEnabled())
         {
@@ -456,7 +467,8 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
                 {
                     if (_closedStack != null)
                     {
-                        _logger.trace(_consumerTag + " close():" + Arrays.asList(Thread.currentThread().getStackTrace()).subList(3, 6));
+                        _logger.trace(_consumerTag + " close():"
+                                      + Arrays.asList(Thread.currentThread().getStackTrace()).subList(3, 6));
                         _logger.trace(_consumerTag + " previously:" + _closedStack.toString());
                     }
                     else
@@ -464,14 +476,14 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
                         _closedStack = Arrays.asList(Thread.currentThread().getStackTrace()).subList(3, 6);
                     }
                 }
+
                 if (sendClose)
                 {
                     // TODO: Be aware of possible changes to parameter order as versions change.
-                    final AMQFrame cancelFrame = BasicCancelBody.createAMQFrame(_channelId,
-                                                                                _protocolHandler.getProtocolMajorVersion(),
-                                                                                _protocolHandler.getProtocolMinorVersion(),
-                                                                                _consumerTag,    // consumerTag
-                                                                                false);    // nowait
+                    final AMQFrame cancelFrame =
+                            BasicCancelBody.createAMQFrame(_channelId, _protocolHandler.getProtocolMajorVersion(),
+                                                           _protocolHandler.getProtocolMinorVersion(), _consumerTag, // consumerTag
+                                                           false); // nowait
 
                     try
                     {
@@ -485,25 +497,28 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
                     }
                     catch (AMQException e)
                     {
-                        _logger.error("Error closing consumer: " + e, e);
-                        throw new JMSException("Error closing consumer: " + e);
+                        // _logger.error("Error closing consumer: " + e, e);            
+                        JMSException jmse = new JMSException("Error closing consumer: " + e);
+                        jmse.setLinkedException(e);
+                        throw jmse;
                     }
                 }
                 else
                 {
-//                    //fixme this probably is not right
-//                    if (!isNoConsume())
-                    {   //done in BasicCancelOK Handler but not sending one so just deregister.
+                    // //fixme this probably is not right
+                    // if (!isNoConsume())
+                    { // done in BasicCancelOK Handler but not sending one so just deregister.
                         deregisterConsumer();
                     }
                 }
 
-                if (_messageListener != null && _receiving.get())
+                if ((_messageListener != null) && _receiving.get())
                 {
                     if (_logger.isInfoEnabled())
                     {
                         _logger.info("Interrupting thread: " + _receivingThread);
                     }
+
                     _receivingThread.interrupt();
                 }
             }
@@ -516,7 +531,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
      */
     void markClosed()
     {
-//        synchronized (_closed)
+        // synchronized (_closed)
         {
             _closed.set(true);
 
@@ -524,7 +539,8 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
             {
                 if (_closedStack != null)
                 {
-                    _logger.trace(_consumerTag + " markClosed():" + Arrays.asList(Thread.currentThread().getStackTrace()).subList(3, 8));
+                    _logger.trace(_consumerTag + " markClosed():"
+                                  + Arrays.asList(Thread.currentThread().getStackTrace()).subList(3, 8));
                     _logger.trace(_consumerTag + " previously:" + _closedStack.toString());
                 }
                 else
@@ -533,6 +549,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
                 }
             }
         }
+
         deregisterConsumer();
     }
 
@@ -551,22 +568,22 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
         {
             _logger.debug("notifyMessage called with message number " + messageFrame.getDeliverBody().deliveryTag);
         }
+
         try
         {
-            AbstractJMSMessage jmsMessage = _messageFactory.createMessage(messageFrame.getDeliverBody().deliveryTag,
-                                                                          messageFrame.getDeliverBody().redelivered,
-                                                                          messageFrame.getDeliverBody().exchange,
-                                                                          messageFrame.getDeliverBody().routingKey,
-                                                                          messageFrame.getContentHeader(),
-                                                                          messageFrame.getBodies());
+            AbstractJMSMessage jmsMessage =
+                    _messageFactory.createMessage(messageFrame.getDeliverBody().deliveryTag,
+                                                  messageFrame.getDeliverBody().redelivered, messageFrame.getDeliverBody().exchange,
+                                                  messageFrame.getDeliverBody().routingKey, messageFrame.getContentHeader(), messageFrame.getBodies());
 
             if (debug)
             {
                 _logger.debug("Message is of type: " + jmsMessage.getClass().getName());
             }
-//            synchronized (_closed)
+            // synchronized (_closed)
+
             {
-//                if (!_closed.get())
+                // if (!_closed.get())
                 {
 
                     jmsMessage.setConsumer(this);
@@ -575,12 +592,12 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
 
                     notifyMessage(jmsMessage, channelId);
                 }
-//                else
-//                {
-//                    _logger.error("MESSAGE REJECTING!");
-//                    _session.rejectMessage(jmsMessage, true);
-//                    //_logger.error("MESSAGE JUST DROPPED!");
-//                }
+                // else
+                // {
+                // _logger.error("MESSAGE REJECTING!");
+                // _session.rejectMessage(jmsMessage, true);
+                // //_logger.error("MESSAGE JUST DROPPED!");
+                // }
             }
         }
         catch (Exception e)
@@ -606,11 +623,11 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
         {
             if (isMessageListenerSet())
             {
-                //we do not need a lock around the test above, and the dispatch below as it is invalid
-                //for an application to alter an installed listener while the session is started
-//                synchronized (_closed)
+                // we do not need a lock around the test above, and the dispatch below as it is invalid
+                // for an application to alter an installed listener while the session is started
+                // synchronized (_closed)
                 {
-//                    if (!_closed.get())
+                    // if (!_closed.get())
                     {
 
                         preApplicationProcessing(jmsMessage);
@@ -641,9 +658,11 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
     {
         switch (_acknowledgeMode)
         {
+
             case Session.PRE_ACKNOWLEDGE:
                 _session.acknowledgeMessage(msg.getDeliveryTag(), false);
                 break;
+
             case Session.CLIENT_ACKNOWLEDGE:
                 // we set the session so that when the user calls acknowledge() it can call the method on session
                 // to send out the appropriate frame
@@ -657,17 +676,21 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
         msg.setJMSDestination(_destination);
         switch (_acknowledgeMode)
         {
+
             case Session.CLIENT_ACKNOWLEDGE:
                 if (isNoConsume())
                 {
                     _session.acknowledgeMessage(msg.getDeliveryTag(), false);
                 }
+
                 break;
+
             case Session.DUPS_OK_ACKNOWLEDGE:
                 if (++_outstanding >= _prefetchHigh)
                 {
                     _dups_ok_acknowledge_send = true;
                 }
+
                 if (_outstanding <= _prefetchLow)
                 {
                     _dups_ok_acknowledge_send = false;
@@ -680,14 +703,18 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
                         _session.acknowledgeMessage(msg.getDeliveryTag(), true);
                     }
                 }
+
                 break;
+
             case Session.AUTO_ACKNOWLEDGE:
                 // we do not auto ack a message if the application code called recover()
                 if (!_session.isInRecovery())
                 {
                     _session.acknowledgeMessage(msg.getDeliveryTag(), false);
                 }
+
                 break;
+
             case Session.SESSION_TRANSACTED:
                 if (isNoConsume())
                 {
@@ -697,6 +724,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
                 {
                     _receivedDeliveryTags.add(msg.getDeliveryTag());
                 }
+
                 break;
         }
     }
@@ -721,14 +749,15 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
 
     void notifyError(Throwable cause)
     {
-//        synchronized (_closed)
+        // synchronized (_closed)
         {
             _closed.set(true);
             if (_logger.isTraceEnabled())
             {
                 if (_closedStack != null)
                 {
-                    _logger.trace(_consumerTag + " notifyError():" + Arrays.asList(Thread.currentThread().getStackTrace()).subList(3, 8));
+                    _logger.trace(_consumerTag + " notifyError():"
+                                  + Arrays.asList(Thread.currentThread().getStackTrace()).subList(3, 8));
                     _logger.trace(_consumerTag + " previously" + _closedStack.toString());
                 }
                 else
@@ -737,7 +766,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
                 }
             }
         }
-        //QPID-293 can "request redelivery of this error through dispatcher"
+        // QPID-293 can "request redelivery of this error through dispatcher"
 
         // we have no way of propagating the exception to a message listener - a JMS limitation - so we
         // deal with the case where we have a synchronous receive() waiting for a message to arrive
@@ -749,10 +778,10 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
                 _logger.debug("Passed exception to synchronous queue for propagation to receive()");
             }
         }
+
         deregisterConsumer();
     }
 
-
     /**
      * Perform cleanup to deregister this consumer. This occurs when closing the consumer in both the clean case and in
      * the case of an error occurring.
@@ -782,7 +811,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
 
         this.checkNotClosed();
 
-        if (_session == null || _session.isClosed())
+        if ((_session == null) || _session.isClosed())
         {
             throw new javax.jms.IllegalStateException("Invalid Session");
         }
@@ -817,7 +846,6 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
         return _autoClose;
     }
 
-
     public boolean isNoConsume()
     {
         return _noConsume;
@@ -827,10 +855,7 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
     {
         _closeWhenNoMessages = b;
 
-        if (_closeWhenNoMessages
-            && _synchronousQueue.isEmpty()
-            && _receiving.get()
-            && _messageListener != null)
+        if (_closeWhenNoMessages && _synchronousQueue.isEmpty() && _receiving.get() && (_messageListener != null))
         {
             _receivingThread.interrupt();
         }
@@ -846,13 +871,13 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
             _logger.debug("Rejecting received messages in _receivedDTs (RQ)");
         }
 
-        //rollback received but not committed messages
+        // rollback received but not committed messages
         while (!_receivedDeliveryTags.isEmpty())
         {
             if (_logger.isDebugEnabled())
             {
-                _logger.debug("Rejecting the messages(" + _receivedDeliveryTags.size() + ") in _receivedDTs (RQ)" +
-                              "for consumer with tag:" + _consumerTag);
+                _logger.debug("Rejecting the messages(" + _receivedDeliveryTags.size() + ") in _receivedDTs (RQ)"
+                              + "for consumer with tag:" + _consumerTag);
             }
 
             Long tag = _receivedDeliveryTags.poll();
@@ -876,14 +901,15 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
             }
         }
 
-        //rollback pending messages
+        // rollback pending messages
         if (_synchronousQueue.size() > 0)
         {
             if (_logger.isDebugEnabled())
             {
-                _logger.debug("Rejecting the messages(" + _synchronousQueue.size() + ") in _syncQueue (PRQ)" +
-                              "for consumer with tag:" + _consumerTag);
+                _logger.debug("Rejecting the messages(" + _synchronousQueue.size() + ") in _syncQueue (PRQ)"
+                              + "for consumer with tag:" + _consumerTag);
             }
+
             Iterator iterator = _synchronousQueue.iterator();
 
             while (iterator.hasNext())
@@ -898,13 +924,14 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
                     {
                         _logger.trace("Rejected message:" + ((AbstractJMSMessage) o).getDeliveryTag());
                     }
+
                     iterator.remove();
 
                 }
                 else
                 {
-                    _logger.error("Queue contained a :" + o.getClass() +
-                                  " unable to reject as it is not an AbstractJMSMessage. Will be cleared");
+                    _logger.error("Queue contained a :" + o.getClass()
+                                  + " unable to reject as it is not an AbstractJMSMessage. Will be cleared");
                     iterator.remove();
                 }
             }
@@ -919,7 +946,6 @@ public class BasicMessageConsumer extends Closeable implements MessageConsumer
         }
     }
 
-
     public String debugIdentity()
     {
         return String.valueOf(_consumerTag);
diff --git a/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java b/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
index b01e087ce1..bd7cc94582 100644
--- a/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
+++ b/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
@@ -21,6 +21,7 @@
 package org.apache.qpid.client;
 
 import java.io.UnsupportedEncodingException;
+import java.util.UUID;
 
 import javax.jms.BytesMessage;
 import javax.jms.DeliveryMode;
@@ -118,6 +119,9 @@ public class BasicMessageProducer extends Closeable implements org.apache.qpid.j
     private final boolean _mandatory;
 
     private final boolean _waitUntilSent;
+    
+    private boolean _disableMessageId;
+        
     private static final ContentBody[] NO_CONTENT_BODIES = new ContentBody[0];
 
     protected BasicMessageProducer(AMQConnection connection, AMQDestination destination, boolean transacted, int channelId,
@@ -172,15 +176,14 @@ public class BasicMessageProducer extends Closeable implements org.apache.qpid.j
     {
         checkPreConditions();
         checkNotClosed();
-        // IGNORED
+        _disableMessageId = b;
     }
 
     public boolean getDisableMessageID() throws JMSException
     {
         checkNotClosed();
 
-        // Always false for AMQP
-        return false;
+        return _disableMessageId;
     }
 
     public void setDisableMessageTimestamp(boolean b) throws JMSException
@@ -450,6 +453,18 @@ public class BasicMessageProducer extends Closeable implements org.apache.qpid.j
         origMessage.setJMSDestination(destination);
 
         AbstractJMSMessage message = convertToNativeMessage(origMessage);
+        
+        if(_disableMessageId)
+        {
+        	message.setJMSMessageID(null);
+        }
+        else
+        {
+        	if (message.getJMSMessageID() == null)
+        	{
+        		message.setJMSMessageID(UUID.randomUUID().toString());
+        	}
+        }
 
         int type;
         if (destination instanceof Topic)
@@ -667,4 +682,9 @@ public class BasicMessageProducer extends Closeable implements org.apache.qpid.j
     {
         return _session;
     }
+
+    public boolean isBound(AMQDestination destination) throws JMSException
+    {
+        return _session.isQueueBound(destination.getExchangeName(),null,destination.getRoutingKey());
+    }
 }
diff --git a/java/client/src/main/java/org/apache/qpid/client/JMSAMQException.java b/java/client/src/main/java/org/apache/qpid/client/JMSAMQException.java
index d2ab6bd2c2..d1237cff49 100644
--- a/java/client/src/main/java/org/apache/qpid/client/JMSAMQException.java
+++ b/java/client/src/main/java/org/apache/qpid/client/JMSAMQException.java
@@ -1,5 +1,25 @@
 /*
  *
+ * 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.
+ *
+ */
+/*
+ *
  * Copyright (c) 2006 The Apache Software Foundation
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -22,10 +42,35 @@ import javax.jms.JMSException;
 import org.apache.qpid.AMQException;
 
 /**
+ * JMSException does not accept wrapped exceptions in its constructor. Presumably this is because it is a relatively old
+ * Java exception class, before this was added as a default to Throwable. This exception class accepts wrapped exceptions
+ * as well as error messages, through its constructor, but is a JMSException.
+ *
+ * <p/><table id="crc"><caption>CRC Card</caption>
+ * <tr><th> Responsibilities <th> Collaborations
+ * <tr><td> Accept wrapped exceptions as a JMSException.
+ * </table>
+ *
  * @author Apache Software Foundation
  */
 public class JMSAMQException extends JMSException
 {
+    /**
+     * Creates a JMSException, wrapping another exception class.
+     *
+     * @param message The error message.
+     * @param cause   The underlying exception that caused this one. May be null if none is to be set.
+     */
+    public JMSAMQException(String message, Exception cause)
+    {
+        super(message);
+
+        if (cause != null)
+        {
+            setLinkedException(cause);
+        }
+    }
+
     public JMSAMQException(AMQException s)
     {
         super(s.getMessage(), String.valueOf(s.getErrorCode()));
diff --git a/java/client/src/main/java/org/apache/qpid/client/QueueSenderAdapter.java b/java/client/src/main/java/org/apache/qpid/client/QueueSenderAdapter.java
index c9d29d8077..e0c4b61333 100644
--- a/java/client/src/main/java/org/apache/qpid/client/QueueSenderAdapter.java
+++ b/java/client/src/main/java/org/apache/qpid/client/QueueSenderAdapter.java
@@ -7,14 +7,15 @@ import javax.jms.Message;
 import javax.jms.MessageProducer;
 import javax.jms.Queue;
 import javax.jms.QueueSender;
+import javax.jms.InvalidDestinationException;
 
 public class QueueSenderAdapter implements QueueSender {
 
-	private MessageProducer _delegate;
+	private BasicMessageProducer _delegate;
 	private Queue _queue;
 	private boolean closed = false;
 	
-	public QueueSenderAdapter(MessageProducer msgProducer, Queue queue){
+	public QueueSenderAdapter(BasicMessageProducer msgProducer, Queue queue){
 		_delegate = msgProducer;
 		_queue = queue;
 	}
@@ -122,12 +123,13 @@ public class QueueSenderAdapter implements QueueSender {
 		_delegate.setTimeToLive(timeToLive);
 	}
 
-    private void checkPreConditions() throws IllegalStateException, IllegalStateException
+    private void checkPreConditions() throws JMSException
     {
         checkPreConditions(_queue);
     }
 
-    private void checkPreConditions(Queue queue) throws IllegalStateException, IllegalStateException {
+    private void checkPreConditions(Queue queue) throws JMSException
+    {
 		if (closed){
 			throw new javax.jms.IllegalStateException("Publisher is closed");
 		}
@@ -137,5 +139,28 @@ public class QueueSenderAdapter implements QueueSender {
 		if(session == null || session.isClosed()){
 			throw new javax.jms.IllegalStateException("Invalid Session");
 		}
-	}
+
+        if(!(queue instanceof AMQDestination))
+        {
+            throw new InvalidDestinationException("Queue: " + queue + " is not a valid Qpid queue");
+        }
+        AMQDestination destination = (AMQDestination) queue;
+        if(!destination.isValidated() && checkQueueBeforePublish())
+        {
+
+            if (_delegate.isBound(destination))
+            {
+                destination.setValidated(true);
+            }
+            else
+            {
+                throw new InvalidDestinationException("Queue: " + queue + " is not a valid destination (no bindings on server");
+            }
+        }
+    }
+
+    private boolean checkQueueBeforePublish()
+    {
+        return "true".equalsIgnoreCase(System.getProperty("org.apache.qpid.client.verifyQueueBindingBeforePublish", "true"));
+    }
 }
diff --git a/java/client/src/main/java/org/apache/qpid/client/TopicPublisherAdapter.java b/java/client/src/main/java/org/apache/qpid/client/TopicPublisherAdapter.java
index f67b984658..02a408465b 100644
--- a/java/client/src/main/java/org/apache/qpid/client/TopicPublisherAdapter.java
+++ b/java/client/src/main/java/org/apache/qpid/client/TopicPublisherAdapter.java
@@ -175,5 +175,10 @@ public class TopicPublisherAdapter implements TopicPublisher
         {
             throw new InvalidDestinationException("Destination " + topic + " is not a topic");
         }
+        if(!(topic instanceof AMQDestination))
+        {
+            throw new InvalidDestinationException("Destination " + topic + " is not a Qpid topic");
+        }
+
     }
 }
diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java
index e2b101ab79..f62baf2c3a 100644
--- a/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java
+++ b/java/client/src/main/java/org/apache/qpid/client/handler/ChannelCloseMethodHandler.java
@@ -94,6 +94,8 @@ public class ChannelCloseMethodHandler implements StateAwareMethodListener
             }
 
         }
+        //fixme why is this only done when the close is expected...
+        // should the above forced closes not also cause a close?
         protocolSession.channelClosed(evt.getChannelId(), errorCode, String.valueOf(reason));
     }
 }
diff --git a/java/client/src/main/java/org/apache/qpid/client/message/AbstractBytesMessage.java b/java/client/src/main/java/org/apache/qpid/client/message/AbstractBytesMessage.java
index 8938130417..af254fbbaf 100644
--- a/java/client/src/main/java/org/apache/qpid/client/message/AbstractBytesMessage.java
+++ b/java/client/src/main/java/org/apache/qpid/client/message/AbstractBytesMessage.java
@@ -27,6 +27,7 @@ import javax.jms.JMSException;
 import javax.jms.MessageEOFException;
 
 import org.apache.mina.common.ByteBuffer;
+
 import org.apache.qpid.AMQException;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.BasicContentHeaderProperties;
@@ -72,7 +73,7 @@ public abstract class AbstractBytesMessage extends AbstractJMSMessage
     }
 
     AbstractBytesMessage(long messageNbr, ContentHeaderBody contentHeader, AMQShortString exchange,
-                         AMQShortString routingKey, ByteBuffer data) throws AMQException
+        AMQShortString routingKey, ByteBuffer data) throws AMQException
     {
         // TODO: this casting is ugly. Need to review whole ContentHeaderBody idea
         super(messageNbr, (BasicContentHeaderProperties) contentHeader.properties, exchange, routingKey, data);
@@ -93,7 +94,9 @@ public abstract class AbstractBytesMessage extends AbstractJMSMessage
         }
         catch (IOException e)
         {
-            throw new JMSException(e.toString());
+            JMSException jmse = new JMSException(e.toString());
+            jmse.setLinkedException(e);
+            throw jmse;
         }
     }
 
@@ -112,6 +115,7 @@ public abstract class AbstractBytesMessage extends AbstractJMSMessage
         {
             return null;
         }
+
         int pos = _data.position();
         _data.rewind();
         // one byte left is for the end of frame marker
@@ -119,12 +123,14 @@ public abstract class AbstractBytesMessage extends AbstractJMSMessage
         {
             // this is really redundant since pos must be zero
             _data.position(pos);
+
             return null;
         }
         else
         {
             String data = _data.getString(Charset.forName("UTF8").newDecoder());
             _data.position(pos);
+
             return data;
         }
     }
diff --git a/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java b/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
index 66524edce3..f87b4027f6 100644
--- a/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
+++ b/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessage.java
@@ -23,6 +23,7 @@ package org.apache.qpid.client.message;
 import java.util.Collections;
 import java.util.Enumeration;
 import java.util.Map;
+import java.util.UUID;
 
 import javax.jms.Destination;
 import javax.jms.JMSException;
@@ -32,12 +33,7 @@ import javax.jms.MessageNotWriteableException;
 import org.apache.commons.collections.map.ReferenceMap;
 import org.apache.mina.common.ByteBuffer;
 import org.apache.qpid.AMQException;
-import org.apache.qpid.client.AMQDestination;
-import org.apache.qpid.client.AMQQueue;
-import org.apache.qpid.client.AMQTopic;
-import org.apache.qpid.client.AMQUndefinedDestination;
-import org.apache.qpid.client.BasicMessageConsumer;
-import org.apache.qpid.client.CustomJMSXProperty;
+import org.apache.qpid.client.*;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.BasicContentHeaderProperties;
 import org.apache.qpid.framing.FieldTable;
@@ -123,7 +119,7 @@ public abstract class AbstractJMSMessage extends AMQMessage implements org.apach
     {
         if (getContentHeaderProperties().getMessageIdAsString() == null)
         {
-            getContentHeaderProperties().setMessageId("ID:" + _deliveryTag);
+            getContentHeaderProperties().setMessageId("ID:" + UUID.randomUUID());
         }
 
         return getContentHeaderProperties().getMessageIdAsString();
@@ -183,7 +179,7 @@ public abstract class AbstractJMSMessage extends AMQMessage implements org.apach
                 }
                 catch (URLSyntaxException e)
                 {
-                    throw new JMSException("Illegal value in JMS_ReplyTo property: " + replyToEncoding);
+                    throw new JMSAMQException("Illegal value in JMS_ReplyTo property: " + replyToEncoding, e);
                 }
 
                 _destinationCache.put(replyToEncoding, dest);
diff --git a/java/client/src/main/java/org/apache/qpid/client/message/JMSHeaderAdapter.java b/java/client/src/main/java/org/apache/qpid/client/message/JMSHeaderAdapter.java
index 6352f7029f..348a0bd152 100644
--- a/java/client/src/main/java/org/apache/qpid/client/message/JMSHeaderAdapter.java
+++ b/java/client/src/main/java/org/apache/qpid/client/message/JMSHeaderAdapter.java
@@ -384,7 +384,9 @@ public final class JMSHeaderAdapter
         }

         catch (AMQPInvalidClassException aice)

         {

-            throw new MessageFormatException("Only primatives are allowed object is:" + object.getClass());

+            MessageFormatException mfe = new MessageFormatException("Only primatives are allowed object is:" + object.getClass());

+            mfe.setLinkedException(aice);

+            throw mfe;

         }

     }

 

diff --git a/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessage.java b/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessage.java
index df1400b167..caf8741280 100644
--- a/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessage.java
+++ b/java/client/src/main/java/org/apache/qpid/client/message/JMSObjectMessage.java
@@ -33,6 +33,7 @@ import javax.jms.MessageFormatException;
 import javax.jms.ObjectMessage;
 
 import org.apache.mina.common.ByteBuffer;
+
 import org.apache.qpid.AMQException;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.BasicContentHeaderProperties;
@@ -61,14 +62,15 @@ public class JMSObjectMessage extends AbstractJMSMessage implements ObjectMessag
             _data = ByteBuffer.allocate(DEFAULT_BUFFER_SIZE);
             _data.setAutoExpand(true);
         }
+
         getContentHeaderProperties().setContentType(MIME_TYPE_SHORT_STRING);
     }
 
     /**
      * Creates read only message for delivery to consumers
      */
-    JMSObjectMessage(long messageNbr, ContentHeaderBody contentHeader, AMQShortString exchange,
-                     AMQShortString routingKey, ByteBuffer data) throws AMQException
+    JMSObjectMessage(long messageNbr, ContentHeaderBody contentHeader, AMQShortString exchange, AMQShortString routingKey,
+        ByteBuffer data) throws AMQException
     {
         super(messageNbr, (BasicContentHeaderProperties) contentHeader.properties, exchange, routingKey, data);
     }
@@ -79,6 +81,7 @@ public class JMSObjectMessage extends AbstractJMSMessage implements ObjectMessag
         {
             _data.release();
         }
+
         _data = null;
 
     }
@@ -116,11 +119,13 @@ public class JMSObjectMessage extends AbstractJMSMessage implements ObjectMessag
         }
         catch (IOException e)
         {
-            throw new MessageFormatException("Message not serializable: " + e);
+            MessageFormatException mfe = new MessageFormatException("Message not serializable: " + e);
+            mfe.setLinkedException(e);
+            throw mfe;
         }
 
     }
-  
+
     public Serializable getObject() throws JMSException
     {
         ObjectInputStream in = null;
@@ -133,17 +138,20 @@ public class JMSObjectMessage extends AbstractJMSMessage implements ObjectMessag
         {
             _data.rewind();
             in = new ObjectInputStream(_data.asInputStream());
+
             return (Serializable) in.readObject();
         }
         catch (IOException e)
         {
-            e.printStackTrace();
-            throw new MessageFormatException("Could not deserialize message: " + e);
+            MessageFormatException mfe = new MessageFormatException("Could not deserialize message: " + e);
+            mfe.setLinkedException(e);
+            throw mfe;
         }
         catch (ClassNotFoundException e)
         {
-            e.printStackTrace();
-            throw new MessageFormatException("Could not deserialize message: " + e);
+            MessageFormatException mfe = new MessageFormatException("Could not deserialize message: " + e);
+            mfe.setLinkedException(e);
+            throw mfe;
         }
         finally
         {
@@ -162,8 +170,7 @@ public class JMSObjectMessage extends AbstractJMSMessage implements ObjectMessag
             }
         }
         catch (IOException ignore)
-        {
-        }
+        { }
     }
 
     private static String toString(ByteBuffer data)
@@ -172,6 +179,7 @@ public class JMSObjectMessage extends AbstractJMSMessage implements ObjectMessag
         {
             return null;
         }
+
         int pos = data.position();
         try
         {
diff --git a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
index d0cc52271a..5bc1555df7 100644
--- a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
+++ b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
@@ -34,6 +34,7 @@ import org.apache.qpid.AMQConnectionClosedException;
 import org.apache.qpid.AMQDisconnectedException;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.AMQTimeoutException;
+import org.apache.qpid.AMQChannelClosedException;
 import org.apache.qpid.client.AMQConnection;
 import org.apache.qpid.client.AMQSession;
 import org.apache.qpid.client.SSLConfiguration;
@@ -248,6 +249,12 @@ public class AMQProtocolHandler extends IoHandlerAdapter
 
                 sessionClosed(session);
             }
+
+            //FIXME Need to correctly handle other exceptions. Things like ...
+//            if (cause instanceof AMQChannelClosedException)
+            // which will cause the JMSSession to end due to a channel close and so that Session needs
+            // to be removed from the map so we can correctly still call close without an exception when trying to close
+            // the server closed session.  See also CloseChannelMethodHandler as the sessionClose is never called on exception
         }
         // we reach this point if failover was attempted and failed therefore we need to let the calling app
         // know since we cannot recover the situation
@@ -510,11 +517,6 @@ public class AMQProtocolHandler extends IoHandlerAdapter
         _protocolSession.closeSession(session);
     }
 
-    public void closeConnection() throws AMQException
-    {
-        closeConnection(-1);
-    }
-
     public void closeConnection(long timeout) throws AMQException
     {
         getStateManager().changeState(AMQState.CONNECTION_CLOSING);
diff --git a/java/client/src/main/java/org/apache/qpid/client/security/CallbackHandlerRegistry.properties b/java/client/src/main/java/org/apache/qpid/client/security/CallbackHandlerRegistry.properties
index 50e6f1efaa..89ee8337f8 100644
--- a/java/client/src/main/java/org/apache/qpid/client/security/CallbackHandlerRegistry.properties
+++ b/java/client/src/main/java/org/apache/qpid/client/security/CallbackHandlerRegistry.properties
@@ -16,5 +16,6 @@
 # specific language governing permissions and limitations
 # under the License.
 #
+CallbackHandler.CRAM-MD5-HASHED=org.apache.qpid.client.security.UsernameHashedPasswordCallbackHandler
 CallbackHandler.CRAM-MD5=org.apache.qpid.client.security.UsernamePasswordCallbackHandler
 CallbackHandler.PLAIN=org.apache.qpid.client.security.UsernamePasswordCallbackHandler
diff --git a/java/client/src/main/java/org/apache/qpid/client/security/DynamicSaslRegistrar.java b/java/client/src/main/java/org/apache/qpid/client/security/DynamicSaslRegistrar.java
index f8ee22a5d9..04db8044de 100644
--- a/java/client/src/main/java/org/apache/qpid/client/security/DynamicSaslRegistrar.java
+++ b/java/client/src/main/java/org/apache/qpid/client/security/DynamicSaslRegistrar.java
@@ -20,10 +20,6 @@
  */
 package org.apache.qpid.client.security;
 
-import java.io.BufferedInputStream;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.security.Security;
@@ -34,6 +30,7 @@ import java.util.TreeMap;
 
 import javax.security.sasl.SaslClientFactory;
 
+
 import org.apache.log4j.Logger;
 
 import org.apache.qpid.util.FileUtils;
@@ -50,14 +47,11 @@ import org.apache.qpid.util.FileUtils;
  * mechanism=fully.qualified.class.name
  * </pre>
  *
- * <p/>Where mechanism is an IANA-registered mechanism name and the fully qualified class name refers to a
- * class that implements javax.security.sasl.SaslClientFactory and provides the specified mechanism.
+ * <p/>Where mechanism is an IANA-registered mechanism name and the fully qualified class name refers to a class that
+ * implements javax.security.sasl.SaslClientFactory and provides the specified mechanism.
  *
- * <p><table id="crc"><caption>CRC Card</caption>
- * <tr><th> Responsibilities <th> Collaborations
- * <tr><td> Parse SASL mechanism properties.
- * <tr><td> Create and register security provider for SASL mechanisms.
- * </table>
+ * <p><table id="crc"><caption>CRC Card</caption> <tr><th> Responsibilities <th> Collaborations <tr><td> Parse SASL
+ * mechanism properties. <tr><td> Create and register security provider for SASL mechanisms. </table>
  */
 public class DynamicSaslRegistrar
 {
@@ -69,10 +63,7 @@ public class DynamicSaslRegistrar
     /** The default name of the SASL properties file resource. */
     public static final String DEFAULT_RESOURCE_NAME = "org/apache/qpid/client/security/DynamicSaslRegistrar.properties";
 
-    /**
-     * Reads the properties file, and creates a dynamic security provider to register the SASL implementations
-     * with.
-     */
+    /** Reads the properties file, and creates a dynamic security provider to register the SASL implementations with. */
     public static void registerSaslProviders()
     {
         _logger.debug("public static void registerSaslProviders(): called");
@@ -80,8 +71,8 @@ public class DynamicSaslRegistrar
         // Open the SASL properties file, using the default name is one is not specified.
         String filename = System.getProperty(FILE_PROPERTY);
         InputStream is =
-            FileUtils.openFileOrDefaultResource(filename, DEFAULT_RESOURCE_NAME,
-                                                DynamicSaslRegistrar.class.getClassLoader());
+                FileUtils.openFileOrDefaultResource(filename, DEFAULT_RESOURCE_NAME,
+                                                    DynamicSaslRegistrar.class.getClassLoader());
 
         try
         {
@@ -94,7 +85,7 @@ public class DynamicSaslRegistrar
 
             if (factories.size() > 0)
             {
-                Security.addProvider(new JCAProvider(factories));
+                Security.insertProviderAt(new JCAProvider(factories), 0);
                 _logger.debug("Dynamic SASL provider added as a security provider");
             }
         }
@@ -170,15 +161,15 @@ public class DynamicSaslRegistrar
      * @return A map from SASL mechanism names to implementing client factory classes.
      *
      * @todo Why tree map here? Do really want mechanisms in alphabetical order? Seems more likely that the declared
-     *       order of the mechanisms is intended to be preserved, so that they are registered in the declared order
-     *       of preference. Consider LinkedHashMap instead.
+     * order of the mechanisms is intended to be preserved, so that they are registered in the declared order of
+     * preference. Consider LinkedHashMap instead.
      */
     private static Map<String, Class<? extends SaslClientFactory>> parseProperties(Properties props)
     {
         Enumeration e = props.propertyNames();
 
         TreeMap<String, Class<? extends SaslClientFactory>> factoriesToRegister =
-            new TreeMap<String, Class<? extends SaslClientFactory>>();
+                new TreeMap<String, Class<? extends SaslClientFactory>>();
 
         while (e.hasMoreElements())
         {
diff --git a/java/client/src/main/java/org/apache/qpid/client/security/DynamicSaslRegistrar.properties b/java/client/src/main/java/org/apache/qpid/client/security/DynamicSaslRegistrar.properties
index c2a7d7928c..1bff43142b 100644
--- a/java/client/src/main/java/org/apache/qpid/client/security/DynamicSaslRegistrar.properties
+++ b/java/client/src/main/java/org/apache/qpid/client/security/DynamicSaslRegistrar.properties
@@ -17,3 +17,4 @@
 # under the License.
 #
 AMQPLAIN=org.apache.qpid.client.security.amqplain.AmqPlainSaslClientFactory
+CRAM-MD5-HASHED=org.apache.qpid.client.security.crammd5hashed.CRAMMD5HashedSaslClientFactory
diff --git a/java/client/src/main/java/org/apache/qpid/client/security/JCAProvider.java b/java/client/src/main/java/org/apache/qpid/client/security/JCAProvider.java
index 2fa8dcddde..5bf120454e 100644
--- a/java/client/src/main/java/org/apache/qpid/client/security/JCAProvider.java
+++ b/java/client/src/main/java/org/apache/qpid/client/security/JCAProvider.java
@@ -52,7 +52,7 @@ public class JCAProvider extends Provider
         super("AMQSASLProvider", 1.0, "A JCA provider that registers all "
               + "AMQ SASL providers that want to be registered");
         register(providerMap);
-        Security.addProvider(this);
+//        Security.addProvider(this);
     }
 
     /**
diff --git a/java/client/src/main/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandler.java b/java/client/src/main/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandler.java
new file mode 100644
index 0000000000..46323e8c09
--- /dev/null
+++ b/java/client/src/main/java/org/apache/qpid/client/security/UsernameHashedPasswordCallbackHandler.java
@@ -0,0 +1,104 @@
+/*
+ *
+ * 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.client.security;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.PasswordCallback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.sasl.RealmCallback;
+
+import com.sun.crypto.provider.HmacMD5;
+
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.client.protocol.AMQProtocolSession;
+
+public class UsernameHashedPasswordCallbackHandler implements AMQCallbackHandler
+{
+    private static final Logger _logger = Logger.getLogger(UsernameHashedPasswordCallbackHandler.class);
+
+    private AMQProtocolSession _protocolSession;
+
+    public void initialise(AMQProtocolSession protocolSession)
+    {
+        _protocolSession = protocolSession;
+    }
+
+    public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException
+    {
+        for (int i = 0; i < callbacks.length; i++)
+        {
+            Callback cb = callbacks[i];
+            if (cb instanceof NameCallback)
+            {
+                ((NameCallback) cb).setName(_protocolSession.getUsername());
+            }
+            else if (cb instanceof PasswordCallback)
+            {
+                try
+                {
+                    ((PasswordCallback) cb).setPassword(getHash(_protocolSession.getPassword()));
+                }
+                catch (NoSuchAlgorithmException e)
+                {
+                    UnsupportedCallbackException uce = new UnsupportedCallbackException(cb);
+                    uce.initCause(e);
+                    throw uce;
+                }
+            }
+            else
+            {
+                throw new UnsupportedCallbackException(cb);
+            }
+        }
+    }
+
+    private char[] getHash(String text) throws NoSuchAlgorithmException, UnsupportedEncodingException
+    {
+
+        byte[] data = text.getBytes("utf-8");
+
+        MessageDigest md = MessageDigest.getInstance("MD5");
+
+        for (byte b : data)
+        {
+            md.update(b);
+        }
+
+        byte[] digest = md.digest();
+
+        char[] hash = new char[digest.length ];
+
+        int index = 0;
+        for (byte b : digest)
+        {            
+            hash[index++] = (char) b;
+        }
+
+        return hash;
+    }
+}
diff --git a/java/client/src/main/java/org/apache/qpid/client/security/crammd5hashed/CRAMMD5HashedSaslClientFactory.java b/java/client/src/main/java/org/apache/qpid/client/security/crammd5hashed/CRAMMD5HashedSaslClientFactory.java
new file mode 100644
index 0000000000..22bb1ac156
--- /dev/null
+++ b/java/client/src/main/java/org/apache/qpid/client/security/crammd5hashed/CRAMMD5HashedSaslClientFactory.java
@@ -0,0 +1,72 @@
+/*
+ *  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.client.security.crammd5hashed;
+
+import org.apache.qpid.client.security.amqplain.AmqPlainSaslClient;
+
+import javax.security.sasl.SaslClientFactory;
+import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.Sasl;
+import javax.security.auth.callback.CallbackHandler;
+import java.util.Map;
+import java.security.Security;
+
+public class CRAMMD5HashedSaslClientFactory implements SaslClientFactory
+{
+    /** The name of this mechanism */
+    public static final String MECHANISM = "CRAM-MD5-HASHED";
+
+
+    public SaslClient createSaslClient(String[] mechanisms, String authorizationId, String protocol, String serverName, Map<String, ?> props, CallbackHandler cbh) throws SaslException
+    {
+        for (int i = 0; i < mechanisms.length; i++)
+        {
+            if (mechanisms[i].equals(MECHANISM))
+            {
+                if (cbh == null)
+                {
+                    throw new SaslException("CallbackHandler must not be null");
+                }
+
+                String[] mechs = {"CRAM-MD5"};
+                return Sasl.createSaslClient(mechs, authorizationId, protocol, serverName, props, cbh);
+            }
+        }
+        return null;
+    }
+
+    public String[] getMechanismNames(Map props)
+    {
+        if (props != null)
+        {
+            if (props.containsKey(Sasl.POLICY_NOPLAINTEXT) ||
+                props.containsKey(Sasl.POLICY_NODICTIONARY) ||
+                props.containsKey(Sasl.POLICY_NOACTIVE))
+            {
+                // returned array must be non null according to interface documentation
+                return new String[0];
+            }
+        }
+
+        return new String[]{MECHANISM};
+    }
+}
diff --git a/java/client/src/main/java/org/apache/qpid/client/transport/AMQNoTransportForProtocolException.java b/java/client/src/main/java/org/apache/qpid/client/transport/AMQNoTransportForProtocolException.java
index 104c5bfc44..1ec3adc2eb 100644
--- a/java/client/src/main/java/org/apache/qpid/client/transport/AMQNoTransportForProtocolException.java
+++ b/java/client/src/main/java/org/apache/qpid/client/transport/AMQNoTransportForProtocolException.java
@@ -7,9 +7,9 @@
  * 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
@@ -33,7 +33,7 @@ public class AMQNoTransportForProtocolException extends AMQTransportConnectionEx
 
     public AMQNoTransportForProtocolException(BrokerDetails details, String message)
     {
-        super(message);
+        super(null, message, null);
 
         _details = details;
     }
diff --git a/java/client/src/main/java/org/apache/qpid/client/transport/AMQTransportConnectionException.java b/java/client/src/main/java/org/apache/qpid/client/transport/AMQTransportConnectionException.java
index 4b17661bc3..fec7ff693c 100644
--- a/java/client/src/main/java/org/apache/qpid/client/transport/AMQTransportConnectionException.java
+++ b/java/client/src/main/java/org/apache/qpid/client/transport/AMQTransportConnectionException.java
@@ -7,9 +7,9 @@
  * 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
@@ -21,12 +21,12 @@
 package org.apache.qpid.client.transport;
 
 import org.apache.qpid.AMQException;
+import org.apache.qpid.protocol.AMQConstant;
 
 public class AMQTransportConnectionException extends AMQException
 {
-    public AMQTransportConnectionException(String message)
+    public AMQTransportConnectionException(AMQConstant errorCode, String message, Throwable cause)
     {
-        super(message);
-
+        super(errorCode, message, cause);
     }
 }
diff --git a/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java b/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java
index 8368eee125..0bc83e9804 100644
--- a/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java
+++ b/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java
@@ -7,9 +7,9 @@
  * 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
@@ -26,12 +26,14 @@ import java.util.Iterator;
 import java.util.Map;
 
 import org.apache.log4j.Logger;
+
 import org.apache.mina.common.IoConnector;
 import org.apache.mina.common.IoHandlerAdapter;
 import org.apache.mina.common.IoServiceConfig;
 import org.apache.mina.transport.socket.nio.SocketConnector;
 import org.apache.mina.transport.vmpipe.VmPipeAcceptor;
 import org.apache.mina.transport.vmpipe.VmPipeAddress;
+
 import org.apache.qpid.client.AMQBrokerDetails;
 import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
 import org.apache.qpid.jms.BrokerDetails;
@@ -64,13 +66,11 @@ public class TransportConnection
         int transport = getTransport(details.getTransport());
 
         if (transport == -1)
-
         {
             throw new AMQNoTransportForProtocolException(details);
         }
 
         if (transport == _currentInstance)
-
         {
             if (transport == VM)
             {
@@ -88,40 +88,42 @@ public class TransportConnection
         _currentInstance = transport;
 
         switch (transport)
-
         {
-            case TCP:
-                _instance = new SocketTransportConnection(new SocketTransportConnection.SocketConnectorFactory()
-                {
-                    public IoConnector newSocketConnector()
+
+        case TCP:
+            _instance = new SocketTransportConnection(new SocketTransportConnection.SocketConnectorFactory()
                     {
-                        SocketConnector result;
-                        //FIXME - this needs to be sorted to use the new Mina MultiThread SA.
-                        if (Boolean.getBoolean("qpidnio"))
+                        public IoConnector newSocketConnector()
                         {
-                            _logger.fatal("Using Qpid NIO - sysproperty 'qpidnio' is set.");
-//                            result = new org.apache.qpid.nio.SocketConnector(); // non-blocking connector
+                            SocketConnector result;
+                            // FIXME - this needs to be sorted to use the new Mina MultiThread SA.
+                            if (Boolean.getBoolean("qpidnio"))
+                            {
+                                _logger.fatal("Using Qpid NIO - sysproperty 'qpidnio' is set.");
+                                // result = new org.apache.qpid.nio.SocketConnector(); // non-blocking connector
+                            }
+                            // else
+
+                            {
+                                _logger.info("Using Mina NIO");
+                                result = new SocketConnector(); // non-blocking connector
+                            }
+
+                            // Don't have the connector's worker thread wait around for other connections (we only use
+                            // one SocketConnector per connection at the moment anyway). This allows short-running
+                            // clients (like unit tests) to complete quickly.
+                            result.setWorkerTimeout(0);
+
+                            return result;
                         }
-//                        else
-                        {
-                            _logger.info("Using Mina NIO");
-                            result = new SocketConnector(); // non-blocking connector
-                        }
-
-                        // Don't have the connector's worker thread wait around for other connections (we only use
-                        // one SocketConnector per connection at the moment anyway). This allows short-running
-                        // clients (like unit tests) to complete quickly.
-                        result.setWorkerTimeout(0);
+                    });
+            break;
 
-                        return result;
-                    }
-                });
-                break;
-            case VM:
-            {
-                _instance = getVMTransport(details, Boolean.getBoolean("amqj.AutoCreateVMBroker"));
-                break;
-            }
+        case VM:
+        {
+            _instance = getVMTransport(details, Boolean.getBoolean("amqj.AutoCreateVMBroker"));
+            break;
+        }
         }
 
         return _instance;
@@ -142,7 +144,8 @@ public class TransportConnection
         return -1;
     }
 
-    private static ITransportConnection getVMTransport(BrokerDetails details, boolean AutoCreate) throws AMQVMBrokerCreationException
+    private static ITransportConnection getVMTransport(BrokerDetails details, boolean AutoCreate)
+        throws AMQVMBrokerCreationException
     {
         int port = details.getPort();
 
@@ -154,14 +157,14 @@ public class TransportConnection
             }
             else
             {
-                throw new AMQVMBrokerCreationException(port, "VM Broker on port " + port + " does not exist. Auto create disabled.");
+                throw new AMQVMBrokerCreationException(null, port, "VM Broker on port " + port
+                    + " does not exist. Auto create disabled.", null);
             }
         }
 
         return new VmPipeTransportConnection(port);
     }
 
-
     public static void createVMBroker(int port) throws AMQVMBrokerCreationException
     {
         if (_acceptor == null)
@@ -192,7 +195,7 @@ public class TransportConnection
             {
                 _logger.error(e);
 
-                //Try and unbind provider
+                // Try and unbind provider
                 try
                 {
                     VmPipeAddress pipe = new VmPipeAddress(port);
@@ -203,7 +206,7 @@ public class TransportConnection
                     }
                     catch (Exception ignore)
                     {
-                        //ignore
+                        // ignore
                     }
 
                     if (provider == null)
@@ -227,7 +230,7 @@ public class TransportConnection
                         because = e.getCause().toString();
                     }
 
-                    throw new AMQVMBrokerCreationException(port, because + " Stopped binding of InVM Qpid.AMQP");
+                    throw new AMQVMBrokerCreationException(null, port, because + " Stopped binding of InVM Qpid.AMQP", e);
                 }
             }
         }
@@ -246,14 +249,14 @@ public class TransportConnection
         // can't use introspection to get Provider as it is a server class.
         // need to go straight to IoHandlerAdapter but that requries the queues and exchange from the ApplicationRegistry which we can't access.
 
-        //get right constructor and pass in instancec ID - "port"
+        // get right constructor and pass in instancec ID - "port"
         IoHandlerAdapter provider;
         try
         {
-            Class[] cnstr = {Integer.class};
-            Object[] params = {port};
+            Class[] cnstr = { Integer.class };
+            Object[] params = { port };
             provider = (IoHandlerAdapter) Class.forName(protocolProviderClass).getConstructor(cnstr).newInstance(params);
-            //Give the broker a second to create
+            // Give the broker a second to create
             _logger.info("Created VMBroker Instance:" + port);
         }
         catch (Exception e)
@@ -270,8 +273,10 @@ public class TransportConnection
                 because = e.getCause().toString();
             }
 
-
-            throw new AMQVMBrokerCreationException(port, because + " Stopped InVM Qpid.AMQP creation");
+            AMQVMBrokerCreationException amqbce =
+                new AMQVMBrokerCreationException(null, port, because + " Stopped InVM Qpid.AMQP creation", null);
+            amqbce.initCause(e);
+            throw amqbce;
         }
 
         return provider;
diff --git a/java/client/src/main/java/org/apache/qpid/client/vmbroker/AMQVMBrokerCreationException.java b/java/client/src/main/java/org/apache/qpid/client/vmbroker/AMQVMBrokerCreationException.java
index 607ddcc26a..4b2982fe9c 100644
--- a/java/client/src/main/java/org/apache/qpid/client/vmbroker/AMQVMBrokerCreationException.java
+++ b/java/client/src/main/java/org/apache/qpid/client/vmbroker/AMQVMBrokerCreationException.java
@@ -7,9 +7,9 @@
  * 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
@@ -21,19 +21,25 @@
 package org.apache.qpid.client.vmbroker;
 
 import org.apache.qpid.client.transport.AMQTransportConnectionException;
+import org.apache.qpid.protocol.AMQConstant;
 
 public class AMQVMBrokerCreationException extends AMQTransportConnectionException
 {
     private int _port;
 
+    /**
+     * @param port
+     *
+     * @deprecated
+     */
     public AMQVMBrokerCreationException(int port)
     {
-        this(port, "Unable to create vm broker");
+        this(null, port, "Unable to create vm broker", null);
     }
 
-    public AMQVMBrokerCreationException(int port, String message)
+    public AMQVMBrokerCreationException(AMQConstant errorCode, int port, String message, Throwable cause)
     {
-        super(message);
+        super(errorCode, message, cause);
         _port = port;
     }
 
diff --git a/java/client/src/main/java/org/apache/qpid/jms/FailoverPolicy.java b/java/client/src/main/java/org/apache/qpid/jms/FailoverPolicy.java
index 9adf04e182..6ad3fb4bae 100644
--- a/java/client/src/main/java/org/apache/qpid/jms/FailoverPolicy.java
+++ b/java/client/src/main/java/org/apache/qpid/jms/FailoverPolicy.java
@@ -101,7 +101,7 @@ public class FailoverPolicy
                     }
                     catch (Exception cnfe)
                     {
-                        throw new IllegalArgumentException("Unknown failover method:" + failoverMethod);
+                        throw new IllegalArgumentException("Unknown failover method:" + failoverMethod, cnfe);
                     }
                 }
             }
diff --git a/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerTest.java b/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerTest.java
index a406f9f86e..794fd5c8c1 100644
--- a/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerTest.java
+++ b/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerTest.java
@@ -65,6 +65,7 @@ public class MessageListenerMultiConsumerTest extends TestCase
 
     private final CountDownLatch _allMessagesSent = new CountDownLatch(2); //all messages Sent Lock
 
+
     protected void setUp() throws Exception
     {
         super.setUp();
@@ -122,30 +123,39 @@ public class MessageListenerMultiConsumerTest extends TestCase
         TransportConnection.killAllVMBrokers();
     }
 
+//    public void testRecieveC1thenC2() throws Exception
+//    {
+//
+//        for (int msg = 0; msg < MSG_COUNT / 2; msg++)
+//        {
+//
+//            assertTrue(_consumer1.receive() != null);
+//        }
+//
+//        for (int msg = 0; msg < MSG_COUNT / 2; msg++)
+//        {
+//            assertTrue(_consumer2.receive() != null);
+//        }
+//    }
 
-    public void testRecieveC1thenC2() throws Exception
+    public void testRecieveInterleaved() throws Exception
     {
-
-        for (int msg = 0; msg < MSG_COUNT / 2; msg++)
+        int msg = 0;
+        int MAX_LOOPS = MSG_COUNT * 2;
+        for (int loops = 0; msg < MSG_COUNT || loops < MAX_LOOPS; loops++)
         {
 
-            assertTrue(_consumer1.receive() != null);
-        }
-
-        for (int msg = 0; msg < MSG_COUNT / 2; msg++)
-        {
-            assertTrue(_consumer2.receive() != null);
+            if (_consumer1.receive(100) != null)
+            {
+                msg++;
+            }
+            if (_consumer2.receive(100) != null)
+            {
+                msg++;
+            }
         }
-    }
-
-    public void testRecieveInterleaved() throws Exception
-    {
 
-        for (int msg = 0; msg < MSG_COUNT / 2; msg++)
-        {
-            assertTrue(_consumer1.receive() != null);
-            assertTrue(_consumer2.receive() != null);
-        }
+        assertEquals("Not all messages received.", MSG_COUNT, msg);
     }
 
 
@@ -161,7 +171,7 @@ public class MessageListenerMultiConsumerTest extends TestCase
 
                 if (receivedCount1 == MSG_COUNT / 2)
                 {
-                    _allMessagesSent.countDown();                    
+                    _allMessagesSent.countDown();
                 }
 
             }
@@ -196,6 +206,18 @@ public class MessageListenerMultiConsumerTest extends TestCase
         assertEquals(MSG_COUNT, receivedCount1 + receivedCount2);
     }
 
+    public void testRecieveC2Only_OnlyRunWith_REGISTER_CONSUMERS_FLOWED() throws Exception
+    {
+        if (Boolean.parseBoolean(System.getProperties().getProperty("REGISTER_CONSUMERS_FLOWED", "false")))
+        {
+            for (int msg = 0; msg < MSG_COUNT; msg++)
+            {
+                assertTrue(MSG_COUNT + " msg should be received. Only received:" + msg,
+                           _consumer2.receive(1000) != null);
+            }
+        }
+    }
+
 
     public static junit.framework.Test suite()
     {
diff --git a/java/client/src/test/java/org/apache/qpid/client/MessageListenerTest.java b/java/client/src/test/java/org/apache/qpid/client/MessageListenerTest.java
index 5fb77af4db..7b5957ac8c 100644
--- a/java/client/src/test/java/org/apache/qpid/client/MessageListenerTest.java
+++ b/java/client/src/test/java/org/apache/qpid/client/MessageListenerTest.java
@@ -144,6 +144,36 @@ public class MessageListenerTest extends TestCase implements MessageListener
 
     }
 
+    public void testRecieveTheUseMessageListener() throws Exception
+     {
+
+         _logger.error("Test disabled as initial receive is not called first");
+         // Perform initial receive to start connection
+//         assertTrue(_consumer.receive(2000) != null);
+//         receivedCount++;
+
+         // Sleep to ensure remaining 4 msgs end up on _synchronousQueue
+//         Thread.sleep(1000);
+
+         // Set the message listener and wait for the messages to come in.
+         _consumer.setMessageListener(this);
+
+         _logger.info("Waiting 3 seconds for messages");
+
+         try
+         {
+             _awaitMessages.await(3000, TimeUnit.MILLISECONDS);
+         }
+         catch (InterruptedException e)
+         {
+             //do nothing
+         }
+         //Should have recieved all async messages
+         assertEquals(MSG_COUNT, receivedCount);
+
+     }
+    
+
     public void onMessage(Message message)
     {
         _logger.info("Received Message(" + receivedCount + "):" + message);
diff --git a/java/client/src/test/java/org/apache/qpid/client/ResetMessageListenerTest.java b/java/client/src/test/java/org/apache/qpid/client/ResetMessageListenerTest.java
index 10bf1a8d6d..42594fff8e 100644
--- a/java/client/src/test/java/org/apache/qpid/client/ResetMessageListenerTest.java
+++ b/java/client/src/test/java/org/apache/qpid/client/ResetMessageListenerTest.java
@@ -83,7 +83,7 @@ public class ResetMessageListenerTest extends TestCase
         Hashtable<String, String> env = new Hashtable<String, String>();
 
         env.put("connectionfactory.connection", "amqp://guest:guest@MLT_ID/test?brokerlist='vm://:1'");
-        env.put("queue.queue", "direct://amq.direct//MessageListenerTest");
+        env.put("queue.queue", "direct://amq.direct//ResetMessageListenerTest");
 
         _context = factory.getInitialContext(env);
 
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/basic/InvalidDestinationTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/basic/InvalidDestinationTest.java
new file mode 100644
index 0000000000..1b5da2631d
--- /dev/null
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/basic/InvalidDestinationTest.java
@@ -0,0 +1,109 @@
+package org.apache.qpid.test.unit.basic;

+

+import org.apache.qpid.client.AMQConnection;

+import org.apache.qpid.client.AMQDestination;

+import org.apache.qpid.client.AMQSession;

+import org.apache.qpid.client.AMQQueue;

+import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;

+import org.apache.qpid.client.transport.TransportConnection;

+

+import junit.framework.TestCase;

+

+import javax.jms.MessageConsumer;

+import javax.jms.Session;

+import javax.jms.QueueSession;

+import javax.jms.Queue;

+import javax.jms.QueueSender;

+import javax.jms.TextMessage;

+import javax.jms.InvalidDestinationException;

+

+public class InvalidDestinationTest extends TestCase

+{

+    private AMQConnection _connection;

+    private AMQDestination _destination;

+    private AMQSession _session;

+    private MessageConsumer _consumer;

+

+    private static final String VM_BROKER = "vm://:1";

+

+

+    protected void setUp() throws Exception

+    {

+        super.setUp();

+        createVMBroker();

+        _connection = new AMQConnection(VM_BROKER, "guest", "guest", "ReceiveTestClient", "test");

+    }

+

+    public void createVMBroker()

+    {

+        try

+        {

+            TransportConnection.createVMBroker(1);

+        }

+        catch (AMQVMBrokerCreationException e)

+        {

+            fail("Unable to create broker: " + e);

+        }

+    }

+

+    protected void tearDown() throws Exception

+    {

+        _connection.close();

+        TransportConnection.killVMBroker(1);

+        super.tearDown();

+    }

+

+

+

+    public void testInvalidDestination() throws Exception

+    {

+        Queue invalidDestination = new AMQQueue("amq.direct","unknownQ");

+        AMQQueue validDestination = new AMQQueue("amq.direct","knownQ");

+        QueueSession queueSession = _connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

+

+        // This is the only easy way to create and bind a queue from the API :-(

+        MessageConsumer consumer = queueSession.createConsumer(validDestination);

+

+        QueueSender sender = queueSession.createSender(invalidDestination);

+        TextMessage msg = queueSession.createTextMessage("Hello");

+        try

+        {

+            sender.send(msg);

+            fail("Expected InvalidDestinationException");

+        }

+        catch (InvalidDestinationException ex)

+        {

+            // pass

+        }

+        sender.close();

+

+        sender = queueSession.createSender(null);

+        invalidDestination = new AMQQueue("amq.direct","unknownQ");

+

+        try

+        {

+            sender.send(invalidDestination,msg);

+            fail("Expected InvalidDestinationException");

+        }

+        catch (InvalidDestinationException ex)

+        {

+            // pass

+        }

+        sender.send(validDestination,msg);

+        sender.close();

+        validDestination = new AMQQueue("amq.direct","knownQ");

+        sender = queueSession.createSender(validDestination);

+        sender.send(msg);

+

+

+

+

+    }

+

+

+    public static junit.framework.Test suite()

+    {

+

+        return new junit.framework.TestSuite(InvalidDestinationTest.class);

+    }

+}

diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/close/MessageRequeueTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/close/MessageRequeueTest.java
index 7762cb3fe9..62234ad21f 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/close/MessageRequeueTest.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/close/MessageRequeueTest.java
@@ -330,7 +330,7 @@ public class MessageRequeueTest extends TestCase
     public void testRequeue() throws JMSException, AMQException, URLSyntaxException
     {
         int run = 0;
-        while (run < 10)
+//        while (run < 10)
         {
             run++;
 
@@ -350,17 +350,10 @@ public class MessageRequeueTest extends TestCase
             _logger.debug("Create Consumer");
             MessageConsumer consumer = session.createConsumer(q);
 
-            try
-            {
-                Thread.sleep(2000);
-            }
-            catch (InterruptedException e)
-            {
-                //
-            }
+            conn.start();
 
             _logger.debug("Receiving msg");
-            Message msg = consumer.receive(1000);
+            Message msg = consumer.receive(2000);
 
             assertNotNull("Message should not be null", msg);
 
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
index 0828ab398c..190b3861f0 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
@@ -100,7 +100,9 @@ public class DurableSubscriptionTest extends TestCase
         AMQTopic topic = new AMQTopic(con,"MyTopic");
         Session session1 = con.createSession(false, AMQSession.NO_ACKNOWLEDGE);
         MessageConsumer consumer1 = session1.createConsumer(topic);
-        MessageProducer producer = session1.createProducer(topic);
+
+        Session sessionProd = con.createSession(false, AMQSession.NO_ACKNOWLEDGE);
+        MessageProducer producer = sessionProd.createProducer(topic);
 
         Session session2 = con.createSession(false, AMQSession.NO_ACKNOWLEDGE);
         TopicSubscriber consumer2 = session2.createDurableSubscriber(topic, "MySubscription");
@@ -112,12 +114,12 @@ public class DurableSubscriptionTest extends TestCase
         Message msg;
         msg = consumer1.receive();
         assertEquals("A", ((TextMessage) msg).getText());
-        msg = consumer1.receive(1000);
+        msg = consumer1.receive(100);
         assertEquals(null, msg);
 
         msg = consumer2.receive();
         assertEquals("A", ((TextMessage) msg).getText());
-        msg = consumer2.receive(1000);
+        msg = consumer2.receive(100);
         assertEquals(null, msg);
 
         consumer2.close();
@@ -127,14 +129,14 @@ public class DurableSubscriptionTest extends TestCase
 
         producer.send(session1.createTextMessage("B"));
 
-        msg = consumer1.receive();
+        msg = consumer1.receive(100);
         assertEquals("B", ((TextMessage) msg).getText());
-        msg = consumer1.receive(1000);
+        msg = consumer1.receive(100);
         assertEquals(null, msg);
 
-        msg = consumer3.receive();
+        msg = consumer3.receive(100);
         assertEquals("B", ((TextMessage) msg).getText());
-        msg = consumer3.receive(1000);
+        msg = consumer3.receive(100);
         assertEquals(null, msg);
 
         con.close();
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java
index 2abc139ced..685fe20048 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java
@@ -53,12 +53,15 @@ public class CommitRollbackTest extends TestCase
     Queue _jmsQueue;
 
     private static final Logger _logger = Logger.getLogger(CommitRollbackTest.class);
+    private static final String BROKER = "vm://:1";
 
     protected void setUp() throws Exception
     {
         super.setUp();
-        TransportConnection.createVMBroker(1);
-
+        if (BROKER.startsWith("vm"))
+        {
+            TransportConnection.createVMBroker(1);
+        }
         testMethod++;
         queue += testMethod;
 
@@ -68,7 +71,7 @@ public class CommitRollbackTest extends TestCase
 
     private void newConnection() throws AMQException, URLSyntaxException, JMSException
     {
-        conn = new AMQConnection("amqp://guest:guest@client/test?brokerlist='vm://:1'");
+        conn = new AMQConnection("amqp://guest:guest@client/test?brokerlist='" + BROKER + "'");
 
         _session = conn.createSession(true, Session.CLIENT_ACKNOWLEDGE);
 
@@ -87,7 +90,10 @@ public class CommitRollbackTest extends TestCase
         super.tearDown();
 
         conn.close();
-        TransportConnection.killVMBroker(1);
+        if (BROKER.startsWith("vm"))
+        {
+            TransportConnection.killVMBroker(1);
+        }
     }
 
     /**
@@ -261,7 +267,7 @@ public class CommitRollbackTest extends TestCase
         assertTrue("session is not transacted", _pubSession.getTransacted());
 
         _logger.info("sending test message");
-        String MESSAGE_TEXT = "testGetThenDisconnect";
+        String MESSAGE_TEXT = "testGetThenRollback";
         _publisher.send(_pubSession.createTextMessage(MESSAGE_TEXT));
 
         _pubSession.commit();
@@ -394,16 +400,60 @@ public class CommitRollbackTest extends TestCase
         _logger.info("receiving result");
         result = _consumer.receive(1000);
         assertNotNull("test message was consumed and rolled back, but is gone", result);
-        assertEquals("1", ((TextMessage) result).getText());
-        assertTrue("Messasge is not marked as redelivered" + result, result.getJMSRedelivered());
-
-        result = _consumer.receive(1000);
-        assertNotNull("test message was consumed and rolled back, but is gone", result);
-        assertEquals("2", ((TextMessage) result).getText());
-        assertTrue("Messasge is not marked as redelivered" + result, result.getJMSRedelivered());
-
+        if (result.getJMSRedelivered())
+        {
+            assertEquals("1", ((TextMessage) result).getText());
+
+            result = _consumer.receive(1000);
+            assertNotNull("test message was consumed and rolled back, but is gone", result);
+            assertEquals("2", ((TextMessage) result).getText());
+            assertTrue("Messasge is not marked as redelivered" + result, result.getJMSRedelivered());
+        }
+        else
+        {
+            assertEquals("2", ((TextMessage) result).getText());
+            assertTrue("Messasge is marked as redelivered" + result, !result.getJMSRedelivered());
+
+            result = _consumer.receive(1000);
+            assertNotNull("test message was consumed and rolled back, but is gone", result);
+            assertEquals("1", ((TextMessage) result).getText());
+            assertTrue("Messasge is not marked as redelivered" + result, result.getJMSRedelivered());
+
+        }
         result = _consumer.receive(1000);
         assertNull("test message should be null:" + result, result);
+
+    }
+
+
+    public void testPutThenRollbackThenGet() throws Exception
+    {
+        assertTrue("session is not transacted", _session.getTransacted());
+        assertTrue("session is not transacted", _pubSession.getTransacted());
+
+        _logger.info("sending test message");
+        String MESSAGE_TEXT = "testPutThenRollbackThenGet";
+
+        _publisher.send(_pubSession.createTextMessage(MESSAGE_TEXT));
+        _pubSession.commit();
+
+        assertNotNull(_consumer.receive(100));
+
+        _publisher.send(_pubSession.createTextMessage(MESSAGE_TEXT));
+
+        _logger.info("rolling back");
+        _pubSession.rollback();
+
+        _logger.info("receiving result");
+        Message result = _consumer.receive(1000);
+        assertNull("test message was put and rolled back, but is still present", result);
+
+        _publisher.send(_pubSession.createTextMessage(MESSAGE_TEXT));
+
+        _pubSession.commit();
+
+        assertNotNull(_consumer.receive(100));
+
     }
 
 }
diff --git a/java/client/src/test/java/org/apache/qpid/testutil/Config.java b/java/client/src/test/java/org/apache/qpid/testutil/Config.java
index 8109d20a33..b777cf93b6 100644
--- a/java/client/src/test/java/org/apache/qpid/testutil/Config.java
+++ b/java/client/src/test/java/org/apache/qpid/testutil/Config.java
@@ -172,7 +172,7 @@ public class Config
             }
             catch(NumberFormatException e)
             {
-                throw new RuntimeException("Bad port number: " + value);
+                throw new RuntimeException("Bad port number: " + value, e);
             }
         }
         else if("-name".equalsIgnoreCase(key))
diff --git a/java/client/src/test/java/org/apache/qpid/testutil/QpidClientConnection.java b/java/client/src/test/java/org/apache/qpid/testutil/QpidClientConnection.java
index f2afa472ab..195ed79dab 100644
--- a/java/client/src/test/java/org/apache/qpid/testutil/QpidClientConnection.java
+++ b/java/client/src/test/java/org/apache/qpid/testutil/QpidClientConnection.java
@@ -3,6 +3,7 @@ package org.apache.qpid.testutil;
 import org.apache.qpid.client.AMQConnectionFactory;
 import org.apache.qpid.client.AMQConnectionURL;
 import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.JMSAMQException;
 import org.apache.qpid.url.URLSyntaxException;
 import org.apache.log4j.Logger;
 
@@ -70,7 +71,7 @@ public class QpidClientConnection implements ExceptionListener
             }
             catch (URLSyntaxException e)
             {
-                throw new JMSException("URL syntax error in [" + brokerUrl + "]: " + e.getMessage());
+                throw new JMSAMQException("URL syntax error in [" + brokerUrl + "]: " + e.getMessage(), e);
             }
         }
     }