Patch for Qpid-239 applied. BlockingSocketTransport instantiated directly.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@493064 13f79535-47bb-0310-9956-ffa450edef68

7 files changed, 309 insertions, 12 deletions

diff --git a/dotnet/Qpid.Client.Tests/Qpid.Client.Tests.csproj b/dotnet/Qpid.Client.Tests/Qpid.Client.Tests.csproj
index 00c3d54a82..a7b6e49879 100644
--- a/dotnet/Qpid.Client.Tests/Qpid.Client.Tests.csproj
+++ b/dotnet/Qpid.Client.Tests/Qpid.Client.Tests.csproj
@@ -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>

diff --git a/dotnet/Qpid.Client/Client/AMQConnection.cs b/dotnet/Qpid.Client/Client/AMQConnection.cs
index 64a30c1981..c620412cab 100644
--- a/dotnet/Qpid.Client/Client/AMQConnection.cs
+++ b/dotnet/Qpid.Client/Client/AMQConnection.cs
@@ -29,6 +29,7 @@ using Qpid.Client.Protocol;
 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 @@ namespace Qpid.Client
             }
         }
 
-        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 @@ namespace Qpid.Client
             _log.Info("transport = " + result);
 
             return result;
-        }
+        }*/
 
         public void Disconnect()
         {
@@ -688,12 +689,16 @@ namespace Qpid.Client
                 _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();                
diff --git a/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketProcessor.cs b/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketProcessor.cs
new file mode 100644
index 0000000000..bdec584b7b
--- /dev/null
+++ b/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketProcessor.cs
@@ -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; }

+        }

+    }

+}

+

diff --git a/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketTransport.cs b/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketTransport.cs
new file mode 100644
index 0000000000..e18eefd493
--- /dev/null
+++ b/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/BlockingSocketTransport.cs
@@ -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;

+            }

+        }

+    }

+}

+

diff --git a/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/ByteChannel.cs b/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/ByteChannel.cs
new file mode 100644
index 0000000000..a520815f84
--- /dev/null
+++ b/dotnet/Qpid.Client/Client/Transport/Socket/Blocking/ByteChannel.cs
@@ -0,0 +1,63 @@
+/*

+ *

+ * Licensed to the Apache Software Foundation (ASF) under one

+ * or more contributor license agreements.  See the NOTICE file

+ * distributed with this work for additional information

+ * regarding copyright ownership.  The ASF licenses this file

+ * to you under the Apache License, Version 2.0 (the

+ * "License"); you may not use this file except in compliance

+ * with the License.  You may obtain a copy of the License at

+ * 

+ *   http://www.apache.org/licenses/LICENSE-2.0

+ * 

+ * Unless required by applicable law or agreed to in writing,

+ * software distributed under the License is distributed on an

+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

+ * KIND, either express or implied.  See the License for the

+ * specific language governing permissions and limitations

+ * under the License.

+ *

+ */

+using 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
diff --git a/dotnet/Qpid.Client/Qpid.Client.csproj b/dotnet/Qpid.Client/Qpid.Client.csproj
index 84ada2a94b..83609b0757 100644
--- a/dotnet/Qpid.Client/Qpid.Client.csproj
+++ b/dotnet/Qpid.Client/Qpid.Client.csproj
@@ -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" />

diff --git a/dotnet/Qpid.NET.sln b/dotnet/Qpid.NET.sln
index a58028c413..c4b518e51c 100644
--- a/dotnet/Qpid.NET.sln
+++ b/dotnet/Qpid.NET.sln
@@ -6,8 +6,6 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Qpid.Common.Tests", "Qpid.C
 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 @@ Global
 		{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