QPID-128 Initial import of the C# sources.

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

1 files changed, 215 insertions, 0 deletions

diff --git a/dotnet/Qpid.Client.Tests/requestreply1/ServiceRequestingClient.cs b/dotnet/Qpid.Client.Tests/requestreply1/ServiceRequestingClient.cs
new file mode 100644
index 0000000000..582f022719
--- /dev/null
+++ b/dotnet/Qpid.Client.Tests/requestreply1/ServiceRequestingClient.cs
@@ -0,0 +1,215 @@
+/*
+ *
+ * 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 log4net;
+using NUnit.Framework;
+using Qpid.Messaging;
+
+namespace Qpid.Client.Tests
+{
+    [TestFixture]
+    public class ServiceRequestingClient : BaseMessagingTestFixture
+    {
+        private const int MESSAGE_SIZE = 1024;
+        private static string MESSAGE_DATA = new string('x', MESSAGE_SIZE);
+
+        private const int NUM_MESSAGES = 10000;
+
+        private static ILog _log = LogManager.GetLogger(typeof(ServiceRequestingClient));
+
+        AutoResetEvent _finishedEvent = new AutoResetEvent(false);
+
+        private int _expectedMessageCount = NUM_MESSAGES;
+
+        private long _startTime;        
+        
+        private string _commandQueueName = "ServiceQ1";
+        
+        private IMessagePublisher _publisher;
+
+        Avergager averager = new Avergager();
+        
+        private void InitialiseProducer()
+        {
+            try
+            {
+                _publisher = _channel.CreatePublisherBuilder()
+                    .withRoutingKey(_commandQueueName)
+                    .Create();
+                _publisher.DisableMessageTimestamp = true; // XXX: need a "with" for this in builder?
+                _publisher.DeliveryMode = DeliveryMode.NonPersistent;  // XXX: need a "with" for this in builder?
+            }
+            catch (QpidException e)
+            {
+                _log.Error("Error: " + e, e);
+            }
+        }
+        
+        [Test]
+        public void SendMessages()
+        {
+            InitialiseProducer();
+
+            string replyQueueName = _channel.GenerateUniqueName();
+            
+            _channel.DeclareQueue(replyQueueName, false, true, true);
+
+            IMessageConsumer messageConsumer = _channel.CreateConsumerBuilder(replyQueueName)
+                .withPrefetch(100)
+                .withNoLocal(true)
+                .withExclusive(true).Create();
+ 
+            _startTime = DateTime.Now.Ticks;
+
+            messageConsumer.OnMessage = new MessageReceivedDelegate(OnMessage);
+            _connection.Start();
+            for (int i = 0; i < _expectedMessageCount; i++)
+            {
+                ITextMessage msg;
+                try
+                {
+                    msg = _channel.CreateTextMessage(MESSAGE_DATA + i);
+                }
+                catch (Exception e)
+                {
+                    _log.Error("Error creating message: " + e, e);
+                    break;
+                }
+                msg.ReplyToRoutingKey = replyQueueName;
+
+                // Added timestamp.
+                long timeNow = DateTime.Now.Ticks;
+                string timeSentString = String.Format("{0:G}", timeNow);
+//                _log.Info(String.Format("timeSent={0} timeSentString={1}", timeNow, timeSentString));
+                msg.Headers.SetString("timeSent", timeSentString);
+                //msg.Headers.SetLong("sentAt", timeNow);
+                
+                try
+                {
+                    _publisher.Send(msg);
+                }
+                catch (Exception e)
+                {
+                    _log.Error("Error sending message: " + e, e);
+                    //base._port = 5673;
+                    _log.Info("Reconnecting but on port 5673");
+                    try
+                    {
+                        base.Init();
+                        InitialiseProducer();
+                        // cheesy but a quick test
+                        _log.Info("Calling SendMessages again");
+                        SendMessages();
+                    }
+                    catch (Exception ex)
+                    {
+                        _log.Error("Totally busted: failed to reconnect: " + ex, ex);
+                    }
+                }
+            }
+
+            // Assert that the test finishes within a reasonable amount of time.
+            const int waitSeconds = 10;
+            const int waitMilliseconds = waitSeconds * 1000;
+            _log.Info("Finished sending " + _expectedMessageCount + " messages");
+            _log.Info(String.Format("Waiting {0} seconds to receive last message...", waitSeconds));
+            Assert.IsTrue(_finishedEvent.WaitOne(waitMilliseconds, false), 
+                          String.Format("Expected to finish in {0} seconds", waitSeconds));
+        }
+
+        public void OnMessage(IMessage m)
+        {
+            if (_log.IsDebugEnabled)
+            {
+                _log.Debug("Message received: " + m);
+            }
+
+            //if (m.Headers.Contains("sentAt"))
+            if (!m.Headers.Contains("timeSent"))
+            {
+                throw new Exception("Set timeSent!");
+            }
+            //long sentAt = m.Headers.GetLong("sentAt");
+            long sentAt = Int64.Parse(m.Headers.GetString("timeSent"));
+            long now = DateTime.Now.Ticks;
+            long latencyTicks = now - sentAt;
+//                _log.Info(String.Format("latency = {0} ticks ", latencyTicks));
+            long latencyMilliseconds = latencyTicks / TimeSpan.TicksPerMillisecond;
+//                _log.Info(String.Format("latency = {0} ms", latencyMilliseconds));
+
+            averager.Add(latencyMilliseconds);
+
+            // Output average every 1000 messages.
+            if (averager.Num % 1000 == 0)
+            {
+                _log.Info("Ticks per millisecond = " + TimeSpan.TicksPerMillisecond);
+                _log.Info(String.Format("Average latency (ms) = {0}", averager));
+                _log.Info("Received message count: " + averager.Num);
+            }
+
+            if (averager.Num == _expectedMessageCount)
+            {
+                _log.Info(String.Format("Final average latency (ms) = {0}", averager));
+
+                double timeTakenSeconds = (DateTime.Now.Ticks - _startTime) * 1.0 / (TimeSpan.TicksPerMillisecond * 1000);
+                _log.Info("Total time taken to receive " + _expectedMessageCount + " messages was " +
+                          timeTakenSeconds + "s, equivalent to " +
+                          (_expectedMessageCount/timeTakenSeconds) + " messages per second");
+
+                _finishedEvent.Set(); // Notify main thread to quit.
+            }
+        }
+        
+        public static void Main(String[] args)
+        {
+            ServiceRequestingClient c = new ServiceRequestingClient();
+            c.Init();
+            c.SendMessages();
+        }
+    }
+    
+    class Avergager
+    {
+        long num = 0;
+        long sum = 0;
+
+        long min = long.MaxValue;
+        long max = long.MinValue;
+        
+        public void Add(long item)
+        {
+            ++num;
+            sum += item;
+            if (item < min) min = item;
+            if (item > max) max = item;
+        }
+        
+        public long Average { get { return sum/num; }}
+
+        public long Num { get { return num;  } }
+        
+        public override string ToString()
+        {
+            return String.Format("average={0} min={1} max={2}", Average, min, max);
+        }
+    }
+}