You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2014/10/03 19:41:01 UTC
svn commit: r1629270 [3/3] - in
/qpid/branches/QPID-6125-ProtocolRefactoring/java:
bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/
broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/
broker-plugins/amqp-...
Modified: qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/MethodRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/MethodRegistry.java?rev=1629270&r1=1629269&r2=1629270&view=diff
==============================================================================
--- qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/MethodRegistry.java (original)
+++ qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/MethodRegistry.java Fri Oct 3 17:40:59 2014
@@ -29,302 +29,23 @@
package org.apache.qpid.framing;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.qpid.codec.MarkableDataInput;
-import org.apache.qpid.protocol.AMQConstant;
-
-
public final class MethodRegistry
{
- private static final Map<ProtocolVersion, MethodRegistry> _registries = new HashMap<>();
-
+ private final FrameCreatingMethodProcessor _methodProcessor;
+ private ProtocolVersion _protocolVersion;
- public static final MethodRegistry registry_0_9 =
- new MethodRegistry(ProtocolVersion.v0_9);
- public static final MethodRegistry registry_0_91 =
- new MethodRegistry(ProtocolVersion.v0_91);
-
- public static final MethodRegistry registry_8_0 =
- new MethodRegistry(ProtocolVersion.v8_0);
-
- private final ProtocolVersion _protocolVersion;
-
- public final AMQMethodBody convertToBody(MarkableDataInput in, long size)
- throws AMQFrameDecodingException, IOException
- {
- final int classAndMethod = in.readInt();
-
- AMQMethodBody methodBody;
- switch(classAndMethod)
- {
- //CONNECTION_CLASS:
- case 0x000a000a:
- methodBody = new ConnectionStartBody(in);
- break;
- case 0x000a000b:
- methodBody = new ConnectionStartOkBody(in);
- break;
- case 0x000a0014:
- methodBody = new ConnectionSecureBody(in);
- break;
- case 0x000a0015:
- methodBody = new ConnectionSecureOkBody(in);
- break;
- case 0x000a001e:
- methodBody = new ConnectionTuneBody(in);
- break;
- case 0x000a001f:
- methodBody = new ConnectionTuneOkBody(in);
- break;
- case 0x000a0028:
- methodBody = new ConnectionOpenBody(in);
- break;
- case 0x000a0029:
- methodBody = new ConnectionOpenOkBody(in);
- break;
- case 0x000a002a:
- methodBody = new ConnectionRedirectBody(in, _protocolVersion);
- break;
- case 0x000a0032:
- if(_protocolVersion.equals(ProtocolVersion.v8_0))
- {
- methodBody = new ConnectionRedirectBody(in, _protocolVersion);
- }
- else
- {
- methodBody = new ConnectionCloseBody(in, _protocolVersion);
- }
- break;
- case 0x000a0033:
- if(_protocolVersion.equals(ProtocolVersion.v8_0))
- {
- throw newUnknownMethodException((classAndMethod >> 16), (classAndMethod & 0xFFFF));
- }
- else
- {
- methodBody = ConnectionCloseOkBody.CONNECTION_CLOSE_OK_0_9;
- }
- break;
- case 0x000a003c:
- if(_protocolVersion.equals(ProtocolVersion.v8_0))
- {
- methodBody = new ConnectionCloseBody(in, _protocolVersion);
- }
- else
- {
- throw newUnknownMethodException((classAndMethod >> 16), (classAndMethod & 0xFFFF));
- }
- break;
- case 0x000a003d:
- if(_protocolVersion.equals(ProtocolVersion.v8_0))
- {
- methodBody = ConnectionCloseOkBody.CONNECTION_CLOSE_OK_0_8;
- }
- else
- {
- throw newUnknownMethodException((classAndMethod >> 16), (classAndMethod & 0xFFFF));
- }
- break;
-
- // CHANNEL_CLASS:
-
- case 0x0014000a:
- methodBody = new ChannelOpenBody(in);
- break;
- case 0x0014000b:
- methodBody = ChannelOpenOkBody.getInstance(_protocolVersion, in);
- break;
- case 0x00140014:
- methodBody = new ChannelFlowBody(in);
- break;
- case 0x00140015:
- methodBody = new ChannelFlowOkBody(in);
- break;
- case 0x0014001e:
- methodBody = new ChannelAlertBody(in);
- break;
- case 0x00140028:
- methodBody = new ChannelCloseBody(in);
- break;
- case 0x00140029:
- methodBody = new ChannelCloseOkBody(in);
- break;
- // ACCESS_CLASS:
-
- case 0x001e000a:
- methodBody = new AccessRequestBody(in);
- break;
- case 0x001e000b:
- methodBody = new AccessRequestOkBody(in);
- break;
-
- // EXCHANGE_CLASS:
-
- case 0x0028000a:
- methodBody = new ExchangeDeclareBody(in);
- break;
- case 0x0028000b:
- methodBody = new ExchangeDeclareOkBody(in);
- break;
- case 0x00280014:
- methodBody = new ExchangeDeleteBody(in);
- break;
- case 0x00280015:
- methodBody = new ExchangeDeleteOkBody(in);
- break;
- case 0x00280016:
- methodBody = new ExchangeBoundBody(in);
- break;
- case 0x00280017:
- methodBody = new ExchangeBoundOkBody(in);
- break;
-
-
- // QUEUE_CLASS:
-
- case 0x0032000a:
- methodBody = new QueueDeclareBody(in);
- break;
- case 0x0032000b:
- methodBody = new QueueDeclareOkBody(in);
- break;
- case 0x00320014:
- methodBody = new QueueBindBody(in);
- break;
- case 0x00320015:
- methodBody = new QueueBindOkBody(in);
- break;
- case 0x0032001e:
- methodBody = new QueuePurgeBody(in);
- break;
- case 0x0032001f:
- methodBody = new QueuePurgeOkBody(in);
- break;
- case 0x00320028:
- methodBody = new QueueDeleteBody(in);
- break;
- case 0x00320029:
- methodBody = new QueueDeleteOkBody(in);
- break;
- case 0x00320032:
- methodBody = new QueueUnbindBody(in);
- break;
- case 0x00320033:
- methodBody = new QueueUnbindOkBody(in);
- break;
-
-
- // BASIC_CLASS:
-
- case 0x003c000a:
- methodBody = new BasicQosBody(in);
- break;
- case 0x003c000b:
- methodBody = new BasicQosOkBody(in);
- break;
- case 0x003c0014:
- methodBody = new BasicConsumeBody(in);
- break;
- case 0x003c0015:
- methodBody = new BasicConsumeOkBody(in);
- break;
- case 0x003c001e:
- methodBody = new BasicCancelBody(in);
- break;
- case 0x003c001f:
- methodBody = new BasicCancelOkBody(in);
- break;
- case 0x003c0028:
- methodBody = new BasicPublishBody(in);
- break;
- case 0x003c0032:
- methodBody = new BasicReturnBody(in);
- break;
- case 0x003c003c:
- methodBody = new BasicDeliverBody(in);
- break;
- case 0x003c0046:
- methodBody = new BasicGetBody(in);
- break;
- case 0x003c0047:
- methodBody = new BasicGetOkBody(in);
- break;
- case 0x003c0048:
- methodBody = new BasicGetEmptyBody(in);
- break;
- case 0x003c0050:
- methodBody = new BasicAckBody(in);
- break;
- case 0x003c005a:
- methodBody = new BasicRejectBody(in);
- break;
- case 0x003c0064:
- methodBody = new BasicRecoverBody(in);
- break;
- case 0x003c0065:
- methodBody = new BasicRecoverSyncOkBody(_protocolVersion);
- break;
- case 0x003c0066:
- methodBody = new BasicRecoverSyncBody(in, _protocolVersion);
- break;
- case 0x003c006e:
- methodBody = new BasicRecoverSyncBody(in, _protocolVersion);
- break;
- case 0x003c006f:
- methodBody = new BasicRecoverSyncOkBody(_protocolVersion);
- break;
-
- // TX_CLASS:
-
- case 0x005a000a:
- methodBody = TxSelectBody.INSTANCE;
- break;
- case 0x005a000b:
- methodBody = TxSelectOkBody.INSTANCE;
- break;
- case 0x005a0014:
- methodBody = TxCommitBody.INSTANCE;
- break;
- case 0x005a0015:
- methodBody = TxCommitOkBody.INSTANCE;
- break;
- case 0x005a001e:
- methodBody = TxRollbackBody.INSTANCE;
- break;
- case 0x005a001f:
- methodBody = TxRollbackOkBody.INSTANCE;
- break;
-
- default:
- throw newUnknownMethodException((classAndMethod >> 16), (classAndMethod & 0xFFFF));
-
- }
- return methodBody;
- }
-
- private AMQFrameDecodingException newUnknownMethodException(final int classId, final int methodId)
- {
- return new AMQFrameDecodingException(AMQConstant.COMMAND_INVALID,
- "Method " + methodId + " unknown in AMQP version " + _protocolVersion
- + " (while trying to decode class " + classId + " method " + methodId + ".");
- }
-
- private MethodRegistry(ProtocolVersion pv)
+ public MethodRegistry(ProtocolVersion pv)
{
- _registries.put(pv, this);
_protocolVersion = pv;
+ _methodProcessor = new FrameCreatingMethodProcessor(this);
}
- public static MethodRegistry getMethodRegistry(ProtocolVersion pv)
+ public void setProtocolVersion(final ProtocolVersion protocolVersion)
{
- return _registries.get(pv);
+ _protocolVersion = protocolVersion;
}
-
public final AccessRequestBody createAccessRequestBody(final AMQShortString realm,
final boolean exclusive,
final boolean passive,
@@ -502,7 +223,7 @@ public final class MethodRegistry
public final ChannelOpenBody createChannelOpenBody(final AMQShortString outOfBand)
{
- return new ChannelOpenBody(outOfBand);
+ return new ChannelOpenBody();
}
public final ChannelOpenOkBody createChannelOpenOkBody(byte[] channelId)
@@ -540,7 +261,7 @@ public final class MethodRegistry
public final ChannelCloseOkBody createChannelCloseOkBody()
{
- return new ChannelCloseOkBody();
+ return ChannelCloseOkBody.INSTANCE;
}
@@ -829,4 +550,15 @@ public final class MethodRegistry
return TxRollbackOkBody.INSTANCE;
}
+ public ProtocolVersion getProtocolVersion()
+ {
+ return _protocolVersion;
+ }
+
+ public FrameCreatingMethodProcessor getMethodProcessor()
+ {
+ return _methodProcessor;
+ }
+
+
}
Modified: qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueBindBody.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueBindBody.java?rev=1629270&r1=1629269&r2=1629270&view=diff
==============================================================================
--- qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueBindBody.java (original)
+++ qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueBindBody.java Fri Oct 3 17:40:59 2014
@@ -50,12 +50,12 @@ public class QueueBindBody extends AMQMe
// Constructor
public QueueBindBody(MarkableDataInput buffer) throws AMQFrameDecodingException, IOException
{
- _ticket = readUnsignedShort( buffer );
- _queue = readAMQShortString( buffer );
- _exchange = readAMQShortString( buffer );
- _routingKey = readAMQShortString( buffer );
- _bitfield0 = readBitfield( buffer );
- _arguments = readFieldTable( buffer );
+ _ticket = buffer.readUnsignedShort();
+ _queue = buffer.readAMQShortString();
+ _exchange = buffer.readAMQShortString();
+ _routingKey = buffer.readAMQShortString();
+ _bitfield0 = buffer.readByte();
+ _arguments = EncodingUtils.readFieldTable(buffer);
}
public QueueBindBody(
@@ -165,4 +165,17 @@ public class QueueBindBody extends AMQMe
return buf.toString();
}
+ public static <T> T process(final int channelId,
+ final MarkableDataInput buffer,
+ final MethodProcessor<T> dispatcher) throws IOException, AMQFrameDecodingException
+ {
+
+ int ticket = buffer.readUnsignedShort();
+ AMQShortString queue = buffer.readAMQShortString();
+ AMQShortString exchange = buffer.readAMQShortString();
+ AMQShortString bindingKey = buffer.readAMQShortString();
+ boolean nowait = (buffer.readByte() & 0x01) == 0x01;
+ FieldTable arguments = EncodingUtils.readFieldTable(buffer);
+ return dispatcher.queueBind(channelId, queue, exchange, bindingKey, nowait, arguments);
+ }
}
Modified: qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueDeclareBody.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueDeclareBody.java?rev=1629270&r1=1629269&r2=1629270&view=diff
==============================================================================
--- qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueDeclareBody.java (original)
+++ qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueDeclareBody.java Fri Oct 3 17:40:59 2014
@@ -48,10 +48,10 @@ public class QueueDeclareBody extends AM
// Constructor
public QueueDeclareBody(MarkableDataInput buffer) throws AMQFrameDecodingException, IOException
{
- _ticket = readUnsignedShort( buffer );
- _queue = readAMQShortString( buffer );
- _bitfield0 = readBitfield( buffer );
- _arguments = readFieldTable( buffer );
+ _ticket = buffer.readUnsignedShort();
+ _queue = buffer.readAMQShortString();
+ _bitfield0 = buffer.readByte();
+ _arguments = EncodingUtils.readFieldTable(buffer);
}
public QueueDeclareBody(
@@ -191,4 +191,21 @@ public class QueueDeclareBody extends AM
return buf.toString();
}
+ public static <T> T process(final int channelId,
+ final MarkableDataInput buffer,
+ final MethodProcessor<T> dispatcher) throws IOException, AMQFrameDecodingException
+ {
+
+ int ticket = buffer.readUnsignedShort();
+ AMQShortString queue = buffer.readAMQShortString();
+ byte bitfield = buffer.readByte();
+
+ boolean passive = (bitfield & 0x01 ) == 0x01;
+ boolean durable = (bitfield & 0x02 ) == 0x02;
+ boolean exclusive = (bitfield & 0x04 ) == 0x04;
+ boolean autoDelete = (bitfield & 0x08 ) == 0x08;
+ boolean nowait = (bitfield & 0x010 ) == 0x010;
+ FieldTable arguments = EncodingUtils.readFieldTable(buffer);
+ return dispatcher.queueDeclare(channelId, queue, passive, durable, exclusive, autoDelete, nowait, arguments);
+ }
}
Modified: qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueDeclareOkBody.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueDeclareOkBody.java?rev=1629270&r1=1629269&r2=1629270&view=diff
==============================================================================
--- qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueDeclareOkBody.java (original)
+++ qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueDeclareOkBody.java Fri Oct 3 17:40:59 2014
@@ -47,9 +47,9 @@ public class QueueDeclareOkBody extends
// Constructor
public QueueDeclareOkBody(MarkableDataInput buffer) throws AMQFrameDecodingException, IOException
{
- _queue = readAMQShortString( buffer );
- _messageCount = readUnsignedInteger( buffer );
- _consumerCount = readUnsignedInteger( buffer );
+ _queue = buffer.readAMQShortString();
+ _messageCount = EncodingUtils.readUnsignedInteger(buffer);
+ _consumerCount = EncodingUtils.readUnsignedInteger(buffer);
}
public QueueDeclareOkBody(
@@ -120,4 +120,13 @@ public class QueueDeclareOkBody extends
return buf.toString();
}
+ public static <T> T process(final int channelId,
+ final MarkableDataInput buffer,
+ final MethodProcessor<T> dispatcher) throws IOException
+ {
+ AMQShortString queue = buffer.readAMQShortString();
+ long messageCount = EncodingUtils.readUnsignedInteger(buffer);
+ long consumerCount = EncodingUtils.readUnsignedInteger(buffer);
+ return dispatcher.queueDeclareOk(channelId, queue, messageCount, consumerCount);
+ }
}
Modified: qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueDeleteBody.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueDeleteBody.java?rev=1629270&r1=1629269&r2=1629270&view=diff
==============================================================================
--- qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueDeleteBody.java (original)
+++ qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueDeleteBody.java Fri Oct 3 17:40:59 2014
@@ -47,9 +47,9 @@ public class QueueDeleteBody extends AMQ
// Constructor
public QueueDeleteBody(MarkableDataInput buffer) throws AMQFrameDecodingException, IOException
{
- _ticket = readUnsignedShort( buffer );
- _queue = readAMQShortString( buffer );
- _bitfield0 = readBitfield( buffer );
+ _ticket = buffer.readUnsignedShort();
+ _queue = buffer.readAMQShortString();
+ _bitfield0 = buffer.readByte();
}
public QueueDeleteBody(
@@ -151,4 +151,18 @@ public class QueueDeleteBody extends AMQ
return buf.toString();
}
+ public static <T> T process(final int channelId,
+ final MarkableDataInput buffer,
+ final MethodProcessor<T> dispatcher) throws IOException
+ {
+
+ int ticket = buffer.readUnsignedShort();
+ AMQShortString queue = buffer.readAMQShortString();
+ byte bitfield = buffer.readByte();
+
+ boolean ifUnused = (bitfield & 0x01) == 0x01;
+ boolean ifEmpty = (bitfield & 0x02) == 0x02;
+ boolean nowait = (bitfield & 0x04) == 0x04;
+ return dispatcher.queueDelete(channelId, queue, ifUnused, ifEmpty, nowait);
+ }
}
Modified: qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueDeleteOkBody.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueDeleteOkBody.java?rev=1629270&r1=1629269&r2=1629270&view=diff
==============================================================================
--- qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueDeleteOkBody.java (original)
+++ qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueDeleteOkBody.java Fri Oct 3 17:40:59 2014
@@ -45,7 +45,7 @@ public class QueueDeleteOkBody extends A
// Constructor
public QueueDeleteOkBody(MarkableDataInput buffer) throws AMQFrameDecodingException, IOException
{
- _messageCount = readUnsignedInteger( buffer );
+ _messageCount = EncodingUtils.readUnsignedInteger(buffer);
}
public QueueDeleteOkBody(
@@ -95,4 +95,11 @@ public class QueueDeleteOkBody extends A
return buf.toString();
}
+ public static <T> T process(final int channelId,
+ final MarkableDataInput buffer,
+ final MethodProcessor<T> dispatcher) throws IOException
+ {
+ long messageCount = EncodingUtils.readUnsignedInteger(buffer);
+ return dispatcher.queueDeleteOk(channelId, messageCount);
+ }
}
Modified: qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueuePurgeBody.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueuePurgeBody.java?rev=1629270&r1=1629269&r2=1629270&view=diff
==============================================================================
--- qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueuePurgeBody.java (original)
+++ qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueuePurgeBody.java Fri Oct 3 17:40:59 2014
@@ -47,9 +47,9 @@ public class QueuePurgeBody extends AMQM
// Constructor
public QueuePurgeBody(MarkableDataInput buffer) throws AMQFrameDecodingException, IOException
{
- _ticket = readUnsignedShort( buffer );
- _queue = readAMQShortString( buffer );
- _bitfield0 = readBitfield( buffer );
+ _ticket = buffer.readUnsignedShort();
+ _queue = buffer.readAMQShortString();
+ _bitfield0 = buffer.readByte();
}
public QueuePurgeBody(
@@ -125,4 +125,14 @@ public class QueuePurgeBody extends AMQM
return buf.toString();
}
+ public static <T> T process(final int channelId,
+ final MarkableDataInput buffer,
+ final MethodProcessor<T> dispatcher) throws IOException
+ {
+
+ int ticket = buffer.readUnsignedShort();
+ AMQShortString queue = buffer.readAMQShortString();
+ boolean nowait = (buffer.readByte() & 0x01) == 0x01;
+ return dispatcher.queuePurge(channelId, queue, nowait);
+ }
}
Modified: qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueuePurgeOkBody.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueuePurgeOkBody.java?rev=1629270&r1=1629269&r2=1629270&view=diff
==============================================================================
--- qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueuePurgeOkBody.java (original)
+++ qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueuePurgeOkBody.java Fri Oct 3 17:40:59 2014
@@ -45,7 +45,7 @@ public class QueuePurgeOkBody extends AM
// Constructor
public QueuePurgeOkBody(MarkableDataInput buffer) throws AMQFrameDecodingException, IOException
{
- _messageCount = readUnsignedInteger( buffer );
+ _messageCount = EncodingUtils.readUnsignedInteger(buffer);
}
public QueuePurgeOkBody(
@@ -95,4 +95,11 @@ public class QueuePurgeOkBody extends AM
return buf.toString();
}
+ public static <T> T process(final int channelId,
+ final MarkableDataInput buffer,
+ final MethodProcessor<T> dispatcher) throws IOException
+ {
+ long messageCount = EncodingUtils.readUnsignedInteger(buffer);
+ return dispatcher.queuePurgeOk(channelId, messageCount);
+ }
}
Modified: qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueUnbindBody.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueUnbindBody.java?rev=1629270&r1=1629269&r2=1629270&view=diff
==============================================================================
--- qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueUnbindBody.java (original)
+++ qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueUnbindBody.java Fri Oct 3 17:40:59 2014
@@ -49,11 +49,11 @@ public class QueueUnbindBody extends AMQ
// Constructor
public QueueUnbindBody(MarkableDataInput buffer) throws AMQFrameDecodingException, IOException
{
- _ticket = readUnsignedShort( buffer );
- _queue = readAMQShortString( buffer );
- _exchange = readAMQShortString( buffer );
- _routingKey = readAMQShortString( buffer );
- _arguments = readFieldTable( buffer );
+ _ticket = buffer.readUnsignedShort();
+ _queue = buffer.readAMQShortString();
+ _exchange = buffer.readAMQShortString();
+ _routingKey = buffer.readAMQShortString();
+ _arguments = EncodingUtils.readFieldTable(buffer);
}
public QueueUnbindBody(
@@ -147,4 +147,16 @@ public class QueueUnbindBody extends AMQ
return buf.toString();
}
+ public static <T> T process(final int channelId,
+ final MarkableDataInput buffer,
+ final MethodProcessor<T> dispatcher) throws IOException, AMQFrameDecodingException
+ {
+
+ int ticket = buffer.readUnsignedShort();
+ AMQShortString queue = buffer.readAMQShortString();
+ AMQShortString exchange = buffer.readAMQShortString();
+ AMQShortString routingKey = buffer.readAMQShortString();
+ FieldTable arguments = EncodingUtils.readFieldTable(buffer);
+ return dispatcher.queueUnbind(channelId, queue, exchange, routingKey, arguments);
+ }
}
Modified: qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueUnbindOkBody.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueUnbindOkBody.java?rev=1629270&r1=1629269&r2=1629270&view=diff
==============================================================================
--- qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueUnbindOkBody.java (original)
+++ qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/main/java/org/apache/qpid/framing/QueueUnbindOkBody.java Fri Oct 3 17:40:59 2014
@@ -27,11 +27,11 @@
package org.apache.qpid.framing;
-import org.apache.qpid.codec.MarkableDataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.qpid.AMQException;
+import org.apache.qpid.codec.MarkableDataInput;
public class QueueUnbindOkBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
Modified: qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/test/java/org/apache/qpid/codec/AMQDecoderTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/test/java/org/apache/qpid/codec/AMQDecoderTest.java?rev=1629270&r1=1629269&r2=1629270&view=diff
==============================================================================
--- qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/test/java/org/apache/qpid/codec/AMQDecoderTest.java (original)
+++ qpid/branches/QPID-6125-ProtocolRefactoring/java/common/src/test/java/org/apache/qpid/codec/AMQDecoderTest.java Fri Oct 3 17:40:59 2014
@@ -34,6 +34,8 @@ import org.apache.qpid.framing.AMQFrame;
import org.apache.qpid.framing.AMQFrameDecodingException;
import org.apache.qpid.framing.AMQProtocolVersionException;
import org.apache.qpid.framing.HeartbeatBody;
+import org.apache.qpid.framing.MethodRegistry;
+import org.apache.qpid.framing.ProtocolVersion;
public class AMQDecoderTest extends TestCase
{
@@ -43,7 +45,7 @@ public class AMQDecoderTest extends Test
public void setUp()
{
- _decoder = new AMQDecoder(false, null);
+ _decoder = new AMQDecoder(false, new MethodRegistry(ProtocolVersion.v0_91));
}
Modified: qpid/branches/QPID-6125-ProtocolRefactoring/java/systests/src/test/java/org/apache/qpid/server/logging/ExchangeLoggingTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6125-ProtocolRefactoring/java/systests/src/test/java/org/apache/qpid/server/logging/ExchangeLoggingTest.java?rev=1629270&r1=1629269&r2=1629270&view=diff
==============================================================================
--- qpid/branches/QPID-6125-ProtocolRefactoring/java/systests/src/test/java/org/apache/qpid/server/logging/ExchangeLoggingTest.java (original)
+++ qpid/branches/QPID-6125-ProtocolRefactoring/java/systests/src/test/java/org/apache/qpid/server/logging/ExchangeLoggingTest.java Fri Oct 3 17:40:59 2014
@@ -38,6 +38,7 @@ import org.apache.qpid.framing.AMQShortS
import org.apache.qpid.framing.ExchangeDeleteBody;
import org.apache.qpid.framing.ExchangeDeleteOkBody;
import org.apache.qpid.framing.MethodRegistry;
+import org.apache.qpid.framing.ProtocolVersion;
/**
* Exchange
@@ -192,7 +193,7 @@ public class ExchangeLoggingTest extends
}
else
{
- MethodRegistry registry = MethodRegistry.registry_8_0;
+ MethodRegistry registry = new MethodRegistry(ProtocolVersion.v8_0);
ExchangeDeleteBody body = registry.createExchangeDeleteBody(0, new AMQShortString(_name), false, true);
Modified: qpid/branches/QPID-6125-ProtocolRefactoring/java/systests/src/test/java/org/apache/qpid/server/store/VirtualHostMessageStoreTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6125-ProtocolRefactoring/java/systests/src/test/java/org/apache/qpid/server/store/VirtualHostMessageStoreTest.java?rev=1629270&r1=1629269&r2=1629270&view=diff
==============================================================================
--- qpid/branches/QPID-6125-ProtocolRefactoring/java/systests/src/test/java/org/apache/qpid/server/store/VirtualHostMessageStoreTest.java (original)
+++ qpid/branches/QPID-6125-ProtocolRefactoring/java/systests/src/test/java/org/apache/qpid/server/store/VirtualHostMessageStoreTest.java Fri Oct 3 17:40:59 2014
@@ -38,7 +38,6 @@ import org.codehaus.jackson.map.ObjectMa
import org.apache.qpid.common.AMQPFilterTypes;
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.BasicConsumeBody;
import org.apache.qpid.framing.BasicContentHeaderProperties;
import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.FieldTable;
@@ -599,7 +598,7 @@ public class VirtualHostMessageStoreTest
MessagePublishInfo messageInfo = new MessagePublishInfo(new AMQShortString(exchange.getName()), false, false, new AMQShortString(routingKey));
- ContentHeaderBody headerBody = new ContentHeaderBody(BasicConsumeBody.CLASS_ID,0,properties,0l);
+ ContentHeaderBody headerBody = new ContentHeaderBody(properties,0l);
MessageMetaData mmd = new MessageMetaData(messageInfo, headerBody, System.currentTimeMillis());
Modified: qpid/branches/QPID-6125-ProtocolRefactoring/java/systests/src/test/java/org/apache/qpid/transport/MaxFrameSizeTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6125-ProtocolRefactoring/java/systests/src/test/java/org/apache/qpid/transport/MaxFrameSizeTest.java?rev=1629270&r1=1629269&r2=1629270&view=diff
==============================================================================
--- qpid/branches/QPID-6125-ProtocolRefactoring/java/systests/src/test/java/org/apache/qpid/transport/MaxFrameSizeTest.java (original)
+++ qpid/branches/QPID-6125-ProtocolRefactoring/java/systests/src/test/java/org/apache/qpid/transport/MaxFrameSizeTest.java Fri Oct 3 17:40:59 2014
@@ -41,20 +41,18 @@ import javax.security.sasl.Sasl;
import javax.security.sasl.SaslClient;
import javax.security.sasl.SaslException;
-import org.apache.qpid.codec.MarkableDataInput;
-import org.apache.qpid.framing.AMQBody;
import org.apache.qpid.framing.AMQDataBlockDecoder;
import org.apache.qpid.framing.AMQFrame;
import org.apache.qpid.framing.AMQFrameDecodingException;
import org.apache.qpid.framing.AMQProtocolVersionException;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.BodyFactory;
import org.apache.qpid.framing.ByteArrayDataInput;
import org.apache.qpid.framing.ConnectionCloseBody;
import org.apache.qpid.framing.ConnectionStartOkBody;
import org.apache.qpid.framing.ConnectionTuneOkBody;
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.framing.MethodRegistry;
+import org.apache.qpid.framing.ProtocolVersion;
import org.apache.qpid.jms.BrokerDetails;
import org.apache.qpid.server.model.AuthenticationProvider;
import org.apache.qpid.server.model.Broker;
@@ -238,21 +236,14 @@ public class MaxFrameSizeTest extends Qp
byte[] serverData = baos.toByteArray();
ByteArrayDataInput badi = new ByteArrayDataInput(serverData);
AMQDataBlockDecoder datablockDecoder = new AMQDataBlockDecoder();
- final MethodRegistry methodRegistry_0_91 = MethodRegistry.registry_0_91;
- BodyFactory methodBodyFactory = new BodyFactory()
- {
- @Override
- public AMQBody createBody(final MarkableDataInput in, final long bodySize)
- throws AMQFrameDecodingException, IOException
- {
- return methodRegistry_0_91.convertToBody(in, bodySize);
- }
- };
+ final MethodRegistry methodRegistry_0_91 = new MethodRegistry(ProtocolVersion.v0_91);
List<AMQFrame> frames = new ArrayList<>();
while (datablockDecoder.decodable(badi))
{
- frames.add(datablockDecoder.createAndPopulateFrame(methodBodyFactory, badi));
+ frames.add(datablockDecoder.createAndPopulateFrame(methodRegistry_0_91.getProtocolVersion(),
+ methodRegistry_0_91.getMethodProcessor(),
+ badi));
}
evaluator.evaluate(socket, frames);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org