Added: incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/AMQUnexpectedBodyTypeException.java URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/AMQUnexpectedBodyTypeException.java?view=auto&rev=511389 ============================================================================== --- incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/AMQUnexpectedBodyTypeException.java (added) +++ incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/AMQUnexpectedBodyTypeException.java Sat Feb 24 17:08:57 2007 @@ -0,0 +1,33 @@ +/* + * + * 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.server.cluster; + +import org.apache.qpid.AMQException; +import org.apache.qpid.framing.AMQBody; + +public class AMQUnexpectedBodyTypeException extends AMQException +{ + + public AMQUnexpectedBodyTypeException(Class<? extends AMQBody> expectedClass, AMQBody body) + { + super("Unexpected body type. Expected: " + expectedClass.getName() + "; got: " + body.getClass().getName()); + } +}
Added: incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/AMQUnexpectedFrameTypeException.java URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/AMQUnexpectedFrameTypeException.java?view=auto&rev=511389 ============================================================================== --- incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/AMQUnexpectedFrameTypeException.java (added) +++ incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/AMQUnexpectedFrameTypeException.java Sat Feb 24 17:08:57 2007 @@ -0,0 +1,31 @@ +/* + * + * 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.server.cluster; + +import org.apache.qpid.AMQException; + +public class AMQUnexpectedFrameTypeException extends AMQException +{ + public AMQUnexpectedFrameTypeException(String s) + { + super(s); + } +} Modified: incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/MinaBrokerProxy.java URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/MinaBrokerProxy.java?view=diff&rev=511389&r1=511388&r2=511389 ============================================================================== --- incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/MinaBrokerProxy.java (original) +++ incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/MinaBrokerProxy.java Sat Feb 24 17:08:57 2007 @@ -39,7 +39,7 @@ import org.apache.qpid.framing.AMQMethodBody; import org.apache.qpid.framing.ConnectionRedirectBody; import org.apache.qpid.framing.ProtocolInitiation; -import org.apache.qpid.framing.ProtocolVersionList; +import org.apache.qpid.framing.ProtocolVersion; import java.io.IOException; import java.net.InetSocketAddress; @@ -138,7 +138,7 @@ } } - public void send(AMQDataBlock data) throws AMQException + public void send(AMQDataBlock data) throws AMQConnectionWaitException { if (_session == null) { @@ -146,9 +146,9 @@ { _connectionMonitor.waitUntilOpen(); } - catch (Exception e) + catch (InterruptedException e) { - throw new AMQException("Failed to send " + data + ": " + e, e); + throw new AMQConnectionWaitException("Failed to send " + data + ": " + e, e); } } _session.write(data); @@ -207,7 +207,7 @@ } else { - throw new AMQException("Client only expects method body, got: " + body); + throw new AMQUnexpectedBodyTypeException(AMQMethodBody.class, body); } } @@ -216,7 +216,7 @@ return "MinaBrokerProxy[" + (_session == null ? super.toString() : _session.getRemoteAddress()) + "]"; } - private class MinaBinding extends IoHandlerAdapter implements ProtocolVersionList + private class MinaBinding extends IoHandlerAdapter { public void sessionCreated(IoSession session) throws Exception { @@ -228,8 +228,8 @@ /* Find last protocol version in protocol version list. Make sure last protocol version listed in the build file (build-module.xml) is the latest version which will be used here. */ - int i = pv.length - 1; - session.write(new ProtocolInitiation(pv[i][PROTOCOL_MAJOR], pv[i][PROTOCOL_MINOR])); + + session.write(new ProtocolInitiation(ProtocolVersion.getLatestSupportedVersion())); } public void sessionOpened(IoSession session) throws Exception @@ -260,7 +260,7 @@ } else { - throw new AMQException("Received message of unrecognised type: " + object); + throw new AMQUnexpectedFrameTypeException("Received message of unrecognised type: " + object); } } Modified: incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQMethodBody.java URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQMethodBody.java?view=diff&rev=511389&r1=511388&r2=511389 ============================================================================== --- incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQMethodBody.java (original) +++ incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQMethodBody.java Sat Feb 24 17:08:57 2007 @@ -86,9 +86,9 @@ public String toString() { - StringBuffer buf = new StringBuffer(getClass().toString()); - buf.append(" Class: ").append(getClazz()); - buf.append(" Method: ").append(getMethod()); + StringBuffer buf = new StringBuffer(getClass().getName()); + buf.append("[ Class: ").append(getClazz()); + buf.append(" Method: ").append(getMethod()).append(']'); return buf.toString(); } Modified: incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/BasicContentHeaderProperties.java URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/BasicContentHeaderProperties.java?view=diff&rev=511389&r1=511388&r2=511389 ============================================================================== --- incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/BasicContentHeaderProperties.java (original) +++ incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/BasicContentHeaderProperties.java Sat Feb 24 17:08:57 2007 @@ -24,7 +24,7 @@ import org.apache.mina.common.ByteBuffer; -public class BasicContentHeaderProperties implements ContentHeaderProperties +public class BasicContentHeaderProperties implements CommonContentHeaderProperties { private static final Logger _logger = Logger.getLogger(BasicContentHeaderProperties.class); @@ -421,14 +421,14 @@ } } - public AMQShortString getContentTypeShortString() + public AMQShortString getContentType() { decodeContentTypeIfNecessary(); return _contentType; } - public String getContentType() + public String getContentTypeAsString() { decodeContentTypeIfNecessary(); return _contentType == null ? null : _contentType.toString(); @@ -444,15 +444,19 @@ public void setContentType(String contentType) { - clearEncodedForm(); - _propertyFlags |= (1 << 15); - _contentType = contentType == null ? null : new AMQShortString(contentType); + setContentType(contentType == null ? null : new AMQShortString(contentType)); + } + + public String getEncodingAsString() + { + + return getEncoding() == null ? null : getEncoding().toString(); } - public String getEncoding() + public AMQShortString getEncoding() { decodeIfNecessary(); - return _encoding == null ? null : _encoding.toString(); + return _encoding; } public void setEncoding(String encoding) @@ -462,6 +466,14 @@ _encoding = encoding == null ? null : new AMQShortString(encoding); } + public void setEncoding(AMQShortString encoding) + { + clearEncodedForm(); + _propertyFlags |= (1 << 14); + _encoding = encoding; + } + + public FieldTable getHeaders() { decodeHeadersIfNecessary(); @@ -508,7 +520,13 @@ _priority = priority; } - public String getCorrelationId() + public AMQShortString getCorrelationId() + { + decodeIfNecessary(); + return _correlationId; + } + + public String getCorrelationIdAsString() { decodeIfNecessary(); return _correlationId == null ? null : _correlationId.toString(); @@ -516,18 +534,23 @@ public void setCorrelationId(String correlationId) { + setCorrelationId(correlationId == null ? null : new AMQShortString(correlationId)); + } + + public void setCorrelationId(AMQShortString correlationId) + { clearEncodedForm(); _propertyFlags |= (1 << 10); - _correlationId = correlationId == null ? null : new AMQShortString(correlationId); + _correlationId = correlationId; } - public String getReplyTo() + public String getReplyToAsString() { decodeIfNecessary(); return _replyTo == null ? null : _replyTo.toString(); } - public AMQShortString getReplyToAsShortString() + public AMQShortString getReplyTo() { decodeIfNecessary(); return _replyTo; @@ -561,7 +584,13 @@ } - public String getMessageId() + public AMQShortString getMessageId() + { + decodeIfNecessary(); + return _messageId; + } + + public String getMessageIdAsString() { decodeIfNecessary(); return _messageId == null ? null : _messageId.toString(); @@ -574,6 +603,14 @@ _messageId = messageId == null ? null : new AMQShortString(messageId); } + public void setMessageId(AMQShortString messageId) + { + clearEncodedForm(); + _propertyFlags |= (1 << 7); + _messageId = messageId; + } + + public long getTimestamp() { decodeIfNecessary(); @@ -587,56 +624,102 @@ _timestamp = timestamp; } - public String getType() + public String getTypeAsString() { decodeIfNecessary(); return _type == null ? null : _type.toString(); } + + public AMQShortString getType() + { + decodeIfNecessary(); + return _type; + } + + public void setType(String type) { + setType(type == null ? null : new AMQShortString(type)); + } + + public void setType(AMQShortString type) + { clearEncodedForm(); _propertyFlags |= (1 << 5); - _type = type == null ? null : new AMQShortString(type); + _type = type; } - public String getUserId() + public String getUserIdAsString() { decodeIfNecessary(); return _userId == null ? null : _userId.toString(); } + public AMQShortString getUserId() + { + decodeIfNecessary(); + return _userId; + } + public void setUserId(String userId) { + setUserId(userId == null ? null : new AMQShortString(userId)); + } + + public void setUserId(AMQShortString userId) + { clearEncodedForm(); _propertyFlags |= (1 << 4); - _userId = userId == null ? null : new AMQShortString(userId); + _userId = userId; } - public String getAppId() + public String getAppIdAsString() { decodeIfNecessary(); return _appId == null ? null : _appId.toString(); } + public AMQShortString getAppId() + { + decodeIfNecessary(); + return _appId; + } + public void setAppId(String appId) { + setAppId(appId == null ? null : new AMQShortString(appId)); + } + + public void setAppId(AMQShortString appId) + { clearEncodedForm(); _propertyFlags |= (1 << 3); - _appId = appId == null ? null : new AMQShortString(appId); + _appId = appId; } - public String getClusterId() + public String getClusterIdAsString() { decodeIfNecessary(); return _clusterId == null ? null : _clusterId.toString(); } + public AMQShortString getClusterId() + { + decodeIfNecessary(); + return _clusterId; + } + public void setClusterId(String clusterId) { + setClusterId(clusterId == null ? null : new AMQShortString(clusterId)); + } + + public void setClusterId(AMQShortString clusterId) + { clearEncodedForm(); _propertyFlags |= (1 << 2); - _clusterId = clusterId == null ? null : new AMQShortString(clusterId); + _clusterId = clusterId; } public String toString() Added: incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/CommonContentHeaderProperties.java URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/CommonContentHeaderProperties.java?view=auto&rev=511389 ============================================================================== --- incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/CommonContentHeaderProperties.java (added) +++ incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/CommonContentHeaderProperties.java Sat Feb 24 17:08:57 2007 @@ -0,0 +1,65 @@ +package org.apache.qpid.framing; + +import org.apache.mina.common.ByteBuffer; + +import org.apache.log4j.Logger; + +public interface CommonContentHeaderProperties extends ContentHeaderProperties +{ + + AMQShortString getContentType(); + + void setContentType(AMQShortString contentType); + + FieldTable getHeaders(); + + void setHeaders(FieldTable headers); + + byte getDeliveryMode(); + + void setDeliveryMode(byte deliveryMode); + + byte getPriority(); + + void setPriority(byte priority); + + AMQShortString getCorrelationId(); + + void setCorrelationId(AMQShortString correlationId); + + AMQShortString getReplyTo(); + + void setReplyTo(AMQShortString replyTo); + + long getExpiration(); + + void setExpiration(long expiration); + + AMQShortString getMessageId(); + + void setMessageId(AMQShortString messageId); + + long getTimestamp(); + + void setTimestamp(long timestamp); + + AMQShortString getType(); + + void setType(AMQShortString type); + + AMQShortString getUserId(); + + void setUserId(AMQShortString userId); + + AMQShortString getAppId(); + + void setAppId(AMQShortString appId); + + AMQShortString getClusterId(); + + void setClusterId(AMQShortString clusterId); + + AMQShortString getEncoding(); + + void setEncoding(AMQShortString encoding); +} Modified: incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java?view=diff&rev=511389&r1=511388&r2=511389 ============================================================================== --- incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java (original) +++ incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java Sat Feb 24 17:08:57 2007 @@ -41,10 +41,14 @@ private LinkedHashMap<AMQShortString, AMQTypedValue> _properties; private long _encodedSize; private static final int INITIAL_HASHMAP_CAPACITY = 16; + private static final int INITIAL_ENCODED_FORM_SIZE = 256; public FieldTable() { super(); +// _encodedForm = ByteBuffer.allocate(INITIAL_ENCODED_FORM_SIZE); +// _encodedForm.setAutoExpand(true); +// _encodedForm.limit(0); } /** @@ -109,11 +113,28 @@ private AMQTypedValue setProperty(AMQShortString key, AMQTypedValue val) { initMapIfNecessary(); - _encodedForm = null; - if(val == null) + if(_properties.containsKey(key)) + { + _encodedForm = null; + + if(val == null) + { + return removeKey(key); + } + } + else if(_encodedForm != null && val != null) + { + EncodingUtils.writeShortStringBytes(_encodedForm, key); + val.writeToBuffer(_encodedForm); + + } + else if (val == null) { - return removeKey(key); + return null; } + + + AMQTypedValue oldVal = _properties.put(key,val); if(oldVal != null) { @@ -134,7 +155,7 @@ { if(_properties == null) { - if(_encodedForm == null) + if(_encodedForm == null || _encodedSize == 0) { _properties = new LinkedHashMap<AMQShortString,AMQTypedValue>(); } @@ -655,6 +676,7 @@ if (trace) { _logger.trace("FieldTable::writeToBuffer: Writing encoded length of " + getEncodedSize() + "..."); + _logger.trace(_properties); } EncodingUtils.writeUnsignedInteger(buffer, getEncodedSize()); @@ -701,6 +723,7 @@ public void addAll(FieldTable fieldTable) { initMapIfNecessary(); + _encodedForm = null; _properties.putAll(fieldTable._properties); recalculateEncodedSize(); } @@ -836,7 +859,13 @@ if(_encodedForm != null) { - buffer.put(_encodedForm); + + if(_encodedForm.position() != 0) + { + _encodedForm.flip(); + } +// _encodedForm.limit((int)getEncodedSize()); + buffer.put(_encodedForm); } else if(_properties != null) { @@ -924,4 +953,33 @@ } } + public int hashCode() + { + initMapIfNecessary(); + return _properties.hashCode(); + } + + + public boolean equals(Object o) + { + if(o == this) + { + return true; + } + if(o == null) + { + return false; + } + if(!(o instanceof FieldTable)) + { + return false; + } + + initMapIfNecessary(); + + FieldTable f = (FieldTable) o; + f.initMapIfNecessary(); + + return _properties.equals(f._properties); + } } Modified: incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java?view=diff&rev=511389&r1=511388&r2=511389 ============================================================================== --- incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java (original) +++ incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java Sat Feb 24 17:08:57 2007 @@ -25,25 +25,50 @@ import org.apache.mina.filter.codec.ProtocolDecoderOutput; import org.apache.qpid.AMQException; +import java.io.UnsupportedEncodingException; + public class ProtocolInitiation extends AMQDataBlock implements EncodableAMQDataBlock { - public char[] header = new char[]{'A','M','Q','P'}; + // TODO: generate these constants automatically from the xml protocol spec file + public static final byte[] AMQP_HEADER = new byte[]{(byte)'A',(byte)'M',(byte)'Q',(byte)'P'}; - private static byte CURRENT_PROTOCOL_CLASS = 1; - private static final int CURRENT_PROTOCOL_INSTANCE = 1; + private static final byte CURRENT_PROTOCOL_CLASS = 1; + private static final byte TCP_PROTOCOL_INSTANCE = 1; + + public final byte[] _protocolHeader; + public final byte _protocolClass; + public final byte _protocolInstance; + public final byte _protocolMajor; + public final byte _protocolMinor; - public byte protocolClass = CURRENT_PROTOCOL_CLASS; - public byte protocolInstance = CURRENT_PROTOCOL_INSTANCE; - public byte protocolMajor; - public byte protocolMinor; // public ProtocolInitiation() {} - public ProtocolInitiation(byte major, byte minor) + public ProtocolInitiation(byte[] protocolHeader, byte protocolClass, byte protocolInstance, byte protocolMajor, byte protocolMinor) + { + _protocolHeader = protocolHeader; + _protocolClass = protocolClass; + _protocolInstance = protocolInstance; + _protocolMajor = protocolMajor; + _protocolMinor = protocolMinor; + } + + public ProtocolInitiation(ProtocolVersion pv) { - protocolMajor = major; - protocolMinor = minor; + this(AMQP_HEADER, CURRENT_PROTOCOL_CLASS, TCP_PROTOCOL_INSTANCE, pv.getMajorVersion(), pv.getMinorVersion()); + } + + + public ProtocolInitiation(ByteBuffer in) + { + _protocolHeader = new byte[4]; + in.get(_protocolHeader); + + _protocolClass = in.get(); + _protocolInstance = in.get(); + _protocolMajor = in.get(); + _protocolMinor = in.get(); } public long getSize() @@ -53,19 +78,12 @@ public void writePayload(ByteBuffer buffer) { - for (int i = 0; i < header.length; i++) - { - buffer.put((byte) header[i]); - } - buffer.put(protocolClass); - buffer.put(protocolInstance); - buffer.put(protocolMajor); - buffer.put(protocolMinor); - } - public void populateFromBuffer(ByteBuffer buffer) throws AMQException - { - throw new AMQException("Method not implemented"); + buffer.put(_protocolHeader); + buffer.put(_protocolClass); + buffer.put(_protocolInstance); + buffer.put(_protocolMajor); + buffer.put(_protocolMinor); } public boolean equals(Object o) @@ -76,36 +94,36 @@ } ProtocolInitiation pi = (ProtocolInitiation) o; - if (pi.header == null) + if (pi._protocolHeader == null) { return false; } - if (header.length != pi.header.length) + if (_protocolHeader.length != pi._protocolHeader.length) { return false; } - for (int i = 0; i < header.length; i++) + for (int i = 0; i < _protocolHeader.length; i++) { - if (header[i] != pi.header[i]) + if (_protocolHeader[i] != pi._protocolHeader[i]) { return false; } } - return (protocolClass == pi.protocolClass && - protocolInstance == pi.protocolInstance && - protocolMajor == pi.protocolMajor && - protocolMinor == pi.protocolMinor); + return (_protocolClass == pi._protocolClass && + _protocolInstance == pi._protocolInstance && + _protocolMajor == pi._protocolMajor && + _protocolMinor == pi._protocolMinor); } public static class Decoder //implements MessageDecoder { /** * - * @param session - * @param in + * @param session the session + * @param in input buffer * @return true if we have enough data to decode the PI frame fully, false if more * data is required */ @@ -115,63 +133,62 @@ } public void decode(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) - throws Exception { - byte[] theHeader = new byte[4]; - in.get(theHeader); - ProtocolInitiation pi = new ProtocolInitiation((byte)0, (byte)0); - pi.header = new char[]{(char) theHeader[0],(char) theHeader[CURRENT_PROTOCOL_INSTANCE],(char) theHeader[2], (char) theHeader[3]}; - String stringHeader = new String(pi.header); - if (!"AMQP".equals(stringHeader)) - { - throw new AMQProtocolHeaderException("Invalid protocol header - read " + stringHeader); - } - pi.protocolClass = in.get(); - pi.protocolInstance = in.get(); - pi.protocolMajor = in.get(); - pi.protocolMinor = in.get(); + ProtocolInitiation pi = new ProtocolInitiation(in); out.write(pi); } } - public void checkVersion(ProtocolVersionList pvl) throws AMQException + public void checkVersion() throws AMQException { - if (protocolClass != CURRENT_PROTOCOL_CLASS) - { - throw new AMQProtocolClassException("Protocol class " + CURRENT_PROTOCOL_CLASS + " was expected; received " + - protocolClass); - } - if (protocolInstance != CURRENT_PROTOCOL_INSTANCE) + + if(_protocolHeader.length != 4) { - throw new AMQProtocolInstanceException("Protocol instance " + CURRENT_PROTOCOL_INSTANCE + " was expected; received " + - protocolInstance); + throw new AMQProtocolHeaderException("Protocol header should have exactly four octets"); } - - /* Look through list of available protocol versions */ - boolean found = false; - for (int i=0; i<pvl.pv.length; i++) + for(int i = 0; i < 4; i++) { - if (pvl.pv[i][pvl.PROTOCOL_MAJOR] == protocolMajor && - pvl.pv[i][pvl.PROTOCOL_MINOR] == protocolMinor) + if(_protocolHeader[i] != AMQP_HEADER[i]) { - found = true; + try + { + throw new AMQProtocolHeaderException("Protocol header is not correct: Got " + new String(_protocolHeader,"ISO-8859-1") + " should be: " + new String(AMQP_HEADER, "ISO-8859-1")); + } + catch (UnsupportedEncodingException e) + { + + } } } - if (!found) + if (_protocolClass != CURRENT_PROTOCOL_CLASS) + { + throw new AMQProtocolClassException("Protocol class " + CURRENT_PROTOCOL_CLASS + " was expected; received " + + _protocolClass); + } + if (_protocolInstance != TCP_PROTOCOL_INSTANCE) + { + throw new AMQProtocolInstanceException("Protocol instance " + TCP_PROTOCOL_INSTANCE + " was expected; received " + + _protocolInstance); + } + + ProtocolVersion pv = new ProtocolVersion(_protocolMajor, _protocolMinor); + + + if (!pv.isSupported()) { // TODO: add list of available versions in list to msg... throw new AMQProtocolVersionException("Protocol version " + - protocolMajor + "." + protocolMinor + " not found in protocol version list."); + _protocolMajor + "." + _protocolMinor + " not suppoerted by this version of the Qpid broker."); } } public String toString() { - StringBuffer buffer = new StringBuffer(new String(header)); - buffer.append(Integer.toHexString(protocolClass)); - buffer.append(Integer.toHexString(protocolInstance)); - buffer.append(Integer.toHexString(protocolMajor)); - buffer.append(Integer.toHexString(protocolMinor)); + StringBuffer buffer = new StringBuffer(new String(_protocolHeader)); + buffer.append(Integer.toHexString(_protocolClass)); + buffer.append(Integer.toHexString(_protocolInstance)); + buffer.append(Integer.toHexString(_protocolMajor)); + buffer.append(Integer.toHexString(_protocolMinor)); return buffer.toString(); } } Modified: incubator/qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/framing/BasicContentHeaderPropertiesTest.java URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/framing/BasicContentHeaderPropertiesTest.java?view=diff&rev=511389&r1=511388&r2=511389 ============================================================================== --- incubator/qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/framing/BasicContentHeaderPropertiesTest.java (original) +++ incubator/qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/framing/BasicContentHeaderPropertiesTest.java Sat Feb 24 17:08:57 2007 @@ -22,8 +22,6 @@ import org.apache.mina.common.ByteBuffer; -import java.util.HashMap; - import junit.framework.TestCase; @@ -94,14 +92,14 @@ { String contentType = "contentType"; _testProperties.setContentType(contentType); - assertEquals(contentType, _testProperties.getContentType()); + assertEquals(contentType, _testProperties.getContentTypeAsString()); } public void testSetGetEncoding() { String encoding = "encoding"; _testProperties.setEncoding(encoding); - assertEquals(encoding, _testProperties.getEncoding()); + assertEquals(encoding, _testProperties.getEncodingAsString()); } public void testSetGetHeaders() @@ -128,14 +126,14 @@ { String correlationId = "correlationId"; _testProperties.setCorrelationId(correlationId); - assertEquals(correlationId, _testProperties.getCorrelationId()); + assertEquals(correlationId, _testProperties.getCorrelationIdAsString()); } public void testSetGetReplyTo() { String replyTo = "replyTo"; _testProperties.setReplyTo(replyTo); - assertEquals(replyTo, _testProperties.getReplyTo()); + assertEquals(replyTo, _testProperties.getReplyToAsString()); } public void testSetGetExpiration() @@ -149,7 +147,7 @@ { String messageId = "messageId"; _testProperties.setMessageId(messageId); - assertEquals(messageId, _testProperties.getMessageId()); + assertEquals(messageId, _testProperties.getMessageIdAsString()); } public void testSetGetTimestamp() @@ -163,28 +161,28 @@ { String type = "type"; _testProperties.setType(type); - assertEquals(type, _testProperties.getType()); + assertEquals(type, _testProperties.getTypeAsString()); } public void testSetGetUserId() { String userId = "userId"; _testProperties.setUserId(userId); - assertEquals(userId, _testProperties.getUserId()); + assertEquals(userId, _testProperties.getUserIdAsString()); } public void testSetGetAppId() { String appId = "appId"; _testProperties.setAppId(appId); - assertEquals(appId, _testProperties.getAppId()); + assertEquals(appId, _testProperties.getAppIdAsString()); } public void testSetGetClusterId() { String clusterId = "clusterId"; _testProperties.setClusterId(clusterId); - assertEquals(clusterId, _testProperties.getClusterId()); + assertEquals(clusterId, _testProperties.getClusterIdAsString()); } } Modified: incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MockProtocolSession.java URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MockProtocolSession.java?view=diff&rev=511389&r1=511388&r2=511389 ============================================================================== --- incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MockProtocolSession.java (original) +++ incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MockProtocolSession.java Sat Feb 24 17:08:57 2007 @@ -26,6 +26,7 @@ import org.apache.qpid.framing.AMQShortString; import org.apache.qpid.framing.VersionSpecificRegistry; import org.apache.qpid.server.AMQChannel; +import org.apache.qpid.server.output.ProtocolOutputConverter; import org.apache.qpid.server.virtualhost.VirtualHost; import org.apache.qpid.server.protocol.AMQProtocolSession; import org.apache.qpid.server.store.MessageStore; @@ -168,6 +169,11 @@ public void removeSessionCloseTask(Task task) { //To change body of implemented methods use File | Settings | File Templates. + } + + public ProtocolOutputConverter getProtocolOutputConverter() + { + return null; //To change body of implemented methods use File | Settings | File Templates. } public byte getProtocolMajorVersion()
