source: branches/thasling/DistributedGA/DistributedGA.Core/Implementation/PeerNetworkMessageHandler.cs @ 13982

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using DistributedGA.Core.Domain;
using DistributedGA.Core.Interface;
using DistributedGA.Core.Util;

namespace DistributedGA.Core.Implementation {
  public class PeerNetworkMessageHandler : IMessageHandler {
    //own peer-instance Information
    private PeerInfo ownInstance = null;

    //uses peer-list from IPeerListManager to decide which peers to contact
    private IPeerListManager peerListManager;

    //uses IMessageSender to send populations to peers
    private IMessageSender sender = null;

    //provides current population for the higher layer IMigrationOperator
    //two queues are used to gather and and provide population more efficiently
    private object activeQueueLocker = new Object();
    private SizedConcurrentQueue<KeyValuePair<PeerInfo,byte[]>> writeQueue;
    private SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[]>> readQueue;

    //uses IMessageService for recieving population from one peer at once
    private IMessageService host = null;

    private double communicationRate;
    private Random rand;

    public event EventHandler<Exception> ExceptionOccurend;

    public void Init(string lanIpPrefix, string contactServerUrl, string problemInstance, int messageCacheCapacity, double communicationRate) {
      try {
        ownInstance = new PeerInfo() {
          IpAddress = GetInternalIpAddress(lanIpPrefix),
          Port = 0,
          ProblemInstance = problemInstance
        }; // TODO: get own peerinfo

        this.communicationRate = communicationRate;
        rand = new Random();

        writeQueue = new SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[]>>();
        readQueue = new SizedConcurrentQueue<KeyValuePair<PeerInfo,byte[]>>();
        writeQueue.Limit = messageCacheCapacity;
        readQueue.Limit = writeQueue.Limit;

        host = new WcfMessageService();
        ownInstance.Port = host.Init(ownInstance.IpAddress); //getting port, on which service is hostet
        host.OnDataRecieved += new EventHandler<MessageRecieveEventArgs>(OnDataRecieved);

        peerListManager = new WcfPeerListManager();
        peerListManager.Init(ownInstance, contactServerUrl);

        sender = new WcfMessageSender();
        sender.Init(ownInstance, messageCacheCapacity);

      }
      catch (Exception ex) {
        AddError("PeerNetworkMessageHandler.Init", ex);
      }
    }

    public void Dispose() {
      try {
        host.Dispose();
        sender.Dispose();
        peerListManager.Dispose();
      }
      catch (Exception ex) {
        AddError("PeerNetworkMessageHandler.Dispose", ex);
      }
    }

    public void PublishDataToNetwork(byte[] data) {
      try {
        var peers = peerListManager.GetPeerList();
        foreach (PeerInfo peer in peerListManager.GetPeerList()) {
          var peersForMessaging = ChoosePeersForMessaging(ref peers);

          //maybe something will go wrong during network communication...
          try {
            sender.SendData(peer, data);
          }
          catch (Exception ex) {
            AddError("PeerNetworkMessageHandler.PublishDataToNetwork(during sending to one peer!)", ex);
          }
        }
      }
      catch (Exception ex) {
        AddError("PeerNetworkMessageHandler.PublishDataToNetwork", ex);
      }
    }

    public List<KeyValuePair<PeerInfo, byte[]>> GetDataFromNetwork() {
      try {
        List<KeyValuePair<PeerInfo, byte[]>> res = new List<KeyValuePair<PeerInfo, byte[]>>();
        KeyValuePair<PeerInfo,byte[]> item;
        lock (activeQueueLocker) {
          //changing the current queue for storing items to send
          //then read from the now unselect queue
          var tmp = readQueue;
          readQueue = writeQueue;
          writeQueue = tmp;
        }

        //creating resultset
        while (!readQueue.IsEmpty) {
          if (readQueue.TryDequeue(out item)) {
            res.Add(item);
          }
        }
        return res;//.ToArray();
      }
      catch (Exception ex) {
        AddError("PeerNetworkMessageHandler.GetDataFromNetwork", ex);
        return null;
      }
    }

    public PeerInfo GetPeerInfo() {
      return ownInstance;
    }

    public List<PeerInfo> GetCurrentNetwork() {
      return peerListManager.GetPeerList();
    }

    private void PropagateException(Exception ex) {
      //if (CountdownCompleted != null)
      //  CountdownCompleted(this, e);
      if (ExceptionOccurend != null) {
        ExceptionOccurend(this, ex);
      }
    }

    private List<PeerInfo> ChoosePeersForMessaging(ref List<PeerInfo> allPeers) {
      Shuffle<PeerInfo>(allPeers);
      int toTake = Convert.ToInt32(allPeers.Count * communicationRate) + 1;
      if (allPeers.Count > 0 && toTake == 0) {
        toTake = 1;
      }
      return allPeers.Take(toTake).ToList(); ;
    }

    private string GetInternalIpAddress(string ipPrefix) {
      try {
        var strHostName = Dns.GetHostName();
        // Then using host name, get the IP address list..
        IPHostEntry ipEntry = Dns.GetHostEntry(strHostName);
        IPAddress[] addr = ipEntry.AddressList;

        return addr
          .Select(ip => ip.ToString())
          .First(str => str.StartsWith(ipPrefix));
      }
      catch { return null; }
    }

    private void Shuffle<T>(IList<T> list) {
      int n = list.Count;
      while (n > 1) {
        n--;
        int k = rand.Next(n + 1);
        T value = list[k];
        list[k] = list[n];
        list[n] = value;
      }
    }

    private void OnDataRecieved(object sender, MessageRecieveEventArgs e) {
      if (e != null && e.Sender.ProblemInstance.Equals(ownInstance.ProblemInstance)) {
        lock (activeQueueLocker) {
            writeQueue.Enqueue(new KeyValuePair<PeerInfo, byte[]>(e.Sender, e.data));
        }

      }
    }

    private void AddError(string source, Exception ex) {
      PropagateException(ex);
      if (peerListManager != null) {
        try {
          peerListManager.SendLogToServer(string.Concat("Source: ", source, ", Exception: ", ex.Message));
        }
        catch { }
      }
    }

  }
}