You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ra...@apache.org on 2007/03/27 18:31:15 UTC
svn commit: r522988 [1/3] - in
/incubator/qpid/branches/qpid.0-9/java/newclient: ./ src/ src/main/
src/main/java/ src/main/java/org/ src/main/java/org/apache/
src/main/java/org/apache/qpid/ src/main/java/org/apache/qpid/nclient/
src/main/java/org/apach...
Author: rajith
Date: Tue Mar 27 09:31:05 2007
New Revision: 522988
URL: http://svn.apache.org/viewvc?view=rev&rev=522988
Log:
This is the initial checkup for the new client
Added:
incubator/qpid/branches/qpid.0-9/java/newclient/
incubator/qpid/branches/qpid.0-9/java/newclient/.project
incubator/qpid/branches/qpid.0-9/java/newclient/pom.xml
incubator/qpid/branches/qpid.0-9/java/newclient/src/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/client.log4j
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPCallBack.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPCallBackSupport.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPChannel.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPConnection.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPExchange.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPMessage.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPMessageCallBack.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPQueue.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/EventManager.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/sample/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/sample/SecurityHelper.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/sample/TestClient.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPState.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateListener.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateMachine.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateManager.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateType.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/IllegalStateTransitionException.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/config/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/config/ClientConfiguration.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/config/client.xml
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/core/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/core/AMQPException.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/core/AbstractPhase.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/core/DefaultPhaseContext.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/core/Phase.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/core/PhaseContext.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/core/PhaseFactory.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/core/QpidConstants.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/core/TransientPhaseContext.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/execution/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/execution/ExecutionPhase.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/execution/RequestManager.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/execution/ResponseManager.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/message/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/message/AMQPApplicationMessage.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/message/MessagePhase.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/message/MessageStore.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/message/TransientMessageStore.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/model/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/model/AMQPMethodEvent.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/model/AMQPMethodListener.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/model/ModelPhase.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/model/handler/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/security/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/security/AMQPCallbackHandler.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/security/CallbackHandlerRegistry.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/security/DynamicSaslRegistrar.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/security/JCAProvider.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/security/UsernamePasswordCallbackHandler.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/transport/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/transport/AMQPBrokerDetails.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/transport/AMQPConnectionURL.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/transport/BrokerDetails.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/transport/ConnectionURL.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/transport/TCPConnection.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/transport/TransportConnection.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/transport/TransportConnectionFactory.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/transport/TransportPhase.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/transport/VMConnection.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/util/
incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/util/AMQPValidator.java
incubator/qpid/branches/qpid.0-9/java/newclient/src/test/
Added: incubator/qpid/branches/qpid.0-9/java/newclient/.project
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/.project?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/.project (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/.project Tue Mar 27 09:31:05 2007
@@ -0,0 +1,17 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<projectDescription>
+ <name>qpid-newclient</name>
+ <comment></comment>
+ <projects>
+ </projects>
+ <buildSpec>
+ <buildCommand>
+ <name>org.eclipse.jdt.core.javabuilder</name>
+ <arguments>
+ </arguments>
+ </buildCommand>
+ </buildSpec>
+ <natures>
+ <nature>org.eclipse.jdt.core.javanature</nature>
+ </natures>
+</projectDescription>
Added: incubator/qpid/branches/qpid.0-9/java/newclient/pom.xml
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/pom.xml?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/pom.xml (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/pom.xml Tue Mar 27 09:31:05 2007
@@ -0,0 +1,188 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <groupId>org.apache.qpid</groupId>
+ <artifactId>qpid-newclient</artifactId>
+ <packaging>jar</packaging>
+ <version>1.0-incubating-M2-SNAPSHOT</version>
+ <name>Qpid New Client</name>
+ <url>http://cwiki.apache.org/confluence/display/qpid</url>
+
+ <parent>
+ <groupId>org.apache.qpid</groupId>
+ <artifactId>qpid</artifactId>
+ <version>1.0-incubating-M2-SNAPSHOT</version>
+ </parent>
+
+ <properties>
+ <topDirectoryLocation>..</topDirectoryLocation>
+ <java.source.version>1.5</java.source.version>
+ <qpid.version>${pom.version}</qpid.version>
+ <qpid.targetDir>${project.build.directory}</qpid.targetDir>
+ </properties>
+
+ <dependencies>
+
+ <dependency>
+ <groupId>org.apache.qpid</groupId>
+ <artifactId>qpid-common</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.geronimo.specs</groupId>
+ <artifactId>geronimo-jms_1.1_spec</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>commons-collections</groupId>
+ <artifactId>commons-collections</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>commons-lang</groupId>
+ <artifactId>commons-lang</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.mina</groupId>
+ <artifactId>mina-filter-ssl</artifactId>
+ </dependency>
+
+ <!-- Test Dependencies -->
+ <dependency> <!-- for inVm Broker -->
+ <groupId>org.apache.qpid</groupId>
+ <artifactId>qpid-broker</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>jmscts</groupId>
+ <artifactId>jmscts</artifactId>
+ <version>0.5-b2</version>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>jms</groupId>
+ <artifactId>jms</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.easymock</groupId>
+ <artifactId>easymockclassextension</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-antrun-plugin</artifactId>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <systemProperties>
+ <property>
+ <name>amqj.noAutoCreateVMBroker</name>
+ <value>true</value>
+ </property>
+ <property>
+ <name>amqj.logging.level</name>
+ <value>${amqj.logging.level}</value>
+ </property>
+ <property>
+ <name>log4j.configuration</name>
+ <value>file:///${basedir}/src/main/java/client.log4j</value>
+ </property>
+ </systemProperties>
+ </configuration>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
+ </plugins>
+
+<!-- The inclusion of this resource causes the build to hang. -->
+ <!--resources>
+ <resource>
+ <targetPath>META-INF/</targetPath>
+ <filtering>false</filtering>
+ <directory>../resources/META-INF</directory>
+ <includes>
+ <include>**</include>
+ </includes>
+ </resource>
+ </resources-->
+
+ <testResources>
+ <testResource>
+ <targetPath>META-INF/</targetPath>
+ <filtering>false</filtering>
+ <directory>../resources/META-INF</directory>
+ <includes>
+ <include>**</include>
+ </includes>
+ </testResource>
+ <testResource>
+ <targetPath>src/</targetPath>
+ <filtering>false</filtering>
+ <directory>src/test/java</directory>
+ <includes>
+ <include>**/*.java</include>
+ </includes>
+ </testResource>
+
+ <testResource>
+ <targetPath></targetPath>
+ <filtering>false</filtering>
+ <directory>src/main/java</directory>
+ <includes>
+ <include>client.log4j</include>
+ </includes>
+ </testResource>
+ </testResources>
+
+ </build>
+
+</project>
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/client.log4j
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/client.log4j?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/client.log4j (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/client.log4j Tue Mar 27 09:31:05 2007
@@ -0,0 +1,28 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+log4j.rootLogger=${root.logging.level}
+
+
+log4j.logger.org.apache.qpid=${amqj.logging.level}, console
+log4j.additivity.org.apache.qpid=false
+
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.Threshold=all
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+log4j.appender.console.layout.ConversionPattern=%t %d %p [%c{4}] %m%n
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPCallBack.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPCallBack.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPCallBack.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPCallBack.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,23 @@
+package org.apache.qpid.nclient.amqp;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQMethodBody;
+
+public abstract class AMQPCallBack
+{
+ private boolean _isComplete = false;
+
+ public abstract void brokerResponded(AMQMethodBody body);
+
+ public abstract void brokerRespondedWithError(AMQException e);
+
+ public void setIsComplete(boolean isComplete)
+ {
+ _isComplete = isComplete;
+ }
+
+ public boolean isComplete()
+ {
+ return _isComplete;
+ }
+}
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPCallBackSupport.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPCallBackSupport.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPCallBackSupport.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPCallBackSupport.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,64 @@
+package org.apache.qpid.nclient.amqp;
+
+import java.security.SecureRandom;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.nclient.core.QpidConstants;
+import org.apache.qpid.nclient.model.AMQPMethodEvent;
+
+public abstract class AMQPCallBackSupport
+{
+ private SecureRandom _localCorrelationIdGenerator = new SecureRandom();
+ protected ConcurrentHashMap<Long,AMQPCallBack> _cbMap = new ConcurrentHashMap<Long,AMQPCallBack>();
+
+ //the channelId assigned for this instance
+ protected int _channelId;
+
+ public AMQPCallBackSupport(int channelId)
+ {
+ _channelId = channelId;
+ }
+
+ private long getNextCorrelationId()
+ {
+ return _localCorrelationIdGenerator.nextLong();
+ }
+
+
+ // For methods that still use nowait, hopefully they will remove nowait
+ protected AMQPMethodEvent handleNoWait(boolean noWait,AMQMethodBody methodBody,AMQPCallBack cb)
+ {
+ if(noWait)
+ {
+ // u only need to register if u are expecting a response
+ long localCorrelationId = getNextCorrelationId();
+ AMQPMethodEvent msg = new AMQPMethodEvent(_channelId,methodBody,QpidConstants.EMPTY_CORRELATION_ID,localCorrelationId);
+ _cbMap.put(localCorrelationId, cb);
+ return msg;
+ }
+ else
+ {
+ AMQPMethodEvent msg = new AMQPMethodEvent(_channelId,methodBody,QpidConstants.EMPTY_CORRELATION_ID);
+ return msg;
+ }
+ }
+
+ protected AMQPMethodEvent handleAsynchronousCall(AMQMethodBody methodBody,AMQPCallBack cb)
+ {
+ long localCorrelationId = getNextCorrelationId();
+ AMQPMethodEvent msg = new AMQPMethodEvent(_channelId,methodBody,QpidConstants.EMPTY_CORRELATION_ID,localCorrelationId);
+ _cbMap.put(localCorrelationId, cb);
+ return msg;
+ }
+
+ protected void invokeCallBack(long localCorrelationId, AMQMethodBody methodBody)
+ {
+ if(_cbMap.contains(localCorrelationId))
+ {
+ AMQPCallBack cb = (AMQPCallBack)_cbMap.get(localCorrelationId);
+ cb.brokerResponded(methodBody);
+ }
+ }
+
+}
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPChannel.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPChannel.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPChannel.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPChannel.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,284 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.nclient.amqp;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.framing.ChannelCloseBody;
+import org.apache.qpid.framing.ChannelCloseOkBody;
+import org.apache.qpid.framing.ChannelFlowBody;
+import org.apache.qpid.framing.ChannelFlowOkBody;
+import org.apache.qpid.framing.ChannelOkBody;
+import org.apache.qpid.framing.ChannelOpenBody;
+import org.apache.qpid.framing.ChannelOpenOkBody;
+import org.apache.qpid.framing.ChannelResumeBody;
+import org.apache.qpid.nclient.amqp.state.AMQPState;
+import org.apache.qpid.nclient.amqp.state.AMQPStateMachine;
+import org.apache.qpid.nclient.config.ClientConfiguration;
+import org.apache.qpid.nclient.core.AMQPException;
+import org.apache.qpid.nclient.core.Phase;
+import org.apache.qpid.nclient.core.QpidConstants;
+import org.apache.qpid.nclient.model.AMQPMethodEvent;
+import org.apache.qpid.nclient.model.AMQPMethodListener;
+import org.apache.qpid.nclient.util.AMQPValidator;
+
+/**
+ * This represents the Channel class defined in the AMQP protocol.
+ * This class is a finite state machine and is thread safe by design.
+ * Only valid state changes are allowed or else an IllegalStateTransitionException will be thrown.
+ * Only one thread can enter the methods that change state, at a given time.
+ * The AMQP protocol recommends one thread per channel by design.
+ *
+ * A JMS Session can wrap an instance of this class.
+ */
+
+public class AMQPChannel extends AMQPStateMachine implements AMQPMethodListener
+{
+private static final Logger _logger = Logger.getLogger(AMQPChannel.class);
+
+ //the channelId assigned for this channel
+ private int _channelId;
+ private Phase _phase;
+ private AMQPState _currentState;
+ private final AMQPState[] _validCloseStates = new AMQPState[]{AMQPState.CHANNEL_OPENED,AMQPState.CHANNEL_SUSPEND};
+ private final AMQPState[] _validResumeStates = new AMQPState[]{AMQPState.CHANNEL_CLOSED,AMQPState.CHANNEL_NOT_OPENED};
+
+ // The wait period until a server sends a respond
+ private long _serverTimeOut = 1000;
+ private final Lock _lock = new ReentrantLock();
+ private final Condition _channelNotOpend = _lock.newCondition();
+ private final Condition _channelNotClosed = _lock.newCondition();
+ private final Condition _channelFlowNotResponded = _lock.newCondition();
+ private final Condition _channelNotResumed = _lock.newCondition();
+
+ private ChannelOpenOkBody _channelOpenOkBody;
+ private ChannelCloseOkBody _channelCloseOkBody;
+ private ChannelFlowOkBody _channelFlowOkBody;
+ private ChannelOkBody _channelOkBody;
+
+ public AMQPChannel(int channelId)
+ {
+ _channelId = channelId;
+ _currentState = AMQPState.CHANNEL_NOT_OPENED;
+ _serverTimeOut = ClientConfiguration.get().getLong(QpidConstants.SERVER_TIMEOUT_IN_MILLISECONDS);
+ }
+
+ /**-------------------------------------------
+ * API Methods
+ *--------------------------------------------
+ */
+
+ /**
+ * Opens the channel
+ */
+ public ChannelOpenOkBody open(ChannelOpenBody channelOpenBody) throws AMQPException
+ {
+ _lock.lock();
+ try {
+ _channelOpenOkBody = null;
+ checkIfValidStateTransition(AMQPState.CHANNEL_NOT_OPENED,_currentState,AMQPState.CHANNEL_OPENED);
+ AMQPMethodEvent msg = new AMQPMethodEvent(_channelId,channelOpenBody,QpidConstants.EMPTY_CORRELATION_ID);
+ _phase.messageSent(msg);
+ _channelNotOpend.await(_serverTimeOut, TimeUnit.MILLISECONDS);
+ AMQPValidator.throwExceptionOnNull(_channelOpenOkBody, "The broker didn't send the ChannelOpenOkBody in time");
+ _currentState = AMQPState.CHANNEL_OPENED;
+ return _channelOpenOkBody;
+ }
+ catch(Exception e)
+ {
+ throw new AMQPException("XXX");
+ }
+ finally
+ {
+ _lock.unlock();
+ }
+ }
+
+ /**
+ * Close the channel
+ */
+ public ChannelCloseOkBody close(ChannelCloseBody channelCloseBody) throws AMQPException
+ {
+ _lock.lock();
+ try {
+ _channelCloseOkBody = null;
+ checkIfValidStateTransition(_validCloseStates,_currentState,AMQPState.CHANNEL_CLOSED);
+ AMQPMethodEvent msg = new AMQPMethodEvent(_channelId,channelCloseBody,QpidConstants.EMPTY_CORRELATION_ID);
+ _phase.messageSent(msg);
+ _channelNotClosed.await(_serverTimeOut, TimeUnit.MILLISECONDS);
+ AMQPValidator.throwExceptionOnNull(_channelCloseOkBody, "The broker didn't send the ChannelCloseOkBody in time");
+ _currentState = AMQPState.CHANNEL_CLOSED;
+ return _channelCloseOkBody;
+ }
+ catch(Exception e)
+ {
+ throw new AMQPException("XXX");
+ }
+ finally
+ {
+ _lock.unlock();
+ }
+ }
+
+ /**
+ * Channel Flow
+ */
+ public ChannelFlowOkBody close(ChannelFlowBody channelFlowBody) throws AMQPException
+ {
+ _lock.lock();
+ try {
+ _channelFlowOkBody = null;
+ if(channelFlowBody.active)
+ {
+ checkIfValidStateTransition(AMQPState.CHANNEL_SUSPEND,_currentState,AMQPState.CHANNEL_OPENED);
+ }
+ else
+ {
+ checkIfValidStateTransition(AMQPState.CHANNEL_OPENED,_currentState,AMQPState.CHANNEL_SUSPEND);
+ }
+ AMQPMethodEvent msg = new AMQPMethodEvent(_channelId,channelFlowBody,QpidConstants.EMPTY_CORRELATION_ID);
+ _phase.messageSent(msg);
+ _channelFlowNotResponded.await(_serverTimeOut, TimeUnit.MILLISECONDS);
+ AMQPValidator.throwExceptionOnNull(_channelFlowOkBody, "The broker didn't send the ChannelFlowOkBody in time");
+ handleChannelFlowState(_channelFlowOkBody.active);
+ return _channelFlowOkBody;
+ }
+ catch(Exception e)
+ {
+ throw new AMQPException("XXX");
+ }
+ finally
+ {
+ _lock.unlock();
+ }
+ }
+
+ /**
+ * Close the channel
+ */
+ public ChannelOkBody resume(ChannelResumeBody channelResumeBody) throws AMQPException
+ {
+ _lock.lock();
+ try {
+ _channelOkBody = null;
+ checkIfValidStateTransition(_validResumeStates,_currentState,AMQPState.CHANNEL_OPENED);
+ AMQPMethodEvent msg = new AMQPMethodEvent(_channelId,channelResumeBody,QpidConstants.EMPTY_CORRELATION_ID);
+ _phase.messageSent(msg);
+ _channelNotResumed.await(_serverTimeOut, TimeUnit.MILLISECONDS);
+ AMQPValidator.throwExceptionOnNull(_channelOkBody, "The broker didn't send the ChannelOkBody in response to the ChannelResumeBody in time");
+ _currentState = AMQPState.CHANNEL_OPENED;
+ return _channelOkBody;
+ }
+ catch(Exception e)
+ {
+ throw new AMQPException("XXX");
+ }
+ finally
+ {
+ _lock.unlock();
+ }
+ }
+
+ /**-------------------------------------------
+ * AMQPMethodListener methods
+ *--------------------------------------------
+ */
+ public <B extends AMQMethodBody> boolean methodReceived(AMQPMethodEvent<B> evt) throws AMQPException
+ {
+ if (evt.getMethod() instanceof ChannelOpenOkBody)
+ {
+ _channelOpenOkBody = (ChannelOpenOkBody)evt.getMethod();
+ _channelNotOpend.signal();
+ return true;
+ }
+ else if (evt.getMethod() instanceof ChannelCloseOkBody)
+ {
+ _channelCloseOkBody = (ChannelCloseOkBody)evt.getMethod();
+ _channelNotClosed.signal();
+ return true;
+ }
+ else if (evt.getMethod() instanceof ChannelCloseBody)
+ {
+ handleChannelClose((ChannelCloseBody)evt.getMethod());
+ return true;
+ }
+ else if (evt.getMethod() instanceof ChannelFlowOkBody)
+ {
+ _channelFlowOkBody = (ChannelFlowOkBody)evt.getMethod();
+ _channelFlowNotResponded.signal();
+ return true;
+ }
+ else if (evt.getMethod() instanceof ChannelFlowBody)
+ {
+ handleChannelFlow((ChannelFlowBody)evt.getMethod());
+ return true;
+ }
+ else if (evt.getMethod() instanceof ChannelOkBody)
+ {
+ _channelOkBody = (ChannelOkBody)evt.getMethod();
+ //In this case the only method expecting channel-ok is channel-resume
+ // haven't implemented ping and pong.
+ _channelNotResumed.signal();
+ return true;
+ }
+ else
+ {
+ return false;
+ }
+ }
+
+ private void handleChannelClose(ChannelCloseBody channelCloseBody)
+ {
+ try
+ {
+ _lock.lock();
+ _currentState = AMQPState.CHANNEL_CLOSED;
+ }
+ finally
+ {
+ _lock.unlock();
+ }
+ }
+
+ private void handleChannelFlow(ChannelFlowBody channelFlowBody)
+ {
+ _lock.lock();
+ try
+ {
+ handleChannelFlowState(channelFlowBody.active);
+ }
+ finally
+ {
+ _lock.unlock();
+ }
+ }
+
+ private void handleChannelFlowState(boolean flow)
+ {
+ _currentState = (flow) ? AMQPState.CHANNEL_OPENED : AMQPState.CHANNEL_SUSPEND;
+ }
+}
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPConnection.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPConnection.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPConnection.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPConnection.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,348 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.nclient.amqp;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.framing.ConnectionCloseBody;
+import org.apache.qpid.framing.ConnectionCloseOkBody;
+import org.apache.qpid.framing.ConnectionOpenBody;
+import org.apache.qpid.framing.ConnectionOpenOkBody;
+import org.apache.qpid.framing.ConnectionSecureBody;
+import org.apache.qpid.framing.ConnectionSecureOkBody;
+import org.apache.qpid.framing.ConnectionStartBody;
+import org.apache.qpid.framing.ConnectionStartOkBody;
+import org.apache.qpid.framing.ConnectionTuneBody;
+import org.apache.qpid.framing.ConnectionTuneOkBody;
+import org.apache.qpid.nclient.amqp.state.AMQPState;
+import org.apache.qpid.nclient.amqp.state.AMQPStateMachine;
+import org.apache.qpid.nclient.config.ClientConfiguration;
+import org.apache.qpid.nclient.core.AMQPException;
+import org.apache.qpid.nclient.core.Phase;
+import org.apache.qpid.nclient.core.QpidConstants;
+import org.apache.qpid.nclient.model.AMQPMethodEvent;
+import org.apache.qpid.nclient.model.AMQPMethodListener;
+import org.apache.qpid.nclient.transport.TransportConnection;
+import org.apache.qpid.nclient.util.AMQPValidator;
+
+/**
+ * This maps directly to the Connection class defined in the AMQP protocol This class is a finite state machine and is
+ * thread safe by design A particular method (state changing) can only be invoked once and only in sequence or else an
+ * IllegalStateTransitionException will be thrown Also only one thread can enter those methods at a given time.
+ */
+public class AMQPConnection extends AMQPStateMachine implements AMQPMethodListener
+{
+ private static final Logger _logger = Logger.getLogger(AMQPConnection.class);
+
+ private Phase _phase;
+
+ private TransportConnection _connection;
+
+ private long _correlationId;
+
+ private AMQPState _currentState;
+
+ private final AMQPState[] _validCloseStates = new AMQPState[] { AMQPState.CONNECTION_NOT_STARTED,
+ AMQPState.CONNECTION_NOT_SECURE, AMQPState.CONNECTION_NOT_TUNED, AMQPState.CONNECTION_NOT_OPENED,
+ AMQPState.CONNECTION_OPEN, };
+
+ // The wait period until a server sends a respond
+ private long _serverTimeOut = 1000;
+
+ private final Lock _lock = new ReentrantLock();
+
+ private final Condition _connectionNotStarted = _lock.newCondition();
+
+ private final Condition _connectionNotSecure = _lock.newCondition();
+
+ private final Condition _connectionNotTuned = _lock.newCondition();
+
+ private final Condition _connectionNotOpened = _lock.newCondition();
+
+ private final Condition _connectionNotClosed = _lock.newCondition();
+
+ private ConnectionStartBody _connectionStartBody;
+
+ private ConnectionSecureBody _connectionSecureBody;
+
+ private ConnectionTuneBody _connectionTuneBody;
+
+ private ConnectionOpenOkBody _connectionOpenOkBody;
+
+ private ConnectionCloseOkBody _connectionCloseOkBody;
+
+ public AMQPConnection(TransportConnection connection)
+ {
+ _connection = connection;
+ _currentState = AMQPState.CONNECTION_UNDEFINED;
+ _serverTimeOut = ClientConfiguration.get().getLong(QpidConstants.SERVER_TIMEOUT_IN_MILLISECONDS);
+ }
+
+ /**
+ * ------------------------------------------- API Methods --------------------------------------------
+ */
+
+ /**
+ * Opens the TCP connection and let the formalities begin.
+ */
+ public ConnectionStartBody openTCPConnection() throws AMQPException
+ {
+ _lock.lock();
+ // open the TCP connection
+ try
+ {
+ _connectionStartBody = null;
+ checkIfValidStateTransition(AMQPState.CONNECTION_UNDEFINED, _currentState, AMQPState.CONNECTION_NOT_STARTED);
+ _phase = _connection.connect();
+
+ // waiting for ConnectionStartBody or error in connection
+ _connectionNotStarted.await(_serverTimeOut, TimeUnit.MILLISECONDS);
+ AMQPValidator.throwExceptionOnNull(_connectionStartBody,
+ "The broker didn't send the ConnectionStartBody in time");
+ _currentState = AMQPState.CONNECTION_NOT_STARTED;
+ return _connectionStartBody;
+ }
+ catch (Exception e)
+ {
+ throw new AMQPException("XXX");
+ }
+ finally
+ {
+ _lock.unlock();
+ }
+ }
+
+ public ConnectionSecureBody startOk(ConnectionStartOkBody connectionStartOkBody) throws AMQPException
+ {
+ _lock.lock();
+ try
+ {
+ _connectionSecureBody = null;
+ checkIfValidStateTransition(AMQPState.CONNECTION_NOT_STARTED, _currentState,
+ AMQPState.CONNECTION_NOT_SECURE);
+ AMQPMethodEvent msg = new AMQPMethodEvent(QpidConstants.CHANNEL_ZERO, connectionStartOkBody, _correlationId);
+ _phase.messageSent(msg);
+ _connectionNotSecure.await(_serverTimeOut, TimeUnit.MILLISECONDS);
+ AMQPValidator.throwExceptionOnNull(_connectionSecureBody,
+ "The broker didn't send the ConnectionSecureBody in time");
+ _currentState = AMQPState.CONNECTION_NOT_SECURE;
+ return _connectionSecureBody;
+ }
+ catch (Exception e)
+ {
+ throw new AMQPException("XXX");
+ }
+ finally
+ {
+ _lock.unlock();
+ }
+ }
+
+ /**
+ * The server will verify the response contained in the secureOK body and send a ConnectionTuneBody or it could
+ * issue a new challenge
+ */
+ public AMQMethodBody secureOk(ConnectionSecureOkBody connectionSecureOkBody) throws AMQPException
+ {
+ _lock.lock();
+ try
+ {
+ _connectionTuneBody = null;
+ _connectionSecureBody = null;
+ checkIfValidStateTransition(AMQPState.CONNECTION_NOT_SECURE, _currentState, AMQPState.CONNECTION_NOT_TUNED);
+ _connectionSecureBody = null; // The server could send a fresh challenge
+ AMQPMethodEvent msg = new AMQPMethodEvent(QpidConstants.CHANNEL_ZERO, connectionSecureOkBody,
+ _correlationId);
+ _phase.messageSent(msg);
+ _connectionNotTuned.await(_serverTimeOut, TimeUnit.MILLISECONDS);
+ if (_connectionTuneBody != null)
+ {
+ _currentState = AMQPState.CONNECTION_NOT_TUNED;
+ return _connectionTuneBody;
+ }
+ else if (_connectionSecureBody != null)
+ { // oops the server sent another challenge
+ _currentState = AMQPState.CONNECTION_NOT_SECURE;
+ return _connectionSecureBody;
+ }
+ else
+ {
+ throw new AMQPException("The broker didn't send the ConnectionTuneBody or ConnectionSecureBody in time");
+ }
+ }
+ catch (Exception e)
+ {
+ throw new AMQPException("XXX");
+ }
+ finally
+ {
+ _lock.unlock();
+ }
+ }
+
+ public void tuneOk(ConnectionTuneOkBody connectionTuneOkBody) throws AMQPException
+ {
+ _lock.lock();
+ try
+ {
+ checkIfValidStateTransition(AMQPState.CONNECTION_NOT_TUNED, _currentState, AMQPState.CONNECTION_NOT_OPENED);
+ _connectionSecureBody = null;
+ AMQPMethodEvent msg = new AMQPMethodEvent(QpidConstants.CHANNEL_ZERO, connectionTuneOkBody, _correlationId);
+ _phase.messageSent(msg);
+ _currentState = AMQPState.CONNECTION_NOT_OPENED;
+ }
+ catch (Exception e)
+ {
+ throw new AMQPException("XXX");
+ }
+ finally
+ {
+ _lock.unlock();
+ }
+ }
+
+ public ConnectionOpenOkBody open(ConnectionOpenBody connectionOpenBody) throws AMQPException
+ {
+ _lock.lock();
+ try
+ {
+ _connectionOpenOkBody = null;
+ checkIfValidStateTransition(AMQPState.CONNECTION_NOT_OPENED, _currentState, AMQPState.CONNECTION_OPEN);
+ AMQPMethodEvent msg = new AMQPMethodEvent(QpidConstants.CHANNEL_ZERO, connectionOpenBody,
+ QpidConstants.EMPTY_CORRELATION_ID);
+ _phase.messageSent(msg);
+ _connectionNotOpened.await(_serverTimeOut, TimeUnit.MILLISECONDS);
+ AMQPValidator.throwExceptionOnNull(_connectionOpenOkBody,
+ "The broker didn't send the ConnectionOpenOkBody in time");
+ _currentState = AMQPState.CONNECTION_OPEN;
+ return _connectionOpenOkBody;
+ }
+ catch (Exception e)
+ {
+ throw new AMQPException("XXX");
+ }
+ finally
+ {
+ _lock.unlock();
+ }
+ }
+
+ public ConnectionCloseOkBody close(ConnectionCloseBody connectioncloseBody) throws AMQPException
+ {
+ _lock.lock();
+ try
+ {
+ _connectionCloseOkBody = null;
+ checkIfValidStateTransition(_validCloseStates, _currentState, AMQPState.CONNECTION_CLOSED);
+ AMQPMethodEvent msg = new AMQPMethodEvent(QpidConstants.CHANNEL_ZERO, connectioncloseBody,
+ QpidConstants.EMPTY_CORRELATION_ID);
+ _phase.messageSent(msg);
+ _connectionNotClosed.await(_serverTimeOut, TimeUnit.MILLISECONDS);
+ AMQPValidator.throwExceptionOnNull(_connectionCloseOkBody,
+ "The broker didn't send the ConnectionCloseOkBody in time");
+ _currentState = AMQPState.CONNECTION_CLOSED;
+ return _connectionCloseOkBody;
+ }
+ catch (Exception e)
+ {
+ throw new AMQPException("XXX");
+ }
+ finally
+ {
+ _lock.unlock();
+ }
+ }
+
+ /**
+ * ------------------------------------------- AMQMethodListener methods
+ * --------------------------------------------
+ */
+ public <B extends AMQMethodBody> boolean methodReceived(AMQPMethodEvent<B> evt) throws AMQPException
+ {
+ _correlationId = evt.getCorrelationId();
+
+ if (evt.getMethod() instanceof ConnectionStartBody)
+ {
+ _connectionStartBody = (ConnectionStartBody) evt.getMethod();
+ _connectionNotStarted.signal();
+ return true;
+ }
+ else if (evt.getMethod() instanceof ConnectionSecureBody)
+ {
+ _connectionSecureBody = (ConnectionSecureBody) evt.getMethod();
+ _connectionNotSecure.signal();
+ _connectionNotTuned.signal(); // in case the server has sent another chanllenge
+ return true;
+ }
+ else if (evt.getMethod() instanceof ConnectionTuneBody)
+ {
+ _connectionTuneBody = (ConnectionTuneBody) evt.getMethod();
+ _connectionNotTuned.signal();
+ return true;
+ }
+ else if (evt.getMethod() instanceof ConnectionOpenOkBody)
+ {
+ _connectionOpenOkBody = (ConnectionOpenOkBody) evt.getMethod();
+ _connectionNotOpened.signal();
+ return true;
+ }
+ else if (evt.getMethod() instanceof ConnectionCloseOkBody)
+ {
+ _connectionCloseOkBody = (ConnectionCloseOkBody) evt.getMethod();
+ _connectionNotClosed.signal();
+ return true;
+ }
+ else if (evt.getMethod() instanceof ConnectionCloseBody)
+ {
+ handleClose();
+ return true;
+ }
+ else
+ {
+ return false;
+ }
+ }
+
+ public void handleClose() throws AMQPException
+ {
+ _lock.lock();
+ try
+ {
+ checkIfValidStateTransition(AMQPState.CONNECTION_OPEN, _currentState, AMQPState.CONNECTION_CLOSING);
+ _currentState = AMQPState.CONNECTION_CLOSING;
+ // do the required cleanup and send a ConnectionCloseOkBody
+ }
+ catch (Exception e)
+ {
+ throw new AMQPException("XXX");
+ }
+ finally
+ {
+ _lock.unlock();
+ }
+ }
+
+}
\ No newline at end of file
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPExchange.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPExchange.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPExchange.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPExchange.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,88 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.qpid.nclient.amqp;
+
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.framing.ExchangeDeclareBody;
+import org.apache.qpid.framing.ExchangeDeclareOkBody;
+import org.apache.qpid.framing.ExchangeDeleteBody;
+import org.apache.qpid.framing.ExchangeDeleteOkBody;
+import org.apache.qpid.nclient.core.AMQPException;
+import org.apache.qpid.nclient.core.Phase;
+import org.apache.qpid.nclient.model.AMQPMethodEvent;
+import org.apache.qpid.nclient.model.AMQPMethodListener;
+
+/**
+ *
+ * This class represents the Exchange class defined in AMQP.
+ * Each method takes an @see AMQPCallBack object if it wants to know
+ * the response from the broker to particular method.
+ * Clients can handle the reponse asynchronously or block for a response
+ * using AMQPCallBack.isComplete() periodically using a loop.
+ */
+public class AMQPExchange extends AMQPCallBackSupport implements AMQPMethodListener
+{
+ private Phase _phase;
+
+ public AMQPExchange(int channelId,Phase phase)
+ {
+ super(channelId);
+ _phase = phase;
+ }
+
+ /**
+ * -----------------------------------------------
+ * API Methods
+ * -----------------------------------------------
+ */
+ public void declare(ExchangeDeclareBody exchangeDeclareBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleNoWait(exchangeDeclareBody.nowait,exchangeDeclareBody,cb);
+ _phase.messageSent(msg);
+ }
+
+ public void delete(ExchangeDeleteBody exchangeDeleteBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleNoWait(exchangeDeleteBody.nowait,exchangeDeleteBody,cb);
+ _phase.messageSent(msg);
+ }
+
+
+ /**-------------------------------------------
+ * AMQPMethodListener methods
+ *--------------------------------------------
+ */
+ public <B extends AMQMethodBody> boolean methodReceived(AMQPMethodEvent<B> evt) throws AMQPException
+ {
+ long localCorrelationId = evt.getLocalCorrelationId();
+ AMQMethodBody methodBody = evt.getMethod();
+ if ( methodBody instanceof ExchangeDeclareOkBody || methodBody instanceof ExchangeDeleteOkBody)
+ {
+ invokeCallBack(localCorrelationId,methodBody);
+ return true;
+ }
+ else
+ {
+ return false;
+ }
+ }
+}
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPMessage.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPMessage.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPMessage.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPMessage.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,232 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.nclient.amqp;
+
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.framing.MessageAppendBody;
+import org.apache.qpid.framing.MessageCancelBody;
+import org.apache.qpid.framing.MessageCheckpointBody;
+import org.apache.qpid.framing.MessageCloseBody;
+import org.apache.qpid.framing.MessageEmptyBody;
+import org.apache.qpid.framing.MessageGetBody;
+import org.apache.qpid.framing.MessageOffsetBody;
+import org.apache.qpid.framing.MessageOkBody;
+import org.apache.qpid.framing.MessageOpenBody;
+import org.apache.qpid.framing.MessageQosBody;
+import org.apache.qpid.framing.MessageRecoverBody;
+import org.apache.qpid.framing.MessageRejectBody;
+import org.apache.qpid.framing.MessageResumeBody;
+import org.apache.qpid.framing.MessageTransferBody;
+import org.apache.qpid.nclient.core.AMQPException;
+import org.apache.qpid.nclient.core.Phase;
+import org.apache.qpid.nclient.model.AMQPMethodEvent;
+import org.apache.qpid.nclient.model.AMQPMethodListener;
+
+/**
+ * This class represents the AMQP Message class.
+ * You need an instance of this class per channel.
+ * A @see AMQPMessageCallBack class is taken as an argument in the constructor.
+ * A client can use this class to issue Message class methods on the broker.
+ * When the broker issues Message class methods on the client, the client is notified
+ * via the AMQPMessageCallBack interface.
+ *
+ * A JMS Message producer implementation can wrap an instance if this and map
+ * JMS method calls to the appropriate AMQP methods.
+ *
+ * AMQPMessageCallBack can be implemented by the JMS MessageConsumer implementation.
+ *
+ */
+public class AMQPMessage extends AMQPCallBackSupport implements AMQPMethodListener
+{
+ private Phase _phase;
+ private AMQPMessageCallBack _messageCb;
+
+ public AMQPMessage(int channelId,Phase phase,AMQPMessageCallBack messageCb)
+ {
+ super(channelId);
+ _phase = phase;
+ _messageCb = messageCb;
+ }
+
+ /**
+ * -----------------------------------------------
+ * API Methods
+ * -----------------------------------------------
+ */
+
+ public void transfer(MessageTransferBody messageTransferBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleAsynchronousCall(messageTransferBody,cb);
+ _phase.messageSent(msg);
+ }
+
+ public void consume(MessageCancelBody messageCancelBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleAsynchronousCall(messageCancelBody,cb);
+ _phase.messageSent(msg);
+ }
+
+ public void cancel(MessageCancelBody messageCancelBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleAsynchronousCall(messageCancelBody,cb);
+ _phase.messageSent(msg);
+ }
+
+ public void get(MessageGetBody messageGetBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleAsynchronousCall(messageGetBody,cb);
+ _phase.messageSent(msg);
+ }
+
+ public void recover(MessageRecoverBody messageRecoverBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleAsynchronousCall(messageRecoverBody,cb);
+ _phase.messageSent(msg);
+ }
+
+ public void open(MessageOpenBody messageOpenBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleAsynchronousCall(messageOpenBody,cb);
+ _phase.messageSent(msg);
+ }
+
+ public void close(MessageCloseBody messageCloseBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleAsynchronousCall(messageCloseBody,cb);
+ _phase.messageSent(msg);
+ }
+
+ public void append(MessageAppendBody messageAppendBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleAsynchronousCall(messageAppendBody,cb);
+ _phase.messageSent(msg);
+ }
+
+ public void checkpoint(MessageCheckpointBody messageCheckpointBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleAsynchronousCall(messageCheckpointBody,cb);
+ _phase.messageSent(msg);
+ }
+
+ public void resume(MessageResumeBody messageResumeBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleAsynchronousCall(messageResumeBody,cb);
+ _phase.messageSent(msg);
+ }
+
+ public void qos(MessageQosBody messageQosBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleAsynchronousCall(messageQosBody,cb);
+ _phase.messageSent(msg);
+ }
+
+ /**
+ * The correlationId from the request.
+ * For example if a message.transfer is sent with correlationId "ABCD"
+ * then u need to pass that in. This correlation id is used by the execution layer
+ * to handle the correlation of method requests and responses
+ */
+ public void ok(MessageOkBody messageOkBody,long correlationId) throws AMQPException
+ {
+ AMQPMethodEvent msg = new AMQPMethodEvent(_channelId,messageOkBody,correlationId);
+ _phase.messageSent(msg);
+ }
+
+ /**
+ * The correlationId from the request.
+ * For example if a message.transfer is sent with correlationId "ABCD"
+ * then u need to pass that in. This correlation id is used by the execution layer
+ * to handle the correlation of method requests and responses
+ */
+ public void reject(MessageRejectBody messageRejectBody,long correlationId) throws AMQPException
+ {
+ AMQPMethodEvent msg = new AMQPMethodEvent(_channelId,messageRejectBody,correlationId);
+ _phase.messageSent(msg);
+ }
+
+ /**
+ * The correlationId from the request.
+ * For example if a message.resume is sent with correlationId "ABCD"
+ * then u need to pass that in. This correlation id is used by the execution layer
+ * to handle the correlation of method requests and responses
+ */
+ public void offset(MessageOffsetBody messageOffsetBody,long correlationId) throws AMQPException
+ {
+ AMQPMethodEvent msg = new AMQPMethodEvent(_channelId,messageOffsetBody,correlationId);
+ _phase.messageSent(msg);
+ }
+
+ /**-------------------------------------------
+ * AMQPMethodListener methods
+ *--------------------------------------------
+ */
+ public <B extends AMQMethodBody> boolean methodReceived(AMQPMethodEvent<B> evt) throws AMQPException
+ {
+ long localCorrelationId = evt.getLocalCorrelationId();
+ AMQMethodBody methodBody = evt.getMethod();
+ if ( methodBody instanceof MessageOkBody ||
+ methodBody instanceof MessageRejectBody ||
+ methodBody instanceof MessageEmptyBody)
+ {
+ invokeCallBack(localCorrelationId,methodBody);
+ return true;
+ }
+ else if (methodBody instanceof MessageTransferBody)
+ {
+ _messageCb.transfer((MessageTransferBody)methodBody, evt.getCorrelationId());
+ return true;
+ }
+ else if (methodBody instanceof MessageAppendBody)
+ {
+ _messageCb.append((MessageAppendBody)methodBody, evt.getCorrelationId());
+ return true;
+ }
+ else if (methodBody instanceof MessageOpenBody)
+ {
+ _messageCb.open((MessageOpenBody)methodBody, evt.getCorrelationId());
+ return true;
+ }
+ else if (methodBody instanceof MessageCloseBody)
+ {
+ _messageCb.close((MessageCloseBody)methodBody, evt.getCorrelationId());
+ return true;
+ }
+ else if (methodBody instanceof MessageCheckpointBody)
+ {
+ _messageCb.checkpoint((MessageCheckpointBody)methodBody, evt.getCorrelationId());
+ return true;
+ }
+ else if (methodBody instanceof MessageRecoverBody)
+ {
+ _messageCb.recover((MessageRecoverBody)methodBody, evt.getCorrelationId());
+ return true;
+ }
+ else if (methodBody instanceof MessageResumeBody)
+ {
+ _messageCb.resume((MessageResumeBody)methodBody, evt.getCorrelationId());
+ return true;
+ }
+ else
+ {
+ return false;
+ }
+ }
+}
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPMessageCallBack.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPMessageCallBack.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPMessageCallBack.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPMessageCallBack.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,78 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.nclient.amqp;
+
+import org.apache.qpid.framing.MessageAppendBody;
+import org.apache.qpid.framing.MessageCancelBody;
+import org.apache.qpid.framing.MessageCheckpointBody;
+import org.apache.qpid.framing.MessageCloseBody;
+import org.apache.qpid.framing.MessageGetBody;
+import org.apache.qpid.framing.MessageOpenBody;
+import org.apache.qpid.framing.MessageRecoverBody;
+import org.apache.qpid.framing.MessageResumeBody;
+import org.apache.qpid.framing.MessageTransferBody;
+import org.apache.qpid.nclient.core.AMQPException;
+
+/**
+ * This class also represents the AMQP Message class.
+ * You need an instance per channel.
+ * This is passed in as an argument in the constructor of an AMQPMessage instance.
+ * A client who implements this interface is notified When the broker issues
+ * Message class methods on the client.
+ *
+ * A Client should use the AMQPMessage class when it wants to issue Message class
+ * methods on the broker.
+ *
+ * A JMS MessageConsumer implementation can implement this interface and map
+ * AMQP Method notifications to the appropriate JMS methods.
+ *
+ * Simillarly a JMS MessageProducer implementation can wrap an AMQPMessage instance.
+ *
+ */
+
+public interface AMQPMessageCallBack
+{
+ /**
+ * -----------------------------------------------------------------------
+ * This provides Notifications for broker initiated Message class methods.
+ * All methods have a correlationId that u need to pass into
+ * the corresponding Message methods when responding to the broker.
+ *
+ * For example the correlationID passed in from Message.trasnfer
+ * should be passed back when u call Message.ok in AMQPMessage
+ * -----------------------------------------------------------------------
+ */
+
+
+ public void transfer(MessageTransferBody messageTransferBody,long correlationId) throws AMQPException;
+
+ public void recover(MessageRecoverBody messageRecoverBody,long correlationId) throws AMQPException;
+
+ public void open(MessageOpenBody messageOpenBody,long correlationId) throws AMQPException ;
+
+ public void close(MessageCloseBody messageCloseBody,long correlationId) throws AMQPException;
+
+ public void append(MessageAppendBody messageAppendBody,long correlationId) throws AMQPException;
+
+ public void checkpoint(MessageCheckpointBody messageCheckpointBody,long correlationId) throws AMQPException;
+
+ public void resume(MessageResumeBody messageResumeBody,long correlationId) throws AMQPException;
+}
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPQueue.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPQueue.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPQueue.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/AMQPQueue.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,117 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.nclient.amqp;
+
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.framing.QueueBindBody;
+import org.apache.qpid.framing.QueueBindOkBody;
+import org.apache.qpid.framing.QueueDeclareBody;
+import org.apache.qpid.framing.QueueDeclareOkBody;
+import org.apache.qpid.framing.QueueDeleteBody;
+import org.apache.qpid.framing.QueueDeleteOkBody;
+import org.apache.qpid.framing.QueuePurgeBody;
+import org.apache.qpid.framing.QueuePurgeOkBody;
+import org.apache.qpid.framing.QueueUnbindBody;
+import org.apache.qpid.framing.QueueUnbindOkBody;
+import org.apache.qpid.nclient.core.AMQPException;
+import org.apache.qpid.nclient.core.Phase;
+import org.apache.qpid.nclient.model.AMQPMethodEvent;
+import org.apache.qpid.nclient.model.AMQPMethodListener;
+
+/**
+ *
+ * This class represents the Queue class defined in AMQP.
+ * Each method takes an @see AMQPCallBack object if it wants to know
+ * the response from the broker to a particular method.
+ * Clients can handle the reponse asynchronously or block for a response
+ * using AMQPCallBack.isComplete() periodically using a loop.
+ */
+public class AMQPQueue extends AMQPCallBackSupport implements AMQPMethodListener
+{
+ private Phase _phase;
+
+ public AMQPQueue(int channelId,Phase phase)
+ {
+ super(channelId);
+ _phase = phase;
+ }
+
+ /**
+ * -----------------------------------------------
+ * API Methods
+ * -----------------------------------------------
+ */
+ public void declare(QueueDeclareBody queueDeclareBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleNoWait(queueDeclareBody.nowait,queueDeclareBody,cb);
+ _phase.messageSent(msg);
+ }
+
+ public void bind(QueueBindBody queueBindBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleNoWait(queueBindBody.nowait,queueBindBody,cb);
+ _phase.messageSent(msg);
+ }
+
+ // Queue.unbind doesn't have nowait
+ public void unbind(QueueUnbindBody queueUnbindBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleAsynchronousCall(queueUnbindBody,cb);
+ _phase.messageSent(msg);
+ }
+
+ public void purge(QueuePurgeBody queuePurgeBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleNoWait(queuePurgeBody.nowait,queuePurgeBody,cb);
+ _phase.messageSent(msg);
+ }
+
+ public void delete(QueueDeleteBody queueDeleteBody,AMQPCallBack cb) throws AMQPException
+ {
+ AMQPMethodEvent msg = handleNoWait(queueDeleteBody.nowait,queueDeleteBody,cb);
+ _phase.messageSent(msg);
+ }
+
+
+ /**-------------------------------------------
+ * AMQPMethodListener methods
+ *--------------------------------------------
+ */
+ public <B extends AMQMethodBody> boolean methodReceived(AMQPMethodEvent<B> evt) throws AMQPException
+ {
+ long localCorrelationId = evt.getLocalCorrelationId();
+ AMQMethodBody methodBody = evt.getMethod();
+ if ( methodBody instanceof QueueDeclareOkBody ||
+ methodBody instanceof QueueBindOkBody ||
+ methodBody instanceof QueueUnbindOkBody ||
+ methodBody instanceof QueuePurgeOkBody ||
+ methodBody instanceof QueueDeleteOkBody
+ )
+ {
+ invokeCallBack(localCorrelationId,methodBody);
+ return true;
+ }
+ else
+ {
+ return false;
+ }
+ }
+}
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/EventManager.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/EventManager.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/EventManager.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/EventManager.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,129 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.nclient.amqp;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.nclient.core.AMQPException;
+import org.apache.qpid.nclient.model.AMQPMethodEvent;
+import org.apache.qpid.nclient.model.AMQPMethodListener;
+
+/**
+ * This class registeres with the ModelPhase as a AMQMethodListener,
+ * to receive method events and then it distributes methods to other listerners
+ * using a filtering criteria. The criteria is channel id and method body class.
+ * The method listeners are added and removed dynamically
+ *
+ * <p/>
+ */
+public class EventManager implements AMQPMethodListener
+{
+ private static final Logger _logger = Logger.getLogger(EventManager.class);
+
+ private Map <Integer,Map> _channelMap = new ConcurrentHashMap<Integer,Map>();
+
+ /**
+ * ------------------------------------------------
+ * methods introduced by AMQPMethodEventManager
+ * ------------------------------------------------
+ */
+ public void addMethodEventListener(int channelId,Class clazz,AMQPMethodListener l)
+ {
+ Map<Class,List> _methodListenerMap;
+ if (_channelMap.containsKey(channelId))
+ {
+ _methodListenerMap = _channelMap.get(channelId);
+
+ }
+ else
+ {
+ _methodListenerMap = new ConcurrentHashMap<Class,List>();
+ _channelMap.put(channelId, _methodListenerMap);
+ }
+
+ List<AMQPMethodListener> _listeners;
+ if (_methodListenerMap.containsKey(clazz))
+ {
+ _listeners = _methodListenerMap.get(clazz);
+
+ }
+ else
+ {
+ _listeners = new ArrayList<AMQPMethodListener>();
+ _methodListenerMap.put(clazz, _listeners);
+ }
+
+ _listeners.add(l);
+
+ }
+
+ public void removeMethodEventListener(int channelId,Class clazz,AMQPMethodListener l)
+ {
+ if (_channelMap.containsKey(channelId))
+ {
+ Map<Class,List> _methodListenerMap = _channelMap.get(channelId);
+
+ if (_methodListenerMap.containsKey(clazz))
+ {
+ List<AMQPMethodListener> _listeners = _methodListenerMap.get(clazz);
+ _listeners.remove(l);
+ }
+
+ }
+ }
+
+
+ /**
+ * ------------------------------------------------
+ * methods introduced by AMQMethodListener
+ * ------------------------------------------------
+ */
+ /* (non-Javadoc)
+ * @see org.apache.qpid.nclient.model.AMQStateManager#methodReceived(org.apache.qpid.protocol.AMQMethodEvent)
+ */
+ public <B extends AMQMethodBody> boolean methodReceived(AMQPMethodEvent<B> evt) throws AMQPException
+ {
+ if (_channelMap.containsKey(evt.getChannelId()))
+ {
+ Map<Class,List> _methodListenerMap = _channelMap.get(evt.getChannelId());
+
+ if (_methodListenerMap.containsKey(evt.getMethod().getClass()))
+ {
+
+ List<AMQPMethodListener> _listeners = _methodListenerMap.get(evt.getMethod().getClass());
+ for (AMQPMethodListener l:_listeners)
+ {
+ l.methodReceived(evt);
+ }
+
+ return (_listeners.size()>0);
+ }
+
+ }
+
+ return false;
+ }
+}
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/sample/SecurityHelper.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/sample/SecurityHelper.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/sample/SecurityHelper.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/sample/SecurityHelper.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,54 @@
+package org.apache.qpid.nclient.amqp.sample;
+
+import java.io.UnsupportedEncodingException;
+import java.util.HashSet;
+import java.util.StringTokenizer;
+
+import org.apache.qpid.nclient.core.AMQPException;
+import org.apache.qpid.nclient.security.AMQPCallbackHandler;
+import org.apache.qpid.nclient.security.CallbackHandlerRegistry;
+import org.apache.qpid.nclient.transport.ConnectionURL;
+
+public class SecurityHelper
+{
+ public static String chooseMechanism(byte[] availableMechanisms) throws UnsupportedEncodingException
+ {
+ final String mechanisms = new String(availableMechanisms, "utf8");
+ StringTokenizer tokenizer = new StringTokenizer(mechanisms, " ");
+ HashSet mechanismSet = new HashSet();
+ while (tokenizer.hasMoreTokens())
+ {
+ mechanismSet.add(tokenizer.nextToken());
+ }
+
+ String preferredMechanisms = CallbackHandlerRegistry.getInstance().getMechanisms();
+ StringTokenizer prefTokenizer = new StringTokenizer(preferredMechanisms, " ");
+ while (prefTokenizer.hasMoreTokens())
+ {
+ String mech = prefTokenizer.nextToken();
+ if (mechanismSet.contains(mech))
+ {
+ return mech;
+ }
+ }
+ return null;
+ }
+
+ public static AMQPCallbackHandler createCallbackHandler(String mechanism, ConnectionURL url)
+ throws AMQPException
+ {
+ Class mechanismClass = CallbackHandlerRegistry.getInstance().getCallbackHandlerClass(mechanism);
+ try
+ {
+ Object instance = mechanismClass.newInstance();
+ AMQPCallbackHandler cbh = (AMQPCallbackHandler) instance;
+ cbh.initialise(url);
+ return cbh;
+ }
+ catch (Exception e)
+ {
+ throw new AMQPException("Unable to create callback handler: " + e, e);
+ }
+ }
+
+}
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/sample/TestClient.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/sample/TestClient.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/sample/TestClient.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/sample/TestClient.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,93 @@
+package org.apache.qpid.nclient.amqp.sample;
+
+import java.util.StringTokenizer;
+
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslClient;
+
+import org.apache.qpid.common.ClientProperties;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.ConnectionSecureBody;
+import org.apache.qpid.framing.ConnectionSecureOkBody;
+import org.apache.qpid.framing.ConnectionStartBody;
+import org.apache.qpid.framing.ConnectionStartOkBody;
+import org.apache.qpid.framing.ConnectionTuneBody;
+import org.apache.qpid.framing.ConnectionTuneOkBody;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.framing.FieldTableFactory;
+import org.apache.qpid.nclient.amqp.AMQPConnection;
+import org.apache.qpid.nclient.transport.AMQPConnectionURL;
+import org.apache.qpid.nclient.transport.ConnectionURL;
+import org.apache.qpid.nclient.transport.TransportConnection;
+import org.apache.qpid.nclient.transport.TransportConnectionFactory;
+import org.apache.qpid.nclient.transport.TransportConnectionFactory.ConnectionType;
+
+/**
+ * This class illustrates the usage of the API
+ * Notes this is just a simple demo.
+ *
+ * I have used Helper classes to keep the code cleaner.
+ */
+public class TestClient
+{
+ private byte major;
+ private byte minor;
+ private ConnectionURL _url;
+
+ public AMQPConnection openConnection() throws Exception
+ {
+ _url = new AMQPConnectionURL("");
+ TransportConnection conn = TransportConnectionFactory.createTransportConnection(_url, ConnectionType.VM);
+ return new AMQPConnection(conn);
+ }
+
+ public void handleProtocolNegotiation(AMQPConnection con) throws Exception
+ {
+ // ConnectionStartBody
+ ConnectionStartBody connectionStartBody = con.openTCPConnection();
+ major = connectionStartBody.getMajor();
+ minor = connectionStartBody.getMajor();
+
+ FieldTable clientProperties = FieldTableFactory.newFieldTable();
+ clientProperties.put(new AMQShortString(ClientProperties.instance.toString()),"Test"); // setting only the client id
+
+ final String locales = new String(connectionStartBody.getLocales(), "utf8");
+ final StringTokenizer tokenizer = new StringTokenizer(locales, " ");
+
+ final String mechanism = SecurityHelper.chooseMechanism(connectionStartBody.getMechanisms());
+
+ SaslClient sc = Sasl.createSaslClient(new String[]{mechanism},
+ null, "AMQP", "localhost",
+ null, SecurityHelper.createCallbackHandler(mechanism,_url));
+
+ ConnectionStartOkBody connectionStartOkBody =
+ ConnectionStartOkBody.createMethodBody(major, minor, clientProperties,
+ new AMQShortString(tokenizer.nextToken()),
+ new AMQShortString(mechanism),
+ (sc.hasInitialResponse() ? sc.evaluateChallenge(new byte[0]) : null));
+ // ConnectionSecureBody
+ ConnectionSecureBody connectionSecureBody = con.startOk(connectionStartOkBody);
+
+ ConnectionSecureOkBody connectionSecureOkBody = ConnectionSecureOkBody.createMethodBody(
+ major,minor,sc.evaluateChallenge(connectionSecureBody.getChallenge()));
+
+ // Assuming the server is not going to send another challenge
+ ConnectionTuneBody connectionTuneBody = (ConnectionTuneBody)con.secureOk(connectionSecureOkBody);
+
+ // Using broker supplied values
+ ConnectionTuneOkBody connectionTuneOkBody =
+ ConnectionTuneOkBody.createMethodBody(major,minor,
+ connectionTuneBody.getChannelMax(),
+ connectionTuneBody.getFrameMax(),
+ connectionTuneBody.getHeartbeat());
+ con.tuneOk(connectionTuneOkBody);
+ }
+
+ public static void main(String[] args)
+ {
+ TestClient test = new TestClient();
+ AMQPConnection con = test.openConnection();
+
+ }
+
+}
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPState.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPState.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPState.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPState.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,59 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.nclient.amqp.state;
+
+/**
+ * States used in the AMQ protocol. Used by the finite state machine to determine
+ * valid responses.
+ */
+public class AMQPState
+{
+ private final int _id;
+
+ private final String _name;
+
+ private AMQPState(int id, String name)
+ {
+ _id = id;
+ _name = name;
+ }
+
+ public String toString()
+ {
+ return "AMQState: id = " + _id + " name: " + _name;
+ }
+
+ // Connection state
+ public static final AMQPState CONNECTION_UNDEFINED = new AMQPState(0, "CONNECTION_UNDEFINED");
+ public static final AMQPState CONNECTION_NOT_STARTED = new AMQPState(1, "CONNECTION_NOT_STARTED");
+ public static final AMQPState CONNECTION_NOT_SECURE = new AMQPState(2, "CONNECTION_NOT_SECURE");
+ public static final AMQPState CONNECTION_NOT_TUNED = new AMQPState(2, "CONNECTION_NOT_TUNED");
+ public static final AMQPState CONNECTION_NOT_OPENED = new AMQPState(3, "CONNECTION_NOT_OPENED");
+ public static final AMQPState CONNECTION_OPEN = new AMQPState(4, "CONNECTION_OPEN");
+ public static final AMQPState CONNECTION_CLOSING = new AMQPState(5, "CONNECTION_CLOSING");
+ public static final AMQPState CONNECTION_CLOSED = new AMQPState(6, "CONNECTION_CLOSED");
+
+ // Channel state
+ public static final AMQPState CHANNEL_NOT_OPENED = new AMQPState(10, "CHANNEL_NOT_OPENED");
+ public static final AMQPState CHANNEL_OPENED = new AMQPState(11, "CHANNEL_OPENED");
+ public static final AMQPState CHANNEL_CLOSED = new AMQPState(11, "CHANNEL_CLOSED");
+ public static final AMQPState CHANNEL_SUSPEND = new AMQPState(11, "CHANNEL_SUSPEND");
+}
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateListener.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateListener.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateListener.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateListener.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,8 @@
+package org.apache.qpid.nclient.amqp.state;
+
+import org.apache.qpid.nclient.core.AMQPException;
+
+public interface AMQPStateListener
+{
+ public void stateChanged(AMQPState oldState, AMQPState newState) throws AMQPException;
+}
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateMachine.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateMachine.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateMachine.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateMachine.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,26 @@
+package org.apache.qpid.nclient.amqp.state;
+
+import org.apache.qpid.nclient.core.AMQPException;
+
+public class AMQPStateMachine
+{
+ protected void checkIfValidStateTransition(AMQPState correctState,AMQPState currentState,AMQPState requiredState) throws IllegalStateTransitionException
+ {
+ if (currentState != correctState)
+ {
+ throw new IllegalStateTransitionException(currentState,requiredState);
+ }
+ }
+
+ protected void checkIfValidStateTransition(AMQPState[] correctStates,AMQPState currentState,AMQPState requiredState) throws IllegalStateTransitionException
+ {
+ for(AMQPState correctState :correctStates)
+ {
+ if (currentState == correctState)
+ {
+ return;
+ }
+ }
+ throw new IllegalStateTransitionException(currentState,requiredState);
+ }
+}
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateManager.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateManager.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateManager.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateManager.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,11 @@
+package org.apache.qpid.nclient.amqp.state;
+
+import org.apache.qpid.AMQException;
+
+public interface AMQPStateManager
+{
+
+ public void addListener(AMQPStateListener l)throws AMQException;
+
+ public void removeListener(AMQPStateListener l)throws AMQException;
+}
\ No newline at end of file
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateType.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateType.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateType.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/AMQPStateType.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,50 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.nclient.amqp.state;
+
+/**
+ * The Type of States used in the AMQ protocol.
+ * This allows to partition listeners by the type of states they want
+ * to listen rather than all.
+ * For example an Object might only be interested in Channel state
+ */
+public class AMQPStateType
+{
+ private final int _typeId;
+
+ private final String _typeName;
+
+ private AMQPStateType(int id, String name)
+ {
+ _typeId = id;
+ _typeName = name;
+ }
+
+ public String toString()
+ {
+ return "AMQState: id = " + _typeId + " name: " + _typeName;
+ }
+
+ // Connection state
+ public static final AMQPStateType CONNECTION_STATE = new AMQPStateType(0, "CONNECTION_STATE");
+ public static final AMQPStateType CHANNEL_STATE = new AMQPStateType(1, "CHANNEL_STATE");
+
+}
Added: incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/IllegalStateTransitionException.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/IllegalStateTransitionException.java?view=auto&rev=522988
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/IllegalStateTransitionException.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/newclient/src/main/java/org/apache/qpid/nclient/amqp/state/IllegalStateTransitionException.java Tue Mar 27 09:31:05 2007
@@ -0,0 +1,47 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.nclient.amqp.state;
+
+import org.apache.qpid.nclient.core.AMQPException;
+
+public class IllegalStateTransitionException extends AMQPException
+{
+ private AMQPState _currentState;
+ private AMQPState _desiredState;
+
+ public IllegalStateTransitionException(AMQPState currentState, AMQPState desiredState)
+ {
+ super("No valid state transition defined from state " + currentState +
+ " to state " + desiredState);
+ _currentState = currentState;
+ _desiredState = desiredState;
+ }
+
+ public AMQPState getCurrentState()
+ {
+ return _currentState;
+ }
+
+ public AMQPState getDesiredState()
+ {
+ return _desiredState;
+ }
+}