KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > org > objectweb > tribe > adapters > PullPushAdapter


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): Emmanuel Cecchet.
22  * Contributor(s): __________________.
23  */
24
25 package org.objectweb.tribe.adapters;
26
27 import java.io.Serializable JavaDoc;
28 import java.util.ArrayList JavaDoc;
29
30 import org.objectweb.tribe.channel.ReliableGroupChannel;
31 import org.objectweb.tribe.common.GroupIdentifier;
32 import org.objectweb.tribe.common.log.Trace;
33 import org.objectweb.tribe.exceptions.ChannelException;
34 import org.objectweb.tribe.exceptions.NotConnectedException;
35 import org.objectweb.tribe.messages.MessageListener;
36
37 /**
38  * Allows a client to be notified when messages have been received instead of
39  * having to actively poll the channel for new messages. Typically used in the
40  * client role (receive()). This blocks works with any kind of reliable group
41  * channel.
42  * <p>
43  * Note that message delivery is single threaded and if the message handler
44  * hangs, it will block further message delivery. Send() methods are just
45  * provided for convenience and wraps calls to the same methods on the channel.
46  * <p>
47  * This adapter is inspired from JGroups PullPushAdapter but does not deal with
48  * views.
49  *
50  * @author <a HREF="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
51  * @author Bela Ban
52  * @version 1.0
53  */
54 public class PullPushAdapter implements Runnable JavaDoc
55 {
56   protected ReliableGroupChannel channel = null;
57   protected MessageListener listener = null;
58   protected Thread JavaDoc readerThread = null;
59   private static Trace logger = Trace
60                                                   .getLogger("org.objectweb.tribe.channel");
61   private boolean isKilled;
62
63   /**
64    * Creates a new <code>PullPushAdapter</code> on a channel and use the
65    * provided listener for callbacks on message reception.
66    *
67    * @param channel channel to read messages from
68    * @param listener MessageListener for callbacks
69    */
70   public PullPushAdapter(ReliableGroupChannel channel, MessageListener listener)
71   {
72     this.channel = channel;
73     this.listener = listener;
74     start();
75   }
76
77   /**
78    * Return the underlying channel.
79    *
80    * @return underlying channel
81    */
82   public ReliableGroupChannel getChannel()
83   {
84     return channel;
85   }
86
87   /**
88    * Start a new thread
89    */
90   public void start()
91   {
92     if (readerThread == null)
93     {
94       readerThread = new Thread JavaDoc(this, "PullPushAdapterThread");
95       isKilled = false;
96       readerThread.start();
97       Thread.yield(); // Let the thread start
98 }
99   }
100
101   /**
102    * Stop the current adapter and its associated thread.
103    */
104   public void stop()
105   {
106     if ((readerThread != null) && readerThread.isAlive())
107     {
108       isKilled = true;
109       readerThread.interrupt();
110       try
111       {
112         channel.send(null);
113       }
114       catch (Exception JavaDoc ignore)
115       {
116       }
117     }
118     try
119     {
120       readerThread.join(1000);
121     }
122     catch (Exception JavaDoc ex)
123     {
124     }
125   }
126
127   /**
128    * Sends the given message to all members of the group.
129    *
130    * @param msg message to send
131    * @return an <code>ArrayList</code> of Members who failed, or null if all
132    * members received successfully the message.
133    * @throws ChannelException if an error occurs
134    * @throws NotConnectedException if the channel is not connected to any group
135    * @see ReliableGroupChannel#send(Serializable)
136    */
137   public ArrayList JavaDoc send(Serializable JavaDoc msg) throws ChannelException,
138       NotConnectedException
139   {
140     return channel.send(msg);
141   }
142
143   /**
144    * Sends a message to a subset of group members.
145    *
146    * @param msg message to send
147    * @param members <code>ArrayList</code> of <code>Member</code> that are
148    * all part of the group
149    * @return an <code>ArrayList</code> of Members who failed, or null if all
150    * members received successfully the message.
151    * @throws ChannelException if an error occurs
152    * @throws NotConnectedException if the channel is not connected to any group
153    * @see ReliableGroupChannel#send(Serializable, ArrayList)
154    */
155   public ArrayList JavaDoc send(Serializable JavaDoc msg, ArrayList JavaDoc members)
156       throws ChannelException, NotConnectedException
157   {
158     return channel.send(msg, members);
159   }
160
161   /**
162    * Sends a message to members of group gid without being necessary member of
163    * the group.
164    *
165    * @param msg message to send
166    * @param members <code>ArrayList</code> of <code>Member</code> that are
167    * all part of the group
168    * @return an <code>ArrayList</code> of Members who failed, or null if all
169    * members received successfully the message.
170    * @throws ChannelException if an error occurs
171    * @throws NotConnectedException if the channel is not connected to any group
172    * @see ReliableGroupChannel#send(Serializable, GroupIdentifier, ArrayList)
173    */
174   public ArrayList JavaDoc send(Serializable JavaDoc msg, GroupIdentifier gid, ArrayList JavaDoc members)
175       throws ChannelException, NotConnectedException
176   {
177     return channel.send(msg, gid, members);
178   }
179
180   /**
181    * Reentrant run(): message reception is serialized, then the listener is
182    * notified of the message reception
183    */
184   public void run()
185   {
186     Serializable JavaDoc msg;
187
188     while (!isKilled)
189     {
190       try
191       {
192         msg = channel.receive();
193         listener.receive(msg);
194       }
195       catch (Exception JavaDoc e)
196       {
197         logger.debug("PullPushAdapter: Error while reading from channel", e);
198       }
199     }
200   }
201
202 }
Popular Tags