You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2007/01/05 17:16:49 UTC
svn commit: r493064 - in /incubator/qpid/trunk/qpid/dotnet: ./
Qpid.Client.Tests/ Qpid.Client/ Qpid.Client/Client/
Qpid.Client/Client/Transport/Socket/Blocking/
Author: rgreig
Date: Fri Jan 5 08:16:48 2007
New Revision: 493064
URL: http://svn.apache.org/viewvc?view=rev&rev=493064
Log:
Patch for Qpid-239 applied. BlockingSocketTransport instantiated directly.
Added:
incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketProcessor.cs
incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketTransport.cs
incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/ByteChannel.cs
Modified:
incubator/qpid/trunk/qpid/dotnet/Qpid.Client.Tests/Qpid.Client.Tests.csproj
incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/AMQConnection.cs
incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Qpid.Client.csproj
incubator/qpid/trunk/qpid/dotnet/Qpid.NET.sln
Modified: incubator/qpid/trunk/qpid/dotnet/Qpid.Client.Tests/Qpid.Client.Tests.csproj
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/dotnet/Qpid.Client.Tests/Qpid.Client.Tests.csproj?view=diff&rev=493064&r1=493063&r2=493064
==============================================================================
--- incubator/qpid/trunk/qpid/dotnet/Qpid.Client.Tests/Qpid.Client.Tests.csproj (original)
+++ incubator/qpid/trunk/qpid/dotnet/Qpid.Client.Tests/Qpid.Client.Tests.csproj Fri Jan 5 08:16:48 2007
@@ -58,10 +58,6 @@
<Compile Include="url\ConnectionUrlTest.cs" />
</ItemGroup>
<ItemGroup>
- <ProjectReference Include="..\Qpid.Client.Transport.Socket.Blocking\Qpid.Client.Transport.Socket.Blocking.csproj">
- <Project>{52AC4940-2077-4104-A753-29A9C8C16957}</Project>
- <Name>Qpid.Client.Transport.Socket.Blocking</Name>
- </ProjectReference>
<ProjectReference Include="..\Qpid.Messaging\Qpid.Messaging.csproj">
<Project>{6688F826-C58E-4C1B-AA1F-22AFAB4B7D07}</Project>
<Name>Qpid.Messaging</Name>
Modified: incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/AMQConnection.cs
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/AMQConnection.cs?view=diff&rev=493064&r1=493063&r2=493064
==============================================================================
--- incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/AMQConnection.cs (original)
+++ incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/AMQConnection.cs Fri Jan 5 08:16:48 2007
@@ -29,6 +29,7 @@
using Qpid.Client.qms;
using Qpid.Client.State;
using Qpid.Client.Transport;
+using Qpid.Client.Transport.Socket.Blocking;
using Qpid.Collections;
using Qpid.Framing;
using Qpid.Messaging;
@@ -177,7 +178,7 @@
}
}
- private ITransport LoadTransportFromAssembly(string host, int port, String assemblyName, String transportType)
+ /*private ITransport LoadTransportFromAssembly(string host, int port, String assemblyName, String transportType)
{
//Assembly assembly = Assembly.LoadFrom(assemblyName);
Assembly assembly = Assembly.Load(assemblyName);
@@ -205,7 +206,7 @@
_log.Info("transport = " + result);
return result;
- }
+ }*/
public void Disconnect()
{
@@ -688,12 +689,16 @@
_protocolListener = new AMQProtocolListener(this, _stateManager);
_protocolListener.AddFrameListener(_stateManager);
+ /*
// Currently there is only one transport option - BlockingSocket.
String assemblyName = "Qpid.Client.Transport.Socket.Blocking.dll";
String transportType = "Qpid.Client.Transport.Socket.Blocking.BlockingSocketTransport";
// Load the transport assembly dynamically.
_transport = LoadTransportFromAssembly(brokerDetail.getHost(), brokerDetail.getPort(), assemblyName, transportType);
+ */
+
+ _transport = new BlockingSocketTransport(brokerDetail.getHost(), brokerDetail.getPort(), this);
// Connect.
_transport.Open();
Added: incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketProcessor.cs
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketProcessor.cs?view=auto&rev=493064
==============================================================================
--- incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketProcessor.cs (added)
+++ incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketProcessor.cs Fri Jan 5 08:16:48 2007
@@ -0,0 +1,116 @@
+/*
+ *
+ * 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.Net;
+using System.Net.Sockets;
+using log4net;
+using Qpid.Buffer;
+using Qpid.Client.Protocol;
+
+namespace Qpid.Client.Transport.Socket.Blocking
+{
+ class BlockingSocketProcessor : IConnectionCloser
+ {
+ private static readonly ILog _log = LogManager.GetLogger(typeof(BlockingSocketProcessor));
+
+ string _host;
+ int _port;
+ System.Net.Sockets.Socket _socket;
+ private NetworkStream _networkStream;
+ IByteChannel _byteChannel;
+ IProtocolListener _protocolListener;
+
+ public BlockingSocketProcessor(string host, int port, IProtocolListener protocolListener)
+ {
+ _host = host;
+ _port = port;
+ _protocolListener = protocolListener;
+ _byteChannel = new ByteChannel(this);
+ }
+
+ /// <summary>
+ /// Synchronous blocking connect.
+ /// </summary>
+ public void Connect()
+ {
+ _socket = new System.Net.Sockets.Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
+
+ IPHostEntry ipHostInfo = Dns.Resolve(_host); // Note: don't fix this warning. We do this for .NET 1.1 compatibility.
+ IPAddress ipAddress = ipHostInfo.AddressList[0];
+
+ IPEndPoint ipe = new IPEndPoint(ipAddress, _port);
+
+ _socket.Connect(ipe);
+ _networkStream = new NetworkStream(_socket, true);
+ }
+
+ public string getLocalEndPoint()
+ {
+ return _socket.LocalEndPoint.ToString();
+ }
+
+ public void Write(ByteBuffer byteBuffer)
+ {
+ try
+ {
+ _networkStream.Write(byteBuffer.array(), byteBuffer.position(), byteBuffer.limit()); // FIXME
+ }
+ catch (Exception e)
+ {
+ _log.Error("Write caused exception", e);
+ _protocolListener.OnException(e);
+ }
+ }
+
+ public ByteBuffer Read()
+ {
+ const int bufferSize = 4 * 1024; // TODO: Prevent constant allocation of buffers.
+ byte[] bytes = new byte[bufferSize];
+
+ int numOctets = _networkStream.Read(bytes, 0, bytes.Length);
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
+ byteBuffer.limit(numOctets);
+
+ byteBuffer.flip();
+
+ return byteBuffer;
+ }
+
+ public void Disconnect()
+ {
+ _networkStream.Flush();
+ _networkStream.Close();
+ _socket.Close();
+ }
+
+ public void Close()
+ {
+ Disconnect();
+ }
+
+ public IByteChannel ByteChannel
+ {
+ get { return _byteChannel; }
+ }
+ }
+}
+
Added: incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketTransport.cs
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketTransport.cs?view=auto&rev=493064
==============================================================================
--- incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketTransport.cs (added)
+++ incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketTransport.cs Fri Jan 5 08:16:48 2007
@@ -0,0 +1,120 @@
+/*
+ *
+ * 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.Threading;
+using log4net;
+using Qpid.Client.Protocol;
+using Qpid.Framing;
+
+namespace Qpid.Client.Transport.Socket.Blocking
+{
+ public class BlockingSocketTransport : ITransport
+ {
+// static readonly ILog _log = LogManager.GetLogger(typeof(BlockingSocketTransport));
+
+ // Configuration variables.
+ string _host;
+ int _port;
+ IProtocolListener _protocolListener;
+
+ // Runtime variables.
+ private BlockingSocketProcessor _socketProcessor;
+ private AmqpChannel _amqpChannel;
+
+ private ReaderRunner _readerRunner;
+ private Thread _readerThread;
+
+ public BlockingSocketTransport(string host, int port, AMQConnection connection)
+ {
+ _host = host;
+ _port = port;
+ _protocolListener = connection.ProtocolListener;
+ }
+
+ public void Open()
+ {
+ _socketProcessor = new BlockingSocketProcessor(_host, _port, _protocolListener);
+ _socketProcessor.Connect();
+ _amqpChannel = new AmqpChannel(_socketProcessor.ByteChannel);
+ _readerRunner = new ReaderRunner(this);
+ _readerThread = new Thread(new ThreadStart(_readerRunner.Run));
+ _readerThread.Start();
+ }
+
+ public string getLocalEndPoint()
+ {
+ return _socketProcessor.getLocalEndPoint();
+ }
+
+ public void Close()
+ {
+ StopReaderThread();
+ _socketProcessor.Disconnect();
+ }
+
+ public IProtocolChannel ProtocolChannel { get { return _amqpChannel; } }
+ public IProtocolWriter ProtocolWriter { get { return _amqpChannel; } }
+
+ public void StopReaderThread()
+ {
+ _readerRunner.Stop();
+ }
+
+ class ReaderRunner
+ {
+ BlockingSocketTransport _transport;
+ bool _running = true;
+
+ public ReaderRunner(BlockingSocketTransport transport)
+ {
+ _transport = transport;
+ }
+
+ public void Run()
+ {
+ try
+ {
+ while (_running)
+ {
+ Queue frames = _transport.ProtocolChannel.Read();
+
+ foreach (IDataBlock dataBlock in frames)
+ {
+ _transport._protocolListener.OnMessage(dataBlock);
+ }
+ }
+ }
+ catch (Exception e)
+ {
+ _transport._protocolListener.OnException(e);
+ }
+ }
+
+ public void Stop()
+ {
+ // TODO: Check if this is thread safe. running is not volitile....
+ _running = false;
+ }
+ }
+ }
+}
+
Added: incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/ByteChannel.cs
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/ByteChannel.cs?view=auto&rev=493064
==============================================================================
--- incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/ByteChannel.cs (added)
+++ incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/ByteChannel.cs Fri Jan 5 08:16:48 2007
@@ -0,0 +1,63 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+using System;
+using log4net;
+using Qpid.Buffer;
+
+namespace Qpid.Client.Transport.Socket.Blocking
+{
+ class ByteChannel : IByteChannel
+ {
+ // Warning: don't use this log for regular logging.
+ private static readonly ILog _ioTraceLog = LogManager.GetLogger("Qpid.Client.ByteChannel.Tracing");
+
+ BlockingSocketProcessor processor;
+
+ public ByteChannel(BlockingSocketProcessor processor)
+ {
+ this.processor = processor;
+ }
+
+ public ByteBuffer Read()
+ {
+ ByteBuffer result = processor.Read();
+
+ // TODO: Move into decorator.
+ if (_ioTraceLog.IsDebugEnabled)
+ {
+ _ioTraceLog.Debug(String.Format("READ {0}", result));
+ }
+
+ return result;
+ }
+
+ public void Write(ByteBuffer buffer)
+ {
+ // TODO: Move into decorator.
+ if (_ioTraceLog.IsDebugEnabled)
+ {
+ _ioTraceLog.Debug(String.Format("WRITE {0}", buffer));
+ }
+
+ processor.Write(buffer);
+ }
+ }
+}
\ No newline at end of file
Modified: incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Qpid.Client.csproj
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Qpid.Client.csproj?view=diff&rev=493064&r1=493063&r2=493064
==============================================================================
--- incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Qpid.Client.csproj (original)
+++ incubator/qpid/trunk/qpid/dotnet/Qpid.Client/Qpid.Client.csproj Fri Jan 5 08:16:48 2007
@@ -97,6 +97,9 @@
<Compile Include="Client\Transport\IProtocolWriter.cs" />
<Compile Include="Client\Transport\ITransport.cs" />
<Compile Include="Client\Transport\SingleProtocolEncoderOutput.cs" />
+ <Compile Include="Client\Transport\Socket\Blocking\BlockingSocketProcessor.cs" />
+ <Compile Include="Client\Transport\Socket\Blocking\BlockingSocketTransport.cs" />
+ <Compile Include="Client\Transport\Socket\Blocking\ByteChannel.cs" />
<Compile Include="Properties\AssemblyInfo.cs" />
<Compile Include="qms\BrokerInfo.cs" />
<Compile Include="qms\ConnectionInfo.cs" />
Modified: incubator/qpid/trunk/qpid/dotnet/Qpid.NET.sln
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/dotnet/Qpid.NET.sln?view=diff&rev=493064&r1=493063&r2=493064
==============================================================================
--- incubator/qpid/trunk/qpid/dotnet/Qpid.NET.sln (original)
+++ incubator/qpid/trunk/qpid/dotnet/Qpid.NET.sln Fri Jan 5 08:16:48 2007
@@ -6,8 +6,6 @@
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Qpid.Buffer", "Qpid.Buffer\Qpid.Buffer.csproj", "{44384DF2-B0A4-4580-BDBC-EE4BAA87D995}"
EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Qpid.Client.Transport.Socket.Blocking", "Qpid.Client.Transport.Socket.Blocking\Qpid.Client.Transport.Socket.Blocking.csproj", "{52AC4940-2077-4104-A753-29A9C8C16957}"
-EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Qpid.Codec", "Qpid.Codec\Qpid.Codec.csproj", "{22D0D0C2-77AF-4DE3-B456-7FF3893F9F88}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Qpid.Client", "Qpid.Client\Qpid.Client.csproj", "{68987C05-3768-452C-A6FC-6BA1D372852F}"
@@ -34,10 +32,6 @@
{44384DF2-B0A4-4580-BDBC-EE4BAA87D995}.Debug|Any CPU.Build.0 = Debug|Any CPU
{44384DF2-B0A4-4580-BDBC-EE4BAA87D995}.Release|Any CPU.ActiveCfg = Release|Any CPU
{44384DF2-B0A4-4580-BDBC-EE4BAA87D995}.Release|Any CPU.Build.0 = Release|Any CPU
- {52AC4940-2077-4104-A753-29A9C8C16957}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
- {52AC4940-2077-4104-A753-29A9C8C16957}.Debug|Any CPU.Build.0 = Debug|Any CPU
- {52AC4940-2077-4104-A753-29A9C8C16957}.Release|Any CPU.ActiveCfg = Release|Any CPU
- {52AC4940-2077-4104-A753-29A9C8C16957}.Release|Any CPU.Build.0 = Release|Any CPU
{22D0D0C2-77AF-4DE3-B456-7FF3893F9F88}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{22D0D0C2-77AF-4DE3-B456-7FF3893F9F88}.Debug|Any CPU.Build.0 = Debug|Any CPU
{22D0D0C2-77AF-4DE3-B456-7FF3893F9F88}.Release|Any CPU.ActiveCfg = Release|Any CPU