org.modeshape.graph.request
Class CompositeRequestChannel

java.lang.Object
  extended by 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


Copyright © 2008-2012 JBoss, a division of Red Hat. All Rights Reserved.