You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kp...@apache.org on 2013/02/28 17:14:57 UTC
svn commit: r1451244 [37/45] - in /qpid/branches/asyncstore: ./ bin/ cpp/
cpp/bindings/ cpp/bindings/qmf/ cpp/bindings/qmf/python/
cpp/bindings/qmf/ruby/ cpp/bindings/qmf2/ cpp/bindings/qmf2/examples/cpp/
cpp/bindings/qmf2/python/ cpp/bindings/qmf2/rub...
Modified: qpid/branches/asyncstore/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java (original)
+++ qpid/branches/asyncstore/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java Thu Feb 28 16:14:30 2013
@@ -124,7 +124,7 @@ public class TestAMQSession extends AMQS
return false;
}
- public void sendConsume(BasicMessageConsumer_0_8 consumer, AMQShortString queueName, AMQProtocolHandler protocolHandler, boolean nowait, int tag) throws AMQException, FailoverException
+ public void sendConsume(BasicMessageConsumer_0_8 consumer, AMQShortString queueName, boolean nowait, int tag) throws AMQException, FailoverException
{
}
@@ -139,13 +139,13 @@ public class TestAMQSession extends AMQS
return null;
}
- public void sendExchangeDeclare(AMQShortString name, AMQShortString type, AMQProtocolHandler protocolHandler, boolean nowait) throws AMQException, FailoverException
+ public void sendExchangeDeclare(AMQShortString name, AMQShortString type, boolean nowait, boolean durable, boolean autoDelete, boolean internal) throws AMQException, FailoverException
{
}
public void sendQueueDeclare(AMQDestination amqd, AMQProtocolHandler protocolHandler,
- boolean nowait, boolean passive) throws AMQException, FailoverException
+ boolean passive) throws AMQException, FailoverException
{
}
@@ -189,14 +189,6 @@ public class TestAMQSession extends AMQS
{
}
- public void handleAddressBasedDestination(AMQDestination dest,
- boolean isConsumer,
- boolean noWait) throws AMQException
- {
- throw new UnsupportedOperationException("The new addressing based sytanx is "
- + "not supported for AMQP 0-8/0-9 versions");
- }
-
@Override
protected void flushAcknowledgments()
{
Modified: qpid/branches/asyncstore/java/common.xml
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common.xml?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common.xml (original)
+++ qpid/branches/asyncstore/java/common.xml Thu Feb 28 16:14:30 2013
@@ -23,7 +23,10 @@
<dirname property="project.root" file="${ant.file.common}"/>
<property name="project.name" value="qpid"/>
- <property name="project.version" value="0.19"/>
+ <!-- Version used for standard build output -->
+ <property name="project.version" value="0.21"/>
+ <!-- The release version used for maven output. SNAPSHOT added via maven.version.suffix -->
+ <property name="project.version.maven" value="0.22"/>
<property name="project.url" value="http://qpid.apache.org"/>
<property name="project.groupid" value="org.apache.qpid"/>
<property name="project.namever" value="${project.name}-${project.version}"/>
@@ -42,7 +45,7 @@
<property name="build.report" location="${build}/report"/>
<property name="build.release" location="${build}/release"/>
<property name="build.release.prepare" location="${build.release}/prepare"/>
- <property name="build.plugins" location="${build}/lib/plugins"/>
+ <property name="build.scratch.broker.plugins.lib" location="${build.scratch}/broker-plugins/lib"/>
<property name="build.coverage.report" location="${build}/coverage/report"/>
<property name="build.coverage.src" location="${build}/coverage/src"/>
<property name="build.findbugs" location="${build}/findbugs"/>
@@ -93,6 +96,10 @@
<property name="nexus.host" value="repository.apache.org"/>
<property name="nexus.upload.url" value="https://${nexus.host}/service/local/staging/deploy/maven2"/>
+ <!-- properties for deplying snapshot artifacts -->
+ <property name="maven.snapshots.repo.id" value="apache.snapshots.https"/>
+ <property name="maven.snapshots.repo.url" value="https://${nexus.host}/content/repositories/snapshots"/>
+
<!-- properties for downloading ivy, and then our dependencies -->
<property name="ivy.jar.dir" value="${project.root}/lib/ivy" />
<property name="ivy.install.version" value="2.2.0" />
Modified: qpid/branches/asyncstore/java/common/bin/qpid-run
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/bin/qpid-run?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/bin/qpid-run (original)
+++ qpid/branches/asyncstore/java/common/bin/qpid-run Thu Feb 28 16:14:30 2013
@@ -88,10 +88,10 @@ SYSTEM_PROPS[${#SYSTEM_PROPS[@]}]="-DQPI
if [ -n "$QPID_LOG_PREFIX" ]; then
if [ "X$QPID_LOG_PREFIX" = "XPID" ]; then
log $INFO Using pid in qpid log name prefix
- LOG_PREFIX=" -Dlogprefix=$$"
+ LOG_PREFIX="-Dlogprefix=$$"
else
log $INFO Using qpid logprefix property
- LOG_PREFIX=" -Dlogprefix=$QPID_LOG_PREFIX"
+ LOG_PREFIX="-Dlogprefix=$QPID_LOG_PREFIX"
fi
SYSTEM_PROPS[${#SYSTEM_PROPS[@]}]="${LOG_PREFIX}"
fi
@@ -99,10 +99,10 @@ fi
if [ -n "$QPID_LOG_SUFFIX" ]; then
if [ "X$QPID_LOG_SUFFIX" = "XPID" ]; then
log $INFO Using pid in qpid log name suffix
- LOG_SUFFIX=" -Dlogsuffix=$$"
+ LOG_SUFFIX="-Dlogsuffix=$$"
else
log $INFO Using qpig logsuffix property
- LOG_SUFFIX=" -Dlogsuffix=$QPID_LOG_SUFFIX"
+ LOG_SUFFIX="-Dlogsuffix=$QPID_LOG_SUFFIX"
fi
SYSTEM_PROPS[${#SYSTEM_PROPS[@]}]="${LOG_SUFFIX}"
fi
Modified: qpid/branches/asyncstore/java/common/build.xml
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/build.xml?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/build.xml (original)
+++ qpid/branches/asyncstore/java/common/build.xml Thu Feb 28 16:14:30 2013
@@ -53,7 +53,15 @@
</target>
<target name="compile_gentools">
- <ant dir="${gentools.home}" />
+ <mkdir dir="${gentools.build}/classes"/>
+ <javac srcdir="${gentools.home}/src" destdir="${gentools.build}/classes" source="${java.source}" target="${java.target}" fork="true" includeantruntime="false">
+ <classpath>
+ <fileset dir="${project.root}">
+ <include name="${velocity.jar}"/>
+ <include name="${velocity-dep.jar}"/>
+ </fileset>
+ </classpath>
+ </javac>
</target>
<target name="check_gentool_deps">
@@ -64,15 +72,12 @@
<target name="gentools" depends="compile_gentools,check_gentool_deps" unless="gentools.notRequired">
<mkdir dir="${framing.generated.dir}"/>
- <java classname="org.apache.qpid.gentools.Main" fork="true" dir="${gentools.home}/src" failonerror="true">
+ <java classname="org.apache.qpid.gentools.Main" fork="true" dir="${gentools.build}/classes" failonerror="true">
<arg line='-j -o "${framing.generated.dir}" -t "${project.root}/common/templates" ${xml.spec.list}'/>
<classpath>
- <pathelement path="${gentools.home}/src" />
- <fileset dir="${gentools.home}/lib">
- <include name="**/*.jar"/>
- </fileset>
- <pathelement path="${gentools.home}/lib/velocity-1.4.jar" />
- <pathelement path="${gentools.home}/lib/velocity-dep-1.4.jar" />
+ <pathelement path="${gentools.build}/classes" />
+ <pathelement path="${project.root}/${velocity.jar}" />
+ <pathelement path="${project.root}/${velocity-dep.jar}" />
</classpath>
</java>
<touch file="${gentools.timestamp}" />
Modified: qpid/branches/asyncstore/java/common/src/main/java/common.bnd
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/common.bnd?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/common.bnd (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/common.bnd Thu Feb 28 16:14:30 2013
@@ -17,7 +17,7 @@
# under the License.
#
-ver: 0.19.0
+ver: 0.21.0
Bundle-SymbolicName: qpid-common
Bundle-Version: ${ver}
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java Thu Feb 28 16:14:30 2013
@@ -87,6 +87,8 @@ public class ClientProperties
public static final String USE_LEGACY_MAP_MESSAGE_FORMAT = "qpid.use_legacy_map_message";
+ public static final String USE_LEGACY_STREAM_MESSAGE_FORMAT = "qpid.use_legacy_stream_message";
+
public static final String AMQP_VERSION = "qpid.amqp.version";
public static final String QPID_VERIFY_CLIENT_ID = "qpid.verify_client_id";
@@ -190,6 +192,19 @@ public class ClientProperties
*/
public static final long DEFAULT_FLOW_CONTROL_WAIT_NOTIFY_PERIOD = 5000L;
+ /**
+ * System property to control whether the client will declare queues during
+ * consumer creation when using BindingURLs.
+ */
+ public static final String QPID_DECLARE_QUEUES_PROP_NAME = "qpid.declare_queues";
+
+ /**
+ * System property to control whether the client will declare exchanges during
+ * producer/consumer creation when using BindingURLs.
+ */
+ public static final String QPID_DECLARE_EXCHANGES_PROP_NAME = "qpid.declare_exchanges";
+ public static final String VERIFY_QUEUE_ON_SEND = "qpid.verify_queue_on_send";
+
private ClientProperties()
{
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java Thu Feb 28 16:14:30 2013
@@ -110,7 +110,7 @@ public final class AMQShortString implem
{
return new LinkedHashMap<AMQShortString, AMQShortString>()
{
-
+ @Override
protected boolean removeEldestEntry(Map.Entry<AMQShortString, AMQShortString> eldest)
{
return size() > LOCAL_INTERN_CACHE_SIZE;
@@ -845,22 +845,15 @@ public final class AMQShortString implem
return internString;
}
-
- public static void main(String args[])
+ public static String toString(AMQShortString amqShortString)
{
- AMQShortString s = new AMQShortString("a.b.c.d.e.f.g.h.i.j.k");
- AMQShortString s2 = s.substring(2, 7);
-
- AMQShortStringTokenizer t = s2.tokenize((byte) '.');
- while(t.hasMoreTokens())
- {
- System.err.println(t.nextToken());
- }
+ return amqShortString == null ? null : amqShortString.asString();
}
- public static String toString(AMQShortString amqShortString)
+ public static void clearLocalCache()
{
- return amqShortString == null ? null : amqShortString.asString();
+ _localInternMap.remove();
+ _localStringMap.remove();
}
}
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java Thu Feb 28 16:14:30 2013
@@ -855,6 +855,7 @@ public class FieldTable
public void addAll(FieldTable fieldTable)
{
initMapIfNecessary();
+ fieldTable.initMapIfNecessary();
if (fieldTable._properties != null)
{
_encodedForm = null;
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/properties/ConnectionStartProperties.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/properties/ConnectionStartProperties.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/properties/ConnectionStartProperties.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/properties/ConnectionStartProperties.java Thu Feb 28 16:14:30 2013
@@ -49,7 +49,11 @@ public class ConnectionStartProperties
public static final String SESSION_FLOW = "qpid.session_flow";
- public static int getPID()
+ public static int _pid;
+
+ public static final String _platformInfo;
+
+ static
{
RuntimeMXBean rtb = ManagementFactory.getRuntimeMXBean();
String processName = rtb.getName();
@@ -57,23 +61,20 @@ public class ConnectionStartProperties
{
try
{
- return Integer.parseInt(processName.substring(0,processName.indexOf('@')));
+ _pid = Integer.parseInt(processName.substring(0,processName.indexOf('@')));
}
catch(Exception e)
{
LOGGER.warn("Unable to get the PID due to error",e);
- return -1;
+ _pid = -1;
}
}
else
{
LOGGER.warn("Unable to get the PID due to unsupported format : " + processName);
- return -1;
+ _pid = -1;
}
- }
- public static String getPlatformInfo()
- {
StringBuilder fullSystemInfo = new StringBuilder(System.getProperty("java.runtime.name"));
fullSystemInfo.append(", ");
fullSystemInfo.append(System.getProperty("java.runtime.version"));
@@ -88,6 +89,16 @@ public class ConnectionStartProperties
fullSystemInfo.append(", ");
fullSystemInfo.append(System.getProperty("sun.os.patch.level"));
- return fullSystemInfo.toString();
+ _platformInfo = fullSystemInfo.toString();
+ }
+
+ public static int getPID()
+ {
+ return _pid;
+ }
+
+ public static String getPlatformInfo()
+ {
+ return _platformInfo;
}
}
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/protocol/ProtocolEngine.java Thu Feb 28 16:14:30 2013
@@ -23,6 +23,7 @@ package org.apache.qpid.protocol;
import org.apache.qpid.transport.Receiver;
import org.apache.qpid.transport.Sender;
import org.apache.qpid.transport.network.NetworkConnection;
+import org.apache.qpid.transport.network.TransportActivity;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
@@ -31,7 +32,7 @@ import java.nio.ByteBuffer;
* A ProtocolEngine is a Receiver for java.nio.ByteBuffers. It takes the data passed to it in the received
* decodes it and then process the result.
*/
-public interface ProtocolEngine extends Receiver<java.nio.ByteBuffer>
+public interface ProtocolEngine extends Receiver<java.nio.ByteBuffer>, TransportActivity
{
// Returns the remote address of the NetworkDriver
SocketAddress getRemoteAddress();
@@ -56,6 +57,6 @@ public interface ProtocolEngine extends
void readerIdle();
- public void setNetworkConnection(NetworkConnection network, Sender<ByteBuffer> sender);
+ public void setNetworkConnection(NetworkConnection network, Sender<ByteBuffer> sender);
}
\ No newline at end of file
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Connection.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Connection.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Connection.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Connection.java Thu Feb 28 16:14:30 2013
@@ -21,12 +21,7 @@
package org.apache.qpid.transport;
import org.apache.qpid.framing.ProtocolVersion;
-import org.apache.qpid.transport.network.Assembler;
-import org.apache.qpid.transport.network.Disassembler;
-import org.apache.qpid.transport.network.InputHandler;
-import org.apache.qpid.transport.network.NetworkConnection;
-import org.apache.qpid.transport.network.OutgoingNetworkTransport;
-import org.apache.qpid.transport.network.Transport;
+import org.apache.qpid.transport.network.*;
import org.apache.qpid.transport.network.security.SecurityLayer;
import org.apache.qpid.transport.network.security.SecurityLayerFactory;
import org.apache.qpid.transport.util.Logger;
@@ -73,6 +68,8 @@ public class Connection extends Connecti
//Usable channels are numbered 0 to <ChannelMax> - 1
public static final int MAX_CHANNEL_MAX = 0xFFFF;
public static final int MIN_USABLE_CHANNEL_NUM = 0;
+ private long _lastSendTime;
+ private long _lastReadTime;
public enum State { NEW, CLOSED, OPENING, OPEN, CLOSING, CLOSE_RCVD, RESUMING }
@@ -89,15 +86,15 @@ public class Connection extends Connecti
public static interface SessionFactory
{
- Session newSession(Connection conn, Binary name, long expiry);
+ Session newSession(Connection conn, Binary name, long expiry, boolean isNoReplay);
}
private static final class DefaultSessionFactory implements SessionFactory
{
- public Session newSession(final Connection conn, final Binary name, final long expiry)
+ public Session newSession(final Connection conn, final Binary name, final long expiry, final boolean isNoReplay)
{
- return new Session(conn, name, expiry);
+ return new Session(conn, name, expiry, isNoReplay);
}
}
@@ -232,9 +229,10 @@ public class Connection extends Connecti
addConnectionListener((ConnectionListener)secureReceiver);
}
- NetworkConnection network = transport.connect(settings, secureReceiver, null);
- _remoteAddress = network.getRemoteAddress();
- _localAddress = network.getLocalAddress();
+ NetworkConnection network = transport.connect(settings, secureReceiver, new ConnectionActivity());
+
+ setRemoteAddress(network.getRemoteAddress());
+ setLocalAddress(network.getLocalAddress());
final Sender<ByteBuffer> secureSender = securityLayer.sender(network.getSender());
if(secureSender instanceof ConnectionListener)
@@ -298,7 +296,12 @@ public class Connection extends Connecti
public Session createSession(long expiry)
{
- return createSession(UUID.randomUUID().toString(), expiry);
+ return createSession(expiry, false);
+ }
+
+ public Session createSession(long expiry, boolean isNoReplay)
+ {
+ return createSession(UUID.randomUUID().toString(), expiry, isNoReplay);
}
public Session createSession(String name)
@@ -311,6 +314,11 @@ public class Connection extends Connecti
return createSession(Strings.toUTF8(name), expiry);
}
+ public Session createSession(String name, long expiry,boolean isNoReplay)
+ {
+ return createSession(new Binary(Strings.toUTF8(name)), expiry, isNoReplay);
+ }
+
public Session createSession(byte[] name, long expiry)
{
return createSession(new Binary(name), expiry);
@@ -318,6 +326,11 @@ public class Connection extends Connecti
public Session createSession(Binary name, long expiry)
{
+ return createSession(name, expiry, false);
+ }
+
+ public Session createSession(Binary name, long expiry, boolean isNoReplay)
+ {
synchronized (lock)
{
Waiter w = new Waiter(lock, timeout);
@@ -331,7 +344,7 @@ public class Connection extends Connecti
throw new ConnectionException("Timed out waiting for connection to be ready. Current state is :" + state);
}
- Session ssn = _sessionFactory.newSession(this, name, expiry);
+ Session ssn = _sessionFactory.newSession(this, name, expiry, isNoReplay);
registerSession(ssn);
map(ssn);
ssn.attach();
@@ -369,6 +382,7 @@ public class Connection extends Connecti
public void received(ProtocolEvent event)
{
+ _lastReadTime = System.currentTimeMillis();
if(log.isDebugEnabled())
{
log.debug("RECV: [%s] %s", this, event);
@@ -378,6 +392,7 @@ public class Connection extends Connecti
public void send(ProtocolEvent event)
{
+ _lastSendTime = System.currentTimeMillis();
if(log.isDebugEnabled())
{
log.debug("SEND: [%s] %s", this, event);
@@ -728,6 +743,17 @@ public class Connection extends Connecti
return _localAddress;
}
+ protected void setRemoteAddress(SocketAddress remoteAddress)
+ {
+ _remoteAddress = remoteAddress;
+ }
+
+ protected void setLocalAddress(SocketAddress localAddress)
+ {
+ _localAddress = localAddress;
+ }
+
+
private void invokeSessionDetached(int channel, SessionDetachCode sessionDetachCode)
{
SessionDetached sessionDetached = new SessionDetached();
@@ -735,4 +761,38 @@ public class Connection extends Connecti
sessionDetached.setCode(sessionDetachCode);
invoke(sessionDetached);
}
+
+
+ protected void doHeartBeat()
+ {
+ connectionHeartbeat();
+ }
+
+ private class ConnectionActivity implements TransportActivity
+ {
+ @Override
+ public long getLastReadTime()
+ {
+ return _lastReadTime;
+ }
+
+ @Override
+ public long getLastWriteTime()
+ {
+ return _lastSendTime;
+ }
+
+ @Override
+ public void writerIdle()
+ {
+ connectionHeartbeat();
+ }
+
+ @Override
+ public void readerIdle()
+ {
+ // TODO
+
+ }
+ }
}
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/NetworkTransportConfiguration.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/NetworkTransportConfiguration.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/NetworkTransportConfiguration.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/NetworkTransportConfiguration.java Thu Feb 28 16:14:30 2013
@@ -38,14 +38,6 @@ public interface NetworkTransportConfigu
// The amount of memory in bytes to allocate to the outgoing buffer
Integer getSendBufferSize();
- Integer getPort();
-
- String getHost();
-
- String getTransport();
-
- Integer getConnectorProcessors();
-
InetSocketAddress getAddress();
boolean needClientAuth();
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java Thu Feb 28 16:14:30 2013
@@ -71,7 +71,8 @@ public class ServerDelegate extends Conn
if (mechanism == null || mechanism.length() == 0)
{
- tuneAuthorizedConnection(conn);
+ conn.connectionClose(ConnectionCloseCode.CONNECTION_FORCED,
+ "No Sasl mechanism was specified");
return;
}
@@ -82,7 +83,7 @@ public class ServerDelegate extends Conn
if (ss == null)
{
conn.connectionClose(ConnectionCloseCode.CONNECTION_FORCED,
- "null SASL mechanism: " + mechanism);
+ "No SaslServer could be created for mechanism: " + mechanism);
return;
}
conn.setSaslServer(ss);
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Session.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Session.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Session.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Session.java Thu Feb 28 16:14:30 2013
@@ -25,7 +25,6 @@ import org.apache.qpid.configuration.Cli
import org.apache.qpid.transport.network.Frame;
import org.apache.qpid.transport.util.Logger;
import org.apache.qpid.transport.util.Waiter;
-
import static org.apache.qpid.transport.Option.COMPLETED;
import static org.apache.qpid.transport.Option.SYNC;
import static org.apache.qpid.transport.Option.TIMELY_REPLY;
@@ -132,19 +131,31 @@ public class Session extends SessionInvo
private final Object stateLock = new Object();
private final AtomicBoolean _failoverRequired = new AtomicBoolean(false);
+ private boolean _isNoReplay = false;
protected Session(Connection connection, Binary name, long expiry)
{
this(connection, new SessionDelegate(), name, expiry);
}
+ protected Session(Connection connection, Binary name, long expiry, boolean noReplay)
+ {
+ this(connection, new SessionDelegate(), name, expiry, noReplay);
+ }
+
protected Session(Connection connection, SessionDelegate delegate, Binary name, long expiry)
{
+ this(connection, delegate, name, expiry,false);
+ }
+
+ protected Session(Connection connection, SessionDelegate delegate, Binary name, long expiry, boolean noReplay)
+ {
this.connection = connection;
this.delegate = delegate;
this.name = name;
this.expiry = expiry;
this.closing = false;
+ this._isNoReplay = noReplay;
initReceiver();
}
@@ -282,6 +293,7 @@ public class Session extends SessionInvo
void resume()
{
_failoverRequired.set(false);
+
synchronized (commandsLock)
{
attach();
@@ -414,7 +426,7 @@ public class Session extends SessionInvo
if(log.isDebugEnabled())
{
- log.debug("ID: [%s] %s", this.channel, id);
+ log.debug("identify: ch=%s, commandId=%s", this.channel, id);
}
if ((id & 0xff) == 0)
@@ -443,7 +455,7 @@ public class Session extends SessionInvo
{
if(log.isDebugEnabled())
{
- log.debug("%s processed([%d,%d]) %s %s", this, lower, upper, syncPoint, maxProcessed);
+ log.debug("%s ch=%s processed([%d,%d]) %s %s", this, channel, lower, upper, syncPoint, maxProcessed);
}
boolean flush;
@@ -451,7 +463,7 @@ public class Session extends SessionInvo
{
if(log.isDebugEnabled())
{
- log.debug("%s", processed);
+ log.debug("%s processed: %s", this, processed);
}
if (ge(upper, commandsIn))
@@ -740,7 +752,7 @@ public class Session extends SessionInvo
sessionCommandPoint(0, 0);
}
- boolean replayTransfer = !closing && !transacted &&
+ boolean replayTransfer = !_isNoReplay && !closing && !transacted &&
m instanceof MessageTransfer &&
! m.isUnreliable();
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/IncomingNetworkTransport.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/IncomingNetworkTransport.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/IncomingNetworkTransport.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/IncomingNetworkTransport.java Thu Feb 28 16:14:30 2013
@@ -27,5 +27,7 @@ import javax.net.ssl.SSLContext;
public interface IncomingNetworkTransport extends NetworkTransport
{
- public void accept(NetworkTransportConfiguration config, ProtocolEngineFactory factory, SSLContext sslContext);
+ public void accept(NetworkTransportConfiguration config,
+ ProtocolEngineFactory factory,
+ SSLContext sslContext);
}
\ No newline at end of file
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java Thu Feb 28 16:14:30 2013
@@ -50,4 +50,8 @@ public interface NetworkConnection
void setPeerPrincipal(Principal principal);
Principal getPeerPrincipal();
+
+ int getMaxReadIdle();
+
+ int getMaxWriteIdle();
}
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/OutgoingNetworkTransport.java Thu Feb 28 16:14:30 2013
@@ -23,12 +23,13 @@ package org.apache.qpid.transport.networ
import org.apache.qpid.transport.ConnectionSettings;
import org.apache.qpid.transport.Receiver;
-import javax.net.ssl.SSLContext;
import java.nio.ByteBuffer;
public interface OutgoingNetworkTransport extends NetworkTransport
{
public NetworkConnection getConnection();
- public NetworkConnection connect(ConnectionSettings settings, Receiver<ByteBuffer> delegate, SSLContext sslContext);
+ public NetworkConnection connect(ConnectionSettings settings,
+ Receiver<ByteBuffer> delegate,
+ TransportActivity transportActivity);
}
\ No newline at end of file
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java Thu Feb 28 16:14:30 2013
@@ -26,7 +26,9 @@ import java.nio.ByteBuffer;
import java.security.Principal;
import org.apache.qpid.transport.Receiver;
import org.apache.qpid.transport.Sender;
+import org.apache.qpid.transport.network.Ticker;
import org.apache.qpid.transport.network.NetworkConnection;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,14 +40,23 @@ public class IoNetworkConnection impleme
private final IoSender _ioSender;
private final IoReceiver _ioReceiver;
private Principal _principal;
+ private int _maxReadIdle;
+ private int _maxWriteIdle;
public IoNetworkConnection(Socket socket, Receiver<ByteBuffer> delegate,
- int sendBufferSize, int receiveBufferSize, long timeout)
+ int sendBufferSize, int receiveBufferSize, long timeout)
+ {
+ this(socket,delegate,sendBufferSize,receiveBufferSize,timeout,null);
+ }
+
+ public IoNetworkConnection(Socket socket, Receiver<ByteBuffer> delegate,
+ int sendBufferSize, int receiveBufferSize, long timeout, Ticker ticker)
{
_socket = socket;
_timeout = timeout;
_ioReceiver = new IoReceiver(_socket, delegate, receiveBufferSize,_timeout);
+ _ioReceiver.setTicker(ticker);
_ioSender = new IoSender(_socket, 2 * sendBufferSize, _timeout);
@@ -88,14 +99,12 @@ public class IoNetworkConnection impleme
public void setMaxWriteIdle(int sec)
{
- // TODO implement support for setting heartbeating config in this way
- // Currently a socket timeout is used in IoSender
+ _maxWriteIdle = sec;
}
public void setMaxReadIdle(int sec)
{
- // TODO implement support for setting heartbeating config in this way
- // Currently a socket timeout is used in IoSender
+ _maxReadIdle = sec;
}
@Override
@@ -109,4 +118,16 @@ public class IoNetworkConnection impleme
{
return _principal;
}
+
+ @Override
+ public int getMaxReadIdle()
+ {
+ return _maxReadIdle;
+ }
+
+ @Override
+ public int getMaxWriteIdle()
+ {
+ return _maxWriteIdle;
+ }
}
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java Thu Feb 28 16:14:30 2013
@@ -41,9 +41,8 @@ import org.apache.qpid.transport.Connect
import org.apache.qpid.transport.NetworkTransportConfiguration;
import org.apache.qpid.transport.Receiver;
import org.apache.qpid.transport.TransportException;
-import org.apache.qpid.transport.network.IncomingNetworkTransport;
-import org.apache.qpid.transport.network.NetworkConnection;
-import org.apache.qpid.transport.network.OutgoingNetworkTransport;
+import org.apache.qpid.transport.network.*;
+
import org.slf4j.LoggerFactory;
public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNetworkTransport
@@ -56,7 +55,9 @@ public class IoNetworkTransport implemen
private IoNetworkConnection _connection;
private AcceptingThread _acceptor;
- public NetworkConnection connect(ConnectionSettings settings, Receiver<ByteBuffer> delegate, SSLContext sslContext)
+ public NetworkConnection connect(ConnectionSettings settings,
+ Receiver<ByteBuffer> delegate,
+ TransportActivity transportActivity)
{
int sendBufferSize = settings.getWriteBufferSize();
int receiveBufferSize = settings.getReadBufferSize();
@@ -91,7 +92,9 @@ public class IoNetworkTransport implemen
try
{
- _connection = new IoNetworkConnection(_socket, delegate, sendBufferSize, receiveBufferSize, TIMEOUT);
+ IdleTimeoutTicker ticker = new IdleTimeoutTicker(transportActivity, TIMEOUT);
+ _connection = new IoNetworkConnection(_socket, delegate, sendBufferSize, receiveBufferSize, TIMEOUT, ticker);
+ ticker.setConnection(_connection);
_connection.start();
}
catch(Exception e)
@@ -128,9 +131,10 @@ public class IoNetworkTransport implemen
return _connection;
}
- public void accept(NetworkTransportConfiguration config, ProtocolEngineFactory factory, SSLContext sslContext)
+ public void accept(NetworkTransportConfiguration config,
+ ProtocolEngineFactory factory,
+ SSLContext sslContext)
{
-
try
{
_acceptor = new AcceptingThread(config, factory, sslContext);
@@ -141,8 +145,6 @@ public class IoNetworkTransport implemen
{
throw new TransportException("Unable to start server socket", e);
}
-
-
}
private class AcceptingThread extends Thread
@@ -152,15 +154,16 @@ public class IoNetworkTransport implemen
private ProtocolEngineFactory _factory;
private SSLContext _sslContext;
private ServerSocket _serverSocket;
+ private int _timeout;
private AcceptingThread(NetworkTransportConfiguration config,
ProtocolEngineFactory factory,
- SSLContext sslContext)
- throws IOException
+ SSLContext sslContext) throws IOException
{
_config = config;
_factory = factory;
_sslContext = sslContext;
+ _timeout = TIMEOUT;
InetSocketAddress address = config.getAddress();
@@ -172,15 +175,19 @@ public class IoNetworkTransport implemen
{
SSLServerSocketFactory socketFactory = _sslContext.getServerSocketFactory();
_serverSocket = socketFactory.createServerSocket();
- ((SSLServerSocket)_serverSocket).setNeedClientAuth(config.needClientAuth());
- ((SSLServerSocket)_serverSocket).setWantClientAuth(config.wantClientAuth());
+ if(config.needClientAuth())
+ {
+ ((SSLServerSocket)_serverSocket).setNeedClientAuth(true);
+ }
+ else if(config.wantClientAuth())
+ {
+ ((SSLServerSocket)_serverSocket).setWantClientAuth(true);
+ }
}
_serverSocket.setReuseAddress(true);
_serverSocket.bind(address);
-
-
}
@@ -217,6 +224,7 @@ public class IoNetworkTransport implemen
{
socket = _serverSocket.accept();
socket.setTcpNoDelay(_config.getTcpNoDelay());
+ socket.setSoTimeout(_timeout);
final Integer sendBufferSize = _config.getSendBufferSize();
final Integer receiveBufferSize = _config.getReceiveBufferSize();
@@ -224,10 +232,12 @@ public class IoNetworkTransport implemen
socket.setSendBufferSize(sendBufferSize);
socket.setReceiveBufferSize(receiveBufferSize);
-
ProtocolEngine engine = _factory.newProtocolEngine();
- NetworkConnection connection = new IoNetworkConnection(socket, engine, sendBufferSize, receiveBufferSize, TIMEOUT);
+ final IdleTimeoutTicker ticker = new IdleTimeoutTicker(engine, TIMEOUT);
+ NetworkConnection connection = new IoNetworkConnection(socket, engine, sendBufferSize, receiveBufferSize, _timeout,
+ ticker);
+ ticker.setConnection(connection);
if(_sslContext != null)
{
@@ -248,14 +258,14 @@ public class IoNetworkTransport implemen
}
catch(RuntimeException e)
{
- LOGGER.error("Error in Acceptor thread on port " + _config.getPort(), e);
+ LOGGER.error("Error in Acceptor thread on address " + _config.getAddress(), e);
closeSocketIfNecessary(socket);
}
catch(IOException e)
{
if(!_closed)
{
- LOGGER.error("Error in Acceptor thread on port " + _config.getPort(), e);
+ LOGGER.error("Error in Acceptor thread on address " + _config.getAddress(), e);
closeSocketIfNecessary(socket);
try
{
@@ -275,7 +285,7 @@ public class IoNetworkTransport implemen
{
if(LOGGER.isDebugEnabled())
{
- LOGGER.debug("Acceptor exiting, no new connections will be accepted on port " + _config.getPort());
+ LOGGER.debug("Acceptor exiting, no new connections will be accepted on address " + _config.getAddress());
}
}
}
@@ -294,6 +304,7 @@ public class IoNetworkTransport implemen
}
}
}
+
}
}
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java Thu Feb 28 16:14:30 2013
@@ -24,6 +24,7 @@ import org.apache.qpid.common.Closeable;
import org.apache.qpid.thread.Threading;
import org.apache.qpid.transport.Receiver;
import org.apache.qpid.transport.TransportException;
+import org.apache.qpid.transport.network.Ticker;
import org.apache.qpid.transport.util.Logger;
import javax.net.ssl.SSLSocket;
@@ -31,6 +32,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.net.Socket;
import java.net.SocketException;
+import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -51,6 +53,8 @@ final class IoReceiver implements Runnab
private final AtomicBoolean closed = new AtomicBoolean(false);
private final Thread receiverThread;
private static final boolean shutdownBroken;
+
+ private Ticker _ticker;
static
{
String osName = System.getProperty("os.name");
@@ -136,7 +140,7 @@ final class IoReceiver implements Runnab
{
final int threshold = bufferSize / 2;
- // I set the read buffer size simillar to SO_RCVBUF
+ // I set the read buffer size similar to SO_RCVBUF
// Haven't tested with a lower value to see if it's better or worse
byte[] buffer = new byte[bufferSize];
try
@@ -144,27 +148,71 @@ final class IoReceiver implements Runnab
InputStream in = socket.getInputStream();
int read = 0;
int offset = 0;
- while ((read = in.read(buffer, offset, bufferSize-offset)) != -1)
+ long currentTime;
+ while(read != -1)
{
- if (read > 0)
+ try
{
- ByteBuffer b = ByteBuffer.wrap(buffer,offset,read);
- receiver.received(b);
- offset+=read;
- if (offset > threshold)
+ while ((read = in.read(buffer, offset, bufferSize-offset)) != -1)
{
- offset = 0;
- buffer = new byte[bufferSize];
+ if (read > 0)
+ {
+ ByteBuffer b = ByteBuffer.wrap(buffer,offset,read);
+ receiver.received(b);
+ offset+=read;
+ if (offset > threshold)
+ {
+ offset = 0;
+ buffer = new byte[bufferSize];
+ }
+ }
+ currentTime = System.currentTimeMillis();
+
+ if(_ticker != null)
+ {
+ int tick = _ticker.getTimeToNextTick(currentTime);
+ if(tick <= 0)
+ {
+ tick = _ticker.tick(currentTime);
+ }
+ try
+ {
+ if(!socket.isClosed())
+ {
+ socket.setSoTimeout(tick <= 0 ? 1 : tick);
+ }
+ }
+ catch(SocketException e)
+ {
+ // ignore - closed socket
+ }
+ }
+ }
+ }
+ catch (SocketTimeoutException e)
+ {
+ currentTime = System.currentTimeMillis();
+ if(_ticker != null)
+ {
+ final int tick = _ticker.tick(currentTime);
+ if(!socket.isClosed())
+ {
+ try
+ {
+ socket.setSoTimeout(tick <= 0 ? 1 : tick );
+ }
+ catch(SocketException ex)
+ {
+ // ignore - closed socket
+ }
+ }
}
}
}
}
catch (Throwable t)
{
- if (!(shutdownBroken &&
- t instanceof SocketException &&
- t.getMessage().equalsIgnoreCase("socket closed") &&
- closed.get()))
+ if (shouldReport(t))
{
receiver.exception(t);
}
@@ -183,4 +231,30 @@ final class IoReceiver implements Runnab
}
}
+ private boolean shouldReport(Throwable t)
+ {
+ boolean brokenClose = closed.get() &&
+ shutdownBroken &&
+ t instanceof SocketException &&
+ "socket closed".equalsIgnoreCase(t.getMessage());
+
+ boolean sslSocketClosed = closed.get() &&
+ socket instanceof SSLSocket &&
+ t instanceof SocketException &&
+ "Socket is closed".equalsIgnoreCase(t.getMessage());
+
+ return !brokenClose && !sslSocketClosed;
+ }
+
+ public Ticker getTicker()
+ {
+ return _ticker;
+ }
+
+ public void setTicker(Ticker ticker)
+ {
+ _ticker = ticker;
+ }
+
+
}
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/url/BindingURL.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/url/BindingURL.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/url/BindingURL.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/url/BindingURL.java Thu Feb 28 16:14:30 2013
@@ -36,6 +36,9 @@ public interface BindingURL
public static final String OPTION_SUBSCRIPTION = "subscription";
public static final String OPTION_ROUTING_KEY = "routingkey";
public static final String OPTION_BINDING_KEY = "bindingkey";
+ public static final String OPTION_EXCHANGE_AUTODELETE = "exchangeautodelete";
+ public static final String OPTION_EXCHANGE_DURABLE = "exchangedurable";
+ public static final String OPTION_EXCHANGE_INTERNAL = "exchangeinternal";
/**
* This option is only applicable for 0-8/0-9/0-9-1 protocols connection
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/util/FileUtils.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/util/FileUtils.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/util/FileUtils.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/util/FileUtils.java Thu Feb 28 16:14:30 2013
@@ -220,6 +220,19 @@ public class FileUtils
public static void copyCheckedEx(File src, File dst) throws IOException
{
InputStream in = new FileInputStream(src);
+ copy(in, dst);
+ }
+
+ /**
+ * Copies the specified InputStream to the specified destination file. If the destination file does not exist,
+ * it is created.
+ *
+ * @param in The InputStream
+ * @param dst The destination file name.
+ * @throws IOException
+ */
+ public static void copy(InputStream in, File dst) throws IOException
+ {
try
{
if (!dst.exists())
Modified: qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/test/utils/QpidTestCase.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/test/utils/QpidTestCase.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/test/utils/QpidTestCase.java (original)
+++ qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/test/utils/QpidTestCase.java Thu Feb 28 16:14:30 2013
@@ -41,6 +41,7 @@ public class QpidTestCase extends TestCa
public static final String QPID_HOME = System.getProperty("QPID_HOME");
public static final String TEST_RESOURCES_DIR = QPID_HOME + "/../test-profiles/test_resources/";
public static final String TMP_FOLDER = System.getProperty("java.io.tmpdir");
+ public static final String LOG4J_CONFIG_FILE_PATH = System.getProperty("log4j.configuration.file");
private static final Logger _logger = Logger.getLogger(QpidTestCase.class);
@@ -115,12 +116,7 @@ public class QpidTestCase extends TestCa
public QpidTestCase()
{
- this("QpidTestCase");
- }
-
- public QpidTestCase(String name)
- {
- super(name);
+ super();
}
public void run(TestResult testResult)
@@ -204,6 +200,8 @@ public class QpidTestCase extends TestCa
{
System.setProperty(property, value);
}
+
+ _logger.info("Set system property \"" + property + "\" to: \"" + value + "\"");
}
/**
Modified: qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/test/utils/TestFileUtils.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/test/utils/TestFileUtils.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/test/utils/TestFileUtils.java (original)
+++ qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/test/utils/TestFileUtils.java Thu Feb 28 16:14:30 2013
@@ -21,6 +21,12 @@
package org.apache.qpid.test.utils;
import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+
+import java.io.FileOutputStream;
+
+import junit.framework.TestCase;
import org.apache.qpid.util.FileUtils;
@@ -30,6 +36,7 @@ import org.apache.qpid.util.FileUtils;
public class TestFileUtils
{
private static final String SYSTEM_TMP_DIR = System.getProperty("java.io.tmpdir");
+ private static final String SUFFIX = "tmp";
/**
* Create and return a temporary directory that will be deleted on exit.
@@ -60,4 +67,87 @@ public class TestFileUtils
return testDir;
}
+
+ public static File createTempFile(TestCase testcase)
+ {
+ return createTempFile(testcase, SUFFIX);
+ }
+
+ public static File createTempFile(TestCase testcase, String suffix)
+ {
+ String prefix = testcase.getClass().getSimpleName() + "-" + testcase.getName();
+
+ File tmpFile;
+ try
+ {
+ tmpFile = File.createTempFile(prefix, suffix);
+ tmpFile.deleteOnExit();
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException("Cannot create temporary file with prefix " + prefix + " and suffix " + SUFFIX, e);
+ }
+
+ return tmpFile;
+ }
+
+ /**
+ * Creates a temporary file from the resource name given, using the resource name as the file suffix.
+ *
+ * This is required because the tests use the jar files as their class path.
+ */
+ public static File createTempFileFromResource(TestCase testCase, String resourceName)
+ {
+ File dst = createTempFile(testCase, resourceName);
+ InputStream in = testCase.getClass().getResourceAsStream(resourceName);
+ try
+ {
+ FileUtils.copy(in, dst);
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException("Cannot copy resource " + resourceName +
+ " to temp file " + dst.getAbsolutePath(), e);
+ }
+ dst.deleteOnExit();
+ return dst;
+ }
+
+ /**
+ * Creates a temporary file for given test with given suffix in file name.
+ * The given content is stored in the file using UTF-8 encoding.
+ */
+ public static File createTempFile(TestCase testcase, String suffix, String content)
+ {
+ File file = createTempFile(testcase, suffix);
+ if (content != null)
+ {
+ FileOutputStream fos = null;
+ try
+ {
+ fos = new FileOutputStream(file);
+ fos.write(content.getBytes("UTF-8"));
+ fos.flush();
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException("Cannot add the content into temp file " + file.getAbsolutePath(), e);
+ }
+ finally
+ {
+ if (fos != null)
+ {
+ try
+ {
+ fos.close();
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException("Cannot close output stream into temp file " + file.getAbsolutePath(), e);
+ }
+ }
+ }
+ }
+ return file;
+ }
}
Modified: qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java (original)
+++ qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java Thu Feb 28 16:14:30 2013
@@ -155,6 +155,7 @@ public class ConnectionTest extends Qpid
{
final Connection conn = new Connection();
conn.setConnectionDelegate(new ClientDelegate(new ConnectionSettings()));
+
conn.addConnectionListener(new ConnectionListener()
{
public void opened(Connection conn) {}
@@ -225,6 +226,12 @@ public class ConnectionTest extends Qpid
ssn.setSessionListener(ConnectionTest.this);
return ssn;
}
+
+ @Override
+ public void connectionStartOk(Connection conn, ConnectionStartOk ok)
+ {
+ tuneAuthorizedConnection(conn);
+ }
};
try
Modified: qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java (original)
+++ qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java Thu Feb 28 16:14:30 2013
@@ -83,6 +83,18 @@ public class TestNetworkConnection imple
return null;
}
+ @Override
+ public int getMaxReadIdle()
+ {
+ return 0;
+ }
+
+ @Override
+ public int getMaxWriteIdle()
+ {
+ return 0;
+ }
+
public void setMaxWriteIdle(int idleTime)
{
Modified: qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java (original)
+++ qpid/branches/asyncstore/java/common/src/test/java/org/apache/qpid/transport/network/TransportTest.java Thu Feb 28 16:14:30 2013
@@ -128,7 +128,8 @@ public class TransportTest extends QpidT
}
public NetworkConnection connect(ConnectionSettings settings,
- Receiver<ByteBuffer> delegate, SSLContext sslContext)
+ Receiver<ByteBuffer> delegate,
+ TransportActivity transportActivity)
{
throw new UnsupportedOperationException();
}
@@ -148,7 +149,7 @@ public class TransportTest extends QpidT
}
public void accept(NetworkTransportConfiguration config,
- ProtocolEngineFactory factory, SSLContext sslContext)
+ ProtocolEngineFactory factory, SSLContext sslContext)
{
throw new UnsupportedOperationException();
}
Modified: qpid/branches/asyncstore/java/ivy.nexus.xml
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/ivy.nexus.xml?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/ivy.nexus.xml (original)
+++ qpid/branches/asyncstore/java/ivy.nexus.xml Thu Feb 28 16:14:30 2013
@@ -39,6 +39,42 @@
<artifact name="qpid-broker" type="jar.asc" ext="jar.asc"/>
<artifact name="qpid-broker" type="source" ext="jar" e:classifier="sources"/>
<artifact name="qpid-broker" type="source.asc" ext="jar.asc" e:classifier="sources"/>
+ <artifact name="qpid-broker-plugins-access-control" type="pom" ext="pom"/>
+ <artifact name="qpid-broker-plugins-access-control" type="pom.asc" ext="pom.asc"/>
+ <artifact name="qpid-broker-plugins-access-control" type="jar" ext="jar"/>
+ <artifact name="qpid-broker-plugins-access-control" type="jar.asc" ext="jar.asc"/>
+ <artifact name="qpid-broker-plugins-access-control" type="source" ext="jar" e:classifier="sources"/>
+ <artifact name="qpid-broker-plugins-access-control" type="source.asc" ext="jar.asc" e:classifier="sources"/>
+ <artifact name="qpid-broker-plugins-management-http" type="pom" ext="pom"/>
+ <artifact name="qpid-broker-plugins-management-http" type="pom.asc" ext="pom.asc"/>
+ <artifact name="qpid-broker-plugins-management-http" type="jar" ext="jar"/>
+ <artifact name="qpid-broker-plugins-management-http" type="jar.asc" ext="jar.asc"/>
+ <artifact name="qpid-broker-plugins-management-http" type="source" ext="jar" e:classifier="sources"/>
+ <artifact name="qpid-broker-plugins-management-http" type="source.asc" ext="jar.asc" e:classifier="sources"/>
+ <artifact name="qpid-broker-plugins-management-jmx" type="pom" ext="pom"/>
+ <artifact name="qpid-broker-plugins-management-jmx" type="pom.asc" ext="pom.asc"/>
+ <artifact name="qpid-broker-plugins-management-jmx" type="jar" ext="jar"/>
+ <artifact name="qpid-broker-plugins-management-jmx" type="jar.asc" ext="jar.asc"/>
+ <artifact name="qpid-broker-plugins-management-jmx" type="source" ext="jar" e:classifier="sources"/>
+ <artifact name="qpid-broker-plugins-management-jmx" type="source.asc" ext="jar.asc" e:classifier="sources"/>
+ <artifact name="qpid-amqp-1-0-common" type="pom" ext="pom"/>
+ <artifact name="qpid-amqp-1-0-common" type="pom.asc" ext="pom.asc"/>
+ <artifact name="qpid-amqp-1-0-common" type="jar" ext="jar"/>
+ <artifact name="qpid-amqp-1-0-common" type="jar.asc" ext="jar.asc"/>
+ <artifact name="qpid-amqp-1-0-common" type="source" ext="jar" e:classifier="sources"/>
+ <artifact name="qpid-amqp-1-0-common" type="source.asc" ext="jar.asc" e:classifier="sources"/>
+ <artifact name="qpid-amqp-1-0-client" type="pom" ext="pom"/>
+ <artifact name="qpid-amqp-1-0-client" type="pom.asc" ext="pom.asc"/>
+ <artifact name="qpid-amqp-1-0-client" type="jar" ext="jar"/>
+ <artifact name="qpid-amqp-1-0-client" type="jar.asc" ext="jar.asc"/>
+ <artifact name="qpid-amqp-1-0-client" type="source" ext="jar" e:classifier="sources"/>
+ <artifact name="qpid-amqp-1-0-client" type="source.asc" ext="jar.asc" e:classifier="sources"/>
+ <artifact name="qpid-amqp-1-0-client-jms" type="pom" ext="pom"/>
+ <artifact name="qpid-amqp-1-0-client-jms" type="pom.asc" ext="pom.asc"/>
+ <artifact name="qpid-amqp-1-0-client-jms" type="jar" ext="jar"/>
+ <artifact name="qpid-amqp-1-0-client-jms" type="jar.asc" ext="jar.asc"/>
+ <artifact name="qpid-amqp-1-0-client-jms" type="source" ext="jar" e:classifier="sources"/>
+ <artifact name="qpid-amqp-1-0-client-jms" type="source.asc" ext="jar.asc" e:classifier="sources"/>
<artifact name="qpid-management-common" type="pom" ext="pom"/>
<artifact name="qpid-management-common" type="pom.asc" ext="pom.asc"/>
<artifact name="qpid-management-common" type="jar" ext="jar"/>
@@ -51,6 +87,12 @@
<artifact name="qpid-bdbstore" type="jar.asc" ext="jar.asc"/>
<artifact name="qpid-bdbstore" type="source" ext="jar" e:classifier="sources"/>
<artifact name="qpid-bdbstore" type="source.asc" ext="jar.asc" e:classifier="sources"/>
+ <artifact name="qpid-bdbstore-jmx" type="pom" ext="pom"/>
+ <artifact name="qpid-bdbstore-jmx" type="pom.asc" ext="pom.asc"/>
+ <artifact name="qpid-bdbstore-jmx" type="jar" ext="jar"/>
+ <artifact name="qpid-bdbstore-jmx" type="jar.asc" ext="jar.asc"/>
+ <artifact name="qpid-bdbstore-jmx" type="source" ext="jar" e:classifier="sources"/>
+ <artifact name="qpid-bdbstore-jmx" type="source.asc" ext="jar.asc" e:classifier="sources"/>
</publications>
<dependencies/>
Modified: qpid/branches/asyncstore/java/ivy.retrieve.xml
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/ivy.retrieve.xml?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/ivy.retrieve.xml (original)
+++ qpid/branches/asyncstore/java/ivy.retrieve.xml Thu Feb 28 16:14:30 2013
@@ -57,7 +57,6 @@
<dependency org="log4j" name="log4j" rev="1.2.16" transitive="false"/>
<dependency org="org.apache.maven" name="maven-ant-tasks" rev="2.1.1" transitive="false"/>
<dependency org="org.mockito" name="mockito-all" rev="1.9.0" transitive="false"/>
- <dependency org="org.apache.felix" name="org.apache.felix.main" rev="2.0.5" transitive="false"/>
<dependency org="org.slf4j" name="slf4j-api" rev="1.6.4" transitive="false"/>
<dependency org="org.slf4j" name="slf4j-log4j12" rev="1.6.4" transitive="false"/>
<dependency org="org.eclipse.jetty" name="jetty-server" rev="7.6.3.v20120416" transitive="false"/>
@@ -70,6 +69,8 @@
<dependency org="org.eclipse.jetty" name="jetty-util" rev="7.6.3.v20120416" transitive="false"/>
<dependency org="org.dojotoolkit" name="dojo-war" rev="1.7.2" transitive="false"/>
<dependency org="xalan" name="xalan" rev="2.7.0" transitive="false"/>
+ <dependency org="velocity" name="velocity" rev="1.4" transitive="false"/>
+ <dependency org="velocity" name="velocity-dep" rev="1.4" transitive="false"/>
<!-- The following are optional dependencies, for modules providing optional functionlity or
for use in optional build/test steps. Their optional status is usually indicative of licences
Modified: qpid/branches/asyncstore/java/ivysettings.retrieve.xml
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/ivysettings.retrieve.xml?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/ivysettings.retrieve.xml (original)
+++ qpid/branches/asyncstore/java/ivysettings.retrieve.xml Thu Feb 28 16:14:30 2013
@@ -16,7 +16,7 @@
-->
<ivysettings>
<property name="ivy.default.resolver" value="chain" override="false"/>
- <property name="ivy.localfs.root" value="${project.root}/localfs_repo" override="false"/>
+ <property name="ivy.localfs.root" value="${project.root}/lib/localfs_repo" override="false"/>
<property name="ivy.localfs.pattern" value="[artifact]-[revision](-[classifier]).[ext]" override="false"/>
<settings defaultResolver="${ivy.default.resolver}"/>
Modified: qpid/branches/asyncstore/java/jca/README-JBOSS.txt
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/jca/README-JBOSS.txt?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/jca/README-JBOSS.txt (original)
+++ qpid/branches/asyncstore/java/jca/README-JBOSS.txt Thu Feb 28 16:14:30 2013
@@ -61,7 +61,7 @@ XA ConnectionFactory
<xa-transaction/>
<rar-name>qpid-ra-<ra-version>.rar</rar-name>
<connection-definition>org.apache.qpid.ra.QpidRAConnectionFactory</connection-definition>
- <config-property name="connectionURL">amqp://guest:guest@/test?brokerlist='tcp://localhost:5672?sasl_mechs='ANONYMOUS''</config-property>
+ <config-property name="ConnectionURL">amqp://guest:guest@/test?brokerlist='tcp://localhost:5672?sasl_mechs='PLAIN''</config-property>
<max-pool-size>20</max-pool-size>
</tx-connection-factory>
@@ -79,11 +79,11 @@ Local ConnectionFactory
=======================
<tx-connection-factory>
<jndi-name>QpidJMS</jndi-name>
- <rar-name>qpid-ra-0.10.rar</rar-name>
+ <rar-name>qpid-ra-<ra-version>.rar</rar-name>
<local-transaction/>
- <config-property name="useLocalTx" type="java.lang.Boolean">true</config-property>
- <config-property name="connectionURL">amqp://anonymous:@client/test?brokerlist='tcp://localhost:5672?sasl_mechs='ANONYMOUS''</config-property>
- <connection-definition>org.apache.qpid.ra.QpidRAConnectionFactory</connection-definition>
+ <config-property name="UseLocalTx" type="java.lang.Boolean">true</config-property>
+ <config-property name="ConnectionURL">amqp://anonymous:@client/test?brokerlist='tcp://localhost:5672?sasl_mechs='PLAIN''
+ </config-property> <connection-definition>org.apache.qpid.ra.QpidRAConnectionFactory</connection-definition>
<max-pool-size>20</max-pool-size>
</tx-connection-factory>
@@ -100,11 +100,10 @@ provides two such objects
<mbean code="org.jboss.resource.deployment.AdminObject"
name="qpid.jca:name=HelloQueue">
<attribute name="JNDIName">Hello</attribute>
- <depends optional-attribute-name="RARName">jboss.jca:service=RARDeployment,name='qpid-ra-0.10.rar'</depends>
+ <depends optional-attribute-name="RARName">jboss.jca:service=RARDeployment,name='qpid-ra-<ra-version>.rar'</depends>
<attribute name="Type">javax.jms.Destination</attribute>
<attribute name="Properties">
- destinationType=QUEUE
- destinationAddress=amq.direct
+ DestinationAddress=amq.direct
</attribute>
</mbean>
@@ -113,16 +112,15 @@ The above XML defines a JMS Queue which
queue/HelloQueue
This destination can be retrieved from JNDI and be used for the consumption or production of messages. The desinationAddress property
-can be customized for your environment. Please see the Qpid Java Client documentation for specific configuration options.
+can be customized for your environment. Please see the Qpid Java Client documentation for specific configuration options.
<mbean code="org.jboss.resource.deployment.AdminObject"
name="qpid.jca:name=HelloTopic">
<attribute name="JNDIName">HelloTopic</attribute>
- <depends optional-attribute-name="RARName">jboss.jca:service=RARDeployment,name='qpid-ra-0.10.rar'</depends>
+ <depends optional-attribute-name="RARName">jboss.jca:service=RARDeployment,name='qpid-ra-<ra-version>.rar'</depends>
<attribute name="Type">javax.jms.Destination</attribute>
<attribute name="Properties">
- destinationType=TOPIC
- destinationAddress=amq.topic
+ DestinationAddress=amq.topic
</attribute>
</mbean>
@@ -138,10 +136,10 @@ can be customized for your environment.
<mbean code="org.jboss.resource.deployment.AdminObject"
name="qpid.jca:name=QpidConnectionFactory">
<attribute name="JNDIName">QpidConnectionFactory</attribute>
- <depends optional-attribute-name="RARName">jboss.jca:service=RARDeployment,name='qpid-ra-0.10.rar'</depends>
+ <depends optional-attribute-name="RARName">jboss.jca:service=RARDeployment,name='qpid-ra-<ra-version>.rar'</depends>
<attribute name="Type">javax.jms.ConnectionFactory</attribute>
<attribute name="Properties">
- connectionURL=amqp://anonymous:@client/test?brokerlist='tcp://localhost:5672?sasl_mechs='ANONYMOUS''
+ ConnectionURL=amqp://anonymous:@client/test?brokerlist='tcp://localhost:5672?sasl_mechs='PLAIN''
</attribute>
</mbean>
Modified: qpid/branches/asyncstore/java/jca/build.xml
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/jca/build.xml?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/jca/build.xml (original)
+++ qpid/branches/asyncstore/java/jca/build.xml Thu Feb 28 16:14:30 2013
@@ -22,18 +22,26 @@
<property name="module.depends" value="common client"/>
<property name="module.name" value="jca"/>
+ <!-- Hack to make the renamed module jars available on the module test classpath -->
+ <property name="module.test.depends" value="ra ra/tests"/>
+
+ <!-- Import common.xml to make the properties it defines available before importing module.xml -->
+ <import file="../common.xml"/>
+
+ <!-- Override the standard output jar names before importing module.xml, to produce
+ artifacts that use ra in the name instead of jca like the module should -->
+ <property name="module.test.jar" value="${build.lib}/${project.name}-ra-tests-${project.version}.jar"/>
+ <property name="module.jar" value="${build.lib}/${project.name}-ra-${project.version}.jar"/>
+ <property name="module.source.jar" value="${build.lib}/${project.name}-ra-${project.version}-sources.jar"/>
<import file="../module.xml"/>
<property name="module.rar" value="${build.lib}/${project.name}-ra-${project.version}.rar"/>
+ <property name="rar.resources" value="rar/src/main/resources"/>
- <property name="module.resources" value="src/main/resources"/>
-
- <target name="rar" depends="jar">
- <!--Note we need to do this as we need to keep the ra in the name of the artificats but we can't override the module.jar property which is based on the directory name-->
- <move file="${build.lib}/${project.name}-${module.name}-${project.version}.jar" tofile="${build.lib}/${project.name}-ra-${project.version}.jar"/>
+ <target name="rar" depends="jar" description="creates a rar file containing the module jar, client jars, etc">
<jar destfile="${module.rar}">
- <fileset dir="${module.resources}">
+ <fileset dir="${rar.resources}">
<include name="**/*.xml"/>
</fileset>
<fileset dir="${build.lib}">
@@ -67,6 +75,9 @@
<target name="examples" depends="example-properties-file, example-jars"/>
- <target name="build" depends="rar, examples"/>
+ <target name="postbuild" depends="rar, examples"/>
+ <!-- Override module.xml 'libs' target to avoid copying the jar files dependencies
+ into the 'build/lib' dir, since they will be supplied by the app server -->
+ <target name="libs"/>
</project>
Modified: qpid/branches/asyncstore/java/jca/src/main/java/org/apache/qpid/ra/admin/QpidConnectionFactoryProxy.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/jca/src/main/java/org/apache/qpid/ra/admin/QpidConnectionFactoryProxy.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/jca/src/main/java/org/apache/qpid/ra/admin/QpidConnectionFactoryProxy.java (original)
+++ qpid/branches/asyncstore/java/jca/src/main/java/org/apache/qpid/ra/admin/QpidConnectionFactoryProxy.java Thu Feb 28 16:14:30 2013
@@ -27,13 +27,17 @@ import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.io.Serializable;
-import javax.jms.ConnectionFactory;
import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.naming.NamingException;
+import javax.jms.QueueConnection;
+import javax.jms.QueueConnectionFactory;
import javax.naming.Reference;
import javax.naming.Referenceable;
import javax.naming.spi.ObjectFactory;
+import javax.jms.TopicConnection;
+import javax.jms.TopicConnectionFactory;
import org.apache.qpid.client.AMQConnectionFactory;
@@ -44,7 +48,7 @@ import org.slf4j.LoggerFactory;
*
*
*/
-public class QpidConnectionFactoryProxy implements Externalizable, Referenceable, ConnectionFactory, Serializable
+public class QpidConnectionFactoryProxy implements QueueConnectionFactory, TopicConnectionFactory, Externalizable, Referenceable, Serializable
{
private static final Logger _log = LoggerFactory.getLogger(QpidDestinationProxy.class);
@@ -100,13 +104,6 @@ public class QpidConnectionFactoryProxy
try
{
_delegate = new AMQConnectionFactory(getConnectionURL());
- /*
- QpidResourceAdapter ra = new QpidResourceAdapter();
- QpidRAManagedConnectionFactory mcf = new QpidRAManagedConnectionFactory();
- mcf.setResourceAdapter(ra);
- mcf.setConnectionURL(getConnectionURL());
- delegate = new QpidRAConnectionFactoryImpl(mcf, null);
- */
return ((Referenceable) _delegate).getReference();
}
catch(Exception e)
@@ -162,7 +159,63 @@ public class QpidConnectionFactoryProxy
*/
public Connection createConnection(final String userName, final String password) throws JMSException
{
- return _delegate.createConnection(userName, password);
+ try
+ {
+ if(_delegate == null)
+ {
+ getReference();
+ }
+
+ return _delegate.createConnection(userName, password);
+ }
+ catch(Exception e)
+ {
+ throw new JMSException(e.getMessage());
+ }
+ }
+
+ /**
+ * Create a queue connection
+ * @return The queue connection
+ * @exception JMSException Thrown if the operation fails
+ */
+ public QueueConnection createQueueConnection() throws JMSException
+ {
+ return (QueueConnection)createConnection();
+ }
+
+ /**
+ * Create a queue connection
+ * @param userName The user name
+ * @param password The password
+ * @return The connection
+ * @exception JMSException Thrown if the operation fails
+ */
+ public QueueConnection createQueueConnection(final String userName, final String password) throws JMSException
+ {
+ return (QueueConnection)createConnection(userName, password);
+ }
+
+ /**
+ * Create a topic connection
+ * @return The topic connection
+ * @exception JMSException Thrown if the operation fails
+ */
+ public TopicConnection createTopicConnection() throws JMSException
+ {
+ return (TopicConnection)createConnection();
+ }
+
+ /**
+ * Create a topic connection
+ * @param userName The user name
+ * @param password The password
+ * @return The topic connection
+ * @exception JMSException Thrown if the operation fails
+ */
+ public TopicConnection createTopicConnection(final String userName, final String password) throws JMSException
+ {
+ return (TopicConnection)createConnection(userName, password);
}
}
Modified: qpid/branches/asyncstore/java/management/common/src/main/java/management-common.bnd
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/management/common/src/main/java/management-common.bnd?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/management/common/src/main/java/management-common.bnd (original)
+++ qpid/branches/asyncstore/java/management/common/src/main/java/management-common.bnd Thu Feb 28 16:14:30 2013
@@ -17,7 +17,7 @@
# under the License.
#
-ver: 0.19.0
+ver: 0.21.0
Bundle-SymbolicName: qpid-management-common
Bundle-Version: ${ver}
Modified: qpid/branches/asyncstore/java/management/example/src/main/java/org/apache/qpid/example/jmxexample/AddQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/management/example/src/main/java/org/apache/qpid/example/jmxexample/AddQueue.java?rev=1451244&r1=1451243&r2=1451244&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/management/example/src/main/java/org/apache/qpid/example/jmxexample/AddQueue.java (original)
+++ qpid/branches/asyncstore/java/management/example/src/main/java/org/apache/qpid/example/jmxexample/AddQueue.java Thu Feb 28 16:14:30 2013
@@ -36,15 +36,14 @@ import org.apache.qpid.management.common
public class AddQueue
{
-
public static void main(String[] args)
{
//Example: add 'newqueue' to the 'test' virtualhost and bind to the 'amq.direct' exchange
//TODO: take these parameters as arguments
-
+
addQueue("test", "amq.direct", "newqueue");
}
-
+
private static JMXConnector getJMXConnection() throws Exception
{
//TODO: Take these parameters as main+method arguments
@@ -52,52 +51,55 @@ public class AddQueue
int port = 8999;
String username = "admin";
String password = "admin";
-
+
Map<String, Object> env = new HashMap<String, Object>();
JMXServiceURL jmxUrl = new JMXServiceURL("service:jmx:rmi:///jndi/rmi://" + host + ":" + port + "/jmxrmi");
//Add user credential's to environment map for RMIConnector startup.
env.put(JMXConnector.CREDENTIALS, new String[] {username,password});
-
+
return JMXConnectorFactory.connect(jmxUrl, env);
}
-
- public static boolean addQueue(String virHost, String exchName, String queueName) {
+ public static boolean addQueue(String virHost, String exchName, String queueName)
+ {
JMXConnector jmxc = null;
try
{
jmxc = getJMXConnection();
-
+
MBeanServerConnection mbsc = jmxc.getMBeanServerConnection();
ObjectName hostManagerObjectName = new ObjectName(
"org.apache.qpid:" +
"type=VirtualHost.VirtualHostManager," +
- "VirtualHost=" + virHost + ",*");
+ "VirtualHost=" + ObjectName.quote(virHost) + ",*");
Set<ObjectName> vhostManagers = mbsc.queryNames(hostManagerObjectName, null);
-
+
if(vhostManagers.size() == 0)
{
+ System.out.println("VirtualHostManager MBean wasnt found: " + virHost);
+
//The vhostManager MBean wasnt found, cant procede
return false;
}
-
+
ManagedBroker vhostManager = (ManagedBroker) MBeanServerInvocationHandler.newProxyInstance(
mbsc, (ObjectName) vhostManagers.toArray()[0], ManagedBroker.class, false);
ObjectName customExchangeObjectName = new ObjectName(
"org.apache.qpid:" +
"type=VirtualHost.Exchange," +
- "VirtualHost=" + virHost + "," +
- "name=" + exchName + "," +
- "ExchangeType=direct,*");
+ "VirtualHost=" + ObjectName.quote(virHost) + "," +
+ "name=" + ObjectName.quote(exchName) + ",*");
Set<ObjectName> exchanges = mbsc.queryNames(customExchangeObjectName, null);
-
+
if(exchanges.size() == 0)
{
+ System.out.println("Exchange wasnt found: " + exchName);
+
//The exchange doesnt exist, cant procede.
return false;
}
@@ -105,12 +107,14 @@ public class AddQueue
//create the MBean proxy
ManagedExchange managedExchange = (ManagedExchange) MBeanServerInvocationHandler.newProxyInstance(
mbsc, (ObjectName) exchanges.toArray()[0], ManagedExchange.class, false);
-
+
try
{
//create the new durable queue and bind it.
vhostManager.createNewQueue(queueName, null, true);
+ System.out.println("Created queue: " + queueName);
managedExchange.createNewBinding(queueName,queueName);
+ System.out.println("Bound queue to exchange: "+ exchName);
}
catch (Exception e)
{
@@ -126,7 +130,7 @@ public class AddQueue
{
System.out.println("Could not add queue due to error :" + e.getMessage());
e.printStackTrace();
- }
+ }
finally
{
if(jmxc != null)
@@ -141,9 +145,8 @@ public class AddQueue
}
}
}
-
+
return false;
-
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org