KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > org > objectweb > dream > queue > AbstractPushIncomingHandlerOverflowImpl


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.queue;
26
27 import java.util.Map JavaDoc;
28
29 import org.objectweb.dream.AbstractComponent;
30 import org.objectweb.dream.InterruptedPushException;
31 import org.objectweb.dream.Push;
32 import org.objectweb.dream.PushException;
33 import org.objectweb.dream.message.Message;
34 import org.objectweb.dream.message.manager.MessageManager;
35 import org.objectweb.dream.util.Error;
36 import org.objectweb.fractal.api.NoSuchInterfaceException;
37 import org.objectweb.fractal.api.control.IllegalBindingException;
38 import org.objectweb.fractal.api.control.IllegalLifeCycleException;
39
40 /**
41  * Abstract implementation of a Push incoming handler. It implements overflow
42  * policies as defined in {@link PushQueueAttributeController}. Moreover, it
43  * implements the <code>push</code> method: it checks for available space and
44  * then applies appropriate policy.
45  * <p>
46  * Push incoming handler developers can inherit this class. They must implement
47  * two methods:
48  * <ul>
49  * <li><code>canAdd(message)</code> returns a boolean indicating whether the
50  * given message can be added into the buffer.</li>
51  * <li><code>doAdd(message)</code> adds the given message into the buffer.
52  * </li>
53  * </ul>
54  */
55 public abstract class AbstractPushIncomingHandlerOverflowImpl
56     extends
57       AbstractComponent implements Push, PushQueueAttributeController
58 {
59
60   // Ids used in switch instructions
61 protected static final int BLOCK_OVERFLOW_POLICY_ID = 0;
62   protected static final int DROP_QUEUE_MESSAGE_OVERFLOW_POLICY_ID = 1;
63   protected static final int DROP_FIRST_OVERFLOW_POLICY_ID = 2;
64   protected static final int DROP_LAST_OVERFLOW_POLICY_ID = 3;
65   protected static final int DROP_PROCESSED_MESSAGE_OVERFLOW_POLICY_ID = 4;
66   protected static final int EXCEPTION_OVERFLOW_POLICY_ID = 5;
67
68   protected String JavaDoc overflowPolicy;
69   protected short overflowPolicyId;
70
71   // Client interfaces
72 protected MessageManager messageManagerItf;
73   protected Buffer bufferItf;
74   protected BufferRemoveFirstLast bufferRemoveFirstLastItf;
75   protected BufferAddFirstLast bufferAddFirstLastItf;
76
77   // ---------------------------------------------------------------------------
78 // Implementation of the Push interface.
79 // ---------------------------------------------------------------------------
80
81   /**
82    * @see org.objectweb.dream.Push#push(org.objectweb.dream.message.Message,
83    * java.util.Map)
84    */
85   public void push(Message message, Map JavaDoc context) throws PushException
86   {
87     try
88     {
89       if (overflowPolicyId != BLOCK_OVERFLOW_POLICY_ID && !canAdd(message))
90       {
91         switch (overflowPolicyId)
92         {
93           case DROP_QUEUE_MESSAGE_OVERFLOW_POLICY_ID : {
94             messageManagerItf.deleteMessage(bufferItf.remove());
95             break;
96           }
97           case DROP_FIRST_OVERFLOW_POLICY_ID : {
98             if (bufferRemoveFirstLastItf != null)
99             {
100               messageManagerItf.deleteMessage(bufferRemoveFirstLastItf
101                   .removeFirst());
102             }
103             else
104             {
105               throw new UnsupportedOperationException JavaDoc(
106                   "Unable to remove first element of the queue.");
107             }
108             break;
109           }
110           case DROP_LAST_OVERFLOW_POLICY_ID : {
111             if (bufferRemoveFirstLastItf != null)
112             {
113               messageManagerItf.deleteMessage(bufferRemoveFirstLastItf
114                   .removeLast());
115             }
116             else
117             {
118               throw new UnsupportedOperationException JavaDoc(
119                   "Unable to remove last element of the queue.");
120             }
121             break;
122           }
123           case DROP_PROCESSED_MESSAGE_OVERFLOW_POLICY_ID : {
124             messageManagerItf.deleteMessage(message);
125             break;
126           }
127           case EXCEPTION_OVERFLOW_POLICY_ID : {
128             throw new BufferOverflowException("Overflow in buffer");
129           }
130           default : {
131             Error.error("Unknown overflow policy" + overflowPolicyId, logger);
132           }
133         }
134       }
135       else
136       {
137         doAdd(message);
138       }
139     }
140     catch (InterruptedException JavaDoc e)
141     {
142       throw new InterruptedPushException("Interrupted while adding a message",
143           e);
144     }
145   }
146
147   // ---------------------------------------------------------------------------
148 // Methods that must be implemented by concrete super classes.
149 // ---------------------------------------------------------------------------
150
151   /**
152    * Checks whether the given message can be added into the buffer.
153    *
154    * @param message the message to be tested.
155    * @return <code>true</code> if the message can be added.
156    */
157   protected abstract boolean canAdd(Message message);
158
159   /**
160    * Adds a message to the buffer. This method should not check if there is
161    * enough available space provided it has already been done.
162    *
163    * @param message the message to be added.
164    * @throws InterruptedException if it is interrupted while removing the
165    * message.
166    */
167   protected abstract void doAdd(Message message) throws InterruptedException JavaDoc;
168
169   // ---------------------------------------------------------------------------
170 // Implementation of the PushQueueAttributeController interface.
171 // ---------------------------------------------------------------------------
172
173   /**
174    * @see PushQueueAttributeController#getOverflowPolicy()
175    */
176   public String JavaDoc getOverflowPolicy()
177   {
178     return overflowPolicy;
179   }
180
181   /**
182    * @see PushQueueAttributeController#setOverflowPolicy(String)
183    */
184   public synchronized void setOverflowPolicy(String JavaDoc policy)
185   {
186     // set overflow policy attribute reference to interface constant to test
187 // policy efficiently.
188 if (policy.equals(BLOCK_OVERFLOW_POLICY))
189     {
190       overflowPolicy = BLOCK_OVERFLOW_POLICY;
191       overflowPolicyId = BLOCK_OVERFLOW_POLICY_ID;
192     }
193     else if (policy.equals(DROP_QUEUE_MESSAGE_OVERFLOW_POLICY))
194     {
195       overflowPolicy = DROP_QUEUE_MESSAGE_OVERFLOW_POLICY;
196       overflowPolicyId = DROP_QUEUE_MESSAGE_OVERFLOW_POLICY_ID;
197     }
198     else if (policy.equals(DROP_FIRST_OVERFLOW_POLICY))
199     {
200       overflowPolicy = DROP_FIRST_OVERFLOW_POLICY;
201       overflowPolicyId = DROP_FIRST_OVERFLOW_POLICY_ID;
202     }
203     else if (policy.equals(DROP_LAST_OVERFLOW_POLICY))
204     {
205       overflowPolicy = DROP_LAST_OVERFLOW_POLICY;
206       overflowPolicyId = DROP_LAST_OVERFLOW_POLICY_ID;
207     }
208     else if (policy.equals(DROP_PROCESSED_MESSAGE_OVERFLOW_POLICY))
209     {
210       overflowPolicy = DROP_PROCESSED_MESSAGE_OVERFLOW_POLICY;
211       overflowPolicyId = DROP_PROCESSED_MESSAGE_OVERFLOW_POLICY_ID;
212     }
213     else if (policy.equals(EXCEPTION_OVERFLOW_POLICY))
214     {
215       overflowPolicy = EXCEPTION_OVERFLOW_POLICY;
216       overflowPolicyId = EXCEPTION_OVERFLOW_POLICY_ID;
217     }
218     else
219     {
220       throw new IllegalArgumentException JavaDoc("Unknown overflow policy : " + policy);
221     }
222   }
223
224   // ---------------------------------------------------------------------------
225 // Implementation of the BindingController interface.
226 // ---------------------------------------------------------------------------
227
228   /**
229    * @see org.objectweb.fractal.api.control.BindingController#bindFc(String,
230    * Object)
231    */
232   public synchronized void bindFc(String JavaDoc clientItfName, Object JavaDoc serverItf)
233       throws NoSuchInterfaceException, IllegalBindingException,
234       IllegalLifeCycleException
235   {
236     super.bindFc(clientItfName, serverItf);
237     if (clientItfName.equals(MessageManager.ITF_NAME))
238     {
239       messageManagerItf = (MessageManager) serverItf;
240     }
241     else if (clientItfName.equals(Buffer.ITF_NAME))
242     {
243       bufferItf = (Buffer) serverItf;
244     }
245     else if (clientItfName.equals(BufferRemoveFirstLast.ITF_NAME))
246     {
247       bufferRemoveFirstLastItf = (BufferRemoveFirstLast) serverItf;
248     }
249     else if (clientItfName.equals(BufferAddFirstLast.ITF_NAME))
250     {
251       bufferAddFirstLastItf = (BufferAddFirstLast) serverItf;
252     }
253   }
254
255   /**
256    * @see org.objectweb.fractal.api.control.BindingController#listFc()
257    */
258   public String JavaDoc[] listFc()
259   {
260     return new String JavaDoc[]{MessageManager.ITF_NAME, Buffer.ITF_NAME,
261         BufferRemoveFirstLast.ITF_NAME, BufferAddFirstLast.ITF_NAME};
262   }
263
264 }
Popular Tags