You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kw...@apache.org on 2014/02/04 12:13:44 UTC
svn commit: r1564254 [2/3] - in /qpid/branches/java-broker-bdb-ha: ./ qpid/
qpid/cpp/ qpid/cpp/bindings/qpid/dotnet/ qpid/cpp/design_docs/
qpid/cpp/src/ qpid/cpp/src/qpid/ qpid/cpp/src/qpid/acl/
qpid/cpp/src/qpid/amqp_0_10/ qpid/cpp/src/qpid/broker/ qp...
Modified: qpid/branches/java-broker-bdb-ha/qpid/cpp/src/qpid/sys/posix/MemoryMappedFile.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/cpp/src/qpid/sys/posix/MemoryMappedFile.cpp?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/cpp/src/qpid/sys/posix/MemoryMappedFile.cpp (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/cpp/src/qpid/sys/posix/MemoryMappedFile.cpp Tue Feb 4 11:13:42 2014
@@ -29,13 +29,14 @@
namespace qpid {
namespace sys {
namespace {
+const std::string PAGEFILE_PREFIX("pf_");
const std::string PATH_SEPARATOR("/");
const std::string ESCAPE("%");
const std::string VALID("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-.");
std::string getFileName(const std::string& name, const std::string& dir)
{
std::stringstream filename;
- if (dir.size()) filename << dir << PATH_SEPARATOR;
+ if (dir.size()) filename << dir << PATH_SEPARATOR << PAGEFILE_PREFIX;
size_t start = 0;
while (true) {
size_t i = name.find_first_not_of(VALID, start);
@@ -55,31 +56,39 @@ std::string getFileName(const std::strin
class MemoryMappedFilePrivate
{
friend class MemoryMappedFile;
+ std::string path;
int fd;
MemoryMappedFilePrivate() : fd(0) {}
};
MemoryMappedFile::MemoryMappedFile() : state(new MemoryMappedFilePrivate) {}
MemoryMappedFile::~MemoryMappedFile() { delete state; }
-std::string MemoryMappedFile::open(const std::string& name, const std::string& directory)
+void MemoryMappedFile::open(const std::string& name, const std::string& directory)
{
- std::string path = getFileName(name, directory);
+ // Ensure directory exists
+ if ( ::mkdir(directory.c_str(), S_IRWXU | S_IRGRP | S_IXGRP )!=0 && errno!=EEXIST ) {
+ throw qpid::Exception(QPID_MSG("Failed to create memory mapped file directory " << directory << ": " << qpid::sys::strError(errno)));
+ }
+
+ state->path = getFileName(name, directory);
- int flags = O_CREAT | O_EXCL | O_RDWR;
- int fd = ::open(path.c_str(), flags, S_IRUSR | S_IWUSR);
- if (fd == -1) throw qpid::Exception(QPID_MSG("Failed to open memory mapped file " << path << ": " << qpid::sys::strError(errno) << " [flags=" << flags << "]"));
+ int flags = O_CREAT | O_TRUNC | O_RDWR;
+ int fd = ::open(state->path.c_str(), flags, S_IRUSR | S_IWUSR);
+ if (fd == -1) throw qpid::Exception(QPID_MSG("Failed to open memory mapped file " << state->path << ": " << qpid::sys::strError(errno) << " [flags=" << flags << "]"));
state->fd = fd;
- return path;
}
-void MemoryMappedFile::close(const std::string& path)
+
+void MemoryMappedFile::close()
{
::close(state->fd);
- ::unlink(path.c_str());
+ ::unlink(state->path.c_str());
}
+
size_t MemoryMappedFile::getPageSize()
{
return ::sysconf(_SC_PAGE_SIZE);
}
+
char* MemoryMappedFile::map(size_t offset, size_t size)
{
int protection = PROT_READ | PROT_WRITE;
@@ -90,20 +99,24 @@ char* MemoryMappedFile::map(size_t offse
return region;
}
+
void MemoryMappedFile::unmap(char* region, size_t size)
{
::munmap(region, size);
}
+
void MemoryMappedFile::flush(char* region, size_t size)
{
::msync(region, size, MS_ASYNC);
}
+
void MemoryMappedFile::expand(size_t offset)
{
if ((::lseek(state->fd, offset - 1, SEEK_SET) == -1) || (::write(state->fd, "", 1) == -1)) {
throw qpid::Exception(QPID_MSG("Failed to expand paged queue file: " << qpid::sys::strError(errno)));
}
}
+
bool MemoryMappedFile::isSupported()
{
return true;
Modified: qpid/branches/java-broker-bdb-ha/qpid/cpp/src/qpid/sys/windows/MemoryMappedFile.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/cpp/src/qpid/sys/windows/MemoryMappedFile.cpp?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/cpp/src/qpid/sys/windows/MemoryMappedFile.cpp (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/cpp/src/qpid/sys/windows/MemoryMappedFile.cpp Tue Feb 4 11:13:42 2014
@@ -27,11 +27,10 @@ class MemoryMappedFilePrivate {};
MemoryMappedFile::MemoryMappedFile() : state(0) {}
MemoryMappedFile::~MemoryMappedFile() {}
-std::string MemoryMappedFile::open(const std::string& /*name*/, const std::string& /*directory*/)
+void MemoryMappedFile::open(const std::string& /*name*/, const std::string& /*directory*/)
{
- return std::string();
}
-void MemoryMappedFile::close(const std::string& /*path*/)
+void MemoryMappedFile::close()
{
}
size_t MemoryMappedFile::getPageSize()
Propchange: qpid/branches/java-broker-bdb-ha/qpid/cpp/src/tests/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/cpp/src/tests:r1562453-1564250
Modified: qpid/branches/java-broker-bdb-ha/qpid/cpp/src/tests/acl.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/cpp/src/tests/acl.py?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/cpp/src/tests/acl.py (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/cpp/src/tests/acl.py Tue Feb 4 11:13:42 2014
@@ -671,6 +671,118 @@ class ACLTests(TestBase010):
self.fail(result)
+ def test_illegal_pages_lower_limit_spec(self):
+ """
+ Test illegal paged queue policy
+ """
+
+ aclf = self.get_acl_file()
+ aclf.write('acl deny bob@QPID create queue name=q2 pageslowerlimit=-1\n')
+ aclf.write('acl allow all all')
+ aclf.close()
+
+ result = self.reload_acl()
+ expected = "-1 is not a valid value for 'pageslowerlimit', " \
+ "values should be between 0 and 9223372036854775807";
+ if (result.find(expected) == -1):
+ self.fail(result)
+
+ aclf = self.get_acl_file()
+ aclf.write('acl deny bob@QPID create queue name=q2 pageslowerlimit=9223372036854775808\n')
+ aclf.write('acl allow all all')
+ aclf.close()
+
+ result = self.reload_acl()
+ expected = "9223372036854775808 is not a valid value for 'pageslowerlimit', " \
+ "values should be between 0 and 9223372036854775807";
+ if (result.find(expected) == -1):
+ self.fail(result)
+
+
+ def test_illegal_pages_upper_limit_spec(self):
+ """
+ Test illegal paged queue policy
+ """
+
+ aclf = self.get_acl_file()
+ aclf.write('acl deny bob@QPID create queue name=q2 pagesupperlimit=-1\n')
+ aclf.write('acl allow all all')
+ aclf.close()
+
+ result = self.reload_acl()
+ expected = "-1 is not a valid value for 'pagesupperlimit', " \
+ "values should be between 0 and 9223372036854775807";
+ if (result.find(expected) == -1):
+ self.fail(result)
+
+ aclf = self.get_acl_file()
+ aclf.write('acl deny bob@QPID create queue name=q2 pagesupperlimit=9223372036854775808\n')
+ aclf.write('acl allow all all')
+ aclf.close()
+
+ result = self.reload_acl()
+ expected = "9223372036854775808 is not a valid value for 'pagesupperlimit', " \
+ "values should be between 0 and 9223372036854775807";
+ if (result.find(expected) == -1):
+ self.fail(result)
+
+
+ def test_illegal_pagefactor_lower_limit_spec(self):
+ """
+ Test illegal paged queue policy
+ """
+
+ aclf = self.get_acl_file()
+ aclf.write('acl deny bob@QPID create queue name=q2 pagefactorlowerlimit=-1\n')
+ aclf.write('acl allow all all')
+ aclf.close()
+
+ result = self.reload_acl()
+ expected = "-1 is not a valid value for 'pagefactorlowerlimit', " \
+ "values should be between 0 and 9223372036854775807";
+ if (result.find(expected) == -1):
+ self.fail(result)
+
+ aclf = self.get_acl_file()
+ aclf.write('acl deny bob@QPID create queue name=q2 pagefactorlowerlimit=9223372036854775808\n')
+ aclf.write('acl allow all all')
+ aclf.close()
+
+ result = self.reload_acl()
+ expected = "9223372036854775808 is not a valid value for 'pagefactorlowerlimit', " \
+ "values should be between 0 and 9223372036854775807";
+ if (result.find(expected) == -1):
+ self.fail(result)
+
+
+ def test_illegal_pagefactor_upper_limit_spec(self):
+ """
+ Test illegal paged queue policy
+ """
+
+ aclf = self.get_acl_file()
+ aclf.write('acl deny bob@QPID create queue name=q2 pagefactorupperlimit=-1\n')
+ aclf.write('acl allow all all')
+ aclf.close()
+
+ result = self.reload_acl()
+ expected = "-1 is not a valid value for 'pagefactorupperlimit', " \
+ "values should be between 0 and 9223372036854775807";
+ if (result.find(expected) == -1):
+ self.fail(result)
+
+ aclf = self.get_acl_file()
+ aclf.write('acl deny bob@QPID create queue name=q2 pagefactorupperlimit=9223372036854775808\n')
+ aclf.write('acl allow all all')
+ aclf.close()
+
+ result = self.reload_acl()
+ expected = "9223372036854775808 is not a valid value for 'pagefactorupperlimit', " \
+ "values should be between 0 and 9223372036854775807";
+ if (result.find(expected) == -1):
+ self.fail(result)
+
+
#=====================================
# ACL queue tests
#=====================================
@@ -687,6 +799,7 @@ class ACLTests(TestBase010):
aclf.write('acl deny bob@QPID purge queue name=q3\n')
aclf.write('acl deny bob@QPID delete queue name=q4\n')
aclf.write('acl deny bob@QPID create queue name=q5 maxqueuesize=1000 maxqueuecount=100\n')
+ aclf.write('acl deny bob@QPID create queue name=q6 paging=true\n')
aclf.write('acl allow all all')
aclf.close()
@@ -739,6 +852,15 @@ class ACLTests(TestBase010):
try:
queue_options = {}
+ queue_options["qpid.paging"] = True
+ session.queue_declare(queue="q6", arguments=queue_options)
+ self.fail("ACL should deny queue create request with name=q6, qpid.paging=True");
+ except qpid.session.SessionException, e:
+ self.assertEqual(403,e.args[0].error_code)
+ session = self.get_session('bob','bob')
+
+ try:
+ queue_options = {}
queue_options["qpid.max_count"] = 200
queue_options["qpid.max_size"] = 100
session.queue_declare(queue="q2", exclusive=True, arguments=queue_options)
@@ -972,6 +1094,107 @@ class ACLTests(TestBase010):
self.fail("ACL should allow queue delete request for q4");
#=====================================
+ # ACL paged tests
+ #=====================================
+
+ def test_paged_allow_mode(self):
+ """
+ Test cases for paged acl in allow mode
+ """
+ aclf = self.get_acl_file()
+ aclf.write('acl deny bob@QPID create queue name=qf1 pageslowerlimit=1000\n')
+ aclf.write('acl deny bob@QPID create queue name=qf2 pagesupperlimit=100\n')
+ aclf.write('acl deny bob@QPID create queue name=qf3 pagefactorlowerlimit=10\n')
+ aclf.write('acl deny bob@QPID create queue name=qf4 pagefactorupperlimit=1\n')
+ aclf.write('acl allow all all')
+ aclf.close()
+
+ result = self.reload_acl()
+ if (result):
+ self.fail(result)
+
+ session = self.get_session('bob','bob')
+
+ try:
+ queue_options = {}
+ queue_options["qpid.paging"] = True
+ queue_options["qpid.max_pages_loaded"] = 500
+ session.queue_declare(queue="qf1", arguments=queue_options)
+ self.fail("ACL should deny queue create request with name=qf1, qpid.paging=True, qpid.max_pages_loaded=500");
+ except qpid.session.SessionException, e:
+ self.assertEqual(403,e.args[0].error_code)
+ session = self.get_session('bob','bob')
+
+ try:
+ queue_options = {}
+ queue_options["qpid.paging"] = True
+ queue_options["qpid.max_pages_loaded"] = 500
+ session.queue_declare(queue="qf2", arguments=queue_options)
+ self.fail("ACL should deny queue create request with name=qf2, qpid.paging=True, qpid.max_pages_loaded=500");
+ except qpid.session.SessionException, e:
+ self.assertEqual(403,e.args[0].error_code)
+ session = self.get_session('bob','bob')
+
+ try:
+ queue_options = {}
+ queue_options["qpid.paging"] = True
+ queue_options["qpid.page_factor"] = 5
+ session.queue_declare(queue="qf3", arguments=queue_options)
+ self.fail("ACL should deny queue create request with name=qf3, qpid.paging=True, qpid.page_factor=5");
+ except qpid.session.SessionException, e:
+ self.assertEqual(403,e.args[0].error_code)
+ session = self.get_session('bob','bob')
+
+ try:
+ queue_options = {}
+ queue_options["qpid.paging"] = True
+ queue_options["qpid.page_factor"] = 5
+ session.queue_declare(queue="qf4", arguments=queue_options)
+ self.fail("ACL should deny queue create request with name=qf4, qpid.paging=True, qpid.page_factor=5");
+ except qpid.session.SessionException, e:
+ self.assertEqual(403,e.args[0].error_code)
+ session = self.get_session('bob','bob')
+
+
+ def test_paged_deny_mode(self):
+ """
+ Test cases for paged acl in deny mode
+ """
+ aclf = self.get_acl_file()
+ aclf.write('acl allow bob@QPID create queue name=qf1 pageslowerlimit=100 pagesupperlimit=1000\n')
+ aclf.write('acl allow bob@QPID create queue name=qf2 pagefactorlowerlimit=1 pagefactorupperlimit=10\n')
+ aclf.write('acl allow anonymous all all\n')
+ aclf.write('acl deny all all')
+ aclf.close()
+
+ result = self.reload_acl()
+ if (result):
+ self.fail(result)
+
+ session = self.get_session('bob','bob')
+
+ try:
+ queue_options = {}
+ queue_options["qpid.paging"] = True
+ queue_options["qpid.max_pages_loaded"] = 500
+ session.queue_declare(queue="qf1", arguments=queue_options)
+ except qpid.session.SessionException, e:
+ if (403 == e.args[0].error_code):
+ self.fail("ACL should allow queue create request with name=qf1, qpid.paging=True, qpid.max_pages_loaded=500");
+ session = self.get_session('bob','bob')
+
+ try:
+ queue_options = {}
+ queue_options["qpid.paging"] = True
+ queue_options["qpid.page_factor"] = 5
+ session.queue_declare(queue="qf2", arguments=queue_options)
+ except qpid.session.SessionException, e:
+ if (403 == e.args[0].error_code):
+ self.fail("ACL should allow queue create request with name=qf2, qpid.paging=True, qpid.page_factor=5");
+ session = self.get_session('bob','bob')
+
+
+ #=====================================
# ACL file tests
#=====================================
Modified: qpid/branches/java-broker-bdb-ha/qpid/cpp/src/tests/run_paged_queue_tests
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/cpp/src/tests/run_paged_queue_tests?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/cpp/src/tests/run_paged_queue_tests (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/cpp/src/tests/run_paged_queue_tests Tue Feb 4 11:13:42 2014
@@ -26,7 +26,7 @@ trap stop_broker INT TERM QUIT
export PATH=$PWD:$srcdir:$PYTHON_COMMANDS:$PATH
start_broker() {
- QPID_PORT=$($QPIDD_EXEC --daemon --port 0 --interface 127.0.0.1 --no-data-dir $MODULES --auth no) || fail "Could not start broker"
+ QPID_PORT=$($QPIDD_EXEC --daemon --port 0 --interface 127.0.0.1 --no-data-dir --paging-dir=$PWD/pqtest_data $MODULES --auth no) || fail "Could not start broker"
}
stop_broker() {
Modified: qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/cpp-broker/Security.xml
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/cpp-broker/Security.xml?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/cpp-broker/Security.xml (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/cpp-broker/Security.xml Tue Feb 4 11:13:42 2014
@@ -421,7 +421,11 @@ com.sun.security.jgss.initiate {
filemaxsizelowerlimit |
filemaxsizeupperlimit |
filemaxcountlowerlimit |
- filemaxcountupperlimit ]
+ filemaxcountupperlimit |
+ pageslowerlimit |
+ pagesupperlimit |
+ pagefactorlowerlimit |
+ pagefactorupperlimit ]
acl permission {<group-name>|<user-name>|"all"} {action|"all"} [object|"all"
[property=<property-value> ...]]
@@ -728,6 +732,12 @@ com.sun.security.jgss.initiate {
<entry>Indicates the presence of an <parameter>exclusive</parameter> flag</entry>
<entry>CREATE QUEUE, ACCESS QUEUE</entry>
</row>
+ <row>
+ <entry> <command>paging</command> </entry>
+ <entry>Boolean</entry>
+ <entry>Indicates if the queue is paging queue</entry>
+ <entry>CREATE QUEUE, ACCESS QUEUE</entry>
+ </row>
<row>
<entry> <command>type</command> </entry>
<entry>String</entry>
@@ -806,6 +816,30 @@ com.sun.security.jgss.initiate {
<entry>Maximum value for file.max_count (files)</entry>
<entry>CREATE QUEUE, ACCESS QUEUE</entry>
</row>
+ <row>
+ <entry> <command>pageslowerlimit</command> </entry>
+ <entry>Integer</entry>
+ <entry>Minimum value for number of pages in memory of paged queue</entry>
+ <entry>CREATE QUEUE</entry>
+ </row>
+ <row>
+ <entry> <command>pagesupperlimit</command> </entry>
+ <entry>Integer</entry>
+ <entry>Maximum value for number of pages in memory of paged queue</entry>
+ <entry>CREATE QUEUE</entry>
+ </row>
+ <row>
+ <entry> <command>pagefactorlowerlimit</command> </entry>
+ <entry>Integer</entry>
+ <entry>Minimum value for size of one page in paged queue</entry>
+ <entry>CREATE QUEUE</entry>
+ </row>
+ <row>
+ <entry> <command>pagefactorupperlimit</command> </entry>
+ <entry>Integer</entry>
+ <entry>Maximum value for size of one page in paged queue</entry>
+ <entry>CREATE QUEUE</entry>
+ </row>
</tbody>
</tgroup>
</table>
@@ -910,7 +944,7 @@ com.sun.security.jgss.initiate {
<row>
<entry>create</entry>
<entry>queue</entry>
- <entry>name alternate durable exclusive autodelete policy queuemaxsizelowerlimit queuemaxsizeupperlimit queuemaxcountlowerlimit queuemaxcountupperlimit filemaxsizelowerlimit filemaxsizeupperlimit filemaxcountlowerlimit filemaxcountupperlimit</entry>
+ <entry>name alternate durable exclusive autodelete policy queuemaxsizelowerlimit queuemaxsizeupperlimit queuemaxcountlowerlimit queuemaxcountupperlimit filemaxsizelowerlimit filemaxsizeupperlimit filemaxcountlowerlimit filemaxcountupperlimit paging pageslowerlimit pagesupperlimit pagefactorlowerlimit pagefactorupperlimit</entry>
<entry></entry>
</row>
<row>
Modified: qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/java-broker/Java-Broker-Configuring-And-Managing-JMX.xml
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/java-broker/Java-Broker-Configuring-And-Managing-JMX.xml?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/java-broker/Java-Broker-Configuring-And-Managing-JMX.xml (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/java-broker/Java-Broker-Configuring-And-Managing-JMX.xml Tue Feb 4 11:13:42 2014
@@ -1,4 +1,8 @@
<?xml version="1.0" encoding="utf-8"?>
+<!DOCTYPE entities [
+<!ENTITY % entities SYSTEM "commonEntities.xml">
+%entities;
+]>
<!--
Licensed to the Apache Software Foundation (ASF) under one
@@ -25,29 +29,293 @@
<section id="Java-Broker-Configuring-And-Managing-JMX-Management-Introduction">
<title>Introduction</title>
- <para>
- The brokers JMX Management Plugin provides the support for creating JMX MBeans for broker objects such as Queues, Exchanges, Connections etc.
- </para>
- <para>
- It is included into the brokers Initial Configuration by default, and is responsible for servicing the RMI and JMX_RMI ports configured on the broker, with the former serving as the RMI Registry used to advertise the actual JMX Connector Server started on the latter.
- </para>
+ <para>The JMX management plugin provides a series of managed beans (MBeans) allowing you to
+ control and monitor the Broker via an industry compliant interface. This provides a
+ convenient intergration point for a variety of Infrastructure Monitoring Solutions,
+ tools such as Jconsole and VisualVM, as well as custom Java programs and scripts.</para>
+ <para>The following sections describe how to connect to JMX, the configuration of the JMX
+ plugin covering topis including securing with SSL, programmatically interacting with
+ Qpid MBeans and finally a summary of all the MBeans made available from by the
+ plugin.</para>
+ <important>
+ <para>For new development work, the reader is directed towards the strategic <link
+ linkend="Java-Broker-Configuring-And-Managing-Web-Console">Web Management
+ Console</link> and the <link
+ linkend="Java-Broker-Configuring-And-Managing-REST-API">REST API</link>. Use the
+ Web/REST interfaces in preference to JMX whenever possible. The JMX interface may be
+ withdrawn in a future release.</para>
+ </important>
</section>
+ <section id="Java-Broker-Configuring-And-Managing-JMX-Management-Plugin-DefaultConfiguration">
+ <title>Default Configuration</title>
+ <para>By default, the Broker is shipped with JMX enabled.</para>
+ <para>The RMI registry port runs on port <literal>8999</literal> and the JMX connector on
+ port <literal>9099</literal>. The connector is not SSL protected. Qpid will use the
+ <ulink
+ url="&oracleJdkDocUrl;java/lang/management/ManagementFactory.html#getPlatformMBeanServer()"
+ >Platform MBeanServer</ulink>.</para>
+ <para>To change these settings, use the <link
+ linkend="Java-Broker-Configuring-And-Managing-Web-Console">Web Management
+ interface</link>.</para>
+ </section>
+
+ <section id="Java-Broker-Configuring-And-Managing-JMX-Management-Plugin-ConnectingToJMX">
+ <title>Connecting to JMX</title>
+ <para>The following example uses Jconsole to illustrates how to connect to JMX and assume
+ the defaults described above. Jconsole is a management tool that comes with the JDK. It
+ provides a very simple view of the MBeans, but requires no special configuration to be
+ used with Qpid.</para>
+ <para>For full details of Jconsole itself refer to Oracle's <ulink url="&oracleJconsole;"
+ >JConsole Guide</ulink>.</para>
+ <para>Jconsole can be used to connect to local or remote Java processes. On startup, it
+ presents a list of all the Java processes running on the local host and also allows you
+ to specify a service url to connect to a Java process running on a remote host.</para>
+ <para>To start Jconsole on the command line, type:</para>
+ <programlisting><![CDATA[jconsole]]></programlisting>
+ <section
+ id="Java-Broker-Configuring-And-Managing-JMX-Management-Plugin-ConnectingToJMX-Local">
+ <title>Local</title>
+ <para>To connect to a Broker running locally, simply select the process from the list.
+ You can identify the Broker by looking for its classname
+ <literal>org.apache.qpid.server.Main</literal>.</para>
+ </section>
+ <section
+ id="Java-Broker-Configuring-And-Managing-JMX-Management-Plugin-ConnectingToJMX-Remote">
+ <title>Remote</title>
+ <para>To connect to a broker running remotely, provide the hostname and port number of
+ the <emphasis>RMI registry port</emphasis> (e.g. <literal>hostname:8999</literal>)
+ and a valid username and password.</para>
+ <para>You can also provide a service url in the form
+ <literal>service:jmx:rmi:///jndi/rmi://hostname:8999/jmxrmi</literal></para>
+ <figure>
+ <title>Making a remote JMX connection to a Broker using jconsole</title>
+ <graphic fileref="images/JMX-Connect-Remote.png"/>
+ </figure>
+ </section>
+ <para>Once you are connected expand the tree of nodes marked
+ <literal>org.apache.qpid</literal> to begin to interact with the Qpid MBeans.</para>
+ <figure>
+ <title>Qpid MBean hierarchy</title>
+ <graphic fileref="images/JMX-Connect-MBeans.png"/>
+ </figure>
+ <section id="Java-Broker-Configuring-And-Managing-JMX-Management-Plugin-ConnectingToJMX-SSL">
+ <title>Connecting to a remote Broker protected by SSL</title>
+ <para>If you are connecting to a remote Broker whose JMX connector port has been secured
+ with SSL certificate signed by a private CA (or a self-signed certificate), you will
+ need to pass a trust store and trust store password to Jconsole. If this is
+ required, start jconsole with the following options:</para>
+ <programlisting><![CDATA[jconsole -J-Djavax.net.ssl.trustStore=jmxtruststore.jks -J-Djavax.net.ssl.trustStorePassword=password]]></programlisting>
+ </section>
+ </section>
+
+ <section id="Java-Broker-Configuring-And-Managing-JMX-Example-Client">
+ <title>Example JMX Client</title>
+ <para>The following java snippet illustrates a JMX client that connects to Qpid over JMX
+ passing a userid and password, looks up the <ulink
+ url="&qpidManagementCommonSrc;org/apache/qpid/management/common/mbeans/ManagedBroker&qpidSrcSuffix;"
+ >ManagedBroker</ulink> object corresponding to the <literal>myvhost</literal>
+ virtualhost, then invokes a method on the virtualhost to create a new queue.</para>
+ <para>A full introduction to custom JMX clients is beyond the scope of this book. For this
+ the reader is directed toward Oracle's <ulink url="&oracleJmxTutorial;">JMX
+ tutorial.</ulink></para>
+ <example id="Java-Broker-Configuring-And-Managing-JMX-Example-Client-Code">
+ <title>JMX Client illustrating the creation of a new queue</title>
+ <programlisting language="java">
+Map<String, Object< environment = new HashMap<String, Object>();
+environment.put(JMXConnector.CREDENTIALS, new String[] {"admin","password"});
+// Connect to service
+JMXServiceURL url = new JMXServiceURL("service:jmx:rmi:///jndi/rmi://localhost:8999/jmxrmi");
+JMXConnector jmxConnector = JMXConnectorFactory.connect(url, environment);
+MBeanServerConnection mbsc = jmxConnector.getMBeanServerConnection();
+// Object name for ManagedBroker mbean for virtualhost myvhost
+ObjectName objectName = new ObjectName("org.apache.qpid:type=VirtualHost.VirtualHostManager,VirtualHost=\"myvhost\"");
+// Get the ManagedBroker object
+ManagedBroker managedBroker = JMX.newMBeanProxy(mbsc, objectName, ManagedBroker.class);;
+
+// Create the queue named "myqueue"
+managedBroker.createNewQueue("myqueue", null, true);</programlisting>
+ </example>
+ <para>The Qpid classes required for a custom JMX client are included in the
+ <literal>qpid-management-common</literal> artefact.</para>
+ </section>
+
+ <section id="Java-Broker-Configuring-And-Managing-JMX-Management-MBeans">
+ <title>The MBeans</title>
+ <para>The following table summarises the available MBeans. The MBeans are self-describing:
+ each attribute and operation carry a description describing their purpose. This
+ description is visible when using tools such Jconsole. They are also available on
+ Management interfaces themselves (linked below).</para>
+ <table>
+ <title>Qpid Broker MBeans</title>
+ <tgroup cols="2">
+ <thead>
+ <row>
+ <entry>Management Interface</entry>
+ <entry>Object Name</entry>
+ </row>
+ </thead>
+ <tbody>
+ <row>
+ <entry morerows="1">
+ <para>
+ <ulink
+ url="&qpidManagementCommonSrc;org/apache/qpid/management/common/mbeans/ManagedBroker&qpidSrcSuffix;"
+ >ManagedBroker</ulink>
+ </para>
+ </entry>
+ <entry>
+ <para>
+ <literal>org.apache.qpid:type=VirtualHost.VirtualHostManager,VirtualHost="<replaceable>virtualhostname</replaceable>"</literal>
+ </para>
+ </entry>
+ </row>
+ <row>
+ <entry>
+ <para>MBean corresponding to the named virtualhost. Allows operations
+ such as the creation/deletion of queues and exchanges on that
+ virtualhost and virtualhost levell statistics.</para>
+ </entry>
+ </row>
+ <row>
+ <entry morerows="1">
+ <para>
+ <ulink
+ url="&qpidManagementCommonSrc;org/apache/qpid/management/common/mbeans/ManagedQueue&qpidSrcSuffix;"
+ >ManagedQueue</ulink>
+ </para>
+ </entry>
+ <entry>
+ <para>
+ <literal>org.apache.qpid:type=VirtualHost.Queue,VirtualHost="<replaceable>virtualhostname</replaceable>",name="<replaceable>queuename</replaceable>"</literal>
+ </para>
+ </entry>
+ </row>
+ <row>
+ <entry>
+ <para>MBean corresponding to the named queue on the given virtualhost.
+ Allows queue management operations such as view message, move
+ message and clear queue. Exposes attributes such as queue depth and
+ durability.</para>
+ </entry>
+ </row>
+ <row>
+ <entry morerows="1">
+ <para>
+ <ulink
+ url="&qpidManagementCommonSrc;org/apache/qpid/management/common/mbeans/ManagedExchange&qpidSrcSuffix;"
+ >ManagedExchange</ulink>
+ </para>
+ </entry>
+ <entry>
+ <para>
+ <literal>org.apache.qpid:type=VirtualHost.Exchange,VirtualHost="<replaceable>virtualhostname</replaceable>",name="<replaceable>amq.direct</replaceable>",ExchangeType=<replaceable>type</replaceable></literal>
+ </para>
+ </entry>
+ </row>
+ <row>
+ <entry>
+ <para>MBean corresponding to the named exchange on the given
+ virtualhost. Allows exchange management operations such as the
+ creation and removal of bindings. The supported exchange types are
+ exposed by the <literal>exchangeTypes</literal> attribute of the
+ virtualhost.</para>
+ </entry>
+ </row>
+ <row>
+ <entry morerows="1">
+ <para>
+ <ulink
+ url="&qpidManagementCommonSrc;org/apache/qpid/management/common/mbeans/ManagedConnection&qpidSrcSuffix;"
+ >ManagedConnection</ulink>
+ </para>
+ </entry>
+ <entry>
+ <para>
+ <literal>org.apache.qpid:type=VirtualHost.Connection,VirtualHost="<replaceable>virtualhostname</replaceable>",name="<replaceable>/peerid:ephemeralport</replaceable>"</literal>
+ </para>
+ </entry>
+ </row>
+ <row>
+ <entry>
+ <para>MBean representing a active AMQP connection to the named virtual
+ host. Name is formed from the IP and ephemeral port of the peer.
+ Attributes include the client version and connection level
+ statistics.</para>
+ </entry>
+ </row>
+ <row>
+ <entry morerows="1">
+ <para>
+ <ulink
+ url="&qpidManagementCommonSrc;org/apache/qpid/management/common/mbeans/UserManagement&qpidSrcSuffix;"
+ >UserManagement</ulink>
+ </para>
+ </entry>
+ <entry>
+ <para>
+ <literal>org.apache.qpid:type=UserManagement,name="UserManagement-<replaceable>authentication
+ manager name</replaceable>"</literal>
+ </para>
+ </entry>
+ </row>
+ <row>
+ <entry>
+ <para>When using <link
+ linkend="Java-Broker-Security-PlainPasswordFile-Provider">Plain
+ password provider</link> or <link
+ linkend="Java-Broker-Security-Base64MD5PasswordFile-Provider"
+ >Base 64 MD5 password provider</link>, permits user operations
+ such creation and deletion of users. and password changes.</para>
+ </entry>
- <section id="Java-Broker-Configuring-And-Managing-JMX-Management-Plugin-Configuration">
- <title>JMX Management Plugin Configuration</title>
+ </row>
+ <row>
+ <entry morerows="1">
+ <para>
+ <ulink
+ url="&qpidManagementCommonSrc;org/apache/qpid/management/common/mbeans/ServerInformation&qpidSrcSuffix;"
+ >ServerInformation</ulink>
+ </para>
+ </entry>
+ <entry>
+ <para>
+ <literal>org.apache.qpid:type=ServerInformation,name=ServerInformation</literal>
+ </para>
+ </entry>
+ </row>
+ <row>
+ <entry>
+ <para>Exposes broker wide statistics, product version number and JMX
+ management API version number.</para>
+ </entry>
- <para>
- The JMX Management Plugin can be configured through the <link linkend="Java-Broker-Configuring-And-Managing-Web-Console">Web Management Console</link>
- and underlying REST management interface. By double-clicking on the JMX Management Plugin name in the object tree a tab for the plugin
- is displayed with its current settings, which can be changed by clicking on the "Edit" button.
-
- The following attributes can be set on the JMX Management Plugin:
- <itemizedlist>
- <listitem><para><emphasis>Use Platform MBean Server</emphasis>. The JMX Management Plugin can start its own MBean Server or it can use the JVMs 'Platform MBean Server'.
- By default this is true, and the Platform MBean Server is used.</para></listitem>
- </itemizedlist>
- NOTE: Changes to the "Use Platform MBean Server" attribute only take effect at broker restart.
- </para>
+ </row>
+ <row>
+ <entry morerows="1">
+ <para>
+ <ulink
+ url="&qpidManagementCommonSrc;org/apache/qpid/management/common/mbeans/LoggingManagement&qpidSrcSuffix;"
+ >LoggingManagement</ulink>
+ </para>
+ </entry>
+ <entry>
+ <para>
+ <literal>org.apache.qpid:type=LoggingManagement,name=LoggingManagement</literal>
+ </para>
+ </entry>
+ </row>
+ <row>
+ <entry>
+ <para>MBean permitting control of the Broker's logging. Exposes
+ operations allow the logging level to be controlled at runtime
+ (without restarting the Broker) and others that allow changes to be
+ written back to the log4j.xml logging configuration file, or the
+ contents of the log4.xml file to be re-read at runtime.</para>
+ </entry>
+ </row>
+ </tbody>
+ </tgroup>
+ </table>
</section>
</section>
Modified: qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/java-broker/Java-Broker-Ports.xml
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/java-broker/Java-Broker-Ports.xml?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/java-broker/Java-Broker-Ports.xml (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/java-broker/Java-Broker-Ports.xml Tue Feb 4 11:13:42 2014
@@ -63,7 +63,7 @@
</para>
<para>
- SSL can be enabled forPorts with protocols that support it by selecting the 'SSL' transport, at which
+ SSL can be enabled for Ports with protocols that support it by selecting the 'SSL' transport, at which
point a configured <link linkend="Java-Broker-SSL-Keystore">KeyStore</link> must also be selected for the Port.
</para>
@@ -83,7 +83,7 @@
</important>
<important>
- Following deletion of an active Port, the port remains bound until the Broker is restarted. You should restart the broker
+ Following deletion of an active HTTP or JMX Port, the port remains bound until the Broker is restarted. You should restart the broker
immediately if you require preventing new connections on the port or disconnecting existing clients.
</important>
Modified: qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/java-broker/commonEntities.xml
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/java-broker/commonEntities.xml?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/java-broker/commonEntities.xml (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/java-broker/commonEntities.xml Tue Feb 4 11:13:42 2014
@@ -36,6 +36,7 @@
<!ENTITY oracleJdkDocUrl "http://docs.oracle.com/javase/6/docs/api/">
<!ENTITY oracleJeeDocUrl "http://docs.oracle.com/javaee/6/api/">
<!ENTITY oracleKeytool "http://docs.oracle.com/javase/6/docs/technotes/tools/solaris/keytool.html">
+<!ENTITY oracleJconsole "http://java.sun.com/javase/6/docs/technotes/guides/management/jconsole.html">
<!-- Oracle BDB JE-->
<!ENTITY oracleJeDownloadUrl "http://www.oracle.com/technetwork/products/berkeleydb/downloads/index.html?ssSourceSiteId=ocomen">
@@ -44,3 +45,8 @@
<!ENTITY oracleBdbJavaDocUrl "http://docs.oracle.com/cd/E17277_02/html/java/">
<!ENTITY oracleBdbProductVersion "5.0.97">
+<!ENTITY oracleJmxTutorial "http://docs.oracle.com/javase/tutorial/jmx/">
+
+<!ENTITY qpidSrc "http://svn.apache.org/viewvc/qpid/trunk/qpid/java/">
+<!ENTITY qpidManagementCommonSrc "&qpidSrc;management/common/src/main/java/">
+<!ENTITY qpidSrcSuffix ".java?view=co">
Modified: qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/jms-client-0-8/JMS-Client-Examples.xml
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/jms-client-0-8/JMS-Client-Examples.xml?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/jms-client-0-8/JMS-Client-Examples.xml (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/jms-client-0-8/JMS-Client-Examples.xml Tue Feb 4 11:13:42 2014
@@ -226,7 +226,7 @@ public class StocksExample {
Message message = session.createMessage();
message.setStringProperty("instrument", "IBM");
- message.setStringProperty("price", "100");
+ message.setIntProperty("price", 100);
messageProducer.send(message);
session.commit();
Modified: qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/jms-client-0-8/JMS-Client-Understanding.xml
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/jms-client-0-8/JMS-Client-Understanding.xml?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/jms-client-0-8/JMS-Client-Understanding.xml (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/doc/book/src/jms-client-0-8/JMS-Client-Understanding.xml Tue Feb 4 11:13:42 2014
@@ -136,7 +136,7 @@ amqp://username:password@clientid/test
</screen>
</example>
<para>For full details see <xref linkend="JMS-Client-0-8-Connection-URL"/></para>
- <note>Note, that a single broker failover is enabled by default. If the failover behaviour is not desired it can be switched off
+ <note><para>Note, that a single broker failover is enabled by default. If the failover behaviour is not desired it can be switched off
by setting a failover option to <emphasis>nofailover</emphasis> as in the example below
<example>
<title>Connection URL configured with nofailover</title>
@@ -145,7 +145,7 @@ amqp://username:password@clientid/test
?brokerlist='tcp://localhost:15672?failover='nofailover']]>
</screen>
</example>
- </note>
+ </para></note>
<!-- TODO perhaps mention ConnectionListener?-->
</section>
<section id="JMS-Client-0-8-Client-Understanding-Connection-Heartbeating">
@@ -284,7 +284,7 @@ amqp://guest:guest@clientid/?brokerlist=
>MessageConsumer#receive()</ulink> for Consumer B on the same Session, the application
will hang indefinitely as even if messages suitable for B arrive at the Broker. Those
messages can never be sent to the Session as no space is available in prefetch. </para>
- <note>Please note, when the acknowlegement mode <emphasis>Session#SESSION_TRANSACTED</emphasis>
+ <note><para>Please note, when the acknowlegement mode <emphasis>Session#SESSION_TRANSACTED</emphasis>
or <emphasis>Session#CLIENT_ACKNOWLEDGE</emphasis> is set on a consuming session,
the prefetched messages are released from the prefetch buffer on transaction commit/rollback
(in case of acknowledgement mode <emphasis>Session#SESSION_TRANSACTED</emphasis> )
@@ -294,7 +294,7 @@ amqp://guest:guest@clientid/?brokerlist=
the prefetched messages continue to remain in the prefetch buffer preventing the delivery of the following messages.
As result, the application might stop the receiving of the messages
until the transaction is committed/rolled back (for <emphasis>Session#SESSION_TRANSACTED</emphasis> )
- or received messages are acknowledged (for <emphasis>Session#CLIENT_ACKNOWLEDGE</emphasis>).</note>
+ or received messages are acknowledged (for <emphasis>Session#CLIENT_ACKNOWLEDGE</emphasis>).</para></note>
</section>
<section id="JMS-Client-0-8-Client-Understanding-Session-TemporaryQueues">
<title>TemporaryQueues</title>
@@ -379,10 +379,10 @@ amqp://guest:guest@clientid/?brokerlist=
linkend="JMS-Client-0-8-System-Properties-DefaultMandatoryTopic"
><literal>qpid.default_mandatory_topic</literal></link> for Queues and Topics
respectively.</para>
- <note>Please note, according to AMQP specifications the mandatory flag on a message tells the server
+ <note><para>Please note, according to AMQP specifications the mandatory flag on a message tells the server
how to react if the message cannot be routed to a queue. If this flag is set, the server will return an unroutable message with a
Return method. If this flag is zero, the server silently drops the message. Please, refer <ulink url="&amqpSrc;">AMQP specifications</ulink>
- for more details.</note>
+ for more details.</para></note>
</section>
<section id="JMS-Client-0-8-Client-Understanding-MessageProducer-CloseWhenNoRoute">
<title>Close When No Route</title>
@@ -508,8 +508,10 @@ amqp://guest:guest@clientid/?brokerlist=
<literal>server</literal>.</para>
<para>See <ulink
url="&qpidJavaBrokerBook;Java-Broker-Runtime-Handling-Undeliverable-Messages.html#Java-Broker-Runtime-Handling-Undeliverable-Messages-Maximum-Delivery-Count"
- > Unhandling Undeliverable Messages</ulink> within the Java Broker book for full details
- of the functioning of this feature.</para>
+ > Handling Undeliverable Messages</ulink> within the Java Broker book for full details of
+ the functioning of this feature.</para>
+ <note><para>The optional JMS message header <literal>JMSXDeliveryCount</literal> is <emphasis>not</emphasis>
+ supported.</para></note>
</section>
</section>
<section id="JMS-Client-0-8-Client-Understanding-Destinations">
Propchange: qpid/branches/java-broker-bdb-ha/qpid/java/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java:r1562453-1564250
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/amqp-1-0-client-websocket/pom.xml
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/amqp-1-0-client-websocket/pom.xml?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/amqp-1-0-client-websocket/pom.xml (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/amqp-1-0-client-websocket/pom.xml Tue Feb 4 11:13:42 2014
@@ -44,15 +44,15 @@
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
- <artifactId>geronimo-servlet_2.5_spec</artifactId>
- <version>1.2</version>
+ <artifactId>geronimo-servlet_3.0_spec</artifactId>
+ <version>1.0</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-server</artifactId>
- <version>7.6.10.v20130312</version>
+ <version>8.1.14.v20131031</version>
<scope>compile</scope>
<exclusions>
<exclusion>
@@ -73,14 +73,14 @@
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-continuation</artifactId>
- <version>7.6.10.v20130312</version>
+ <version>8.1.14.v20131031</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-security</artifactId>
- <version>7.6.10.v20130312</version>
+ <version>8.1.14.v20131031</version>
<scope>compile</scope>
<exclusions>
<exclusion>
@@ -93,7 +93,7 @@
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-http</artifactId>
- <version>7.6.10.v20130312</version>
+ <version>8.1.14.v20131031</version>
<scope>compile</scope>
<exclusions>
<exclusion>
@@ -105,7 +105,7 @@
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-io</artifactId>
- <version>7.6.10.v20130312</version>
+ <version>8.1.14.v20131031</version>
<scope>compile</scope>
<exclusions>
<exclusion>
@@ -118,7 +118,7 @@
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-servlet</artifactId>
- <version>7.6.10.v20130312</version>
+ <version>8.1.14.v20131031</version>
<scope>compile</scope>
<exclusions>
<exclusion>
@@ -131,14 +131,14 @@
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-util</artifactId>
- <version>7.6.10.v20130312</version>
+ <version>8.1.14.v20131031</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-websocket</artifactId>
- <version>7.6.10.v20130312</version>
+ <version>8.1.14.v20131031</version>
<scope>compile</scope>
<exclusions>
<exclusion>
Propchange: qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker-core:r1562453-1564250
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java Tue Feb 4 11:13:42 2014
@@ -33,12 +33,14 @@ import org.apache.qpid.server.logging.me
import org.apache.qpid.server.logging.subjects.BindingLogSubject;
import org.apache.qpid.server.logging.subjects.ExchangeLogSubject;
import org.apache.qpid.server.message.InstanceProperties;
+import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.UUIDGenerator;
import org.apache.qpid.server.plugin.ExchangeType;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.BaseQueue;
import org.apache.qpid.server.store.DurableConfigurationStoreHelper;
+import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.virtualhost.VirtualHost;
import java.util.Collection;
@@ -374,9 +376,9 @@ public abstract class AbstractExchange i
return getBindings().size();
}
- @Override
- public final List<? extends BaseQueue> route(final ServerMessage message,
- final InstanceProperties instanceProperties)
+
+ final List<? extends BaseQueue> route(final ServerMessage message,
+ final InstanceProperties instanceProperties)
{
_receivedMessageCount.incrementAndGet();
_receivedMessageSize.addAndGet(message.getSize());
@@ -416,6 +418,59 @@ public abstract class AbstractExchange i
return queues;
}
+ public final int send(final ServerMessage message,
+ final InstanceProperties instanceProperties,
+ final ServerTransaction txn,
+ final BaseQueue.PostEnqueueAction postEnqueueAction)
+ {
+ List<? extends BaseQueue> queues = route(message, instanceProperties);
+
+ if(queues == null || queues.isEmpty())
+ {
+ Exchange altExchange = getAlternateExchange();
+ if(altExchange != null)
+ {
+ return altExchange.send(message, instanceProperties, txn, postEnqueueAction);
+ }
+ else
+ {
+ return 0;
+ }
+ }
+ else
+ {
+ final BaseQueue[] baseQueues = queues.toArray(new BaseQueue[queues.size()]);
+
+ txn.enqueue(queues,message, new ServerTransaction.Action()
+ {
+ MessageReference _reference = message.newReference();
+
+ public void postCommit()
+ {
+ for(int i = 0; i < baseQueues.length; i++)
+ {
+ try
+ {
+ baseQueues[i].enqueue(message, postEnqueueAction);
+ }
+ catch (AMQException e)
+ {
+ // TODO
+ throw new RuntimeException(e);
+ }
+ }
+ _reference.release();
+ }
+
+ public void onRollback()
+ {
+ _reference.release();
+ }
+ });
+ return queues.size();
+ }
+ }
+
protected abstract List<? extends BaseQueue> doRoute(final ServerMessage message,
final InstanceProperties instanceProperties);
@@ -679,4 +734,6 @@ public abstract class AbstractExchange i
public void onClose(Exchange exchange) throws AMQSecurityException, AMQInternalException;
}
+
+
}
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultExchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultExchange.java?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultExchange.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultExchange.java Tue Feb 4 11:13:42 2014
@@ -36,11 +36,14 @@ import org.apache.qpid.server.logging.Lo
import org.apache.qpid.server.logging.actors.CurrentActor;
import org.apache.qpid.server.logging.messages.ExchangeMessages;
import org.apache.qpid.server.message.InstanceProperties;
+import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.UUIDGenerator;
import org.apache.qpid.server.plugin.ExchangeType;
import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.BaseQueue;
import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.virtualhost.VirtualHost;
public class DefaultExchange implements Exchange
@@ -204,22 +207,6 @@ public class DefaultExchange implements
}
@Override
- public List<AMQQueue> route(ServerMessage message, final InstanceProperties instanceProperties)
- {
- AMQQueue q = _virtualHost.getQueue(message.getRoutingKey());
- if(q == null)
- {
- List<AMQQueue> noQueues = Collections.emptyList();
- return noQueues;
- }
- else
- {
- return Collections.singletonList(q);
- }
-
- }
-
- @Override
public boolean isBound(AMQQueue queue)
{
return _virtualHost.getQueue(queue.getName()) == queue;
@@ -343,4 +330,47 @@ public class DefaultExchange implements
{
return _id;
}
+
+ public final int send(final ServerMessage message,
+ final InstanceProperties instanceProperties,
+ final ServerTransaction txn,
+ final BaseQueue.PostEnqueueAction postEnqueueAction)
+ {
+ final AMQQueue q = _virtualHost.getQueue(message.getRoutingKey());
+ if(q == null)
+ {
+ return 0;
+ }
+ else
+ {
+ txn.enqueue(q,message, new ServerTransaction.Action()
+ {
+ MessageReference _reference = message.newReference();
+
+ public void postCommit()
+ {
+ try
+ {
+ q.enqueue(message, postEnqueueAction);
+ }
+ catch (AMQException e)
+ {
+ // TODO
+ throw new RuntimeException(e);
+ }
+ finally
+ {
+ _reference.release();
+ }
+ }
+
+ public void onRollback()
+ {
+ _reference.release();
+ }
+ });
+ return 1;
+ }
+ }
+
}
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/Exchange.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/Exchange.java?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/Exchange.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/Exchange.java Tue Feb 4 11:13:42 2014
@@ -29,6 +29,7 @@ import org.apache.qpid.server.message.Se
import org.apache.qpid.server.plugin.ExchangeType;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.virtualhost.VirtualHost;
import java.util.Collection;
@@ -94,13 +95,17 @@ public interface Exchange extends Exchan
void close() throws AMQException;
/**
- * Returns a list of queues to which to route this message. If there are
- * no queues the empty list must be returned.
- *
- * @return list of queues to which to route the message.
- */
- List<? extends BaseQueue> route(ServerMessage message, final InstanceProperties instanceProperties);
-
+ * Routes a message
+ * @param message the message to be routed
+ * @param instanceProperties the instance properties
+ * @param txn the transaction to enqueue within
+ * @param postEnqueueAction action to perform on the result of every enqueue (may be null)
+ * @return the number of queues in which the message was enqueued performed
+ */
+ int send(ServerMessage message,
+ InstanceProperties instanceProperties,
+ ServerTransaction txn,
+ BaseQueue.PostEnqueueAction postEnqueueAction);
/**
* Determines whether a message would be isBound to a particular queue using a specific routing key and arguments
Propchange: qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue:r1562453-1564250
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntry.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntry.java?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntry.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntry.java Tue Feb 4 11:13:42 2014
@@ -22,11 +22,11 @@ package org.apache.qpid.server.queue;
import org.apache.qpid.AMQException;
import org.apache.qpid.server.filter.Filterable;
-import org.apache.qpid.server.message.InstanceProperties;
-import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.txn.ServerTransaction;
-public interface QueueEntry extends Comparable<QueueEntry>
+public interface QueueEntry extends MessageInstance, Comparable<QueueEntry>
{
@@ -177,26 +177,17 @@ public interface QueueEntry extends Comp
AMQQueue getQueue();
- ServerMessage getMessage();
-
long getSize();
boolean getDeliveredToConsumer();
boolean expired() throws AMQException;
- boolean isAvailable();
-
- boolean isAcquired();
-
- boolean acquire();
boolean acquire(Subscription sub);
boolean acquiredBySubscription();
boolean isAcquiredBy(Subscription subscription);
- void release();
-
void setRedelivered();
boolean isRedelivered();
@@ -207,16 +198,7 @@ public interface QueueEntry extends Comp
boolean isRejectedBy(long subscriptionId);
- void delete();
-
- /**
- * Returns true if entry is either DEQUED or DELETED state.
- *
- * @return true if entry is either DEQUED or DELETED state
- */
- boolean isDeleted();
-
- void routeToAlternate();
+ int routeToAlternate(final BaseQueue.PostEnqueueAction action, ServerTransaction txn);
boolean isQueueDeleted();
@@ -241,5 +223,4 @@ public interface QueueEntry extends Comp
Filterable asFilterable();
- InstanceProperties getInstanceProperties();
}
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java Tue Feb 4 11:13:42 2014
@@ -34,7 +34,6 @@ import org.apache.qpid.server.txn.Server
import java.util.EnumMap;
import java.util.HashSet;
-import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
@@ -250,7 +249,7 @@ public abstract class QueueEntryImpl imp
}
else if(acquire())
{
- routeToAlternate();
+ routeToAlternate(null, null);
}
}
@@ -368,65 +367,43 @@ public abstract class QueueEntryImpl imp
dispose();
}
- public void routeToAlternate()
+ public int routeToAlternate(final BaseQueue.PostEnqueueAction action, ServerTransaction txn)
{
final AMQQueue currentQueue = getQueue();
Exchange alternateExchange = currentQueue.getAlternateExchange();
-
+ boolean autocommit = txn == null;
if (alternateExchange != null)
{
- List<? extends BaseQueue> queues = alternateExchange.route(getMessage(), getInstanceProperties());
- final ServerMessage message = getMessage();
- if ((queues == null || queues.size() == 0) && alternateExchange.getAlternateExchange() != null)
+ if(autocommit)
{
- queues = alternateExchange.getAlternateExchange().route(getMessage(), getInstanceProperties());
+ txn = new LocalTransaction(getQueue().getVirtualHost().getMessageStore());
}
+ int enqueues = alternateExchange.send(getMessage(), getInstanceProperties(), txn, action);
-
- if (queues != null && queues.size() != 0)
+ txn.dequeue(currentQueue, getMessage(), new ServerTransaction.Action()
{
- final List<? extends BaseQueue> rerouteQueues = queues;
- ServerTransaction txn = new LocalTransaction(getQueue().getVirtualHost().getMessageStore());
-
- txn.enqueue(rerouteQueues, message, new ServerTransaction.Action()
+ public void postCommit()
{
- public void postCommit()
- {
- try
- {
- for (BaseQueue queue : rerouteQueues)
- {
- queue.enqueue(message);
- }
- }
- catch (AMQException e)
- {
- throw new RuntimeException(e);
- }
- }
-
- public void onRollback()
- {
-
- }
- });
+ delete();
+ }
- txn.dequeue(currentQueue, message, new ServerTransaction.Action()
+ public void onRollback()
{
- public void postCommit()
- {
- delete();
- }
-
- public void onRollback()
- {
- }
- });
+ }
+ });
+ if(autocommit)
+ {
txn.commit();
}
+ return enqueues;
+
+ }
+ else
+ {
+ return 0;
}
}
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java Tue Feb 4 11:13:42 2014
@@ -1355,94 +1355,26 @@ public class SimpleAMQQueue implements A
ServerTransaction txn = new LocalTransaction(getVirtualHost().getMessageStore());
- if(_alternateExchange != null)
- {
-
- for(final QueueEntry entry : entries)
- {
-
- List<? extends BaseQueue> queues = _alternateExchange.route(entry.getMessage(), entry.getInstanceProperties());
- if((queues == null || queues.size() == 0) && _alternateExchange.getAlternateExchange() != null)
- {
- queues = _alternateExchange.getAlternateExchange().route(entry.getMessage(), entry.getInstanceProperties());
- }
-
- final ServerMessage message = entry.getMessage();
- if(queues != null && queues.size() != 0)
- {
- final List<? extends BaseQueue> rerouteQueues = queues;
- txn.enqueue(rerouteQueues, entry.getMessage(),
- new ServerTransaction.Action()
- {
-
- public void postCommit()
- {
- try
- {
- for(BaseQueue queue : rerouteQueues)
- {
- queue.enqueue(message);
- }
- }
- catch (AMQException e)
- {
- throw new RuntimeException(e);
- }
-
- }
-
- public void onRollback()
- {
-
- }
- });
- txn.dequeue(this, entry.getMessage(),
- new ServerTransaction.Action()
- {
-
- public void postCommit()
- {
- entry.delete();
- }
-
- public void onRollback()
- {
- }
- });
- }
-
- }
- _alternateExchange.removeReference(this);
- }
- else
+ for(final QueueEntry entry : entries)
{
- // TODO log discard
+ // TODO log requeues with a post enqueue action
+ int requeues = entry.routeToAlternate(null, txn);
- for(final QueueEntry entry : entries)
+ if(requeues == 0)
{
- final ServerMessage message = entry.getMessage();
- if(message != null)
- {
- txn.dequeue(this, message,
- new ServerTransaction.Action()
- {
-
- public void postCommit()
- {
- entry.delete();
- }
-
- public void onRollback()
- {
- }
- });
- }
+ // TODO log discard
}
}
txn.commit();
+ if(_alternateExchange != null)
+ {
+ _alternateExchange.removeReference(this);
+ }
+
+
for (Task task : _deleteTaskList)
{
task.doTask(this);
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java Tue Feb 4 11:13:42 2014
@@ -312,10 +312,9 @@ public class TopicExchangeTest extends Q
private int routeMessage(String routingKey, long messageNumber) throws AMQException
{
- ServerMessage serverMessage = mock(ServerMessage.class);
- when(serverMessage.getRoutingKey()).thenReturn(routingKey);
- List<? extends BaseQueue> queues = _exchange.route(serverMessage, InstanceProperties.EMPTY);
ServerMessage message = mock(ServerMessage.class);
+ when(message.getRoutingKey()).thenReturn(routingKey);
+ List<? extends BaseQueue> queues = _exchange.route(message, InstanceProperties.EMPTY);
MessageReference ref = mock(MessageReference.class);
when(ref.getMessage()).thenReturn(message);
when(message.newReference()).thenReturn(ref);
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockQueueEntry.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockQueueEntry.java?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockQueueEntry.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockQueueEntry.java Tue Feb 4 11:13:42 2014
@@ -26,6 +26,7 @@ import org.apache.qpid.server.message.AM
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.txn.ServerTransaction;
public class MockQueueEntry implements QueueEntry
{
@@ -62,9 +63,9 @@ public class MockQueueEntry implements Q
}
- public void routeToAlternate()
+ public int routeToAlternate(final BaseQueue.PostEnqueueAction action, final ServerTransaction txn)
{
-
+ return 0;
}
public boolean expired() throws AMQException
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java Tue Feb 4 11:13:42 2014
@@ -46,6 +46,7 @@ import org.apache.qpid.AMQStoreException
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.TransactionTimeoutHelper;
import org.apache.qpid.server.TransactionTimeoutHelper.CloseAction;
+import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.logging.LogActor;
import org.apache.qpid.server.logging.LogMessage;
import org.apache.qpid.server.logging.LogSubject;
@@ -53,6 +54,7 @@ import org.apache.qpid.server.logging.ac
import org.apache.qpid.server.logging.actors.GenericActor;
import org.apache.qpid.server.logging.messages.ChannelMessages;
import org.apache.qpid.server.logging.subjects.ChannelLogSubject;
+import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.protocol.AMQSessionModel;
@@ -102,6 +104,14 @@ public class ServerSession extends Sessi
private final AtomicBoolean _blocking = new AtomicBoolean(false);
private ChannelLogSubject _logSubject;
private final AtomicInteger _outstandingCredit = new AtomicInteger(UNLIMITED_CREDIT);
+ private final BaseQueue.PostEnqueueAction _checkCapacityAction = new BaseQueue.PostEnqueueAction()
+ {
+ @Override
+ public void onEnqueue(final QueueEntry entry)
+ {
+ entry.getQueue().checkCapacity(ServerSession.this);
+ }
+ };
public static interface MessageDispositionChangeListener
{
@@ -182,7 +192,9 @@ public class ServerSession extends Sessi
return isCommandsFull(id);
}
- public void enqueue(final MessageTransferMessage message, final List<? extends BaseQueue> queues)
+ public int enqueue(final MessageTransferMessage message,
+ final InstanceProperties instanceProperties,
+ final Exchange exchange)
{
if(_outstandingCredit.get() != UNLIMITED_CREDIT
&& _outstandingCredit.decrementAndGet() == (Integer.MAX_VALUE - PRODUCER_CREDIT_TOPUP_THRESHOLD))
@@ -190,10 +202,10 @@ public class ServerSession extends Sessi
_outstandingCredit.addAndGet(PRODUCER_CREDIT_TOPUP_THRESHOLD);
invoke(new MessageFlow("",MessageCreditUnit.MESSAGE, PRODUCER_CREDIT_TOPUP_THRESHOLD));
}
+ int enqueues = exchange.send(message, instanceProperties, _transaction, _checkCapacityAction);
getConnectionModel().registerMessageReceived(message.getSize(), message.getArrivalTime());
- PostEnqueueAction postTransactionAction = new PostEnqueueAction(queues, message, isTransactional()) ;
- _transaction.enqueue(queues,message, postTransactionAction);
incrementOutstandingTxnsIfNecessary();
+ return enqueues;
}
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java Tue Feb 4 11:13:42 2014
@@ -39,7 +39,6 @@ import org.apache.qpid.server.model.Queu
import org.apache.qpid.server.model.UUIDGenerator;
import org.apache.qpid.server.plugin.ExchangeType;
import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.BaseQueue;
import org.apache.qpid.server.queue.QueueArgumentsConverter;
import org.apache.qpid.server.security.SecurityManager;
import org.apache.qpid.server.store.DurableConfigurationStore;
@@ -337,28 +336,10 @@ public class ServerSessionDelegate exten
}
};
- List<? extends BaseQueue> queues = exchange.route(message, instanceProperties);
- if(queues.isEmpty() && exchange.getAlternateExchange() != null)
- {
- final Exchange alternateExchange = exchange.getAlternateExchange();
- queues = alternateExchange.route(message, instanceProperties);
- if (!queues.isEmpty())
- {
- exchangeInUse = alternateExchange;
- }
- else
- {
- exchangeInUse = exchange;
- }
- }
- else
- {
- exchangeInUse = exchange;
- }
+ int enqueues = serverSession.enqueue(message, instanceProperties, exchange);
- if(!queues.isEmpty())
+ if(enqueues != 0)
{
- serverSession.enqueue(message, queues);
storeMessage.flushToStore();
}
else
@@ -372,7 +353,7 @@ public class ServerSessionDelegate exten
}
else
{
- serverSession.getLogActor().message(ExchangeMessages.DISCARDMSG(exchangeInUse.getName(), messageMetaData.getRoutingKey()));
+ serverSession.getLogActor().message(ExchangeMessages.DISCARDMSG(exchange.getName(), messageMetaData.getRoutingKey()));
}
}
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Subscription_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Subscription_0_10.java?rev=1564254&r1=1564253&r2=1564254&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Subscription_0_10.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Subscription_0_10.java Tue Feb 4 11:13:42 2014
@@ -59,7 +59,6 @@ import java.text.MessageFormat;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -544,7 +543,7 @@ public class Subscription_0_10 implement
void reject(final QueueEntry entry)
{
entry.setRedelivered();
- entry.routeToAlternate();
+ entry.routeToAlternate(null, null);
if(entry.isAcquiredBy(this))
{
entry.delete();
@@ -575,35 +574,36 @@ public class Subscription_0_10 implement
protected void sendToDLQOrDiscard(QueueEntry entry)
{
- final Exchange alternateExchange = entry.getQueue().getAlternateExchange();
final LogActor logActor = CurrentActor.get();
final ServerMessage msg = entry.getMessage();
- if (alternateExchange != null)
+
+ int requeues = entry.routeToAlternate(new BaseQueue.PostEnqueueAction()
+ {
+ @Override
+ public void onEnqueue(final QueueEntry requeueEntry)
+ {
+ logActor.message( ChannelMessages.DEADLETTERMSG(msg.getMessageNumber(),
+ requeueEntry.getQueue().getName()));
+ }
+ }, null);
+
+ if (requeues == 0)
{
- final List<? extends BaseQueue> destinationQueues = alternateExchange.route(entry.getMessage(), entry.getInstanceProperties());
+ final AMQQueue queue = entry.getQueue();
+ final Exchange alternateExchange = queue.getAlternateExchange();
- if (destinationQueues == null || destinationQueues.isEmpty())
+ if(alternateExchange != null)
{
- entry.delete();
-
- logActor.message( ChannelMessages.DISCARDMSG_NOROUTE(msg.getMessageNumber(), alternateExchange.getName()));
+ logActor.message( ChannelMessages.DISCARDMSG_NOROUTE(msg.getMessageNumber(),
+ alternateExchange.getName()));
}
else
{
- entry.routeToAlternate();
-
- //output operational logging for each delivery post commit
- for (final BaseQueue destinationQueue : destinationQueues)
- {
- logActor.message( ChannelMessages.DEADLETTERMSG(msg.getMessageNumber(), destinationQueue.getName()));
- }
+ logActor.message(ChannelMessages.DISCARDMSG_NOALTEXCH(msg.getMessageNumber(),
+ queue.getName(),
+ msg.getRoutingKey()));
}
}
- else
- {
- entry.delete();
- logActor.message(ChannelMessages.DISCARDMSG_NOALTEXCH(msg.getMessageNumber(), entry.getQueue().getName(), msg.getRoutingKey()));
- }
}
private boolean isMaxDeliveryLimitReached(QueueEntry entry)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org