diff options
Diffstat (limited to 'RC9/qpid/dotnet/client-010/client')
70 files changed, 7713 insertions, 0 deletions
diff --git a/RC9/qpid/dotnet/client-010/client/Client.csproj b/RC9/qpid/dotnet/client-010/client/Client.csproj new file mode 100644 index 0000000000..e449781722 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/Client.csproj @@ -0,0 +1,55 @@ +<Project DefaultTargets="Build" xmlns="http://schemas.microsoft.com/developer/msbuild/2003" ToolsVersion="3.5">
+ <PropertyGroup>
+ <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
+ <Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform>
+ <ProductVersion>8.0.50727</ProductVersion>
+ <SchemaVersion>2.0</SchemaVersion>
+ <ProjectGuid>{B911FFD7-754F-4735-A188-218D5065BE79}</ProjectGuid>
+ <OutputType>Library</OutputType>
+ <AppDesignerFolder>Properties</AppDesignerFolder>
+ <RootNamespace>client</RootNamespace>
+ <AssemblyName>Qpid Client</AssemblyName>
+ <FileUpgradeFlags>
+ </FileUpgradeFlags>
+ <OldToolsVersion>2.0</OldToolsVersion>
+ <UpgradeBackupLocation>
+ </UpgradeBackupLocation>
+ </PropertyGroup>
+ <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
+ <DebugSymbols>true</DebugSymbols>
+ <DebugType>full</DebugType>
+ <Optimize>false</Optimize>
+ <OutputPath>bin\Debug\</OutputPath>
+ <DefineConstants>DEBUG;TRACE</DefineConstants>
+ <ErrorReport>prompt</ErrorReport>
+ <WarningLevel>4</WarningLevel>
+ </PropertyGroup>
+ <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
+ <DebugType>pdbonly</DebugType>
+ <Optimize>true</Optimize>
+ <OutputPath>bin\Release\</OutputPath>
+ <DefineConstants>TRACE</DefineConstants>
+ <ErrorReport>prompt</ErrorReport>
+ <WarningLevel>4</WarningLevel>
+ </PropertyGroup>
+ <ItemGroup>
+ <Reference Include="log4net, Version=1.2.10.0, Culture=neutral, PublicKeyToken=1b44e1d426115821">
+ <SpecificVersion>False</SpecificVersion>
+ <HintPath>..\lib\log4net\log4net.dll</HintPath>
+ </Reference>
+ <Reference Include="System" />
+ <Reference Include="System.Data" />
+ <Reference Include="System.Xml" />
+ </ItemGroup>
+ <ItemGroup>
+ <Compile Include="**\*.cs" />
+ </ItemGroup>
+ <Import Project="$(MSBuildBinPath)\Microsoft.CSharp.targets" />
+ <!-- To modify your build process, add your task inside one of the targets below and uncomment it.
+ Other similar extension points exist, see Microsoft.Common.targets.
+ <Target Name="BeforeBuild">
+ </Target>
+ <Target Name="AfterBuild">
+ </Target>
+ -->
+</Project>
\ No newline at end of file diff --git a/RC9/qpid/dotnet/client-010/client/Properties/AssemblyInfo.cs b/RC9/qpid/dotnet/client-010/client/Properties/AssemblyInfo.cs new file mode 100644 index 0000000000..5ab0be9a99 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/Properties/AssemblyInfo.cs @@ -0,0 +1,56 @@ +/* + * + * 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.Reflection; +using System.Runtime.CompilerServices; +using System.Runtime.InteropServices; + +// General Information about an assembly is controlled through the following +// set of attributes. Change these attribute values to modify the information +// associated with an assembly. +[assembly: AssemblyTitle("Qpid Client")] +[assembly: AssemblyDescription("Built from svn revision number: ")] +[assembly: AssemblyConfiguration("")] +[assembly: AssemblyCompany("Apache Software Foundation")] +[assembly: AssemblyProduct("Qpid Client")] +[assembly: AssemblyCopyright("Apache Software Foundation")] +[assembly: AssemblyTrademark("")] +[assembly: AssemblyCulture("")] + +// Setting ComVisible to false makes the types in this assembly not visible +// to COM components. If you need to access a type in this assembly from +// COM, set the ComVisible attribute to true on that type. +[assembly: ComVisible(false)] + +// The following GUID is for the ID of the typelib if this project is exposed to COM +[assembly: Guid("dac7ef42-e9c8-45a5-8050-1301b6f8160e")] + +// Version information for an assembly consists of the following four values: +// +// Major Version +// Minor Version +// Build Number +// Revision +// +// You can specify all the values or you can default the Revision and Build Numbers +// by using the '*' as shown below: +[assembly: AssemblyVersion("0.10.0.0")] +[assembly: AssemblyFileVersion("1.0.0.0")] diff --git a/RC9/qpid/dotnet/client-010/client/client.sln b/RC9/qpid/dotnet/client-010/client/client.sln new file mode 100644 index 0000000000..c94509e72a --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/client.sln @@ -0,0 +1,104 @@ +
+Microsoft Visual Studio Solution File, Format Version 10.00
+# Visual Studio 2008
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Client", "Client.csproj", "{B911FFD7-754F-4735-A188-218D5065BE79}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Demo", "..\demo\Demo.csproj", "{E4C46FBC-7560-406D-BFEF-CA010E584DF4}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ExcelAddIn", "..\addins\ExcelAddIn\ExcelAddIn.csproj", "{85EFD719-39F6-4471-90FF-9E621430344B}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ExcelAddInProducer", "..\addins\ExcelAddInProducer\ExcelAddInProducer.csproj", "{80F00C3B-EB38-4B3B-9F77-68977A30B155}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "example-direct-producer", "..\examples\direct\example-direct-producer\example-direct-producer.csproj", "{96FCB250-8142-40EE-9BDD-CA839EE21021}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "example-direct-Listener", "..\examples\direct\example-direct-Listener\example-direct-Listener.csproj", "{AE65B1B9-8779-4CB1-91AF-E7F6C7A736D7}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "example-pub-sub-Listener", "..\examples\pub-sub\example-pub-sub-Listener\example-pub-sub-Listener.csproj", "{2BCDC2CC-5BDA-4CC7-944D-2899AD8A53C7}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "example-pub-sub-Publisher", "..\examples\pub-sub\example-pub-sub-Publisher\example-pub-sub-Publisher.csproj", "{F8857634-A134-44E7-A953-F2B22688C599}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Test", "..\test\Test.csproj", "{95CB4C90-7C53-44A9-B11C-96235F158ACA}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "example-request-response-Client", "..\examples\request-response\example-request-response-Client\example-request-response-Client.csproj", "{1BC63815-4029-4039-9207-35E7E06ECC99}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "example-request-response-Server", "..\examples\request-response\example-request-response-Server\example-request-response-Server.csproj", "{922FBA9C-E483-4AEF-ABE8-AC87421E829B}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "example-fanout-Producer", "..\examples\fanout\example-fanout-Producer\example-fanout-Producer.csproj", "{4513BF94-D54A-42FE-8506-FE2CD57B2C51}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "example-fanout-Listener", "..\examples\fanout\example-fanout-Listener\example-fanout-Listener.csproj", "{18A0792B-DC3A-4EC5-93D6-DB8A111D8F15}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "perftest", "..\perftest\perftest.csproj", "{7F7E8DE7-FDF2-4A52-A4CE-D3756B05273C}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ExcelAddInMessageProcessor", "..\addins\ExcelAddInMessageProcessor\ExcelAddInMessageProcessor.csproj", "{C2AE83A3-D5D1-469D-8611-A4738B9997CF}"
+EndProject
+Global
+ GlobalSection(SolutionConfigurationPlatforms) = preSolution
+ Debug|Any CPU = Debug|Any CPU
+ Release|Any CPU = Release|Any CPU
+ EndGlobalSection
+ GlobalSection(ProjectConfigurationPlatforms) = postSolution
+ {B911FFD7-754F-4735-A188-218D5065BE79}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {B911FFD7-754F-4735-A188-218D5065BE79}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {B911FFD7-754F-4735-A188-218D5065BE79}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {B911FFD7-754F-4735-A188-218D5065BE79}.Release|Any CPU.Build.0 = Release|Any CPU
+ {E4C46FBC-7560-406D-BFEF-CA010E584DF4}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {E4C46FBC-7560-406D-BFEF-CA010E584DF4}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {E4C46FBC-7560-406D-BFEF-CA010E584DF4}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {E4C46FBC-7560-406D-BFEF-CA010E584DF4}.Release|Any CPU.Build.0 = Release|Any CPU
+ {85EFD719-39F6-4471-90FF-9E621430344B}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {85EFD719-39F6-4471-90FF-9E621430344B}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {85EFD719-39F6-4471-90FF-9E621430344B}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {85EFD719-39F6-4471-90FF-9E621430344B}.Release|Any CPU.Build.0 = Release|Any CPU
+ {80F00C3B-EB38-4B3B-9F77-68977A30B155}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {80F00C3B-EB38-4B3B-9F77-68977A30B155}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {80F00C3B-EB38-4B3B-9F77-68977A30B155}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {80F00C3B-EB38-4B3B-9F77-68977A30B155}.Release|Any CPU.Build.0 = Release|Any CPU
+ {96FCB250-8142-40EE-9BDD-CA839EE21021}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {96FCB250-8142-40EE-9BDD-CA839EE21021}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {96FCB250-8142-40EE-9BDD-CA839EE21021}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {96FCB250-8142-40EE-9BDD-CA839EE21021}.Release|Any CPU.Build.0 = Release|Any CPU
+ {AE65B1B9-8779-4CB1-91AF-E7F6C7A736D7}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {AE65B1B9-8779-4CB1-91AF-E7F6C7A736D7}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {AE65B1B9-8779-4CB1-91AF-E7F6C7A736D7}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {AE65B1B9-8779-4CB1-91AF-E7F6C7A736D7}.Release|Any CPU.Build.0 = Release|Any CPU
+ {2BCDC2CC-5BDA-4CC7-944D-2899AD8A53C7}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {2BCDC2CC-5BDA-4CC7-944D-2899AD8A53C7}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {2BCDC2CC-5BDA-4CC7-944D-2899AD8A53C7}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {2BCDC2CC-5BDA-4CC7-944D-2899AD8A53C7}.Release|Any CPU.Build.0 = Release|Any CPU
+ {F8857634-A134-44E7-A953-F2B22688C599}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {F8857634-A134-44E7-A953-F2B22688C599}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {F8857634-A134-44E7-A953-F2B22688C599}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {F8857634-A134-44E7-A953-F2B22688C599}.Release|Any CPU.Build.0 = Release|Any CPU
+ {95CB4C90-7C53-44A9-B11C-96235F158ACA}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {95CB4C90-7C53-44A9-B11C-96235F158ACA}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {95CB4C90-7C53-44A9-B11C-96235F158ACA}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {95CB4C90-7C53-44A9-B11C-96235F158ACA}.Release|Any CPU.Build.0 = Release|Any CPU
+ {1BC63815-4029-4039-9207-35E7E06ECC99}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {1BC63815-4029-4039-9207-35E7E06ECC99}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {1BC63815-4029-4039-9207-35E7E06ECC99}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {1BC63815-4029-4039-9207-35E7E06ECC99}.Release|Any CPU.Build.0 = Release|Any CPU
+ {922FBA9C-E483-4AEF-ABE8-AC87421E829B}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {922FBA9C-E483-4AEF-ABE8-AC87421E829B}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {922FBA9C-E483-4AEF-ABE8-AC87421E829B}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {922FBA9C-E483-4AEF-ABE8-AC87421E829B}.Release|Any CPU.Build.0 = Release|Any CPU
+ {4513BF94-D54A-42FE-8506-FE2CD57B2C51}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {4513BF94-D54A-42FE-8506-FE2CD57B2C51}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {4513BF94-D54A-42FE-8506-FE2CD57B2C51}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {4513BF94-D54A-42FE-8506-FE2CD57B2C51}.Release|Any CPU.Build.0 = Release|Any CPU
+ {18A0792B-DC3A-4EC5-93D6-DB8A111D8F15}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {18A0792B-DC3A-4EC5-93D6-DB8A111D8F15}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {18A0792B-DC3A-4EC5-93D6-DB8A111D8F15}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {18A0792B-DC3A-4EC5-93D6-DB8A111D8F15}.Release|Any CPU.Build.0 = Release|Any CPU
+ {7F7E8DE7-FDF2-4A52-A4CE-D3756B05273C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {7F7E8DE7-FDF2-4A52-A4CE-D3756B05273C}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {7F7E8DE7-FDF2-4A52-A4CE-D3756B05273C}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {7F7E8DE7-FDF2-4A52-A4CE-D3756B05273C}.Release|Any CPU.Build.0 = Release|Any CPU
+ {C2AE83A3-D5D1-469D-8611-A4738B9997CF}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {C2AE83A3-D5D1-469D-8611-A4738B9997CF}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {C2AE83A3-D5D1-469D-8611-A4738B9997CF}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {C2AE83A3-D5D1-469D-8611-A4738B9997CF}.Release|Any CPU.Build.0 = Release|Any CPU
+ EndGlobalSection
+ GlobalSection(SolutionProperties) = preSolution
+ HideSolutionNode = FALSE
+ EndGlobalSection
+EndGlobal
diff --git a/RC9/qpid/dotnet/client-010/client/client.suo b/RC9/qpid/dotnet/client-010/client/client.suo Binary files differnew file mode 100644 index 0000000000..cfdbf4be93 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/client.suo diff --git a/RC9/qpid/dotnet/client-010/client/client/Client.cs b/RC9/qpid/dotnet/client-010/client/client/Client.cs new file mode 100644 index 0000000000..4871a1c3d9 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/client/Client.cs @@ -0,0 +1,145 @@ +/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+using System;
+using System.Text;
+using System.Threading;
+using org.apache.qpid.transport;
+using org.apache.qpid.transport.network.io;
+using org.apache.qpid.transport.util;
+
+namespace org.apache.qpid.client
+{
+ public class Client : ClientInterface
+ {
+ private Connection _conn;
+ private static readonly Logger _log = Logger.get(typeof (Client));
+ private const long timeout = 60000;
+ private bool _closed;
+ private readonly Object _closeOK;
+ private ClosedListener _closedListner;
+
+ public bool Closed
+ {
+ get { return _closed; }
+ set { _closed = value; }
+ }
+
+ public Object CloseOk
+ {
+ get { return _closeOK; }
+ }
+
+ public Client()
+ {
+ _closed = false;
+ _closeOK = new object();
+ }
+
+ #region Interface ClientInterface
+
+ /// <summary>
+ /// Establishes a connection with a broker using the provided user auths
+ ///
+ /// </summary>
+ /// <param name="host">Host name on which a broker is deployed</param>
+ /// <param name="port">Broker port </param>
+ /// <param name="virtualHost">virtual host name</param>
+ /// <param name="username">User Name</param>
+ /// <param name="password">Password</param>
+ public void connect(String host, int port, String virtualHost, String username, String password)
+ {
+ _log.debug(String.Format("Client Connecting to host {0}; port {1}; virtualHost {2}; username {3}", host,
+ port, virtualHost, username));
+ ConnectionDelegate connectionDelegate = new ClientConnectionDelegate(this, username, password);
+ ManualResetEvent negotiationComplete = new ManualResetEvent(false);
+ connectionDelegate.setCondition(negotiationComplete);
+ connectionDelegate.VirtualHost = virtualHost;
+ _conn = IoTransport.connect(host, port, connectionDelegate);
+
+ _conn.send(new ProtocolHeader(1, 0, 10));
+ negotiationComplete.WaitOne();
+ }
+
+ /// <summary>
+ /// Establishes a connection with a broker using SSL
+ ///
+ /// </summary>
+ /// <param name="host">Host name on which a broker is deployed</param>
+ /// <param name="port">Broker port </param>
+ /// <param name="virtualHost">virtual host name</param>
+ /// <param name="username">User Name</param>
+ /// <param name="password">Password</param>
+ /// <param name="serverName">Name of the SSL server</param>
+ /// <param name="certPath">Path to the X509 certificate to be used for client authentication</param>
+ /// <param name="rejectUntrusted">If true connection will not be established if the broker is not trusted</param>
+ public void connectSSL(String host, int port, String virtualHost, String username, String password, string serverName, string certPath, bool rejectUntrusted)
+ {
+ _log.debug(String.Format("Client Connecting to host {0}; port {1}; virtualHost {2}; username {3}", host,
+ port, virtualHost, username));
+ _log.debug(String.Format("SSL paramters: serverName: {0}; certPath: {1}; rejectUntrusted: {2}", serverName, certPath, rejectUntrusted));
+ ConnectionDelegate connectionDelegate = new ClientConnectionDelegate(this, username, password);
+ ManualResetEvent negotiationComplete = new ManualResetEvent(false);
+ connectionDelegate.setCondition(negotiationComplete);
+ connectionDelegate.VirtualHost = virtualHost;
+ _conn = IoSSLTransport.connect(host, port, serverName, certPath, rejectUntrusted, connectionDelegate);
+
+ _conn.send(new ProtocolHeader(1, 0, 10));
+ negotiationComplete.WaitOne();
+ }
+
+ public void close()
+ {
+ Channel ch = _conn.getChannel(0);
+ ch.connectionClose(ConnectionCloseCode.NORMAL, "client is closing");
+ lock (CloseOk)
+ {
+ DateTime start = DateTime.Now;
+ long elapsed = 0;
+ while (!Closed && elapsed < timeout)
+ {
+ Monitor.Wait(CloseOk, (int) (timeout - elapsed));
+ elapsed = DateTime.Now.Subtract(start).Milliseconds;
+ }
+ if (!Closed)
+ {
+ throw new Exception("Timed out when closing connection");
+ }
+ _conn.close();
+ }
+ }
+
+ public ClientSession createSession(long expiryInSeconds)
+ {
+ Channel ch = _conn.getChannel();
+ ClientSession ssn = new ClientSession(Encoding.UTF8.GetBytes(UUID.randomUUID().ToString()));
+ ssn.attach(ch);
+ ssn.sessionAttach(ssn.getName());
+ ssn.sessionRequestTimeout(expiryInSeconds);
+ return ssn;
+ }
+
+ public ClosedListener ClosedListener
+ {
+ set { _closedListner = value; }
+ get { return _closedListner; }
+ }
+
+ #endregion
+ }
+}
diff --git a/RC9/qpid/dotnet/client-010/client/client/ClientConnectionDelegate.cs b/RC9/qpid/dotnet/client-010/client/client/ClientConnectionDelegate.cs new file mode 100644 index 0000000000..5ed4187b36 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/client/ClientConnectionDelegate.cs @@ -0,0 +1,97 @@ +/*
+* 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.Text;
+using System.Threading;
+using org.apache.qpid.transport;
+using org.apache.qpid.transport.util;
+
+namespace org.apache.qpid.client
+{
+ internal class ClientConnectionDelegate : ClientDelegate
+ {
+ private static readonly Logger log = Logger.get(typeof (ClientConnectionDelegate));
+ private readonly Client _client;
+ private string _username;
+ private string _password;
+
+ public ClientConnectionDelegate(Client client, string username, string pasword)
+ {
+ _client = client;
+ _username = username;
+ _password = pasword;
+ }
+
+ public override SessionDelegate getSessionDelegate()
+ {
+ return new ClientSessionDelegate();
+ }
+
+ public override void exception(Exception t)
+ {
+ throw t;
+ }
+
+ public override void connectionStart(Channel context, ConnectionStart mystruct)
+ {
+ const string mechanism = "PLAIN";
+ MemoryStream stResponse = new MemoryStream();
+ byte[] part = Encoding.UTF8.GetBytes(_username);
+ stResponse.WriteByte(0);
+ stResponse.Write(part, 0, part.Length);
+ stResponse.WriteByte(0);
+ part = Encoding.UTF8.GetBytes(_password);
+ stResponse.Write(part, 0, part.Length);
+ Dictionary<String, Object> props = new Dictionary<String, Object>();
+ context.connectionStartOk(props, mechanism, stResponse.ToArray(), "utf8");
+ }
+
+ public override void closed()
+ {
+ log.debug("Delegate closed");
+ lock (_client.CloseOk)
+ {
+ try
+ {
+ _client.Closed = true;
+ Monitor.PulseAll(_client.CloseOk);
+ }
+ catch (Exception e)
+ {
+ throw new SystemException("Error when closing client", e);
+ }
+ }
+ }
+
+ public override void connectionClose(Channel context, ConnectionClose connectionClose)
+ {
+ base.connectionClose(context, connectionClose);
+ ErrorCode errorCode = ErrorCode.getErrorCode((int) connectionClose.getReplyCode());
+ if (_client.ClosedListener == null && errorCode.Code != (int) QpidErrorCode.NO_ERROR)
+ {
+ throw new Exception ("Server closed the connection: Reason " +
+ connectionClose.getReplyText());
+ }
+ _client.ClosedListener.onClosed(errorCode, connectionClose.getReplyText(), null);
+ }
+ }
+}
diff --git a/RC9/qpid/dotnet/client-010/client/client/ClientInterface.cs b/RC9/qpid/dotnet/client-010/client/client/ClientInterface.cs new file mode 100644 index 0000000000..85be5886c6 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/client/ClientInterface.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.client
+{
+ public interface ClientInterface
+ {
+ /// <summary>
+ /// Establish a connection with the broker using the given parameters
+ ///
+ /// </summary>
+ /// <param name="host">host name</param>
+ /// <param name="port">port number</param>
+ /// <param name="virtualHost">virtualHost the virtual host name</param>
+ /// <param name="username"> username user name</param>
+ /// <param name="passwor">password password</param>
+ void connect(String host, int port, String virtualHost, String username, String passwor);
+
+ /// <summary>
+ /// Close this client
+ /// </summary>
+ void close();
+
+ /// <summary>
+ /// Create a session for this connection.
+ /// The returned session is suspended
+ /// (i.e. this session is not attached to an underlying channel)
+ /// </summary>
+ /// <param name="expiryInSeconds">Expiry time expressed in seconds, if the value is less than
+ /// or equal to 0 then the session does not expire.</param>
+ /// <returns>A newly created (suspended) session.</returns>
+ ClientSession createSession(long expiryInSeconds);
+
+ /// <summary>
+ /// If the communication layer detects a serious problem with a connection, it
+ // informs the client's ClosedListener
+ /// </summary>
+ ///
+ ClosedListener ClosedListener { set; }
+ }
+}
diff --git a/RC9/qpid/dotnet/client-010/client/client/ClientSession.cs b/RC9/qpid/dotnet/client-010/client/client/ClientSession.cs new file mode 100644 index 0000000000..2dbc27ecd9 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/client/ClientSession.cs @@ -0,0 +1,110 @@ +/* + * + * 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.Text; +using org.apache.qpid.transport; +using org.apache.qpid.transport.util; + +namespace org.apache.qpid.client +{ + /// <summary> Implements a Qpid Sesion.</summary> + public class ClientSession : Session + { + public static short TRANSFER_ACQUIRE_MODE_NO_ACQUIRE = 1; + public static short TRANSFER_ACQUIRE_MODE_PRE_ACQUIRE = 0; + public static short TRANSFER_CONFIRM_MODE_REQUIRED = 0; + public static short TRANSFER_CONFIRM_MODE_NOT_REQUIRED = 1; + public static short MESSAGE_FLOW_MODE_CREDIT = 0; + public static short MESSAGE_FLOW_MODE_WINDOW = 1; + public static short MESSAGE_FLOW_UNIT_MESSAGE = 0; + public static short MESSAGE_FLOW_UNIT_BYTE = 1; + public static long MESSAGE_FLOW_MAX_BYTES = 0xFFFFFFFF; + public static short MESSAGE_REJECT_CODE_GENERIC = 0; + public static short MESSAGE_REJECT_CODE_IMMEDIATE_DELIVERY_FAILED = 1; + public static short MESSAGE_ACQUIRE_ANY_AVAILABLE_MESSAGE = 0; + public static short MESSAGE_ACQUIRE_MESSAGES_IF_ALL_ARE_AVAILABLE = 1; + + private Dictionary<String, IMessageListener> _listeners = new Dictionary<String, IMessageListener>(); + + public ClientSession(byte[] name) : base(name) + { + } + + public void attachMessageListener(IMessageListener listener, string Destination) + { + _listeners.Add(Destination, listener); + } + + public Dictionary<String, IMessageListener> MessageListeners + { + get { return _listeners; } + } + + public void messageTransfer(String destination, string routingkey, IMessage message) + { + message.DeliveryProperties.setRoutingKey(routingkey); + messageTransfer(destination, message); + } + + public void messageTransfer(String destination, IMessage message) + { + byte[] body = new byte[message.Body.Position]; + message.Body.Seek(0, SeekOrigin.Begin); + message.Body.Read(body, 0, body.Length); + message.MessageProperties.setMessageId(UUID.randomUUID()); + messageTransfer(destination, + MessageAcceptMode.NONE, + MessageAcquireMode.PRE_ACQUIRED, + message.Header, + body); + } + + public void queueDeclare(String queue) + { + queueDeclare(queue, null, null); + } + + public void queueDeclare(String queue, params Option[] options) + { + queueDeclare(queue, null, null, options); + } + + public void exchangeBind(String queue, String exchange, String bindingKey) + { + exchangeBind(queue, exchange, bindingKey, null); + } + + public void messageSubscribe(String queue) + { + messageSubscribe(queue, queue, MessageAcceptMode.EXPLICIT, MessageAcquireMode.PRE_ACQUIRED, null, 0, null); + // issue credits + messageSetFlowMode(queue, MessageFlowMode.WINDOW); + messageFlow(queue, MessageCreditUnit.BYTE, ClientSession.MESSAGE_FLOW_MAX_BYTES); + messageFlow(queue, MessageCreditUnit.MESSAGE, 10000); + } + + } +} diff --git a/RC9/qpid/dotnet/client-010/client/client/ClientSessionDelegate.cs b/RC9/qpid/dotnet/client-010/client/client/ClientSessionDelegate.cs new file mode 100644 index 0000000000..1b40e2ba45 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/client/ClientSessionDelegate.cs @@ -0,0 +1,55 @@ +/*
+* 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;
+using org.apache.qpid.transport.util;
+
+namespace org.apache.qpid.client
+{
+ public class ClientSessionDelegate : SessionDelegate
+ {
+ private static readonly Logger _log = Logger.get(typeof (ClientSessionDelegate));
+
+ // --------------------------------------------
+ // Message methods
+ // --------------------------------------------
+ public override void messageTransfer(Session session, MessageTransfer xfr)
+ {
+ if (((ClientSession) session).MessageListeners.ContainsKey(xfr.getDestination()))
+ {
+ IMessageListener listener = ((ClientSession)session).MessageListeners[xfr.getDestination()];
+ listener.messageTransfer( new Message(xfr));
+ }
+ else
+ {
+ _log.warn("No listener set for: {0}", xfr);
+ }
+ }
+
+ public override void messageReject(Session session, MessageReject mstruct)
+ {
+ foreach (Range range in mstruct.getTransfers())
+ {
+ for (long l = range.Lower; l <= range.Upper; l++)
+ {
+ _log.warn("message rejected: " + session.getCommand((int) l));
+ }
+ }
+ }
+ }
+}
\ No newline at end of file diff --git a/RC9/qpid/dotnet/client-010/client/client/ClosedListenerInterface.cs b/RC9/qpid/dotnet/client-010/client/client/ClosedListenerInterface.cs new file mode 100644 index 0000000000..8089c3bd90 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/client/ClosedListenerInterface.cs @@ -0,0 +1,29 @@ +/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+
+using System;
+
+namespace org.apache.qpid.client
+{
+ public interface ClosedListener
+ {
+
+ void onClosed(ErrorCode errorCode, String reason, Exception t);
+ }
+}
diff --git a/RC9/qpid/dotnet/client-010/client/client/ErrorCode.cs b/RC9/qpid/dotnet/client-010/client/client/ErrorCode.cs new file mode 100644 index 0000000000..226f078dac --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/client/ErrorCode.cs @@ -0,0 +1,140 @@ +/*
+* 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.client
+{
+ public enum QpidErrorCode
+ {
+ NO_ERROR = 200,
+ CONTENT_TOO_LARGE = 311,
+ NO_ROUTE = 312,
+ NO_CONSUMERS = 313,
+ CONNECTION_FORCED = 320,
+ INVALID_PATH = 402,
+ ACCESS_REFUSED = 403,
+ NOT_FOUND = 404,
+ RESOURCE_LOCKED = 405,
+ PRE_CONDITION_FAILED = 406,
+ FRAME_ERROR = 501,
+ SYNTAX_ERROR = 502,
+ COMMAND_INVALID = 503,
+ SESSION_ERROR = 504,
+ NOT_ALLOWED = 530,
+ NOT_IMPLEMENTED = 540,
+ INTERNAL_ERROR = 541,
+ INVALID_ARGUMENT = 542,
+ UNDEFINED = 1
+ }
+
+ public struct ErrorCode
+ {
+ private int _code;
+ private String _desc;
+ private readonly bool _hardError;
+
+ public ErrorCode(int code, String desc, bool hardError)
+ {
+ _code = code;
+ _desc = desc;
+ _hardError = hardError;
+ }
+
+ public int Code
+ {
+ get { return _code; }
+ set { _code = value; }
+ }
+
+ public String Description
+ {
+ get { return _desc; }
+ set { _desc = value; }
+ }
+
+ public bool ISHardError
+ {
+ get { return _hardError; }
+ }
+
+ public static ErrorCode getErrorCode(int code)
+ {
+ switch (code)
+ {
+ case 200:
+ return
+ new ErrorCode(200, "reply-success", true);
+ case 311:
+ return
+ new ErrorCode(311, "content-too-large", false);
+ case 312:
+ return
+ new ErrorCode(312, "no-route", false);
+ case 313:
+ return
+ new ErrorCode(313, "content-consumers", false);
+ case 320:
+ return
+ new ErrorCode(320, "connection-forced", true);
+ case 402:
+ return
+ new ErrorCode(402, "invalid-path", true);
+ case 403:
+ return
+ new ErrorCode(403, "access-refused", false);
+ case 404:
+ return
+ new ErrorCode(404, "not-found", false);
+ case 405:
+ return
+ new ErrorCode(405, "resource-locked", false);
+ case 406:
+ return
+ new ErrorCode(406, "precondition-failed", false);
+ case 501:
+ return
+ new ErrorCode(501, "frame_error", true);
+ case 502:
+ return
+ new ErrorCode(502, "syntax_error", true);
+ case 503:
+ return
+ new ErrorCode(503, "command_invalid", true);
+ case 504:
+ return
+ new ErrorCode(504, "sesion_error", true);
+ case 530:
+ return
+ new ErrorCode(530, "not_allowed", true);
+ case 540:
+ return
+ new ErrorCode(540, "not_implemented", true);
+ case 541:
+ return
+ new ErrorCode(541, "internal_error", true);
+ case 542:
+ return
+ new ErrorCode(542, "invalid_argument", true);
+ default:
+ return new ErrorCode(1, "undefined", true);
+ }
+ }
+ }
+}
\ No newline at end of file diff --git a/RC9/qpid/dotnet/client-010/client/client/IMessage.cs b/RC9/qpid/dotnet/client-010/client/client/IMessage.cs new file mode 100644 index 0000000000..10b92de7ff --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/client/IMessage.cs @@ -0,0 +1,48 @@ +/*
+*
+* 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;
+
+namespace org.apache.qpid.client
+{
+ public interface IMessage
+ {
+ int Id { get; }
+
+ Header Header { get; set; }
+
+ MessageProperties MessageProperties { get; set; }
+
+ DeliveryProperties DeliveryProperties { get; set; }
+
+ Dictionary<String, Object> ApplicationHeaders { get; set; }
+
+ void appendData(byte[] bytes);
+
+ MemoryStream Body { get; }
+
+ string Destination { get; }
+
+ void clearData();
+ }
+}
diff --git a/RC9/qpid/dotnet/client-010/client/client/Message.cs b/RC9/qpid/dotnet/client-010/client/client/Message.cs new file mode 100644 index 0000000000..567ca5cae7 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/client/Message.cs @@ -0,0 +1,131 @@ +/*
+*
+* 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;
+using System.IO;
+using org.apache.qpid.transport;
+
+namespace org.apache.qpid.client
+{
+ public class Message : IMessage
+ {
+ private readonly MessageTransfer _message;
+
+ public Message(MessageTransfer m)
+ {
+ _message = m;
+ }
+
+ public Message()
+ {
+ _message = new MessageTransfer();
+ _message.Header = new Header( new MessageProperties(), new DeliveryProperties());
+ ((MessageProperties) _message.Header.Structs[0]).setApplicationHeaders(new Dictionary<string, object>());
+ }
+
+ public MessageProperties MessageProperties
+ {
+ get
+ {
+ if (_message.Header != null && Header.Structs.Length > 1)
+ return (MessageProperties) Header.Structs[0];
+ return null;
+ }
+ set
+ {
+ if (_message.Header != null)
+ {
+ Header.Structs[0] = value;
+ }
+ }
+ }
+
+ public DeliveryProperties DeliveryProperties
+ {
+ get
+ {
+ if (Header != null)
+ {
+ if( Header.Structs.Length > 1 )
+ return (DeliveryProperties)Header.Structs[1];
+ return (DeliveryProperties)Header.Structs[0];
+ }
+
+ return null;
+ }
+ set
+ {
+ if (Header != null)
+ {
+ Header.Structs[1] = value;
+ }
+ }
+ }
+
+ public Dictionary<string, object> ApplicationHeaders
+ {
+ get
+ {
+ if (Header != null)
+ return ((MessageProperties) Header.Structs[0]).getApplicationHeaders();
+ return null;
+ }
+ set
+ {
+ if (Header != null)
+ {
+ ((MessageProperties) Header.Structs[0]).setApplicationHeaders(value);
+ }
+ }
+ }
+
+ public void appendData(byte[] bytes)
+ {
+ Body.Write(bytes, 0, bytes.Length);
+ }
+
+ public void clearData()
+ {
+ Body.Seek(0, SeekOrigin.Begin);
+ }
+
+ public Header Header
+ {
+ get{ return _message.Header;}
+ set{ _message.Header = value;}
+ }
+
+ public MemoryStream Body
+ {
+ get { return _message.Body; }
+ set { _message.Body = value; }
+ }
+
+ public int Id
+ {
+ get { return _message.Id; }
+ }
+
+ public string Destination
+ {
+ get{ return _message.getDestination();}
+ }
+ }
+}
diff --git a/RC9/qpid/dotnet/client-010/client/client/MessageListenerInterface.cs b/RC9/qpid/dotnet/client-010/client/client/MessageListenerInterface.cs new file mode 100644 index 0000000000..978248e04c --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/client/MessageListenerInterface.cs @@ -0,0 +1,31 @@ +/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+
+
+namespace org.apache.qpid.client
+{
+ public interface IMessageListener
+ {
+ /// <summary>
+ /// Inform the listener of the message transfer
+ /// </summary>
+ /// <param name="xfr">The message transfer object</param>
+ void messageTransfer(IMessage xfr);
+ }
+}
diff --git a/RC9/qpid/dotnet/client-010/client/default.build b/RC9/qpid/dotnet/client-010/client/default.build new file mode 100644 index 0000000000..139796a58d --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/default.build @@ -0,0 +1,46 @@ +<?xml version="1.0"?> +<!-- + + 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. + +--> + +<project name="qpid.client" default="build"> + <!-- + Properties that come from master build file + - build.dir: root directory for build + - build.debug: true if building debug release + - build.defines: variables to define during build + --> + + <target name="build"> + <csc target="library" + define="${build.defines}" + debug="${build.debug}" + output="${build.dir}/${project::get-name()}.dll"> + + <sources> + <include name="**/*.cs" /> + </sources> + <references> + <include name="${build.dir}/log4net.dll" /> + </references> + </csc> + </target> +</project> + diff --git a/RC9/qpid/dotnet/client-010/client/transport/Binary.cs b/RC9/qpid/dotnet/client-010/client/transport/Binary.cs new file mode 100644 index 0000000000..ab829dfb69 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/Binding.cs b/RC9/qpid/dotnet/client-010/client/transport/Binding.cs new file mode 100644 index 0000000000..e07e85990d --- /dev/null +++ b/RC9/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<R> receiver<R>(E endpoint) where R : EventArgs;
+ }
+}
\ No newline at end of file diff --git a/RC9/qpid/dotnet/client-010/client/transport/Channel.cs b/RC9/qpid/dotnet/client-010/client/transport/Channel.cs new file mode 100644 index 0000000000..6915c123e7 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/Channel.cs @@ -0,0 +1,174 @@ +/*
+*
+* 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;
+
+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 invoke(Method m, Future future)
+ {
+ throw new Exception("UnsupportedOperation");
+ }
+
+ public String toString()
+ {
+ return String.Format("{0}:{1}", _connection, _channel);
+ }
+ }
+}
\ No newline at end of file diff --git a/RC9/qpid/dotnet/client-010/client/transport/ChannelDelegate.cs b/RC9/qpid/dotnet/client-010/client/transport/ChannelDelegate.cs new file mode 100644 index 0000000000..34126ddbf4 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/ChannelDelegate.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.
+*
+*/
+namespace org.apache.qpid.transport
+{
+ /// <summary>
+ /// ChannelDelegate
+ ///
+ /// </summary>
+ internal class ChannelDelegate : MethodDelegate<Channel>
+ {
+ public override void sessionDetached(Channel channel, SessionDetached closed)
+ {
+ channel.closed();
+ }
+
+ public override void sessionDetach(Channel channel, SessionDetach dtc)
+ {
+ channel.Session.closed();
+ channel.sessionDetached(dtc.getName(), SessionDetachCode.NORMAL);
+ channel.closed();
+ }
+ }
+}
\ No newline at end of file diff --git a/RC9/qpid/dotnet/client-010/client/transport/ClientDelegate.cs b/RC9/qpid/dotnet/client-010/client/transport/ClientDelegate.cs new file mode 100644 index 0000000000..fa0062ef19 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/Connection.cs b/RC9/qpid/dotnet/client-010/client/transport/Connection.cs new file mode 100644 index 0000000000..863f90e093 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/ConnectionDelegate.cs b/RC9/qpid/dotnet/client-010/client/transport/ConnectionDelegate.cs new file mode 100644 index 0000000000..1f76840b98 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/ConnectionDelegate.cs @@ -0,0 +1,108 @@ +/*
+*
+* 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)
+ {
+ 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/RC9/qpid/dotnet/client-010/client/transport/Field.cs b/RC9/qpid/dotnet/client-010/client/transport/Field.cs new file mode 100644 index 0000000000..e8c3cb3497 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/Future.cs b/RC9/qpid/dotnet/client-010/client/transport/Future.cs new file mode 100644 index 0000000000..ddb07f1dda --- /dev/null +++ b/RC9/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
+ {
+ Struct Result
+ {
+ get; set;
+ }
+
+ Session Session
+ { set;
+ }
+ }
+}
\ No newline at end of file diff --git a/RC9/qpid/dotnet/client-010/client/transport/Header.cs b/RC9/qpid/dotnet/client-010/client/transport/Header.cs new file mode 100644 index 0000000000..74b8a7fe2e --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/Method.cs b/RC9/qpid/dotnet/client-010/client/transport/Method.cs new file mode 100644 index 0000000000..c15343ba73 --- /dev/null +++ b/RC9/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 new 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/RC9/qpid/dotnet/client-010/client/transport/ProtocolDelegate.cs b/RC9/qpid/dotnet/client-010/client/transport/ProtocolDelegate.cs new file mode 100644 index 0000000000..d269b01b79 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/ProtocolError.cs b/RC9/qpid/dotnet/client-010/client/transport/ProtocolError.cs new file mode 100644 index 0000000000..933a18c78a --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/ProtocolEvent.cs b/RC9/qpid/dotnet/client-010/client/transport/ProtocolEvent.cs new file mode 100644 index 0000000000..0f6529dfd3 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/ProtocolHeader.cs b/RC9/qpid/dotnet/client-010/client/transport/ProtocolHeader.cs new file mode 100644 index 0000000000..39a3f5670c --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/Range.cs b/RC9/qpid/dotnet/client-010/client/transport/Range.cs new file mode 100644 index 0000000000..a5bc4f0488 --- /dev/null +++ b/RC9/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)
+ {
+ _lower = lower;
+ _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/RC9/qpid/dotnet/client-010/client/transport/RangeSet.cs b/RC9/qpid/dotnet/client-010/client/transport/RangeSet.cs new file mode 100644 index 0000000000..0c502541ec --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/RangeSet.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.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 List<Range> _ranges = new List<Range>();
+
+ IEnumerator IEnumerable.GetEnumerator()
+ {
+ return GetEnumerator();
+ }
+
+ public IEnumerator<Range> GetEnumerator()
+ {
+ return _ranges.GetEnumerator();
+ }
+
+
+ public int size()
+ {
+ return _ranges.Count;
+ }
+
+
+ public Range getFirst()
+ {
+ return _ranges[0];
+ }
+
+ 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)
+ {
+ for (int i = 0; i < _ranges.Count; i++)
+ {
+ Range r = _ranges[i];
+ if (range.touches(r))
+ {
+ _ranges.Remove(r);
+ range = range.span(r);
+ }
+ else if (Serial.lt(range.Upper, r.Lower))
+ {
+ _ranges.Insert(i - 1 , range);
+ return;
+ }
+ }
+ _ranges.Add(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.Add(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/RC9/qpid/dotnet/client-010/client/transport/ReceivedPayload.cs b/RC9/qpid/dotnet/client-010/client/transport/ReceivedPayload.cs new file mode 100644 index 0000000000..792ca4f977 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/Receiver.cs b/RC9/qpid/dotnet/client-010/client/transport/Receiver.cs new file mode 100644 index 0000000000..7b61d7f684 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/Receiver.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;
+
+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/RC9/qpid/dotnet/client-010/client/transport/Sender.cs b/RC9/qpid/dotnet/client-010/client/transport/Sender.cs new file mode 100644 index 0000000000..564dd83789 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/Session.cs b/RC9/qpid/dotnet/client-010/client/transport/Session.cs new file mode 100644 index 0000000000..5de379ab45 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/Session.cs @@ -0,0 +1,522 @@ +/*
+*
+* 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.Text;
+using System.Threading;
+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> _results = new Dictionary<int, Future>();
+ private readonly List<ExecutionException> _exceptions = new List<ExecutionException>();
+
+
+ public bool Closed
+ {
+ get
+ {
+ lock (this)
+ {
+ return _closed;
+ }
+ }
+ set
+ {
+ lock (this)
+ {
+ _closed = value;
+ }
+ }
+ }
+
+ public string Name
+ {
+ get
+ {
+ ASCIIEncoding enc = new ASCIIEncoding();
+ return enc.GetString(_name);
+ }
+ }
+
+ 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);
+ }
+
+ DateTime start = DateTime.Now;
+ 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.Subtract(start).Milliseconds;
+ }
+
+ 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 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 invoke(Method m, Future future)
+ {
+ lock (_commands)
+ {
+ future.Session = this;
+ int command = _commandsOut;
+ lock (_results)
+ {
+ _results.Add(command, 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)
+ {
+ DateTime start = DateTime.Now;
+ long elapsed = 0;
+
+ while (! Closed && elapsed < _timeout)
+ {
+ Monitor.Wait(_commands, (int) (_timeout - elapsed));
+ elapsed = DateTime.Now.Subtract(start).Milliseconds;
+ }
+ }
+ }
+
+ public void exception(Exception t)
+ {
+ log.error(t, "Caught exception");
+ }
+
+ public void closed()
+ {
+ Closed = true;
+ lock (_commands)
+ {
+ Monitor.PulseAll(_commands);
+ }
+ lock (_results)
+ {
+ foreach (Future 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/RC9/qpid/dotnet/client-010/client/transport/SessionDelegate.cs b/RC9/qpid/dotnet/client-010/client/transport/SessionDelegate.cs new file mode 100644 index 0000000000..e4b58b20ff --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/SessionDelegate.cs @@ -0,0 +1,126 @@ +/*
+*
+* 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>
+ /// 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/RC9/qpid/dotnet/client-010/client/transport/Struct.cs b/RC9/qpid/dotnet/client-010/client/transport/Struct.cs new file mode 100644 index 0000000000..fead22268b --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/codec/AbstractDecoder.cs b/RC9/qpid/dotnet/client-010/client/transport/codec/AbstractDecoder.cs new file mode 100644 index 0000000000..44aecb0933 --- /dev/null +++ b/RC9/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)BitConverter.Int64BitsToDouble(readUint32() << 32);
+
+ case Code.BIN64:
+ case Code.UINT64:
+ case Code.INT64:
+ case Code.DATETIME:
+ return readUint64();
+
+ case Code.DOUBLE:
+ return BitConverter.Int64BitsToDouble(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/RC9/qpid/dotnet/client-010/client/transport/codec/AbstractEncoder.cs b/RC9/qpid/dotnet/client-010/client/transport/codec/AbstractEncoder.cs new file mode 100644 index 0000000000..c89b96462e --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/codec/AbstractEncoder.cs @@ -0,0 +1,583 @@ +/*
+*
+* 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.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 == Code.VOID)
+ {
+ throw new Exception
+ ("unable to resolve type: " + klass + ", " + value);
+ }
+ else
+ {
+ return type;
+ }
+ }
+
+ private static Code resolve(Type klass)
+ {
+ Code type;
+ if(ENCODINGS.ContainsKey(klass))
+ {
+ return ENCODINGS[klass];
+ }
+
+ Type sup = klass.BaseType;
+ if (sup != null)
+ {
+ type = resolve(sup);
+
+ if (type != Code.VOID)
+ {
+ return type;
+ }
+ }
+ foreach (Type iface in klass.GetInterfaces())
+ {
+ type = resolve(iface);
+ if (type != Code.VOID)
+ {
+ 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:
+ byte[] b = BitConverter.GetBytes((char) value);
+ put(b[0]);
+ 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(BitConverter.DoubleToInt64Bits((float) value) >> 32);
+ break;
+
+ case Code.BIN64:
+ case Code.UINT64:
+ case Code.INT64:
+ case Code.DATETIME:
+ writeUint64((long) value);
+ break;
+
+ case Code.DOUBLE:
+ writeUint64( BitConverter.DoubleToInt64Bits((double) 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/RC9/qpid/dotnet/client-010/client/transport/codec/Decoder.cs b/RC9/qpid/dotnet/client-010/client/transport/codec/Decoder.cs new file mode 100644 index 0000000000..207bd60df1 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/codec/Encodable.cs b/RC9/qpid/dotnet/client-010/client/transport/codec/Encodable.cs new file mode 100644 index 0000000000..12ab7c52f1 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/codec/Encoder.cs b/RC9/qpid/dotnet/client-010/client/transport/codec/Encoder.cs new file mode 100644 index 0000000000..32950ee02e --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/codec/MSDecoder.cs b/RC9/qpid/dotnet/client-010/client/transport/codec/MSDecoder.cs new file mode 100644 index 0000000000..cc8971d38b --- /dev/null +++ b/RC9/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(reader.ReadInt64());
+ }
+ }
+}
\ No newline at end of file diff --git a/RC9/qpid/dotnet/client-010/client/transport/codec/MSEncoder.cs b/RC9/qpid/dotnet/client-010/client/transport/codec/MSEncoder.cs new file mode 100644 index 0000000000..b2ccbb89ae --- /dev/null +++ b/RC9/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(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/RC9/qpid/dotnet/client-010/client/transport/exception/ConnectionException.cs b/RC9/qpid/dotnet/client-010/client/transport/exception/ConnectionException.cs new file mode 100644 index 0000000000..d380024c39 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/exception/ExceptionArgs.cs b/RC9/qpid/dotnet/client-010/client/transport/exception/ExceptionArgs.cs new file mode 100644 index 0000000000..e9cc1b6cad --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/exception/ProtocolVersionException.cs b/RC9/qpid/dotnet/client-010/client/transport/exception/ProtocolVersionException.cs new file mode 100644 index 0000000000..2182ab9ba4 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/exception/SessionClosedException.cs b/RC9/qpid/dotnet/client-010/client/transport/exception/SessionClosedException.cs new file mode 100644 index 0000000000..9f3080ac96 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/exception/SessionException.cs b/RC9/qpid/dotnet/client-010/client/transport/exception/SessionException.cs new file mode 100644 index 0000000000..b586019802 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/exception/TransportException.cs b/RC9/qpid/dotnet/client-010/client/transport/exception/TransportException.cs new file mode 100644 index 0000000000..1c216d51e4 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/network/Assembler.cs b/RC9/qpid/dotnet/client-010/client/transport/network/Assembler.cs new file mode 100644 index 0000000000..6f81c8edee --- /dev/null +++ b/RC9/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<byte[]>> 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<byte[]>>();
+ 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())
+ {
+ byte[] tmp = new byte[frame.BodySize];
+ frame.Body.Read(tmp, 0, tmp.Length);
+ segment = new MemoryStream();
+ BinaryWriter w = new BinaryWriter(segment);
+ w.Write(tmp);
+ assemble(frame, new MemoryStream(tmp));
+ }
+ else
+ {
+ List<byte[]> frames;
+ if (frame.isFirstFrame())
+ {
+ frames = new List<byte[]>();
+ setSegment(frame, frames);
+ }
+ else
+ {
+ frames = getSegment(frame);
+ }
+ byte[] tmp = new byte[frame.BodySize];
+ frame.Body.Read(tmp, 0, tmp.Length);
+ frames.Add(tmp);
+
+ if (frame.isLastFrame())
+ {
+ clearSegment(frame);
+ segment = new MemoryStream();
+ BinaryWriter w = new BinaryWriter(segment);
+ foreach (byte[] f in frames)
+ {
+ w.Write(f);
+ }
+ 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())
+ {
+ 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];
+ segment.Seek(0, SeekOrigin.Begin);
+ command.Body = segment;
+ 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<byte[]> getSegment(Frame frame)
+ {
+ return segments[segmentKey(frame)];
+ }
+
+ private void setSegment(Frame frame, List<byte[]> 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/RC9/qpid/dotnet/client-010/client/transport/network/Disassembler.cs b/RC9/qpid/dotnet/client-010/client/transport/network/Disassembler.cs new file mode 100644 index 0000000000..4cf16a98fe --- /dev/null +++ b/RC9/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 IIOSender<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(IIOSender<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, size);
+ }
+ }
+
+ 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;
+ buf.Seek(0, SeekOrigin.Begin);
+ 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/RC9/qpid/dotnet/client-010/client/transport/network/Frame.cs b/RC9/qpid/dotnet/client-010/client/transport/network/Frame.cs new file mode 100644 index 0000000000..5b7e15b1aa --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/network/IIoSender.cs b/RC9/qpid/dotnet/client-010/client/transport/network/IIoSender.cs new file mode 100644 index 0000000000..67a52ef707 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/network/IIoSender.cs @@ -0,0 +1,28 @@ +/*
+*
+* 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
+{
+ public interface IIOSender<T>:Sender<T>
+ {
+ void send(T body, int siz);
+ }
+}
diff --git a/RC9/qpid/dotnet/client-010/client/transport/network/InputHandler.cs b/RC9/qpid/dotnet/client-010/client/transport/network/InputHandler.cs new file mode 100644 index 0000000000..f3bffdb821 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/network/InputHandler.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.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;
+ if( input != null )
+ {
+ remaining += (int) input.Length;
+ }
+ try
+ {
+ while (remaining > 0)
+ {
+ if (remaining >= needed)
+ {
+ if (input != null)
+ {
+ byte[] tmp = new byte[buf.Length];
+ buf.Read(tmp, 0, tmp.Length);
+ input.Write(tmp, 0, tmp.Length);
+ 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
+ {
+ byte[] tmp;
+ if (input == null)
+ {
+ input = new MemoryStream();
+ tmp = new byte[remaining];
+ }
+ else
+ {
+ // this is a full buffer
+ tmp = new byte[buf.Length];
+ }
+ buf.Read(tmp, 0, tmp.Length);
+ input.Write(tmp, 0, tmp.Length);
+ 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/RC9/qpid/dotnet/client-010/client/transport/network/NetworkDelegate.cs b/RC9/qpid/dotnet/client-010/client/transport/network/NetworkDelegate.cs new file mode 100644 index 0000000000..9549f5c295 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/network/NetworkEvent.cs b/RC9/qpid/dotnet/client-010/client/transport/network/NetworkEvent.cs new file mode 100644 index 0000000000..c682e77560 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/network/io/IIoTransport.cs b/RC9/qpid/dotnet/client-010/client/transport/network/io/IIoTransport.cs new file mode 100644 index 0000000000..aab017dad1 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/network/io/IIoTransport.cs @@ -0,0 +1,57 @@ +/*
+* 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.IO;
+using System.Net.Sockets;
+
+namespace org.apache.qpid.transport.network.io
+{
+ public interface IIoTransport
+ {
+ Connection Connection
+ {
+ get;
+ set;
+ }
+
+ Receiver<ReceivedPayload<MemoryStream>> Receiver
+ {
+ get;
+ set;
+ }
+
+ IoSender Sender
+ {
+ get;
+ set;
+ }
+
+
+ Stream Stream
+ {
+ get;
+ set;
+ }
+
+ TcpClient Socket
+ {
+ get;
+ set;
+ }
+ }
+}
diff --git a/RC9/qpid/dotnet/client-010/client/transport/network/io/IoReceiver.cs b/RC9/qpid/dotnet/client-010/client/transport/network/io/IoReceiver.cs new file mode 100644 index 0000000000..4c59cf0a6c --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/network/io/IoReceiver.cs @@ -0,0 +1,189 @@ +/*
+*
+* 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 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 (IOException e)
+ {
+ // IOException is thrown when the socket is closed according to the docs
+ }
+ catch (Exception 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/RC9/qpid/dotnet/client-010/client/transport/network/io/IoSSLTransport.cs b/RC9/qpid/dotnet/client-010/client/transport/network/io/IoSSLTransport.cs new file mode 100644 index 0000000000..8c5f161a35 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/network/io/IoSSLTransport.cs @@ -0,0 +1,194 @@ +/*
+* 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.Security;
+using System.Net.Sockets;
+using System.Security.Authentication;
+using System.Security.Cryptography.X509Certificates;
+using org.apache.qpid.transport.util;
+
+namespace org.apache.qpid.transport.network.io
+{
+ public sealed class IoSSLTransport : IIoTransport
+ {
+ // 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 (IoSSLTransport));
+ private Stream m_stream;
+ private IoSender m_sender;
+ private Receiver<ReceivedPayload<MemoryStream>> m_receiver;
+ private TcpClient m_socket;
+ private Connection m_con;
+ private readonly bool _rejectUntrusted;
+
+ public static Connection connect(String host, int port, string serverName, string certPath, bool rejectUntrusted, ConnectionDelegate conndel)
+ {
+ IIoTransport transport = new IoSSLTransport(host, port, serverName, certPath, rejectUntrusted, conndel);
+ return transport.Connection;
+ }
+
+ public IoSSLTransport(String host, int port, string serverName, string certPath, bool rejectUntrusted, ConnectionDelegate conndel)
+ {
+ _rejectUntrusted = rejectUntrusted;
+ createSocket(host, port, serverName, certPath);
+ 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 Stream 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, string serverName, string certPath)
+ {
+ TcpClient socket;
+ try
+ {
+ 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;
+ }
+ catch (Exception e)
+ {
+ throw new TransportException("Error connecting to broker", e);
+ }
+ try
+ {
+ //Initializes a new instance of the SslStream class using the specified Stream, stream closure behavior, certificate validation delegate and certificate selection delegate
+ SslStream sslStream = new SslStream(socket.GetStream(), false, ValidateServerCertificate, LocalCertificateSelection);
+ if (certPath != null)
+ {
+ X509CertificateCollection col = new X509CertificateCollection();
+ X509Certificate cert = X509Certificate.CreateFromCertFile(certPath);
+ col.Add(cert);
+ sslStream.AuthenticateAsClient(serverName, col, SslProtocols.Default, true);
+ }
+ else
+ {
+ sslStream.AuthenticateAsClient(serverName);
+ }
+ Stream = sslStream;
+ }
+ catch (AuthenticationException e)
+ {
+ log.warn("Exception: {0}", e.Message);
+ if (e.InnerException != null)
+ {
+ log.warn("Inner exception: {0}", e.InnerException.Message);
+ }
+ socket.Close();
+ throw new TransportException("Authentication failed - closing the connection.");
+ }
+ }
+
+ // The following method is invoked by the RemoteCertificateValidationDelegate.
+ public bool ValidateServerCertificate(
+ object sender,
+ X509Certificate certificate,
+ X509Chain chain,
+ SslPolicyErrors sslPolicyErrors)
+ {
+ bool result = true;
+ if (sslPolicyErrors != SslPolicyErrors.None && _rejectUntrusted )
+ {
+ log.warn("Certificate error: {0}", sslPolicyErrors);
+ // Do not allow this client to communicate with unauthenticated servers.
+ result = false;
+ }
+ return result;
+ }
+
+ public X509Certificate LocalCertificateSelection(
+ Object sender,
+ string targetHost,
+ X509CertificateCollection localCertificates,
+ X509Certificate remoteCertificate,
+ string[] acceptableIssuers
+ )
+ {
+ return remoteCertificate;
+ }
+
+ #endregion
+ }
+}
diff --git a/RC9/qpid/dotnet/client-010/client/transport/network/io/IoSender.cs b/RC9/qpid/dotnet/client-010/client/transport/network/io/IoSender.cs new file mode 100644 index 0000000000..924d871dd2 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/network/io/IoSender.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.IO;
+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 : IIOSender<MemoryStream>
+ {
+ private static readonly Logger log = Logger.get(typeof (IoReceiver));
+ private readonly Stream bufStream;
+ private bool closed;
+ private readonly Mutex mutClosed = new Mutex();
+ private readonly CircularBuffer<byte[]> queue;
+ private readonly Thread thread;
+ private readonly int timeout;
+ private readonly MemoryStream _tobeSent = new MemoryStream();
+ public IoSender(IIoTransport transport, int queueSize, int timeout)
+ {
+ this.timeout = timeout;
+ bufStream = transport.Stream;
+ queue = new CircularBuffer<byte[]>(queueSize);
+ 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)
+ {
+ int pos = (int) str.Position;
+ str.Seek(0, SeekOrigin.Begin);
+ send(str, pos);
+ }
+
+ public void send(MemoryStream str, int size)
+ {
+ mutClosed.WaitOne();
+ if (closed)
+ {
+ throw new TransportException("sender is closed");
+ }
+ mutClosed.ReleaseMutex();
+ byte[] buf = new byte[size];
+ str.Read(buf, 0, size);
+ _tobeSent.Write(buf, 0, size);
+ }
+
+ 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/RC9/qpid/dotnet/client-010/client/transport/network/io/IoTransport.cs b/RC9/qpid/dotnet/client-010/client/transport/network/io/IoTransport.cs new file mode 100644 index 0000000000..3a2397870d --- /dev/null +++ b/RC9/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 : IIoTransport
+ {
+ // 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 Stream 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 Stream 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/RC9/qpid/dotnet/client-010/client/transport/util/ByteEncoder.cs b/RC9/qpid/dotnet/client-010/client/transport/util/ByteEncoder.cs new file mode 100644 index 0000000000..ac493a9b62 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/util/ByteEncoder.cs @@ -0,0 +1,193 @@ +/*
+*
+* 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 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);
+ }
+ 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);
+ }
+ 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);
+ }
+ 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 long GetBigEndian(long value)
+ {
+ if (BitConverter.IsLittleEndian)
+ {
+ return swapByteOrder(value);
+ }
+ 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;
+ }
+ 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;
+ }
+ 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;
+ }
+ 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 long GetLittleEndian(long value)
+ {
+ if (BitConverter.IsLittleEndian)
+ {
+ return value;
+ }
+ 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 long swapByteOrder(long value)
+ {
+ Byte[] buffer = BitConverter.GetBytes(value);
+ Array.Reverse(buffer, 0, buffer.Length);
+ return BitConverter.ToInt64(buffer, 0);
+ }
+ #endregion
+ }
+
+}
diff --git a/RC9/qpid/dotnet/client-010/client/transport/util/CircularBuffer.cs b/RC9/qpid/dotnet/client-010/client/transport/util/CircularBuffer.cs new file mode 100644 index 0000000000..05c26bcc49 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/util/CircularBuffer.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.Threading;
+
+namespace common.org.apache.qpid.transport.util
+{
+ public 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;
+ updateCount();
+ }
+
+ 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--;
+ return (temp);
+ }
+
+ #endregion
+ }
+}
diff --git a/RC9/qpid/dotnet/client-010/client/transport/util/Functions.cs b/RC9/qpid/dotnet/client-010/client/transport/util/Functions.cs new file mode 100644 index 0000000000..f5777d8ee9 --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/util/Functions.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.
+*
+*/
+
+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/RC9/qpid/dotnet/client-010/client/transport/util/Logger.cs b/RC9/qpid/dotnet/client-010/client/transport/util/Logger.cs new file mode 100644 index 0000000000..04473575c2 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/util/ResultFuture.cs b/RC9/qpid/dotnet/client-010/client/transport/util/ResultFuture.cs new file mode 100644 index 0000000000..5b31c9d71a --- /dev/null +++ b/RC9/qpid/dotnet/client-010/client/transport/util/ResultFuture.cs @@ -0,0 +1,80 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ + +using 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 : Future + { + const long _timeout = 60000; + private Struct _result; + private Session _session; + private static readonly Logger log = Logger.get(typeof(ResultFuture)); + + public Struct get(long timeout) + { + lock (this) + { + DateTime start = DateTime.Now; + 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 = (long) (DateTime.Now.Subtract(start)).TotalMilliseconds; + } + } + if( _session.Closed ) + { + throw new SessionException(_session.getExceptions()); + } + return _result; + } + + public Struct Result + { + get { return 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/RC9/qpid/dotnet/client-010/client/transport/util/Serial.cs b/RC9/qpid/dotnet/client-010/client/transport/util/Serial.cs new file mode 100644 index 0000000000..e47a6745a5 --- /dev/null +++ b/RC9/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/RC9/qpid/dotnet/client-010/client/transport/util/UUID.cs b/RC9/qpid/dotnet/client-010/client/transport/util/UUID.cs new file mode 100644 index 0000000000..788c5b99c8 --- /dev/null +++ b/RC9/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);
+ }
+ }
+}
|