Changeset 13923 for branches/thasling/DistributedGA/DistributedGA.Core/Implementation/WcfMessageSender.cs

Timestamp:

06/19/16 21:37:24 (8 years ago)

Author:

thasling

Message:

#2615:
re-implemented WcfPeerListManager
also sending messages is now done in the background async

File:

• branches/thasling/DistributedGA/DistributedGA.Core/Implementation/WcfMessageSender.cs (modified) (2 diffs)

branches/thasling/DistributedGA/DistributedGA.Core/Implementation/WcfMessageSender.cs

@@ -1 +1 @@
 using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.ServiceModel;
+using System.Threading.Tasks;
+using System.Timers;
 using DistributedGA.Core.Domain;
 using DistributedGA.Core.Interface;
@@ -6 +10 @@
 namespace DistributedGA.Core.Implementation {
   public class WcfMessageSender : IMessageSender {
+
     private PeerInfo myself;
+
+    private ConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>> bufferedMessages;
+
+    private Timer timer; //sends cached messages to network in background
 
     public void Init(PeerInfo source) {
       myself = source;
+      bufferedMessages = new ConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>>();
+      timer = new Timer(1000 * 60); //each 5 minutes
+      timer.Elapsed += GenerateSendingTasks;
+      timer.Start();
     }
 
     public void SendData(PeerInfo destination, byte[][] data) {
+      bufferedMessages.Enqueue(new KeyValuePair<PeerInfo, byte[][]>(destination, data));
+    }
 
-      var serviceUrl = "DistributedGA.svc";
-      var baseUri = new Uri(string.Concat("net.tcp://", destination.IpAddress, ":", destination.Port, "/DistributedGA"));
-      var serviceUri = new Uri(baseUri, serviceUrl);
+    public void Dispose() {
+      timer.Stop();
+      timer.Dispose();
+      timer = null;
+    }
 
-      var binding = new NetTcpBinding();
-      var endpoint = new EndpointAddress(serviceUri);
-      using (var myChannelFactory = new ChannelFactory<IMessageContract>(binding, endpoint)) {
-        using (IClientChannel client = (IClientChannel)myChannelFactory.CreateChannel()) {
-          ((IMessageContract)client).SendData(myself, data); //maybe timout exception...
+    private void GenerateSendingTasks(object sender, ElapsedEventArgs e) {
+      while (!bufferedMessages.IsEmpty) {
+        KeyValuePair<PeerInfo, byte[][]> message;
+        if (bufferedMessages.TryDequeue(out message)) {
+          Task.Factory.StartNew(() => SendDataFromQueue(message.Key, message.Value), TaskCreationOptions.LongRunning);
         }
       }
     }
 
-    public void Dispose() {
+    /// <summary>
+    /// sends data to another peer, is used by a single thread to run in background
+    /// </summary>
+    /// <param name="destination">destination peer</param>
+    /// <param name="data">data to send</param>
+    private void SendDataFromQueue(PeerInfo destination, byte[][] data) {
+      try {
+        Console.WriteLine(string.Format("Sending data to {0}:{1} in the background...", destination.IpAddress, destination.Port));
+        var serviceUrl = "DistributedGA.svc";
+        var baseUri = new Uri(string.Concat("net.tcp://", destination.IpAddress, ":", destination.Port, "/DistributedGA"));
+        var serviceUri = new Uri(baseUri, serviceUrl);
 
+        var binding = new NetTcpBinding();
+        var endpoint = new EndpointAddress(serviceUri);
+        using (var myChannelFactory = new ChannelFactory<IMessageContract>(binding, endpoint)) {
+          using (IClientChannel client = (IClientChannel)myChannelFactory.CreateChannel()) {
+            ((IMessageContract)client).SendData(myself, data); //maybe timout exception...
+          }
+        }
+      }
+      catch { } //if maybe sending failed (because of connection lost, etc.): just ignore
     }
   }