You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kp...@apache.org on 2007/01/18 22:39:30 UTC
svn commit: r497585 - in /incubator/qpid/branches/qpid.0-9:
gentools/src/org/apache/qpid/gentools/
java/broker/src/main/java/org/apache/qpid/server/
java/broker/src/main/java/org/apache/qpid/server/handler/
java/broker/src/main/java/org/apache/qpid/ser...
Author: kpvdr
Date: Thu Jan 18 13:39:29 2007
New Revision: 497585
URL: http://svn.apache.org/viewvc?view=rev&rev=497585
Log:
Changed Content to use ByteBuffer, added Message.Transfer and Message.Cancel handlers
Modified:
incubator/qpid/branches/qpid.0-9/gentools/src/org/apache/qpid/gentools/JavaGenerator.java
incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/handler/MessageCancelHandler.java
incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/handler/MessageTransferHandler.java
incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
incubator/qpid/branches/qpid.0-9/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
incubator/qpid/branches/qpid.0-9/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java
incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/Content.java
Modified: incubator/qpid/branches/qpid.0-9/gentools/src/org/apache/qpid/gentools/JavaGenerator.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/gentools/src/org/apache/qpid/gentools/JavaGenerator.java?view=diff&rev=497585&r1=497584&r2=497585
==============================================================================
--- incubator/qpid/branches/qpid.0-9/gentools/src/org/apache/qpid/gentools/JavaGenerator.java (original)
+++ incubator/qpid/branches/qpid.0-9/gentools/src/org/apache/qpid/gentools/JavaGenerator.java Thu Jan 18 13:39:29 2007
@@ -1054,8 +1054,16 @@
int ordinal, int indentSize, int tabSize)
{
StringBuffer sb = new StringBuffer();
- sb.append(Utils.createSpaces(indentSize) +
- "buf.append(\" " + fieldName + ": \" + " + fieldName + ");" + cr);
+ if (domain.compareTo("longstr") == 0)
+ {
+ sb.append(Utils.createSpaces(indentSize) +
+ "buf.append(\" " + fieldName + ": \" + new String(" + fieldName + "));" + cr);
+ }
+ else
+ {
+ sb.append(Utils.createSpaces(indentSize) +
+ "buf.append(\" " + fieldName + ": \" + " + fieldName + ");" + cr);
+ }
return sb.toString();
}
Modified: incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java?view=diff&rev=497585&r1=497584&r2=497585
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java (original)
+++ incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java Thu Jan 18 13:39:29 2007
@@ -197,7 +197,7 @@
route(message);
break;
case CONTENT_TYPE_REFERENCE:
- getMessages(body.getContent()).add(message);
+ getMessages(body.getContentAsByteArray()).add(message);
break;
}
}
Modified: incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/handler/MessageCancelHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/handler/MessageCancelHandler.java?view=diff&rev=497585&r1=497584&r2=497585
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/handler/MessageCancelHandler.java (original)
+++ incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/handler/MessageCancelHandler.java Thu Jan 18 13:39:29 2007
@@ -21,8 +21,11 @@
package org.apache.qpid.server.handler;
import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQMethodBody;
import org.apache.qpid.framing.MessageCancelBody;
+import org.apache.qpid.framing.MessageOkBody;
import org.apache.qpid.protocol.AMQMethodEvent;
+import org.apache.qpid.server.AMQChannel;
import org.apache.qpid.server.exchange.ExchangeRegistry;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.queue.QueueRegistry;
@@ -48,7 +51,15 @@
AMQMethodEvent<MessageCancelBody> evt)
throws AMQException
{
- // TODO
+ final AMQChannel channel = protocolSession.getChannel(evt.getChannelId());
+ final MessageCancelBody body = evt.getMethod();
+ channel.unsubscribeConsumer(protocolSession, body.destination);
+
+ // AMQP version change: Hardwire the version to 0-9 (major=0, minor=9)
+ // TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
+ // Be aware of possible changes to parameter order as versions change.
+ final AMQMethodBody methodBody = MessageOkBody.createMethodBody((byte)0, (byte)9);
+ protocolSession.writeResponse(evt, methodBody);
}
}
Modified: incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/handler/MessageTransferHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/handler/MessageTransferHandler.java?view=diff&rev=497585&r1=497584&r2=497585
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/handler/MessageTransferHandler.java (original)
+++ incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/handler/MessageTransferHandler.java Thu Jan 18 13:39:29 2007
@@ -74,11 +74,11 @@
protocolSession.closeChannel(evt.getChannelId());
// TODO: modify code gen to make getClazz and getMethod public methods rather than protected
// then we can remove the hardcoded 0,0
- // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
+ // AMQP version change: Hardwire the version to 0-9 (major=0, minor=9)
// TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
// Be aware of possible changes to parameter order as versions change.
AMQMethodBody cf = ChannelCloseBody.createMethodBody
- ((byte)8, (byte)0, // AMQP version (major, minor)
+ ((byte)0, (byte)9, // AMQP version (major, minor)
MessageTransferBody.getClazz((byte)0, (byte)9), // classId
MessageTransferBody.getMethod((byte)0, (byte)9), // methodId
500, // replyCode
Modified: incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java?view=diff&rev=497585&r1=497584&r2=497585
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java (original)
+++ incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java Thu Jan 18 13:39:29 2007
@@ -150,7 +150,7 @@
Content body = _transferBody.getBody();
switch (body.getContentType()) {
case CONTENT_TYPE_INLINE:
- return _transferBody.getBody().getContent().length;
+ return _transferBody.getBody().getContent().limit();
case CONTENT_TYPE_REFERENCE:
return getReferenceSize();
default:
Modified: incubator/qpid/branches/qpid.0-9/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java?view=diff&rev=497585&r1=497584&r2=497585
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java (original)
+++ incubator/qpid/branches/qpid.0-9/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java Thu Jan 18 13:39:29 2007
@@ -557,8 +557,6 @@
messageHeaders.setExpiration(0);
}
}
-// messageHeaders.setDeliveryMode((byte) deliveryMode);
-// messageHeaders.setPriority((byte) priority);
int size = (payload != null) ? payload.limit() : 0;
Content[] content = createContent(payload);
@@ -656,7 +654,7 @@
if (frameCount == 1)
{
- bodies[0] = new Content(Content.ContentTypeEnum.CONTENT_TYPE_INLINE, payload.array());
+ bodies[0] = new Content(Content.ContentTypeEnum.CONTENT_TYPE_INLINE, payload);
}
else
{
@@ -666,7 +664,7 @@
payload.position((int) framePayloadMax * i);
int length = (remaining >= framePayloadMax) ? (int) framePayloadMax : (int) remaining;
payload.limit(payload.position() + length);
- bodies[i] = new Content(Content.ContentTypeEnum.CONTENT_TYPE_INLINE, payload.slice().array());
+ bodies[i] = new Content(Content.ContentTypeEnum.CONTENT_TYPE_INLINE, payload.slice());
remaining -= length;
}
}
Modified: incubator/qpid/branches/qpid.0-9/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java?view=diff&rev=497585&r1=497584&r2=497585
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java (original)
+++ incubator/qpid/branches/qpid.0-9/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java Thu Jan 18 13:39:29 2007
@@ -41,7 +41,7 @@
MessageHeaders contentHeader, Content body) throws AMQException {
ByteBuffer data;
- data = ByteBuffer.allocate(body.content.length);
+ data = ByteBuffer.allocate(body.content.remaining());
data.put(body.content);
data.flip();
Modified: incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/Content.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/Content.java?view=diff&rev=497585&r1=497584&r2=497585
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/Content.java (original)
+++ incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/Content.java Thu Jan 18 13:39:29 2007
@@ -43,7 +43,7 @@
}
public ContentTypeEnum contentType;
- public byte[] content;
+ public ByteBuffer content;
// Constructors
@@ -63,20 +63,13 @@
throw new IllegalArgumentException("Content cannot be empty for a ref type.");
}
this.contentType = contentType;
- this.content = content;
+ this.content = ByteBuffer.allocate(content.length);
+ this.content.put(content);
}
- public Content(ContentTypeEnum contentType, String content)
+ public Content(ContentTypeEnum contentType, String contentStr)
{
- if (contentType == ContentTypeEnum.CONTENT_TYPE_REFERENCE)
- {
- if (content == null)
- throw new IllegalArgumentException("Content cannot be null for a ref type.");
- if (content.length() == 0)
- throw new IllegalArgumentException("Content cannot be empty for a ref type.");
- }
- this.contentType = contentType;
- this.content = content.getBytes();
+ this(contentType, contentStr.getBytes());
}
public Content(ContentTypeEnum contentType, ByteBuffer content)
@@ -89,18 +82,26 @@
throw new IllegalArgumentException("Content cannot be empty for a ref type.");
}
this.contentType = contentType;
- this.content = content.array();
+ this.content = content;
}
// Get functions
public ContentTypeEnum getContentType() { return contentType; }
- public byte[] getContent() { return content; }
+ public ByteBuffer getContent() { return content; }
+
+ public byte[] getContentAsByteArray()
+ {
+ byte[] ba = new byte[content.remaining()];
+ content.get(ba);
+ return ba;
+ }
+
public String getContentAsString()
{
if (content == null)
return null;
- return new String(content);
+ return new String(getContentAsByteArray());
}
// Wire functions
@@ -109,18 +110,22 @@
{
if (content == null)
return 1 + 4;
- return 1 + 4 + content.length;
+ return 1 + 4 + content.remaining();
}
public void writePayload(ByteBuffer buffer)
{
EncodingUtils.writeUnsignedByte(buffer, contentType.toByte());
- EncodingUtils.writeLongStringBytes(buffer, content);
+ EncodingUtils.writeUnsignedInteger(buffer, content.remaining());
+ buffer.put(content);
}
public void populateFromBuffer(ByteBuffer buffer) throws AMQFrameDecodingException
{
contentType = ContentTypeEnum.toContentEnum(buffer.get());
- content = EncodingUtils.readLongstr(buffer);
+ int length = buffer.getInt();
+ content = buffer.slice();
+ buffer.skip(length);
+ content.limit(length);
}
}