diff options
Diffstat (limited to 'qpid/dotnet/client-010/client/transport')
53 files changed, 6328 insertions, 0 deletions
diff --git a/qpid/dotnet/client-010/client/transport/Binary.cs b/qpid/dotnet/client-010/client/transport/Binary.cs new file mode 100644 index 0000000000..ab829dfb69 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/Binary.cs @@ -0,0 +1,129 @@ +/*
+*
+* 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 org.apache.qpid.transport
+{
+
+
+ /// <summary>
+ /// Binary
+ /// </summary>
+
+ public sealed class Binary
+ {
+
+ private readonly byte[] bytes;
+ private readonly int offset_Renamed_Field;
+ private readonly int size_Renamed_Field;
+ private int hash = 0;
+
+ public Binary(byte[] bytes, int offset, int size)
+ {
+ if (offset + size > bytes.Length)
+ {
+ throw new System.IndexOutOfRangeException();
+ }
+
+ this.bytes = bytes;
+ offset_Renamed_Field = offset;
+ size_Renamed_Field = size;
+ }
+
+ public Binary(byte[] bytes):this(bytes, 0, bytes.Length)
+ {
+ }
+
+ public byte[] array()
+ {
+ return bytes;
+ }
+
+ public int offset()
+ {
+ return offset_Renamed_Field;
+ }
+
+ public int size()
+ {
+ return size_Renamed_Field;
+ }
+
+ public Binary slice(int low, int high)
+ {
+ int sz;
+
+ if (high < 0)
+ {
+ sz = size_Renamed_Field + high;
+ }
+ else
+ {
+ sz = high - low;
+ }
+
+ if (sz < 0)
+ {
+ sz = 0;
+ }
+
+ return new Binary(bytes, offset_Renamed_Field + low, sz);
+ }
+
+ public override int GetHashCode()
+ {
+ if (hash == 0)
+ {
+ int hc = 0;
+ for (int i = 0; i < size_Renamed_Field; i++)
+ {
+ hc = 31 * hc + (0xFF & bytes[offset_Renamed_Field + i]);
+ }
+ hash = hc;
+ }
+
+ return hash;
+ }
+
+ public override bool Equals(System.Object o)
+ {
+ if (!(o is Binary))
+ {
+ return false;
+ }
+
+ Binary buf = (Binary) o;
+ if (size_Renamed_Field != buf.size_Renamed_Field)
+ {
+ return false;
+ }
+
+ for (int i = 0; i < size_Renamed_Field; i++)
+ {
+ if (bytes[offset_Renamed_Field + i] != buf.bytes[buf.offset_Renamed_Field + i])
+ {
+ return false;
+ }
+ }
+
+ return true;
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/Binding.cs b/qpid/dotnet/client-010/client/transport/Binding.cs new file mode 100644 index 0000000000..25884fc772 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/Binding.cs @@ -0,0 +1,34 @@ +/*
+*
+* 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;
+
+namespace org.apache.qpid.transport
+{
+ /// <summary>
+ /// Binding
+ /// </summary>
+ internal interface Binding<E, T>
+ {
+ E endpoint(Sender<T> sender);
+
+ Receiver<T> receiver<T>(E endpoint) where T : EventArgs;
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/Channel.cs b/qpid/dotnet/client-010/client/transport/Channel.cs new file mode 100644 index 0000000000..d7ccc6eec8 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/Channel.cs @@ -0,0 +1,175 @@ +/*
+*
+* 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 org.apache.qpid.transport.network;
+using org.apache.qpid.transport.util;
+using SessionDelegate=org.apache.qpid.transport.SessionDelegate;
+
+namespace org.apache.qpid.transport
+{
+ /// <summary>
+ /// Channel
+ /// </summary>
+ public class Channel : Invoker, ProtocolDelegate<Object>
+ {
+ private static readonly Logger log = Logger.get(typeof (Channel));
+
+ private readonly Connection _connection;
+ private readonly int _channel;
+ private readonly MethodDelegate<Channel> _methoddelegate;
+ private readonly SessionDelegate _sessionDelegate;
+ // session may be null
+ private Session _session;
+
+ public Channel(Connection connection, int channel, SessionDelegate sessionDelegate)
+ {
+ _connection = connection;
+ _channel = channel;
+ _methoddelegate = new ChannelDelegate();
+ _sessionDelegate = sessionDelegate;
+ }
+
+ public Connection Connection
+ {
+ get { return _connection; }
+ }
+
+ // Invoked when a network event is received
+ public void On_ReceivedEvent(object sender, ReceivedPayload<ProtocolEvent> payload)
+ {
+ if (payload.Payload.Channel == _channel)
+ {
+ payload.Payload.ProcessProtocolEvent(null, this);
+ }
+ }
+
+ #region ProtocolDelegate<T>
+
+ public void Init(Object v, ProtocolHeader hdr)
+ {
+ _connection.ConnectionDelegate.init(this, hdr);
+ }
+
+ public void Control(Object v, Method method)
+ {
+ switch (method.EncodedTrack)
+ {
+ case Frame.L1:
+ method.dispatch(this, _connection.ConnectionDelegate);
+ break;
+ case Frame.L2:
+ method.dispatch(this, _methoddelegate);
+ break;
+ case Frame.L3:
+ method.ProcessProtocolEvent(_session, _sessionDelegate);
+ break;
+ default:
+ throw new Exception("unknown track: " + method.EncodedTrack);
+ }
+ }
+
+ public void Command(Object v, Method method)
+ {
+ method.ProcessProtocolEvent(_session, _sessionDelegate);
+ }
+
+ public void Error(Object v, ProtocolError error)
+ {
+ throw new Exception(error.Message);
+ }
+
+ #endregion
+
+ public void exception(Exception t)
+ {
+ _session.exception(t);
+ }
+
+ public void closedFromConnection()
+ {
+ log.debug("channel closed: ", this);
+ if (_session != null)
+ {
+ _session.closed();
+ }
+ }
+
+ public void closed()
+ {
+ log.debug("channel closed: ", this);
+ if (_session != null)
+ {
+ _session.closed();
+ }
+ _connection.removeChannel(_channel);
+ }
+
+ public int EncodedChannel
+ {
+ get { return _channel; }
+ }
+
+ public Session Session
+ {
+ get { return _session; }
+ set { _session = value; }
+ }
+
+ public void closeCode(ConnectionClose close)
+ {
+ if (_session != null)
+ {
+ _session.closeCode(close);
+ }
+ }
+
+ private void emit(ProtocolEvent pevent)
+ {
+ pevent.Channel = _channel;
+ _connection.send(pevent);
+ }
+
+ public void method(Method m)
+ {
+ emit(m);
+
+ if (!m.Batch)
+ {
+ _connection.flush();
+ }
+ }
+
+ protected override void invoke(Method m)
+ {
+ method(m);
+ }
+
+ public override Future<T> invoke<T>(Method m, Future<T> future)
+ {
+ throw new Exception("UnsupportedOperation");
+ }
+
+ public String toString()
+ {
+ return String.Format("{0}:{1}", _connection, _channel);
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/ChannelDelegate.cs b/qpid/dotnet/client-010/client/transport/ChannelDelegate.cs new file mode 100644 index 0000000000..5e9fad6e40 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/ChannelDelegate.cs @@ -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.
+*
+*/
+namespace org.apache.qpid.transport
+{
+
+
+ /// <summary>
+ /// ChannelDelegate
+ ///
+ /// </summary>
+
+ class ChannelDelegate : MethodDelegate<Channel>
+ {
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/ClientDelegate.cs b/qpid/dotnet/client-010/client/transport/ClientDelegate.cs new file mode 100644 index 0000000000..fa0062ef19 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/ClientDelegate.cs @@ -0,0 +1,35 @@ +/*
+*
+* 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 org.apache.qpid.transport;
+
+namespace org.apache.qpid.transport
+{
+ abstract class ClientDelegate : ConnectionDelegate
+ {
+ public override void init(Channel ch, ProtocolHeader hdr)
+ {
+ if (hdr.Major != 0 && hdr.Minor != 10)
+ {
+ throw new ProtocolVersionException((sbyte) hdr.Major, (sbyte) hdr.Minor);
+ }
+ }
+ }
+}
diff --git a/qpid/dotnet/client-010/client/transport/Connection.cs b/qpid/dotnet/client-010/client/transport/Connection.cs new file mode 100644 index 0000000000..863f90e093 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/Connection.cs @@ -0,0 +1,172 @@ +/*
+*
+* 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.Generic;
+using Logger = org.apache.qpid.transport.util.Logger;
+
+namespace org.apache.qpid.transport
+{
+ /// <summary>
+ /// Connection
+ /// </summary>
+ public class Connection
+ {
+ private static readonly Logger log = Logger.get(typeof (Connection));
+
+ private readonly Sender<ProtocolEvent> _sender;
+ private readonly ConnectionDelegate _connDdelegate;
+ private int _channelMax = 1;
+ private int _connectionId;
+ private readonly Receiver<ReceivedPayload<ProtocolEvent>> _receiver;
+
+ private readonly Dictionary<int, Channel> _channels = new Dictionary<int, Channel>();
+
+ public Connection(Receiver<ReceivedPayload<ProtocolEvent>> receiver, Sender<ProtocolEvent> sender, ConnectionDelegate connDdelegate)
+ {
+ _receiver = receiver;
+ _sender = sender;
+ _connDdelegate = connDdelegate;
+ }
+
+ public int ConnectionId
+ {
+ get { return _connectionId; }
+ set { _connectionId = value; }
+ }
+
+ public ConnectionDelegate ConnectionDelegate
+ {
+ get { return _connDdelegate; }
+ }
+
+ public int ChannelMax
+ {
+ get { return _channelMax; }
+ set { _channelMax = value; }
+ }
+
+ public void send(ProtocolEvent pevent)
+ {
+ log.debug("SEND: [{0}] {1}", this, pevent);
+ _sender.send(pevent);
+ }
+
+ public void flush()
+ {
+ log.debug("FLUSH: [{0}]", this);
+ _sender.flush();
+ }
+
+
+ public Channel getChannel()
+ {
+ lock (_channels)
+ {
+ for (int i = 0; i < ChannelMax; i++)
+ {
+ if (!_channels.ContainsKey(i))
+ {
+ return getChannel(i);
+ }
+ }
+ throw new Exception("no more _channels available");
+ }
+ }
+
+ public Channel getChannel(int number)
+ {
+ lock (_channels)
+ {
+ Channel channel = null;
+ if (_channels.Count > 0)
+ {
+ if( _channels.ContainsKey(number))
+ channel = _channels[number];
+ }
+ if (channel == null)
+ {
+ channel = new Channel(this, number, _connDdelegate.getSessionDelegate());
+ _receiver.Received += channel.On_ReceivedEvent;
+ _channels.Add(number, channel);
+ }
+ return channel;
+ }
+ }
+
+ public void removeChannel(int number)
+ {
+ lock (_channels)
+ {
+ _receiver.Received -= _channels[number].On_ReceivedEvent;
+ _channels.Remove(number);
+ }
+ }
+
+ public void On_ReceivedEvent(object sender, ReceivedPayload<ProtocolEvent> payload)
+ {
+ log.debug("RECV: [{0}] {1}", this, payload.Payload);
+ if (_channels.ContainsKey(payload.Payload.Channel)) return;
+ Channel channel = getChannel(payload.Payload.Channel);
+ channel.On_ReceivedEvent(sender, payload);
+ }
+
+ public void On_ReceivedException(Object sender, ExceptionArgs arg)
+ {
+ _connDdelegate.exception(arg.Exception);
+ }
+
+ public void On_ReceivedClosed(Object sender, EventArgs arg)
+ {
+ log.debug("Connection closed: {0}", this);
+ lock (_channels)
+ {
+ foreach (Channel ch in _channels.Values)
+ {
+ ch.closedFromConnection();
+ }
+ }
+ _channels.Clear();
+ _connDdelegate.closed();
+ }
+
+
+ public void closeCode(ConnectionClose close)
+ {
+ lock (_channels)
+ {
+ foreach (Channel ch in _channels.Values)
+ {
+ ch.closeCode(close);
+ }
+ }
+ }
+
+ public void close()
+ {
+ _sender.close();
+ }
+
+ public String toString()
+ {
+ return String.Format("conn:{0}", this);
+ }
+ }
+}
diff --git a/qpid/dotnet/client-010/client/transport/ConnectionDelegate.cs b/qpid/dotnet/client-010/client/transport/ConnectionDelegate.cs new file mode 100644 index 0000000000..ae3fb11fdd --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/ConnectionDelegate.cs @@ -0,0 +1,109 @@ +/*
+*
+* 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.Generic;
+using System.Threading;
+using Logger = org.apache.qpid.transport.util.Logger;
+
+namespace org.apache.qpid.transport
+{
+ /// <summary>
+ /// ConnectionDelegate
+ ///
+ /// Currently only implemented client specific methods
+ /// </summary>
+ public abstract class ConnectionDelegate : MethodDelegate<Channel>
+ {
+ private static readonly Logger log = Logger.get(typeof(ConnectionDelegate));
+ private String _virtualHost;
+
+ private ManualResetEvent _negotiationComplete;
+
+ public abstract SessionDelegate getSessionDelegate();
+
+ public abstract void exception(Exception t);
+
+ public abstract void closed();
+
+ public void setCondition(ManualResetEvent negotiationComplete)
+ {
+ _negotiationComplete = negotiationComplete;
+ }
+
+ public virtual void init(Channel ch, ProtocolHeader hdr)
+ {
+ ch.Connection.send(new ProtocolHeader((byte)1, hdr.Major, hdr.Minor));
+ List<Object> plain = new List<Object>();
+ plain.Add("PLAIN");
+ List<Object> utf8 = new List<Object>();
+ utf8.Add("utf8");
+ ch.connectionStart(null, plain, utf8);
+ }
+
+ public String VirtualHost
+ {
+ get { return _virtualHost; }
+ set { _virtualHost = value; }
+ }
+
+ // ----------------------------------------------
+ // Client side
+ //-----------------------------------------------
+ public override void connectionStart(Channel context, ConnectionStart mstruct)
+ {
+ // todo SASL
+ Dictionary<String, Object> props = new Dictionary<String, Object>();
+ context.connectionStartOk(props, null, null, "utf8");
+ }
+
+ public override void connectionSecure(Channel context, ConnectionSecure mstruct)
+ { // todo SASL
+ context.connectionSecureOk(new byte[0]);
+ }
+
+ public override void connectionTune(Channel context, ConnectionTune mstruct)
+ {
+ context.Connection.ChannelMax = mstruct.getChannelMax();
+ context.connectionTuneOk(mstruct.getChannelMax(), mstruct.getMaxFrameSize(), mstruct.getHeartbeatMax());
+ context.connectionOpen(_virtualHost, null, Option.INSIST);
+ }
+
+ public override void connectionOpenOk(Channel context, ConnectionOpenOk mstruct)
+ {
+ List<Object> knownHosts = mstruct.getKnownHosts();
+ if (_negotiationComplete != null)
+ {
+ _negotiationComplete.Set();
+ }
+ }
+
+ public override void connectionRedirect(Channel context, ConnectionRedirect mstruct)
+ {
+ // not going to bother at the moment
+ }
+
+ public override void connectionClose(Channel ch, ConnectionClose close)
+ {
+ ch.Connection.closeCode(close);
+ ch.connectionCloseOk();
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/Field.cs b/qpid/dotnet/client-010/client/transport/Field.cs new file mode 100644 index 0000000000..e8c3cb3497 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/Field.cs @@ -0,0 +1,75 @@ +/*
+*
+* 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 Decoder = org.apache.qpid.transport.codec.Decoder;
+using Encoder = org.apache.qpid.transport.codec.Encoder;
+
+namespace org.apache.qpid.transport
+{
+ /// <summary>
+ /// Field
+ /// </summary>
+ public abstract class Field<C, T>
+ {
+ private C container;
+ private T type;
+ private String name;
+ private int index;
+
+ protected Field(C container, T type, String name, int index)
+ {
+ this.container = container;
+ this.type = type;
+ this.name = name;
+ this.index = index;
+ }
+
+ public C Container
+ {
+ get { return container; }
+ }
+
+ public T Type
+ {
+ get { return type; }
+ }
+
+ public String Name
+ {
+ get { return name; }
+ }
+
+ public int Index
+ {
+ get { return index; }
+ }
+
+ public abstract bool has(Object mystruct);
+
+ public abstract void has(Object mystruct, bool value);
+
+ public abstract T get(Object mystruct);
+
+ public abstract void read(Decoder dec, Object mystruct);
+
+ public abstract void write(Encoder enc, Object mystruct);
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/Future.cs b/qpid/dotnet/client-010/client/transport/Future.cs new file mode 100644 index 0000000000..f636e91914 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/Future.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.
+*
+*/
+
+namespace org.apache.qpid.transport
+{
+ /// <summary>
+ /// Future
+ /// </summary>
+ public interface Future<T>
+ {
+ T Result
+ {
+ get; set;
+ }
+
+ Session Session
+ { set;
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/Header.cs b/qpid/dotnet/client-010/client/transport/Header.cs new file mode 100644 index 0000000000..74b8a7fe2e --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/Header.cs @@ -0,0 +1,83 @@ +/*
+*
+* 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.Generic;
+using System.Text;
+
+namespace org.apache.qpid.transport
+{
+ /// <summary>
+ /// Header
+ /// </summary>
+ public class Header
+ {
+ private readonly Struct[] _mystructs;
+
+ public Header(List<Struct> structs)
+ : this(structs.ToArray())
+ {
+ }
+
+ public Header(params Struct[] structs)
+ {
+ _mystructs = structs;
+ }
+
+ public Struct[] Structs
+ {
+ get { return _mystructs; }
+ }
+
+
+ public Struct get(Struct klass)
+ {
+ foreach (Struct st in _mystructs)
+ {
+ if (Equals(st.GetType(), klass.GetType()))
+ {
+ return st;
+ }
+ }
+ return null;
+ }
+
+ public String toString()
+ {
+ StringBuilder str = new StringBuilder();
+ str.Append(" Header(");
+ bool first = true;
+ foreach (Struct s in _mystructs)
+ {
+ if (first)
+ {
+ first = false;
+ }
+ else
+ {
+ str.Append(", ");
+ }
+ str.Append(s);
+ }
+ str.Append(")");
+ return str.ToString();
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/Method.cs b/qpid/dotnet/client-010/client/transport/Method.cs new file mode 100644 index 0000000000..0f9a84d778 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/Method.cs @@ -0,0 +1,150 @@ +/*
+*
+* 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.IO;
+using System.Text;
+using Frame = org.apache.qpid.transport.network.Frame;
+
+namespace org.apache.qpid.transport
+{
+ /// <summary>
+ /// Method
+ /// </summary>
+ public abstract class Method : Struct, ProtocolEvent
+ {
+ public static Method create(int type)
+ {
+ return (Method) StructFactory.createInstruction(type);
+ }
+
+ // XXX: command subclass?
+ private int id;
+ private int channel;
+ private bool idSet;
+ private bool sync;
+ private bool batch;
+
+ public int Id
+ {
+ get { return id; }
+ set
+ {
+ id = value;
+ idSet = true;
+ }
+ }
+
+
+ public bool Sync
+ {
+ get { return sync; }
+ set { sync = value; }
+ }
+
+ public bool Batch
+ {
+ get { return batch; }
+ set { batch = value; }
+ }
+
+ public abstract bool hasPayload();
+
+ public virtual Header Header
+ {
+ get { return null; }
+ set { throw new Exception(); }
+ }
+
+ public virtual MemoryStream Body
+ {
+ get { return null; }
+ set { throw new Exception(); }
+ }
+
+
+ public abstract void dispatch<C>(C context, MethodDelegate<C> mdelegate );
+
+ #region ProtocolEvent
+
+ public int Channel
+ {
+ get { return channel; }
+ set { channel = value; }
+ }
+
+ public abstract byte EncodedTrack { get; set; }
+
+ public void ProcessProtocolEvent<C>(C context, ProtocolDelegate<C> protocoldelegate)
+ {
+ if (EncodedTrack == Frame.L4)
+ {
+ protocoldelegate.Command(context, this);
+ }
+ else
+ {
+ protocoldelegate.Control(context, this);
+ }
+ }
+
+ #endregion
+
+ public new String toString()
+ {
+ StringBuilder str = new StringBuilder();
+
+ str.Append("ch=");
+ str.Append(channel);
+
+ if (EncodedTrack == Frame.L4 && idSet)
+ {
+ str.Append(" id=");
+ str.Append(id);
+ }
+
+ if (sync || batch)
+ {
+ str.Append(" ");
+ str.Append("[");
+ if (Sync)
+ {
+ str.Append("S");
+ }
+ if (Batch)
+ {
+ str.Append("B");
+ }
+ str.Append("]");
+ }
+ str.Append(" ");
+ str.Append(base.toString());
+ if (Header != null)
+ {
+ str.Append(Header.toString());
+ }
+ if (Body != null)
+ {
+ str.Append("\n body=");
+ str.Append(Body.ToString());
+ }
+ return str.ToString();
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/ProtocolDelegate.cs b/qpid/dotnet/client-010/client/transport/ProtocolDelegate.cs new file mode 100644 index 0000000000..d269b01b79 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/ProtocolDelegate.cs @@ -0,0 +1,37 @@ +/*
+*
+* 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 org.apache.qpid.transport
+{
+ /// <summary>
+ /// ProtocolDelegate
+ /// </summary>
+ public interface ProtocolDelegate<T>
+ {
+ void Init(T context, ProtocolHeader header);
+
+ void Control(T context, Method control);
+
+ void Command(T context, Method command);
+
+ void Error(T context, ProtocolError error);
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/ProtocolError.cs b/qpid/dotnet/client-010/client/transport/ProtocolError.cs new file mode 100644 index 0000000000..933a18c78a --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/ProtocolError.cs @@ -0,0 +1,86 @@ +/*
+*
+* 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 NetworkDelegate = org.apache.qpid.transport.network.NetworkDelegate;
+using NetworkEvent = org.apache.qpid.transport.network.NetworkEvent;
+
+namespace org.apache.qpid.transport
+{
+ /// <summary>
+ /// ProtocolError
+ /// </summary>
+ public sealed class ProtocolError : NetworkEvent, ProtocolEvent
+ {
+ private int channel;
+ private byte track;
+ private String format;
+ private Object[] args;
+
+ public ProtocolError(byte track, String format, params Object[] args)
+ {
+ this.track = track;
+ this.format = format;
+ this.args = args;
+ }
+
+ #region NetworkEvent Methods
+
+ public void ProcessNetworkEvent(NetworkDelegate ndelegate)
+ {
+ ndelegate.Error(this);
+ }
+
+ #endregion
+
+ #region ProtocolEvent Methods
+
+ public int Channel
+ {
+ get { return channel; }
+ set { channel = value; }
+ }
+
+ public byte EncodedTrack
+ {
+ get { return track; }
+ set { throw new NotImplementedException(); }
+ }
+
+ public void ProcessProtocolEvent<C>(C context, ProtocolDelegate<C> protocoldelegate)
+ {
+ protocoldelegate.Error(context, this);
+ }
+
+ #endregion
+
+ public String Message
+ {
+ get { return String.Format(format, args); }
+ }
+
+
+ public String toString()
+ {
+ return String.Format("protocol error: {0}", Message);
+ }
+
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/ProtocolEvent.cs b/qpid/dotnet/client-010/client/transport/ProtocolEvent.cs new file mode 100644 index 0000000000..0f6529dfd3 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/ProtocolEvent.cs @@ -0,0 +1,42 @@ +/*
+*
+* 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 org.apache.qpid.transport
+{
+ /// <summary>
+ /// ProtocolEvent
+ /// </summary>
+ public interface ProtocolEvent
+ {
+ int Channel
+ {
+ get;
+ set;
+ }
+
+ byte EncodedTrack
+ {
+ set;
+ get;
+ }
+
+ void ProcessProtocolEvent<C>(C context, ProtocolDelegate<C> protocoldelegate);
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/ProtocolHeader.cs b/qpid/dotnet/client-010/client/transport/ProtocolHeader.cs new file mode 100644 index 0000000000..39a3f5670c --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/ProtocolHeader.cs @@ -0,0 +1,125 @@ +/*
+*
+* 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.IO;
+using System.Text;
+using NetworkDelegate = org.apache.qpid.transport.network.NetworkDelegate;
+using NetworkEvent = org.apache.qpid.transport.network.NetworkEvent;
+using Frame = org.apache.qpid.transport.network.Frame;
+
+namespace org.apache.qpid.transport
+{
+ /// <summary> ProtocolHeader
+ ///
+ /// </summary>
+ public sealed class ProtocolHeader : NetworkEvent, ProtocolEvent
+ {
+ private readonly char[] AMQP = new char[] {'A', 'M', 'Q', 'P'};
+ private const byte CLASS = 1;
+
+ private readonly byte instance;
+ private readonly byte major;
+ private readonly byte minor;
+ private int channel;
+
+ public ProtocolHeader(byte instance, byte major, byte minor)
+ {
+ this.instance = instance;
+ this.major = major;
+ this.minor = minor;
+ }
+
+ public ProtocolHeader(int instance, int major, int minor) : this((byte)instance, (byte)major, (byte)minor)
+ {
+ }
+
+ #region NetworkEvent Methods
+
+ public void ProcessNetworkEvent(NetworkDelegate ndelegate)
+ {
+ ndelegate.Init(this);
+ }
+
+ #endregion
+
+ #region ProtocolEvent Methods
+
+ public int Channel
+ {
+ get
+ {
+ return channel;
+ }
+ set
+ {
+ channel = value;
+ }
+ }
+
+ public byte EncodedTrack
+ {
+ get
+ {
+ return Frame.L1;
+ }
+ set { throw new NotImplementedException(); }
+ }
+
+ public void ProcessProtocolEvent<C>(C context, ProtocolDelegate<C> protocoldelegate)
+ {
+ protocoldelegate.Init(context, this);
+ }
+
+ #endregion
+
+ public byte Instance
+ {
+ get { return instance; }
+ }
+
+ public byte Major
+ {
+ get { return major; }
+ }
+
+ public byte Minor
+ {
+ get { return minor; }
+ }
+
+ public MemoryStream ToMemoryStream()
+ {
+ MemoryStream buf = new MemoryStream(8);
+ BinaryWriter writer = new BinaryWriter(buf);
+ writer.Write(AMQP);
+ writer.Write(CLASS);
+ writer.Write(instance);
+ writer.Write((sbyte) major);
+ writer.Write((sbyte) minor);
+ return buf;
+ }
+
+ public String toString()
+ {
+ return String.Format("AMQP.{0:d} {1:d}-{2:d}", instance, major, minor);
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/Range.cs b/qpid/dotnet/client-010/client/transport/Range.cs new file mode 100644 index 0000000000..37e0761794 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/Range.cs @@ -0,0 +1,117 @@ +/*
+*
+* 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.Generic;
+using org.apache.qpid.transport.util;
+
+namespace org.apache.qpid.transport
+{
+
+ /// <summary>
+ /// Range
+ /// </summary>
+
+
+ public sealed class Range
+ {
+ private int lower;
+ private int upper;
+
+ public Range(int lower, int upper)
+ {
+ this.lower = lower;
+ this.upper = upper;
+ }
+
+ public int Lower
+ {
+ get { return lower; }
+ set { lower = value; }
+ }
+ public int Upper
+ {
+ get { return upper; }
+ set { upper = value; }
+ }
+
+ public bool includes(int value)
+ {
+ return Serial.le(lower, value) && Serial.le(value, upper);
+ }
+
+ public bool includes(Range range)
+ {
+ return includes(range.lower) && includes(range.upper);
+ }
+
+ public bool intersects(Range range)
+ {
+ return (includes(range.lower) || includes(range.upper) ||
+ range.includes(lower) || range.includes(upper));
+ }
+
+ public bool touches(Range range)
+ {
+ return (intersects(range) ||
+ includes(range.upper + 1) || includes(range.lower - 1) ||
+ range.includes(upper + 1) || range.includes(lower - 1));
+ }
+
+ public Range span(Range range)
+ {
+ return new Range(Serial.min(lower, range.lower), Serial.max(upper, range.upper));
+ }
+
+ public List<Range> subtract(Range range)
+ {
+ List<Range> result = new List<Range>();
+
+ if (includes(range.lower) && Serial.le(lower, range.lower - 1))
+ {
+ result.Add(new Range(lower, range.lower - 1));
+ }
+
+ if (includes(range.upper) && Serial.le(range.upper + 1, upper))
+ {
+ result.Add(new Range(range.upper + 1, upper));
+ }
+
+ if (result.Count == 0 && !range.includes(this))
+ {
+ result.Add(this);
+ }
+
+ return result;
+ }
+
+ public Range intersect(Range range)
+ {
+ int l = Serial.max(lower, range.lower);
+ int r = Serial.min(upper, range.upper);
+ return Serial.gt(l, r) ? null : new Range(l, r);
+ }
+
+ public String toString()
+ {
+ return "[" + lower + ", " + upper + "]";
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/RangeSet.cs b/qpid/dotnet/client-010/client/transport/RangeSet.cs new file mode 100644 index 0000000000..d0ee1d4916 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/RangeSet.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 System.Collections.Generic;
+using System.Text;
+using org.apache.qpid.transport.util;
+
+namespace org.apache.qpid.transport
+{
+
+
+ /// <summary>
+ /// RangeSet
+ /// </summary>
+
+ public sealed class RangeSet : IEnumerable<Range>
+ {
+ private readonly LinkedList<Range> ranges = new LinkedList<Range>();
+
+ IEnumerator IEnumerable.GetEnumerator()
+ {
+ return GetEnumerator();
+ }
+
+ public IEnumerator<Range> GetEnumerator()
+ {
+ return ranges.GetEnumerator();
+ }
+
+
+ public int size()
+ {
+ return ranges.Count;
+ }
+
+
+
+ public Range getFirst()
+ {
+ return ranges.First.Value;
+ }
+
+ public bool includes(Range range)
+ {
+ foreach (Range r in this)
+ {
+ if (r.includes(range))
+ {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ public bool includes(int n)
+ {
+ foreach (Range r in this)
+ {
+ if (r.includes(n))
+ {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ public void add(Range range)
+ {
+ foreach (Range r in ranges)
+ {
+ if (range.touches(r))
+ {
+ ranges.Remove(r);
+ range = range.span(r);
+ }
+ else if (Serial.lt(range.Upper, r.Lower ))
+ {
+ ranges.AddBefore(ranges.Find(r), range);
+ return;
+ }
+ }
+ ranges.AddLast(range);
+ }
+
+ public void add(int lower, int upper)
+ {
+ add(new Range(lower, upper));
+ }
+
+ public void add(int value)
+ {
+ add(value, value);
+ }
+
+ public void clear()
+ {
+ ranges.Clear();
+ }
+
+ public RangeSet copy()
+ {
+ RangeSet copy = new RangeSet();
+ foreach( Range r in ranges)
+ {
+ copy.ranges.AddLast(r);
+ }
+ return copy;
+ }
+
+ public String toString()
+ {
+ StringBuilder str = new StringBuilder();
+ str.Append("{");
+ bool first = true;
+ foreach (Range range in ranges)
+ {
+ if (first)
+ {
+ first = false;
+ }
+ else
+ {
+ str.Append(", ");
+ }
+ str.Append(range);
+ }
+ str.Append("}");
+ return str.ToString();
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/ReceivedPayload.cs b/qpid/dotnet/client-010/client/transport/ReceivedPayload.cs new file mode 100644 index 0000000000..792ca4f977 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/ReceivedPayload.cs @@ -0,0 +1,43 @@ +/*
+*
+* 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;
+
+namespace org.apache.qpid.transport
+{
+ public class ReceivedPayload<T> : EventArgs
+ {
+ public ReceivedPayload()
+ {
+ }
+
+ public ReceivedPayload(T payload)
+ {
+ m_payload = payload;
+ }
+ private T m_payload;
+
+ public T Payload
+ {
+ get { return m_payload; }
+ set { m_payload = value; }
+ }
+ }
+}
diff --git a/qpid/dotnet/client-010/client/transport/Receiver.cs b/qpid/dotnet/client-010/client/transport/Receiver.cs new file mode 100644 index 0000000000..be28d4ab6e --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/Receiver.cs @@ -0,0 +1,39 @@ +/*
+*
+* 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 org.apache.qpid.transport;
+
+namespace org.apache.qpid.transport
+{
+ /// <summary>
+ /// a receiver will raise an event when:
+ /// - data is received
+ /// - an exception is thrown
+ /// - it is closed
+ /// </summary>
+ public interface Receiver <T> where T : EventArgs
+ {
+ event EventHandler<T> Received;
+ event EventHandler<ExceptionArgs> Exception;
+ event EventHandler Closed;
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/Result.cs b/qpid/dotnet/client-010/client/transport/Result.cs new file mode 100644 index 0000000000..bde1a2f98c --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/Result.cs @@ -0,0 +1,35 @@ +/*
+*
+* 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;
+namespace org.apache.qpid.transport
+{
+
+
+ /// <summary> Result
+ ///
+ /// </summary>
+ /// <author> Rafael H. Schloming
+ /// </author>
+
+ public abstract class Result:Struct
+ {
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/Sender.cs b/qpid/dotnet/client-010/client/transport/Sender.cs new file mode 100644 index 0000000000..564dd83789 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/Sender.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.
+*
+*/
+namespace org.apache.qpid.transport
+{
+ /// <summary>
+ /// Sender
+ /// </summary>
+ public interface Sender<T>
+ {
+ void send(T msg);
+ void flush();
+ void close();
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/Session.cs b/qpid/dotnet/client-010/client/transport/Session.cs new file mode 100644 index 0000000000..54c5c2c462 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/Session.cs @@ -0,0 +1,513 @@ +/*
+*
+* 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.Generic;
+using System.IO;
+using System.Threading;
+using org.apache.qpid.transport;
+using org.apache.qpid.transport.util;
+using Frame = org.apache.qpid.transport.network.Frame;
+using Logger = org.apache.qpid.transport.util.Logger;
+
+
+namespace org.apache.qpid.transport
+{
+ /// <summary>
+ /// Session
+ ///
+ /// </summary>
+ public class Session : Invoker
+ {
+ private static readonly Logger log = Logger.get(typeof (Session));
+ private static readonly bool ENABLE_REPLAY;
+
+ static Session()
+ {
+ const string enableReplay = "enable_command_replay";
+ try
+ {
+ String var = Environment.GetEnvironmentVariable(enableReplay);
+ if (var != null)
+ {
+ ENABLE_REPLAY = bool.Parse(var);
+ }
+ }
+ catch (Exception)
+ {
+ ENABLE_REPLAY = false;
+ }
+ }
+
+ private readonly byte[] _name;
+ private const long _timeout = 600000;
+ private bool _autoSync = false;
+
+ // channel may be null
+ private Channel _channel;
+
+ // incoming command count
+ private int _commandsIn = 0;
+ // completed incoming commands
+ private readonly Object _processedLock = new Object();
+ private RangeSet _processed = new RangeSet();
+ private int _maxProcessed = - 1;
+ private int _syncPoint = -1;
+
+ // outgoing command count
+ private int _commandsOut = 0;
+ private readonly Dictionary<int, Method> _commands = new Dictionary<int, Method>();
+ private int _maxComplete = - 1;
+ private bool _needSync = false;
+ private bool _closed;
+ private readonly Dictionary<int, Future<Struct>> _results = new Dictionary<int, Future<Struct>>();
+ private readonly List<ExecutionException> _exceptions = new List<ExecutionException>();
+
+
+ public bool Closed
+ {
+ get
+ {
+ lock (this)
+ {
+ return _closed;
+ }
+ }
+ set
+ {
+ lock (this)
+ {
+ _closed = value;
+ }
+ }
+ }
+
+ public Session(byte[] name)
+ {
+ _name = name;
+ }
+
+ public byte[] getName()
+ {
+ return _name;
+ }
+
+ public void setAutoSync(bool value)
+ {
+ lock (_commands)
+ {
+ _autoSync = value;
+ }
+ }
+
+ public Dictionary<int, Method> getOutstandingCommands()
+ {
+ return _commands;
+ }
+
+ public int getCommandsOut()
+ {
+ return _commandsOut;
+ }
+
+ public int CommandsIn
+ {
+ get { return _commandsIn; }
+ set { _commandsIn = value; }
+ }
+
+ public int nextCommandId()
+ {
+ return _commandsIn++;
+ }
+
+ public void identify(Method cmd)
+ {
+ int id = nextCommandId();
+ cmd.Id = id;
+
+ if (log.isDebugEnabled())
+ {
+ log.debug("ID: [{0}] %{1}", _channel, id);
+ }
+
+ //if ((id % 65536) == 0)
+ if ((id & 0xff) == 0)
+ {
+ flushProcessed(Option.TIMELY_REPLY);
+ }
+ }
+
+ public void processed(Method command)
+ {
+ processed(command.Id);
+ }
+
+ public void processed(int command)
+ {
+ processed(new Range(command, command));
+ }
+
+ public void processed(int lower, int upper)
+ {
+ processed(new Range(lower, upper));
+ }
+
+ public void processed(Range range)
+ {
+ log.debug("{0} processed({1})", this, range);
+
+ bool flush;
+ lock (_processedLock)
+ {
+ _processed.add(range);
+ Range first = _processed.getFirst();
+ int lower = first.Lower;
+ int upper = first.Upper;
+ int old = _maxProcessed;
+ if (Serial.le(lower, _maxProcessed + 1))
+ {
+ _maxProcessed = Serial.max(_maxProcessed, upper);
+ }
+ flush = Serial.lt(old, _syncPoint) && Serial.ge(_maxProcessed, _syncPoint);
+ _syncPoint = _maxProcessed;
+ }
+ if (flush)
+ {
+ flushProcessed();
+ }
+ }
+
+ public void flushProcessed(params Option[] options)
+ {
+ RangeSet copy;
+ lock (_processedLock)
+ {
+ copy = _processed.copy();
+ }
+ sessionCompleted(copy, options);
+ }
+
+ public void knownComplete(RangeSet kc)
+ {
+ lock (_processedLock)
+ {
+ RangeSet newProcessed = new RangeSet();
+ foreach (Range pr in _processed)
+ {
+ foreach (Range kr in kc)
+ {
+ foreach (Range r in pr.subtract(kr))
+ {
+ newProcessed.add(r);
+ }
+ }
+ }
+ _processed = newProcessed;
+ }
+ }
+
+ public void syncPoint()
+ {
+ int id = CommandsIn - 1;
+ log.debug("{0} synced to {1}", this, id);
+ bool flush;
+ lock (_processedLock)
+ {
+ _syncPoint = id;
+ flush = Serial.ge(_maxProcessed, _syncPoint);
+ }
+ if (flush)
+ {
+ flushProcessed();
+ }
+ }
+
+ public void attach(Channel channel)
+ {
+ _channel = channel;
+ _channel.Session = this;
+ }
+
+ public Method getCommand(int id)
+ {
+ lock (_commands)
+ {
+ return _commands[id];
+ }
+ }
+
+ public bool complete(int lower, int upper)
+ {
+ //avoid autoboxing
+ if (log.isDebugEnabled())
+ {
+ log.debug("{0} complete({1}, {2})", this, lower, upper);
+ }
+ lock (_commands)
+ {
+ int old = _maxComplete;
+ for (int id = Serial.max(_maxComplete, lower); Serial.le(id, upper); id++)
+ {
+ _commands.Remove(id);
+ }
+ if (Serial.le(lower, _maxComplete + 1))
+ {
+ _maxComplete = Serial.max(_maxComplete, upper);
+ }
+ log.debug("{0} commands remaining: {1}", this, _commands);
+ Monitor.PulseAll(_commands);
+ return Serial.gt(_maxComplete, old);
+ }
+ }
+
+ protected override void invoke(Method m)
+ {
+ if (Closed)
+ {
+ List<ExecutionException> exc = getExceptions();
+ if (exc.Count > 0)
+ {
+ throw new SessionException(exc);
+ }
+ else if (_close != null)
+ {
+ throw new ConnectionException(_close);
+ }
+ else
+ {
+ throw new SessionClosedException();
+ }
+ }
+
+ if (m.EncodedTrack == Frame.L4)
+ {
+ lock (_commands)
+ {
+ int next = _commandsOut++;
+ m.Id = next;
+ if (next == 0)
+ {
+ sessionCommandPoint(0, 0);
+ }
+ if (ENABLE_REPLAY)
+ {
+ _commands.Add(next, m);
+ }
+ if (_autoSync)
+ {
+ m.Sync = true;
+ }
+ _needSync = ! m.Sync;
+ _channel.method(m);
+ if (_autoSync)
+ {
+ sync();
+ }
+
+ // flush every 64K commands to avoid ambiguity on
+ // wraparound
+ if ((next%65536) == 0)
+ {
+ sessionFlush(Option.COMPLETED);
+ }
+ }
+ }
+ else
+ {
+ _channel.method(m);
+ }
+ }
+
+ public void sync()
+ {
+ sync(_timeout);
+ }
+
+ public void sync(long timeout)
+ {
+ log.debug("{0} sync()", this);
+ lock (_commands)
+ {
+ int point = _commandsOut - 1;
+
+ if (_needSync && Serial.lt(_maxComplete, point))
+ {
+ executionSync(Option.SYNC);
+ }
+
+ long start = DateTime.Now.Millisecond;
+ long elapsed = 0;
+
+ while (! Closed && elapsed < timeout && Serial.lt(_maxComplete, point))
+ {
+ log.debug("{0} waiting for[{1}]: {2}, {3}", this, point,
+ _maxComplete, _commands);
+ Monitor.Wait(_commands, (int) (timeout - elapsed));
+ elapsed = DateTime.Now.Millisecond - start;
+ }
+
+ if (Serial.lt(_maxComplete, point))
+ {
+ if (Closed)
+ {
+ throw new SessionException(getExceptions());
+ }
+ else
+ {
+ throw new Exception
+ (String.Format
+ ("timed out waiting for sync: complete = {0}, point = {1}", _maxComplete, point));
+ }
+ }
+ }
+ }
+
+
+ public void result(int command, Struct result)
+ {
+ Future<Struct> future;
+ lock (_results)
+ {
+ if (_results.ContainsKey(command))
+ {
+ future = _results[command];
+ _results.Remove(command);
+ }
+ else
+ {
+ throw new Exception(String.Format("Cannot ger result {0} for {1}", command, result));
+ }
+ }
+ future.Result = result;
+ }
+
+ public void addException(ExecutionException exc)
+ {
+ lock (_exceptions)
+ {
+ _exceptions.Add(exc);
+ }
+ }
+
+ private ConnectionClose _close = null;
+
+ public void closeCode(ConnectionClose close)
+ {
+ _close = close;
+ }
+
+ public List<ExecutionException> getExceptions()
+ {
+ lock (_exceptions)
+ {
+ return new List<ExecutionException>(_exceptions);
+ }
+ }
+
+ public override Future<T> invoke<T>(Method m, Future<T> future)
+ {
+ lock (_commands)
+ {
+ future.Session = this;
+ int command = _commandsOut;
+ lock (_results)
+ {
+ _results.Add(command, (Future<Struct>) future);
+ }
+ invoke(m);
+ }
+ return future;
+ }
+
+
+ public void messageTransfer(String destination,
+ MessageAcceptMode acceptMode,
+ MessageAcquireMode acquireMode,
+ Header header,
+ byte[] body,
+ params Option[] options)
+ {
+ MemoryStream mbody = new MemoryStream();
+ mbody.Write(body,0, body.Length);
+ messageTransfer(destination, acceptMode, acquireMode, header,
+ mbody, options);
+ }
+
+ public void messageTransfer(String destination,
+ MessageAcceptMode acceptMode,
+ MessageAcquireMode acquireMode,
+ Header header,
+ String body,
+ params Option[] options)
+ {
+ messageTransfer(destination, acceptMode, acquireMode, header,
+ new MemoryStream(Convert.ToByte(body)), options);
+ }
+
+ public void close()
+ {
+ sessionRequestTimeout(0);
+ sessionDetach(_name);
+ lock (_commands)
+ {
+ long start = DateTime.Now.Millisecond;
+ long elapsed = 0;
+
+ while (! Closed && elapsed < _timeout)
+ {
+ Monitor.Wait(_commands, (int) (_timeout - elapsed));
+ elapsed = DateTime.Now.Millisecond - start;
+ }
+ }
+ }
+
+ public void exception(Exception t)
+ {
+ log.error(t, "Caught exception");
+ }
+
+ public void closed()
+ {
+ Closed = true;
+ lock (_commands)
+ {
+ Monitor.PulseAll(_commands);
+ }
+ lock (_results)
+ {
+ foreach (Future<Struct> result in _results.Values)
+ {
+ lock (result)
+ {
+ Monitor.PulseAll(result);
+ }
+ }
+ }
+ _channel.Session = null;
+ _channel = null;
+ }
+
+ public String toString()
+ {
+ return String.Format("session:{0}", _name);
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/SessionDelegate.cs b/qpid/dotnet/client-010/client/transport/SessionDelegate.cs new file mode 100644 index 0000000000..daae528eb0 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/SessionDelegate.cs @@ -0,0 +1,128 @@ +/*
+*
+* 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;
+
+namespace org.apache.qpid.transport
+{
+ /// <summary>
+ /// SessionDelegate
+ ///
+ /// </summary>
+ public abstract class SessionDelegate : MethodDelegate<Session>, ProtocolDelegate<Session>
+ {
+ public void Init(Session ssn, ProtocolHeader hdr)
+ {
+ }
+
+ public void Control(Session ssn, Method method)
+ {
+ method.dispatch(ssn, this);
+ }
+
+ public void Command(Session ssn, Method method)
+ {
+ ssn.identify(method);
+ method.dispatch(ssn, this);
+ if (!method.hasPayload())
+ {
+ ssn.processed(method);
+ }
+ }
+
+ public void Error(Session ssn, ProtocolError error)
+ {
+ }
+
+ public override void executionResult(Session ssn, ExecutionResult result)
+ {
+ ssn.result(result.getCommandId(), result.getValue());
+ }
+
+ public override void executionException(Session ssn, ExecutionException exc)
+ {
+ ssn.addException(exc);
+ }
+
+ public override void sessionCompleted(Session ssn, SessionCompleted cmp)
+ {
+ RangeSet ranges = cmp.getCommands();
+ RangeSet known = null;
+ if (cmp.getTimelyReply())
+ {
+ known = new RangeSet();
+ }
+
+ if (ranges != null)
+ {
+ foreach (Range range in ranges)
+ {
+ bool advanced = ssn.complete(range.Lower, range.Upper);
+ if (advanced && known != null)
+ {
+ known.add(range);
+ }
+ }
+ }
+
+ if (known != null)
+ {
+ ssn.sessionKnownCompleted(known);
+ }
+ }
+
+ public override void sessionKnownCompleted(Session ssn, SessionKnownCompleted kcmp)
+ {
+ RangeSet kc = kcmp.getCommands();
+ if (kc != null)
+ {
+ ssn.knownComplete(kc);
+ }
+ }
+
+ public override void sessionFlush(Session ssn, SessionFlush flush)
+ {
+ if (flush.getCompleted())
+ {
+ ssn.flushProcessed();
+ }
+ if (flush.getConfirmed())
+ {
+ ssn.flushProcessed();
+ }
+ if (flush.getExpected())
+ {
+ // to be done
+ //throw new Exception("not implemented");
+ }
+ }
+
+ public override void sessionCommandPoint(Session ssn, SessionCommandPoint scp)
+ {
+ ssn.CommandsIn = scp.getCommandId();
+ }
+
+ public override void executionSync(Session ssn, ExecutionSync sync)
+ {
+ ssn.syncPoint();
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/Struct.cs b/qpid/dotnet/client-010/client/transport/Struct.cs new file mode 100644 index 0000000000..718dcbc978 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/Struct.cs @@ -0,0 +1,122 @@ +/*
+*
+* 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.Generic;
+using System.Text;
+using Decoder = org.apache.qpid.transport.codec.Decoder;
+using Encodable = org.apache.qpid.transport.codec.Encodable;
+using Encoder = org.apache.qpid.transport.codec.Encoder;
+namespace org.apache.qpid.transport
+{
+ /// <summary>
+ /// Struct
+ /// </summary>
+
+ public abstract class Struct : Encodable
+ {
+ public static Struct create(int type)
+ {
+ return StructFactory.create(type);
+ }
+
+ bool dirty = true;
+
+ public bool Dirty
+ {
+ get { return dirty; }
+ set { dirty = value; }
+ }
+
+ public abstract int getStructType();
+
+ public abstract int getSizeWidth();
+
+ public abstract int getPackWidth();
+
+ public int getEncodedType()
+ {
+ int type = getStructType();
+ if (type < 0)
+ {
+ throw new Exception();
+ }
+ return type;
+ }
+
+ private bool isBit<C, T>(Field<C, T> f)
+ {
+ return Equals(f.Type, typeof(Boolean));
+ }
+
+ private bool packed()
+ {
+ return getPackWidth() > 0;
+ }
+
+ private bool encoded<C, T>(Field<C, T> f)
+ {
+ return !packed() || !isBit(f) && f.has(this);
+ }
+
+ private int getFlagWidth()
+ {
+ return (Fields.Count + 7) / 8;
+ }
+
+ private int getFlagCount()
+ {
+ return 8 * getPackWidth();
+ }
+
+ public abstract void read(Decoder dec);
+
+ public abstract void write(Encoder enc);
+
+ public abstract Dictionary<String, Object> Fields
+ {
+ get;
+ }
+
+ public String toString()
+ {
+ StringBuilder str = new StringBuilder();
+ str.Append(GetType());
+ str.Append("(");
+ bool first = true;
+ foreach (KeyValuePair<String, Object> me in Fields)
+ {
+ if (first)
+ {
+ first = false;
+ }
+ else
+ {
+ str.Append(", ");
+ }
+ str.Append(me.Key);
+ str.Append("=");
+ str.Append(me.Value);
+ }
+ str.Append(")");
+ return str.ToString();
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/codec/AbstractDecoder.cs b/qpid/dotnet/client-010/client/transport/codec/AbstractDecoder.cs new file mode 100644 index 0000000000..eeb08ce2ae --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/codec/AbstractDecoder.cs @@ -0,0 +1,392 @@ +/*
+*
+* 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.Generic;
+using System.Diagnostics;
+using System.Text;
+using org.apache.qpid.transport.util;
+
+namespace org.apache.qpid.transport.codec
+{
+ /// <summary>
+ /// AbstractDecoder
+ /// </summary>
+ public abstract class AbstractDecoder : Decoder
+ {
+ private readonly Dictionary<Binary, String> str8cache = new Dictionary<Binary, String>();
+
+ protected abstract byte doGet();
+
+ protected abstract void doGet(byte[] bytes);
+ public abstract bool hasRemaining();
+
+ protected byte get()
+ {
+ return doGet();
+ }
+
+ protected void get(byte[] bytes)
+ {
+ doGet(bytes);
+ }
+
+ protected Binary get(int size)
+ {
+ byte[] bytes = new byte[size];
+ get(bytes);
+ return new Binary(bytes);
+ }
+
+ protected short uget()
+ {
+ return (short) (0xFF & get());
+ }
+
+ public virtual short readUint8()
+ {
+ return uget();
+ }
+
+ public abstract int readUint16();
+
+
+ public abstract long readUint32();
+
+
+ public int readSequenceNo()
+ {
+ return (int) readUint32();
+ }
+
+ public virtual long readUint64()
+ {
+ long l = 0;
+ for (int i = 0; i < 8; i++)
+ {
+ l |= ((long) (0xFF & get())) << (56 - i*8);
+ }
+ return l;
+ }
+
+ public long readDatetime()
+ {
+ return readUint64();
+ }
+
+ private static String decode(byte[] bytes, int offset, int length, Encoding encoding)
+ {
+ return encoding.GetString(bytes, offset, length);
+ }
+
+ private static String decode(byte[] bytes, Encoding encoding)
+ {
+ return decode(bytes, 0, bytes.Length, encoding);
+ }
+
+ public String readStr8()
+ {
+ short size = readUint8();
+ Binary bin = get(size);
+ String str;
+ if (! str8cache.TryGetValue(bin, out str))
+ {
+ str = decode(bin.array(), bin.offset(), bin.size(), Encoding.UTF8);
+ str8cache.Add(bin, str);
+ }
+ return str;
+ }
+
+ public String readStr16()
+ {
+ int size = readUint16();
+ byte[] bytes = new byte[size];
+ get(bytes);
+ return decode(bytes, Encoding.UTF8);
+ }
+
+ public byte[] readVbin8()
+ {
+ int size = readUint8();
+ byte[] bytes = new byte[size];
+ get(bytes);
+ return bytes;
+ }
+
+ public byte[] readVbin16()
+ {
+ int size = readUint16();
+ byte[] bytes = new byte[size];
+ get(bytes);
+ return bytes;
+ }
+
+ public byte[] readVbin32()
+ {
+ int size = (int) readUint32();
+ byte[] bytes = new byte[size];
+ get(bytes);
+ return bytes;
+ }
+
+ public RangeSet readSequenceSet()
+ {
+ int count = readUint16()/8;
+ if (count == 0)
+ {
+ return null;
+ }
+ RangeSet ranges = new RangeSet();
+ for (int i = 0; i < count; i++)
+ {
+ ranges.add(readSequenceNo(), readSequenceNo());
+ }
+ return ranges;
+ }
+
+ public RangeSet readByteRanges()
+ {
+ throw new Exception("not implemented");
+ }
+
+ public UUID readUuid()
+ {
+ long msb = readUint64();
+ long lsb = readUint64();
+ return new UUID(msb, lsb);
+ }
+
+ public String readContent()
+ {
+ throw new Exception("Deprecated");
+ }
+
+ public Struct readStruct(int type)
+ {
+ Struct st = Struct.create(type);
+ int width = st.getSizeWidth();
+ if (width > 0)
+ {
+ long size = readSize(width);
+ if (size == 0)
+ {
+ return null;
+ }
+ }
+ if (type > 0)
+ {
+ int code = readUint16();
+ Debug.Assert(code == type);
+ }
+ st.read(this);
+ return st;
+ }
+
+ public Struct readStruct32()
+ {
+ long size = readUint32();
+ if (size == 0)
+ {
+ return null;
+ }
+ int type = readUint16();
+ Struct result = Struct.create(type);
+ result.read(this);
+ return result;
+ }
+
+ public Dictionary<String, Object> readMap()
+ {
+ long size = readUint32();
+
+ if (size == 0)
+ {
+ return null;
+ }
+
+ long count = readUint32();
+
+ Dictionary<String, Object> result = new Dictionary<String, Object>();
+ for (int i = 0; i < count; i++)
+ {
+ String key = readStr8();
+ byte code = get();
+ QpidType t = getType(code);
+ Object value = read(t);
+ result.Add(key, value);
+ }
+
+ return result;
+ }
+
+ public List<Object> readList()
+ {
+ long size = readUint32();
+
+ if (size == 0)
+ {
+ return null;
+ }
+
+ long count = readUint32();
+
+ List<Object> result = new List<Object>();
+ for (int i = 0; i < count; i++)
+ {
+ byte code = get();
+ QpidType t = getType(code);
+ Object value = read(t);
+ result.Add(value);
+ }
+ return result;
+ }
+
+ public List<Object> readArray()
+ {
+ long size = readUint32();
+
+ if (size == 0)
+ {
+ return null;
+ }
+
+ byte code = get();
+ QpidType t = getType(code);
+ long count = readUint32();
+
+ List<Object> result = new List<Object>();
+ for (int i = 0; i < count; i++)
+ {
+ Object value = read(t);
+ result.Add(value);
+ }
+ return result;
+ }
+
+ private QpidType getType(byte code)
+ {
+ return QpidType.get(code);
+ }
+
+ private long readSize(QpidType t)
+ {
+ return t.Fixed ? t.Width : readSize(t.Width);
+ }
+
+ private long readSize(int width)
+ {
+ switch (width)
+ {
+ case 1:
+ return readUint8();
+ case 2:
+ return readUint16();
+ case 4:
+ return readUint32();
+ default:
+ throw new Exception("illegal width: " + width);
+ }
+ }
+
+ private byte[] readBytes(QpidType t)
+ {
+ long size = readSize(t);
+ byte[] result = new byte[(int) size];
+ get(result);
+ return result;
+ }
+
+ private Object read(QpidType t)
+ {
+ switch (t.Code)
+ {
+ case Code.BIN8:
+ case Code.UINT8:
+ return readUint8();
+ case Code.INT8:
+ return get();
+ case Code.CHAR:
+ return (char) get();
+ case Code.BOOLEAN:
+ return get() > 0;
+
+ case Code.BIN16:
+ case Code.UINT16:
+ return readUint16();
+ case Code.INT16:
+ return (short) readUint16();
+
+ case Code.BIN32:
+ case Code.UINT32:
+ return readUint32();
+
+ case Code.CHAR_UTF32:
+ case Code.INT32:
+ return (int) readUint32();
+
+ case Code.FLOAT:
+ return (float) readUint32();
+
+ case Code.BIN64:
+ case Code.UINT64:
+ case Code.INT64:
+ case Code.DATETIME:
+ return readUint64();
+
+ case Code.DOUBLE:
+ return (double) readUint64();
+ case Code.UUID:
+ return readUuid();
+ case Code.STR8:
+ return readStr8();
+ case Code.STR16:
+ return readStr16();
+ case Code.STR8_LATIN:
+ case Code.STR8_UTF16:
+ case Code.STR16_LATIN:
+ case Code.STR16_UTF16:
+ // XXX: need to do character conversion
+ return Encoding.UTF8.GetString(readBytes(t));
+
+ case Code.MAP:
+ return readMap();
+ case Code.LIST:
+ return readList();
+ case Code.ARRAY:
+ return readArray();
+ case Code.STRUCT32:
+ return readStruct32();
+
+ case Code.BIN40:
+ case Code.DEC32:
+ case Code.BIN72:
+ case Code.DEC64:
+ // XXX: what types are we supposed to use here?
+ return readBytes(t);
+
+ case Code.VOID:
+ return null;
+
+ default:
+ return readBytes(t);
+ }
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/codec/AbstractEncoder.cs b/qpid/dotnet/client-010/client/transport/codec/AbstractEncoder.cs new file mode 100644 index 0000000000..a34bd43331 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/codec/AbstractEncoder.cs @@ -0,0 +1,582 @@ +/*
+*
+* 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.Generic;
+using System.Diagnostics;
+using System.IO;
+using System.Text;
+using org.apache.qpid.transport.codec;
+using org.apache.qpid.transport.util;
+
+namespace org.apache.qpid.transport.codec
+{
+ /// <summary>
+ /// AbstractEncoder
+ /// </summary>
+ public abstract class AbstractEncoder : Encoder
+ {
+ private static readonly Dictionary<Type, Code> ENCODINGS = new Dictionary<Type, Code>();
+ private readonly Dictionary<String, byte[]> str8cache = new Dictionary<String, byte[]>();
+
+ static AbstractEncoder()
+ {
+ ENCODINGS.Add(typeof (Boolean), Code.BOOLEAN);
+ ENCODINGS.Add(typeof (String), Code.STR16);
+ ENCODINGS.Add(typeof (long), Code.INT64);
+ ENCODINGS.Add(typeof (int), Code.INT32);
+ ENCODINGS.Add(typeof (short), Code.INT16);
+ ENCODINGS.Add(typeof (Byte), Code.INT8);
+ ENCODINGS.Add(typeof (Dictionary<String, Object>), Code.MAP);
+ ENCODINGS.Add(typeof (List<Object>), Code.LIST);
+ ENCODINGS.Add(typeof (float), Code.FLOAT);
+ ENCODINGS.Add(typeof (Double), Code.DOUBLE);
+ ENCODINGS.Add(typeof (char), Code.CHAR);
+ ENCODINGS.Add(typeof (byte[]), Code.VBIN32);
+ }
+
+ protected abstract void doPut(byte b);
+
+ protected abstract void doPut(MemoryStream src);
+
+
+ protected void put(byte b)
+ {
+ doPut(b);
+ }
+
+ protected void put(MemoryStream src)
+ {
+ doPut(src);
+ }
+
+ protected virtual void put(byte[] bytes)
+ {
+ put(new MemoryStream(bytes));
+ }
+
+ protected abstract int beginSize8();
+ protected abstract void endSize8(int pos);
+
+ protected abstract int beginSize16();
+ protected abstract void endSize16(int pos);
+
+ protected abstract int beginSize32();
+ protected abstract void endSize32(int pos);
+
+ public virtual void writeUint8(short b)
+ {
+ Debug.Assert(b < 0x100);
+ put((byte) b);
+ }
+
+ public virtual void writeUint16(int s)
+ {
+ Debug.Assert(s < 0x10000);
+ put((byte) Functions.lsb(s >> 8));
+ put((byte) Functions.lsb(s));
+ }
+
+ public virtual void writeUint32(long i)
+ {
+ Debug.Assert(i < 0x100000000L);
+ put((byte) Functions.lsb(i >> 24));
+ put((byte) Functions.lsb(i >> 16));
+ put((byte) Functions.lsb(i >> 8));
+ put((byte) Functions.lsb(i));
+ }
+
+ public void writeSequenceNo(int i)
+ {
+ writeUint32(i);
+ }
+
+ public virtual void writeUint64(long l)
+ {
+ for (int i = 0; i < 8; i++)
+ {
+ put((byte) Functions.lsb(l >> (56 - i*8)));
+ }
+ }
+
+
+ public void writeDatetime(long l)
+ {
+ writeUint64(l);
+ }
+
+ private static byte[] encode(String s, Encoding encoding)
+ {
+ return encoding.GetBytes(s);
+ }
+
+ public void writeStr8(String s)
+ {
+ if (s == null)
+ {
+ s = "";
+ }
+
+ byte[] bytes;
+ if (! str8cache.ContainsKey(s))
+ {
+ bytes = encode(s, Encoding.UTF8);
+ str8cache.Add(s, bytes);
+ }
+ else
+ {
+ bytes = str8cache[s];
+ }
+ writeUint8((short) bytes.Length);
+ put(bytes);
+ }
+
+ public void writeStr16(String s)
+ {
+ if (s == null)
+ {
+ s = "";
+ }
+
+ byte[] bytes = encode(s, Encoding.UTF8);
+ writeUint16(bytes.Length);
+ put(bytes);
+ }
+
+ public void writeVbin8(byte[] bytes)
+ {
+ if (bytes == null)
+ {
+ bytes = new byte[0];
+ }
+ if (bytes.Length > 255)
+ {
+ throw new Exception("array too long: " + bytes.Length);
+ }
+ writeUint8((short) bytes.Length);
+ put(bytes);
+ }
+
+ public void writeVbin16(byte[] bytes)
+ {
+ if (bytes == null)
+ {
+ bytes = new byte[0];
+ }
+ writeUint16(bytes.Length);
+ put(bytes);
+ }
+
+ public void writeVbin32(byte[] bytes)
+ {
+ if (bytes == null)
+ {
+ bytes = new byte[0];
+ }
+ writeUint32(bytes.Length);
+ put(bytes);
+ }
+
+ public void writeSequenceSet(RangeSet ranges)
+ {
+ if (ranges == null)
+ {
+ writeUint16(0);
+ }
+ else
+ {
+ writeUint16(ranges.size()*8);
+ foreach (Range range in ranges)
+ {
+ writeSequenceNo(range.Lower);
+ writeSequenceNo(range.Upper);
+ }
+ }
+ }
+
+ public void writeByteRanges(RangeSet ranges)
+ {
+ throw new Exception("not implemented");
+ }
+
+ public void writeUuid(UUID uuid)
+ {
+ long msb = 0;
+ long lsb = 0;
+ if (uuid != null)
+ {
+ msb = uuid.MostSignificantBits;
+ lsb = uuid.LeastSignificantBits;
+ }
+ writeUint64(msb);
+ writeUint64(lsb);
+ }
+
+ public void writeStruct(int type, Struct s)
+ {
+ if (s == null)
+ {
+ s = Struct.create(type);
+ }
+
+ int width = s.getSizeWidth();
+ int pos = -1;
+ if (width > 0)
+ {
+ pos = beginSize(width);
+ }
+
+ if (type > 0)
+ {
+ writeUint16(type);
+ }
+
+ s.write(this);
+
+ if (width > 0)
+ {
+ endSize(width, pos);
+ }
+ }
+
+ public void writeStruct32(Struct s)
+ {
+ if (s == null)
+ {
+ writeUint32(0);
+ }
+ else
+ {
+ int pos = beginSize32();
+ writeUint16(s.getEncodedType());
+ s.write(this);
+ endSize32(pos);
+ }
+ }
+
+ private Code encoding(Object value)
+ {
+ if (value == null)
+ {
+ return Code.VOID;
+ }
+
+ Type klass = value.GetType();
+ Code type = resolve(klass);
+
+ if (type == null)
+ {
+ throw new Exception
+ ("unable to resolve type: " + klass + ", " + value);
+ }
+ else
+ {
+ return type;
+ }
+ }
+
+ private static Code resolve(Type klass)
+ {
+ Code type = ENCODINGS[klass];
+ if (type != null)
+ {
+ return type;
+ }
+ Type sup = klass.BaseType;
+ if (sup != null)
+ {
+ type = resolve(sup);
+
+ if (type != null)
+ {
+ return type;
+ }
+ }
+ foreach (Type iface in klass.GetInterfaces())
+ {
+ type = resolve(iface);
+ if (type != null)
+ {
+ return type;
+ }
+ }
+ return Code.VOID;
+ }
+
+ public void writeMap(Dictionary<String, Object> map)
+ {
+ int pos = beginSize32();
+ if (map != null)
+ {
+ writeUint32(map.Count);
+ writeMapEntries(map);
+ }
+ endSize32(pos);
+ }
+
+ protected void writeMapEntries(Dictionary<String, Object> map)
+ {
+ foreach (KeyValuePair<String, Object> entry in map)
+ {
+ String key = entry.Key;
+ Object value = entry.Value;
+ Code type = encoding(value);
+ writeStr8(key);
+ put((byte) type);
+ write(type, value);
+ }
+ }
+
+ public void writeList(List<Object> list)
+ {
+ int pos = beginSize32();
+ if (list != null)
+ {
+ writeUint32(list.Count);
+ writeListEntries(list);
+ }
+ endSize32(pos);
+ }
+
+ protected void writeListEntries(List<Object> list)
+ {
+ foreach (Object value in list)
+ {
+ Code type = encoding(value);
+ put((byte) type);
+ write(type, value);
+ }
+ }
+
+ public void writeArray(List<Object> array)
+ {
+ int pos = beginSize32();
+ if (array != null)
+ {
+ writeArrayEntries(array);
+ }
+ endSize32(pos);
+ }
+
+ protected void writeArrayEntries(List<Object> array)
+ {
+ Code type;
+
+ if (array.Count == 0)
+ {
+ return;
+ }
+ else
+ {
+ type = encoding(array[0]);
+ }
+ put((byte) type);
+ writeUint32(array.Count);
+
+ foreach (Object value in array)
+ {
+ write(type, value);
+ }
+ }
+
+ private void writeSize(QpidType t, int size)
+ {
+ if (t.Fixed)
+ {
+ if (size != t.width)
+ {
+ throw new Exception("size does not match fixed width " + t.width + ": " + size);
+ }
+ }
+ else
+ {
+ writeSize(t.width, size);
+ }
+ }
+
+ private void writeSize(int width, int size)
+ {
+ // XXX: should check lengths
+ switch (width)
+ {
+ case 1:
+ writeUint8((short) size);
+ break;
+ case 2:
+ writeUint16(size);
+ break;
+ case 4:
+ writeUint32(size);
+ break;
+ default:
+ throw new Exception("illegal width: " + width);
+ }
+ }
+
+ private int beginSize(int width)
+ {
+ switch (width)
+ {
+ case 1:
+ return beginSize8();
+ case 2:
+ return beginSize16();
+ case 4:
+ return beginSize32();
+ default:
+ throw new Exception("illegal width: " + width);
+ }
+ }
+
+ private void endSize(int width, int pos)
+ {
+ switch (width)
+ {
+ case 1:
+ endSize8(pos);
+ break;
+ case 2:
+ endSize16(pos);
+ break;
+ case 4:
+ endSize32(pos);
+ break;
+ default:
+ throw new Exception("illegal width: " + width);
+ }
+ }
+
+ private void writeBytes(QpidType t, byte[] bytes)
+ {
+ writeSize(t, bytes.Length);
+ put(bytes);
+ }
+
+ private void write(Code t, Object value)
+ {
+ switch (t)
+ {
+ case Code.BIN8:
+ case Code.UINT8:
+ writeUint8((short) value);
+ break;
+ case Code.INT8:
+ put((Byte) value);
+ break;
+ case Code.CHAR:
+ put((byte) value);
+ break;
+ case Code.BOOLEAN:
+ if ((bool) value)
+ {
+ put(1);
+ }
+ else
+ {
+ put(0);
+ }
+
+ break;
+
+ case Code.BIN16:
+ case Code.UINT16:
+ writeUint16((int) value);
+ break;
+
+ case Code.INT16:
+ writeUint16((short) value);
+ break;
+
+ case Code.BIN32:
+ case Code.UINT32:
+ writeUint32((long) value);
+ break;
+
+ case Code.CHAR_UTF32:
+ case Code.INT32:
+ writeUint32((int) value);
+ break;
+
+ case Code.FLOAT:
+ writeUint32((long) value);
+ break;
+
+ case Code.BIN64:
+ case Code.UINT64:
+ case Code.INT64:
+ case Code.DATETIME:
+ writeUint64((long) value);
+ break;
+
+ case Code.DOUBLE:
+ writeUint64((long) value);
+ break;
+
+ case Code.UUID:
+ writeUuid((UUID) value);
+ break;
+
+ case Code.STR8:
+ writeStr8((string) value);
+ break;
+
+ case Code.STR16:
+ writeStr16((string) value);
+ break;
+
+ case Code.STR8_LATIN:
+ case Code.STR8_UTF16:
+ case Code.STR16_LATIN:
+ case Code.STR16_UTF16:
+ // XXX: need to do character conversion
+ writeBytes(QpidType.get((byte) t), encode((string) value, Encoding.Unicode));
+ break;
+
+ case Code.MAP:
+ writeMap((Dictionary<String, Object>) value);
+ break;
+ case Code.LIST:
+ writeList((List<Object>) value);
+ break;
+ case Code.ARRAY:
+ writeList((List<Object>) value);
+ break;
+ case Code.STRUCT32:
+ writeStruct32((Struct) value);
+ break;
+
+ case Code.BIN40:
+ case Code.DEC32:
+ case Code.BIN72:
+ case Code.DEC64:
+ // XXX: what types are we supposed to use here?
+ writeBytes(QpidType.get((byte) t), (byte[]) value);
+ break;
+
+ case Code.VOID:
+ break;
+
+ default:
+ writeBytes(QpidType.get((byte) t), (byte[]) value);
+ break;
+ }
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/codec/Decoder.cs b/qpid/dotnet/client-010/client/transport/codec/Decoder.cs new file mode 100644 index 0000000000..207bd60df1 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/codec/Decoder.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;
+using System.Collections.Generic;
+using org.apache.qpid.transport.util;
+
+namespace org.apache.qpid.transport.codec
+{
+ /// <summary>
+ /// Decoder
+ /// </summary>
+
+ public interface Decoder
+ {
+
+ bool hasRemaining();
+
+ short readUint8();
+ int readUint16();
+ long readUint32();
+ long readUint64();
+
+ long readDatetime();
+
+ UUID readUuid();
+
+ int readSequenceNo();
+ RangeSet readSequenceSet(); // XXX
+ RangeSet readByteRanges(); // XXX
+
+ String readStr8();
+ String readStr16();
+
+ byte[] readVbin8();
+ byte[] readVbin16();
+ byte[] readVbin32();
+
+ Struct readStruct32();
+ Dictionary<String, Object> readMap();
+ List<Object> readList();
+ List<Object> readArray();
+
+ Struct readStruct(int type);
+ }
+
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/codec/Encodable.cs b/qpid/dotnet/client-010/client/transport/codec/Encodable.cs new file mode 100644 index 0000000000..12ab7c52f1 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/codec/Encodable.cs @@ -0,0 +1,37 @@ +/*
+*
+* 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 org.apache.qpid.transport.codec
+{
+
+
+ /// <summary>
+ /// Encodable
+ /// </summary>
+
+ public interface Encodable
+ {
+
+ void write(Encoder enc);
+
+ void read(Decoder dec);
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/codec/Encoder.cs b/qpid/dotnet/client-010/client/transport/codec/Encoder.cs new file mode 100644 index 0000000000..32950ee02e --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/codec/Encoder.cs @@ -0,0 +1,62 @@ +/*
+*
+* 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.Generic;
+using org.apache.qpid.transport.util;
+using RangeSet = org.apache.qpid.transport.RangeSet;
+using Struct = org.apache.qpid.transport.Struct;
+namespace org.apache.qpid.transport.codec
+{
+ /// <summary>
+ /// Encoder
+ /// </summary>
+
+ public interface Encoder
+ {
+
+ void writeUint8(short b);
+ void writeUint16(int s);
+ void writeUint32(long i);
+ void writeUint64(long l);
+
+ void writeDatetime(long l);
+ void writeUuid(UUID uuid);
+
+ void writeSequenceNo(int s);
+ void writeSequenceSet(RangeSet ranges); // XXX
+ void writeByteRanges(RangeSet ranges); // XXX
+
+ void writeStr8(string s);
+ void writeStr16(string s);
+
+ void writeVbin8(byte[] bytes);
+ void writeVbin16(byte[] bytes);
+ void writeVbin32(byte[] bytes);
+
+ void writeStruct32(Struct s);
+ void writeMap(Dictionary<String, Object> map);
+ void writeList(List<Object> list);
+ void writeArray(List<Object> array);
+
+ void writeStruct(int type, Struct s);
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/codec/MSDecoder.cs b/qpid/dotnet/client-010/client/transport/codec/MSDecoder.cs new file mode 100644 index 0000000000..a4f2a24801 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/codec/MSDecoder.cs @@ -0,0 +1,82 @@ +/*
+*
+* 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.IO;
+using System.Text;
+using org.apache.qpid.transport.util;
+
+namespace org.apache.qpid.transport.codec
+{
+
+
+ /// <summary>
+ /// MSDecoder
+ ///
+ /// </summary>
+
+
+ public sealed class MSDecoder:AbstractDecoder
+ {
+
+ private BinaryReader reader;
+
+ public void init(MemoryStream st)
+ {
+ reader = new BinaryReader(st, Encoding.BigEndianUnicode);
+ }
+
+ protected override byte doGet()
+ {
+ return reader.ReadByte();
+ }
+
+ protected override void doGet(byte[] bytes)
+ {
+ reader.Read(bytes, 0, bytes.Length);
+ }
+
+ public override bool hasRemaining()
+ {
+ return (reader.BaseStream.Position < reader.BaseStream.Length);
+ }
+
+ public override short readUint8()
+ {
+ return (short) (0xFF & reader.ReadByte());
+ }
+
+ public override int readUint16()
+ {
+ return ByteEncoder.GetBigEndian((UInt16) reader.ReadInt16());
+ }
+
+ public override long readUint32()
+ {
+ return ByteEncoder.GetBigEndian((UInt32) reader.ReadInt32());
+ }
+
+ public override long readUint64()
+ {
+ return (long) ByteEncoder.GetBigEndian((Double) reader.ReadInt64());
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/codec/MSEncoder.cs b/qpid/dotnet/client-010/client/transport/codec/MSEncoder.cs new file mode 100644 index 0000000000..bb78aa8680 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/codec/MSEncoder.cs @@ -0,0 +1,141 @@ +/*
+*
+* 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.Diagnostics;
+using System.IO;
+using org.apache.qpid.transport.util;
+
+namespace org.apache.qpid.transport.codec
+{
+ /// <summary>
+ /// MSEncoder
+ /// </summary>
+ public sealed class MSEncoder : AbstractEncoder
+ {
+ private readonly MemoryStream _out;
+ private readonly BinaryWriter _writer;
+
+ public MSEncoder(int capacity)
+ {
+ _out = new MemoryStream(capacity);
+ _writer = new BinaryWriter(_out);
+ }
+
+ public void init()
+ {
+ _out.Seek(0, SeekOrigin.Begin);
+ }
+
+ public MemoryStream segment()
+ {
+ int length = (int) _out.Position;
+ MemoryStream result = new MemoryStream(_out.ToArray(), 0, length);
+ result.Seek(length, SeekOrigin.Begin);
+ _out.Seek(0, SeekOrigin.Begin);
+ return result;
+ }
+
+
+ protected override void doPut(byte b)
+ {
+ _writer.Write(b);
+ }
+
+ protected override void doPut(MemoryStream src)
+ {
+ _writer.Write(src.ToArray());
+ }
+
+ protected override void put(byte[] bytes)
+ {
+ _writer.Write(bytes);
+ }
+
+ public override void writeUint8(short b)
+ {
+ Debug.Assert(b < 0x100);
+ _writer.Write((byte) b);
+ }
+
+ public override void writeUint16(int s)
+ {
+ Debug.Assert(s < 0x10000);
+ _writer.Write(ByteEncoder.GetBigEndian((UInt16) s));
+ }
+
+ public override void writeUint32(long i)
+ {
+ Debug.Assert(i < 0x100000000L);
+ _writer.Write(ByteEncoder.GetBigEndian((UInt32) i));
+ }
+
+ public override void writeUint64(long l)
+ {
+ _writer.Write(ByteEncoder.GetBigEndian((Double) l));
+ }
+
+ protected override int beginSize8()
+ {
+ int pos = (int) _out.Position;
+ _writer.Write((byte) 0);
+ return pos;
+ }
+
+ protected override void endSize8(int pos)
+ {
+ int cur = (int) _out.Position;
+ _out.Seek(pos, SeekOrigin.Begin);
+ _writer.Write((byte) (cur - pos - 1));
+ _out.Seek(cur, SeekOrigin.Begin);
+ }
+
+ protected override int beginSize16()
+ {
+ int pos = (int) _out.Position;
+ _writer.Write((short) 0);
+ return pos;
+ }
+
+ protected override void endSize16(int pos)
+ {
+ int cur = (int) _out.Position;
+ _out.Seek(pos, SeekOrigin.Begin);
+ _writer.Write((short) (cur - pos - 2));
+ _out.Seek(cur, SeekOrigin.Begin);
+ }
+
+ protected override int beginSize32()
+ {
+ int pos = (int) _out.Position;
+ _writer.Write(0);
+ return pos;
+ }
+
+ protected override void endSize32(int pos)
+ {
+ int cur = (int) _out.Position;
+ _out.Seek(pos, SeekOrigin.Begin);
+ _writer.Write(ByteEncoder.GetBigEndian((Int32) cur - pos - 4));
+ _out.Seek(cur, SeekOrigin.Begin);
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/exception/ConnectionException.cs b/qpid/dotnet/client-010/client/transport/exception/ConnectionException.cs new file mode 100644 index 0000000000..d380024c39 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/exception/ConnectionException.cs @@ -0,0 +1,49 @@ +/*
+*
+* 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;
+namespace org.apache.qpid.transport
+{
+
+
+ /// <summary>
+ /// ConnectionException
+ /// </summary>
+
+ [Serializable]
+ public class ConnectionException : Exception
+ {
+ virtual public ConnectionClose Close
+ {
+ get
+ {
+ return _close;
+ }
+
+ }
+
+ private ConnectionClose _close;
+
+ public ConnectionException(ConnectionClose close):base(close.getReplyText())
+ {
+ _close = close;
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/exception/ExceptionArgs.cs b/qpid/dotnet/client-010/client/transport/exception/ExceptionArgs.cs new file mode 100644 index 0000000000..e9cc1b6cad --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/exception/ExceptionArgs.cs @@ -0,0 +1,41 @@ +/*
+*
+* 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;
+
+namespace org.apache.qpid.transport
+{
+ public class ExceptionArgs : EventArgs
+ {
+ public ExceptionArgs(Exception e)
+ {
+ m_exception = e;
+ }
+ private Exception m_exception;
+
+ public Exception Exception
+ {
+ get { return m_exception; }
+ set { m_exception = value; }
+ }
+
+ }
+}
diff --git a/qpid/dotnet/client-010/client/transport/exception/ProtocolVersionException.cs b/qpid/dotnet/client-010/client/transport/exception/ProtocolVersionException.cs new file mode 100644 index 0000000000..2182ab9ba4 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/exception/ProtocolVersionException.cs @@ -0,0 +1,59 @@ +/*
+*
+* 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;
+namespace org.apache.qpid.transport
+{
+
+
+ /// <summary> ProtocolVersionException
+ ///
+ /// </summary>
+
+ [Serializable]
+ public sealed class ProtocolVersionException:TransportException
+ {
+ public sbyte Major
+ {
+ get
+ {
+ return this.major;
+ }
+
+ }
+ public sbyte Minor
+ {
+ get
+ {
+ return this.minor;
+ }
+
+ }
+
+ private sbyte major;
+ private sbyte minor;
+
+ public ProtocolVersionException(sbyte major, sbyte minor):base(String.Format("version missmatch: %{0}-{1}", major, minor))
+ {
+ this.major = major;
+ this.minor = minor;
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/exception/SessionClosedException.cs b/qpid/dotnet/client-010/client/transport/exception/SessionClosedException.cs new file mode 100644 index 0000000000..9f3080ac96 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/exception/SessionClosedException.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 System.Collections.Generic;
+
+namespace org.apache.qpid.transport
+{
+
+
+ /// <summary>
+ /// SessionClosedException
+ /// </summary>
+
+ public class SessionClosedException : SessionException
+ {
+
+ public SessionClosedException(): base(new List<ExecutionException>())
+ {
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/exception/SessionException.cs b/qpid/dotnet/client-010/client/transport/exception/SessionException.cs new file mode 100644 index 0000000000..b586019802 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/exception/SessionException.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 System;
+using System.Collections.Generic;
+
+namespace org.apache.qpid.transport
+{
+ /// <summary>
+ /// SessionException
+ /// </summary>
+ public class SessionException : Exception
+ {
+ private readonly List<ExecutionException> _exceptions;
+
+ public SessionException(List<ExecutionException> exceptions)
+ : base(exceptions.Count == 0 ? "" : exceptions.ToString())
+
+ {
+ _exceptions = exceptions;
+ }
+
+ public List<ExecutionException> Exceptions
+ {
+ get { return _exceptions; }
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/exception/TransportException.cs b/qpid/dotnet/client-010/client/transport/exception/TransportException.cs new file mode 100644 index 0000000000..1c216d51e4 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/exception/TransportException.cs @@ -0,0 +1,46 @@ +/*
+*
+* 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;
+namespace org.apache.qpid.transport
+{
+
+
+ /// <summary>
+ /// TransportException
+ /// </summary>
+
+
+ public class TransportException : Exception
+ {
+ public TransportException(String msg) : base(msg)
+ {
+ }
+
+ public TransportException(String msg, Exception cause) : base(msg, cause)
+ {
+ }
+
+ public TransportException(Exception cause): base("Transport Exception", cause)
+ {
+ }
+
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/network/Assembler.cs b/qpid/dotnet/client-010/client/transport/network/Assembler.cs new file mode 100644 index 0000000000..89bd592f96 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/network/Assembler.cs @@ -0,0 +1,282 @@ +/*
+*
+* 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.Generic;
+using System.IO;
+using org.apache.qpid.transport.codec;
+using org.apache.qpid.transport.util;
+
+namespace org.apache.qpid.transport.network
+{
+ /// <summary>
+ /// Assembler
+ /// </summary>
+ public delegate void Processor(NetworkDelegate ndelegate);
+
+ public class Assembler : NetworkDelegate, Receiver<ReceivedPayload<ProtocolEvent>>
+ {
+ private static readonly Logger log = Logger.get(typeof (Assembler));
+ private readonly Dictionary<int, List<Frame>> segments;
+ private readonly Method[] incomplete;
+ [ThreadStatic] static MSDecoder _decoder;
+ private readonly Object m_objectLock = new object();
+
+ // the event raised when a buffer is read from the wire
+ public event EventHandler<ReceivedPayload<ProtocolEvent>> ReceivedEvent;
+ public event EventHandler<ExceptionArgs> ExceptionProcessing;
+ public event EventHandler HandlerClosed;
+
+ event EventHandler<ReceivedPayload<ProtocolEvent>> Receiver<ReceivedPayload<ProtocolEvent>>.Received
+ {
+ add
+ {
+ lock (m_objectLock)
+ {
+ ReceivedEvent += value;
+ }
+ }
+ remove
+ {
+ lock (m_objectLock)
+ {
+ ReceivedEvent -= value;
+ }
+ }
+ }
+
+ event EventHandler<ExceptionArgs> Receiver<ReceivedPayload<ProtocolEvent>>.Exception
+ {
+ add
+ {
+ lock (m_objectLock)
+ {
+ ExceptionProcessing += value;
+ }
+ }
+ remove
+ {
+ lock (m_objectLock)
+ {
+ ExceptionProcessing -= value;
+ }
+ }
+ }
+
+ event EventHandler Receiver<ReceivedPayload<ProtocolEvent>>.Closed
+ {
+ add
+ {
+ lock (m_objectLock)
+ {
+ HandlerClosed += value;
+ }
+ }
+ remove
+ {
+ lock (m_objectLock)
+ {
+ HandlerClosed -= value;
+ }
+ }
+ }
+
+ public Assembler()
+ {
+ segments = new Dictionary<int, List<Frame>>();
+ incomplete = new Method[64*1024];
+ }
+
+ // Invoked when a network event is received
+ public void On_ReceivedEvent(object sender, ReceivedPayload<NetworkEvent> payload)
+ {
+ payload.Payload.ProcessNetworkEvent(this);
+ }
+
+ #region Interface NetworkDelegate
+
+ public void Init(ProtocolHeader header)
+ {
+ Emit(0, header);
+ }
+
+ public void Error(ProtocolError error)
+ {
+ Emit(0, error);
+ }
+
+ public void Frame(Frame frame)
+ {
+ MemoryStream segment;
+ if (frame.isFirstFrame() && frame.isLastFrame())
+ {
+ segment = frame.Body;
+ assemble(frame, segment);
+ }
+ else
+ {
+ List<Frame> frames;
+ if (frame.isFirstFrame())
+ {
+ frames = new List<Frame>();
+ setSegment(frame, frames);
+ }
+ else
+ {
+ frames = getSegment(frame);
+ }
+
+ frames.Add(frame);
+
+ if (frame.isLastFrame())
+ {
+ clearSegment(frame);
+ segment = new MemoryStream();
+ BinaryWriter w = new BinaryWriter(segment);
+ foreach (Frame f in frames)
+ {
+ w.Write(f.Body.ToArray());
+ }
+ assemble(frame, segment);
+ }
+ }
+ }
+
+ #endregion
+
+ #region Private Support Functions
+
+
+ private MSDecoder getDecoder()
+ {
+ if( _decoder == null )
+ {
+ _decoder = new MSDecoder();
+ }
+ return _decoder;
+ }
+
+ private void assemble(Frame frame, MemoryStream segment)
+ {
+ MSDecoder decoder = getDecoder();
+ decoder.init(segment);
+ int channel = frame.Channel;
+ Method command;
+ switch (frame.Type)
+ {
+ case SegmentType.CONTROL:
+ int controlType = decoder.readUint16();
+ Method control = Method.create(controlType);
+ control.read(decoder);
+ Emit(channel, control);
+ break;
+ case SegmentType.COMMAND:
+ int commandType = decoder.readUint16();
+ // read in the session header, right now we don't use it
+ decoder.readUint16();
+ command = Method.create(commandType);
+ command.read(decoder);
+ if (command.hasPayload())
+ {
+ incomplete[channel] = command;
+ }
+ else
+ {
+ Emit(channel, command);
+ }
+ break;
+ case SegmentType.HEADER:
+ command = incomplete[channel];
+ List<Struct> structs = new List<Struct>();
+ //while (decoder.hasRemaining())
+ for (int i = 0; i < 2; i++ )
+ {
+ structs.Add(decoder.readStruct32());
+ }
+ command.Header = new Header(structs);
+ if (frame.isLastSegment())
+ {
+ incomplete[channel] = null;
+ Emit(channel, command);
+ }
+ break;
+ case SegmentType.BODY:
+ command = incomplete[channel];
+ byte[] b = new byte[frame.BodySize];
+ MemoryStream body = new MemoryStream();
+ segment.Read(b, 0, b.Length);
+ body.Write(b, 0, b.Length);
+ body.Seek(0, SeekOrigin.Begin);
+ command.Body = body;
+ incomplete[channel] = null;
+ Emit(channel, command);
+ break;
+ default:
+ throw new Exception("unknown frame type: " + frame.Type);
+ }
+ }
+
+ private int segmentKey(Frame frame)
+ {
+ return (frame.Track + 1)*frame.Channel;
+ }
+
+ private List<Frame> getSegment(Frame frame)
+ {
+ return segments[segmentKey(frame)];
+ }
+
+ private void setSegment(Frame frame, List<Frame> segment)
+ {
+ int key = segmentKey(frame);
+ if (segments.ContainsKey(key))
+ {
+ Error(new ProtocolError(network.Frame.L2, "segment in progress: %s",
+ frame));
+ }
+ segments.Add(segmentKey(frame), segment);
+ }
+
+ private void clearSegment(Frame frame)
+ {
+ segments.Remove(segmentKey(frame));
+ }
+
+ // Emit a protocol event
+ private void Emit(int channel, ProtocolEvent protevent)
+ {
+ protevent.Channel = channel;
+ log.debug("Assembler: protocol event:", protevent);
+ ReceivedPayload<ProtocolEvent> payload = new ReceivedPayload<ProtocolEvent>();
+ payload.Payload = protevent;
+ if (ReceivedEvent != null)
+ {
+ ReceivedEvent(this, payload);
+ }
+ else
+ {
+ log.debug("No listener for event: {0}", protevent);
+ }
+ }
+
+ #endregion
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/network/Disassembler.cs b/qpid/dotnet/client-010/client/transport/network/Disassembler.cs new file mode 100644 index 0000000000..c1e6744f2c --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/network/Disassembler.cs @@ -0,0 +1,222 @@ +/*
+*
+* 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.IO;
+using org.apache.qpid.transport.codec;
+using org.apache.qpid.transport.util;
+
+namespace org.apache.qpid.transport.network
+{
+ /// <summary>
+ /// Disassembler
+ /// </summary>
+ public sealed class Disassembler : Sender<ProtocolEvent>, ProtocolDelegate<Object>
+ {
+ private readonly Sender<MemoryStream> _sender;
+ private readonly int _maxPayload;
+ private readonly MemoryStream _header;
+ private readonly BinaryWriter _writer;
+ private readonly Object _sendlock = new Object();
+ [ThreadStatic] static MSEncoder _encoder;
+
+
+ public Disassembler(Sender<MemoryStream> sender, int maxFrame)
+ {
+ if (maxFrame <= Frame.HEADER_SIZE || maxFrame >= 64*1024)
+ {
+ throw new Exception(String.Format("maxFrame must be > {0} and < 64K: ", Frame.HEADER_SIZE) + maxFrame);
+ }
+ _sender = sender;
+ _maxPayload = maxFrame - Frame.HEADER_SIZE;
+ _header = new MemoryStream(Frame.HEADER_SIZE);
+ _writer = new BinaryWriter(_header);
+ }
+
+ #region Sender Interface
+
+ public void send(ProtocolEvent pevent)
+ {
+ pevent.ProcessProtocolEvent(null, this);
+ }
+
+ public void flush()
+ {
+ lock (_sendlock)
+ {
+ _sender.flush();
+ }
+ }
+
+ public void close()
+ {
+ lock (_sendlock)
+ {
+ _sender.close();
+ }
+ }
+
+ #endregion
+
+ #region ProtocolDelegate<Object> Interface
+
+ public void Init(Object v, ProtocolHeader header)
+ {
+ lock (_sendlock)
+ {
+ _sender.send(header.ToMemoryStream());
+ _sender.flush();
+ }
+ }
+
+ public void Control(Object v, Method method)
+ {
+ invokeMethod(method, SegmentType.CONTROL);
+ }
+
+ public void Command(Object v, Method method)
+ {
+ invokeMethod(method, SegmentType.COMMAND);
+ }
+
+ public void Error(Object v, ProtocolError error)
+ {
+ throw new Exception("Error: " + error);
+ }
+
+ #endregion
+
+ #region private
+
+ private void frame(byte flags, byte type, byte track, int channel, int size, MemoryStream buf)
+ {
+ lock (_sendlock)
+ {
+ _writer.Write(flags);
+ _writer.Write(type);
+ _writer.Write(ByteEncoder.GetBigEndian((UInt16)(size + Frame.HEADER_SIZE)));
+ _writer.Write((byte)0);
+ _writer.Write(track);
+ _writer.Write(ByteEncoder.GetBigEndian((UInt16)( channel)));
+ _writer.Write((byte)0);
+ _writer.Write((byte)0);
+ _writer.Write((byte)0);
+ _writer.Write((byte)0);
+ _sender.send(_header);
+ _header.Seek(0, SeekOrigin.Begin);
+ _sender.send(buf);
+ }
+ }
+
+ private void fragment(byte flags, SegmentType type, ProtocolEvent mevent, MemoryStream buf)
+ {
+ byte typeb = (byte) type;
+ byte track = mevent.EncodedTrack == Frame.L4 ? (byte) 1 : (byte) 0;
+
+ int remaining = (int) buf.Length;
+ bool first = true;
+ while (true)
+ {
+ int size = Math.Min(_maxPayload, remaining);
+ remaining -= size;
+
+ byte newflags = flags;
+ if (first)
+ {
+ newflags |= Frame.FIRST_FRAME;
+ first = false;
+ }
+ if (remaining == 0)
+ {
+ newflags |= Frame.LAST_FRAME;
+ }
+
+ frame(newflags, typeb, track, mevent.Channel, size, buf);
+
+ if (remaining == 0)
+ {
+ break;
+ }
+ }
+ }
+
+ private MSEncoder getEncoder()
+ {
+ if( _encoder == null)
+ {
+ _encoder = new MSEncoder(4 * 1024);
+ }
+ return _encoder;
+ }
+
+ private void invokeMethod(Method method, SegmentType type)
+ {
+ MSEncoder encoder = getEncoder();
+ encoder.init();
+ encoder.writeUint16(method.getEncodedType());
+ if (type == SegmentType.COMMAND)
+ {
+ if (method.Sync)
+ {
+ encoder.writeUint16(0x0101);
+ }
+ else
+ {
+ encoder.writeUint16(0x0100);
+ }
+ }
+ method.write(_encoder);
+ MemoryStream methodSeg = encoder.segment();
+
+ byte flags = Frame.FIRST_SEG;
+
+ bool payload = method.hasPayload();
+ if (!payload)
+ {
+ flags |= Frame.LAST_SEG;
+ }
+
+ MemoryStream headerSeg = null;
+ if (payload)
+ {
+ Header hdr = method.Header;
+ Struct[] structs = hdr.Structs;
+
+ foreach (Struct st in structs)
+ {
+ encoder.writeStruct32(st);
+ }
+ headerSeg = encoder.segment();
+ }
+
+ lock (_sendlock)
+ {
+ fragment(flags, type, method, methodSeg);
+ if (payload)
+ {
+ fragment( 0x0, SegmentType.HEADER, method, headerSeg);
+ fragment(Frame.LAST_SEG, SegmentType.BODY, method, method.Body);
+ }
+ }
+ }
+
+ #endregion
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/network/Frame.cs b/qpid/dotnet/client-010/client/transport/network/Frame.cs new file mode 100644 index 0000000000..5b7e15b1aa --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/network/Frame.cs @@ -0,0 +1,143 @@ +/*
+*
+* 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.IO;
+
+namespace org.apache.qpid.transport.network
+{
+ public sealed class Frame : NetworkEvent
+ {
+ internal static int HEADER_SIZE = 12;
+
+ // XXX: enums?
+ public const byte L1 = 0;
+ public const byte L2 = 1;
+ public const byte L3 = 2;
+ public const byte L4 = 3;
+
+ public static byte RESERVED = 0x0;
+
+ public static byte VERSION = 0x0;
+
+ public static byte FIRST_SEG = 0x8;
+ public static byte LAST_SEG = 0x4;
+ public static byte FIRST_FRAME = 0x2;
+ public static byte LAST_FRAME = 0x1;
+
+ private readonly byte flags;
+ private readonly SegmentType type;
+ private readonly byte track;
+ private readonly int channel;
+ private readonly MemoryStream body;
+ private int _bodySize;
+
+
+ public Frame(byte flags, SegmentType type, byte track, int channel, int bodySize,
+ MemoryStream body)
+ {
+ this.flags = flags;
+ this.type = type;
+ this.track = track;
+ this.channel = channel;
+ this.body = body;
+ _bodySize = bodySize;
+ }
+
+ public int BodySize
+ {
+ get { return _bodySize; }
+ }
+
+ public MemoryStream Body
+ {
+ get { return body; }
+ }
+
+ public byte Flags
+ {
+ get { return flags; }
+ }
+
+ public int Channel
+ {
+ get { return channel; }
+ }
+
+ public int Size
+ {
+ get { return (int) body.Length;}
+ }
+
+ public SegmentType Type
+ {
+ get { return type; }
+ }
+
+ public byte Track
+ {
+ get { return track; }
+ }
+
+ private bool flag(byte mask)
+ {
+ return (flags & mask) != 0;
+ }
+
+ public bool isFirstSegment()
+ {
+ return flag(FIRST_SEG);
+ }
+
+ public bool isLastSegment()
+ {
+ return flag(LAST_SEG);
+ }
+
+ public bool isFirstFrame()
+ {
+ return flag(FIRST_FRAME);
+ }
+
+ public bool isLastFrame()
+ {
+ return flag(LAST_FRAME);
+ }
+
+ #region NetworkEvent Methods
+
+ public void ProcessNetworkEvent(NetworkDelegate ndelegate)
+ {
+ ndelegate.Frame(this);
+ }
+
+ #endregion
+
+ public String toString()
+ {
+ return String.Format
+ ("[{0:d} {1:d} {2:d} {3} {4}{5}{6}{7}] ", Channel, Size, Track, Type,
+ isFirstSegment() ? 1 : 0, isLastSegment() ? 1 : 0,
+ isFirstFrame() ? 1 : 0, isLastFrame() ? 1 : 0);
+ }
+
+
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/network/InputHandler.cs b/qpid/dotnet/client-010/client/transport/network/InputHandler.cs new file mode 100644 index 0000000000..244bf5a3ff --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/network/InputHandler.cs @@ -0,0 +1,273 @@ +/*
+*
+* 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.IO;
+using System.Text;
+using org.apache.qpid.transport.util;
+
+namespace org.apache.qpid.transport.network
+{
+ /// <summary>
+ /// InputHandler
+ /// </summary>
+ public sealed class InputHandler : Receiver<ReceivedPayload<NetworkEvent>>
+ {
+ public enum State
+ {
+ PROTO_HDR,
+ FRAME_HDR,
+ FRAME_BODY,
+ ERROR
+ }
+
+ private static readonly Logger log = Logger.get(typeof(InputHandler));
+ private readonly Object m_objectLock = new object();
+
+ // the event raised when a buffer is read from the wire
+ public event EventHandler<ReceivedPayload<NetworkEvent>> ReceivedEvent;
+ public event EventHandler<ExceptionArgs> ExceptionProcessing;
+ public event EventHandler HandlerClosed;
+
+ event EventHandler<ReceivedPayload<NetworkEvent>> Receiver<ReceivedPayload<NetworkEvent>>.Received
+ {
+ add
+ {
+ lock (m_objectLock)
+ {
+ ReceivedEvent += value;
+ }
+ }
+ remove
+ {
+ lock (m_objectLock)
+ {
+ ReceivedEvent -= value;
+ }
+ }
+ }
+
+ event EventHandler<ExceptionArgs> Receiver<ReceivedPayload<NetworkEvent>>.Exception
+ {
+ add
+ {
+ lock (m_objectLock)
+ {
+ ExceptionProcessing += value;
+ }
+ }
+ remove
+ {
+ lock (m_objectLock)
+ {
+ ExceptionProcessing -= value;
+ }
+ }
+ }
+
+ event EventHandler Receiver<ReceivedPayload<NetworkEvent>>.Closed
+ {
+ add
+ {
+ lock (m_objectLock)
+ {
+ HandlerClosed += value;
+ }
+ }
+ remove
+ {
+ lock (m_objectLock)
+ {
+ HandlerClosed -= value;
+ }
+ }
+ }
+
+ private State state;
+ private MemoryStream input;
+ private int needed;
+
+ private byte flags;
+ private SegmentType type;
+ private byte track;
+ private int channel;
+
+ public InputHandler(State state)
+ {
+ this.state = state;
+ switch (state)
+ {
+ case State.PROTO_HDR:
+ needed = 8;
+ break;
+ case State.FRAME_HDR:
+ needed = Frame.HEADER_SIZE;
+ break;
+ }
+ }
+
+ // The command listening for a buffer read.
+ public void On_ReceivedBuffer(object sender, ReceivedPayload<MemoryStream> payload)
+ {
+ MemoryStream buf = payload.Payload;
+ int remaining = (int)buf.Length;
+ try
+ {
+ while (remaining > 0)
+ {
+ if (remaining >= needed)
+ {
+ if (input != null)
+ {
+ remaining += (int) input.Length;
+ byte[] tmp = new byte[remaining];
+ buf.Read(tmp, 0, remaining);
+ input.Write(tmp, 0, remaining);
+ input.Seek(0, SeekOrigin.Begin);
+ buf = input;
+ }
+ int startPos = (int)buf.Position;
+ int consumed = needed;
+ state = next(buf);
+ if ((buf.Position - startPos) < consumed)
+ {
+ buf.Seek(consumed - (buf.Position - startPos), SeekOrigin.Current);
+ }
+ remaining -= consumed;
+ input = null;
+ }
+ else
+ {
+ if (input == null)
+ {
+ input = new MemoryStream();
+ }
+ byte[] tmp = new byte[remaining];
+ buf.Read(tmp, 0, remaining);
+ input.Write(tmp, 0, remaining);
+ remaining = 0;
+ }
+ }
+ }
+ catch (Exception t)
+ {
+ Console.Write(t);
+ if (ExceptionProcessing != null)
+ {
+ ExceptionProcessing(this, new ExceptionArgs(t));
+ }
+ }
+ }
+
+ #region Private Support Functions
+
+ private State next(MemoryStream buf)
+ {
+ BinaryReader reader = new BinaryReader(buf);
+
+ switch (state)
+ {
+ case State.PROTO_HDR:
+ char a = reader.ReadChar();
+ char m = reader.ReadChar();
+ char q = reader.ReadChar();
+ char p = reader.ReadChar();
+ if (a != 'A' &&
+ m != 'M' &&
+ q != 'Q' &&
+ p != 'P')
+ {
+ Error("bad protocol header: {0}", buf.ToString());
+ return State.ERROR;
+ }
+ reader.ReadByte();
+ byte instance = reader.ReadByte();
+ byte major = reader.ReadByte();
+ byte minor = reader.ReadByte();
+ Fire_NetworkEvent(new ProtocolHeader(instance, major, minor));
+ needed = Frame.HEADER_SIZE;
+ return State.FRAME_HDR;
+ case State.FRAME_HDR:
+ reader = new BinaryReader(buf, Encoding.BigEndianUnicode);
+ flags = reader.ReadByte();
+ type = SegmentTypeGetter.get(reader.ReadByte()); // generated code
+ int size = reader.ReadUInt16();
+ size = ByteEncoder.GetBigEndian((UInt16)size);
+ size -= Frame.HEADER_SIZE;
+ if (size < 0 || size > (64 * 1024 - 12))
+ {
+ Error("bad frame size: {0:d}", size);
+ return State.ERROR;
+ }
+ reader.ReadByte();
+ byte b = reader.ReadByte();
+ if ((b & 0xF0) != 0)
+ {
+ Error("non-zero reserved bits in upper nibble of " +
+ "frame header byte 5: {0}", b);
+ return State.ERROR;
+ }
+ track = (byte)(b & 0xF);
+ channel = reader.ReadUInt16();
+ channel = ByteEncoder.GetBigEndian((UInt16)channel);
+ if (size == 0)
+ {
+ Fire_NetworkEvent(new Frame(flags, type, track, channel, 0, new MemoryStream()));
+ needed = Frame.HEADER_SIZE;
+ return State.FRAME_HDR;
+ }
+ needed = size;
+ return State.FRAME_BODY;
+ case State.FRAME_BODY:
+ Fire_NetworkEvent(new Frame(flags, type, track, channel, needed, buf));
+ needed = Frame.HEADER_SIZE;
+ return State.FRAME_HDR;
+ default:
+ if (ExceptionProcessing != null)
+ {
+ ExceptionProcessing(this, new ExceptionArgs(new Exception("Error creating frame")));
+ }
+ throw new Exception("Error creating frame");
+ }
+ }
+
+ private void Error(String fmt, params Object[] args)
+ {
+ Fire_NetworkEvent(new ProtocolError(Frame.L1, fmt, args));
+ }
+
+ private void Fire_NetworkEvent(NetworkEvent netevent)
+ {
+ log.debug("InputHandler: network event:", netevent);
+ ReceivedPayload<NetworkEvent> payload = new ReceivedPayload<NetworkEvent>();
+ payload.Payload = netevent;
+ if (ReceivedEvent != null)
+ {
+ ReceivedEvent(this, payload);
+ }
+ else
+ {
+ log.debug("Nobody listening for event: {0}");
+ }
+ }
+
+ #endregion
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/network/NetworkDelegate.cs b/qpid/dotnet/client-010/client/transport/network/NetworkDelegate.cs new file mode 100644 index 0000000000..9549f5c295 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/network/NetworkDelegate.cs @@ -0,0 +1,40 @@ +/*
+*
+* 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 ProtocolError = org.apache.qpid.transport.ProtocolError;
+using ProtocolHeader = org.apache.qpid.transport.ProtocolHeader;
+namespace org.apache.qpid.transport.network
+{
+
+
+ /// <summary>
+ /// NetworkDelegate
+ /// </summary>
+
+ public interface NetworkDelegate
+ {
+
+ void Init(ProtocolHeader header);
+
+ void Frame(Frame frame);
+
+ void Error(ProtocolError error);
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/network/NetworkEvent.cs b/qpid/dotnet/client-010/client/transport/network/NetworkEvent.cs new file mode 100644 index 0000000000..c682e77560 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/network/NetworkEvent.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.
+*
+*/
+namespace org.apache.qpid.transport.network
+{
+
+ /// <summary>
+ /// NetworkEvent
+ /// </summary>
+
+ public interface NetworkEvent
+ {
+ void ProcessNetworkEvent(NetworkDelegate networkDelegate);
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/network/io/IoReceiver.cs b/qpid/dotnet/client-010/client/transport/network/io/IoReceiver.cs new file mode 100644 index 0000000000..b4544fe072 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/network/io/IoReceiver.cs @@ -0,0 +1,187 @@ +/*
+*
+* 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.IO;
+using System.Threading;
+using TransportException = org.apache.qpid.transport.TransportException;
+using Logger = org.apache.qpid.transport.util.Logger;
+
+
+namespace org.apache.qpid.transport.network.io
+{
+ /// <summary>
+ /// IoReceiver
+ /// </summary>
+ public sealed class IoReceiver : Receiver<ReceivedPayload<MemoryStream>>
+ {
+ private static readonly Logger log = Logger.get(typeof(IoReceiver));
+ private readonly int m_bufferSize;
+ private readonly Stream m_bufStream;
+ private readonly int m_timeout;
+ private readonly Thread m_thread;
+ private bool m_closed;
+ private readonly Object m_objectLock = new object();
+
+ // the event raised when a buffer is read from the wire
+ event EventHandler<ReceivedPayload<MemoryStream>> ReceivedBuffer;
+ event EventHandler<ExceptionArgs> ExceptionReading;
+ event EventHandler ReceiverClosed;
+
+ event EventHandler<ReceivedPayload<MemoryStream>> Receiver<ReceivedPayload<MemoryStream>>.Received
+ {
+ add
+ {
+ lock (m_objectLock)
+ {
+ ReceivedBuffer += value;
+ }
+ }
+ remove
+ {
+ lock (m_objectLock)
+ {
+ ReceivedBuffer -= value;
+ }
+ }
+ }
+
+ event EventHandler<ExceptionArgs> Receiver<ReceivedPayload<MemoryStream>>.Exception
+ {
+ add
+ {
+ lock (m_objectLock)
+ {
+ ExceptionReading += value;
+ }
+ }
+ remove
+ {
+ lock (m_objectLock)
+ {
+ ExceptionReading -= value;
+ }
+ }
+ }
+
+ event EventHandler Receiver<ReceivedPayload<MemoryStream>>.Closed
+ {
+ add
+ {
+ lock (m_objectLock)
+ {
+ ReceiverClosed += value;
+ }
+ }
+ remove
+ {
+ lock (m_objectLock)
+ {
+ ReceiverClosed -= value;
+ }
+ }
+ }
+
+ public IoReceiver(Stream stream, int bufferSize, int timeout)
+ {
+ m_bufferSize = bufferSize;
+ m_bufStream = stream;
+ m_timeout = timeout;
+ m_thread = new Thread(Go);
+ m_thread.Name = String.Format("IoReceiver - {0}", stream);
+ m_thread.IsBackground = true;
+ m_thread.Start();
+ }
+
+ public void close()
+ {
+ Mutex mut = new Mutex();
+ mut.WaitOne();
+ if (!m_closed)
+ {
+ m_closed = true;
+ try
+ {
+ log.debug("Receiver closing");
+ m_bufStream.Close();
+ m_thread.Join(m_timeout);
+ if (m_thread.IsAlive)
+ {
+ throw new TransportException("join timed out");
+ }
+ }
+ catch (ThreadInterruptedException e)
+ {
+ throw new TransportException(e);
+ }
+ catch (IOException e)
+ {
+ throw new TransportException(e);
+ }
+ }
+ mut.ReleaseMutex();
+ }
+
+ void Go()
+ {
+ // create a BufferedStream on top of the NetworkStream.
+ int threshold = m_bufferSize/2;
+ byte[] buffer = new byte[m_bufferSize];
+ try
+ {
+ int read;
+ int offset = 0;
+ ReceivedPayload<MemoryStream> payload = new ReceivedPayload<MemoryStream>();
+ while ((read = m_bufStream.Read(buffer, offset, m_bufferSize - offset)) > 0)
+ {
+ MemoryStream memStream = new MemoryStream(buffer, offset, read);
+ if(ReceivedBuffer != null)
+ {
+ // call the event
+ payload.Payload = memStream;
+ ReceivedBuffer(this, payload);
+ }
+ offset += read;
+ if (offset > threshold)
+ {
+ offset = 0;
+ buffer = new byte[m_bufferSize];
+ }
+ }
+ log.debug("Receiver thread terminating");
+ }
+ catch (Exception t)
+ {
+ Console.WriteLine(t);
+ if (ExceptionReading != null)
+ {
+ ExceptionReading(this, new ExceptionArgs(t));
+ }
+ }
+ finally
+ {
+ if (ReceiverClosed != null)
+ {
+ ReceiverClosed(this, new EventArgs());
+ }
+ }
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/network/io/IoSender.cs b/qpid/dotnet/client-010/client/transport/network/io/IoSender.cs new file mode 100644 index 0000000000..38fda3e1e2 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/network/io/IoSender.cs @@ -0,0 +1,132 @@ +/*
+* 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.Generic;
+using System.IO;
+using System.Net.Sockets;
+using System.Threading;
+using common.org.apache.qpid.transport.util;
+using org.apache.qpid.transport.util;
+
+namespace org.apache.qpid.transport.network.io
+{
+ public sealed class IoSender : Sender<MemoryStream>
+ {
+ private static readonly Logger log = Logger.get(typeof (IoReceiver));
+ private readonly NetworkStream bufStream;
+ private bool closed;
+ private readonly Mutex mutClosed = new Mutex();
+ private readonly CircularBuffer<byte[]> queue;
+ private readonly Thread thread;
+ private readonly int timeout;
+ private MemoryStream _tobeSent = new MemoryStream();
+ public IoSender(IoTransport transport, int queueSize, int timeout)
+ {
+ this.timeout = timeout;
+ bufStream = transport.Stream;
+ queue = new CircularBuffer<byte[]>(2);
+ thread = new Thread(Go);
+ log.debug("Creating IoSender thread");
+ thread.Name = String.Format("IoSender - {0}", transport.Socket) ;
+ thread.IsBackground = true;
+ thread.Start();
+ }
+
+
+ public void send(MemoryStream str)
+ {
+ mutClosed.WaitOne();
+ if (closed)
+ {
+ throw new TransportException("sender is closed");
+ }
+ mutClosed.ReleaseMutex();
+ int length = (int)str.Position;
+ str.Seek(0, SeekOrigin.Begin);
+ byte[] buf = new byte[length];
+ str.Read(buf, 0, length);
+ _tobeSent.Write(buf, 0, length);
+ }
+
+ public void flush()
+ {
+ int length = (int)_tobeSent.Position;
+ byte[] buf = new byte[length];
+ _tobeSent.Seek(0, SeekOrigin.Begin);
+ _tobeSent.Read(buf, 0, length);
+ queue.Enqueue(buf);
+ // bufStream.Write(buf, 0, length);
+ // _tobeSent = new MemoryStream();
+ // _writer.Write(buf, 0, length);
+ // _writer.Flush();
+ _tobeSent.Seek(0, SeekOrigin.Begin);
+ }
+
+ public void close()
+ {
+ log.debug("Closing Sender");
+ mutClosed.WaitOne();
+ if (!closed)
+ {
+ try
+ {
+ closed = true;
+ queue.close();
+ thread.Join(timeout);
+ if (thread.IsAlive)
+ {
+ throw new TransportException("join timed out");
+ }
+ }
+ catch (ThreadInterruptedException e)
+ {
+ throw new TransportException(e);
+ }
+ catch (IOException e)
+ {
+ throw new TransportException(e);
+ }
+ }
+ mutClosed.ReleaseMutex();
+ }
+
+ private void Go()
+ {
+ while (! closed)
+ {
+ //MemoryStream st = queue.Dequeue();
+ byte[] st = queue.Dequeue();
+ if (st != null)
+ {
+ try
+ {
+ // int length = (int) st.Length;
+ // byte[] buf = new byte[length];
+ // st.Read(buf, 0, length);
+ bufStream.Write(st, 0, st.Length);
+ }
+ catch (Exception e)
+ {
+ Console.WriteLine(e);
+ }
+ }
+ }
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/network/io/IoTransport.cs b/qpid/dotnet/client-010/client/transport/network/io/IoTransport.cs new file mode 100644 index 0000000000..f2ebab67c9 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/network/io/IoTransport.cs @@ -0,0 +1,143 @@ +/*
+* 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.IO;
+using System.Net.Sockets;
+using org.apache.qpid.transport.util;
+
+namespace org.apache.qpid.transport.network.io
+{
+ /// <summary>
+ /// This class provides a socket based transport using sync io classes.
+ ///
+ /// The following params are configurable via JVM arguments
+ /// TCP_NO_DELAY - qpid.tcpNoDelay
+ /// SO_RCVBUF - qpid.readBufferSize
+ /// SO_SNDBUF - qpid.writeBufferSize
+ /// </summary>
+ public sealed class IoTransport
+ {
+ // constants
+ private const int DEFAULT_READ_WRITE_BUFFER_SIZE = 64*1024;
+ private const int TIMEOUT = 60000;
+ private const int QUEUE_SIZE = 1000;
+ // props
+ private static readonly Logger log = Logger.get(typeof (IoTransport));
+ private NetworkStream m_stream;
+ private IoSender m_sender;
+ private Receiver<ReceivedPayload<MemoryStream>> m_receiver;
+ private TcpClient m_socket;
+ private Connection m_con;
+
+ public static Connection connect(String host, int port, ConnectionDelegate conndel)
+ {
+ IoTransport transport = new IoTransport(host, port, conndel);
+ return transport.Connection;
+ }
+
+ public IoTransport(String host, int port, ConnectionDelegate conndel)
+ {
+ createSocket(host, port);
+ Sender = new IoSender(this, QUEUE_SIZE, TIMEOUT);
+ Receiver = new IoReceiver(Stream, Socket.ReceiveBufferSize * 2, TIMEOUT);
+ Assembler assembler = new Assembler();
+ InputHandler inputHandler = new InputHandler(InputHandler.State.PROTO_HDR);
+ Connection = new Connection(assembler, new Disassembler(Sender, 64 * 1024 - 1), conndel);
+ // Input handler listen to Receiver events
+ Receiver.Received += inputHandler.On_ReceivedBuffer;
+ // Assembler listen to inputhandler events
+ inputHandler.ReceivedEvent += assembler.On_ReceivedEvent;
+ // Connection listen to asembler protocol event
+ Receiver.Closed += Connection.On_ReceivedClosed;
+ Receiver.Exception += Connection.On_ReceivedException;
+ inputHandler.HandlerClosed += Connection.On_ReceivedClosed;
+ inputHandler.ExceptionProcessing += Connection.On_ReceivedException;
+ assembler.HandlerClosed += Connection.On_ReceivedClosed;
+ assembler.ExceptionProcessing += Connection.On_ReceivedException;
+ assembler.ReceivedEvent += Connection.On_ReceivedEvent;
+ }
+
+ public Connection Connection
+ {
+ get { return m_con; }
+ set { m_con = value; }
+ }
+
+ public Receiver<ReceivedPayload<MemoryStream>> Receiver
+ {
+ get { return m_receiver; }
+ set { m_receiver = value; }
+ }
+
+ public IoSender Sender
+ {
+ get { return m_sender; }
+ set { m_sender = value; }
+ }
+
+
+ public NetworkStream Stream
+ {
+ get { return m_stream; }
+ set { m_stream = value; }
+ }
+
+ public TcpClient Socket
+ {
+ get { return m_socket; }
+ set { m_socket = value; }
+ }
+
+ #region Private Support Functions
+
+ private void createSocket(String host, int port)
+ {
+ try
+ {
+ TcpClient socket = new TcpClient();
+ String noDelay = Environment.GetEnvironmentVariable("qpid.tcpNoDelay");
+ String writeBufferSize = Environment.GetEnvironmentVariable("qpid.writeBufferSize");
+ String readBufferSize = Environment.GetEnvironmentVariable("qpid.readBufferSize");
+ socket.NoDelay = noDelay != null && bool.Parse(noDelay);
+ socket.ReceiveBufferSize = readBufferSize == null ? DEFAULT_READ_WRITE_BUFFER_SIZE : int.Parse(readBufferSize);
+ socket.SendBufferSize = writeBufferSize == null ? DEFAULT_READ_WRITE_BUFFER_SIZE : int.Parse(writeBufferSize);
+
+ log.debug("NoDelay : {0}", socket.NoDelay);
+ log.debug("ReceiveBufferSize : {0}", socket.ReceiveBufferSize);
+ log.debug("SendBufferSize : {0}", socket.SendBufferSize);
+ log.debug("Openning connection with host : {0}; port: {1}", host, port);
+
+ socket.Connect(host, port);
+ Socket = socket;
+ Stream = socket.GetStream();
+ }
+ catch (SocketException e)
+ {
+ Console.WriteLine(e.StackTrace);
+ throw new TransportException("Error connecting to broker", e);
+ }
+ catch (IOException e)
+ {
+ throw new TransportException("Error connecting to broker", e);
+ }
+ }
+
+ #endregion
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/util/ByteEncoder.cs b/qpid/dotnet/client-010/client/transport/util/ByteEncoder.cs new file mode 100644 index 0000000000..75526eb432 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/util/ByteEncoder.cs @@ -0,0 +1,199 @@ +using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace org.apache.qpid.transport.util
+{
+ public static class ByteEncoder
+ {
+ #region Endian conversion helper routines
+ /// <summary>
+ /// Returns the value encoded in Big Endian (PPC, XDR) format.
+ /// </summary>
+ /// <param name="value">Value to encode.</param>
+ /// <returns>Big-endian encoded value.</returns>
+ public static Int32 GetBigEndian(Int32 value)
+ {
+ if (BitConverter.IsLittleEndian)
+ {
+ return swapByteOrder(value);
+ }
+ else
+ {
+ return value;
+ }
+ }
+
+ /// <summary>
+ /// Returns the value encoded in Big Endian (PPC, XDR) format.
+ /// </summary>
+ /// <param name="value">Value to encode.</param>
+ /// <returns>Big-endian encoded value.</returns>
+ public static UInt16 GetBigEndian(UInt16 value)
+ {
+ if (BitConverter.IsLittleEndian)
+ {
+ return swapByteOrder(value);
+ }
+ else
+ {
+ return value;
+ }
+ }
+
+ /// <summary>
+ /// Returns the value encoded in Big Endian (PPC, XDR) format.
+ /// </summary>
+ /// <param name="value">Value to encode.</param>
+ /// <returns>Big-endian encoded value.</returns>
+ public static UInt32 GetBigEndian(UInt32 value)
+ {
+ if (BitConverter.IsLittleEndian)
+ {
+ return swapByteOrder(value);
+ }
+ else
+ {
+ return value;
+ }
+ }
+
+ /// <summary>
+ /// Returns the value encoded in Big Endian (PPC, XDR) format.
+ /// </summary>
+ /// <param name="value">Value to encode.</param>
+ /// <returns>Big-endian encoded value.</returns>
+ public static Double GetBigEndian(Double value)
+ {
+ if (BitConverter.IsLittleEndian)
+ {
+ return swapByteOrder(value);
+ }
+ else
+ {
+ return value;
+ }
+ }
+
+ /// <summary>
+ /// Returns the value encoded in Little Endian (x86, NDR) format.
+ /// </summary>
+ /// <param name="value">Value to encode.</param>
+ /// <returns>Little-endian encoded value.</returns>
+ public static Int32 GetLittleEndian(Int32 value)
+ {
+ if (BitConverter.IsLittleEndian)
+ {
+ return value;
+ }
+ else
+ {
+ return swapByteOrder(value);
+ }
+ }
+
+ /// <summary>
+ /// Returns the value encoded in Little Endian (x86, NDR) format.
+ /// </summary>
+ /// <param name="value">Value to encode.</param>
+ /// <returns>Little-endian encoded value.</returns>
+ public static UInt32 GetLittleEndian(UInt32 value)
+ {
+ if (BitConverter.IsLittleEndian)
+ {
+ return value;
+ }
+ else
+ {
+ return swapByteOrder(value);
+ }
+ }
+
+ /// <summary>
+ /// Returns the value encoded in Little Endian (x86, NDR) format.
+ /// </summary>
+ /// <param name="value">Value to encode.</param>
+ /// <returns>Little-endian encoded value.</returns>
+ public static UInt16 GetLittleEndian(UInt16 value)
+ {
+ if (BitConverter.IsLittleEndian)
+ {
+ return value;
+ }
+ else
+ {
+ return swapByteOrder(value);
+ }
+ }
+
+ /// <summary>
+ /// Returns the value encoded in Little Endian (x86, NDR) format.
+ /// </summary>
+ /// <param name="value">Value to encode.</param>
+ /// <returns>Little-endian encoded value.</returns>
+ public static Double GetLittleEndian(Double value)
+ {
+ if (BitConverter.IsLittleEndian)
+ {
+ return value;
+ }
+ else
+ {
+ return swapByteOrder(value);
+ }
+ }
+
+ /// <summary>
+ /// Swaps the Byte order of an <see cref="Int32"/>.
+ /// </summary>
+ /// <param name="value"><see cref="Int32"/> to swap the bytes of.</param>
+ /// <returns>Byte order swapped <see cref="Int32"/>.</returns>
+ private static Int32 swapByteOrder(Int32 value)
+ {
+ Int32 swapped = (Int32)((0x000000FF) & (value >> 24)
+ | (0x0000FF00) & (value >> 8)
+ | (0x00FF0000) & (value << 8)
+ | (0xFF000000) & (value << 24));
+ return swapped;
+ }
+
+ /// <summary>
+ /// Swaps the byte order of a <see cref="UInt16"/>.
+ /// </summary>
+ /// <param name="value"><see cref="UInt16"/> to swap the bytes of.</param>
+ /// <returns>Byte order swapped <see cref="UInt16"/>.</returns>
+ private static UInt16 swapByteOrder(UInt16 value)
+ {
+ return (UInt16)((0x00FF & (value >> 8))
+ | (0xFF00 & (value << 8)));
+ }
+
+ /// <summary>
+ /// Swaps the byte order of a <see cref="UInt32"/>.
+ /// </summary>
+ /// <param name="value"><see cref="UInt32"/> to swap the bytes of.</param>
+ /// <returns>Byte order swapped <see cref="UInt32"/>.</returns>
+ private static UInt32 swapByteOrder(UInt32 value)
+ {
+ UInt32 swapped = ((0x000000FF) & (value >> 24)
+ | (0x0000FF00) & (value >> 8)
+ | (0x00FF0000) & (value << 8)
+ | (0xFF000000) & (value << 24));
+ return swapped;
+ }
+
+ /// <summary>
+ /// Swaps the byte order of a <see cref="Double"/> (double precision IEEE 754)
+ /// </summary>
+ /// <param name="value"><see cref="Double"/> to swap.</param>
+ /// <returns>Byte order swapped <see cref="Double"/> value.</returns>
+ private static Double swapByteOrder(Double value)
+ {
+ Byte[] buffer = BitConverter.GetBytes(value);
+ Array.Reverse(buffer, 0, buffer.Length);
+ return BitConverter.ToDouble(buffer, 0);
+ }
+ #endregion
+ }
+
+}
diff --git a/qpid/dotnet/client-010/client/transport/util/CircularBuffer.cs b/qpid/dotnet/client-010/client/transport/util/CircularBuffer.cs new file mode 100644 index 0000000000..13bdc8caf1 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/util/CircularBuffer.cs @@ -0,0 +1,134 @@ +/*
+*
+* 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.Threading;
+
+namespace common.org.apache.qpid.transport.util
+{
+ internal class CircularBuffer<T>
+ {
+ private readonly T[] buffer;
+ private Int32 nrp, nwp;
+ private readonly Int32 len;
+ private Int32 countValue;
+ private readonly Int32 add;
+
+
+ /// <summary>
+ /// Constructor creates N=len element
+ /// Circular Buffer that olds MemoryStream
+ /// </summary>
+ public CircularBuffer(Int32 len)
+ {
+ buffer = new T[len];
+ this.len = len;
+ add = 1 - len;
+ nrp = 0;
+ nwp = 0;
+ countValue = 0;
+ }
+
+
+ public void Enqueue(T t)
+ {
+ lock (this)
+ {
+ if (countValue >= (len - 1))
+ {
+ // wait for room to be available
+ Monitor.Wait(this);
+ }
+ bool notifyDequeue = countValue <= 0;
+ load(t);
+ if (notifyDequeue) //notifyDequeue)
+ {
+ Monitor.PulseAll(this);
+ }
+ }
+ }
+
+
+ public T Dequeue()
+ {
+ lock (this)
+ {
+ if (countValue <= 0)
+ {
+ Monitor.Wait(this);
+ }
+ bool notifyEnqueue = countValue >= (len - 1);
+ T temp = get();
+ if (notifyEnqueue) //notifyEnqueue)
+ {
+ Monitor.PulseAll(this);
+ }
+ return temp;
+ }
+ }
+
+ public void close()
+ {
+ nrp = 0;
+ nwp = 0;
+ countValue = 0;
+ Array.Clear(buffer, 0, len);
+ lock (this)
+ {
+ Monitor.PulseAll(this);
+ }
+ }
+
+ #region Private Support Functions
+
+ private void load(T t)
+ {
+ Int32 i = nwp;
+ buffer[i] = t;
+ i += add;
+ if (i < 0) i += len;
+ nwp = i;
+ countValue = nwp - nrp;
+ if (countValue == 0)
+ countValue += len;
+ }
+
+ private void updateCount()
+ {
+ countValue = nwp - nrp;
+ if (countValue <= 0 )
+ countValue += len; // modulo buffer size
+ }
+
+ private T get()
+ {
+ Int32 i = nrp;
+ T temp = buffer[i];
+ i += add;
+ if (i < 0) i += len;
+ nrp = i;
+ countValue = nwp - nrp;
+ return (temp);
+ }
+
+ #endregion
+ }
+}
diff --git a/qpid/dotnet/client-010/client/transport/util/Functions.cs b/qpid/dotnet/client-010/client/transport/util/Functions.cs new file mode 100644 index 0000000000..0ced53c335 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/util/Functions.cs @@ -0,0 +1,42 @@ +/*
+*
+* 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;
+
+namespace org.apache.qpid.transport.util
+{
+
+ /// <summary>
+ /// Functions
+ /// </summary>
+
+ public class Functions
+ {
+ public static sbyte lsb(int i)
+ {
+ return (sbyte) (0xFF & i);
+ }
+
+ public static sbyte lsb(long l)
+ {
+ return (sbyte) (0xFF & l);
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/util/Logger.cs b/qpid/dotnet/client-010/client/transport/util/Logger.cs new file mode 100644 index 0000000000..04473575c2 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/util/Logger.cs @@ -0,0 +1,114 @@ +/*
+*
+* 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 org.apache.qpid.transport.util
+{
+
+ /// <summary> Logger
+ ///
+ /// </summary>
+
+ public sealed class Logger
+ {
+ private readonly ILog log;
+
+ public static Logger get(Type type)
+ {
+ return new Logger(LogManager.GetLogger(type));
+ }
+
+ private Logger(ILog log)
+ {
+ this.log = log;
+ }
+
+ public bool isDebugEnabled()
+ {
+ return log.IsDebugEnabled;
+ }
+
+ public void debug(String message, params Object[] args)
+ {
+ if (log.IsDebugEnabled)
+ {
+ log.Debug(String.Format(message, args));
+ }
+ }
+
+ public void debug(Exception t, String message, params Object[] args)
+ {
+ if (log.IsDebugEnabled)
+ {
+ log.Debug(String.Format(message, args), t);
+ }
+ }
+
+ public void error(String message, params Object[] args)
+ {
+ if (log.IsErrorEnabled)
+ {
+ log.Error(String.Format(message, args));
+ }
+ }
+
+ public void error(Exception t, String message, params Object[] args)
+ {
+ if (log.IsErrorEnabled)
+ {
+ log.Error(String.Format(message, args), t);
+ }
+ }
+
+ public void warn(String message, params Object[] args)
+ {
+ if (log.IsWarnEnabled)
+ {
+ log.Warn(String.Format(message, args));
+ }
+ }
+
+ public void warn(Exception t, String message, params Object[] args)
+ {
+ if (log.IsWarnEnabled)
+ {
+ log.Warn(String.Format(message, args), t);
+ }
+ }
+
+ public void info(String message, params Object[] args)
+ {
+ if (log.IsInfoEnabled)
+ {
+ log.Info(String.Format(message, args));
+ }
+ }
+
+ public void info(Exception t, String message, params Object[] args)
+ {
+ if (log.IsInfoEnabled)
+ {
+ log.Info(String.Format(message, args), t);
+ }
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/util/ResultFuture.cs b/qpid/dotnet/client-010/client/transport/util/ResultFuture.cs new file mode 100644 index 0000000000..8a38cdb356 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/util/ResultFuture.cs @@ -0,0 +1,59 @@ +using System;
+using System.Threading;
+using org.apache.qpid.transport;
+using org.apache.qpid.transport.util;
+
+namespace common.org.apache.qpid.transport.util
+{
+ public class ResultFuture<T> : Future<T> where T : Struct
+ {
+ const long _timeout = 60000;
+ private Struct _result;
+ private Session _session;
+ private static readonly Logger log = Logger.get(typeof(ResultFuture<T>));
+
+ public Struct get(long timeout)
+ {
+ lock (this)
+ {
+ long start = DateTime.Now.Millisecond;
+ long elapsed = 0;
+ while (! _session.Closed && _timeout - elapsed > 0 && _result == null)
+ {
+ log.debug("{0} waiting for result: {1}", _session, this );
+ Monitor.Wait(this, (int) (timeout - elapsed));
+ elapsed = DateTime.Now.Millisecond - start;
+ }
+ }
+ if( _session.Closed )
+ {
+ throw new SessionException(_session.getExceptions());
+ }
+ return _result;
+ }
+
+ public T Result
+ {
+ get { return (T) get(_timeout); }
+ set
+ {
+ lock (this)
+ {
+ _result = value;
+ Monitor.PulseAll(this);
+ }
+ }
+ }
+
+ public Session Session
+ {
+ set { _session = value; }
+ }
+
+ public String toString()
+ {
+ return String.Format("Future({0})", _result);
+ }
+
+ }
+}
diff --git a/qpid/dotnet/client-010/client/transport/util/Serial.cs b/qpid/dotnet/client-010/client/transport/util/Serial.cs new file mode 100644 index 0000000000..e47a6745a5 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/util/Serial.cs @@ -0,0 +1,94 @@ +/*
+*
+* 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 org.apache.qpid.transport.util
+{
+ /// <summary>
+ /// This class provides basic serial number comparisons as defined in
+ /// RFC 1982.
+ /// </summary>
+ public class Serial
+ {
+ ///
+ ///
+ ///Compares two numbers using serial arithmetic.
+ ///
+ /// param s1 the first serial number
+ /// param s2 the second serial number
+ ///
+ /// return a negative integer, zero, or a positive integer as the
+ /// first argument is less than, equal to, or greater than the
+ /// second
+ ///
+ public static int compare(int s1, int s2)
+ {
+ return s1 - s2;
+ }
+
+ public static bool lt(int s1, int s2)
+ {
+ return compare(s1, s2) < 0;
+ }
+
+ public static bool le(int s1, int s2)
+ {
+ return compare(s1, s2) <= 0;
+ }
+
+ public static bool gt(int s1, int s2)
+ {
+ return compare(s1, s2) > 0;
+ }
+
+ public static bool ge(int s1, int s2)
+ {
+ return compare(s1, s2) >= 0;
+ }
+
+ public static bool eq(int s1, int s2)
+ {
+ return s1 == s2;
+ }
+
+ public static int min(int s1, int s2)
+ {
+ if (lt(s1, s2))
+ {
+ return s1;
+ }
+ else
+ {
+ return s2;
+ }
+ }
+
+ public static int max(int s1, int s2)
+ {
+ if (gt(s1, s2))
+ {
+ return s1;
+ }
+ else
+ {
+ return s2;
+ }
+ }
+ }
+}
\ No newline at end of file diff --git a/qpid/dotnet/client-010/client/transport/util/UUID.cs b/qpid/dotnet/client-010/client/transport/util/UUID.cs new file mode 100644 index 0000000000..788c5b99c8 --- /dev/null +++ b/qpid/dotnet/client-010/client/transport/util/UUID.cs @@ -0,0 +1,88 @@ +/*
+*
+* 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;
+
+namespace org.apache.qpid.transport.util
+{
+ public class UUID
+ {
+ private long _mostSigBits;
+
+ private long _leastSigBits;
+
+
+ public UUID(long mostSigBits, long leastSigBits)
+ {
+ _mostSigBits = mostSigBits;
+ _leastSigBits = leastSigBits;
+ }
+
+ public long MostSignificantBits
+ {
+ get { return _mostSigBits; }
+ set { _mostSigBits = value; }
+ }
+
+ public long LeastSignificantBits
+ {
+ get { return _leastSigBits; }
+ set { _leastSigBits = value; }
+ }
+
+ private UUID(byte[] r)
+ {
+ MostSignificantBits = 0;
+ LeastSignificantBits = 0;
+ for (int i = 0; i < 8; i++)
+ MostSignificantBits = (MostSignificantBits << 8) | (r[i] & 0xff);
+ for (int i = 8; i < 16; i++)
+ LeastSignificantBits = (LeastSignificantBits << 8) | (r[i] & 0xff);
+ }
+
+ public static UUID randomUUID()
+ {
+ byte[] randomBytes = new byte[16];
+ Random random = new Random();
+ random.NextBytes(randomBytes);
+ randomBytes[6] &= 0x0f;
+ randomBytes[6] |= 0x40;
+ randomBytes[8] &= 0x3f;
+ randomBytes[8] |= 0x80;
+ return new UUID(randomBytes);
+ }
+
+ public new String ToString()
+ {
+ return (digits(_mostSigBits >> 32, 8) + "-" +
+ digits(_mostSigBits >> 16, 4) + "-" +
+ digits(_mostSigBits, 4) + "-" +
+ digits(_leastSigBits >> 48, 4) + "-" +
+ digits(_leastSigBits, 12));
+ }
+
+ private static String digits(long val, int digits)
+ {
+ long hi = 1L << (digits * 4);
+ return Convert.ToString((hi | (val & (hi - 1))), 16);
+ }
+ }
+}
|