You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ta...@apache.org on 2012/03/14 21:00:44 UTC
svn commit: r1300701 - in /activemq/activemq-cpp/trunk/activemq-cpp/src:
examples/cmstemplate/ main/ main/decaf/internal/util/concurrent/
main/decaf/util/concurrent/ test-benchmarks/ test/
test/activemq/wireformat/openwire/marshal/generated/ test/decaf...
Author: tabish
Date: Wed Mar 14 20:00:43 2012
New Revision: 1300701
URL: http://svn.apache.org/viewvc?rev=1300701&view=rev
Log:
Adds in the support for submitting task to the ExecutorService objects and getting back a Future to track their progress and Fetch their result. Adds additional tests and fixes a memory leak as well.
Added:
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/ExecutorsSupport.cpp (with props)
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/ExecutorsSupport.h (with props)
activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/AbstractExecutorServiceTest.cpp (with props)
activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/AbstractExecutorServiceTest.h (with props)
Modified:
activemq/activemq-cpp/trunk/activemq-cpp/src/examples/cmstemplate/ (props changed)
activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/Threading.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/AbstractExecutorService.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/ExecutorService.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/Executors.h
activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/FutureTask.h
activemq/activemq-cpp/trunk/activemq-cpp/src/test-benchmarks/ (props changed)
activemq/activemq-cpp/trunk/activemq-cpp/src/test/Makefile.am
activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/wireformat/openwire/marshal/generated/ (props changed)
activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/security/ (props changed)
activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/ExecutorsTestSupport.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/ExecutorsTestSupport.h
activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/FutureTaskTest.cpp
activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/FutureTaskTest.h
activemq/activemq-cpp/trunk/activemq-cpp/src/test/testRegistry.cpp
Propchange: activemq/activemq-cpp/trunk/activemq-cpp/src/examples/cmstemplate/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Wed Mar 14 20:00:43 2012
@@ -0,0 +1 @@
+.deps
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am?rev=1300701&r1=1300700&r2=1300701&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/Makefile.am Wed Mar 14 20:00:43 2012
@@ -319,6 +319,7 @@ cc_sources = \
decaf/internal/util/Resource.cpp \
decaf/internal/util/ResourceLifecycleManager.cpp \
decaf/internal/util/TimerTaskHeap.cpp \
+ decaf/internal/util/concurrent/ExecutorsSupport.cpp \
decaf/internal/util/concurrent/SynchronizableImpl.cpp \
decaf/internal/util/concurrent/Threading.cpp \
decaf/internal/util/concurrent/unix/Atomics.cpp \
@@ -840,6 +841,7 @@ h_sources = \
decaf/internal/util/ResourceLifecycleManager.h \
decaf/internal/util/TimerTaskHeap.h \
decaf/internal/util/concurrent/Atomics.h \
+ decaf/internal/util/concurrent/ExecutorsSupport.h \
decaf/internal/util/concurrent/PlatformThread.h \
decaf/internal/util/concurrent/SynchronizableImpl.h \
decaf/internal/util/concurrent/Threading.h \
Added: activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/ExecutorsSupport.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/ExecutorsSupport.cpp?rev=1300701&view=auto
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/ExecutorsSupport.cpp (added)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/ExecutorsSupport.cpp Wed Mar 14 20:00:43 2012
@@ -0,0 +1,19 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+#include "ExecutorsSupport.h"
+
Propchange: activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/ExecutorsSupport.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Added: activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/ExecutorsSupport.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/ExecutorsSupport.h?rev=1300701&view=auto
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/ExecutorsSupport.h (added)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/ExecutorsSupport.h Wed Mar 14 20:00:43 2012
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+#ifndef _DECAF_INTERNAL_UTIL_CONCURRENT_EXECUTORSSUPPORT_H_
+#define _DECAF_INTERNAL_UTIL_CONCURRENT_EXECUTORSSUPPORT_H_
+
+namespace decaf {
+namespace internal {
+namespace util {
+namespace concurrent {
+
+ /**
+ * Various support methods for use in Executors and surrounding classes.
+ *
+ * @since 1.0
+ */
+ class ExecutorsSupport {
+ private:
+
+ ExecutorsSupport() {}
+ ExecutorsSupport(const ExecutorsSupport&);
+
+ public:
+
+ };
+
+}}}}
+
+#endif /* _DECAF_INTERNAL_UTIL_CONCURRENT_EXECUTORSSUPPORT_H_ */
Propchange: activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/ExecutorsSupport.h
------------------------------------------------------------------------------
svn:eol-style = native
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/Threading.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/Threading.cpp?rev=1300701&r1=1300700&r2=1300701&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/Threading.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/internal/util/concurrent/Threading.cpp Wed Mar 14 20:00:43 2012
@@ -19,6 +19,7 @@
#include <decaf/lang/Thread.h>
#include <decaf/lang/System.h>
+#include <decaf/lang/Integer.h>
#include <decaf/lang/exceptions/NullPointerException.h>
#include <decaf/util/concurrent/Executors.h>
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/AbstractExecutorService.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/AbstractExecutorService.cpp?rev=1300701&r1=1300700&r2=1300701&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/AbstractExecutorService.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/AbstractExecutorService.cpp Wed Mar 14 20:00:43 2012
@@ -17,6 +17,8 @@
#include "AbstractExecutorService.h"
+#include <decaf/lang/exceptions/NullPointerException.h>
+
using namespace decaf;
using namespace decaf::util;
using namespace decaf::util::concurrent;
@@ -34,5 +36,16 @@ AbstractExecutorService::~AbstractExecut
////////////////////////////////////////////////////////////////////////////////
void AbstractExecutorService::doSubmit(FutureType* future) {
- throw UnsupportedOperationException();
+ try {
+ // Its supposed to be a RunnableFuture<?> be we should double check.
+ Runnable* task = dynamic_cast<Runnable*>(future);
+ if (task == NULL) {
+ throw NullPointerException(__FILE__, __LINE__, "Could not cast FutureType to a Runnabke");
+ }
+
+ // Ensure that we tell the subclass it owns the Future.
+ this->execute(task, true);
+ }
+ DECAF_CATCH_RETHROW(NullPointerException)
+ DECAF_CATCH_RETHROW(RejectedExecutionException)
}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/ExecutorService.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/ExecutorService.h?rev=1300701&r1=1300700&r2=1300701&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/ExecutorService.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/ExecutorService.h Wed Mar 14 20:00:43 2012
@@ -23,6 +23,7 @@
#include <decaf/lang/Runnable.h>
#include <decaf/util/ArrayList.h>
#include <decaf/util/concurrent/Future.h>
+#include <decaf/util/concurrent/FutureTask.h>
#include <decaf/util/concurrent/Executor.h>
#include <decaf/util/concurrent/TimeUnit.h>
#include <decaf/lang/exceptions/InterruptedException.h>
@@ -72,7 +73,7 @@ namespace concurrent {
*
* @throws InterruptedException if this call is interrupted while awaiting termination.
*/
- virtual bool awaitTermination( long long timeout, const TimeUnit& unit ) = 0;
+ virtual bool awaitTermination(long long timeout, const TimeUnit& unit) = 0;
/**
* Performs an orderly shutdown of this Executor. Previously queued tasks are allowed
@@ -108,26 +109,153 @@ namespace concurrent {
virtual bool isTerminated() const = 0;
/**
+ * Submits a value-returning task for execution and returns a Future pointer
+ * representing the pending results of the task. The Future's <tt>get</tt> method
+ * will return the task's result upon successful completion. The caller owns the
+ * returned pointer and is responsible for deleting it. The returned value is a
+ * proxy to the actual FutureTask that is submitted for execution so is legal for
+ * the caller to delete this value before its execution has completed.
+ *
+ * @param task
+ * Pointer to the Callable<?> task to submit.
+ * @param takeOwnership
+ * Boolean value indicating if the Executor now owns the pointer to the task.
+ *
+ * @return a Future<?> pointer representing pending completion of the task.
+ *
+ * @throws RejectedExecutionException if the task cannot be scheduled for execution
+ * @throws NullPointerException if the task is null
+ */
+ template<typename E>
+ Future<E>* submit(Callable<E>* task, bool takeOwnership = true) {
+
+ // Creates a new FutureTask to wrap the target task, and then creates a clone
+ // that will act as the proxy to return to the caller.
+ Pointer< FutureTask<E> > newTask(new FutureTask<E>(task, takeOwnership));
+ Pointer< FutureTask<E> > proxy(newTask->clone());
+
+ try {
+ // Its safe to submit and allow the task to be executed only after creating
+ // the proxy so that if its run on the current thread and destroyed the
+ // proxy still holds a vlid reference to the inner FutureTask data keeping it
+ // from being destroyed before the caller has a chance to call get().
+ this->doSubmit(newTask.get());
+
+ // No exception so we can release our ref, the executor owns it now.
+ newTask.release();
+
+ return proxy.release();
+ } catch(decaf::util::concurrent::RejectedExecutionException& ex) {
+ // Policy will delete the submitted task
+ newTask.release();
+ ex.setMark(__FILE__, __LINE__);
+ throw;
+ }
+ DECAF_CATCH_RETHROW(decaf::lang::exceptions::NullPointerException)
+ DECAF_CATCHALL_THROW(decaf::lang::Exception)
+ }
+
+ /**
+ * Submits a Runnable task for execution and returns a Future representing that
+ * task. The Future's <tt>get</tt> method will return the given result upon successful
+ * completion. The caller owns the returned pointer and is responsible for deleting it.
+ * The returned value is a proxy to the actual FutureTask that is submitted for execution
+ * so is legal for the caller to delete this value before its execution has completed.
+ *
+ * @param task
+ * The pointer to the task to submit.
+ * @param result
+ * The result to return
+ * @param takeOwnership
+ * Boolean value indicating if the Executor now owns the pointer to the task.
+ *
+ * @return a Future<?> pointer representing pending completion of the task,
+ *
+ * @throws RejectedExecutionException if the task cannot be scheduled for execution
+ * @throws NullPointerException if the task is null
+ */
+ template<typename E>
+ Future<E>* submit(decaf::lang::Runnable* task, const E& result, bool takeOwnership = true) {
+
+ // Creates a new FutureTask to wrap the target task, and then creates a clone
+ // that will act as the proxy to return to the caller.
+ Pointer< FutureTask<E> > newTask(new FutureTask<E>(task, result, takeOwnership));
+ Pointer< FutureTask<E> > proxy(newTask->clone());
+
+ try {
+ // Its safe to submit and allow the task to be executed only after creating
+ // the proxy so that if its run on the current thread and destroyed the
+ // proxy still holds a vlid reference to the inner FutureTask data keeping it
+ // from being destroyed before the caller has a chance to call get().
+ this->doSubmit(newTask.get());
+
+ // No exception so we can release our reference, the executor owns it now.
+ newTask.release();
+
+ return proxy.release();
+ } catch(decaf::util::concurrent::RejectedExecutionException& ex) {
+ // Policy will delete the submitted task
+ newTask.release();
+ ex.setMark(__FILE__, __LINE__);
+ throw;
+ }
+ DECAF_CATCH_RETHROW(decaf::lang::exceptions::NullPointerException)
+ DECAF_CATCHALL_THROW(decaf::lang::Exception)
+ }
+
+ /**
* Submits a Runnable object for execution. A Future object is created and returned
- * that will return the default value of the template type upon completion.
+ * that will return the default value of the template type upon completion. The caller
+ * owns the returned pointer and is responsible for deleting it. The returned value is
+ * a proxy to the actual FutureTask that is submitted for execution so is legal for the
+ * caller to delete this value before its execution has completed.
*
* @param task
* Pointer to a Runnable object that will be executed by this ExecutorService.
+ * @param takeOwnership
+ * Boolean value indicating if the Executor now owns the pointer to the task.
*
* @returns a new Future<?> pointer that is owned by the caller.
*
+ * @throws RejectedExecutionException if the task cannot be scheduled for execution
* @throws NullPointerException if the Runnable pointer passed is NULL.
*/
template<typename E>
- Future<E>* submit(decaf::lang::Runnable* task) {
- return NULL;
+ Future<E>* submit(decaf::lang::Runnable* task, bool takeOwnership = true) {
+
+ // Creates a new FutureTask to wrap the target task, and then creates a clone
+ // that will act as the proxy to return to the caller.
+ Pointer< FutureTask<E> > newTask(new FutureTask<E>(task, E(), takeOwnership));
+ Pointer< FutureTask<E> > proxy(newTask->clone());
+
+ try {
+ // Its safe to submit and allow the task to be executed only after creating
+ // the proxy so that if its run on the current thread and destroyed the
+ // proxy still holds a vlid reference to the inner FutureTask data keeping it
+ // from being destroyed before the caller has a chance to call get().
+ this->doSubmit(newTask.get());
+
+ // No exception so we can release our ref, the executor owns it now.
+ newTask.release();
+
+ return proxy.release();
+ } catch(decaf::util::concurrent::RejectedExecutionException& ex) {
+ // Policy will delete the submitted task
+ newTask.release();
+ ex.setMark(__FILE__, __LINE__);
+ throw;
+ }
+ DECAF_CATCH_RETHROW(decaf::lang::exceptions::NullPointerException)
+ DECAF_CATCHALL_THROW(decaf::lang::Exception)
}
protected:
/**
* Perform the actual submit of a FutureType instance, the caller is responsible for
- * creating the properly typed Future<E> object and returning that to its caller.
+ * creating the properly typed Future<E> object and returning that to its caller. The
+ * pointer provided is the property of this Executor and must be deleted by this executor
+ * once its completed.
*
* @param future
* Pointer to a base FutureType instance that is to be submitted to the Executor.
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/Executors.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/Executors.h?rev=1300701&r1=1300700&r2=1300701&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/Executors.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/Executors.h Wed Mar 14 20:00:43 2012
@@ -22,8 +22,6 @@
#include <decaf/lang/Thread.h>
#include <decaf/lang/Runnable.h>
-#include <decaf/util/concurrent/ExecutorService.h>
-#include <decaf/util/concurrent/ThreadFactory.h>
#include <decaf/util/concurrent/Callable.h>
#include <decaf/lang/exceptions/NullPointerException.h>
@@ -32,6 +30,9 @@ namespace decaf {
namespace util {
namespace concurrent {
+ class ThreadFactory;
+ class ExecutorService;
+
/**
* Implements a set of utilities for use with Executors, ExecutorService, ThreadFactory,
* and Callable types, as well as providing factory methods for instance of these
@@ -42,11 +43,17 @@ namespace concurrent {
class DECAF_API Executors {
private:
+ Executors();
+ Executors(const Executors&);
+ Executors& operator= (const Executors&);
+
+ private:
+
/**
* A Callable subclass that runs given task and returns given result
*/
template<typename E>
- class RunnableAdapter : public Callable<E> {
+ class RunnableAdapter : public decaf::util::concurrent::Callable<E> {
private:
decaf::lang::Runnable* task;
@@ -56,7 +63,7 @@ namespace concurrent {
public:
RunnableAdapter(decaf::lang::Runnable* task, bool owns, const E& result) :
- Callable<E>(), task(task), owns(owns), result(result) {
+ decaf::util::concurrent::Callable<E>(), task(task), owns(owns), result(result) {
}
virtual ~RunnableAdapter() {
@@ -74,12 +81,6 @@ namespace concurrent {
}
};
- private:
-
- Executors();
- Executors(const Executors&);
- Executors& operator= (const Executors&);
-
public:
virtual ~Executors();
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/FutureTask.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/FutureTask.h?rev=1300701&r1=1300700&r2=1300701&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/FutureTask.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/decaf/util/concurrent/FutureTask.h Wed Mar 14 20:00:43 2012
@@ -26,7 +26,6 @@
#include <decaf/util/concurrent/RunnableFuture.h>
#include <decaf/util/concurrent/Callable.h>
-#include <decaf/util/concurrent/Executors.h>
#include <decaf/util/concurrent/CancellationException.h>
#include <decaf/util/concurrent/ExecutionException.h>
#include <decaf/util/concurrent/TimeoutException.h>
@@ -50,7 +49,7 @@ namespace concurrent {
* A FutureTask can be used to wrap a Callable or Runnable object. Because FutureTask
* implements Runnable, a FutureTask can be submitted to an Executor for execution.
*
- * In addition to serving as a standalone class, this class provides protected functionality
+ * In addition to serving as a stand-alone class, this class provides protected functionality
* that may be useful when creating customized task classes.
*
* @since 1.0
@@ -60,6 +59,48 @@ namespace concurrent {
private:
/**
+ * A Callable subclass that runs given task and returns given result, used to
+ * wrap either a Runnable or Callable pointer and
+ */
+ class FutureTaskAdapter : public decaf::util::concurrent::Callable<T> {
+ private:
+
+ decaf::lang::Runnable* task;
+ decaf::util::concurrent::Callable<T>* callable;
+ bool owns;
+ T result;
+
+ public:
+
+ FutureTaskAdapter(decaf::lang::Runnable* task, const T& result, bool owns = true) :
+ decaf::util::concurrent::Callable<T>(), task(task), callable(NULL), owns(owns), result(result) {
+ }
+
+ FutureTaskAdapter(decaf::util::concurrent::Callable<T>* task, bool owns = true) :
+ decaf::util::concurrent::Callable<T>(), task(NULL), callable(task), owns(owns), result(T()) {
+ }
+
+ virtual ~FutureTaskAdapter() {
+ try{
+ if (owns) {
+ delete this->task;
+ delete this->callable;
+ }
+ }
+ DECAF_CATCHALL_NOTHROW()
+ }
+
+ virtual T call() {
+ if (this->task != NULL) {
+ this->task->run();
+ return result;
+ } else {
+ return this->callable->call();
+ }
+ }
+ };
+
+ /**
* Synchronization control for FutureTask.
*
* Uses AQS sync state to represent run status
@@ -103,7 +144,6 @@ namespace concurrent {
}
virtual ~FutureTaskSync() {
-
}
bool innerIsCancelled() const {
@@ -210,7 +250,7 @@ namespace concurrent {
return;
}
- runner = decaf::lang::Thread::currentThread();
+ this->runner = decaf::lang::Thread::currentThread();
if (getState() == RUNNING) { // recheck after setting thread
T result;
try {
@@ -218,6 +258,13 @@ namespace concurrent {
} catch(decaf::lang::Exception& ex) {
this->parent->setException(ex);
return;
+ } catch(std::exception& stdex) {
+ this->parent->setException(decaf::lang::Exception(&stdex));
+ return;
+ } catch(...) {
+ this->parent->setException(decaf::lang::Exception(
+ __FILE__, __LINE__, "FutureTask Caught Unknown exception during task execution."));
+ return;
}
this->parent->set(result);
} else {
@@ -240,6 +287,13 @@ namespace concurrent {
} catch(decaf::lang::Exception& ex) {
this->parent->setException(ex);
return false;
+ } catch(std::exception& stdex) {
+ this->parent->setException(decaf::lang::Exception(&stdex));
+ return false;
+ } catch(...) {
+ this->parent->setException(decaf::lang::Exception(
+ __FILE__, __LINE__, "FutureTask Caught Unknown exception during task execution."));
+ return false;
}
}
@@ -270,7 +324,7 @@ namespace concurrent {
private:
- FutureTaskSync* sync;
+ Pointer<FutureTaskSync> sync;
public:
@@ -280,16 +334,18 @@ namespace concurrent {
*
* @param callable
* The callable task that will be invoked when run.
+ * @param takeOwnership
+ * Boolean value indicating if the Executor now owns the pointer to the task.
*
* @throws NullPointerException if callable pointer is NULL
*/
- FutureTask(Callable<T>* callable) : sync(NULL) {
+ FutureTask(Callable<T>* callable, bool takeOwnership = true) : sync(NULL) {
if (callable == NULL ) {
throw decaf::lang::exceptions::NullPointerException(__FILE__, __LINE__,
"The Callable pointer passed to the constructor was NULL");
}
- this->sync = new FutureTaskSync(this, callable);
+ this->sync.reset(new FutureTaskSync(this, new FutureTaskAdapter(callable, takeOwnership)));
}
/**
@@ -301,23 +357,21 @@ namespace concurrent {
* The runnable task that the future will execute.
* @param result
* The result to return on successful completion.
+ * @param takeOwnership
+ * Boolean value indicating if the Executor now owns the pointer to the task.
*
* @throws NullPointerException if runnable is NULL.
*/
- FutureTask(decaf::lang::Runnable* runnable, const T& result) {
+ FutureTask(decaf::lang::Runnable* runnable, const T& result, bool takeOwnership = true) : sync(NULL) {
if (runnable == NULL ) {
throw decaf::lang::exceptions::NullPointerException(__FILE__, __LINE__,
"The Runnable pointer passed to the constructor was NULL");
}
- sync = new FutureTaskSync(this, Executors::callable<T>(runnable, result));
+ this->sync.reset(new FutureTaskSync(this, new FutureTaskAdapter(runnable, result, takeOwnership)));
}
virtual ~FutureTask() {
- try{
- delete this->sync;
- }
- DECAF_CATCHALL_NOTHROW()
}
virtual bool isCancelled() const {
@@ -340,6 +394,10 @@ namespace concurrent {
return this->sync->innerGet(unit.toNanos(timeout));
}
+ FutureTask<T>* clone() {
+ return new FutureTask<T>(*this);
+ }
+
public:
/**
@@ -394,6 +452,16 @@ namespace concurrent {
return this->sync->innerRunAndReset();
}
+ public:
+
+ FutureTask(const FutureTask<T>& source) : RunnableFuture<T>(), sync(source.sync) {
+ }
+
+ FutureTask<T>& operator= (const FutureTask<T>& source) {
+ this->sync = source.sync;
+ return *this;
+ }
+
};
}}}
Propchange: activemq/activemq-cpp/trunk/activemq-cpp/src/test-benchmarks/
------------------------------------------------------------------------------
--- svn:ignore (original)
+++ svn:ignore Wed Mar 14 20:00:43 2012
@@ -1,2 +1,3 @@
.deps
Makefile.in
+Makefile
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/test/Makefile.am
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/test/Makefile.am?rev=1300701&r1=1300700&r2=1300701&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/test/Makefile.am (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/test/Makefile.am Wed Mar 14 20:00:43 2012
@@ -203,6 +203,7 @@ cc_sources = \
decaf/util/StringTokenizerTest.cpp \
decaf/util/TimerTest.cpp \
decaf/util/UUIDTest.cpp \
+ decaf/util/concurrent/AbstractExecutorServiceTest.cpp \
decaf/util/concurrent/ConcurrentStlMapTest.cpp \
decaf/util/concurrent/CopyOnWriteArrayListTest.cpp \
decaf/util/concurrent/CopyOnWriteArraySetTest.cpp \
@@ -431,6 +432,7 @@ h_sources = \
decaf/util/StringTokenizerTest.h \
decaf/util/TimerTest.h \
decaf/util/UUIDTest.h \
+ decaf/util/concurrent/AbstractExecutorServiceTest.h \
decaf/util/concurrent/ConcurrentStlMapTest.h \
decaf/util/concurrent/CopyOnWriteArrayListTest.h \
decaf/util/concurrent/CopyOnWriteArraySetTest.h \
Propchange: activemq/activemq-cpp/trunk/activemq-cpp/src/test/activemq/wireformat/openwire/marshal/generated/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Wed Mar 14 20:00:43 2012
@@ -0,0 +1 @@
+.deps
Propchange: activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/security/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Wed Mar 14 20:00:43 2012
@@ -0,0 +1 @@
+*.deps
Added: activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/AbstractExecutorServiceTest.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/AbstractExecutorServiceTest.cpp?rev=1300701&view=auto
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/AbstractExecutorServiceTest.cpp (added)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/AbstractExecutorServiceTest.cpp Wed Mar 14 20:00:43 2012
@@ -0,0 +1,427 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+#include "AbstractExecutorServiceTest.h"
+
+#include <string>
+
+#include <decaf/lang/Runnable.h>
+#include <decaf/lang/Pointer.h>
+#include <decaf/lang/exceptions/NullPointerException.h>
+#include <decaf/util/ArrayList.h>
+#include <decaf/util/concurrent/LinkedBlockingQueue.h>
+#include <decaf/util/concurrent/TimeUnit.h>
+#include <decaf/util/concurrent/Future.h>
+#include <decaf/util/concurrent/AbstractExecutorService.h>
+#include <decaf/util/concurrent/ThreadPoolExecutor.h>
+
+using namespace std;
+using namespace decaf;
+using namespace decaf::lang;
+using namespace decaf::lang::exceptions;
+using namespace decaf::util;
+using namespace decaf::util::concurrent;
+
+////////////////////////////////////////////////////////////////////////////////
+namespace {
+
+ /**
+ * A no-frills implementation of AbstractExecutorService, designed
+ * to test the submit methods only.
+ */
+ class DirectExecutorService : public AbstractExecutorService {
+ private:
+
+ volatile bool isshutdown;
+
+ public:
+
+ DirectExecutorService() : AbstractExecutorService(), isshutdown(false) {
+ }
+
+ virtual ~DirectExecutorService() {}
+
+ virtual void execute(Runnable* r) {
+ this->execute(r, true);
+ }
+
+ virtual void execute(Runnable* r, bool own) {
+ if (r == NULL) {
+ throw NullPointerException(__FILE__, __LINE__, "Runnable was NULL");
+ }
+
+ try {
+ r->run();
+ } catch(Exception& e) {
+ if (own) {
+ delete r;
+ }
+
+ throw;
+ }
+
+ if (own) {
+ delete r;
+ }
+ }
+
+ virtual void shutdown() {
+ isshutdown = true;
+ }
+
+ virtual ArrayList<Runnable*> shutdownNow() {
+ isshutdown = true;
+
+ return ArrayList<Runnable*>();
+ }
+
+ virtual bool isShutdown() const {
+ return isshutdown;
+ }
+
+ virtual bool isTerminated() const {
+ return isShutdown();
+ }
+
+ virtual bool awaitTermination(long long timeout, const TimeUnit& unit) {
+ return isShutdown();
+ }
+ };
+}
+
+////////////////////////////////////////////////////////////////////////////////
+AbstractExecutorServiceTest::AbstractExecutorServiceTest() {
+}
+
+////////////////////////////////////////////////////////////////////////////////
+AbstractExecutorServiceTest::~AbstractExecutorServiceTest() {
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void AbstractExecutorServiceTest::testExecuteRunnable() {
+
+ try {
+ bool done = false;
+ DirectExecutorService e;
+ TrackedShortRunnable task(&done);
+
+ CPPUNIT_ASSERT(!done);
+ Future<int>* future = e.submit<int>(&task, false);
+ future->get();
+ CPPUNIT_ASSERT(done);
+ delete future;
+ } catch (ExecutionException& ex) {
+ unexpectedException();
+ } catch (InterruptedException& ex) {
+ unexpectedException();
+ }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void AbstractExecutorServiceTest::testSubmitCallable() {
+
+ try {
+ DirectExecutorService e;
+ Future<string>* future = e.submit(new StringTask());
+ string result = future->get();
+ CPPUNIT_ASSERT_EQUAL(TEST_STRING, result);
+ delete future;
+ } catch (ExecutionException& ex) {
+ unexpectedException();
+ } catch (InterruptedException& ex) {
+ unexpectedException();
+ }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void AbstractExecutorServiceTest::testSubmitRunnable() {
+ try {
+ DirectExecutorService e;
+ Future<int>* future = e.submit<int>(new NoOpRunnable());
+ future->get();
+ CPPUNIT_ASSERT(future->isDone());
+ delete future;
+ } catch (ExecutionException& ex) {
+ unexpectedException();
+ } catch (InterruptedException& ex) {
+ unexpectedException();
+ }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void AbstractExecutorServiceTest::testSubmitRunnable2() {
+ try {
+ DirectExecutorService e;
+ Future<string>* future = e.submit(new NoOpRunnable(), TEST_STRING);
+ string result = future->get();
+ CPPUNIT_ASSERT_EQUAL(TEST_STRING, result);
+ delete future;
+ } catch (ExecutionException& ex) {
+ unexpectedException();
+ } catch (InterruptedException& ex) {
+ unexpectedException();
+ }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void AbstractExecutorServiceTest::testExecuteNullRunnable() {
+ try {
+ DirectExecutorService e;
+ TrackedShortRunnable* task = NULL;
+ e.submit<int>(task);
+ shouldThrow();
+ } catch (NullPointerException& success) {
+ } catch (Exception& ex) {
+ unexpectedException();
+ }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void AbstractExecutorServiceTest::testSubmitNullCallable() {
+ try {
+ DirectExecutorService e;
+ StringTask* t = NULL;
+ e.submit(t);
+ shouldThrow();
+ } catch (NullPointerException& success) {
+ } catch (Exception& ex) {
+ unexpectedException();
+ }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void AbstractExecutorServiceTest::testExecute1() {
+ ThreadPoolExecutor p(1, 1, 60, TimeUnit::SECONDS, new LinkedBlockingQueue<Runnable*>(1));
+ MediumRunnable task(this);
+
+ try {
+ for(int i = 0; i < 5; ++i) {
+ Pointer< Future<int> > future(p.submit<int>(&task, false));
+ }
+
+ shouldThrow();
+ } catch(RejectedExecutionException& success) {
+ }
+
+ joinPool(p);
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void AbstractExecutorServiceTest::testExecute2() {
+ ThreadPoolExecutor p(1, 1, 60, TimeUnit::SECONDS, new LinkedBlockingQueue<Runnable*>(1));
+ try {
+
+ for(int i = 0; i < 5; ++i) {
+ delete p.submit(new SmallCallable<int>(this));
+ }
+
+ shouldThrow();
+ } catch(RejectedExecutionException& e) {
+ }
+
+ joinPool(p);
+}
+
+////////////////////////////////////////////////////////////////////////////////
+namespace {
+
+ template<typename E>
+ class TestInterruptedSubmitCallable : public Callable<E> {
+ private:
+
+ AbstractExecutorServiceTest* parent;
+
+ public:
+
+ TestInterruptedSubmitCallable(AbstractExecutorServiceTest* parent) :
+ decaf::util::concurrent::Callable<E>(), parent(parent) {
+ }
+
+ virtual ~TestInterruptedSubmitCallable() {}
+
+ virtual E call() {
+ try {
+ Thread::sleep(AbstractExecutorServiceTest::LONG_DELAY_MS);
+ parent->threadShouldThrow();
+ } catch(InterruptedException& e){
+ }
+
+ return E();
+ }
+ };
+
+ class TestInterruptedSubmitRunnable : public Runnable {
+ private:
+
+ AbstractExecutorServiceTest* parent;
+ ThreadPoolExecutor* executor;
+
+ public:
+
+ TestInterruptedSubmitRunnable(AbstractExecutorServiceTest* parent, ThreadPoolExecutor* executor) :
+ Runnable(), parent(parent), executor(executor) {
+ }
+
+ virtual ~TestInterruptedSubmitRunnable() {}
+
+ virtual void run() {
+ try {
+ Pointer< Future<int> > future(executor->submit(new TestInterruptedSubmitCallable<int>(parent)));
+ future->get();
+ } catch(InterruptedException& success){
+ } catch(Exception& e) {
+ parent->unexpectedException();
+ }
+ }
+ };
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void AbstractExecutorServiceTest::testInterruptedSubmit() {
+
+ // TODO
+// ThreadPoolExecutor p(1, 1, 60, TimeUnit::SECONDS, new LinkedBlockingQueue<Runnable*>(10));
+// TestInterruptedSubmitRunnable runner(this, &p);
+// Thread t(&runner);
+//
+// try {
+//
+// t.start();
+// Thread::sleep(SHORT_DELAY_MS);
+// t.interrupt();
+//
+// } catch(Exception& e){
+// unexpectedException();
+// }
+//
+// joinPool(p);
+}
+
+////////////////////////////////////////////////////////////////////////////////
+namespace {
+
+ template<typename E>
+ class CallingRunnable : public decaf::lang::Runnable {
+ private:
+
+ AbstractExecutorServiceTest* parent;
+ Callable<E>* target;
+
+ public:
+
+ CallingRunnable(AbstractExecutorServiceTest* parent, Callable<E>* target) :
+ Runnable(), parent(parent), target(target) {
+ }
+
+ virtual ~CallingRunnable() {
+ delete target;
+ }
+
+ virtual void run() {
+ try {
+ target->call();
+ } catch(Exception& e) {
+ }
+ }
+ };
+
+ template<typename E>
+ class TestSubmitIECallable : public Callable<E> {
+ private:
+
+ AbstractExecutorServiceTest* parent;
+ ThreadPoolExecutor* executor;
+
+ public:
+
+ TestSubmitIECallable(AbstractExecutorServiceTest* parent, ThreadPoolExecutor* executor) :
+ Callable<E>(), parent(parent), executor(executor) {
+ }
+
+ virtual ~TestSubmitIECallable() {}
+
+ virtual E call() {
+ try {
+ Pointer< Future<E> >(executor->submit(
+ new AbstractExecutorServiceTest::SmallCallable<E>(parent)))->get();
+ parent->threadShouldThrow();
+ } catch(InterruptedException& e){
+ } catch(RejectedExecutionException& e2){
+ } catch(ExecutionException& e3){
+ }
+
+ return E();
+ }
+ };
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void AbstractExecutorServiceTest::testSubmitIE() {
+
+ ThreadPoolExecutor p(1, 1, 60, TimeUnit::SECONDS, new LinkedBlockingQueue<Runnable*>(10));
+
+ TestSubmitIECallable<int>* callable = new TestSubmitIECallable<int>(this, &p);
+ CallingRunnable<int> runner(this, callable);
+ Thread t(&runner);
+
+ try {
+ t.start();
+ Thread::sleep(SHORT_DELAY_MS);
+ t.interrupt();
+ t.join();
+ } catch(InterruptedException& e){
+ unexpectedException();
+ }
+
+ joinPool(p);
+}
+
+////////////////////////////////////////////////////////////////////////////////
+namespace {
+
+ class testSubmitEECallable : public Callable<int> {
+ public:
+
+ virtual ~testSubmitEECallable() {}
+
+ virtual int call() {
+ throw NumberFormatException(__FILE__, __LINE__, "Throwing a common exception");
+ return 1;
+ }
+ };
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void AbstractExecutorServiceTest::testSubmitEE() {
+ ThreadPoolExecutor p(1, 1, 60, TimeUnit::SECONDS, new LinkedBlockingQueue<Runnable*>(10));
+
+ testSubmitEECallable c;
+
+ try {
+
+ for(int i = 0; i < 5; i++) {
+ Pointer< Future<int> >(p.submit(&c, false))->get();
+ }
+
+ shouldThrow();
+ } catch(ExecutionException& success) {
+ } catch(Exception& e) {
+ unexpectedException();
+ }
+
+ joinPool(p);
+}
Propchange: activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/AbstractExecutorServiceTest.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Added: activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/AbstractExecutorServiceTest.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/AbstractExecutorServiceTest.h?rev=1300701&view=auto
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/AbstractExecutorServiceTest.h (added)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/AbstractExecutorServiceTest.h Wed Mar 14 20:00:43 2012
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+#ifndef _DECAF_UTIL_CONCURRENT_ABSTRACTEXECUTORSERVICETEST_H_
+#define _DECAF_UTIL_CONCURRENT_ABSTRACTEXECUTORSERVICETEST_H_
+
+#include <decaf/util/concurrent/ExecutorsTestSupport.h>
+
+namespace decaf {
+namespace util {
+namespace concurrent {
+
+ class AbstractExecutorServiceTest : public ExecutorsTestSupport
+ {
+ CPPUNIT_TEST_SUITE( AbstractExecutorServiceTest );
+ CPPUNIT_TEST( testExecuteRunnable );
+ CPPUNIT_TEST( testSubmitCallable );
+ CPPUNIT_TEST( testSubmitRunnable );
+ CPPUNIT_TEST( testSubmitRunnable2 );
+ CPPUNIT_TEST( testExecuteNullRunnable );
+ CPPUNIT_TEST( testSubmitNullCallable );
+ CPPUNIT_TEST( testExecute1 );
+ CPPUNIT_TEST( testExecute2 );
+ CPPUNIT_TEST( testInterruptedSubmit );
+ CPPUNIT_TEST( testSubmitIE );
+ CPPUNIT_TEST( testSubmitEE );
+ CPPUNIT_TEST_SUITE_END();
+
+ public:
+
+ AbstractExecutorServiceTest();
+ virtual ~AbstractExecutorServiceTest();
+
+ void testExecuteRunnable();
+ void testSubmitCallable();
+ void testSubmitRunnable();
+ void testSubmitRunnable2();
+ void testExecuteNullRunnable();
+ void testSubmitNullCallable();
+ void testExecute1();
+ void testExecute2();
+ void testInterruptedSubmit();
+ void testSubmitIE();
+ void testSubmitEE();
+
+ };
+
+}}}
+
+#endif /* _DECAF_UTIL_CONCURRENT_ABSTRACTEXECUTORSERVICETEST_H_ */
Propchange: activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/AbstractExecutorServiceTest.h
------------------------------------------------------------------------------
svn:eol-style = native
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/ExecutorsTestSupport.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/ExecutorsTestSupport.cpp?rev=1300701&r1=1300700&r2=1300701&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/ExecutorsTestSupport.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/ExecutorsTestSupport.cpp Wed Mar 14 20:00:43 2012
@@ -30,6 +30,8 @@ const int ExecutorsTestSupport::SMALL_DE
const int ExecutorsTestSupport::MEDIUM_DELAY_MS = 50 * 10;
const int ExecutorsTestSupport::LONG_DELAY_MS = 50 * 50;
+const std::string ExecutorsTestSupport::TEST_STRING = "a test string";
+
///////////////////////////////////////////////////////////////////////////////
ExecutorsTestSupport::ExecutorsTestSupport() : threadFailed(false) {
}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/ExecutorsTestSupport.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/ExecutorsTestSupport.h?rev=1300701&r1=1300700&r2=1300701&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/ExecutorsTestSupport.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/ExecutorsTestSupport.h Wed Mar 14 20:00:43 2012
@@ -21,6 +21,7 @@
#include <cppunit/TestFixture.h>
#include <cppunit/extensions/HelperMacros.h>
+#include <decaf/lang/Boolean.h>
#include <decaf/lang/Thread.h>
#include <decaf/lang/Runnable.h>
@@ -45,6 +46,8 @@ namespace concurrent {
static const int MEDIUM_DELAY_MS;
static const int LONG_DELAY_MS;
+ static const std::string TEST_STRING;
+
protected:
bool threadFailed;
@@ -145,6 +148,30 @@ namespace concurrent {
}
};
+ template<typename E>
+ class SmallCallable : public Callable<E> {
+ private:
+
+ ExecutorsTestSupport* parent;
+
+ public:
+
+ SmallCallable(ExecutorsTestSupport* parent) : decaf::util::concurrent::Callable<E>(), parent(parent) {
+ }
+
+ virtual ~SmallCallable() {}
+
+ virtual E call() {
+ try {
+ Thread::sleep(SMALL_DELAY_MS);
+ } catch(decaf::lang::Exception& e) {
+ parent->threadUnexpectedException(e);
+ }
+
+ return E();
+ }
+ };
+
class SmallInterruptedRunnable : public decaf::lang::Runnable {
private:
@@ -336,6 +363,27 @@ namespace concurrent {
}
};
+ class TrackedShortRunnable : public Runnable {
+ private:
+
+ bool* done;
+
+ public:
+
+ TrackedShortRunnable(bool* done) : decaf::lang::Runnable(), done(done) {
+ }
+
+ virtual ~TrackedShortRunnable() {}
+
+ virtual void run() {
+ try {
+ Thread::sleep(SMALL_DELAY_MS);
+ *done = true;
+ } catch(decaf::lang::Exception& e) {
+ }
+ }
+ };
+
class TrackedNoOpRunnable : public Runnable {
private:
@@ -374,6 +422,19 @@ namespace concurrent {
}
};
+ class StringTask : public decaf::util::concurrent::Callable<std::string> {
+ public:
+
+ StringTask() : decaf::util::concurrent::Callable<std::string>() {
+ }
+
+ virtual ~StringTask() {}
+
+ std::string call() {
+ return TEST_STRING;
+ }
+ };
+
};
}}}
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/FutureTaskTest.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/FutureTaskTest.cpp?rev=1300701&r1=1300700&r2=1300701&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/FutureTaskTest.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/FutureTaskTest.cpp Wed Mar 14 20:00:43 2012
@@ -24,6 +24,8 @@
#include <decaf/lang/Thread.h>
#include <decaf/lang/exceptions/IllegalArgumentException.h>
+#include <typeinfo>
+
using namespace std;
using namespace decaf;
using namespace decaf::lang;
@@ -54,6 +56,58 @@ namespace {
}
};
+ class FutureRunnable : public decaf::lang::Runnable {
+ public:
+
+ FutureRunnable() {
+ }
+
+ virtual ~FutureRunnable() {
+ }
+
+ virtual void run() {
+ }
+ };
+
+ template<typename E>
+ class FutureCallable : public Callable<E> {
+ public:
+
+ FutureCallable() {
+ }
+
+ virtual ~FutureCallable() {
+ }
+
+ virtual E call() {
+ return E();
+ }
+ };
+
+ template<typename E>
+ class MediumSleepCallable : public Callable<E> {
+ private:
+
+ FutureTaskTest* parent;
+
+ public:
+
+ MediumSleepCallable(FutureTaskTest* parent) : Callable<E>(), parent(parent) {
+ }
+
+ virtual ~MediumSleepCallable() {
+ }
+
+ virtual E call() {
+ try {
+ Thread::sleep(FutureTaskTest::MEDIUM_DELAY_MS);
+ this->parent->threadShouldThrow();
+ } catch (InterruptedException& success) {
+ }
+
+ return E();
+ }
+ };
}
////////////////////////////////////////////////////////////////////////////////
@@ -83,6 +137,31 @@ void FutureTaskTest::testConstructor2()
}
////////////////////////////////////////////////////////////////////////////////
+void FutureTaskTest::testConstructor3() {
+ FutureTask<int> task(new FutureRunnable(), 10);
+ CPPUNIT_ASSERT(!task.isCancelled());
+ CPPUNIT_ASSERT(!task.isDone());
+
+ task.run();
+
+ CPPUNIT_ASSERT(!task.isCancelled());
+ CPPUNIT_ASSERT(task.isDone());
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void FutureTaskTest::testConstructor4() {
+
+ FutureTask<std::string> task(new FutureCallable<std::string>());
+ CPPUNIT_ASSERT(!task.isCancelled());
+ CPPUNIT_ASSERT(!task.isDone());
+
+ task.run();
+
+ CPPUNIT_ASSERT(!task.isCancelled());
+ CPPUNIT_ASSERT(task.isDone());
+}
+
+////////////////////////////////////////////////////////////////////////////////
void FutureTaskTest::testIsDone() {
FutureTask<int> task(new NoOpCallable<int>());
@@ -118,8 +197,6 @@ void FutureTaskTest::testSet() {
}
}
-#include <typeinfo>
-
////////////////////////////////////////////////////////////////////////////////
void FutureTaskTest::testSetException() {
NoSuchElementException nse;
@@ -165,35 +242,6 @@ void FutureTaskTest::testCancelAfterRun(
}
////////////////////////////////////////////////////////////////////////////////
-namespace {
-
- template<typename E>
- class MediumSleepCallable : public Callable<E> {
- private:
-
- FutureTaskTest* parent;
-
- public:
-
- MediumSleepCallable(FutureTaskTest* parent) : Callable<E>(), parent(parent) {
- }
-
- virtual ~MediumSleepCallable() {
- }
-
- virtual E call() {
- try {
- Thread::sleep(FutureTaskTest::MEDIUM_DELAY_MS);
- this->parent->threadShouldThrow();
- } catch (InterruptedException& success) {
- }
-
- return E();
- }
- };
-}
-
-////////////////////////////////////////////////////////////////////////////////
void FutureTaskTest::testCancelInterrupt() {
FutureTask<std::string> task(new MediumSleepCallable<std::string>(this));
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/FutureTaskTest.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/FutureTaskTest.h?rev=1300701&r1=1300700&r2=1300701&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/FutureTaskTest.h (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/test/decaf/util/concurrent/FutureTaskTest.h Wed Mar 14 20:00:43 2012
@@ -29,6 +29,8 @@ namespace concurrent {
CPPUNIT_TEST_SUITE( FutureTaskTest );
CPPUNIT_TEST( testConstructor1 );
CPPUNIT_TEST( testConstructor2 );
+ CPPUNIT_TEST( testConstructor3 );
+ CPPUNIT_TEST( testConstructor4 );
CPPUNIT_TEST( testIsDone );
CPPUNIT_TEST( testRunAndReset );
CPPUNIT_TEST( testResetAfterCancel );
@@ -57,6 +59,8 @@ namespace concurrent {
void testConstructor1();
void testConstructor2();
+ void testConstructor3();
+ void testConstructor4();
void testIsDone();
void testRunAndReset();
void testResetAfterCancel();
Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/test/testRegistry.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/test/testRegistry.cpp?rev=1300701&r1=1300700&r2=1300701&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/test/testRegistry.cpp (original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/test/testRegistry.cpp Wed Mar 14 20:00:43 2012
@@ -275,16 +275,18 @@
//CPPUNIT_TEST_SUITE_REGISTRATION( decaf::util::concurrent::MutexTest );
//#include <decaf/util/concurrent/ThreadPoolExecutorTest.h>
//CPPUNIT_TEST_SUITE_REGISTRATION( decaf::util::concurrent::ThreadPoolExecutorTest );
-#include <decaf/util/concurrent/ExecutorsTest.h>
-CPPUNIT_TEST_SUITE_REGISTRATION( decaf::util::concurrent::ExecutorsTest );
+//#include <decaf/util/concurrent/ExecutorsTest.h>
+//CPPUNIT_TEST_SUITE_REGISTRATION( decaf::util::concurrent::ExecutorsTest );
//#include <decaf/util/concurrent/TimeUnitTest.h>
//CPPUNIT_TEST_SUITE_REGISTRATION( decaf::util::concurrent::TimeUnitTest );
//#include <decaf/util/concurrent/LinkedBlockingQueueTest.h>
//CPPUNIT_TEST_SUITE_REGISTRATION( decaf::util::concurrent::LinkedBlockingQueueTest );
//#include <decaf/util/concurrent/SemaphoreTest.h>
//CPPUNIT_TEST_SUITE_REGISTRATION( decaf::util::concurrent::SemaphoreTest );
-#include <decaf/util/concurrent/FutureTaskTest.h>
-CPPUNIT_TEST_SUITE_REGISTRATION( decaf::util::concurrent::FutureTaskTest );
+//#include <decaf/util/concurrent/FutureTaskTest.h>
+//CPPUNIT_TEST_SUITE_REGISTRATION( decaf::util::concurrent::FutureTaskTest );
+#include <decaf/util/concurrent/AbstractExecutorServiceTest.h>
+CPPUNIT_TEST_SUITE_REGISTRATION( decaf::util::concurrent::AbstractExecutorServiceTest );
//
//#include <decaf/util/concurrent/atomic/AtomicBooleanTest.h>
//CPPUNIT_TEST_SUITE_REGISTRATION( decaf::util::concurrent::atomic::AtomicBooleanTest );