PersistentQueue


1   /*
2    * Copyright 2002 by
3    * <a HREF="http://www.coridan.com">Coridan</a>
4    * <a HREF="mailto: support@coridan.com ">support@coridan.com</a>
5    *
6    * The contents of this file are subject to the Mozilla Public License Version
7    * 1.1 (the "License"); you may not use this file except in compliance with the
8    * License.  You may obtain a copy of the License at
9    * http://www.mozilla.org/MPL/
10   *
11   * Software distributed under the License is distributed on an "AS IS" basis,
12   * WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
13   * for the specific language governing rights and limitations under the
14   * License.
15   *
16   * The Original Code is "MantaRay" (TM).
17   *
18   * The Initial Developer of the Original Code is Amir Shevat.
19   * Portions created by the Initial Developer are Copyright (C) 2006
20   * Coridan Inc.  All Rights Reserved.
21   *
22   * Contributor(s): all the names of the contributors are added in the source
23   * code where applicable.
24   *
25   * Alternatively, the contents of this file may be used under the terms of the
26   * LGPL license (the "GNU LESSER GENERAL PUBLIC LICENSE"), in which case the
27   * provisions of LGPL are applicable instead of those above.  If you wish to
28   * allow use of your version of this file only under the terms of the LGPL
29   * License and not to allow others to use your version of this file under
30   * the MPL, indicate your decision by deleting the provisions above and
31   * replace them with the notice and other provisions required by the LGPL.
32   * If you do not delete the provisions above, a recipient may use your version
33   * of this file under either the MPL or the GNU LESSER GENERAL PUBLIC LICENSE.
34   
35   *
36   * This library is free software; you can redistribute it and/or modify it
37   * under the terms of the MPL as stated above or under the terms of the GNU
38   * Lesser General Public License as published by the Free Software Foundation;
39   * either version 2.1 of the License, or any later version.
40   *
41   * This library is distributed in the hope that it will be useful, but WITHOUT
42   * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
43   * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
44   * License for more details.
45   */
46  /*
47   * Created on Dec 11, 2003
48   *
49   */
50  package org.mr.core.persistent;
51  
52  import java.io.IOException  ;
53  
54  import java.util.Iterator  ;
55  import java.util.LinkedList  ;
56  import java.util.List  ;
57  
58  import org.apache.commons.logging.Log;
59  import org.apache.commons.logging.LogFactory;
60  
61  import org.mr.core.util.SynchronizedQueue;
62  import org.mr.core.util.byteable.Byteable;
63  
64  
65  
66  /**
67   * PersistentQueue acks like any other SynchronizedQueue but keeps data in the disk for Persistent
68   * @author Amir Shevat
69   */
70  public class PersistentQueue extends SynchronizedQueue{
71  
72  // the file of persistent in the system
73      private Log log;
74  // every element is given a name (file name) this is it
75      private int itemsCount = 0;
76      // the default persistent of the queue
77      private boolean defaultPersistent ;
78      
79      private PersistentManager  persistentManager;
80      // list of the real underline elements - good for get queue copy
81      private LinkedList   underLineElementsCopy= new LinkedList  ();
82      
83      /**
84       * @param name - the name of the file this queue will be save to.
85       * @param defaultPersistent - when invoking method 'enqueue(Object o)' the file will 
86       * be persistent if this value is true.
87       * @param blocking - true if return only after file has been written else false
88       */
89      public PersistentQueue(String   name ,boolean defaultPersistent ,  boolean blocking){
90          // call papa
91          super();
92          this.defaultPersistent =defaultPersistent;
93          log=LogFactory.getLog("PersistentQueue");
94          persistentManager = PersistentManagerFactory.getPersistentManager(name);
95          
96          try {
97              recover(name);
98          } catch (IOException   e) {
99              if(log.isFatalEnabled())
100                 log.fatal("Can not init persistent stracture. ",e);
101         }   
102         
103     }//PersistentQueue
104     
105     
106     /**
107      * recovers the data of the stract from file
108      * @param name the name of the file this stract was save to
109      * @throws IOException
110      */
111     private  void recover(String   name) throws IOException  {
112         persistentManager.recover();
113         int[] keySet = persistentManager.getKeys();
114         if(log.isInfoEnabled() && keySet.length >0 ){
115             log.info("Recoverd "+name+". There are "+keySet.length+" elements there.");
116         }
117         for(int index =0 ; index < keySet.length; index++){
118             
119             Object   entry =persistentManager.getPersistentObject(keySet[index]);
120             PersistentEvent event = new PersistentEvent();
121             event.setEntry((Byteable)entry);
122             event.setEntryName(keySet[index]);
123             event.setPersistentState(true);
124             super.enqueue(event);
125             underLineElementsCopy.add(entry);
126             //persistentManager.deletePersistentObject(keySet[index]);
127             //this.enqueue(entry);
128             itemsCount++;
129         }
130     }//recover
131     
132     /**
133      * same as super but with file save
134      */
135     synchronized public boolean enqueue(Object   o){
136         return enqueue(o,defaultPersistent);
137     }//enqueue
138     
139     /**
140      * puts an object in the queue and saves it if persistent == tue
141      * @param o the object to be enqueued
142      * @param persistent if true save
143      */
144     synchronized public boolean enqueue(Object   o , boolean persistent){
145         PersistentEvent event = new PersistentEvent();
146         event.setEntry((Byteable)o);
147         int name = getValidFreePersistentName();
148         event.setEntryName(name);
149         event.setPersistentState(persistent);
150         
151         if(persistent){
152             try{
153                 persistentManager.savePersistentObject(event.getEntryName(),event.getEntry());
154                 
155             }catch(IOException   e){
156                 if(log.isFatalEnabled())
157                     log.fatal("Can not save persistent stracture. ",e);
158             }
159         }
160         boolean b =super.enqueue(event);
161         underLineElementsCopy.add(o);
162         return b;
163     }//enqueue
164     
165     /**
166      * @return a valid file name with the flowing disclamer that thew queue is not bigger then the max file of fat 
167      */
168     private int getValidFreePersistentName(){
169         int result = itemsCount;
170         itemsCount++;
171         if(itemsCount >= PersistentManager.MAX_FILES_PER_STORAGE){
172             itemsCount = 1;
173         }
174         
175         return result;
176     }
177     
178     /**
179      * same as super but with file delete
180      */
181     synchronized public Object   dequeue(){
182         PersistentEvent event = (PersistentEvent)super.dequeue();
183         if(event.isPersistentState())
184             persistentManager.deletePersistentObject(event.getEntryName());
185         underLineElementsCopy.remove(event.getEntry());
186         return event.getEntry();
187     }
188     
189     /**
190      * removes an element from the queue
191      * @param toBRemoved the element to be removed
192      * @return true if found and removed else false
193      */
194     public synchronized boolean removeElement(Object   toBRemoved){
195         List   list = getUnderlineList();
196         Iterator   elements = list.iterator();
197         while(elements.hasNext()){
198             PersistentEvent event =(PersistentEvent)elements.next();
199             if(event.getEntry() ==toBRemoved ){
200                 elements.remove();
201                 if(event.isPersistentState())
202                     persistentManager.deletePersistentObject(event.getEntryName());
203                 underLineElementsCopy.remove(toBRemoved);
204                 
205                 return true;
206             }
207         }
208         return false;
209         
210     }
211     /**
212      * same as super but with file delete
213      */
214     synchronized public Object   dequeueNoBlock(){
215         PersistentEvent event = (PersistentEvent)super.dequeueNoBlock();
216         if(event != null){
217             persistentManager.deletePersistentObject(event.getEntryName());
218             return event.getEntry();
219         }else{
220             return null;
221         }
222     }
223     
224     synchronized public  LinkedList   copyUnderlineElementsList(){
225         
226         return new LinkedList  (underLineElementsCopy);
227     }
228     
229     
230 
231 }
232
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags