16 files changed, 936 insertions, 0 deletions
diff --git a/cpp/common/concurrent/inc/APRBase.h b/cpp/common/concurrent/inc/APRBase.h
new file mode 100644
index 0000000000..e0b526faa1
--- /dev/null
+++ b/cpp/common/concurrent/inc/APRBase.h
@@ -0,0 +1,63 @@
+/*
+ *
+ * Copyright (c) 2006 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.
+ *
+ */
+#ifndef _APRBase_
+#define _APRBase_
+
+#include <string>
+#include "apr_thread_mutex.h"
+#include "apr_errno.h"
+
+namespace qpid {
+namespace concurrent {
+
+    /**
+     * Use of APR libraries necessitates explicit init and terminate
+     * calls. Any class using APR libs should obtain the reference to
+     * this singleton and increment on construction, decrement on
+     * destruction. This class can then correctly initialise apr
+     * before the first use and terminate after the last use.
+     */
+    class APRBase{
+	static APRBase* instance;
+	apr_pool_t* pool;
+	apr_thread_mutex_t* mutex;
+	int count;
+
+	APRBase();
+	~APRBase();
+	static APRBase* getInstance();
+	bool _increment();
+	void _decrement();
+    public:
+	static void increment();
+	static void decrement();
+    };
+
+    //this is also a convenient place for a helper function for error checking:
+    void check(apr_status_t status, const std::string& file, const int line);
+    std::string get_desc(apr_status_t status);
+
+#define CHECK_APR_SUCCESS(A) check(A, __FILE__, __LINE__); 
+
+}
+}
+
+
+
+
+#endif
diff --git a/cpp/common/concurrent/inc/APRMonitor.h b/cpp/common/concurrent/inc/APRMonitor.h
new file mode 100644
index 0000000000..bf72596564
--- /dev/null
+++ b/cpp/common/concurrent/inc/APRMonitor.h
@@ -0,0 +1,48 @@
+/*
+ *
+ * Copyright (c) 2006 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.
+ *
+ */
+#ifndef _APRMonitor_
+#define _APRMonitor_
+
+#include "apr_thread_mutex.h"
+#include "apr_thread_cond.h"
+#include "Monitor.h"
+
+namespace qpid {
+namespace concurrent {
+
+    class APRMonitor : public virtual Monitor 
+    {
+	apr_pool_t* pool;
+	apr_thread_mutex_t* mutex;
+	apr_thread_cond_t* condition;
+
+    public:
+	APRMonitor();
+	virtual ~APRMonitor();
+	virtual void wait();
+	virtual void wait(u_int64_t time);
+	virtual void notify();
+	virtual void notifyAll();
+	virtual void acquire();
+	virtual void release();
+    };
+}
+}
+
+
+#endif
diff --git a/cpp/common/concurrent/inc/APRThread.h b/cpp/common/concurrent/inc/APRThread.h
new file mode 100644
index 0000000000..d5034ce3b7
--- /dev/null
+++ b/cpp/common/concurrent/inc/APRThread.h
@@ -0,0 +1,48 @@
+/*
+ *
+ * Copyright (c) 2006 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.
+ *
+ */
+#ifndef _APRThread_
+#define _APRThread_
+
+#include "apr_thread_proc.h"
+#include "APRThread.h"
+#include "Runnable.h"
+#include "Thread.h"
+
+namespace qpid {
+namespace concurrent {
+
+    class APRThread : public virtual Thread
+    {
+	const Runnable* runnable;
+	apr_pool_t* pool;
+	apr_thread_t* runner;
+
+    public:
+	APRThread(apr_pool_t* pool, Runnable* runnable);
+	virtual ~APRThread();
+	virtual void start();
+	virtual void join();
+	virtual void interrupt();
+        static unsigned int currentThread();
+    };
+
+}
+}
+
+
+#endif
diff --git a/cpp/common/concurrent/inc/APRThreadFactory.h b/cpp/common/concurrent/inc/APRThreadFactory.h
new file mode 100644
index 0000000000..87b240025d
--- /dev/null
+++ b/cpp/common/concurrent/inc/APRThreadFactory.h
@@ -0,0 +1,44 @@
+/*
+ *
+ * Copyright (c) 2006 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.
+ *
+ */
+#ifndef _APRThreadFactory_
+#define _APRThreadFactory_
+
+#include "apr_thread_proc.h"
+
+#include "APRThread.h"
+#include "Thread.h"
+#include "ThreadFactory.h"
+#include "Runnable.h"
+
+namespace qpid {
+namespace concurrent {
+
+    class APRThreadFactory : public virtual ThreadFactory
+    {
+	apr_pool_t* pool;
+    public:
+	APRThreadFactory();
+	virtual ~APRThreadFactory();
+	virtual Thread* create(Runnable* runnable);
+    };
+
+}
+}
+
+
+#endif
diff --git a/cpp/common/concurrent/inc/APRThreadPool.h b/cpp/common/concurrent/inc/APRThreadPool.h
new file mode 100644
index 0000000000..cf6d30774c
--- /dev/null
+++ b/cpp/common/concurrent/inc/APRThreadPool.h
@@ -0,0 +1,67 @@
+/*
+ *
+ * Copyright (c) 2006 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.
+ *
+ */
+#ifndef _APRThreadPool_
+#define _APRThreadPool_
+
+#include <queue>
+#include <vector>
+#include "APRMonitor.h"
+#include "Thread.h"
+#include "ThreadFactory.h"
+#include "ThreadPool.h"
+#include "Runnable.h"
+
+namespace qpid {
+namespace concurrent {
+
+    class APRThreadPool : public virtual ThreadPool
+    {
+        class Worker : public virtual Runnable{
+            APRThreadPool* pool;
+        public:
+            inline Worker(APRThreadPool* _pool) : pool(_pool){}
+            inline virtual void run(){
+                while(pool->running){
+                    pool->runTask();
+                }
+            }
+        };
+        const bool deleteFactory;
+        const int size;
+        ThreadFactory* factory;
+        APRMonitor lock; 
+        std::vector<Thread*> threads;
+        std::queue<Runnable*> tasks;
+        Worker* worker;
+        volatile bool running;
+
+        void runTask();
+    public:
+        APRThreadPool(int size);
+        APRThreadPool(int size, ThreadFactory* factory);
+        virtual void start();
+        virtual void stop();
+	virtual void addTask(Runnable* task);
+        virtual ~APRThreadPool();
+    };
+
+}
+}
+
+
+#endif
diff --git a/cpp/common/concurrent/inc/LMonitor.h b/cpp/common/concurrent/inc/LMonitor.h
new file mode 100644
index 0000000000..8e2569921d
--- /dev/null
+++ b/cpp/common/concurrent/inc/LMonitor.h
@@ -0,0 +1,44 @@
+/*
+ *
+ * Copyright (c) 2006 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.
+ *
+ */
+#ifndef _LMonitor_
+#define _LMonitor_
+
+/* Native Linux Monitor - Based of Kernel patch 19/20 */
+
+#include "Monitor.h"
+
+namespace qpid {
+namespace concurrent {
+
+    class LMonitor : public virtual Monitor 
+    {
+
+    public:
+	LMonitor();
+	virtual ~LMonitor();
+	virtual void wait();
+	virtual void notify();
+	virtual void notifyAll();
+	virtual void acquire();
+	virtual void release();
+    };
+}
+}
+
+
+#endif
diff --git a/cpp/common/concurrent/inc/LThreadFactory.h b/cpp/common/concurrent/inc/LThreadFactory.h
new file mode 100644
index 0000000000..4a573d1bd1
--- /dev/null
+++ b/cpp/common/concurrent/inc/LThreadFactory.h
@@ -0,0 +1,37 @@
+/*
+ *
+ * Copyright (c) 2006 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.
+ *
+ */
+#ifndef _LAPRThreadFactory_
+#define _LAPRThreadFactory_
+
+
+namespace qpid {
+namespace concurrent {
+
+    class LThreadFactory 
+    {
+    public:
+	LThreadFactory();
+	virtual ~LThreadFactory();
+	virtual Thread* create(Runnable* runnable);
+    };
+
+}
+}
+
+
+#endif
diff --git a/cpp/common/concurrent/inc/LockedQueue.h b/cpp/common/concurrent/inc/LockedQueue.h
new file mode 100644
index 0000000000..ef3f0b8381
--- /dev/null
+++ b/cpp/common/concurrent/inc/LockedQueue.h
@@ -0,0 +1,68 @@
+/*
+ *
+ * Copyright (c) 2006 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.
+ *
+ */
+#ifndef _LockedQueue_
+#define _LockedQueue_
+
+#include <queue>
+#include "Monitor.h"
+
+/**
+ * A threadsafe queue abstraction
+ */
+namespace qpid {
+namespace concurrent {
+    template<class T, class L> class LockedQueue
+    {
+        L lock;
+        std::queue<T*> queue;
+
+    public:
+        void put(T* item);
+        T* take();
+        bool empty();
+    };
+
+    template<class T, class L> void LockedQueue<T, L>::put(T* item){
+        lock.acquire();
+        queue.push(item);
+        lock.release();
+    }
+
+    template<class T, class L> T* LockedQueue<T, L>::take(){
+        lock.acquire();        
+        T* item = 0;
+        if(!queue.empty()){
+            item = queue.front();
+            queue.pop();
+        }
+        lock.release();
+        return item;
+    }
+
+    template<class T, class L> bool LockedQueue<T, L>::empty(){
+        lock.acquire();
+        bool result = queue.empty();
+        lock.release();
+        return result;
+    }
+
+}
+}
+
+
+#endif
diff --git a/cpp/common/concurrent/inc/Monitor.h b/cpp/common/concurrent/inc/Monitor.h
new file mode 100644
index 0000000000..7f1a299c6a
--- /dev/null
+++ b/cpp/common/concurrent/inc/Monitor.h
@@ -0,0 +1,59 @@
+/*
+ *
+ * Copyright (c) 2006 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.
+ *
+ */
+#ifndef _Monitor_
+#define _Monitor_
+
+#include "amqp_types.h"
+
+namespace qpid {
+namespace concurrent {
+
+class Monitor
+{
+  public:
+    virtual ~Monitor(){}
+    virtual void wait() = 0;
+    virtual void wait(u_int64_t time) = 0;
+    virtual void notify() = 0;
+    virtual void notifyAll() = 0;
+    virtual void acquire() = 0;
+    virtual void release() = 0;
+};
+
+/**
+ * Scoped locker for a monitor.
+ */
+class Locker
+{
+  public:
+    Locker(Monitor&  lock_) : lock(lock_) { lock.acquire(); }
+    ~Locker() { lock.release(); }
+
+  private:
+    Monitor& lock;
+
+    // private and unimplemented to prevent copying
+    Locker(const Locker&);
+    void operator=(const Locker&);
+};
+
+}
+}
+
+
+#endif
diff --git a/cpp/common/concurrent/inc/MonitorImpl.h b/cpp/common/concurrent/inc/MonitorImpl.h
new file mode 100644
index 0000000000..e96e81d795
--- /dev/null
+++ b/cpp/common/concurrent/inc/MonitorImpl.h
@@ -0,0 +1,57 @@
+/*
+ *
+ * Copyright (c) 2006 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.
+ *
+ */
+ 
+ 
+#ifndef _MonitorImpl_
+#define _MonitorImpl_
+
+#ifdef _USE_APR_IO_
+#include "APRMonitor.h"
+#else /* use POSIX Monitor */
+#include "LMonitor.h"  
+#endif
+
+
+namespace qpid {
+namespace concurrent {
+
+#ifdef _USE_APR_IO_
+    class MonitorImpl : public virtual APRMonitor 
+    {
+
+    public:
+	MonitorImpl() : APRMonitor(){};
+	virtual ~MonitorImpl(){};
+
+    };
+#else
+    class MonitorImpl : public virtual LMonitor 
+    {
+
+    public:
+	MonitorImpl() : LMonitor(){};
+	virtual ~MonitorImpl(){};
+
+    };   
+#endif
+    
+}
+}
+
+
+#endif
diff --git a/cpp/common/concurrent/inc/Runnable.h b/cpp/common/concurrent/inc/Runnable.h
new file mode 100644
index 0000000000..523ad813f7
--- /dev/null
+++ b/cpp/common/concurrent/inc/Runnable.h
@@ -0,0 +1,34 @@
+/*
+ *
+ * Copyright (c) 2006 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.
+ *
+ */
+#ifndef _Runnable_
+#define _Runnable_
+
+namespace qpid {
+namespace concurrent {
+
+    class Runnable
+    {
+    public:
+	virtual void run() = 0;
+    };
+
+}
+}
+
+
+#endif
diff --git a/cpp/common/concurrent/inc/TaskQueue.h b/cpp/common/concurrent/inc/TaskQueue.h
new file mode 100644
index 0000000000..e06a3ce069
--- /dev/null
+++ b/cpp/common/concurrent/inc/TaskQueue.h
@@ -0,0 +1,200 @@
+/*
+ *
+ * Copyright (c) 2006 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.
+ *
+ */
+#ifndef _TaskQueue_
+#define _TaskQueue_
+
+#include <iostream>
+#include <memory>
+#include <queue>
+#include "LockedQueue.h"
+#include "Runnable.h"
+#include "ThreadPool.h"
+
+namespace qpid {
+namespace concurrent {
+    template<class T, class L> class TaskQueue : public virtual Runnable
+    {
+        const int max_iterations_per_run;
+        L lock;
+        //LockedQueue<T, L> queue;
+        std::queue<T*> queue;
+        ThreadPool* const pool;
+        T* work;
+        bool running;        
+        volatile bool stopped;        
+        TaskQueue<T, L>* next;
+
+        volatile bool inrun;
+
+        bool hasWork();
+        void completed();
+
+        T* take();
+
+    protected:
+        /**
+         * Callback though which the task is executed 
+         */
+        virtual void execute(T* t) = 0;
+        /**
+         * Allows a task to be completed asynchronously to the
+         * execute() call if required.
+         */
+        virtual bool isComplete(T* t);
+        /**
+         * Should be called to signal completion of a task that was
+         * signalled as not complete through the isComplete() methods
+         * return value. This will allow normal processing to resume.
+         */
+        virtual void complete();
+
+    public:
+        TaskQueue(ThreadPool* const pool, int max_iterations_per_run = 100);
+        virtual void run();
+        void trigger();
+        bool append(T* t);
+        void stop(bool drain);
+        inline void setNext(TaskQueue<T, L>* next){ this->next = next; }
+    };
+
+    template<class T, class L> TaskQueue<T, L>::TaskQueue(ThreadPool* const _pool, int _max_iterations_per_run) : 
+        pool(_pool), 
+        max_iterations_per_run(_max_iterations_per_run),
+        work(0), 
+        running(false),
+        stopped(false),
+        next(0), inrun(false){
+    }
+
+    template<class T, class L> void TaskQueue<T, L>::run(){
+        if(inrun) std::cout << "Already running" << std::endl;
+        inrun = true;
+
+        bool blocked = false;
+        int count = max_iterations_per_run;
+        while(!blocked && hasWork() && count){
+            execute(work);
+            if(isComplete(work)){
+                completed();
+            }else{
+                blocked = true;
+            }
+            count--;
+        }
+        inrun = false;
+
+        if(!blocked && count == 0){//performed max_iterations_per_run, requeue task to ensure fairness 
+            //running will still be true at this point
+            lock.acquire();
+            running = false;
+            if(stopped) lock.notify();
+            lock.release();
+
+            trigger();
+        }else if(hasWork()){//task was added to queue after we exited the loop above; should not need this?
+            trigger();
+        }
+    }
+
+    template<class T, class L> void TaskQueue<T, L>::trigger(){
+        lock.acquire();
+        if(!running){
+            running = true;
+            pool->addTask(this);
+        }
+        lock.release();
+    }
+
+    template<class T, class L> bool TaskQueue<T, L>::hasWork(){
+        lock.acquire();
+        if(!work) work = take();//queue.take();
+        if(!work){
+            running = false;
+            if(stopped) lock.notify();
+        }
+        lock.release();
+        return work;
+    }
+
+    template<class T, class L> bool TaskQueue<T, L>::append(T* item){
+        if(!stopped){
+            lock.acquire();
+
+            //queue.put(item);
+            queue.push(item);
+
+            if(!running){
+                running = true;
+                pool->addTask(this);
+            }
+            lock.release();
+                //}
+            return true;
+        }else{
+            return false;
+        }
+    }
+
+    template<class T, class L> bool TaskQueue<T, L>::isComplete(T* item){
+        return true;//by default assume all tasks are synchronous w.r.t. execute()
+    }
+
+
+    template<class T, class L> void TaskQueue<T, L>::completed(){
+        if(next){
+            if(!next->append(work)){
+                std::cout << "Warning: dropping task as next queue appears to have stopped." << std::endl;
+            }
+        }else{
+            delete work;
+        }
+        work = 0;        
+    }
+        
+    template<class T, class L> void TaskQueue<T, L>::complete(){
+        completed();
+        lock.acquire();
+        running = false;
+        if(stopped) lock.notify();
+        lock.release();
+    }
+
+    template<class T, class L> void TaskQueue<T, L>::stop(bool drain){
+        //prevent new tasks from being added
+        stopped = true;
+        //wait until no longer running
+        lock.acquire();
+        while(running && (drain && hasWork())){
+            lock.wait();
+        }
+        lock.release();
+    }
+
+    template<class T, class L> T* TaskQueue<T, L>::take(){
+        T* item = 0;
+        if(!queue.empty()){
+            item = queue.front();
+            queue.pop();
+        }
+        return item;
+    }
+}
+}
+
+
+#endif
diff --git a/cpp/common/concurrent/inc/Thread.h b/cpp/common/concurrent/inc/Thread.h
new file mode 100644
index 0000000000..6bd2a379ce
--- /dev/null
+++ b/cpp/common/concurrent/inc/Thread.h
@@ -0,0 +1,37 @@
+/*
+ *
+ * Copyright (c) 2006 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.
+ *
+ */
+#ifndef _Thread_
+#define _Thread_
+
+namespace qpid {
+namespace concurrent {
+
+    class Thread
+    {
+    public:
+        virtual ~Thread(){}
+	virtual void start() = 0;
+	virtual void join() = 0;
+	virtual void interrupt() = 0;
+    };
+
+}
+}
+
+
+#endif
diff --git a/cpp/common/concurrent/inc/ThreadFactory.h b/cpp/common/concurrent/inc/ThreadFactory.h
new file mode 100644
index 0000000000..53be000ff3
--- /dev/null
+++ b/cpp/common/concurrent/inc/ThreadFactory.h
@@ -0,0 +1,38 @@
+/*
+ *
+ * Copyright (c) 2006 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.
+ *
+ */
+#ifndef _ThreadFactory_
+#define _ThreadFactory_
+
+#include "Thread.h"
+#include "Runnable.h"
+
+namespace qpid {
+namespace concurrent {
+
+    class ThreadFactory
+    {
+    public:
+        virtual ~ThreadFactory(){}
+	virtual Thread* create(Runnable* runnable) = 0;
+    };
+
+}
+}
+
+
+#endif
diff --git a/cpp/common/concurrent/inc/ThreadFactoryImpl.h b/cpp/common/concurrent/inc/ThreadFactoryImpl.h
new file mode 100644
index 0000000000..a534b3c1e2
--- /dev/null
+++ b/cpp/common/concurrent/inc/ThreadFactoryImpl.h
@@ -0,0 +1,52 @@
+/*
+ *
+ * Copyright (c) 2006 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.
+ *
+ */
+#ifndef _ThreadFactoryImpl_
+#define _ThreadFactoryImpl_
+
+
+#ifdef _USE_APR_IO_
+#include "APRThreadFactory.h"
+#else
+#include "LThreadFactory.h"
+#endif
+
+
+namespace qpid {
+namespace concurrent {
+
+
+#ifdef _USE_APR_IO_
+    class ThreadFactoryImpl : public virtual APRThreadFactory
+    {
+    public:
+	ThreadFactoryImpl(): APRThreadFactory() {};
+	virtual ~ThreadFactoryImpl() {};
+    };
+#else
+    class ThreadFactoryImpl : public virtual LThreadFactory
+    {
+    public:
+	ThreadFactoryImpl(): LThreadFactory() {};
+	virtual ~ThreadFactoryImpl() {};
+    };
+#endif
+}
+}
+
+
+#endif
diff --git a/cpp/common/concurrent/inc/ThreadPool.h b/cpp/common/concurrent/inc/ThreadPool.h
new file mode 100644
index 0000000000..679c889ff3
--- /dev/null
+++ b/cpp/common/concurrent/inc/ThreadPool.h
@@ -0,0 +1,40 @@
+/*
+ *
+ * Copyright (c) 2006 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.
+ *
+ */
+#ifndef _ThreadPool_
+#define _ThreadPool_
+
+#include "Thread.h"
+#include "Runnable.h"
+
+namespace qpid {
+namespace concurrent {
+
+    class ThreadPool
+    {
+    public:
+        virtual void start() = 0;
+        virtual void stop() = 0;
+	virtual void addTask(Runnable* runnable) = 0;
+        virtual ~ThreadPool(){}
+    };
+
+}
+}
+
+
+#endif