commons-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From nutty...@apache.org
Subject svn commit: r462575 - in /jakarta/commons/sandbox/pipeline/trunk/src: main/java/org/apache/commons/pipeline/driver/ test/java/org/apache/commons/pipeline/driver/
Date Tue, 10 Oct 2006 21:21:03 GMT
Author: nuttycom
Date: Tue Oct 10 14:21:02 2006
New Revision: 462575

URL: http://svn.apache.org/viewvc?view=rev&rev=462575
Log:
Initial import. Stage implementations contributed by Steve Christensen.

Added:
    jakarta/commons/sandbox/pipeline/trunk/src/main/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriver.java
  (with props)
    jakarta/commons/sandbox/pipeline/trunk/src/main/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriverFactory.java
  (with props)
    jakarta/commons/sandbox/pipeline/trunk/src/test/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriverTest.java
  (with props)

Added: jakarta/commons/sandbox/pipeline/trunk/src/main/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriver.java
URL: http://svn.apache.org/viewvc/jakarta/commons/sandbox/pipeline/trunk/src/main/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriver.java?view=auto&rev=462575
==============================================================================
--- jakarta/commons/sandbox/pipeline/trunk/src/main/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriver.java
(added)
+++ jakarta/commons/sandbox/pipeline/trunk/src/main/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriver.java
Tue Oct 10 14:21:02 2006
@@ -0,0 +1,372 @@
+/*
+ * Copyright 2004 The Apache Software Foundation
+ *
+ * Licensed 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.commons.pipeline.driver;
+
+import java.lang.Thread.UncaughtExceptionHandler;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.commons.pipeline.Feeder;
+import org.apache.commons.pipeline.StageDriver;
+import org.apache.commons.pipeline.Stage;
+import org.apache.commons.pipeline.StageContext;
+import org.apache.commons.pipeline.StageException;
+import org.apache.commons.pipeline.driver.AbstractStageDriver;
+import org.apache.commons.pipeline.driver.FaultTolerance;
+
+import static org.apache.commons.pipeline.StageDriver.State.*;
+import static org.apache.commons.pipeline.driver.FaultTolerance.*;
+
+/**
+ * This is a very simple implementation of a AbstractStageDriver which spawns
+ * a single thread to process a stage.
+ */
+public class ThreadPoolStageDriver extends AbstractStageDriver {
+    private final Log log = LogFactory.getLog(ThreadPoolStageDriver.class);
+    
+    //wait timeout to ensure deadlock cannot occur on thread termination
+    private long timeout;
+    
+    //flag describing whether or not the driver is fault tolerant
+    private FaultTolerance faultTolerance = FaultTolerance.NONE;
+    
+    //signal telling threads to start polling queue
+    final private CountDownLatch startSignal;
+    
+    //signal threads use to tell driver they have finished
+    final private CountDownLatch doneSignal;
+    
+    // number of threads polling queue
+    private int numThreads = 1;
+    
+    //queue to hold data to be processed
+    private BlockingQueue queue;
+    
+    //current state of thread processing
+    private volatile State currentState = State.STOPPED;
+    
+    //feeder used to feed data to this stage's queue
+    private final Feeder feeder = new Feeder() {
+        public void feed(Object obj) {
+            if (log.isDebugEnabled()) log.debug(obj + " is being fed to stage " + stage
+                    + " (" + ThreadPoolStageDriver.this.queue.remainingCapacity() + " available
slots in queue)");
+            try {
+                ThreadPoolStageDriver.this.queue.put(obj);
+            } catch (InterruptedException e) {
+                throw new IllegalStateException("Unexpected interrupt while waiting for space
to become available for object "
+                        + obj + " in queue for stage " + stage, e);
+            }
+            synchronized(ThreadPoolStageDriver.this) {
+                ThreadPoolStageDriver.this.notifyAll();
+            }
+        }
+    };
+    
+    /**
+     * Creates a new ThreadPoolStageDriver with the specified thread wait
+     * timeout and fault tolerance values.
+     * @param stage The stage that the driver will run
+     * @param context the context in which to run the stage
+     * @param queue The object queue to use for storing objects prior to processing. The
+     * default is {@link LinkedBlockingQueue}
+     * @param timeout The amount of time, in milliseconds, that the worker thread
+     * will wait before checking the processing state if no objects are available
+     * in the thread's queue.
+     * @param faultTolerance Flag determining the behavior of the driver when
+     * an error is encountered in execution of {@link Stage#process(Object)}.
+     * If this is set to false, any exception thrown during {@link Stage#process(Object)}
+     * will cause the worker thread to halt without executing {@link Stage#postprocess()}
+     * ({@link Stage#release()} will be called.)
+     * @param numThreads Number of threads that will be simultaneously reading from queue
+     */
+    public ThreadPoolStageDriver(Stage stage, StageContext context,
+            BlockingQueue queue,
+            long timeout,
+            FaultTolerance faultTolerance,
+            int numThreads) {
+        super(stage, context);
+        this.numThreads = numThreads;
+        
+        this.startSignal = new CountDownLatch(1);
+        this.doneSignal = new CountDownLatch(this.numThreads);
+        
+        this.queue = queue;
+        this.timeout = timeout;
+        this.faultTolerance = faultTolerance;
+    }
+    
+    /**
+     * Return the Feeder used to feed data to the queue of objects to be processed.
+     * @return The feeder for objects processed by this driver's stage.
+     */
+    public Feeder getFeeder() {
+        return this.feeder;
+    }
+    
+    /**
+     * Start the processing of the stage. Creates threads to poll items
+     * from queue.
+     * @throws org.apache.commons.pipeline.StageException Thrown if the driver is in an illegal
state during startup
+     */
+    public synchronized void start() throws StageException {
+        if (this.currentState == STOPPED) {
+            setState(STARTED);
+            
+            if (log.isDebugEnabled()) log.debug("Preprocessing stage " + stage + "...");
+            stage.preprocess();
+            if (log.isDebugEnabled()) log.debug("Preprocessing for stage " + stage + " complete.");
+            
+            log.debug("Starting worker threads for stage " + stage + ".");
+            
+            for (int i=0;i<this.numThreads;i++) {
+                new LatchWorkerThread(i).start();
+            }
+            
+            // let threads know they can start
+            testAndSetState(STARTED, RUNNING);
+            startSignal.countDown();
+            
+            log.debug("Worker threads for stage " + stage + " started.");
+            
+            //wait to ensure that the stage starts up correctly
+            try {
+                while ( !(this.currentState == RUNNING || this.currentState == ERROR) ) this.wait();
+            } catch (InterruptedException e) {
+                throw new StageException(this.getStage(), "Worker thread unexpectedly interrupted
while waiting for thread startup.", e);
+            }
+        } else {
+            throw new IllegalStateException("Attempt to start driver in state " + this.currentState);
+        }
+    }
+    
+    /**
+     * Causes processing to shut down gracefully. Waits until all worker threads
+     * have completed.
+     * @throws org.apache.commons.pipeline.StageException Thrown if the driver is in an illegal
state for shutdown.
+     */
+    public synchronized void finish() throws StageException {
+        if (currentState == STOPPED) {
+            throw new IllegalStateException("The driver is not currently running.");
+        }
+        
+        try {
+            while ( !(this.currentState == RUNNING || this.currentState == ERROR) ) this.wait();
+            
+            //ask the worker threads to shut down
+            testAndSetState(RUNNING, STOP_REQUESTED);
+            
+            if (log.isDebugEnabled()) log.debug("Waiting for worker threads to stop for stage
" + stage + ".");
+            doneSignal.await();
+            
+            if (log.isDebugEnabled()) log.debug("Postprocessing stage " + stage + "...");
+            ThreadPoolStageDriver.this.stage.postprocess();
+            if (log.isDebugEnabled()) log.debug("Postprocessing for stage " + stage + " complete.");
+            
+            //do not transition into finished state if an error has occurred
+            testAndSetState(STOP_REQUESTED, FINISHED);
+            
+            while ( !(this.currentState == FINISHED || this.currentState == ERROR) ) this.wait();
+            
+            log.debug("Worker threads for stage " + stage + " halted");
+        } catch (StageException e) {
+            log.error("An error occurred during postprocess for stage " + stage , e);
+            recordFatalError(e);
+            setState(ERROR);
+        } catch (InterruptedException e) {
+            throw new StageException(this.getStage(), "StageDriver unexpectedly interrupted
while waiting for shutdown of worker threads.", e);
+        } finally {
+            if (log.isDebugEnabled()) log.debug("Releasing resources for stage " + stage
+ "...");
+            stage.release();
+            if (log.isDebugEnabled()) log.debug("Stage " + stage + " released.");
+        }
+        
+        setState(STOPPED);
+    }
+    
+    /**
+     * Return the current state of stage processing.
+     * @return the current state of processing
+     */
+    public StageDriver.State getState() {
+        return this.currentState;
+    }
+    
+    /**
+     * Atomically tests to determine whether or not the driver is in the one of
+     * the specified states.
+     */
+    private synchronized boolean isInState(State... states) {
+        for (State state : states) if (state == currentState) return true;
+        return false;
+    }
+    
+    /**
+     * Set the current state of stage processing and notify any listeners
+     * that may be waiting on a state change.
+     */
+    private synchronized void setState(State nextState) {
+        if (log.isDebugEnabled()) log.debug("State change for " + stage + ": " + this.currentState
+ " -> " + nextState);
+        this.currentState = nextState;
+        this.notifyAll();
+    }
+    
+    /**
+     * This method performs an atomic conditional state transition change
+     * to the value specified by the nextState parameter if and only if the
+     * current state is equal to the test state.
+     */
+    private synchronized boolean testAndSetState(State testState, State nextState) {
+        if (currentState == testState) {
+            setState(nextState);
+            return true;
+        } else {
+            return false;
+        }
+    }
+    
+    /**
+     * Sets the failure tolerance flag for the worker thread. If faultTolerance
+     * is set to CHECKED, {@link StageException StageException}s thrown by
+     * the {@link Stage#process(Object)} method will not interrupt queue
+     * processing, but will simply be logged with a severity of ERROR.
+     * If faultTolerance is set to ALL, runtime exceptions will also be
+     * logged and otherwise ignored.
+     * @param faultTolerance the flag value
+     */
+    public final void setFaultTolerance(String faultTolerance) {
+        this.faultTolerance = FaultTolerance.valueOf(faultTolerance);
+    }
+    
+    /**
+     * Sets the failure tolerance flag for the worker thread. If faultTolerance
+     * is set to CHECKED, {@link StageException StageException}s thrown by
+     * the {@link Stage#process(Object)} method will not interrupt queue
+     * processing, but will simply be logged with a severity of ERROR.
+     * If faultTolerance is set to ALL, runtime exceptions will also be
+     * logged and otherwise ignored.
+     * @param faultTolerance the flag value
+     */
+    public final void setFaultTolerance(FaultTolerance faultTolerance) {
+        this.faultTolerance = faultTolerance;
+    }
+    
+    /**
+     * Getter for property faultTolerant.
+     * @return Value of property faultTolerant.
+     */
+    public FaultTolerance getFaultTolerance() {
+        return this.faultTolerance;
+    }
+    
+    /*********************************
+     * WORKER THREAD IMPLEMENTATIONS *
+     *********************************/
+    private UncaughtExceptionHandler workerThreadExceptionHandler = new UncaughtExceptionHandler()
{
+        public void uncaughtException(Thread t, Throwable e) {
+            setState(ERROR);
+            recordFatalError(e);
+            log.error("Uncaught exception in stage " + stage, e);
+        }
+    };
+    
+    /**
+     * This worker thread removes and processes data objects from the incoming
+     * synchronize queue. It calls the Stage's process() method to process data
+     * from the queue. This loop runs until State has changed to
+     * STOP_REQUESTED. To break the loop the calling code must run the writer's
+     * finish() method to set the running property to false.
+     *
+     * @throws StageException if an error is encountered during data processing
+     * and faultTolerant is set to false.
+     */
+    private class LatchWorkerThread extends Thread {
+        final int threadID;
+        
+        LatchWorkerThread(int threadID) {
+            this.setUncaughtExceptionHandler(workerThreadExceptionHandler);
+            this.threadID = threadID;
+        }        
+        
+        public final void run() {
+            try {
+                ThreadPoolStageDriver.this.startSignal.await();
+                //do not transition into running state if an error has occurred or a stop
requested
+                running: while (currentState != ERROR) {
+                    try {
+                        Object obj = queue.poll(timeout, TimeUnit.MILLISECONDS);
+                        if (obj == null) {
+                            if (currentState == STOP_REQUESTED) break running;
+                        } else {
+                            try {
+                                stage.process(obj);
+                            } catch (StageException e) {
+                                recordProcessingException(obj, e);
+                                if (faultTolerance == NONE) throw e;
+                            } catch (RuntimeException e) {
+                                recordProcessingException(obj, e);
+                                if (faultTolerance == CHECKED || faultTolerance == NONE)
throw e;
+                            }
+                        }
+                    } catch (InterruptedException e) {
+                        throw new RuntimeException("Worker thread " + this.threadID + " unexpectedly
interrupted while waiting on data for stage " + stage, e);
+                    }
+                }
+                if (log.isDebugEnabled()) log.debug("Stage " + stage + " (threadID: " + this.threadID
+ ") exited running state.");
+                
+            } catch (StageException e) {
+                log.error("An error occurred in the stage " + stage + " (threadID: " + this.threadID
+ ")", e);
+                recordFatalError(e);
+                setState(ERROR);
+            } catch (InterruptedException e) {
+                log.error("Stage " + stage + " (threadID: " + threadID + ") interrupted while
waiting for barrier",e);
+                recordFatalError(e);
+                setState(ERROR);
+            } finally {
+                doneSignal.countDown();
+                synchronized (ThreadPoolStageDriver.this) {
+                    ThreadPoolStageDriver.this.notifyAll();
+                }
+            }
+        }
+    }
+    
+    /**
+     * Get the size of the queue used by this StageDriver.
+     * @return the queue capacity
+     */
+    public int getQueueSize() {
+        return this.queue.size() + this.queue.remainingCapacity();
+    }
+    
+    /**
+     * Get the timeout value (in milliseconds) used by this StageDriver on
+     * thread termination.
+     * @return the timeout setting in milliseconds
+     */
+    public long getTimeout() {
+        return this.timeout;
+    }
+    
+    /**
+     * Returns the number of threads allocated to the thread pool.
+     */
+    public int getNumThreads() {
+        return numThreads;
+    }
+}

Propchange: jakarta/commons/sandbox/pipeline/trunk/src/main/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriver.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: jakarta/commons/sandbox/pipeline/trunk/src/main/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriverFactory.java
URL: http://svn.apache.org/viewvc/jakarta/commons/sandbox/pipeline/trunk/src/main/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriverFactory.java?view=auto&rev=462575
==============================================================================
--- jakarta/commons/sandbox/pipeline/trunk/src/main/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriverFactory.java
(added)
+++ jakarta/commons/sandbox/pipeline/trunk/src/main/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriverFactory.java
Tue Oct 10 14:21:02 2006
@@ -0,0 +1,146 @@
+/*
+ * Copyright 2005 The Apache Software Foundation
+ *
+ * Licensed 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.commons.pipeline.driver;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.apache.commons.pipeline.Stage;
+import org.apache.commons.pipeline.StageContext;
+import org.apache.commons.pipeline.StageDriver;
+import org.apache.commons.pipeline.StageDriverFactory;
+
+/**
+ * This factory is used to create ThreadPoolStageDriver instances configured
+ * to run specific stages.
+ */
+public class ThreadPoolStageDriverFactory implements StageDriverFactory {
+    
+    private int numThreads = 1;
+    
+    /** Creates a new instance of ThreadPoolStageDriverFactory */
+    public ThreadPoolStageDriverFactory() {
+    }
+    
+    /**
+     * Creates the new {@link ThreadPoolStageDriver} based upon the configuration
+     * of this factory instance
+     * @param stage The stage to be run by the newly created driver
+     * @param context The context in which the stage will be run
+     * @return the newly created driver
+     */
+    public StageDriver createStageDriver(Stage stage, StageContext context) {
+        try {
+            return new ThreadPoolStageDriver(stage, context, queueClass.newInstance(), timeout,
faultTolerance, numThreads);
+        } catch (Exception e) {
+            throw new IllegalStateException("Instantiation of driver failed due to illegal
factory state.", e);
+        }
+    }
+    
+    /**
+     * Holds value of property queueClass.
+     */
+    private Class<? extends BlockingQueue> queueClass = LinkedBlockingQueue.class;
+    
+    /**
+     * Getter for property queueClass.
+     * @return Value of property queueClass.
+     */
+    public Class<? extends BlockingQueue> getQueueClass() {
+        return this.queueClass;
+    }
+    
+    /**
+     * Setter for property queueClass.
+     * @param queueClass New value of property queueClass.
+     */
+    public void setQueueClass(Class<? extends BlockingQueue> queueClass) {
+        if (queueClass == null) throw new IllegalArgumentException("Queue class may not be
null.");
+        this.queueClass = queueClass;
+    }
+    
+    /**
+     * Holds value of property timeout.
+     */
+    private long timeout = 500;
+    
+    /**
+     * Timeout for wait to ensure deadlock cannot occur on thread termination.
+     * Default is 500
+     * @return Value of property timeout.
+     */
+    public long getTimeout() {
+        return this.timeout;
+    }
+    
+    /**
+     * Setter for property timeout.
+     * @param timeout New value of property timeout.
+     */
+    public void setTimeout(long timeout) {
+        this.timeout = timeout;
+    }
+    
+    /**
+     * Holds value of property faultTolerance.
+     */
+    private FaultTolerance faultTolerance = FaultTolerance.NONE;
+    
+    /**
+     * Getter for property faultTolerance. See {@link FaultTolerance} for valid values
+     * and enumation meanings.
+     * @return Value of property faultTolerance.
+     */
+    public FaultTolerance getFaultTolerance() {
+        return this.faultTolerance;
+    }
+    
+    /**
+     * Setter for property faultTolerance.
+     *
+     * @param faultTolerance New value of property faultTolerance.
+     */
+    public void setFaultTolerance(FaultTolerance faultTolerance) {
+        this.faultTolerance = faultTolerance;
+    }
+    
+    /**
+     * Convenience setter for property faultTolerance for use by Digester.
+     *
+     * @param level New value of property level ("ALL","CHECKED", or "NONE").
+     */
+    public void setFaultToleranceLevel(String level) {
+        this.faultTolerance = FaultTolerance.valueOf(level);
+    }
+    
+    /**
+     * Returns the number of threads that will be allocated to the thread
+     * pool of a driver created by this factory.
+     */
+    public int getNumThreads() {
+        return numThreads;
+    }
+    
+    /**
+     * Sets the number of threads that will be allocated to the thread
+     * pool of a driver created by this factory.
+     */
+    public void setNumThreads(int numThreads) {
+        this.numThreads = numThreads;
+    }
+    
+}

Propchange: jakarta/commons/sandbox/pipeline/trunk/src/main/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriverFactory.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: jakarta/commons/sandbox/pipeline/trunk/src/test/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriverTest.java
URL: http://svn.apache.org/viewvc/jakarta/commons/sandbox/pipeline/trunk/src/test/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriverTest.java?view=auto&rev=462575
==============================================================================
--- jakarta/commons/sandbox/pipeline/trunk/src/test/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriverTest.java
(added)
+++ jakarta/commons/sandbox/pipeline/trunk/src/test/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriverTest.java
Tue Oct 10 14:21:02 2006
@@ -0,0 +1,98 @@
+/*
+ * Copyright 2005 The Apache Software Foundation
+ *
+ * Licensed 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.commons.pipeline.driver;
+
+import junit.framework.*;
+import java.util.concurrent.LinkedBlockingQueue;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.commons.pipeline.Feeder;
+import org.apache.commons.pipeline.StageDriver.State;
+import static org.apache.commons.pipeline.StageDriver.State.*;
+
+/**
+ *
+ *
+ */
+public class ThreadPoolStageDriverTest extends AbstractStageDriverTest {
+    private Log log;
+    
+    public ThreadPoolStageDriverTest(String testName) {
+        super(testName);
+        this.log = LogFactory.getLog(ThreadPoolStageDriverTest.class);
+    }
+    
+    public static Test suite() {
+        TestSuite suite = new TestSuite(ThreadPoolStageDriverTest.class);
+        
+        return suite;
+    }
+        /**
+     * Test of getFeeder method, of class org.apache.commons.pipeline.driver.SynchronousStageDriver.
+     */
+    public void testGetFeeder() {
+        log.debug("testGetFeeder ---------------------------------------------");
+        ThreadPoolStageDriver instance = new ThreadPoolStageDriver(stage, context, new LinkedBlockingQueue(),
500, FaultTolerance.NONE, 5);
+        
+        Feeder feeder = instance.getFeeder();
+        assertNotNull(feeder);        
+    }
+    
+    /**
+     * Due to the design of the ThreadPoolStageDriver, it is meaningless
+     * to independently test the start or finish methods; however, testing 
+     * both together is meaningful. This test also provides verification of
+     * proper behavior of the getState() method.
+     */
+    public void testStartFinish() throws Exception {
+        log.debug("testStartFinish -------------------------------------------");
+        ThreadPoolStageDriver instance = new ThreadPoolStageDriver(stage, context, new LinkedBlockingQueue(),
500, FaultTolerance.NONE, 5);
+        
+        assertEquals(State.STOPPED, instance.getState());
+        
+        instance.start();
+        
+        assertTrue(instance.getState() == State.STARTED || instance.getState() == State.RUNNING);
+        
+        instance.finish();
+        
+        assertEquals(State.STOPPED, instance.getState());
+    }
+
+        
+    /*********************
+     * INTEGRATION TESTS *
+     *********************/
+    
+    public void testSingleStage() throws Exception {        
+        log.debug("testSingleStage -------------------------------------------");
+        StageDriverTestUtils.testSingleStage(this, new ThreadPoolStageDriverFactory());
+    }
+    
+    public void testMultiStage() throws Exception {        
+        log.debug("testMultiStage --------------------------------------------");
+        StageDriverTestUtils.testMultiStage(this, new ThreadPoolStageDriverFactory());
+    }
+    
+    public void testMultiFaultingStage() throws Exception {       
+        log.debug("testMultiFaultingStage ------------------------------------");
+        ThreadPoolStageDriverFactory factory = new ThreadPoolStageDriverFactory();
+        factory.setFaultTolerance(FaultTolerance.CHECKED);
+        
+        StageDriverTestUtils.testMultiFaultingStage(this, factory);
+    }
+}

Propchange: jakarta/commons/sandbox/pipeline/trunk/src/test/java/org/apache/commons/pipeline/driver/ThreadPoolStageDriverTest.java
------------------------------------------------------------------------------
    svn:eol-style = native



---------------------------------------------------------------------
To unsubscribe, e-mail: commons-dev-unsubscribe@jakarta.apache.org
For additional commands, e-mail: commons-dev-help@jakarta.apache.org


Mime
View raw message