org.modeshape.graph.request
Class CompositeRequestChannel

java.lang.Object
  org.modeshape.graph.request.CompositeRequestChannel

public class CompositeRequestChannel
extends Object

A channel for Request objects that can be submitted to a consumer (typically a RequestProcessor or RepositoryConnection) while allowing the channel owner to continue adding more Request objects into the channel.

The owner of this channel is responsible for starting the processing using one of the two start(...) methods, adding CompositeRequestChannel.ChannelCompositeRequests via add(...) methods, closing the channel when there are no more requests to be added, and finally awaiting until all of the submitted requests have been processed. Note that the owner can optionally pre-fill the channel with Request objects before calling start(...).

The consumer will be handed a CompositeRequestChannel.ChannelCompositeRequest, and should use the CompositeRequestChannel.ChannelCompositeRequest.iterator() method to obtain an Iterator<Request>. The Iterator.hasNext() method will block until there is another Request available in the channel, or until the channel is closed (at which point Iterator.hasNext() will return false. (Notice that the Iterator.next() method will also block if it is not preceeded by a Iterator.hasNext(), but will throw a NoSuchElementException when there are no more Request objects and the channel is closed.)

Because the CompositeRequest's iterator will block, the consumer will block while processing the request. Therefore, this channel submits the CompositeRequest to the consumer asynchronously, via an ExecutorService supplied in one of the two start methods.

Nested Class Summary

protected class	CompositeRequestChannel.ChannelCompositeRequest
protected static class	CompositeRequestChannel.LastRequest A psuedo Request that is used by the CompositeRequestChannel to insert into a request queue so that the queue's iterator knows when there are no more requests to process.

Field Summary

protected LinkedList<Request>	allRequests The list of all requests that are or have been processed as part of this channel
protected AtomicBoolean	closed Flag that defines whether the channel has processed all requests
protected CompositeRequest	composite The CompositeRequest that is submitted to the underlying processor
protected Throwable	compositeError
protected Future<String>	future The Future that is submitted to the ExecutorService to do the processing, which is used to await() until the processing is completed or cancel the work
protected String	sourceName

Constructor Summary

CompositeRequestChannel(String sourceName)
Create a new channel with the supplied channel name.

CompositeRequestChannel(String sourceName, boolean keepRequests)
Create a new channel with the supplied channel name.

Method Summary

void	add(Request request) Add the request to this channel for asynchronous processing.
CountDownLatch	add(Request request, CountDownLatch latch) Add the request to this channel for asynchronous processing, and supply a count-down latch that should be decremented when this request is completed.
void	addAndAwait(Request request) Add the request to this channel for processing, but wait to return until the request has been processed.
List<Request>	allRequests() Get all the requests that were submitted to this queue.
void	await() Await until this channel has completed.
void	cancel(boolean mayInterruptIfRunning) Cancel this forked channel, stopping work as soon as possible.
void	close() Mark this source as having no more requests to process.
protected Iterator<Request>	createIterator() Utility method to create an iterator over the requests in this channel.
boolean	isClosed() Return whether this channel has been closed.
boolean	isComplete() Return whether this channel has completed all of its work.
boolean	isStarted() Return whether this channel has been started .
String	sourceName() Get the name of the source that this channel uses.
void	start(ExecutorService executor, ExecutionContext context, RepositoryConnectionFactory connectionFactory) Begins processing any requests that have been added to this channel.
void	start(ExecutorService executor, RequestProcessor processor, boolean closeProcessorWhenCompleted) Begins processing any requests that have been added to this channel.

Methods inherited from class java.lang.Object

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

Field Detail

sourceName

protected final String sourceName

allRequests

protected final LinkedList<Request> allRequests

The list of all requests that are or have been processed as part of this channel

composite

protected final CompositeRequest composite

The CompositeRequest that is submitted to the underlying processor

future

protected Future<String> future

The Future that is submitted to the ExecutorService to do the processing, which is used to await() until the processing is completed or cancel the work

closed

protected final AtomicBoolean closed

Flag that defines whether the channel has processed all requests

compositeError

protected Throwable compositeError

Constructor Detail

CompositeRequestChannel

public CompositeRequestChannel(String sourceName)

Create a new channel with the supplied channel name.

Parameters:

sourceName - the name of the repository source used to execute this channel's requests; may not be null or empty

CompositeRequestChannel

public CompositeRequestChannel(String sourceName,
                               boolean keepRequests)

Create a new channel with the supplied channel name.

Parameters:

sourceName - the name of the repository source used to execute this channel's requests; may not be null or empty

keepRequests - true if this channel should keep all of the requests that it processes, or false otherwise

Method Detail

createIterator

protected Iterator<Request> createIterator()

Utility method to create an iterator over the requests in this channel. This really should be called once

Returns:

the iterator over the channels

start

public void start(ExecutorService executor,
                  ExecutionContext context,
                  RepositoryConnectionFactory connectionFactory)

Begins processing any requests that have been added to this channel. Processing is done by submitting the channel to the supplied executor.

Parameters:

executor - the executor that is to do the work; may not be null

context - the execution context in which the work is to be performed; may not be null

connectionFactory - the connection factory that should be used to create connections; may not be null

Throws:

IllegalStateException - if this channel has already been started

start

public void start(ExecutorService executor,
                  RequestProcessor processor,
                  boolean closeProcessorWhenCompleted)

Begins processing any requests that have been added to this channel. Processing is done by submitting the channel to the supplied executor.

Parameters:

executor - the executor that is to do the work; may not be null

processor - the request processor that will be used to execute the requests; may not be null

closeProcessorWhenCompleted - true if this method should call RequestProcessor.close() when the channel is completed, or false if the caller is responsible for doing this

Throws:

IllegalStateException - if this channel has already been started

add

public void add(Request request)

Add the request to this channel for asynchronous processing.

Parameters:

request - the request to be submitted; may not be null

Throws:

IllegalStateException - if this channel has already been closed

add

public CountDownLatch add(Request request,
                          CountDownLatch latch)

Add the request to this channel for asynchronous processing, and supply a count-down latch that should be decremented when this request is completed.

Parameters:

request - the request to be submitted; may not be null

latch - the count-down latch that should be decremented when request has been completed; may not be null

Returns:

the same latch that was supplied, for method chaining purposes; never null

Throws:

IllegalStateException - if this channel has already been closed

addAndAwait

public void addAndAwait(Request request)
                 throws InterruptedException

Add the request to this channel for processing, but wait to return until the request has been processed.

Parameters:

request - the request to be submitted; may not be null

Throws:

InterruptedException - if the current thread is interrupted while waiting

close

public void close()

Mark this source as having no more requests to process.

isClosed

public boolean isClosed()

Return whether this channel has been closed.

Returns:

true if the channel was marked as done, or false otherwise

cancel

public void cancel(boolean mayInterruptIfRunning)

Cancel this forked channel, stopping work as soon as possible. If the channel has not yet been started, this method

Parameters:

mayInterruptIfRunning - true if the channel is still being worked on, and the thread on which its being worked on may be interrupted, or false if the channel should be allowed to finish if it is already in work.

isStarted

public boolean isStarted()

Return whether this channel has been started .

Returns:

true if this channel was started, or false otherwise

isComplete

public boolean isComplete()

Return whether this channel has completed all of its work.

Returns:

true if the channel was started and is complete, or false otherwise

await

public void await()
           throws ExecutionException,
                  InterruptedException,
                  CancellationException

Await until this channel has completed.

Throws:

CancellationException - if the channel was cancelled

ExecutionException - if the channel execution threw an exception

InterruptedException - if the current thread is interrupted while waiting

allRequests

public List<Request> allRequests()

Get all the requests that were submitted to this queue. The resulting list is the actual list that is appended when requests are added, and may change until the channel is closed.

Returns:

all of the requests that were submitted to this channel; never null

sourceName

public String sourceName()

Get the name of the source that this channel uses.

Returns:

the source name; never null