KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > org > objectweb > dream > protocol > utobcast > UTOImpl


1 /**
2  * Dream
3  * Copyright (C) 2003-2004 INRIA Rhone-Alpes
4  *
5  * This library is free software; you can redistribute it and/or
6  * modify it under the terms of the GNU Lesser General Public
7  * License as published by the Free Software Foundation; either
8  * version 2 of the License, or (at your option) any later version.
9  *
10  * This library is distributed in the hope that it will be useful,
11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13  * Lesser General Public License for more details.
14  *
15  * You should have received a copy of the GNU Lesser General Public
16  * License along with this library; if not, write to the Free Software
17  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18  *
19  * Contact: dream@objectweb.org
20  *
21  * Initial developer(s): Vivien Quema
22  * Contributor(s):
23  */
24
25 package org.objectweb.dream.protocol.utobcast;
26
27 import java.util.Map JavaDoc;
28
29 import org.objectweb.dream.AbstractComponent;
30 import org.objectweb.dream.IOPushException;
31 import org.objectweb.dream.InterruptedPullException;
32 import org.objectweb.dream.InterruptedPushException;
33 import org.objectweb.dream.Pull;
34 import org.objectweb.dream.PullException;
35 import org.objectweb.dream.Push;
36 import org.objectweb.dream.PushException;
37 import org.objectweb.dream.message.Message;
38 import org.objectweb.dream.message.manager.MessageManager;
39 import org.objectweb.dream.protocol.utobcast.message.UTOBcastChunk;
40 import org.objectweb.dream.util.Error;
41 import org.objectweb.fractal.api.NoSuchInterfaceException;
42 import org.objectweb.fractal.api.control.IllegalBindingException;
43 import org.objectweb.fractal.api.control.IllegalLifeCycleException;
44 import org.objectweb.util.monolog.api.BasicLevel;
45
46 /**
47  * Implementation of the UTO component. This component handles UTO messages.
48  * Upon message reception,
49  * <ul>
50  * <li>either it delivers the message and all the messages stored in the
51  * <code>toBeDelivered</code> queue that can be delivered.</li>
52  * <li>or it stores the message in the <code>toBeDelivered</code>.</li>
53  * </ul>
54  * In both cases, an ACK is sent to pleader and pbackup.
55  */
56 public class UTOImpl extends AbstractComponent implements Push
57 {
58
59   // ---------------------------------------------------------------------------
60 // Fields of this class
61 // ---------------------------------------------------------------------------
62
63   /**
64    * The commonly used name to refer to the <code>deliveryItf</code>
65    * interface.
66    */
67   public static final String JavaDoc DELIVERY_ITF_NAME = "delivery";
68
69   /**
70    * The commonly used name to refer to the <code>toBeDeliveredInItf</code>
71    * interface.
72    */
73   public static final String JavaDoc TO_BE_DELIVERED_IN_ITF_NAME = "to-be-delivered-in";
74
75   /**
76    * The commonly used name to refer to the <code>toBeDeliveredOutItf</code>
77    * interface.
78    */
79   public static final String JavaDoc TO_BE_DELIVERED_OUT_ITF_NAME = "to-be-delivered-out";
80
81   /** The interface to deliver messages. */
82   protected Push deliveryItf;
83
84   /** The interface to store messages in the <code>ToBeDelivered</code> queue. */
85   protected Push toBeDeliveredInItf;
86
87   /**
88    * The interface to retrieve messages to be delivered from the
89    * <code>toBeDelivered</code> queue.
90    */
91   protected Pull toBeDeliveredOutItf;
92
93   /** The interface to send ACK messages. */
94   protected Push outPushItf;
95
96   /** The interface to retrieve process membership information. */
97   protected ProcessMembership processMembershipItf;
98
99   /** The interface to manage message lifecycle. */
100   protected MessageManager messageManagerItf;
101
102   // ---------------------------------------------------------------------------
103 // Constructor
104 // ---------------------------------------------------------------------------
105
106   /**
107    * Constructor.
108    */
109   public UTOImpl()
110   {
111   }
112
113   // ---------------------------------------------------------------------------
114 // Implementation of the Push interface
115 // ---------------------------------------------------------------------------
116
117   /**
118    * @see org.objectweb.dream.Push#push(org.objectweb.dream.message.Message,
119    * java.util.Map)
120    */
121   public void push(Message message, Map JavaDoc context) throws PushException
122   {
123     if (logger.isLoggable(BasicLevel.DEBUG))
124     {
125       logger.log(BasicLevel.DEBUG, "Received UTO message " + message);
126     }
127     //////////////////////////////////////////////////////////
128 // Add the message to the ToBeDelivered list
129 //////////////////////////////////////////////////////////
130
131     if (logger.isLoggable(BasicLevel.DEBUG))
132     {
133       logger.log(BasicLevel.DEBUG, "Add UTO message " + message
134           + " to the toBeDelivered queue");
135     }
136     Message msg = messageManagerItf.duplicateMessage(message, true);
137     toBeDeliveredInItf.push(msg, null);
138
139     //////////////////////////////////////////////////////////
140 // Deliver all messages that are stored in the ToBeDelivered list and that
141 // can be delivered
142 //////////////////////////////////////////////////////////
143 try
144     {
145       while ((msg = toBeDeliveredOutItf.pull(null)) != null)
146       {
147         if (logger.isLoggable(BasicLevel.DEBUG))
148         {
149           logger.log(BasicLevel.DEBUG, "Deliver UTO message " + msg);
150         }
151         deliveryItf.push(msg, null);
152       }
153     }
154     catch (InterruptedPullException e)
155     {
156       // Happens if the queue is in blocking mode
157 throw new InterruptedPushException("Interrupted while pulling a message");
158     }
159     catch (PullException e)
160     {
161       // This is a bug: queue must either return null or block
162 Error.bug(logger, e);
163     }
164     //////////////////////////////////////////////////////////
165 // Send ACK message to leader and backup
166 //////////////////////////////////////////////////////////
167
168     UTOBcastChunk chunk = (UTOBcastChunk) message
169         .getChunk(UTOBcastChunk.DEFAULT_NAME);
170     chunk.setUTOBcastMessageType(UTOBcastChunk.ACK);
171
172     // ... to leader
173 try
174     {
175       chunk.setProcessTo(processMembershipItf.getLeader());
176     }
177     catch (InterruptedException JavaDoc e1)
178     {
179       throw new InterruptedPushException(e1);
180     }
181     try
182     {
183       // We do not clone message
184 messageManagerItf.duplicateMessage(message, false);
185       if (logger.isLoggable(BasicLevel.DEBUG))
186       {
187         logger.log(BasicLevel.DEBUG, "Send ACK message " + message
188             + " to leader");
189       }
190       outPushItf.push(message, null);
191     }
192     catch (IOPushException e)
193     {
194       // We detect that the leader has crashed
195 // TODO
196 e.printStackTrace();
197     }
198
199     // ... to backup
200 try
201     {
202       chunk.setProcessTo(processMembershipItf.getBackup());
203     }
204     catch (InterruptedException JavaDoc e1)
205     {
206       throw new InterruptedPushException(e1);
207     }
208     try
209     {
210       // We do not duplicate messages
211 if (logger.isLoggable(BasicLevel.DEBUG))
212       {
213         logger.log(BasicLevel.DEBUG, "Send ACK message " + message
214             + " to backup");
215       }
216       outPushItf.push(message, null);
217     }
218     catch (IOPushException e)
219     {
220       // We detect that the backup has crashed
221 // TODO
222 e.printStackTrace();
223     }
224
225   }
226
227   // ---------------------------------------------------------------------------
228 // Implementation of the BindingController interface
229 // ---------------------------------------------------------------------------
230
231   /**
232    * @see org.objectweb.fractal.api.control.BindingController#bindFc(java.lang.String,
233    * java.lang.Object)
234    */
235   public void bindFc(String JavaDoc clientItfName, Object JavaDoc serverItf)
236       throws NoSuchInterfaceException, IllegalBindingException,
237       IllegalLifeCycleException
238   {
239     super.bindFc(clientItfName, serverItf);
240     if (clientItfName.equals(DELIVERY_ITF_NAME))
241     {
242       deliveryItf = (Push) serverItf;
243     }
244     else if (clientItfName.equals(TO_BE_DELIVERED_IN_ITF_NAME))
245     {
246       toBeDeliveredInItf = (Push) serverItf;
247     }
248     else if (clientItfName.equals(TO_BE_DELIVERED_OUT_ITF_NAME))
249     {
250       toBeDeliveredOutItf = (Pull) serverItf;
251     }
252     else if (clientItfName.equals(Push.OUT_PUSH_ITF_NAME))
253     {
254       outPushItf = (Push) serverItf;
255     }
256     else if (clientItfName.equals(ProcessMembership.ITF_NAME))
257     {
258       processMembershipItf = (ProcessMembership) serverItf;
259     }
260     else if (clientItfName.equals(MessageManager.ITF_NAME))
261     {
262       messageManagerItf = (MessageManager) serverItf;
263     }
264   }
265
266   /**
267    * @see org.objectweb.fractal.api.control.BindingController#listFc()
268    */
269   public String JavaDoc[] listFc()
270   {
271     return new String JavaDoc[]{DELIVERY_ITF_NAME, TO_BE_DELIVERED_IN_ITF_NAME,
272         TO_BE_DELIVERED_OUT_ITF_NAME, Push.OUT_PUSH_ITF_NAME,
273         ProcessMembership.ITF_NAME, MessageManager.ITF_NAME};
274   }
275
276 }
Popular Tags