source: branches/thasling/DistributedGA/DistributedGA.ContactServer/ContactServiceImpl.cs @ 13808

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.ServiceModel;
using System.Timers;
using DistributedGA.Core.Domain;

namespace DistributedGA.ContactServer {

  [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single)]
  public class ContactServiceImpl : IContactService {

    private ConcurrentDictionary<PeerInfo, DateTime> allPeers = null;

    private Timer timer = null;

    private Object logLock = new Object();

    public ContactServiceImpl() {
      allPeers = new ConcurrentDictionary<PeerInfo, DateTime>();

      timer = new Timer(60 * 1000); //each hour
      timer.Elapsed += CleanUpContactTable;
      timer.Start();
    }

    public void RegisterPeer(PeerInfo source) {
      try {
        UpdateHeartbeat(source);
      } catch (Exception ex) {
        AddError("ContactServiceImpl.RegisterPeer", ex);
      }
    }

    public List<PeerInfo> GetPeerList(PeerInfo source) {
      try {
        UpdateHeartbeat(source);
        //only return peers of the same work group and not the sender itself
        return allPeers.Keys.Where(x => {
          if (source.ProblemInstance.Equals(x.ProblemInstance) &&
              (!(x.IpAddress.Equals(source.IpAddress) && (x.Port.Equals(source.Port)))))
            return true;
          else
            return false;
        }).ToList();
      } catch (Exception ex) {
        AddError("ContactServiceImpl.GetPeerList", ex);
        return null;
      }
    }

    public void MakeLog(PeerInfo source, string msg) {
      try {
        // TODO 
        lock (logLock) {
          File.AppendAllText("Log.txt", string.Concat(source.IpAddress, ":", source.Port, ",", source.ProblemInstance, ",", msg, Environment.NewLine));
        }
      } catch (Exception ex) {
        //Nothing to do because maybe called from adderror
      }
    }

    private void UpdateHeartbeat(PeerInfo source) {
      Console.WriteLine("hb from {0}:{1}", source.IpAddress, source.Port);
      DateTime now = DateTime.Now;
      allPeers.AddOrUpdate(source, now, (k, v) => v = now);
    }

    private void CleanUpContactTable(object sender, ElapsedEventArgs e) {
      DateTime deadline = DateTime.Now;
      //collect items to remove
      List<PeerInfo> itemsToDelete = new List<PeerInfo>();
      foreach (PeerInfo pi in allPeers.Keys) {
        DateTime tmp;
        if (allPeers.TryGetValue(pi, out tmp)) {
          if (tmp.AddHours(1f) < deadline) {
            //if (tmp < deadline.AddHours(1f)) {
            itemsToDelete.Add(pi);
          }
        }
      }
      //remove items
      foreach (PeerInfo pi in itemsToDelete) {
        DateTime tmp;
        allPeers.TryRemove(pi, out tmp);
      }
    }

    private void AddError(string source, Exception ex) {
      MakeLog(new PeerInfo() { ProblemInstance = "ContactServer  Error at " + source }, ex.Message);
    }

  }
}