12:06 a.m.
Author: david.lloyd(a)jboss.com
Date: 2009-02-26 18:06:03 -0500 (Thu, 26 Feb 2009)
New Revision: 4863
Added:
remoting3/trunk/api/src/main/java/org/jboss/remoting3/stream/ObjectPipe.java
Modified:
remoting3/trunk/api/src/main/java/org/jboss/remoting3/stream/ObjectSource.java
Log:
Clarify ObjectSource semantics. Implement an object pipe for connecting a remote
ObjectSink to a local ObjectSource.
Added: remoting3/trunk/api/src/main/java/org/jboss/remoting3/stream/ObjectPipe.java
===================================================================
--- remoting3/trunk/api/src/main/java/org/jboss/remoting3/stream/ObjectPipe.java
(rev 0)
+++
remoting3/trunk/api/src/main/java/org/jboss/remoting3/stream/ObjectPipe.java 2009-02-26
23:06:03 UTC (rev 4863)
@@ -0,0 +1,211 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2008, JBoss Inc., and individual contributors as indicated
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.jboss.remoting3.stream;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.io.EOFException;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.locks.Condition;
+import java.util.Queue;
+import java.util.LinkedList;
+import java.util.NoSuchElementException;
+
+/**
+ * A pipe for objects. Typically, data is written to the sink side of the pipe from one
thread while being read from
+ * the source side in another thread. Object pipes are useful in the case that you send
an {@link
+ * org.jboss.remoting3.stream.ObjectSink ObjectSink} to a remote system in a request, but
you want to read the objects
+ * from an {@link org.jboss.remoting3.stream.ObjectSource ObjectSource}.
+ */
+public final class ObjectPipe<T> {
+
+ private final Lock queueLock = new ReentrantLock();
+ // signal on write, await on read
+ private final Condition writeCondition = queueLock.newCondition();
+ // signal on read, await on write
+ private final Condition readCondition = queueLock.newCondition();
+
+ private final Queue<T> queue = new LinkedList<T>();
+
+ private final Source source = new Source();
+ private final Sink sink = new Sink();
+ private final int max;
+ private boolean open = true;
+
+ /**
+ * Create an object pipe with the given maximum buffer size.
+ *
+ * @param max the maximum number of buffered objects
+ */
+ public ObjectPipe(int max) {
+ this.max = max;
+ }
+
+ /**
+ * Get the source end of the pipe, from which objects may be read.
+ *
+ * @return the source end
+ */
+ public ObjectSource<T> getSource() {
+ return source;
+ }
+
+ /**
+ * Get the sink end of the pipe, to which objects may be written.
+ *
+ * @return the sink end
+ */
+ public ObjectSink<T> getSink() {
+ return sink;
+ }
+
+ private class Source implements ObjectSource<T> {
+
+ public boolean hasNext() throws IOException {
+ final Lock queueLock = ObjectPipe.this.queueLock;
+ final Condition writeCondition = ObjectPipe.this.writeCondition;
+ final Queue<T> queue = ObjectPipe.this.queue;
+ try {
+ queueLock.lockInterruptibly();
+ try {
+ while (open && queue.isEmpty()) {
+ writeCondition.await();
+ }
+ return open || !queue.isEmpty();
+ } finally {
+ queueLock.unlock();
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new InterruptedIOException("hasNext() was interrupted");
+ }
+ }
+
+ public T next() throws IOException {
+ final Lock queueLock = ObjectPipe.this.queueLock;
+ final Queue<T> queue = ObjectPipe.this.queue;
+ try {
+ queueLock.lockInterruptibly();
+ try {
+ final T t = queue.poll();
+ if (t == null) {
+ if (open) {
+ throw new NoSuchElementException();
+ } else {
+ throw new EOFException("EOF on next()");
+ }
+ }
+ readCondition.signal();
+ return t;
+ } finally {
+ queueLock.unlock();
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new InterruptedIOException("hasNext() was interrupted");
+ }
+ }
+
+ /**
+ * Closing the reader breaks everything. All unread items are discarded, all
waiters are woken up.
+ */
+ public void close() {
+ final Lock queueLock = ObjectPipe.this.queueLock;
+ queueLock.lock();
+ try {
+ if (open) {
+ open = false;
+ queue.clear();
+ writeCondition.signalAll();
+ readCondition.signalAll();
+ }
+ } finally {
+ queueLock.unlock();
+ }
+ }
+
+ protected void finalize() throws Throwable {
+ close();
+ super.finalize();
+ }
+ }
+
+ private class Sink implements ObjectSink<T> {
+
+ public void accept(final T instance) throws IOException {
+ final int max = ObjectPipe.this.max;
+ final Queue<T> queue = ObjectPipe.this.queue;
+ final Lock queueLock = ObjectPipe.this.queueLock;
+ try {
+ queueLock.lockInterruptibly();
+ try {
+ while (open && queue.size() == max) {
+ readCondition.await();
+ }
+ if (!open) {
+ throw new EOFException("pipe closed");
+ }
+ } finally {
+ queueLock.unlock();
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new InterruptedIOException("accept(T) was interrupted");
+ }
+ }
+
+ /**
+ * Do nothing since nothing can be done - clearing our internal state depends on
the reader.
+ */
+ public void flush() {
+ }
+
+ /**
+ * Closing the writer just clears the open flag and notifies waiters.
+ */
+ public void close() {
+ final Lock queueLock = ObjectPipe.this.queueLock;
+ queueLock.lock();
+ try {
+ if (!open) return;
+ open = false;
+ // readers might be waiting
+ if (queue.isEmpty()) {
+ readCondition.signalAll();
+ } else {
+ readCondition.signal();
+ }
+ // other writers might also be waiting - they should be killed
+ writeCondition.signalAll();
+ } finally {
+ queueLock.unlock();
+ }
+ }
+
+ protected void finalize() throws Throwable {
+ close();
+ super.finalize();
+ }
+ }
+}
Modified: remoting3/trunk/api/src/main/java/org/jboss/remoting3/stream/ObjectSource.java
===================================================================
---
remoting3/trunk/api/src/main/java/org/jboss/remoting3/stream/ObjectSource.java 2009-02-26
21:56:19 UTC (rev 4862)
+++
remoting3/trunk/api/src/main/java/org/jboss/remoting3/stream/ObjectSource.java 2009-02-26
23:06:03 UTC (rev 4863)
@@ -2,6 +2,7 @@
import java.io.Closeable;
import java.io.IOException;
+import java.util.NoSuchElementException;
/**
* A streaming source for objects.
@@ -12,29 +13,31 @@
/**
* Indicate whether there are more objects to retrieve. If this method returns
{@code true}, an object is
- * guaranteed to be available. If this method returns {@code false}, there is no
object available; however, an
- * object may become available at a later time, depending on the implementation.
+ * guaranteed to be available. If this method returns {@code false}, the end of
stream has been reached.
* <p/>
* If this method returns {@code true}, it will continue to return {@code true} on
every subsequent invocation until
- * the next object is pulled using the {@code next()} method, or until the object
source is closed.
+ * the next object is pulled using the {@code next()} method, or until the object
source is closed. This method
+ * may block until the presence of the next object in the stream has been
ascertained.
*
* @return {@code true} if there are more objects in this stream
*/
boolean hasNext() throws IOException;
/**
- * Get the next object in the stream.
+ * Get the next object in the stream. The {@code hasNext()} method should be called
before this method is called
+ * to avoid receiving a {@code NoSuchElementException}.
*
* @return the next object
*
- * @throws IOException if the stream can no longer be read
+ * @throws NoSuchElementException if no object is available
+ * @throws IOException if an I/O error occurs
*/
- T next() throws IOException;
+ T next() throws NoSuchElementException, IOException;
/**
* Close the stream. No more objects may be read from this stream after it has been
closed.
*
- * @throws IOException if an error occurs
+ * @throws IOException if an I/O error occurs
*/
void close() throws IOException;
}