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

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;
using DistributedGA.Core.Util;

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

    private PeerInfo myself;

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


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

    private Object timerLock = new Object();

    public void Init(PeerInfo source, int messageCacheCapacity) {
      myself = source;
      bufferedMessagesRead = new SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>>();
      bufferedMessagesRead.Limit = messageCacheCapacity;
      bufferedMessagesWrite = new SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>>();
      bufferedMessagesWrite.Limit = bufferedMessagesRead.Limit;
      timer = new Timer(1000 * 60); //each 5 minutes
      timer.Elapsed += GenerateSendingTasks;
      timer.Start();
    }

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

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

    private void GenerateSendingTasks(object sender, ElapsedEventArgs e) {
      lock (timerLock) {
        //changing the queues...
        var tmp = bufferedMessagesRead;
        bufferedMessagesRead = bufferedMessagesWrite;
        bufferedMessagesWrite = tmp;
        List<KeyValuePair<PeerInfo, byte[][]>> messages = new List<KeyValuePair<PeerInfo, byte[][]>>();
        while (!bufferedMessagesRead.IsEmpty) {
          KeyValuePair<PeerInfo, byte[][]> 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, byte[][]>> mergedMessages = MergeMessages(messages);
        foreach (var item in mergedMessages) {
          Task.Factory.StartNew(() => SendDataFromQueue(item.Key, item.Value), TaskCreationOptions.LongRunning);
        }
      }
    }

    /// <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, byte[][]>> MergeMessages(List<KeyValuePair<PeerInfo, byte[][]>> messages) {
      List<KeyValuePair<PeerInfo, byte[][]>> res = new List<KeyValuePair<PeerInfo, byte[][]>>();
      Dictionary<PeerInfo, List<byte[]>> cache = new Dictionary<PeerInfo, List<byte[]>>();
      foreach (var messagePackage in messages) {
        if (!cache.ContainsKey(messagePackage.Key)) {
          cache.Add(messagePackage.Key, new List<byte[]>());
        }
        for (int i = 0; i <= messagePackage.Value.GetUpperBound(0); i++) {
          cache[messagePackage.Key].Add(messagePackage.Value[i]);
        }
      }
      //now we have a dictionary with all messages per destionation
      //so: create a byte[][] again and return
      foreach (var dest in cache.Keys) {
        byte[][] messagesPerDest = new byte[cache[dest].Count][];
        for (int i = 0; i <= messagesPerDest.GetUpperBound(0); i++) {
          messagesPerDest[i] = cache[dest][i];
        }
        res.Add(new KeyValuePair<PeerInfo, byte[][]>(dest, messagesPerDest));
      }
      return res;
    }

    /// <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 {
        int arrayLength = 3;
        if (data.GetUpperBound(0) > arrayLength) {
          //HACK: SEND MAX 10 items
          byte[][] fake = new byte[arrayLength][];
          for (int i = 0; i < arrayLength; i++) {
            fake[i] = data[i];
          }
          data = fake;
        }


        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.ReaderQuotas.MaxArrayLength = 2147483647;
        binding.ReaderQuotas.MaxNameTableCharCount = 2147483647;
        binding.ReaderQuotas.MaxBytesPerRead = 2147483647;
        binding.ReaderQuotas.MaxStringContentLength = 2147483647;

        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 (Exception ex) {
        Console.WriteLine(ex.Message);
      }
    }

  }
}