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

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

#2615:
made minor changes

File size: 4.9 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
31    public void Init(string lanIpPrefix, string contactServerUrl, string problemInstance, int messageCacheCapacity, double communicationRate) {
32      try {
33        ownInstance = new PeerInfo() {
34          IpAddress = GetInternalIpAddress(lanIpPrefix),
35          Port = 0,
36          ProblemInstance = problemInstance
37        }; // TODO: get own peerinfo
38
39        writeQueue = new SizedConcurrentQueue<byte[]>();
40        readQueue = new SizedConcurrentQueue<byte[]>();
41        writeQueue.Limit = messageCacheCapacity;
42        readQueue.Limit = writeQueue.Limit;
43
44        host = new WcfMessageService();
45        ownInstance.Port = host.Init(ownInstance.IpAddress); //getting port, on which service is hostet
46        host.OnDataRecieved += new EventHandler<MessageRecieveEventArgs>(OnDataRecieved);
47
48        peerListManager = new WcfPeerListManager();
49        peerListManager.Init(ownInstance, contactServerUrl, communicationRate);
50
51        sender = new WcfMessageSender();
52        sender.Init(ownInstance, messageCacheCapacity);
53
54      }
55      catch (Exception ex) {
56        AddError("PeerNetworkMessageHandler.Init", ex);
57      }
58    }
59
60    public void Dispose() {
61      try {
62        host.Dispose();
63        sender.Dispose();
64        peerListManager.Dispose();
65      }
66      catch (Exception ex) {
67        AddError("PeerNetworkMessageHandler.Dispose", ex);
68      }
69    }
70
71    public void PublishDataToNetwork(byte[][] data) {
72      try {
73        foreach (PeerInfo peer in peerListManager.GetPeerList()) {
74          //maybe something will go wrong during network communication...
75          try {
76            sender.SendData(peer, data);
77          }
78          catch (Exception ex) {
79            AddError("PeerNetworkMessageHandler.PublishDataToNetwork(during sending to one peer!)", ex);
80          }
81        }
82      }
83      catch (Exception ex) {
84        AddError("PeerNetworkMessageHandler.PublishDataToNetwork", ex);
85      }
86    }
87
88    public byte[][] GetDataFromNetwork() {
89      try {
90        List<byte[]> res = new List<byte[]>();
91        byte[] item = null;
92        lock (activeQueueLocker) {
93          //changing the current queue for storing items to send
94          //then read from the now unselect queue
95          var tmp = readQueue;
96          readQueue = writeQueue;
97          writeQueue = tmp;
98        }
99
100        //creating resultset
101        while (!readQueue.IsEmpty) {
102          if (readQueue.TryDequeue(out item)) {
103            res.Add(item);
104          }
105        }
106        return res.ToArray();
107      }
108      catch (Exception ex) {
109        AddError("PeerNetworkMessageHandler.GetDataFromNetwork", ex);
110        return null;
111      }
112    }
113
114    public PeerInfo GetPeerInfo() {
115      return ownInstance;
116    }
117
118    public List<PeerInfo> GetCurrentNetwork() {
119      return peerListManager.GetPeerList();
120    }
121
122    private string GetInternalIpAddress(string ipPrefix) {
123      try {
124        var strHostName = Dns.GetHostName();
125        // Then using host name, get the IP address list..
126        IPHostEntry ipEntry = Dns.GetHostEntry(strHostName);
127        IPAddress[] addr = ipEntry.AddressList;
128
129        return addr
130          .Select(ip => ip.ToString())
131          .First(str => str.StartsWith(ipPrefix));
132      }
133      catch { return null; }
134    }
135
136    private void OnDataRecieved(object sender, MessageRecieveEventArgs e) {
137      if (e != null && e.Sender.ProblemInstance.Equals(ownInstance.ProblemInstance)) {
138        lock (activeQueueLocker) {
139          foreach (byte[] item in e.data) {
140            writeQueue.Enqueue(item);
141          }
142        }
143     
144      }
145    }
146
147    private void AddError(string source, Exception ex) {
148      if (peerListManager != null) {
149        try {
150          peerListManager.SendLogToServer(string.Concat("Source: ", source, ", Exception: ", ex.Message));
151        }
152        catch { }
153      }
154    }
155
156  }
157}
Note: See TracBrowser for help on using the repository browser.