com.bigdata.service.proxy
Class ClientAsynchronousIterator<E>

java.lang.Object
  com.bigdata.service.proxy.ClientAsynchronousIterator<E>

All Implemented Interfaces:

IAsynchronousIterator<E>, ICloseableIterator<E>, Serializable, Iterator<E>

public class ClientAsynchronousIterator<E>
extends Object
implements IAsynchronousIterator<E>, Serializable

Wraps an RemoteAsynchronousIterator so that it looks like an IAsynchronousIterator again.

This implementation uses the caller's ExecutorService to run a task that migrates data from the remote IAsynchronousIterator into a local BlockingBuffer. The client implementation uses purely local operations to test the contents of the local BlockingBuffer. This means that operations like hasNext(long, TimeUnit) may be used with short timeouts without possibility of blocking IO.

Note: This class is NOT thread-safe. There should be only a single process draining the iterator.

Note: In order to conserve resources, the caller is advised to use start(ExecutorService) before making any other requests. This will allow the caller to specify an ExecutorService that will be tasked with buffering elements from the RemoteAsynchronousIterator. When start(ExecutorService) is not invoked explicitly, a new Thread will be allocated. Typically, an ExecutorService can be much more efficient than creating a new Thread.

Version:

$Id: ClientAsynchronousIterator.java 2265 2009-10-26 12:51:06Z thompsonbry $

Author:

Bryan Thompson

See Also:

Serialized Form

TODO:

Note that we only use three methods to implement ClientAsynchronousIterator - hasNext(timeout,unit), isExhausted(), and next(). The rest of the RemoteAsynchronousIterator API does not need to be implemented.

Field Summary

protected static boolean	DEBUG
protected static boolean	INFO
protected static org.apache.log4j.Logger	log
protected static long	timeout Timeout for the ReaderTask when it invokes RemoteAsynchronousIterator.hasNext(long, TimeUnit).
protected static boolean	trace Enables low-level trace of the ReaderTask.
protected static TimeUnit	unit The units for timeout.

Constructor Summary

ClientAsynchronousIterator(RemoteAsynchronousIterator<E> remoteIterator, int capacity)

Method Summary

void	close() Notes that the iterator is closed and hence may no longer be read.
boolean	hasNext()
boolean	hasNext(long timeout, TimeUnit unit) Return true iff there is at least one element that can be visited.
boolean	isExhausted() Return true iff this iterator will not visit any more elements (non-blocking).
E	next()
E	next(long timeout, TimeUnit unit) Waits up to the timeout to return the next element.
void	remove() Operation is not supported.
protected void	start() Start the ReaderTask iff it is not already running.
void	start(ExecutorService executorService) Start the ReaderTask that will populate the local buffer with elements from the remote iterator.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

log

protected static final transient org.apache.log4j.Logger log

INFO

protected static final transient boolean INFO

DEBUG

protected static final transient boolean DEBUG

timeout

protected static final transient long timeout

Timeout for the ReaderTask when it invokes RemoteAsynchronousIterator.hasNext(long, TimeUnit).

Note: A timeout is used instead of RemoteAsynchronousIterator.hasNext() in order to avoid blocking during an RMI call on a remote thread.

See Also:

Constant Field Values

unit

protected static final transient TimeUnit unit

The units for timeout.

trace

protected static final transient boolean trace

Enables low-level trace of the ReaderTask.

See Also:

Constant Field Values

Constructor Detail

ClientAsynchronousIterator

public ClientAsynchronousIterator(RemoteAsynchronousIterator<E> remoteIterator,
                                  int capacity)

Parameters:

remoteIterator - A proxy for the remote iterator from which the elements are being read.

capacity - The capacity of the internal IBlockingBuffer.

Method Detail

start

public void start(ExecutorService executorService)

Start the ReaderTask that will populate the local buffer with elements from the remote iterator.

When start(ExecutorService) is not invoked explicitly, a new Thread will be allocated and the ReaderTask will be run on that Thread. Typically, an ExecutorService can be much more efficient than creating a new Thread. Therefore the caller is encourged to break encapsulation and specify the ExecutorService on which the ReaderTask will run directly.

Throws:

IllegalStateException - if the ReaderTask is already running.

start

protected final void start()

Start the ReaderTask iff it is not already running.

close

public void close()

Description copied from interface: IAsynchronousIterator

Notes that the iterator is closed and hence may no longer be read. It is safe to invoke this method even if the iterator is closed. However, if the producer is still running then it will be canceled (interrupted) in order to prevent the IBlockingBuffer from filling up and deadlocking. For this reason, IAsynchronousIterator.close() has consequences NOT entailed by ICloseableIterator.

Note: Depending on the semantics of the producer, it MAY choose to treat an interrupt() as normal (but eager) termination. For example, rule execution treats an interrupt() as normal (but eager) termination with the consequence that queries may be safely interrupted once some limit has been satisfied. However, the preferred way to treat LIMIT is using IRule with an IQueryOptions that specifies a LIMIT.

Specified by:

close in interface IAsynchronousIterator<E>

Specified by:

close in interface ICloseableIterator<E>

hasNext

public boolean hasNext(long timeout,
                       TimeUnit unit)
                throws InterruptedException

Description copied from interface: IAsynchronousIterator

Return true iff there is at least one element that can be visited. If the buffer is empty then this will block until: (a) an element appears in the buffer; (b) the buffer is BlockingBuffer.close()ed; or (c) the timeout expires.

Note that a false return DOES NOT signify that the iterator is exhausted. However, if you specify an infinite timeout using Long.MAX_VALUE TimeUnit.SECONDS then you MAY safely interpret a false return as an indication that the iterator is exhausted.

Specified by:

hasNext in interface IAsynchronousIterator<E>

Parameters:

timeout - The length of time that the method may block awaiting an element to appear.

unit - The units in which the timeout is expressed.

Returns:

true iff there is an element available.

Throws:

InterruptedException - if the current thread is interrupted while waiting another element.

isExhausted

public boolean isExhausted()

Description copied from interface: IAsynchronousIterator

Return true iff this iterator will not visit any more elements (non-blocking).

Specified by:

isExhausted in interface IAsynchronousIterator<E>

Returns:

true iff the iterator is known to be exhausted without blocking. A false return does NOT imply that the iterator will visit more elements, only that it MIGHT visit more elements.

next

public E next(long timeout,
              TimeUnit unit)
       throws InterruptedException

Description copied from interface: IAsynchronousIterator

Waits up to the timeout to return the next element. When the generic type of the elements in the buffer is an array type and there is more than one element available within the specified time, then multiple elements in the buffer MAY be combined into a single element and that element returned to the caller.

Specified by:

next in interface IAsynchronousIterator<E>

Parameters:

timeout - The timeout (overrides the chunkTimeout specified to the BlockingBuffer ctor).

unit - The unit in which the timeout is expressed.

Returns:

The element -or- null iff the timeout was exceeded before IAsynchronousIterator.hasNext(long, TimeUnit) returned true.

In order to have a definitive indication that the iterator is exhausted you need to invoke Iterator.hasNext() without a timeout (blocking) or IAsynchronousIterator.isExhausted() (non-blocking).

Throws:

InterruptedException - if the current thread is interrupted while waiting for another element.

hasNext

public boolean hasNext()

Specified by:

hasNext in interface Iterator<E>

next

public E next()

Specified by:

next in interface Iterator<E>

remove

public void remove()

Operation is not supported.

Specified by:

remove in interface Iterator<E>

Throws:

UnsupportedOperationException - always.