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

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

#2615:
re-enabled log in P2PMigrationAnalyzer
made code formatings

File size: 5.3 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, IDisposable {
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<KeyValuePair<PeerInfo, ByteArrayWrapper>> writeQueue;
25    private SizedConcurrentQueue<KeyValuePair<PeerInfo, ByteArrayWrapper>> 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<KeyValuePair<PeerInfo, ByteArrayWrapper>>();
41        readQueue = new SizedConcurrentQueue<KeyValuePair<PeerInfo, ByteArrayWrapper>>();
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(ByteArrayWrapper data) {
73      try {
74        var allPeers = peerListManager.GetPeerList();
75        foreach (PeerInfo peer in allPeers) {
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      }
85      catch (Exception ex) {
86        AddError("PeerNetworkMessageHandler.PublishDataToNetwork", ex);
87      }
88    }
89
90    public List<KeyValuePair<PeerInfo, ByteArrayWrapper>> GetDataFromNetwork() {
91      try {
92        List<KeyValuePair<PeerInfo, ByteArrayWrapper>> res = new List<KeyValuePair<PeerInfo, ByteArrayWrapper>>();
93        KeyValuePair<PeerInfo, ByteArrayWrapper> item;
94        lock (activeQueueLocker) {
95          //changing the current queue for storing items to send
96          //then read from the now unselect queue
97          var tmp = readQueue;
98          readQueue = writeQueue;
99          writeQueue = tmp;
100        }
101
102        //creating resultset
103        while (!readQueue.IsEmpty) {
104          if (readQueue.TryDequeue(out item)) {
105            res.Add(item);
106          }
107        }
108        return res;//.ToArray();
109      }
110      catch (Exception ex) {
111        AddError("PeerNetworkMessageHandler.GetDataFromNetwork", ex);
112        return null;
113      }
114    }
115
116    public PeerInfo GetPeerInfo() {
117      return ownInstance;
118    }
119
120    public List<PeerInfo> GetCurrentNetwork() {
121      return peerListManager.GetPeerList();
122    }
123
124    private void PropagateException(Exception ex) {
125      if (ExceptionOccurend != null) {
126        ExceptionOccurend(this, ex);
127      }
128    }
129
130    private string GetInternalIpAddress(string ipPrefix) {
131      try {
132        var strHostName = Dns.GetHostName();
133        // Then using host name, get the IP address list..
134        IPHostEntry ipEntry = Dns.GetHostEntry(strHostName);
135        IPAddress[] addr = ipEntry.AddressList;
136
137        return addr
138          .Select(ip => ip.ToString())
139          .First(str => str.StartsWith(ipPrefix));
140      }
141      catch { return null; }
142    }
143
144    private void OnDataRecieved(object sender, MessageRecieveEventArgs e) {
145      if (e != null && e.Sender.ProblemInstance.Equals(ownInstance.ProblemInstance)) {
146        lock (activeQueueLocker) {
147          writeQueue.Enqueue(new KeyValuePair<PeerInfo, ByteArrayWrapper>(e.Sender, ByteArrayWrapper.CreateByteArrayWrapper(e.data)));
148        }
149
150      }
151    }
152
153    private void AddError(string source, Exception ex) {
154      PropagateException(ex);
155    }
156
157  }
158}
Note: See TracBrowser for help on using the repository browser.