More reorg to separate APR/posix code, work in progress.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@472545 13f79535-47bb-0310-9956-ffa450edef68

19 files changed, 1608 insertions, 0 deletions

diff --git a/cpp/src/qpid/apr/APRBase.cpp b/cpp/src/qpid/apr/APRBase.cpp
new file mode 100644
index 0000000000..f629a5381d
--- /dev/null
+++ b/cpp/src/qpid/apr/APRBase.cpp
@@ -0,0 +1,96 @@
+/*
+ *
+ * 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.
+ *
+ */
+#include <iostream>
+#include <qpid/QpidError.h>
+#include "APRBase.h"
+
+using namespace qpid::sys;
+
+APRBase* APRBase::instance = 0;
+
+APRBase* APRBase::getInstance(){
+    if(instance == 0){
+	instance = new APRBase();
+    }
+    return instance;
+}
+
+
+APRBase::APRBase() : count(0){
+    apr_initialize();
+    CHECK_APR_SUCCESS(apr_pool_create(&pool, 0));
+    CHECK_APR_SUCCESS(apr_thread_mutex_create(&mutex, APR_THREAD_MUTEX_NESTED, pool));
+}
+
+APRBase::~APRBase(){
+    CHECK_APR_SUCCESS(apr_thread_mutex_destroy(mutex));
+    apr_pool_destroy(pool);
+    apr_terminate();  
+}
+
+bool APRBase::_increment(){
+    bool deleted(false);
+    CHECK_APR_SUCCESS(apr_thread_mutex_lock(mutex));
+    if(this == instance){
+	count++;
+    }else{
+	deleted = true;
+    }
+    CHECK_APR_SUCCESS(apr_thread_mutex_unlock(mutex));
+    return !deleted;
+}
+
+void APRBase::_decrement(){
+    APRBase* copy = 0;
+    CHECK_APR_SUCCESS(apr_thread_mutex_lock(mutex));
+    if(--count == 0){
+	copy = instance;
+	instance = 0;
+    }
+    CHECK_APR_SUCCESS(apr_thread_mutex_unlock(mutex));
+    if(copy != 0){
+	delete copy;
+    }
+}
+
+void APRBase::increment(){
+    int count = 0;
+    while(count++ < 2 && !getInstance()->_increment()){
+        std::cout << "WARNING: APR initialization triggered concurrently with termination." << std::endl;
+    }
+}
+
+void APRBase::decrement(){
+    getInstance()->_decrement();
+}
+
+void qpid::sys::check(apr_status_t status, const std::string& file, const int line){
+    if (status != APR_SUCCESS){
+        const int size = 50;
+        char tmp[size];
+        std::string msg(apr_strerror(status, tmp, size));
+        throw QpidError(APR_ERROR + ((int) status), msg, file, line);
+    }
+}
+
+std::string qpid::sys::get_desc(apr_status_t status){
+    const int size = 50;
+    char tmp[size];
+    return std::string(apr_strerror(status, tmp, size));
+}
+
diff --git a/cpp/src/qpid/apr/APRBase.h b/cpp/src/qpid/apr/APRBase.h
new file mode 100644
index 0000000000..b84e9860df
--- /dev/null
+++ b/cpp/src/qpid/apr/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-1/apr_thread_mutex.h"
+#include "apr-1/apr_errno.h"
+
+namespace qpid {
+namespace sys {
+
+    /**
+     * 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) qpid::sys::check(A, __FILE__, __LINE__); 
+
+}
+}
+
+
+
+
+#endif
diff --git a/cpp/src/qpid/apr/APRPool.cpp b/cpp/src/qpid/apr/APRPool.cpp
new file mode 100644
index 0000000000..e465a6b40d
--- /dev/null
+++ b/cpp/src/qpid/apr/APRPool.cpp
@@ -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.
+ *
+ */
+
+#include "APRPool.h"
+#include "APRBase.h"
+#include <boost/pool/detail/singleton.hpp>
+
+using namespace qpid::sys;
+
+APRPool::APRPool(){
+    APRBase::increment();
+    CHECK_APR_SUCCESS(apr_pool_create(&pool, NULL));
+}
+
+APRPool::~APRPool(){
+    apr_pool_destroy(pool);
+    APRBase::decrement();
+}
+
+apr_pool_t* APRPool::get() {
+    return boost::details::pool::singleton_default<APRPool>::instance().pool;
+}
+
diff --git a/cpp/src/qpid/apr/APRPool.h b/cpp/src/qpid/apr/APRPool.h
new file mode 100644
index 0000000000..2196cd64e7
--- /dev/null
+++ b/cpp/src/qpid/apr/APRPool.h
@@ -0,0 +1,47 @@
+#ifndef _APRPool_
+#define _APRPool_
+
+/*
+ *
+ * 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.
+ *
+ */
+#include <boost/noncopyable.hpp>
+#include <apr-1/apr_pools.h>
+
+namespace qpid {
+namespace sys {
+/**
+ * Singleton APR memory pool.
+ */
+class APRPool : private boost::noncopyable {
+  public:
+    APRPool();
+    ~APRPool();
+
+    /** Get singleton instance */
+    static apr_pool_t* get();
+
+  private:
+    apr_pool_t* pool;
+};
+
+}}
+
+
+
+
+
+#endif  /*!_APRPool_*/
diff --git a/cpp/src/qpid/apr/APRSocket.cpp b/cpp/src/qpid/apr/APRSocket.cpp
new file mode 100644
index 0000000000..0c5a29c216
--- /dev/null
+++ b/cpp/src/qpid/apr/APRSocket.cpp
@@ -0,0 +1,75 @@
+/*
+ *
+ * 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.
+ *
+ */
+#include "APRBase.h"
+#include "APRSocket.h"
+#include <assert.h>
+#include <iostream>
+
+using namespace qpid::sys;
+using namespace qpid::framing;
+
+APRSocket::APRSocket(apr_socket_t* _socket) : socket(_socket), closed(false){
+
+}
+
+void APRSocket::read(qpid::framing::Buffer& buffer){
+    apr_size_t bytes;
+    bytes = buffer.available();
+    apr_status_t s = apr_socket_recv(socket, buffer.start(), &bytes);
+    buffer.move(bytes);
+    if(APR_STATUS_IS_TIMEUP(s)){
+        //timed out
+    }else if(APR_STATUS_IS_EOF(s)){
+        close();
+    }
+}
+
+void APRSocket::write(qpid::framing::Buffer& buffer){
+    apr_size_t bytes;
+    do{
+        bytes = buffer.available();
+        apr_socket_send(socket, buffer.start(), &bytes);
+        buffer.move(bytes);    
+    }while(bytes > 0);
+}
+
+void APRSocket::close(){
+    if(!closed){
+        std::cout << "Closing socket " << socket << "@" << this << std::endl;
+        CHECK_APR_SUCCESS(apr_socket_close(socket));
+        closed = true;
+    }
+}
+
+bool APRSocket::isOpen(){
+    return !closed;
+}
+
+u_int8_t APRSocket::read(){
+    char data[1];
+    apr_size_t bytes = 1;
+    apr_status_t s = apr_socket_recv(socket, data, &bytes);
+    if(APR_STATUS_IS_EOF(s) || bytes == 0){
+        return 0;
+    }else{
+        return *data;
+    }
+}
+
+APRSocket::~APRSocket(){
+}
diff --git a/cpp/src/qpid/apr/APRSocket.h b/cpp/src/qpid/apr/APRSocket.h
new file mode 100644
index 0000000000..f7e7ad107b
--- /dev/null
+++ b/cpp/src/qpid/apr/APRSocket.h
@@ -0,0 +1,45 @@
+/*
+ *
+ * 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 _APRSocket_
+#define _APRSocket_
+
+#include "apr-1/apr_network_io.h"
+#include "qpid/framing/Buffer.h"
+
+namespace qpid {
+namespace sys {
+
+    class APRSocket
+    {
+	apr_socket_t* const socket;
+        volatile bool closed;
+    public:
+	APRSocket(apr_socket_t* socket);
+        void read(qpid::framing::Buffer& b);
+        void write(qpid::framing::Buffer& b);
+        void close();
+        bool isOpen();
+        u_int8_t read();
+	~APRSocket();
+    };
+
+}
+}
+
+
+#endif
diff --git a/cpp/src/qpid/apr/Acceptor.cpp b/cpp/src/qpid/apr/Acceptor.cpp
new file mode 100644
index 0000000000..cbeea9902b
--- /dev/null
+++ b/cpp/src/qpid/apr/Acceptor.cpp
@@ -0,0 +1,77 @@
+/*
+ *
+ * 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.
+ *
+ */
+#include "Acceptor.h"
+#include "APRBase.h"
+#include "APRPool.h"
+
+using namespace qpid::sys;
+
+Acceptor::Acceptor(int16_t port_, int backlog, int threads) :
+    port(port_),
+    processor(APRPool::get(), threads, 1000, 5000000)
+{
+    apr_sockaddr_t* address;
+    CHECK_APR_SUCCESS(apr_sockaddr_info_get(&address, APR_ANYADDR, APR_UNSPEC, port, APR_IPV4_ADDR_OK, APRPool::get()));
+    CHECK_APR_SUCCESS(apr_socket_create(&socket, APR_INET, SOCK_STREAM, APR_PROTO_TCP, APRPool::get()));
+    CHECK_APR_SUCCESS(apr_socket_opt_set(socket, APR_SO_REUSEADDR, 1));
+    CHECK_APR_SUCCESS(apr_socket_bind(socket, address));
+    CHECK_APR_SUCCESS(apr_socket_listen(socket, backlog));
+}
+
+int16_t Acceptor::getPort() const {
+    apr_sockaddr_t* address;
+    CHECK_APR_SUCCESS(apr_socket_addr_get(&address, APR_LOCAL, socket));
+    return address->port;
+}
+
+void Acceptor::run(SessionHandlerFactory* factory) {
+    running = true;
+    processor.start();
+    std::cout << "Listening on port " << getPort() << "..." << std::endl;
+    while(running){
+        apr_socket_t* client;
+        apr_status_t status = apr_socket_accept(&client, socket, APRPool::get());
+        if(status == APR_SUCCESS){
+            //make this socket non-blocking:
+            CHECK_APR_SUCCESS(apr_socket_timeout_set(client, 0));
+            CHECK_APR_SUCCESS(apr_socket_opt_set(client, APR_SO_NONBLOCK, 1));
+            CHECK_APR_SUCCESS(apr_socket_opt_set(client, APR_TCP_NODELAY, 1));
+            CHECK_APR_SUCCESS(apr_socket_opt_set(client, APR_SO_SNDBUF, 32768));
+            CHECK_APR_SUCCESS(apr_socket_opt_set(client, APR_SO_RCVBUF, 32768));
+            LFSessionContext* session = new LFSessionContext(APRPool::get(), client, &processor, false);
+            session->init(factory->create(session));
+        }else{
+            running = false;
+            if(status != APR_EINTR){
+                std::cout << "ERROR: " << get_desc(status) << std::endl;
+            }
+        }
+    }
+    shutdown();
+}
+
+void Acceptor::shutdown() {
+    // TODO aconway 2006-10-12: Cleanup, this is not thread safe.
+    if (running) {
+        running = false;
+        processor.stop();
+        CHECK_APR_SUCCESS(apr_socket_close(socket));
+    }
+}
+
+
diff --git a/cpp/src/qpid/apr/Acceptor.h b/cpp/src/qpid/apr/Acceptor.h
new file mode 100644
index 0000000000..1813b391c1
--- /dev/null
+++ b/cpp/src/qpid/apr/Acceptor.h
@@ -0,0 +1,55 @@
+/*
+ *
+ * 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 _LFAcceptor_
+#define _LFAcceptor_
+
+#include "LFProcessor.h"
+#include "LFSessionContext.h"
+#include "apr-1/apr_network_io.h"
+#include "apr-1/apr_poll.h"
+#include "apr-1/apr_time.h"
+#include "Monitor.h"
+#include "qpid/sys/Runnable.h"
+#include "qpid/sys/SessionContext.h"
+#include "qpid/sys/SessionHandlerFactory.h"
+#include <qpid/SharedObject.h>
+
+namespace qpid {
+namespace sys {
+
+/** APR Acceptor. */
+class Acceptor : public qpid::SharedObject<Acceptor>
+{
+  public:
+    Acceptor(int16_t port, int backlog, int threads);
+    virtual int16_t getPort() const;
+    virtual void run(qpid::sys::SessionHandlerFactory* factory);
+    virtual void shutdown();
+
+  private:
+    int16_t port;
+    LFProcessor processor;
+    apr_socket_t* socket;
+    volatile bool running;
+};
+
+}
+}
+
+
+#endif
diff --git a/cpp/src/qpid/apr/Connector.cpp b/cpp/src/qpid/apr/Connector.cpp
new file mode 100644
index 0000000000..4446731654
--- /dev/null
+++ b/cpp/src/qpid/apr/Connector.cpp
@@ -0,0 +1,190 @@
+/*
+ *
+ * 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.
+ *
+ */
+#include <iostream>
+#include <qpid/QpidError.h>
+#include "APRBase.h"
+#include "Connector.h"
+
+using namespace qpid::sys;
+using namespace qpid::sys;
+using namespace qpid::framing;
+using qpid::QpidError;
+
+Connector::Connector(bool _debug, u_int32_t buffer_size) :
+    debug(_debug), 
+    receive_buffer_size(buffer_size),
+    send_buffer_size(buffer_size),
+    closed(true),
+    lastIn(0), lastOut(0),
+    timeout(0),
+    idleIn(0), idleOut(0), 
+    timeoutHandler(0),
+    shutdownHandler(0),
+    inbuf(receive_buffer_size), 
+    outbuf(send_buffer_size){
+
+    APRBase::increment();
+
+    CHECK_APR_SUCCESS(apr_pool_create(&pool, NULL));
+    CHECK_APR_SUCCESS(apr_socket_create(&socket, APR_INET, SOCK_STREAM, APR_PROTO_TCP, pool));
+}
+
+Connector::~Connector(){
+    apr_pool_destroy(pool);
+
+    APRBase::decrement();
+}
+
+void Connector::connect(const std::string& host, int port){
+    apr_sockaddr_t* address;
+    CHECK_APR_SUCCESS(apr_sockaddr_info_get(&address, host.c_str(), APR_UNSPEC, port, APR_IPV4_ADDR_OK, pool));
+    CHECK_APR_SUCCESS(apr_socket_connect(socket, address));
+    closed = false;
+    receiver = Thread(this);
+}
+
+void Connector::init(ProtocolInitiation* header){
+    writeBlock(header);
+    delete header;
+}
+
+void Connector::close(){
+    closed = true;
+    CHECK_APR_SUCCESS(apr_socket_close(socket));
+    receiver.join();
+}
+
+void Connector::setInputHandler(InputHandler* handler){
+    input = handler;
+}
+
+void Connector::setShutdownHandler(ShutdownHandler* handler){
+    shutdownHandler = handler;
+}
+
+OutputHandler* Connector::getOutputHandler(){ 
+    return this; 
+}
+
+void Connector::send(AMQFrame* frame){
+    writeBlock(frame);    
+    if(debug) std::cout << "SENT: " << *frame << std::endl; 
+    delete frame;
+}
+
+void Connector::writeBlock(AMQDataBlock* data){
+    Mutex::ScopedLock l(writeLock);
+    data->encode(outbuf);
+    //transfer data to wire
+    outbuf.flip();
+    writeToSocket(outbuf.start(), outbuf.available());
+    outbuf.clear();
+}
+
+void Connector::writeToSocket(char* data, size_t available){
+    apr_size_t bytes(available);
+    apr_size_t written(0);
+    while(written < available && !closed){
+	apr_status_t status = apr_socket_send(socket, data + written, &bytes);
+        if(status == APR_TIMEUP){
+            std::cout << "Write request timed out." << std::endl;
+        }
+        if(bytes == 0){
+            std::cout << "Write request wrote 0 bytes." << std::endl;
+        }
+        lastOut = apr_time_as_msec(apr_time_now());
+	written += bytes;
+	bytes = available - written;
+    }
+}
+
+void Connector::checkIdle(apr_status_t status){
+    if(timeoutHandler){
+        int64_t now = apr_time_as_msec(apr_time_now());
+        if(APR_STATUS_IS_TIMEUP(status)){
+            if(idleIn && (now - lastIn > idleIn)){
+                timeoutHandler->idleIn();
+            }
+        }else if(APR_STATUS_IS_EOF(status)){
+            closed = true;
+            CHECK_APR_SUCCESS(apr_socket_close(socket));
+            if(shutdownHandler) shutdownHandler->shutdown();
+        }else{
+            lastIn = now;
+        }
+        if(idleOut && (now - lastOut > idleOut)){
+            timeoutHandler->idleOut();
+        }
+    }
+}
+
+void Connector::setReadTimeout(u_int16_t t){
+    idleIn = t * 1000;//t is in secs
+    if(idleIn && (!timeout || idleIn < timeout)){
+        timeout = idleIn;
+        setSocketTimeout();
+    }
+
+}
+
+void Connector::setWriteTimeout(u_int16_t t){
+    idleOut = t * 1000;//t is in secs
+    if(idleOut && (!timeout || idleOut < timeout)){
+        timeout = idleOut;
+        setSocketTimeout();
+    }
+}
+
+void Connector::setSocketTimeout(){
+    //interval is in microseconds, timeout in milliseconds
+    //want the interval to be a bit shorter than the timeout, hence multiply
+    //by 800 rather than 1000.
+    apr_interval_time_t interval(timeout * 800);
+    apr_socket_timeout_set(socket, interval);
+}
+
+void Connector::setTimeoutHandler(TimeoutHandler* handler){
+    timeoutHandler = handler;
+}
+
+void Connector::run(){
+    try{
+	while(!closed){
+	    apr_size_t bytes(inbuf.available());
+            if(bytes < 1){
+                THROW_QPID_ERROR(INTERNAL_ERROR, "Frame exceeds buffer size.");
+            }
+	    checkIdle(apr_socket_recv(socket, inbuf.start(), &bytes));
+
+	    if(bytes > 0){
+		inbuf.move(bytes);
+		inbuf.flip();//position = 0, limit = total data read
+		
+		AMQFrame frame;
+		while(frame.decode(inbuf)){
+                    if(debug) std::cout << "RECV: " << frame << std::endl; 
+		    input->received(&frame);
+		}
+                //need to compact buffer to preserve any 'extra' data
+                inbuf.compact();
+	    }
+	}
+    }catch(QpidError error){
+	std::cout << "Error [" << error.code << "] " << error.msg << " (" << error.file << ":" << error.line << ")" << std::endl;
+    }
+}
diff --git a/cpp/src/qpid/apr/Connector.h b/cpp/src/qpid/apr/Connector.h
new file mode 100644
index 0000000000..e69a7205f3
--- /dev/null
+++ b/cpp/src/qpid/apr/Connector.h
@@ -0,0 +1,93 @@
+/*
+ *
+ * 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 _Connector_
+#define _Connector_
+
+#include "apr-1/apr_network_io.h"
+#include "apr-1/apr_time.h"
+
+#include "qpid/framing/InputHandler.h"
+#include "qpid/framing/OutputHandler.h"
+#include "qpid/framing/InitiationHandler.h"
+#include "qpid/framing/ProtocolInitiation.h"
+#include "qpid/sys/ShutdownHandler.h"
+#include "qpid/sys/TimeoutHandler.h"
+#include "qpid/sys/Thread.h"
+#include "qpid/sys/Connector.h"
+#include "qpid/sys/Monitor.h"
+
+namespace qpid {
+namespace sys {
+
+    class Connector : public qpid::framing::OutputHandler, 
+                      private qpid::sys::Runnable
+    {
+        const bool debug;
+	const int receive_buffer_size;
+	const int send_buffer_size;
+
+	bool closed;
+
+        int64_t lastIn;
+        int64_t lastOut;
+        int64_t timeout;
+        u_int32_t idleIn;
+        u_int32_t idleOut;
+
+        TimeoutHandler* timeoutHandler;
+        ShutdownHandler* shutdownHandler;
+	qpid::framing::InputHandler* input;
+	qpid::framing::InitiationHandler* initialiser;
+	qpid::framing::OutputHandler* output;
+	
+	qpid::framing::Buffer inbuf;
+	qpid::framing::Buffer outbuf;
+
+        qpid::sys::Mutex writeLock;
+	qpid::sys::Thread receiver;
+
+	apr_pool_t* pool;
+	apr_socket_t* socket;
+
+        void checkIdle(apr_status_t status);
+	void writeBlock(qpid::framing::AMQDataBlock* data);
+	void writeToSocket(char* data, size_t available);
+        void setSocketTimeout();
+
+	void run();
+
+    public:
+	Connector(bool debug = false, u_int32_t buffer_size = 1024);
+	virtual ~Connector();
+	virtual void connect(const std::string& host, int port);
+	virtual void init(qpid::framing::ProtocolInitiation* header);
+	virtual void close();
+	virtual void setInputHandler(qpid::framing::InputHandler* handler);
+	virtual void setTimeoutHandler(TimeoutHandler* handler);
+	virtual void setShutdownHandler(ShutdownHandler* handler);
+	virtual qpid::framing::OutputHandler* getOutputHandler();
+	virtual void send(qpid::framing::AMQFrame* frame);
+        virtual void setReadTimeout(u_int16_t timeout);
+        virtual void setWriteTimeout(u_int16_t timeout);
+    };
+
+}
+}
+
+
+#endif
diff --git a/cpp/src/qpid/apr/LFProcessor.cpp b/cpp/src/qpid/apr/LFProcessor.cpp
new file mode 100644
index 0000000000..f4d4258f6f
--- /dev/null
+++ b/cpp/src/qpid/apr/LFProcessor.cpp
@@ -0,0 +1,176 @@
+/*
+ *
+ * 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.
+ *
+ */
+#include <sstream>
+#include <qpid/QpidError.h>
+#include "LFProcessor.h"
+#include "APRBase.h"
+#include "LFSessionContext.h"
+
+using namespace qpid::sys;
+using qpid::QpidError;
+
+// TODO aconway 2006-10-12: stopped is read outside locks.
+//
+
+LFProcessor::LFProcessor(apr_pool_t* pool, int _workers, int _size, int _timeout) :
+    size(_size),
+    timeout(_timeout), 
+    signalledCount(0),
+    current(0),
+    count(0),
+    workerCount(_workers),
+    hasLeader(false),
+    workers(new Thread[_workers]),
+    stopped(false)
+{
+
+    CHECK_APR_SUCCESS(apr_pollset_create(&pollset, size, pool, APR_POLLSET_THREADSAFE));
+}
+
+
+LFProcessor::~LFProcessor(){
+    if (!stopped) stop();
+    delete[] workers;
+    CHECK_APR_SUCCESS(apr_pollset_destroy(pollset));
+}
+
+void LFProcessor::start(){
+    for(int i = 0; i < workerCount; i++){
+        workers[i] = Thread(this);
+    }
+}
+
+void LFProcessor::add(const apr_pollfd_t* const fd){
+    CHECK_APR_SUCCESS(apr_pollset_add(pollset, fd));
+    Monitor::ScopedLock l(countLock);
+    sessions.push_back(reinterpret_cast<LFSessionContext*>(fd->client_data));
+    count++;
+}
+
+void LFProcessor::remove(const apr_pollfd_t* const fd){
+    CHECK_APR_SUCCESS(apr_pollset_remove(pollset, fd));
+    Monitor::ScopedLock l(countLock);
+    sessions.erase(find(sessions.begin(), sessions.end(), reinterpret_cast<LFSessionContext*>(fd->client_data)));
+    count--;
+}
+
+void LFProcessor::reactivate(const apr_pollfd_t* const fd){
+    CHECK_APR_SUCCESS(apr_pollset_add(pollset, fd));
+}
+
+void LFProcessor::deactivate(const apr_pollfd_t* const fd){
+    CHECK_APR_SUCCESS(apr_pollset_remove(pollset, fd));
+}
+
+void LFProcessor::update(const apr_pollfd_t* const fd){
+    CHECK_APR_SUCCESS(apr_pollset_remove(pollset, fd));
+    CHECK_APR_SUCCESS(apr_pollset_add(pollset, fd));
+}
+
+bool LFProcessor::full(){
+    Mutex::ScopedLock locker(countLock);
+    return count == size; 
+}
+
+bool LFProcessor::empty(){
+    Mutex::ScopedLock locker(countLock);
+    return count == 0; 
+}
+
+void LFProcessor::poll() {
+    apr_status_t status = APR_EGENERAL;
+    do{
+        current = 0;
+        if(!stopped){
+            status = apr_pollset_poll(pollset, timeout, &signalledCount, &signalledFDs);
+        }
+    }while(status != APR_SUCCESS && !stopped);
+}
+
+void LFProcessor::run(){
+    try{
+        while(!stopped){
+            const apr_pollfd_t* event = 0;
+            LFSessionContext* session = 0;
+            {
+                Monitor::ScopedLock l(leadLock);
+                waitToLead();
+                event = getNextEvent();
+                if(!event) return;
+                session = reinterpret_cast<LFSessionContext*>(
+                    event->client_data);
+                session->startProcessing();
+                relinquishLead();
+            }
+
+            //process event:
+            if(event->rtnevents & APR_POLLIN) session->read();
+            if(event->rtnevents & APR_POLLOUT) session->write();
+
+            if(session->isClosed()){
+                session->handleClose();
+                Monitor::ScopedLock l(countLock);
+                sessions.erase(find(sessions.begin(),sessions.end(), session));
+                count--;
+            }else{
+                session->stopProcessing();
+            }
+        }
+    }catch(QpidError error){
+	std::cout << "Error [" << error.code << "] " << error.msg << " (" << error.file << ":" << error.line << ")" << std::endl;
+    }
+}
+
+void LFProcessor::waitToLead(){
+    while(hasLeader && !stopped) leadLock.wait();
+    hasLeader = !stopped;
+}
+
+void LFProcessor::relinquishLead(){
+    hasLeader = false;
+    leadLock.notify();
+}
+
+const apr_pollfd_t* LFProcessor::getNextEvent(){
+    while(true){
+        if(stopped){
+            return 0;
+        }else if(current < signalledCount){
+            //use result of previous poll if one is available
+            return signalledFDs + (current++);
+        }else{
+            //else poll to get new events
+            poll();
+        }
+    }
+}
+
+void LFProcessor::stop(){
+    stopped = true;
+    {
+        Monitor::ScopedLock l(leadLock);
+        leadLock.notifyAll();
+    }
+    for(int i = 0; i < workerCount; i++){
+        workers[i].join();
+    }
+    for(iterator i = sessions.begin(); i < sessions.end(); i++){
+        (*i)->shutdown();
+    }
+}
+
diff --git a/cpp/src/qpid/apr/LFProcessor.h b/cpp/src/qpid/apr/LFProcessor.h
new file mode 100644
index 0000000000..dd85ad9e84
--- /dev/null
+++ b/cpp/src/qpid/apr/LFProcessor.h
@@ -0,0 +1,118 @@
+/*
+ *
+ * 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 _LFProcessor_
+#define _LFProcessor_
+
+#include "apr-1/apr_poll.h"
+#include <iostream>
+#include <vector>
+#include <qpid/sys/Monitor.h>
+#include "qpid/sys/Runnable.h"
+#include "qpid/sys/Thread.h"
+
+namespace qpid {
+namespace sys {
+
+    class LFSessionContext;
+
+    /**
+     * This class processes a poll set using the leaders-followers
+     * pattern for thread synchronization: the leader will poll and on
+     * the poll returning, it will remove a session, promote a
+     * follower to leadership, then process the session.
+     */
+    class LFProcessor : private virtual qpid::sys::Runnable
+    {
+        typedef std::vector<LFSessionContext*>::iterator iterator;
+        
+        const int size;
+        const apr_interval_time_t timeout;
+        apr_pollset_t* pollset;
+        int signalledCount;
+        int current;
+        const apr_pollfd_t* signalledFDs;
+        int count;
+        const int workerCount;
+        bool hasLeader;
+        qpid::sys::Thread* workers;
+        qpid::sys::Monitor leadLock;
+        qpid::sys::Mutex countLock;
+        std::vector<LFSessionContext*> sessions;
+        volatile bool stopped;
+
+        const apr_pollfd_t* getNextEvent();
+        void waitToLead();
+        void relinquishLead();
+        void poll();        
+        virtual void run();        
+
+    public:
+        LFProcessor(apr_pool_t* pool, int workers, int size, int timeout);
+        /**
+         * Add the fd to the poll set. Relies on the client_data being
+         * an instance of LFSessionContext.
+         */
+        void add(const apr_pollfd_t* const fd);
+        /**
+         * Remove the fd from the poll set.
+         */
+        void remove(const apr_pollfd_t* const fd);
+        /**
+         * Signal that the fd passed in, already part of the pollset,
+         * has had its flags altered.
+         */
+        void update(const apr_pollfd_t* const fd);
+        /**
+         * Add an fd back to the poll set after deactivation.
+         */
+        void reactivate(const apr_pollfd_t* const fd);
+        /**
+         * Temporarily remove the fd from the poll set. Called when processing
+         * is about to begin.
+         */
+        void deactivate(const apr_pollfd_t* const fd);
+        /**
+         * Indicates whether the capacity of this processor has been
+         * reached (or whether it can still handle further fd's).
+         */
+        bool full();
+        /**
+         * Indicates whether there are any fd's registered.
+         */
+        bool empty();
+        /**
+         * Stop processing.
+         */
+        void stop();
+        /**
+         * Start processing.
+         */
+        void start();
+        /**
+         * Is processing stopped?
+         */
+        bool isStopped();
+        
+	~LFProcessor();
+    };
+
+}
+}
+
+
+#endif
diff --git a/cpp/src/qpid/apr/LFSessionContext.cpp b/cpp/src/qpid/apr/LFSessionContext.cpp
new file mode 100644
index 0000000000..4a704013a8
--- /dev/null
+++ b/cpp/src/qpid/apr/LFSessionContext.cpp
@@ -0,0 +1,170 @@
+/*
+ *
+ * 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.
+ *
+ */
+#include "LFSessionContext.h"
+#include "APRBase.h"
+#include "qpid/QpidError.h"
+#include <assert.h>
+
+using namespace qpid::sys;
+using namespace qpid::sys;
+using namespace qpid::framing;
+
+LFSessionContext::LFSessionContext(apr_pool_t* _pool, apr_socket_t* _socket, 
+                                   LFProcessor* const _processor,
+                                   bool _debug) :
+    debug(_debug),
+    socket(_socket),
+    initiated(false),
+    in(32768),
+    out(32768),
+    processor(_processor),
+    processing(false),
+    closing(false)
+{
+    
+    fd.p = _pool;
+    fd.desc_type = APR_POLL_SOCKET;
+    fd.reqevents = APR_POLLIN;
+    fd.client_data = this;
+    fd.desc.s = _socket;
+
+    out.flip();
+}
+
+LFSessionContext::~LFSessionContext(){
+
+}
+
+void LFSessionContext::read(){
+    socket.read(in);
+    in.flip();
+    if(initiated){
+        AMQFrame frame;
+        while(frame.decode(in)){
+            if(debug) log("RECV", &frame);
+            handler->received(&frame);
+        }
+    }else{
+        ProtocolInitiation protocolInit;
+        if(protocolInit.decode(in)){
+            handler->initiated(&protocolInit);
+            initiated = true;
+            if(debug) std::cout << "INIT [" << &socket << "]" << std::endl;
+        }
+    }
+    in.compact();
+}
+
+void LFSessionContext::write(){
+    bool done = isClosed();
+    while(!done){
+        if(out.available() > 0){
+            socket.write(out);
+            if(out.available() > 0){
+
+                //incomplete write, leave flags to receive notification of readiness to write
+                done = true;//finished processing for now, but write is still in progress
+            }
+        }else{
+            //do we have any frames to write?
+            Mutex::ScopedLock l(writeLock);
+            if(!framesToWrite.empty()){
+                out.clear();
+                bool encoded(false);
+                AMQFrame* frame = framesToWrite.front();
+                while(frame && out.available() >= frame->size()){
+                    encoded = true;
+                    frame->encode(out);
+                    if(debug) log("SENT", frame);
+                    delete frame;
+                    framesToWrite.pop();
+                    frame = framesToWrite.empty() ? 0 : framesToWrite.front();
+                }
+                if(!encoded) THROW_QPID_ERROR(FRAMING_ERROR, "Could not write frame, too large for buffer.");
+                out.flip();
+            }else{
+                //reset flags, don't care about writability anymore
+                fd.reqevents = APR_POLLIN;
+                done = true;
+
+                if(closing){
+                    socket.close();
+                }
+            }
+        }
+    }
+}
+
+void LFSessionContext::send(AMQFrame* frame){
+    Mutex::ScopedLock l(writeLock);
+    if(!closing){
+        framesToWrite.push(frame);
+        if(!(fd.reqevents & APR_POLLOUT)){
+            fd.reqevents |= APR_POLLOUT;
+            if(!processing){
+                processor->update(&fd);
+            }
+        }
+    }
+}
+
+void LFSessionContext::startProcessing(){
+    Mutex::ScopedLock l(writeLock);
+    processing = true;
+    processor->deactivate(&fd);
+}
+
+void LFSessionContext::stopProcessing(){
+    Mutex::ScopedLock l(writeLock);
+    processor->reactivate(&fd);
+    processing = false;
+}
+
+void LFSessionContext::close(){
+    closing = true;
+    Mutex::ScopedLock l(writeLock);
+    if(!processing){
+        //allow pending frames to be written to socket
+        fd.reqevents = APR_POLLOUT;
+        processor->update(&fd);
+    }
+}
+
+void LFSessionContext::handleClose(){
+    handler->closed();
+    std::cout << "Session closed [" << &socket << "]" << std::endl;
+    delete handler;
+    delete this;
+}
+
+void LFSessionContext::shutdown(){
+    socket.close();
+    handleClose();
+}
+
+void LFSessionContext::init(SessionHandler* _handler){
+    handler = _handler;
+    processor->add(&fd);
+}
+
+void LFSessionContext::log(const std::string& desc, AMQFrame* const frame){
+    Mutex::ScopedLock l(logLock);
+    std::cout << desc << " [" << &socket << "]: " << *frame << std::endl;
+}
+
+Mutex LFSessionContext::logLock;
diff --git a/cpp/src/qpid/apr/LFSessionContext.h b/cpp/src/qpid/apr/LFSessionContext.h
new file mode 100644
index 0000000000..9b3104b085
--- /dev/null
+++ b/cpp/src/qpid/apr/LFSessionContext.h
@@ -0,0 +1,87 @@
+/*
+ *
+ * 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 _LFSessionContext_
+#define _LFSessionContext_
+
+#include <queue>
+
+#include "apr-1/apr_network_io.h"
+#include "apr-1/apr_poll.h"
+#include "apr-1/apr_time.h"
+
+#include "qpid/framing/AMQFrame.h"
+#include "qpid/framing/Buffer.h"
+#include "qpid/sys/Monitor.h"
+#include "qpid/sys/SessionContext.h"
+#include "qpid/sys/SessionHandler.h"
+
+#include "APRSocket.h"
+#include "LFProcessor.h"
+
+namespace qpid {
+namespace sys {
+
+
+class LFSessionContext : public virtual qpid::sys::SessionContext
+{
+    const bool debug;
+    APRSocket socket;
+    bool initiated;
+        
+    qpid::framing::Buffer in;
+    qpid::framing::Buffer out;
+        
+    qpid::sys::SessionHandler* handler;
+    LFProcessor* const processor;
+
+    apr_pollfd_t fd;
+
+    std::queue<qpid::framing::AMQFrame*> framesToWrite;
+    qpid::sys::Mutex writeLock;
+        
+    bool processing;
+    bool closing;
+
+    static qpid::sys::Mutex logLock;
+    void log(const std::string& desc,
+             qpid::framing::AMQFrame* const frame);
+        
+
+  public:
+    LFSessionContext(apr_pool_t* pool, apr_socket_t* socket, 
+                     LFProcessor* const processor, 
+                     bool debug = false);
+    virtual ~LFSessionContext();
+    virtual void send(qpid::framing::AMQFrame* frame);
+    virtual void close();        
+    void read();
+    void write();
+    void init(qpid::sys::SessionHandler* handler);
+    void startProcessing();
+    void stopProcessing();
+    void handleClose();        
+    void shutdown();        
+    inline apr_pollfd_t* const getFd(){ return &fd; }
+    inline bool isClosed(){ return !socket.isOpen(); }
+};
+
+}
+}
+
+
+#endif
diff --git a/cpp/src/qpid/apr/Monitor.cpp b/cpp/src/qpid/apr/Monitor.cpp
new file mode 100644
index 0000000000..69fb2f6ffd
--- /dev/null
+++ b/cpp/src/qpid/apr/Monitor.cpp
@@ -0,0 +1,64 @@
+/*
+ *
+ * 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.
+ *
+ */
+#include "Monitor.h"
+#include "APRPool.h"
+
+using namespace qpid::sys;
+
+Mutex::Mutex()
+{
+    APRBase::increment();
+    // TODO aconway 2006-11-08: Switch to non-nested.
+    CHECK_APR_SUCCESS(apr_thread_mutex_create(&mutex, APR_THREAD_MUTEX_NESTED, APRPool::get()));
+}
+
+Mutex::~Mutex(){
+    CHECK_APR_SUCCESS(apr_thread_mutex_destroy(mutex));
+    APRBase::decrement();
+}
+
+Monitor::Monitor() 
+{
+    CHECK_APR_SUCCESS(apr_thread_cond_create(&condition, APRPool::get()));
+}
+
+Monitor::~Monitor(){
+    CHECK_APR_SUCCESS(apr_thread_cond_destroy(condition));
+}
+
+
+
+void Monitor::wait(){
+    CHECK_APR_SUCCESS(apr_thread_cond_wait(condition, mutex));
+}
+
+void Monitor::wait(int64_t nsecs){
+    // APR uses microseconds.
+    apr_status_t status = apr_thread_cond_timedwait(
+        condition, mutex, nsecs/1000);
+    if(!status == APR_TIMEUP) CHECK_APR_SUCCESS(status);
+}
+
+void Monitor::notify(){
+    CHECK_APR_SUCCESS(apr_thread_cond_signal(condition));
+}
+
+void Monitor::notifyAll(){
+    CHECK_APR_SUCCESS(apr_thread_cond_broadcast(condition));
+}
+
diff --git a/cpp/src/qpid/apr/Monitor.h b/cpp/src/qpid/apr/Monitor.h
new file mode 100644
index 0000000000..a51baf8d94
--- /dev/null
+++ b/cpp/src/qpid/apr/Monitor.h
@@ -0,0 +1,75 @@
+/*
+ *
+ * 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 <boost/noncopyable.hpp>
+#include <qpid/sys/Time.h>
+#include "apr-1/apr_thread_mutex.h"
+#include "apr-1/apr_thread_cond.h"
+#include "APRBase.h"
+
+namespace qpid {
+namespace sys {
+
+template <class L>
+class ScopedLock
+{
+  public:
+    ScopedLock(L& l) : mutex(l) { l.lock(); }
+    ~ScopedLock() { mutex.unlock(); }
+  private:
+    L& mutex;
+};
+
+
+class Mutex : private boost::noncopyable
+{
+  public:
+    typedef ScopedLock<Mutex> ScopedLock;
+    
+    Mutex();
+    ~Mutex();
+    void lock() { CHECK_APR_SUCCESS(apr_thread_mutex_lock(mutex)); }
+    void unlock() { CHECK_APR_SUCCESS(apr_thread_mutex_unlock(mutex)); }
+    void trylock() { CHECK_APR_SUCCESS(apr_thread_mutex_trylock(mutex)); }
+
+  protected:
+    apr_thread_mutex_t* mutex;
+};
+
+/** A condition variable and a mutex */
+class Monitor : public Mutex
+{
+  public:
+    Monitor();
+    ~Monitor();
+    void wait();
+    void wait(int64_t nsecs);
+    void notify();
+    void notifyAll();
+
+  private:
+    apr_thread_cond_t* condition;
+};
+
+
+}}
+
+
+#endif
diff --git a/cpp/src/qpid/apr/Thread.cpp b/cpp/src/qpid/apr/Thread.cpp
new file mode 100644
index 0000000000..6d5cadb009
--- /dev/null
+++ b/cpp/src/qpid/apr/Thread.cpp
@@ -0,0 +1,55 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#include "Thread.h"
+#include "APRPool.h"
+#include "APRBase.h"
+#include <apr-1/apr_portable.h>
+
+using namespace qpid::sys;
+using qpid::sys::Runnable;
+
+namespace {
+void* APR_THREAD_FUNC run(apr_thread_t* thread, void *data) {
+    reinterpret_cast<Runnable*>(data)->run();
+    CHECK_APR_SUCCESS(apr_thread_exit(thread, APR_SUCCESS));
+    return NULL;
+} 
+}
+
+Thread::Thread() : thread(0) {}
+
+Thread::Thread(Runnable* runnable) {
+    CHECK_APR_SUCCESS(
+        apr_thread_create(&thread, NULL, run, runnable, APRPool::get()));
+}
+
+void Thread::join(){
+    apr_status_t status;
+    if (thread != 0) 
+        CHECK_APR_SUCCESS(apr_thread_join(&status, thread));
+}
+
+Thread::Thread(apr_thread_t* t) : thread(t) {}
+
+Thread Thread::current(){
+    apr_thread_t* thr;
+    apr_os_thread_t osthr = apr_os_thread_current();
+    CHECK_APR_SUCCESS(apr_os_thread_put(&thr, &osthr, APRPool::get()));
+    return Thread(thr);
+}
diff --git a/cpp/src/qpid/apr/Thread.h b/cpp/src/qpid/apr/Thread.h
new file mode 100644
index 0000000000..0c717dea70
--- /dev/null
+++ b/cpp/src/qpid/apr/Thread.h
@@ -0,0 +1,45 @@
+#ifndef _apr_Thread_h
+#define _apr_Thread_h
+
+/*
+ *
+ * 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.
+ *
+ */
+
+#include <apr-1/apr_thread_proc.h>
+#include <qpid/sys/Runnable.h>
+
+namespace qpid {
+namespace sys {
+
+class Thread
+{
+
+  public:
+    Thread();
+    explicit Thread(qpid::sys::Runnable*);
+    void join();
+    static Thread current();
+
+  private:
+    Thread(apr_thread_t*);
+    
+    apr_thread_t* thread;
+};
+
+}}
+
+#endif  /*!_apr_Thread_h*/
diff --git a/cpp/src/qpid/apr/Time.cpp b/cpp/src/qpid/apr/Time.cpp
new file mode 100644
index 0000000000..8b5590481d
--- /dev/null
+++ b/cpp/src/qpid/apr/Time.cpp
@@ -0,0 +1,39 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#include <qpid/sys/Time.h>
+#include "apr-1/apr_time.h"
+
+namespace qpid {
+namespace sys {
+
+int64_t getTimeNsecs() 
+{
+    // APR returns microseconds.
+    return apr_time_now() * 1000;
+}
+
+int64_t getTimeMsecs() 
+{
+    // APR returns microseconds.
+    return apr_time_now() / 1000;
+}
+
+
+}}
+