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

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;

    private SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>> bufferedMessages;

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

    private Object timerLock = new Object();

    public void Init(PeerInfo source, int messageCacheCapacity) {
      myself = source;
      bufferedMessages = new SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>>();
      bufferedMessages.Limit = messageCacheCapacity;
      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));
    }

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

    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);
        }
      }
    }

    /// <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) {
      lock (timerLock) {
        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
      }
    }

  }
}