PingManagerThread


1   /**
2    * Tribe: Group communication library.
3    * Copyright (C) 2004 French National Institute For Research In Computer
4    * Science And Control (INRIA).
5    * Contact: tribe@objectweb.org
6    * 
7    * This library is free software; you can redistribute it and/or modify it
8    * under the terms of the GNU Lesser General Public License as published by the
9    * Free Software Foundation; either version 2.1 of the License, or any later
10   * version.
11   * 
12   * This library is distributed in the hope that it will be useful, but WITHOUT
13   * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
14   * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License
15   * for more details.
16   * 
17   * You should have received a copy of the GNU Lesser General Public License
18   * along with this library; if not, write to the Free Software Foundation,
19   * Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
20   *
21   * Initial developer(s): Nicolas Modrzyk.
22   * Contributor(s): ______________________.
23   */
24  
25  package org.objectweb.tribe.faultdetection;
26  
27  import java.net.SocketException  ;
28  import java.util.ArrayList  ;
29  import java.util.Hashtable  ;
30  
31  import org.objectweb.tribe.common.Address;
32  import org.objectweb.tribe.common.IpAddress;
33  
34  /**
35   * This class defines a PingManagerThread. A ping thread will periodically send
36   * ping to addresses and send events to registered listeners.
37   * 
38   * @author <a HREF="mailto:Nicolas.Modrzyk@inrialpes.fr">Nicolas Modrzyk </a>
39   * @version 1.0
40   */
41  public class PingManagerThread implements Runnable  
42  {
43  
44    ArrayList   addresses;
45    ArrayList   listeners;
46    Hashtable   sockets;
47    long      waitInterval;
48    boolean   stop;
49    IpAddress localAddress;
50  
51    /**
52     * Creates a new <code>PingManagerThread</code> object
53     * 
54     * @param localAddress
55     * @param waitInterval the time between two ping
56     */
57    public PingManagerThread(IpAddress localAddress, long waitInterval)
58    {
59      super();
60      addresses = new ArrayList  ();
61      listeners = new ArrayList  ();
62      sockets = new Hashtable  ();
63      this.waitInterval = waitInterval;
64      this.localAddress = localAddress;
65      stop = false;
66    }
67  
68    /**
69     * Tell the thread to stop as soon as possible
70     * 
71     * @param stop The stop to set.
72     */
73    public void setStop(boolean stop)
74    {
75      this.stop = stop;
76    }
77  
78    /**
79     * Returns the waitInterval value.
80     * 
81     * @return Returns the waitInterval.
82     */
83    public long getWaitInterval()
84    {
85      return waitInterval;
86    }
87  
88    /**
89     * Sets the waitInterval value.
90     * 
91     * @param waitInterval The waitInterval to set.
92     */
93    public void setWaitInterval(long waitInterval)
94    {
95      this.waitInterval = waitInterval;
96    }
97  
98    /**
99     * Tell the thread it shoudl start pinging a new address
100    * 
101    * @param address <code>Address</code> to ping
102    */
103   public void addPingHook(Address address)
104   {
105     synchronized (addresses)
106     {
107       addresses.add(address);
108     }
109   }
110 
111   /**
112    * Tell the thread to stop pinging an address
113    * 
114    * @param address <code>Address</code> to stop pinging
115    */
116   public void removePingHook(Address address)
117   {
118     synchronized (addresses)
119     {
120       addresses.remove(address);
121     }
122   }
123 
124   /**
125    * Register a new ping listener, that will receive ping events
126    * 
127    * @param listener <code>FaultDetectionListener</code> object that will then
128    *                    receive events and results of ping action
129    */
130   public void addListener(FaultDetectionListener listener)
131   {
132     synchronized (listeners)
133     {
134       listeners.add(listener);
135     }
136   }
137 
138   /**
139    * Unregister a ping listener
140    * 
141    * @param listener <code>FaultDetectionListener</code> will not receive
142    *                    further events
143    */
144   public void removeListener(FaultDetectionListener listener)
145   {
146     synchronized (listeners)
147     {
148       listeners.remove(listener);
149     }
150   }
151 
152   /**
153    * Execute a ping on an <code>Address</code>. This does not collect the
154    * result
155    * 
156    * @param address <code>Address</code> address object to check for
157    */
158   private void doPing(IpAddress address)
159   {
160     Thread   t = (Thread  ) sockets.get(address);
161     if (t == null)
162     {
163       try
164       {
165         t = new UDPPingThread(localAddress, address, (int) waitInterval);
166       }
167       catch (SocketException   e)
168       {
169         notifyListenerOfEvent(FaultDetectionListener.EVENT_ERROR_LOCAL_ADDRESS,
170             localAddress);
171         return;
172       }
173     }
174     t.start();
175   }
176 
177   /**
178    * Get the result and call back events on listeners 
179    * 
180    * @param address <code>IpAddress</code> of the result to get
181    */
182   public void getResult(IpAddress address)
183   {
184     UDPPingThread t = (UDPPingThread) sockets.get(address);
185     if (!t.isTargetAlive())
186     {
187       notifyListenerOfEvent(FaultDetectionListener.THREAD_SUSPECTED, address);
188     }
189     else
190     {
191       notifyListenerOfEvent(FaultDetectionListener.THREAD_ALIVE, address);
192     }
193   }
194 
195   /**
196    * Notify listeners of an event on the specified address
197    * 
198    * @see FaultDetectionService
199    * @param event an int representing the event value
200    * @param address the address where the event occured
201    */
202   public void notifyListenerOfEvent(int event, IpAddress address)
203   {
204     switch (event)
205     {
206       case FaultDetectionListener.THREAD_ALIVE :
207         synchronized (listeners)
208         {
209           for (int i = 0; i < listeners.size(); i++)
210             ((FaultDetectionListener) listeners.get(i)).alive(address);
211         }
212         break;
213       case FaultDetectionListener.THREAD_SUSPECTED :
214         synchronized (listeners)
215         {
216           for (int i = 0; i < listeners.size(); i++)
217             ((FaultDetectionListener) listeners.get(i)).noResponse(address);
218         }
219         break;
220       default :
221         synchronized (listeners)
222         {
223           for (int i = 0; i < listeners.size(); i++)
224             ((FaultDetectionListener) listeners.get(i)).pingServiceEvent(event);
225         }
226         break;
227     }
228   }
229 
230   /**
231    * @see java.lang.Runnable#run()
232    */
233   public void run()
234   {
235     while (!stop)
236     {
237 
238       // UDPPingThread all addresses
239       synchronized (addresses)
240       {
241         for (int i = 0; i < addresses.size(); i++)
242           doPing(((IpAddress) addresses.get(i)));
243       }
244 
245       try
246       {
247         // Wait till next ping time
248         synchronized (this)
249         {
250           wait(waitInterval);
251         }
252       }
253       catch (InterruptedException   e)
254       {
255         // Notify listeners
256         notifyListenerOfEvent(FaultDetectionListener.EVENT_INTERRUPTED,
257             localAddress);
258       }
259 
260       // Get results
261       synchronized (addresses)
262       {
263         for (int i = 0; i < addresses.size(); i++)
264           getResult((IpAddress) addresses.get(i));
265       }
266     }
267 
268     // Quitting
269     notifyListenerOfEvent(FaultDetectionListener.EVENT_QUIT, localAddress);
270   }
271 }
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags