001 /*
002 *
003 * Licensed to the Apache Software Foundation (ASF) under one
004 * or more contributor license agreements. See the NOTICE file
005 * distributed with this work for additional information
006 * regarding copyright ownership. The ASF licenses this file
007 * to you under the Apache License, Version 2.0 (the
008 * "License"); you may not use this file except in compliance
009 * with the License. You may obtain a copy of the License at
010 *
011 * http://www.apache.org/licenses/LICENSE-2.0
012 *
013 * Unless required by applicable law or agreed to in writing,
014 * software distributed under the License is distributed on an
015 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
016 * KIND, either express or implied. See the License for the
017 * specific language governing permissions and limitations
018 * under the License.
019 *
020 */
021 package org.apache.qpid.framing;
022
023 import org.apache.mina.common.ByteBuffer;
024 import org.apache.mina.common.IoSession;
025 import org.apache.mina.filter.codec.ProtocolDecoderOutput;
026 import org.apache.qpid.AMQException;
027
028 import java.io.UnsupportedEncodingException;
029
030 public class ProtocolInitiation extends AMQDataBlock implements EncodableAMQDataBlock
031 {
032
033 // TODO: generate these constants automatically from the xml protocol spec file
034 public static final byte[] AMQP_HEADER = new byte[]{(byte)'A',(byte)'M',(byte)'Q',(byte)'P'};
035
036 private static final byte CURRENT_PROTOCOL_CLASS = 1;
037 private static final byte TCP_PROTOCOL_INSTANCE = 1;
038
039 public final byte[] _protocolHeader;
040 public final byte _protocolClass;
041 public final byte _protocolInstance;
042 public final byte _protocolMajor;
043 public final byte _protocolMinor;
044
045
046 // public ProtocolInitiation() {}
047
048 public ProtocolInitiation(byte[] protocolHeader, byte protocolClass, byte protocolInstance, byte protocolMajor, byte protocolMinor)
049 {
050 _protocolHeader = protocolHeader;
051 _protocolClass = protocolClass;
052 _protocolInstance = protocolInstance;
053 _protocolMajor = protocolMajor;
054 _protocolMinor = protocolMinor;
055 }
056
057 public ProtocolInitiation(ProtocolVersion pv)
058 {
059 this(AMQP_HEADER, CURRENT_PROTOCOL_CLASS, TCP_PROTOCOL_INSTANCE, pv.getMajorVersion(), pv.getMinorVersion());
060 }
061
062
063 public ProtocolInitiation(ByteBuffer in)
064 {
065 _protocolHeader = new byte[4];
066 in.get(_protocolHeader);
067
068 _protocolClass = in.get();
069 _protocolInstance = in.get();
070 _protocolMajor = in.get();
071 _protocolMinor = in.get();
072 }
073
074 public long getSize()
075 {
076 return 4 + 1 + 1 + 1 + 1;
077 }
078
079 public void writePayload(ByteBuffer buffer)
080 {
081
082 buffer.put(_protocolHeader);
083 buffer.put(_protocolClass);
084 buffer.put(_protocolInstance);
085 buffer.put(_protocolMajor);
086 buffer.put(_protocolMinor);
087 }
088
089 public boolean equals(Object o)
090 {
091 if (!(o instanceof ProtocolInitiation))
092 {
093 return false;
094 }
095
096 ProtocolInitiation pi = (ProtocolInitiation) o;
097 if (pi._protocolHeader == null)
098 {
099 return false;
100 }
101
102 if (_protocolHeader.length != pi._protocolHeader.length)
103 {
104 return false;
105 }
106
107 for (int i = 0; i < _protocolHeader.length; i++)
108 {
109 if (_protocolHeader[i] != pi._protocolHeader[i])
110 {
111 return false;
112 }
113 }
114
115 return (_protocolClass == pi._protocolClass &&
116 _protocolInstance == pi._protocolInstance &&
117 _protocolMajor == pi._protocolMajor &&
118 _protocolMinor == pi._protocolMinor);
119 }
120
121 public static class Decoder //implements MessageDecoder
122 {
123 /**
124 *
125 * @param session the session
126 * @param in input buffer
127 * @return true if we have enough data to decode the PI frame fully, false if more
128 * data is required
129 */
130 public boolean decodable(IoSession session, ByteBuffer in)
131 {
132 return (in.remaining() >= 8);
133 }
134
135 public void decode(IoSession session, ByteBuffer in, ProtocolDecoderOutput out)
136 {
137 ProtocolInitiation pi = new ProtocolInitiation(in);
138 out.write(pi);
139 }
140 }
141
142 public ProtocolVersion checkVersion() throws AMQException
143 {
144
145 if(_protocolHeader.length != 4)
146 {
147 throw new AMQProtocolHeaderException("Protocol header should have exactly four octets", null);
148 }
149 for(int i = 0; i < 4; i++)
150 {
151 if(_protocolHeader[i] != AMQP_HEADER[i])
152 {
153 try
154 {
155 throw new AMQProtocolHeaderException("Protocol header is not correct: Got " + new String(_protocolHeader,"ISO-8859-1") + " should be: " + new String(AMQP_HEADER, "ISO-8859-1"), null);
156 }
157 catch (UnsupportedEncodingException e)
158 {
159
160 }
161 }
162 }
163 if (_protocolClass != CURRENT_PROTOCOL_CLASS)
164 {
165 throw new AMQProtocolClassException("Protocol class " + CURRENT_PROTOCOL_CLASS + " was expected; received " +
166 _protocolClass, null);
167 }
168 if (_protocolInstance != TCP_PROTOCOL_INSTANCE)
169 {
170 throw new AMQProtocolInstanceException("Protocol instance " + TCP_PROTOCOL_INSTANCE + " was expected; received " +
171 _protocolInstance, null);
172 }
173
174 ProtocolVersion pv = new ProtocolVersion(_protocolMajor, _protocolMinor);
175
176
177 if (!pv.isSupported())
178 {
179 // TODO: add list of available versions in list to msg...
180 throw new AMQProtocolVersionException("Protocol version " +
181 _protocolMajor + "." + _protocolMinor + " not suppoerted by this version of the Qpid broker.", null);
182 }
183 return pv;
184 }
185
186 public String toString()
187 {
188 StringBuffer buffer = new StringBuffer(new String(_protocolHeader));
189 buffer.append(Integer.toHexString(_protocolClass));
190 buffer.append(Integer.toHexString(_protocolInstance));
191 buffer.append(Integer.toHexString(_protocolMajor));
192 buffer.append(Integer.toHexString(_protocolMinor));
193 return buffer.toString();
194 }
195 }
|