diff options
Diffstat (limited to 'dotnet/Qpid.Common/Framing')
24 files changed, 2212 insertions, 0 deletions
diff --git a/dotnet/Qpid.Common/Framing/AMQDataBlockDecoder.cs b/dotnet/Qpid.Common/Framing/AMQDataBlockDecoder.cs new file mode 100644 index 0000000000..f76ac005e4 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/AMQDataBlockDecoder.cs @@ -0,0 +1,153 @@ +/* + * + * 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. + * + */ +using System; +using System.Collections; +using log4net; +using Qpid.Buffer; +using Qpid.Codec; +using Qpid.Codec.Demux; + +namespace Qpid.Framing +{ + public class AMQDataBlockDecoder : IMessageDecoder + { + private static ILog _logger = LogManager.GetLogger(typeof(AMQDataBlockDecoder)); + + private Hashtable _supportedBodies = new Hashtable(); + + private bool _disabled = false; + + public AMQDataBlockDecoder() + { + _supportedBodies[AMQMethodBody.TYPE] = AMQMethodBodyFactory.GetInstance(); + _supportedBodies[ContentHeaderBody.TYPE] = ContentHeaderBodyFactory.GetInstance(); + _supportedBodies[ContentBody.TYPE] = ContentBodyFactory.GetInstance(); + _supportedBodies[HeartbeatBody.TYPE] = new HeartbeatBodyFactory(); + } + + public MessageDecoderResult Decodable(ByteBuffer input) + { + if (_disabled) + { + return MessageDecoderResult.NOT_OK; + } + // final +1 represents the command end which we know we must require even + // if there is an empty body + if (input.Remaining < 1) + { + return MessageDecoderResult.NEED_DATA; + } + byte type = input.Get(); + + // we have to check this isn't a protocol initiation frame here - we can't tell later on and we end up + // waiting for more data. This could be improved if MINA supported some kind of state awareness when decoding + if ((char)type == 'A') + { + _logger.Error("Received what appears to be a protocol initiation frame"); + return MessageDecoderResult.NOT_OK; + } + // zero, channel, body size and end byte + if (input.Remaining < (1 + 2 + 4 + 1)) + { + return MessageDecoderResult.NEED_DATA; + } + + int channel = input.GetUnsignedShort(); + long bodySize = input.GetUnsignedInt(); + + // bodySize can be zero + if (type <= 0 || channel < 0 || bodySize < 0) + { + _logger.Error(String.Format("Error decoding frame: Type={0}, Channel={1}, BodySize={2}", type, channel, bodySize)); + return MessageDecoderResult.NOT_OK; + } + + if (input.Remaining < (bodySize + 1)) + { + return MessageDecoderResult.NEED_DATA; + } + + if (IsSupportedFrameType(type)) + { + if (_logger.IsDebugEnabled) + { + // we have read 7 bytes so far, so output 7 + bodysize + 1 (for end byte) to get complete data block size + // this logging statement is useful when looking at exactly what size of data is coming in/out + // the broker + _logger.Debug("Able to decode data block of size " + (bodySize + 8)); + } + return MessageDecoderResult.OK; + } + else + { + return MessageDecoderResult.NOT_OK; + } + } + + private bool IsSupportedFrameType(byte frameType) + { + bool result = _supportedBodies.ContainsKey(frameType); + + if (!result) + { + _logger.Warn("AMQDataBlockDecoder does not handle frame type " + frameType); + } + + return result; + } + + protected Object CreateAndPopulateFrame(ByteBuffer input) + { + byte type = input.Get(); + ushort channel = input.GetUnsignedShort(); + uint bodySize = input.GetUnsignedInt(); + + IBodyFactory bodyFactory = (IBodyFactory)_supportedBodies[type]; + if (bodyFactory == null) + { + throw new AMQFrameDecodingException("Unsupported body type: " + type); + } + AMQFrame frame = new AMQFrame(); + + frame.PopulateFromBuffer(input, channel, bodySize, bodyFactory); + + byte marker = input.Get(); + //assert marker == 0xCE; + return frame; + } + + public MessageDecoderResult Decode(ByteBuffer input, IProtocolDecoderOutput output) + { + + output.Write(CreateAndPopulateFrame(input)); + + return MessageDecoderResult.OK; + } + + public bool Disabled + { + set + { + _disabled = value; + } + } + } +} diff --git a/dotnet/Qpid.Common/Framing/AMQDataBlockEncoder.cs b/dotnet/Qpid.Common/Framing/AMQDataBlockEncoder.cs new file mode 100644 index 0000000000..b180e1ac95 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/AMQDataBlockEncoder.cs @@ -0,0 +1,65 @@ +/* + * + * 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. + * + */ +using System.Collections; +using log4net; +using Qpid.Buffer; +using Qpid.Codec; +using Qpid.Codec.Demux; + +namespace Qpid.Framing +{ + public class AMQDataBlockEncoder : IMessageEncoder + { + private static ILog _logger = LogManager.GetLogger(typeof(AMQDataBlockEncoder)); + + private Hashtable _messageTypes; + + public AMQDataBlockEncoder() + { + _messageTypes = new Hashtable(); + _messageTypes[typeof (IEncodableAMQDataBlock)] = 1; + } + + + public Hashtable MessageTypes + { + get + { + return _messageTypes; + } + } + + public void Encode(object message, IProtocolEncoderOutput output) + { + IDataBlock frame = (IDataBlock) message; + int frameSize = (int)frame.Size; // TODO: sort out signed/unsigned + ByteBuffer buffer = ByteBuffer.Allocate(frameSize); + frame.WritePayload(buffer); + + if (_logger.IsDebugEnabled) + { + _logger.Debug("Encoded frame byte-buffer is '" + ByteBufferHexDumper.GetHexDump(buffer) + "'"); + } + buffer.Flip(); + output.Write(buffer); + } + } +} diff --git a/dotnet/Qpid.Common/Framing/AMQFrame.cs b/dotnet/Qpid.Common/Framing/AMQFrame.cs new file mode 100644 index 0000000000..2708c331b3 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/AMQFrame.cs @@ -0,0 +1,107 @@ +/* + * + * 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. + * + */ +using Qpid.Buffer; + +namespace Qpid.Framing +{ + public class AMQFrame : IDataBlock + { + private ushort _channel; + + private IBody _bodyFrame; + + public AMQFrame() + { + } + + public AMQFrame(ushort channel, IBody bodyFrame) + { + _channel = channel; + _bodyFrame = bodyFrame; + } + + public ushort Channel + { + get + { + return _channel; + } + set + { + _channel = value; + } + } + + public IBody BodyFrame + { + get + { + return _bodyFrame; + } + set + { + _bodyFrame = value; + } + } + + #region IDataBlock Members + + public uint Size + { + get + { + return (uint) (1 + 2 + 4 + _bodyFrame.Size + 1); + } + } + + public void WritePayload(ByteBuffer buffer) + { + buffer.Put(_bodyFrame.BodyType); + // TODO: how does channel get populated + buffer.Put(_channel); + buffer.Put(_bodyFrame.Size); + _bodyFrame.WritePayload(buffer); + buffer.Put((byte) 0xCE); + } + + #endregion + + /// <summary> + /// Populates the frame instance data from the supplied buffer. + /// </summary> + /// <param name="buffer">The buffer.</param> + /// <param name="channel">The channel.</param> + /// <param name="bodySize">Size of the body in bytes</param> + /// <param name="bodyFactory">The body factory.</param> + /// <exception cref="AMQFrameDecodingException">Thrown if the buffer cannot be decoded</exception> + public void PopulateFromBuffer(ByteBuffer buffer, ushort channel, uint bodySize, IBodyFactory bodyFactory) + { + _channel = channel; + _bodyFrame = bodyFactory.CreateBody(buffer); + _bodyFrame.PopulateFromBuffer(buffer, bodySize); + } + + public override string ToString() + { + return "Frame channelId: " + _channel + ", bodyFrame: " + _bodyFrame.ToString(); + } + } +} diff --git a/dotnet/Qpid.Common/Framing/AMQFrameDecodingException.cs b/dotnet/Qpid.Common/Framing/AMQFrameDecodingException.cs new file mode 100644 index 0000000000..e7223c9850 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/AMQFrameDecodingException.cs @@ -0,0 +1,52 @@ +/* + * + * 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. + * + */ +using System; +using log4net; + +namespace Qpid.Framing +{ + /// <summary> + /// Thrown when a frame cannot be decoded. This generally indicates a mismatch between the broker and the + /// client. + /// </summary> + public class AMQFrameDecodingException : AMQException + { + public AMQFrameDecodingException(string message) + : base(message) + { + } + + public AMQFrameDecodingException(string message, Exception innerException) + : base(message, innerException) + { + } + + public AMQFrameDecodingException(ILog logger, string message) + : base(logger, message) + { + } + + public AMQFrameDecodingException(ILog logger, string message, Exception innerException) + : base(logger, message, innerException) + { + } + } +} diff --git a/dotnet/Qpid.Common/Framing/AMQMethodBody.cs b/dotnet/Qpid.Common/Framing/AMQMethodBody.cs new file mode 100644 index 0000000000..804e6a4039 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/AMQMethodBody.cs @@ -0,0 +1,93 @@ +/* + * + * 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. + * + */ +using System; +using Qpid.Buffer; + +namespace Qpid.Framing +{ + public abstract class AMQMethodBody : IBody + { + public const byte TYPE = 1; + + protected abstract uint BodySize + { + get; + } + + protected abstract ushort Clazz + { + get; + } + + protected abstract ushort Method + { + get; + } + + protected abstract void WriteMethodPayload(ByteBuffer buffer); + + public byte BodyType + { + get + { + return TYPE; + } + } + + public uint Size + { + get + { + return (uint) (2 + 2 + BodySize); + } + } + + public void WritePayload(ByteBuffer buffer) + { + buffer.Put(Clazz); + buffer.Put(Method); + WriteMethodPayload(buffer); + } + + /// <summary> + /// Populates the method body by decoding the specified buffer + /// </summary> + /// <param name="buffer">The buffer to decode.</param> + /// <exception cref="AMQFrameDecodingException">If the buffer cannot be decoded</exception> + protected abstract void PopulateMethodBodyFromBuffer(ByteBuffer buffer); + + /// <summary> + /// Populates this instance from a buffer of data. + /// </summary> + /// <param name="buffer">The buffer.</param> + /// <param name="size">The size.</param> + /// <exception cref="AMQFrameDecodingException">If the buffer contains data that cannot be decoded</exception> + public void PopulateFromBuffer(ByteBuffer buffer, uint size) + { + PopulateMethodBodyFromBuffer(buffer); + } + + public override string ToString() + { + return String.Format("{0}{{ Class: {1} Method: {2} }}", GetType().Name, Clazz, Method); + } + } +} diff --git a/dotnet/Qpid.Common/Framing/AMQMethodBodyFactory.cs b/dotnet/Qpid.Common/Framing/AMQMethodBodyFactory.cs new file mode 100644 index 0000000000..5df9720678 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/AMQMethodBodyFactory.cs @@ -0,0 +1,45 @@ +/* + * + * 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. + * + */ +using Qpid.Buffer; + +namespace Qpid.Framing +{ + public class AMQMethodBodyFactory : IBodyFactory + { + private static readonly AMQMethodBodyFactory _instance = new AMQMethodBodyFactory(); + + public static AMQMethodBodyFactory GetInstance() + { + return _instance; + } + + /// <summary> + /// Creates the body. + /// </summary> + /// <param name="inbuf">The ByteBuffer containing data from the network</param> + /// <returns></returns> + /// <exception>AMQFrameDecodingException</exception> + public IBody CreateBody(ByteBuffer inbuf) + { + return MethodBodyDecoderRegistry.Get(inbuf.GetUnsignedShort(), inbuf.GetUnsignedShort()); + } + } +} diff --git a/dotnet/Qpid.Common/Framing/AMQProtocolHeaderException.cs b/dotnet/Qpid.Common/Framing/AMQProtocolHeaderException.cs new file mode 100644 index 0000000000..7ea509c78a --- /dev/null +++ b/dotnet/Qpid.Common/Framing/AMQProtocolHeaderException.cs @@ -0,0 +1,29 @@ +/* + * + * 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. + * + */ +namespace Qpid.Framing +{ + public class AMQProtocolHeaderException : AMQException + { + public AMQProtocolHeaderException(string message) : base(message) + { + } + } +} diff --git a/dotnet/Qpid.Common/Framing/BasicContentHeaderProperties.cs b/dotnet/Qpid.Common/Framing/BasicContentHeaderProperties.cs new file mode 100644 index 0000000000..10d22b0e30 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/BasicContentHeaderProperties.cs @@ -0,0 +1,168 @@ +/* + * + * 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. + * + */ +using System; +using log4net; +using Qpid.Buffer; +using Qpid.Messaging; + +namespace Qpid.Framing +{ + public class BasicContentHeaderProperties : IContentHeaderProperties + { + private static readonly ILog _log = LogManager.GetLogger(typeof(BasicContentHeaderProperties)); + + public string ContentType; + + public string Encoding; + + public FieldTable Headers; + + public byte DeliveryMode; + + public byte Priority; + + public string CorrelationId; + + public uint Expiration; + + public string ReplyTo; + + public string MessageId; + + public ulong Timestamp; + + public string Type; + + public string UserId; + + public string AppId; + + public string ClusterId; + + public BasicContentHeaderProperties() + { + } + + public uint PropertyListSize + { + get + { + return (uint)(EncodingUtils.EncodedShortStringLength(ContentType) + + EncodingUtils.EncodedShortStringLength(Encoding) + + EncodingUtils.EncodedFieldTableLength(Headers) + + 1 + 1 + + EncodingUtils.EncodedShortStringLength(CorrelationId) + + EncodingUtils.EncodedShortStringLength(ReplyTo) + + EncodingUtils.EncodedShortStringLength(String.Format("{0:D}", Expiration)) + + EncodingUtils.EncodedShortStringLength(MessageId) + + 8 + + EncodingUtils.EncodedShortStringLength(Type) + + EncodingUtils.EncodedShortStringLength(UserId) + + EncodingUtils.EncodedShortStringLength(AppId) + + EncodingUtils.EncodedShortStringLength(ClusterId)); + + } + } + + public ushort PropertyFlags + { + get + { + int value = 0; + + // for now we just blast in all properties + for (int i = 0; i < 14; i++) + { + value += (1 << (15-i)); + } + return (ushort) value; + } + } + + public void WritePropertyListPayload(ByteBuffer buffer) + { + EncodingUtils.WriteShortStringBytes(buffer, ContentType); + EncodingUtils.WriteShortStringBytes(buffer, Encoding); + EncodingUtils.WriteFieldTableBytes(buffer, Headers); + buffer.Put(DeliveryMode); + buffer.Put(Priority); + EncodingUtils.WriteShortStringBytes(buffer, CorrelationId); + EncodingUtils.WriteShortStringBytes(buffer, ReplyTo); + EncodingUtils.WriteShortStringBytes(buffer, String.Format("{0:D}", Expiration)); + EncodingUtils.WriteShortStringBytes(buffer, MessageId); + buffer.Put(Timestamp); + EncodingUtils.WriteShortStringBytes(buffer, Type); + EncodingUtils.WriteShortStringBytes(buffer, UserId); + EncodingUtils.WriteShortStringBytes(buffer, AppId); + EncodingUtils.WriteShortStringBytes(buffer, ClusterId); + } + + public void PopulatePropertiesFromBuffer(ByteBuffer buffer, ushort propertyFlags) + { + _log.Debug("Property flags: " + propertyFlags); + if ((propertyFlags & (1 << 15)) > 0) + ContentType = EncodingUtils.ReadShortString(buffer); + if ((propertyFlags & (1 << 14)) > 0) + Encoding = EncodingUtils.ReadShortString(buffer); + if ((propertyFlags & (1 << 13)) > 0) + Headers = EncodingUtils.ReadFieldTable(buffer); + if ((propertyFlags & (1 << 12)) > 0) + DeliveryMode = buffer.Get(); + if ((propertyFlags & (1 << 11)) > 0) + Priority = buffer.Get(); + if ((propertyFlags & (1 << 10)) > 0) + CorrelationId = EncodingUtils.ReadShortString(buffer); + if ((propertyFlags & (1 << 9)) > 0) + ReplyTo = EncodingUtils.ReadShortString(buffer); + if ((propertyFlags & (1 << 8)) > 0) + Expiration = UInt32.Parse(EncodingUtils.ReadShortString(buffer)); + if ((propertyFlags & (1 << 7)) > 0) + MessageId = EncodingUtils.ReadShortString(buffer); + if ((propertyFlags & (1 << 6)) > 0) + Timestamp = buffer.GetUnsignedLong(); + if ((propertyFlags & (1 << 5)) > 0) + Type = EncodingUtils.ReadShortString(buffer); + if ((propertyFlags & (1 << 4)) > 0) + UserId = EncodingUtils.ReadShortString(buffer); + if ((propertyFlags & (1 << 3)) > 0) + AppId = EncodingUtils.ReadShortString(buffer); + if ((propertyFlags & (1 << 2)) > 0) + ClusterId = EncodingUtils.ReadShortString(buffer); + } + + public void SetDeliveryMode(DeliveryMode deliveryMode) + { + if (deliveryMode == Messaging.DeliveryMode.NonPersistent) + { + DeliveryMode = 1; + } + else + { + DeliveryMode = 2; + } + } + + public override string ToString() + { + return "Properties: " + ContentType + " " + Encoding + " " + Timestamp + " " + Type; + } + } +} diff --git a/dotnet/Qpid.Common/Framing/CompositeAMQDataBlock.cs b/dotnet/Qpid.Common/Framing/CompositeAMQDataBlock.cs new file mode 100644 index 0000000000..19d12b8039 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/CompositeAMQDataBlock.cs @@ -0,0 +1,85 @@ +/* + * + * 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. + * + */ +using System.Text; +using Qpid.Buffer; + +namespace Qpid.Framing +{ + public class CompositeAMQDataBlock : IDataBlock, IEncodableAMQDataBlock + { + private IDataBlock[] _blocks; + + public CompositeAMQDataBlock(IDataBlock[] blocks) + { + _blocks = blocks; + } + + public IDataBlock[] Blocks + { + get + { + return _blocks; + } + } + + public uint Size + { + get + { + uint frameSize = 0; + foreach (IDataBlock block in _blocks) + { + frameSize += block.Size; + } + return frameSize; + } + } + + public void WritePayload(ByteBuffer buffer) + { + foreach (IDataBlock block in _blocks) + { + block.WritePayload(buffer); + } + } + + public override string ToString() + { + if (_blocks == null) + { + return "No blocks contained in composite frame"; + } + else + { + StringBuilder buf = new StringBuilder(GetType().Name); + buf.Append("{"); + //buf.Append("encodedBlock=").Append(_encodedBlock); + for (int i = 0; i < _blocks.Length; i++) + { + buf.Append(" ").Append(i).Append("=[").Append(_blocks[i].ToString()).Append("]"); + } + buf.Append("}"); + return buf.ToString(); + } + } + + } +} diff --git a/dotnet/Qpid.Common/Framing/ContentBody.cs b/dotnet/Qpid.Common/Framing/ContentBody.cs new file mode 100644 index 0000000000..e8bc003f65 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/ContentBody.cs @@ -0,0 +1,80 @@ +/* + * + * 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. + * + */ +using Qpid.Buffer; + +namespace Qpid.Framing +{ + public class ContentBody : IBody + { + public const byte TYPE = 3; + + /// <summary> + /// + /// </summary> + /// TODO: consider whether this should be a pointer into the ByteBuffer to avoid copying */ + public byte[] Payload; + + #region IBody Members + + public byte BodyType + { + get + { + return TYPE; + } + } + + public uint Size + { + get + { + return (ushort)(Payload == null ? 0 : Payload.Length); + } + } + + public void WritePayload(ByteBuffer buffer) + { + if (Payload != null) + { + buffer.Put(Payload); + } + } + + public void PopulateFromBuffer(ByteBuffer buffer, uint size) + { + if (size > 0) + { + Payload = new byte[size]; + buffer.Get(Payload); + } + } + + #endregion + + public static AMQFrame CreateAMQFrame(ushort channelId, ContentBody body) + { + AMQFrame frame = new AMQFrame(); + frame.Channel = channelId; + frame.BodyFrame = body; + return frame; + } + } +} diff --git a/dotnet/Qpid.Common/Framing/ContentBodyFactory.cs b/dotnet/Qpid.Common/Framing/ContentBodyFactory.cs new file mode 100644 index 0000000000..ec0e89b421 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/ContentBodyFactory.cs @@ -0,0 +1,53 @@ +/* + * + * 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. + * + */ +using log4net; +using Qpid.Buffer; + +namespace Qpid.Framing +{ + public class ContentBodyFactory : IBodyFactory + { + private static readonly ILog _log = LogManager.GetLogger(typeof(ContentBodyFactory)); + + private static readonly ContentBodyFactory _instance = new ContentBodyFactory(); + + public static ContentBodyFactory GetInstance() + { + return _instance; + } + + private ContentBodyFactory() + { + _log.Debug("Creating content body factory"); + } + + /// <summary> + /// Creates the body. + /// </summary> + /// <param name="inbuf">The ByteBuffer containing data from the network</param> + /// <returns></returns> + /// <exception>AMQFrameDecodingException</exception> + public IBody CreateBody(ByteBuffer inbuf) + { + return new ContentBody(); + } + } +} diff --git a/dotnet/Qpid.Common/Framing/ContentHeaderBody.cs b/dotnet/Qpid.Common/Framing/ContentHeaderBody.cs new file mode 100644 index 0000000000..a3f71c41aa --- /dev/null +++ b/dotnet/Qpid.Common/Framing/ContentHeaderBody.cs @@ -0,0 +1,118 @@ +/* + * + * 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. + * + */ +using System; +using Qpid.Buffer; + +namespace Qpid.Framing +{ + public class ContentHeaderBody : IBody + { + public static readonly byte TYPE = 2; + + public ushort ClassId; + + public ushort Weight; + + public ulong BodySize; + + /** must never be null */ + public IContentHeaderProperties Properties; + + public ContentHeaderBody() + { + } + + public ContentHeaderBody(IContentHeaderProperties props, ushort classId) + { + Properties = props; + ClassId = classId; + } + + public ContentHeaderBody(ushort classId, ushort weight, IContentHeaderProperties props, uint bodySize) + : this(props, classId) + { + Weight = weight; + BodySize = bodySize; + } + + #region IBody Members + + public byte BodyType + { + get + { + return TYPE; + } + } + + public uint Size + { + get + { + return (2 + 2 + 8 + 2 + Properties.PropertyListSize); + } + } + + public void WritePayload(ByteBuffer buffer) + { + buffer.Put(ClassId); + buffer.Put(Weight); + buffer.Put(BodySize); + buffer.Put(Properties.PropertyFlags); + Properties.WritePropertyListPayload(buffer); + } + + public void PopulateFromBuffer(ByteBuffer buffer, uint size) + { + ClassId = buffer.GetUnsignedShort(); + Weight = buffer.GetUnsignedShort(); + BodySize = buffer.GetUnsignedLong(); + ushort propertyFlags = buffer.GetUnsignedShort(); + ContentHeaderPropertiesFactory factory = ContentHeaderPropertiesFactory.GetInstance(); + Properties = factory.CreateContentHeaderProperties(ClassId, propertyFlags, buffer); + } + + #endregion + + public static AMQFrame CreateAMQFrame(ushort channelId, ushort classId, ushort weight, BasicContentHeaderProperties properties, + uint bodySize) + { + AMQFrame frame = new AMQFrame(); + frame.Channel = channelId; + frame.BodyFrame = new ContentHeaderBody(classId, weight, properties, bodySize); + return frame; + } + + public static AMQFrame CreateAMQFrame(ushort channelId, ContentHeaderBody body) + { + AMQFrame frame = new AMQFrame(); + frame.Channel = channelId; + frame.BodyFrame = body; + return frame; + } + + public override string ToString() + { + return String.Format("ContentHeaderBody: ClassId {0}, Weight {1}, BodySize {2}, Properties {3}", ClassId, Weight, + BodySize, Properties); + } + } +} diff --git a/dotnet/Qpid.Common/Framing/ContentHeaderBodyFactory.cs b/dotnet/Qpid.Common/Framing/ContentHeaderBodyFactory.cs new file mode 100644 index 0000000000..27cc5aeb04 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/ContentHeaderBodyFactory.cs @@ -0,0 +1,53 @@ +/* + * + * 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. + * + */ +using log4net; +using Qpid.Buffer; + +namespace Qpid.Framing +{ + public class ContentHeaderBodyFactory : IBodyFactory + { + private static readonly ILog _log = LogManager.GetLogger(typeof(ContentHeaderBodyFactory)); + + private static readonly ContentHeaderBodyFactory _instance = new ContentHeaderBodyFactory(); + + public static ContentHeaderBodyFactory GetInstance() + { + return _instance; + } + + private ContentHeaderBodyFactory() + { + _log.Debug("Creating content header body factory"); + } + + #region IBodyFactory Members + + public IBody CreateBody(ByteBuffer inbuf) + { + // all content headers are the same - it is only the properties that differ. + // the content header body further delegates construction of properties + return new ContentHeaderBody(); + } + + #endregion + } +} diff --git a/dotnet/Qpid.Common/Framing/ContentHeaderPropertiesFactory.cs b/dotnet/Qpid.Common/Framing/ContentHeaderPropertiesFactory.cs new file mode 100644 index 0000000000..16a24d6b92 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/ContentHeaderPropertiesFactory.cs @@ -0,0 +1,63 @@ +/* + * + * 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. + * + */ +using Qpid.Buffer; + +namespace Qpid.Framing +{ + public class ContentHeaderPropertiesFactory + { + + private static readonly ContentHeaderPropertiesFactory _instance = new ContentHeaderPropertiesFactory(); + + public static ContentHeaderPropertiesFactory GetInstance() + { + return _instance; + } + + private ContentHeaderPropertiesFactory() + { + } + + /// <summary> + /// Creates the content header properties from a buffer. + /// </summary> + /// <param name="classId">The class id.</param> + /// <param name="propertyFlags">The property flags.</param> + /// <param name="buffer">The buffer.</param> + /// <returns>a populated properties structure</returns> + /// <exception cref="AMQFrameDecodingException">if the buffer cannot be decoded</exception> + public IContentHeaderProperties CreateContentHeaderProperties(ushort classId, ushort propertyFlags, + ByteBuffer buffer) + { + IContentHeaderProperties properties; + switch (classId) + { + case 60: + properties = new BasicContentHeaderProperties(); + break; + default: + throw new AMQFrameDecodingException("Unsupport content header class id: " + classId); + } + properties.PopulatePropertiesFromBuffer(buffer, propertyFlags); + return properties; + } + } +} diff --git a/dotnet/Qpid.Common/Framing/EncodingUtils.cs b/dotnet/Qpid.Common/Framing/EncodingUtils.cs new file mode 100644 index 0000000000..748f3c9a0c --- /dev/null +++ b/dotnet/Qpid.Common/Framing/EncodingUtils.cs @@ -0,0 +1,251 @@ +/* + * + * 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. + * + */ +using System; +using System.Text; +using Qpid.Buffer; + +namespace Qpid.Framing +{ + public class EncodingUtils + { + private static readonly Encoding DEFAULT_ENCODER = Encoding.ASCII; + + public static ushort EncodedShortStringLength(string s) + { + if (s == null) + { + return 1; + } + else + { + return (ushort)(1 + s.Length); + } + } + + public static uint EncodedLongStringLength(string s) + { + if (s == null) + { + return 4; + } + else + { + return (uint)(4 + s.Length); + } + } + + public static int EncodedLongstrLength(byte[] bytes) + { + if (bytes == null) + { + return 4; + } + else + { + return 4 + bytes.Length; + } + } + + public static uint EncodedFieldTableLength(FieldTable table) + { + if (table == null) + { + // size is encoded as 4 octets + return 4; + } + else + { + // size of the table plus 4 octets for the size + return table.EncodedSize + 4; + } + } + + public static void WriteShortStringBytes(ByteBuffer buffer, string s) + { + if (s != null) + { + //try + //{ + //final byte[] encodedString = s.getBytes(STRING_ENCODING); + byte[] encodedString; + lock (DEFAULT_ENCODER) + { + encodedString = DEFAULT_ENCODER.GetBytes(s); + } + // TODO: check length fits in an unsigned byte + buffer.Put((byte) encodedString.Length); + buffer.Put(encodedString); + + } + else + { + // really writing out unsigned byte + buffer.Put((byte) 0); + } + } + + public static void WriteLongStringBytes(ByteBuffer buffer, string s) + { + if (!(s == null || s.Length <= 0xFFFE)) + { + throw new ArgumentException("String too long"); + } + if (s != null) + { + buffer.Put((uint)s.Length); + byte[] encodedString = null; + lock (DEFAULT_ENCODER) + { + encodedString = DEFAULT_ENCODER.GetBytes(s); + } + buffer.Put(encodedString); + } + else + { + buffer.Put((uint) 0); + } + } + + public static void WriteFieldTableBytes(ByteBuffer buffer, FieldTable table) + { + if (table != null) + { + table.WriteToBuffer(buffer); + } + else + { + buffer.Put((uint) 0); + } + } + + public static void WriteBooleans(ByteBuffer buffer, bool[] values) + { + byte packedValue = 0; + for (int i = 0; i < values.Length; i++) + { + if (values[i]) + { + packedValue = (byte) (packedValue | (1 << i)); + } + } + + buffer.Put(packedValue); + } + + public static void WriteLongstr(ByteBuffer buffer, byte[] data) + { + if (data != null) + { + buffer.Put((uint) data.Length); + buffer.Put(data); + } + else + { + buffer.Put((uint) 0); + } + } + + public static bool[] ReadBooleans(ByteBuffer buffer) + { + byte packedValue = buffer.Get(); + bool[] result = new bool[8]; + + for (int i = 0; i < 8; i++) + { + result[i] = ((packedValue & (1 << i)) != 0); + } + return result; + } + + /// <summary> + /// Reads the field table uaing the data in the specified buffer + /// </summary> + /// <param name="buffer">The buffer to read from.</param> + /// <returns>a populated field table</returns> + /// <exception cref="AMQFrameDecodingException">if the buffer does not contain a decodable field table</exception> + public static FieldTable ReadFieldTable(ByteBuffer buffer) + { + uint length = buffer.GetUnsignedInt(); + if (length == 0) + { + return null; + } + else + { + return new FieldTable(buffer, length); + } + } + + /// <summary> + /// Read a short string from the buffer + /// </summary> + /// <param name="buffer">The buffer to read from.</param> + /// <returns>a string</returns> + /// <exception cref="AMQFrameDecodingException">if the buffer does not contain a decodable short string</exception> + public static string ReadShortString(ByteBuffer buffer) + { + byte length = buffer.Get(); + if (length == 0) + { + return null; + } + else + { + lock (DEFAULT_ENCODER) + { + return buffer.GetString(length, DEFAULT_ENCODER); + } + } + } + + public static string ReadLongString(ByteBuffer buffer) + { + uint length = buffer.GetUnsignedInt(); + if (length == 0) + { + return null; + } + else + { + lock (DEFAULT_ENCODER) + { + return buffer.GetString(length, DEFAULT_ENCODER); + } + } + } + + public static byte[] ReadLongstr(ByteBuffer buffer) + { + uint length = buffer.GetUnsignedInt(); + if (length == 0) + { + return null; + } + else + { + byte[] result = new byte[length]; + buffer.Get(result); + return result; + } + } + } + +} diff --git a/dotnet/Qpid.Common/Framing/FieldTable.cs b/dotnet/Qpid.Common/Framing/FieldTable.cs new file mode 100644 index 0000000000..a7abc9d6e5 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/FieldTable.cs @@ -0,0 +1,300 @@ +/* + * + * 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. + * + */ +using System; +using System.Collections; +using System.Text; +using Qpid.Buffer; +using Qpid.Collections; +using Qpid.Messaging; + +namespace Qpid.Framing +{ + /// + /// From the protocol document: + /// field-table = short-integer *field-value-pair + /// field-value-pair = field-name field-value + /// field-name = short-string + /// field-value = 'S' long-string + /// 'I' long-integer + /// 'D' decimal-value + /// 'T' long-integer + /// decimal-value = decimals long-integer + /// decimals = OCTET + public class FieldTable : IFieldTable + { + IDictionary _hash = new LinkedHashtable(); + + private uint _encodedSize = 0; + + public FieldTable() + { + } + + /** + * Construct a new field table. + * @param buffer the buffer from which to read data. The length byte must be read already + * @param length the length of the field table. Must be > 0. + * @throws AMQFrameDecodingException if there is an error decoding the table + */ + public FieldTable(ByteBuffer buffer, uint length) + { + _encodedSize = length; + int sizeRead = 0; + while (sizeRead < _encodedSize) + { + int sizeRemaining = buffer.Remaining; + string key = EncodingUtils.ReadShortString(buffer); + // TODO: use proper charset decoder + char type = (char)buffer.Get(); + object value; + switch (type) + { + case 'S': + value = EncodingUtils.ReadLongString(buffer); + break; + case 'I': + value = buffer.GetUnsignedInt(); + break; + default: + throw new AMQFrameDecodingException("Unsupported field table type: " + type); + } + sizeRead += (sizeRemaining - buffer.Remaining); + + _hash.Add(key, value); + } + } + + public uint EncodedSize + { + get + { + return _encodedSize; + } + } + + public int Count + { + get { return _hash.Count; } + } + + public object this[string key] + { + get + { + CheckKey(key); + return _hash[key]; + } + + set + { + CheckKey(key); + CheckValue(value); + + + object oldValue = _hash[key]; + if (oldValue != null) + { + AdjustEncodingSizeWhenRemoving(key, oldValue); + } + + _hash[key] = value; + AdjustEncodingSizeWhenAdding(key, value); + } + } + + public void WriteToBuffer(ByteBuffer buffer) + { + // Write out the total length, which we have kept up to date as data is added. + buffer.Put(_encodedSize); + WritePayload(buffer); + } + + private void WritePayload(ByteBuffer buffer) + { + foreach (DictionaryEntry lde in _hash) + { + string key = (string) lde.Key; + EncodingUtils.WriteShortStringBytes(buffer, key); + object value = lde.Value; + if (value is byte[]) + { + buffer.Put((byte) 'S'); + EncodingUtils.WriteLongstr(buffer, (byte[]) value); + } + else if (value is string) + { + // TODO: look at using proper charset encoder + buffer.Put((byte) 'S'); + EncodingUtils.WriteLongStringBytes(buffer, (string) value); + } + else if (value is uint) + { + // TODO: look at using proper charset encoder + buffer.Put((byte) 'I'); + buffer.Put((uint) value); + } + else + { + // Should never get here. + throw new InvalidOperationException("Unsupported type in FieldTable: " + value.GetType()); + } + } + } + + public byte[] GetDataAsBytes() + { + ByteBuffer buffer = ByteBuffer.Allocate((int)_encodedSize); + WritePayload(buffer); + byte[] result = new byte[_encodedSize]; + buffer.Flip(); + buffer.Get(result); + //buffer.Release(); + return result; + } + + /// <summary> + /// Adds all the items from one field table in this one. Will overwrite any items in the current table + /// with the same key. + /// </summary> + /// <param name="ft">the source field table</param> + public void AddAll(IFieldTable ft) + { + foreach (DictionaryEntry dictionaryEntry in ft) + { + this[(string)dictionaryEntry.Key] = dictionaryEntry.Value; + } + } + + private void CheckKey(object key) + { + if (key == null) + { + throw new ArgumentException("All keys must be Strings - was passed: null"); + } + else if (!(key is string)) + { + throw new ArgumentException("All keys must be Strings - was passed: " + key.GetType()); + } + } + + private void CheckValue(object value) + { + if (!(value is string || value is uint || value is int || value is long)) + { + throw new ArgumentException("All values must be type string or int or long or uint, was passed: " + + value.GetType()); + } + } + + void AdjustEncodingSizeWhenAdding(object key, object value) + { + _encodedSize += EncodingUtils.EncodedShortStringLength((string) key); + // the extra byte if for the type indicator what is written out + if (value is string) + { + _encodedSize += 1 + EncodingUtils.EncodedLongStringLength((string) value); + } + else if (value is int || value is uint || value is long) + { + _encodedSize += 1 + 4; + } + else + { + // Should never get here since was already checked + throw new Exception("Unsupported value type: " + value.GetType()); + } + } + + private void AdjustEncodingSizeWhenRemoving(object key, object value) + { + _encodedSize -= EncodingUtils.EncodedShortStringLength((string) key); + if (value != null) + { + if (value is string) + { + _encodedSize -= 1 + EncodingUtils.EncodedLongStringLength((string) value); + } + else if (value is int || value is uint || value is long) + { + _encodedSize -= 5; + } + else + { + // Should never get here + throw new Exception("Illegal value type: " + value.GetType()); + } + } + } + + public IEnumerator GetEnumerator() + { + return _hash.GetEnumerator(); + } + + public bool Contains(string s) + { + return _hash.Contains(s); + } + + public void Clear() + { + _hash.Clear(); + _encodedSize = 0; + } + + public void Remove(string key) + { + object value = _hash[key]; + if (value != null) + { + AdjustEncodingSizeWhenRemoving(key, value); + } + _hash.Remove(key); + } + + public IDictionary AsDictionary() + { + return _hash; + } + + public override string ToString() + { + StringBuilder sb = new StringBuilder("FieldTable{"); + + bool first = true; + foreach (DictionaryEntry entry in _hash) + { + if (first) + { + first = !first; + } + else + { + sb.Append(", "); + } + sb.Append(entry.Key).Append(" => ").Append(entry.Value); + } + + sb.Append("}"); + return sb.ToString(); + } + } +} diff --git a/dotnet/Qpid.Common/Framing/HeartbeatBody.cs b/dotnet/Qpid.Common/Framing/HeartbeatBody.cs new file mode 100644 index 0000000000..3fdaa7e850 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/HeartbeatBody.cs @@ -0,0 +1,64 @@ +/* + * + * 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. + * + */ +using Qpid.Buffer; + +namespace Qpid.Framing +{ + public class HeartbeatBody : IBody +{ + public const byte TYPE = 8; + public static AMQFrame FRAME = new HeartbeatBody().ToFrame(); + + public byte BodyType + { + get + { + return TYPE; + } + } + + public uint Size + { + get + { + return 0;//heartbeats we generate have no payload + } + } + + public void WritePayload(ByteBuffer buffer) + { + } + + public void PopulateFromBuffer(ByteBuffer buffer, uint size) + { + if (size > 0) + { + //allow other implementations to have a payload, but ignore it: + buffer.Skip((int) size); + } + } + + public AMQFrame ToFrame() + { + return new AMQFrame(0, this); + } +} +} diff --git a/dotnet/Qpid.Common/Framing/HeartbeatBodyFactory.cs b/dotnet/Qpid.Common/Framing/HeartbeatBodyFactory.cs new file mode 100644 index 0000000000..1e62f26c0d --- /dev/null +++ b/dotnet/Qpid.Common/Framing/HeartbeatBodyFactory.cs @@ -0,0 +1,32 @@ +/* + * + * 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. + * + */ +using Qpid.Buffer; + +namespace Qpid.Framing +{ + public class HeartbeatBodyFactory : IBodyFactory + { + public IBody CreateBody(ByteBuffer input) + { + return new HeartbeatBody(); + } + } +} diff --git a/dotnet/Qpid.Common/Framing/IBody.cs b/dotnet/Qpid.Common/Framing/IBody.cs new file mode 100644 index 0000000000..cc0dbd9b59 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/IBody.cs @@ -0,0 +1,63 @@ +/* + * + * 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. + * + */ +using Qpid.Buffer; + +namespace Qpid.Framing +{ + /// <summary> + /// An IBody is contained within a top level frame. As such, it is not en/decodable on its own but + /// is decoded as a step within a the overall en/decoding process. + /// </summary> + public interface IBody + { + /// <summary> + /// Gets the type. See RFC 006 for the meaning of "type" in this context. + /// </summary> + /// <value>The type.</value> + byte BodyType + { + get; + } + + /// <summary> + /// Get the size of the body + /// </summary> + /// <value>The size in bytes.</value> + uint Size + { + get; + } + + /// <summary> + /// Writes this instance to a buffer. + /// </summary> + /// <param name="buffer">The buffer.</param> + void WritePayload(ByteBuffer buffer); + + /// <summary> + /// Populates this instance from a buffer of data. + /// </summary> + /// <param name="buffer">The buffer.</param> + /// <param name="size">The size.</param> + /// <exception cref="AMQFrameDecodingException">If the buffer contains data that cannot be decoded</exception> + void PopulateFromBuffer(ByteBuffer buffer, uint size); + } +} diff --git a/dotnet/Qpid.Common/Framing/IBodyFactory.cs b/dotnet/Qpid.Common/Framing/IBodyFactory.cs new file mode 100644 index 0000000000..4f6bc7155e --- /dev/null +++ b/dotnet/Qpid.Common/Framing/IBodyFactory.cs @@ -0,0 +1,38 @@ +/* + * + * 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. + * + */ +using Qpid.Buffer; + +namespace Qpid.Framing +{ + /// <summary> + /// Any class that is capable of turning a stream of bytes into an AMQ structure must implement this interface. + /// </summary> + public interface IBodyFactory + { + /// <summary> + /// Creates the body. + /// </summary> + /// <param name="inbuf">The ByteBuffer containing data from the network</param> + /// <returns></returns> + /// <exception>AMQFrameDecodingException</exception> + IBody CreateBody(ByteBuffer inbuf); + } +} diff --git a/dotnet/Qpid.Common/Framing/IContentHeaderProperties.cs b/dotnet/Qpid.Common/Framing/IContentHeaderProperties.cs new file mode 100644 index 0000000000..df38738f1d --- /dev/null +++ b/dotnet/Qpid.Common/Framing/IContentHeaderProperties.cs @@ -0,0 +1,65 @@ +/* + * + * 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. + * + */ +using Qpid.Buffer; + +namespace Qpid.Framing +{ + /// <summary> + /// There will be an implementation of this interface for each content type. All content types have associated + /// header properties and this provides a way to encode and decode them. + /// </summary> + public interface IContentHeaderProperties + { + /// <summary> + /// Writes the property list to the buffer, in a suitably encoded form. + /// </summary> + /// <param name="buffer">The buffer to write to</param> + void WritePropertyListPayload(ByteBuffer buffer); + + /// <summary> + /// Populates the properties from buffer. + /// </summary> + /// <param name="buffer">The buffer to read from.</param> + /// <param name="propertyFlags">The property flags.</param> + /// <exception cref="AMQFrameDecodingException">Thrown when the buffer does not contain valid data</exception> + void PopulatePropertiesFromBuffer(ByteBuffer buffer, ushort propertyFlags); + + /// <summary> + /// Gets the size of the encoded property list in bytes. + /// </summary> + /// <value>The size of the property list in bytes</value> + uint PropertyListSize + { + get; + } + + /// <summary> + /// Gets the property flags. Property flags indicate which properties are set in the list. The + /// position and meaning of each flag is defined in the protocol specification for the particular + /// content type with which these properties are associated. + /// </summary> + /// <value>the flags as an unsigned integer</value> + ushort PropertyFlags + { + get; + } + } +} diff --git a/dotnet/Qpid.Common/Framing/IDataBlock.cs b/dotnet/Qpid.Common/Framing/IDataBlock.cs new file mode 100644 index 0000000000..57b2a8ed37 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/IDataBlock.cs @@ -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. + * + */ +using Qpid.Buffer; + +namespace Qpid.Framing +{ + /// <summary> + /// A data block represents something that has a size in bytes and the ability to write itself to a byte + /// buffer (similar to a byte array). It represents "top level" frames in the protocol specification. + /// </summary> + public interface IDataBlock : IEncodableAMQDataBlock + { + /// <summary> + /// Get the size of buffer needed to store the byte representation of this + /// frame. + /// </summary> + /// <returns>size in bytes</returns> + uint Size + { + get; + } + + /// <summary> + /// Writes the datablock to the specified buffer. + /// </summary> + /// <param name="buffer">The buffer to write to. Must be the correct size.</param> + void WritePayload(ByteBuffer buffer); + } +} diff --git a/dotnet/Qpid.Common/Framing/IEncodableAMQDataBlock.cs b/dotnet/Qpid.Common/Framing/IEncodableAMQDataBlock.cs new file mode 100644 index 0000000000..1d81d5ea82 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/IEncodableAMQDataBlock.cs @@ -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. + * + */ +namespace Qpid.Framing +{ + + /// <summary> + /// Marker interface to indicate to MINA that a data block should be encoded with the + /// single encoder/decoder that we have defined. + /// </summary> + public interface IEncodableAMQDataBlock + { + } +} diff --git a/dotnet/Qpid.Common/Framing/ProtocolInitiation.cs b/dotnet/Qpid.Common/Framing/ProtocolInitiation.cs new file mode 100644 index 0000000000..44c9dafd97 --- /dev/null +++ b/dotnet/Qpid.Common/Framing/ProtocolInitiation.cs @@ -0,0 +1,157 @@ +/* + * + * 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. + * + */ +using System; +using System.Collections; +using System.Configuration; +using System.Reflection; +using System.Xml; +using log4net; +using Qpid.Buffer; +using Qpid.Codec; +using Qpid.Codec.Demux; +using Qpid.Common; + +namespace Qpid.Framing +{ + public class ProtocolInitiation : IDataBlock, IEncodableAMQDataBlock + { + private static readonly ILog _log = LogManager.GetLogger(typeof(ProtocolInitiation)); + + public char[] Header = new char[]{'A','M','Q','P'}; + + private const byte CURRENT_PROTOCOL_CLASS = 1; + private const int CURRENT_PROTOCOL_INSTANCE = 1; + // FIXME: Needs to be tweakable from GRM.dll.config file. i.e. Major version 7 or 8 + + // FIXME: a configuration item for avoiding Basic.Qos (for OpenAMQ compatibility) + public static int CURRENT_PROTOCOL_VERSION_MAJOR = 8; // FIXME: put back to 7 for OpenAMQ! + private const int CURRENT_PROTOCOL_VERSION_MINOR = 0; + + public byte ProtocolClass = CURRENT_PROTOCOL_CLASS; + public byte ProtocolInstance = CURRENT_PROTOCOL_INSTANCE; + public byte ProtocolMajor = (byte)CURRENT_PROTOCOL_VERSION_MAJOR; + public byte ProtocolMinor = CURRENT_PROTOCOL_VERSION_MINOR; + + static ProtocolInitiation() + { + AssemblySettings settings = new AssemblySettings(); + string openAMQ = settings["OpenAMQ1d4Compatibility"]; + if (openAMQ.Equals("true")) + { + _log.Warn("Starting in OpenAMQ-1.0d4 compatibility mode. ProtocolMajorVersion is 7 and Basic.Qos will not be sent."); + CURRENT_PROTOCOL_VERSION_MAJOR = 7; + } + } + + public uint Size + { + get + { + return 4 + 1 + 1 + 1 + 1; + } + } + + public void WritePayload(ByteBuffer buffer) + { + foreach (char c in Header) + { + buffer.Put((byte) c); + } + buffer.Put(ProtocolClass); + buffer.Put(ProtocolInstance); + buffer.Put(ProtocolMajor); + buffer.Put(ProtocolMinor); + } + + /// <summary> + /// Populates from buffer. + /// </summary> + /// <param name="buffer">The buffer.</param> + public void PopulateFromBuffer(ByteBuffer buffer) + { + throw new AMQException("Method not implemented"); + } + + public class Decoder : IMessageDecoder + { + private bool _disabled = false; + + public MessageDecoderResult Decodable(ByteBuffer inbuf) + { + if (_disabled) + { + return MessageDecoderResult.NOT_OK; + } + if (inbuf.Remaining < 8) + { + return MessageDecoderResult.NEED_DATA; + } + else + { + char[] expected = new char[]{'A', 'M', 'Q', 'P'}; + for (int i = 0; i < 4; i++) + { + if (((char) inbuf.Get()) != expected[i]) + { + return MessageDecoderResult.NOT_OK; + } + } + return MessageDecoderResult.OK; + } + } + + /// <summary> + /// Decodes the specified session. + /// </summary> + /// <param name="session">The session.</param> + /// <param name="inbuf">The inbuf.</param> + /// <param name="outbuf">The outbuf.</param> + /// <returns></returns> + /// <exception cref="ProtocolViolationException">thrown if the frame violates the protocol</exception> + public MessageDecoderResult Decode(ByteBuffer inbuf, IProtocolDecoderOutput output) + { + byte[] header = new byte[4]; + inbuf.Get(header); + ProtocolInitiation pi = new ProtocolInitiation(); + pi.Header = new char[]{'A','M','Q','P'}; + pi.ProtocolClass = inbuf.Get(); + pi.ProtocolInstance = inbuf.Get(); + pi.ProtocolMajor = inbuf.Get(); + pi.ProtocolMinor = inbuf.Get(); + output.Write(pi); + return MessageDecoderResult.OK; + } + + public bool Disabled + { + set + { + _disabled = value; + } + } + } + + public override string ToString() + { + return String.Format("{0}{{Class={1} Instance={2} Major={3} Minor={4}}}", + GetType().Name, ProtocolClass, ProtocolInstance, ProtocolMajor, ProtocolMinor); + } + } +} |