source: branches/thasling/DistributedGA/DistributedGA.Core/Implementation/WcfMessageSender.cs @ 17169

using System;
using System.CodeDom;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.ServiceModel;
using System.Threading.Tasks;
using System.Timers;
using DistributedGA.Core.Domain;
using DistributedGA.Core.Interface;
using DistributedGA.Core.Util;

namespace DistributedGA.Core.Implementation {
  public class WcfMessageSender : IMessageSender, IDisposable {

    private PeerInfo myself;

    //providing two queues for faster access
    private SizedConcurrentQueue<KeyValuePair<PeerInfo, ByteArrayWrapper>> bufferedMessagesWrite;
    private SizedConcurrentQueue<KeyValuePair<PeerInfo, ByteArrayWrapper>> bufferedMessagesRead;

    private Timer timer; //sends cached messages to network in background

    private Object timerLock = new Object();
    private bool isActive = false;

    public void Init(PeerInfo source, int messageCacheCapacity) {
      myself = source;
      bufferedMessagesRead = new SizedConcurrentQueue<KeyValuePair<PeerInfo, ByteArrayWrapper>>();
      bufferedMessagesRead.Limit = messageCacheCapacity;
      bufferedMessagesWrite = new SizedConcurrentQueue<KeyValuePair<PeerInfo, ByteArrayWrapper>>();
      bufferedMessagesWrite.Limit = bufferedMessagesRead.Limit;
      timer = new Timer(1000 * 10); //each 10 seconds
      timer.Elapsed += GenerateSendingTasks;
      timer.Start();
    }

    public void SendData(PeerInfo destination, ByteArrayWrapper data) {
      bufferedMessagesWrite.Enqueue(new KeyValuePair<PeerInfo, ByteArrayWrapper>(destination, data));
    }

    public void Dispose() {
      timer.Stop();
      timer.Dispose();
      timer = null;
    }

    private void GenerateSendingTasks(object sender, ElapsedEventArgs e) {
      if (isActive) return;
      lock (timerLock) {
        try {
          isActive = true;
          //changing the queues...
          var tmp = bufferedMessagesRead;
          bufferedMessagesRead = bufferedMessagesWrite;
          bufferedMessagesWrite = tmp;
          List<KeyValuePair<PeerInfo, ByteArrayWrapper>> messages = new List<KeyValuePair<PeerInfo, ByteArrayWrapper>>();
          while (!bufferedMessagesRead.IsEmpty) {
            KeyValuePair<PeerInfo, ByteArrayWrapper> message;
            if (bufferedMessagesRead.TryDequeue(out message)) {
              messages.Add(message);
              //Task.Factory.StartNew(() => SendDataFromQueue(message.Key, message.Value), TaskCreationOptions.LongRunning);
            }
          }
          //now: merge them and start sending tasks

          List<KeyValuePair<PeerInfo, ByteArrayWrapper[]>> mergedMessages = MergeMessages(messages);

          var runningTasks = new List<Task>();
          foreach (var item in mergedMessages) {
            var itemToSend = item;
            var t = Task.Factory.StartNew(() => SendDataFromQueue(itemToSend.Key, itemToSend.Value),
              TaskCreationOptions.LongRunning);
            runningTasks.Add(t);
          }

          Task.WaitAll(runningTasks.ToArray());
        }
        finally {
          isActive = false;
        }
      }
    }

    /// <summary>
    /// Merges and resorts all Messages by their destination
    /// </summary>
    /// <param name="messages">a list with multiple messages to the same  destination</param>
    /// <returns></returns>
    private List<KeyValuePair<PeerInfo, ByteArrayWrapper[]>> MergeMessages(List<KeyValuePair<PeerInfo, ByteArrayWrapper>> messages) {
      Dictionary<PeerInfo, List<ByteArrayWrapper>> cache = new Dictionary<PeerInfo, List<ByteArrayWrapper>>();
      foreach (var messagePackage in messages) {
        if (!cache.ContainsKey(messagePackage.Key)) {
          cache.Add(messagePackage.Key, new List<ByteArrayWrapper>());
        }
        //for (int i = 0; i <= messagePackage.Value.GetUpperBound(0); i++) {
        cache[messagePackage.Key].Add(messagePackage.Value);
        //}
      }

      return cache.Keys.Select(dest => new KeyValuePair<PeerInfo, ByteArrayWrapper[]>(dest, cache[dest].ToArray())).ToList();
    }

    /// <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, ByteArrayWrapper[] 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();
        binding.MaxReceivedMessageSize = 20000000;
        binding.MaxBufferSize = 20000000;
        binding.MaxBufferPoolSize = 20000000;
        binding.ReaderQuotas.MaxArrayLength = 20000000;
        binding.ReaderQuotas.MaxDepth = 32;
        binding.ReaderQuotas.MaxStringContentLength = 20000000;

        var endpoint = new EndpointAddress(serviceUri);
        using (var myChannelFactory = new ChannelFactory<IMessageContract>(binding, endpoint)) {
          using (IClientChannel client = (IClientChannel)myChannelFactory.CreateChannel()) {
            for (int i = 0; i < data.GetLength(0); i++) {
              ((IMessageContract)client).SendData(myself, data[i].Array); //maybe exception...
            }
          }
        }
      }
      catch (Exception ex) {
        //ignore 
        Console.WriteLine(ex.Message);
      }
    }

  }
}