001: /*
002: * Copyright 1999,2004 The Apache Software Foundation.
003: *
004: * Licensed under the Apache License, Version 2.0 (the "License");
005: * you may not use this file except in compliance with the License.
006: * You may obtain a copy of the License at
007: *
008: * http://www.apache.org/licenses/LICENSE-2.0
009: *
010: * Unless required by applicable law or agreed to in writing, software
011: * distributed under the License is distributed on an "AS IS" BASIS,
012: * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013: * See the License for the specific language governing permissions and
014: * limitations under the License.
015: */
016:
017: package org.apache.catalina.cluster.util;
018:
019: /**
020: * A smart queue, used for async replication<BR>
021: * the "smart" part of this queue is that if the session is already queued for replication,
022: * and it is updated again, the session will simply be replaced, hence we don't
023: * replicate stuff that is obsolete.
024: * Put this into util, since it is quite generic.
025: *
026: * @author Filip Hanik
027: * @version 1.0
028: */
029:
030: import java.util.LinkedList;
031: import java.util.HashMap;
032:
033: public class SmartQueue {
034:
035: public static org.apache.commons.logging.Log log = org.apache.commons.logging.LogFactory
036: .getLog(SmartQueue.class);
037:
038: /**
039: * This is the actual queue
040: */
041: private LinkedList queue = new LinkedList();
042: /**
043: * And this is only for performance, fast lookups
044: */
045: private HashMap queueMap = new HashMap();
046:
047: private Object mutex = new Object();
048: public static int debug = 0;
049:
050: public SmartQueue() {
051: }
052:
053: /**
054: * Add an object to the queue
055: * @param entry - the smart entry
056: */
057: public void add(SmartEntry entry) {
058: /*make sure we are within a synchronized block since we are dealing with two
059: unsync collections*/
060: synchronized (mutex) {
061: /*check to see if this object has already been queued*/
062: SmartEntry current = (SmartEntry) queueMap.get(entry
063: .getKey());
064: if (current == null) {
065: /*the object has not been queued, at it to the end of the queue*/
066: if (debug != 0)
067: log.debug("[" + Thread.currentThread().getName()
068: + "][SmartQueue] Adding new object="
069: + entry);
070: queue.addLast(entry);
071: queueMap.put(entry.getKey(), entry);
072: } else {
073: /*the object has been queued, replace the value*/
074: if (debug != 0)
075: log.debug("[" + Thread.currentThread().getName()
076: + "][SmartQueue] Replacing old object="
077: + current);
078: current.setValue(entry.getValue());
079: if (debug != 0)
080: log.debug("with new object=" + current);
081: }
082: /*wake up all the threads that are waiting for the lock to be released*/
083: mutex.notifyAll();
084: }
085: }
086:
087: public int size() {
088: synchronized (mutex) {
089: return queue.size();
090: }
091: }
092:
093: /**
094: * Blocks forever until an element has been added to the queue
095: */
096: public SmartEntry remove() {
097: return remove(0);
098: }
099:
100: public SmartEntry remove(long timeout) {
101: SmartEntry result = null;
102: long startEntry = System.currentTimeMillis();
103: synchronized (mutex) {
104: while (size() == 0) {
105: try {
106: if (debug != 0)
107: log
108: .debug("["
109: + Thread.currentThread()
110: .getName()
111: + "][SmartQueue] Queue sleeping until object added size="
112: + size() + ".");
113: if ((timeout != 0)
114: && ((System.currentTimeMillis() - startEntry) > timeout)) {
115: return null;
116: }
117: mutex.wait(timeout);
118: if (debug != 0)
119: log
120: .debug("["
121: + Thread.currentThread()
122: .getName()
123: + "][SmartQueue] Queue woke up or interrupted size="
124: + size() + ".");
125: } catch (IllegalMonitorStateException ex) {
126: throw ex;
127: } catch (InterruptedException ex) {
128: }//catch
129: }//while
130: /*guaranteed that we are not empty by now*/
131: result = (SmartEntry) queue.removeFirst();
132: queueMap.remove(result.getKey());
133: if (debug != 0)
134: log.debug("[" + Thread.currentThread().getName()
135: + "][SmartQueue] Returning=" + result);
136: }
137: return result;
138: }
139:
140: public static class SmartEntry {
141: protected Object key;
142: protected Object value;
143:
144: public SmartEntry(Object key, Object value) {
145: if (key == null)
146: throw new IllegalArgumentException(
147: "SmartEntry key can not be null.");
148: if (value == null)
149: throw new IllegalArgumentException(
150: "SmartEntry value can not be null.");
151: this .key = key;
152: this .value = value;
153: }
154:
155: public Object getKey() {
156: return key;
157: }
158:
159: public Object getValue() {
160: return value;
161: }
162:
163: public void setValue(Object value) {
164: if (value == null)
165: throw new IllegalArgumentException(
166: "SmartEntry value can not be null.");
167: this .value = value;
168: }
169:
170: public int hashCode() {
171: return key.hashCode();
172: }
173:
174: public boolean equals(Object o) {
175: if (!(o instanceof SmartEntry))
176: return false;
177: SmartEntry other = (SmartEntry) o;
178: return other.getKey().equals(getKey());
179: }
180:
181: public String toString() {
182: return "[SmartyEntry key=" + key + " value=" + value + "]";
183: }
184: }
185:
186: }
|