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

Last change on this file since 13971 was 13971, checked in by thasling, 5 years ago

#2615:
increased message size to 20MB
made changes in some interfaces

File size: 5.2 KB
Line 
1using System;
2using System.Collections.Concurrent;
3using System.Collections.Generic;
4using System.Linq;
5using System.Net;
6using DistributedGA.Core.Domain;
7using DistributedGA.Core.Interface;
8using DistributedGA.Core.Util;
9
10namespace DistributedGA.Core.Implementation {
11  public class PeerNetworkMessageHandler : IMessageHandler {
12    //own peer-instance Information
13    private PeerInfo ownInstance = null;
14
15    //uses peer-list from IPeerListManager to decide which peers to contact
16    private IPeerListManager peerListManager;
17
18    //uses IMessageSender to send populations to peers
19    private IMessageSender sender = null;
20
21    //provides current population for the higher layer IMigrationOperator
22    //two queues are used to gather and and provide population more efficiently
23    private object activeQueueLocker = new Object();
24    private SizedConcurrentQueue<byte[]> writeQueue;
25    private SizedConcurrentQueue<byte[]> readQueue;
26
27    //uses IMessageService for recieving population from one peer at once
28    private IMessageService host = null;
29
30    public event EventHandler<Exception> ExceptionOccurend;
31
32    public void Init(string lanIpPrefix, string contactServerUrl, string problemInstance, int messageCacheCapacity, double communicationRate) {
33      try {
34        ownInstance = new PeerInfo() {
35          IpAddress = GetInternalIpAddress(lanIpPrefix),
36          Port = 0,
37          ProblemInstance = problemInstance
38        }; // TODO: get own peerinfo
39
40        writeQueue = new SizedConcurrentQueue<byte[]>();
41        readQueue = new SizedConcurrentQueue<byte[]>();
42        writeQueue.Limit = messageCacheCapacity;
43        readQueue.Limit = writeQueue.Limit;
44
45        host = new WcfMessageService();
46        ownInstance.Port = host.Init(ownInstance.IpAddress); //getting port, on which service is hostet
47        host.OnDataRecieved += new EventHandler<MessageRecieveEventArgs>(OnDataRecieved);
48
49        peerListManager = new WcfPeerListManager();
50        peerListManager.Init(ownInstance, contactServerUrl, communicationRate);
51
52        sender = new WcfMessageSender();
53        sender.Init(ownInstance, messageCacheCapacity);
54
55      }
56      catch (Exception ex) {
57        AddError("PeerNetworkMessageHandler.Init", ex);
58      }
59    }
60
61    public void Dispose() {
62      try {
63        host.Dispose();
64        sender.Dispose();
65        peerListManager.Dispose();
66      }
67      catch (Exception ex) {
68        AddError("PeerNetworkMessageHandler.Dispose", ex);
69      }
70    }
71
72    public void PublishDataToNetwork(byte[][] data) {
73      try {
74        foreach (PeerInfo peer in peerListManager.GetPeerList()) {
75          //maybe something will go wrong during network communication...
76          try {
77            sender.SendData(peer, data);
78          }
79          catch (Exception ex) {
80            AddError("PeerNetworkMessageHandler.PublishDataToNetwork(during sending to one peer!)", ex);
81          }
82        }
83      }
84      catch (Exception ex) {
85        AddError("PeerNetworkMessageHandler.PublishDataToNetwork", ex);
86      }
87    }
88
89    public byte[][] GetDataFromNetwork() {
90      try {
91        List<byte[]> res = new List<byte[]>();
92        byte[] item = null;
93        lock (activeQueueLocker) {
94          //changing the current queue for storing items to send
95          //then read from the now unselect queue
96          var tmp = readQueue;
97          readQueue = writeQueue;
98          writeQueue = tmp;
99        }
100
101        //creating resultset
102        while (!readQueue.IsEmpty) {
103          if (readQueue.TryDequeue(out item)) {
104            res.Add(item);
105          }
106        }
107        return res.ToArray();
108      }
109      catch (Exception ex) {
110        AddError("PeerNetworkMessageHandler.GetDataFromNetwork", ex);
111        return null;
112      }
113    }
114
115    public PeerInfo GetPeerInfo() {
116      return ownInstance;
117    }
118
119    public List<PeerInfo> GetCurrentNetwork() {
120      return peerListManager.GetPeerList();
121    }
122
123    private void PropagateException(Exception ex) {
124      //if (CountdownCompleted != null)
125      //  CountdownCompleted(this, e);
126      if (ExceptionOccurend != null) {
127        ExceptionOccurend(this, ex);
128      }
129    }
130
131    private string GetInternalIpAddress(string ipPrefix) {
132      try {
133        var strHostName = Dns.GetHostName();
134        // Then using host name, get the IP address list..
135        IPHostEntry ipEntry = Dns.GetHostEntry(strHostName);
136        IPAddress[] addr = ipEntry.AddressList;
137
138        return addr
139          .Select(ip => ip.ToString())
140          .First(str => str.StartsWith(ipPrefix));
141      }
142      catch { return null; }
143    }
144
145    private void OnDataRecieved(object sender, MessageRecieveEventArgs e) {
146      if (e != null && e.Sender.ProblemInstance.Equals(ownInstance.ProblemInstance)) {
147        lock (activeQueueLocker) {
148            writeQueue.Enqueue(e.data);
149        }
150
151      }
152    }
153
154    private void AddError(string source, Exception ex) {
155      PropagateException(ex);
156      if (peerListManager != null) {
157        try {
158          peerListManager.SendLogToServer(string.Concat("Source: ", source, ", Exception: ", ex.Message));
159        }
160        catch { }
161      }
162    }
163
164  }
165}
Note: See TracBrowser for help on using the repository browser.