You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mina.apache.org by tr...@apache.org on 2008/05/22 06:26:49 UTC
svn commit: r659000 - in
/mina/branches/buffer/core/src/main/java/org/apache/mina/queue:
AbstractIoQueue.java CircularIoQueue.java IoQueue.java
SynchronizedIoQueue.java
Author: trustin
Date: Wed May 21 21:26:48 2008
New Revision: 659000
URL: http://svn.apache.org/viewvc?rev=659000&view=rev
Log:
* Added CircularIoQueue, which is based on existing CircularQueue
* Added SynchronizedIoQueue, which is based on existing SynchronizedQueue
* Added IoQueue.element(index)
* Added AbstractIoQueue.iterator()
Added:
mina/branches/buffer/core/src/main/java/org/apache/mina/queue/CircularIoQueue.java (with props)
mina/branches/buffer/core/src/main/java/org/apache/mina/queue/SynchronizedIoQueue.java (with props)
Modified:
mina/branches/buffer/core/src/main/java/org/apache/mina/queue/AbstractIoQueue.java
mina/branches/buffer/core/src/main/java/org/apache/mina/queue/IoQueue.java
Modified: mina/branches/buffer/core/src/main/java/org/apache/mina/queue/AbstractIoQueue.java
URL: http://svn.apache.org/viewvc/mina/branches/buffer/core/src/main/java/org/apache/mina/queue/AbstractIoQueue.java?rev=659000&r1=658999&r2=659000&view=diff
==============================================================================
--- mina/branches/buffer/core/src/main/java/org/apache/mina/queue/AbstractIoQueue.java (original)
+++ mina/branches/buffer/core/src/main/java/org/apache/mina/queue/AbstractIoQueue.java Wed May 21 21:26:48 2008
@@ -20,6 +20,9 @@
package org.apache.mina.queue;
import java.util.AbstractQueue;
+import java.util.ConcurrentModificationException;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,6 +40,7 @@
private static final Logger LOG = LoggerFactory.getLogger(AbstractIoQueue.class);
private volatile IoQueueListener<? super E>[] listeners = newListenerArray(0);
+ private int modCount;
/**
* {@inheritDoc}
@@ -101,6 +105,10 @@
* {@inheritDoc}
*/
public final boolean offer(E e) {
+ if (e == null) {
+ throw new NullPointerException("element");
+ }
+
try {
if (!accept(e)) {
return false;
@@ -114,6 +122,8 @@
}
doOffer(e);
+
+ modCount ++;
offered(e);
return true;
}
@@ -127,10 +137,43 @@
return null;
}
+ modCount ++;
polled(e);
return e;
}
+ @Override
+ public Iterator<E> iterator() {
+ final int expectedModCount = modCount;
+ return new Iterator<E>() {
+ private int index = 0;
+
+ public boolean hasNext() {
+ return index != size();
+ }
+
+ public E next() {
+ try {
+ E next = element(index++);
+ return next;
+ } catch (IndexOutOfBoundsException e) {
+ checkForComodification();
+ throw new NoSuchElementException();
+ }
+ }
+
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ private void checkForComodification() {
+ if (modCount != expectedModCount) {
+ throw new ConcurrentModificationException();
+ }
+ }
+ };
+ }
+
/**
* Performs the actual insertion operation.
*
Added: mina/branches/buffer/core/src/main/java/org/apache/mina/queue/CircularIoQueue.java
URL: http://svn.apache.org/viewvc/mina/branches/buffer/core/src/main/java/org/apache/mina/queue/CircularIoQueue.java?rev=659000&view=auto
==============================================================================
--- mina/branches/buffer/core/src/main/java/org/apache/mina/queue/CircularIoQueue.java (added)
+++ mina/branches/buffer/core/src/main/java/org/apache/mina/queue/CircularIoQueue.java Wed May 21 21:26:48 2008
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.mina.queue;
+
+import java.util.NoSuchElementException;
+
+/**
+ * A {@link IoQueue} which is based on
+ * <a href="http://en.wikipedia.org/wiki/Circular_buffer">an array-backed
+ * circular buffer</a>. This queue is unbound - it automatically expands
+ * or shrinks its capacity as needed.
+ * <p>
+ * This queue is not thread safe. {@link SynchronizedIoQueue} should be used
+ * together to guarantee thread safety.
+ *
+ * @param <E> the type of the queue's elements
+ *
+ * @author The Apache MINA project (dev@mina.apache.org)
+ * @version $Rev$, $Date$
+ */
+public class CircularIoQueue<E> extends AbstractIoQueue<E> {
+
+ private static final int DEFAULT_CAPACITY = 4;
+
+ private final int initialCapacity;
+ // XXX: This volatile keyword here is a workaround for SUN Java Compiler
+ // bug, which produces buggy byte code. I don't event know why adding
+ // a volatile fixes the problem. Eclipse Java Compiler seems to
+ // produce correct byte code.
+ private volatile E[] items;
+ private int mask;
+ private int first;
+ private int last;
+ private boolean full;
+ private int shrinkThreshold;
+
+ /**
+ * Creates a new empty queue.
+ */
+ public CircularIoQueue() {
+ this(DEFAULT_CAPACITY);
+ }
+
+ /**
+ * Creates a new empty queue with the specified initial capacity.
+ */
+ public CircularIoQueue(int initialCapacity) {
+ int actualCapacity = normalizeCapacity(initialCapacity);
+ items = newElementArray(actualCapacity);
+ mask = actualCapacity - 1;
+ this.initialCapacity = actualCapacity;
+ this.shrinkThreshold = 0;
+ }
+
+ private static int normalizeCapacity(int initialCapacity) {
+ int actualCapacity = 1;
+ while (actualCapacity < initialCapacity) {
+ actualCapacity <<= 1;
+ if (actualCapacity < 0) {
+ actualCapacity = 1 << 30;
+ break;
+ }
+ }
+ return actualCapacity;
+ }
+
+ @Override
+ protected final void doOffer(E e) {
+ if (e == null) {
+ throw new NullPointerException("element");
+ }
+
+ expandIfNeeded();
+ items[last] = e;
+ increaseSize();
+ }
+
+ @Override
+ protected final E doPoll() {
+ if (isEmpty()) {
+ return null;
+ }
+
+ E ret = items[first];
+ items[first] = null;
+ decreaseSize();
+
+ if (first == last) {
+ first = last = 0;
+ }
+
+ shrinkIfNeeded();
+ return ret;
+ }
+
+ public final E element(int index) {
+ checkIndex(index);
+ return items[getRealIndex(index)];
+ }
+
+ @Override
+ public int size() {
+ if (full) {
+ return items.length;
+ }
+
+ if (last >= first) {
+ return last - first;
+ } else {
+ return last - first + items.length;
+ }
+ }
+
+ public E peek() {
+ if (isEmpty()) {
+ return null;
+ }
+
+ return items[first];
+ }
+
+ private void checkIndex(int idx) {
+ if (idx < 0 || idx >= size()) {
+ throw new NoSuchElementException(String.valueOf(idx));
+ }
+ }
+
+ private int getRealIndex(int idx) {
+ return first + idx & mask;
+ }
+
+ private void increaseSize() {
+ last = last + 1 & mask;
+ full = first == last;
+ }
+
+ private void decreaseSize() {
+ first = first + 1 & mask;
+ full = false;
+ }
+
+ private void expandIfNeeded() {
+ if (full) {
+ // expand queue
+ final int oldLen = items.length;
+ final int newLen = oldLen << 1;
+ E[] tmp = newElementArray(newLen);
+
+ if (first < last) {
+ System.arraycopy(items, first, tmp, 0, last - first);
+ } else {
+ System.arraycopy(items, first, tmp, 0, oldLen - first);
+ System.arraycopy(items, 0, tmp, oldLen - first, last);
+ }
+
+ first = 0;
+ last = oldLen;
+ items = tmp;
+ mask = tmp.length - 1;
+ if (newLen >>> 3 > initialCapacity) {
+ shrinkThreshold = newLen >>> 3;
+ }
+ }
+ }
+
+ private void shrinkIfNeeded() {
+ int size = size();
+ if (size <= shrinkThreshold) {
+ // shrink queue
+ final int oldLen = items.length;
+ int newLen = normalizeCapacity(size);
+ if (size == newLen) {
+ newLen <<= 1;
+ }
+
+ if (newLen >= oldLen) {
+ return;
+ }
+
+ if (newLen < initialCapacity) {
+ if (oldLen == initialCapacity) {
+ return;
+ } else {
+ newLen = initialCapacity;
+ }
+ }
+
+ E[] tmp = newElementArray(newLen);
+
+ // Copy only when there's something to copy.
+ if (size > 0) {
+ if (first < last) {
+ System.arraycopy(items, first, tmp, 0, last - first);
+ } else {
+ System.arraycopy(items, first, tmp, 0, oldLen - first);
+ System.arraycopy(items, 0, tmp, oldLen - first, last);
+ }
+ }
+
+ first = 0;
+ last = size;
+ items = tmp;
+ mask = tmp.length - 1;
+ shrinkThreshold = 0;
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private E[] newElementArray(int length) {
+ return (E[]) new Object[length];
+ }
+}
Propchange: mina/branches/buffer/core/src/main/java/org/apache/mina/queue/CircularIoQueue.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: mina/branches/buffer/core/src/main/java/org/apache/mina/queue/CircularIoQueue.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: mina/branches/buffer/core/src/main/java/org/apache/mina/queue/IoQueue.java
URL: http://svn.apache.org/viewvc/mina/branches/buffer/core/src/main/java/org/apache/mina/queue/IoQueue.java?rev=659000&r1=658999&r2=659000&view=diff
==============================================================================
--- mina/branches/buffer/core/src/main/java/org/apache/mina/queue/IoQueue.java (original)
+++ mina/branches/buffer/core/src/main/java/org/apache/mina/queue/IoQueue.java Wed May 21 21:26:48 2008
@@ -19,6 +19,7 @@
*/
package org.apache.mina.queue;
+import java.util.NoSuchElementException;
import java.util.Queue;
/**
@@ -44,6 +45,13 @@
public interface IoQueue<E> extends Queue<E> {
/**
+ * Returns the element at the specified position.
+ *
+ * @throws NoSuchElementException if a wrong index is specified
+ */
+ E element(int index);
+
+ /**
* Adds the specified {@link IoQueueListener} to this queue. Once added,
* the listener will be notified whenever the state of this queue changes.
*/
Added: mina/branches/buffer/core/src/main/java/org/apache/mina/queue/SynchronizedIoQueue.java
URL: http://svn.apache.org/viewvc/mina/branches/buffer/core/src/main/java/org/apache/mina/queue/SynchronizedIoQueue.java?rev=659000&view=auto
==============================================================================
--- mina/branches/buffer/core/src/main/java/org/apache/mina/queue/SynchronizedIoQueue.java (added)
+++ mina/branches/buffer/core/src/main/java/org/apache/mina/queue/SynchronizedIoQueue.java Wed May 21 21:26:48 2008
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.mina.queue;
+
+import java.util.Collection;
+import java.util.Iterator;
+
+/**
+ * A decorator that makes the specified {@link IoQueue} thread-safe.
+ * Like any other synchronizing wrappers, its iteration is not thread-safe.
+ *
+ * @author The Apache Directory Project (mina-dev@directory.apache.org)
+ * @version $Rev$, $Date$
+ */
+public class SynchronizedIoQueue<E> implements IoQueue<E> {
+
+ private final IoQueue<E> q;
+
+ public SynchronizedIoQueue(IoQueue<E> q) {
+ this.q = q;
+ }
+
+ public synchronized boolean add(E e) {
+ return q.add(e);
+ }
+
+ public synchronized E element() {
+ return q.element();
+ }
+
+ public synchronized boolean offer(E e) {
+ return q.offer(e);
+ }
+
+ public synchronized E peek() {
+ return q.peek();
+ }
+
+ public synchronized E poll() {
+ return q.poll();
+ }
+
+ public synchronized E remove() {
+ return q.remove();
+ }
+
+ public synchronized boolean addAll(Collection<? extends E> c) {
+ return q.addAll(c);
+ }
+
+ public synchronized void clear() {
+ q.clear();
+ }
+
+ public synchronized boolean contains(Object o) {
+ return q.contains(o);
+ }
+
+ public synchronized boolean containsAll(Collection<?> c) {
+ return q.containsAll(c);
+ }
+
+ public synchronized boolean isEmpty() {
+ return q.isEmpty();
+ }
+
+ public synchronized Iterator<E> iterator() {
+ return q.iterator();
+ }
+
+ public synchronized boolean remove(Object o) {
+ return q.remove(o);
+ }
+
+ public synchronized boolean removeAll(Collection<?> c) {
+ return q.removeAll(c);
+ }
+
+ public synchronized boolean retainAll(Collection<?> c) {
+ return q.retainAll(c);
+ }
+
+ public synchronized int size() {
+ return q.size();
+ }
+
+ public synchronized Object[] toArray() {
+ return q.toArray();
+ }
+
+ public synchronized <T> T[] toArray(T[] a) {
+ return q.toArray(a);
+ }
+
+ @Override
+ public synchronized boolean equals(Object obj) {
+ return q.equals(obj);
+ }
+
+ @Override
+ public synchronized int hashCode() {
+ return q.hashCode();
+ }
+
+ @Override
+ public synchronized String toString() {
+ return q.toString();
+ }
+
+ public synchronized E element(int index) {
+ return q.element(index);
+ }
+
+ public synchronized void addListener(IoQueueListener<? super E> listener) {
+ q.addListener(listener);
+ }
+
+ public synchronized void removeListener(IoQueueListener<? super E> listener) {
+ q.removeListener(listener);
+ }
+}
\ No newline at end of file
Propchange: mina/branches/buffer/core/src/main/java/org/apache/mina/queue/SynchronizedIoQueue.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: mina/branches/buffer/core/src/main/java/org/apache/mina/queue/SynchronizedIoQueue.java
------------------------------------------------------------------------------
svn:keywords = Rev Date