summaryrefslogtreecommitdiff
path: root/qpid/cpp-0-9/lib/common/sys/apr
diff options
context:
space:
mode:
Diffstat (limited to 'qpid/cpp-0-9/lib/common/sys/apr')
-rw-r--r--qpid/cpp-0-9/lib/common/sys/apr/APRAcceptor.cpp122
-rw-r--r--qpid/cpp-0-9/lib/common/sys/apr/APRBase.cpp90
-rw-r--r--qpid/cpp-0-9/lib/common/sys/apr/APRBase.h78
-rw-r--r--qpid/cpp-0-9/lib/common/sys/apr/APRPool.cpp41
-rw-r--r--qpid/cpp-0-9/lib/common/sys/apr/APRPool.h50
-rw-r--r--qpid/cpp-0-9/lib/common/sys/apr/APRSocket.cpp78
-rw-r--r--qpid/cpp-0-9/lib/common/sys/apr/APRSocket.h48
-rw-r--r--qpid/cpp-0-9/lib/common/sys/apr/LFProcessor.cpp179
-rw-r--r--qpid/cpp-0-9/lib/common/sys/apr/LFProcessor.h121
-rw-r--r--qpid/cpp-0-9/lib/common/sys/apr/LFSessionContext.cpp179
-rw-r--r--qpid/cpp-0-9/lib/common/sys/apr/LFSessionContext.h90
-rw-r--r--qpid/cpp-0-9/lib/common/sys/apr/Socket.cpp86
-rw-r--r--qpid/cpp-0-9/lib/common/sys/apr/Thread.cpp33
13 files changed, 0 insertions, 1195 deletions
diff --git a/qpid/cpp-0-9/lib/common/sys/apr/APRAcceptor.cpp b/qpid/cpp-0-9/lib/common/sys/apr/APRAcceptor.cpp
deleted file mode 100644
index 52384857ed..0000000000
--- a/qpid/cpp-0-9/lib/common/sys/apr/APRAcceptor.cpp
+++ /dev/null
@@ -1,122 +0,0 @@
-/*
- *
- * 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 <sys/Acceptor.h>
-#include <sys/ConnectionInputHandlerFactory.h>
-#include "LFProcessor.h"
-#include "LFSessionContext.h"
-#include "APRBase.h"
-#include "APRPool.h"
-
-namespace qpid {
-namespace sys {
-
-class APRAcceptor : public Acceptor
-{
- public:
- APRAcceptor(int16_t port, int backlog, int threads, bool trace);
- virtual int16_t getPort() const;
- virtual void run(qpid::sys::ConnectionInputHandlerFactory* factory);
- virtual void shutdown();
-
- private:
- void shutdownImpl();
-
- private:
- int16_t port;
- bool trace;
- LFProcessor processor;
- apr_socket_t* socket;
- volatile bool running;
- Mutex shutdownLock;
-};
-
-// Define generic Acceptor::create() to return APRAcceptor.
-Acceptor::shared_ptr Acceptor::create(int16_t port, int backlog, int threads, bool trace)
-{
- return Acceptor::shared_ptr(new APRAcceptor(port, backlog, threads, trace));
-}
-// Must define Acceptor virtual dtor.
-Acceptor::~Acceptor() {}
-
-APRAcceptor::APRAcceptor(int16_t port_, int backlog, int threads, bool trace_) :
- port(port_),
- trace(trace_),
- processor(APRPool::get(), threads, 1000, 5000000),
- running(false)
-{
- 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 APRAcceptor::getPort() const {
- apr_sockaddr_t* address;
- CHECK_APR_SUCCESS(apr_socket_addr_get(&address, APR_LOCAL, socket));
- return address->port;
-}
-
-void APRAcceptor::run(ConnectionInputHandlerFactory* 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, trace);
- session->init(factory->create(session));
- }else{
- Mutex::ScopedLock locker(shutdownLock);
- if(running) {
- if(status != APR_EINTR){
- std::cout << "ERROR: " << get_desc(status) << std::endl;
- }
- shutdownImpl();
- }
- }
- }
-}
-
-void APRAcceptor::shutdown() {
- Mutex::ScopedLock locker(shutdownLock);
- if (running) {
- shutdownImpl();
- }
-}
-
-void APRAcceptor::shutdownImpl() {
- Mutex::ScopedLock locker(shutdownLock);
- running = false;
- processor.stop();
- CHECK_APR_SUCCESS(apr_socket_close(socket));
-}
-
-
-}}
diff --git a/qpid/cpp-0-9/lib/common/sys/apr/APRBase.cpp b/qpid/cpp-0-9/lib/common/sys/apr/APRBase.cpp
deleted file mode 100644
index 861071499f..0000000000
--- a/qpid/cpp-0-9/lib/common/sys/apr/APRBase.cpp
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- *
- * 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 <iostream>
-#include <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();
-}
-
-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/qpid/cpp-0-9/lib/common/sys/apr/APRBase.h b/qpid/cpp-0-9/lib/common/sys/apr/APRBase.h
deleted file mode 100644
index 6a866a554a..0000000000
--- a/qpid/cpp-0-9/lib/common/sys/apr/APRBase.h
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- *
- * 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 _APRBase_
-#define _APRBase_
-
-#include <string>
-#include <apr_thread_mutex.h>
-#include <apr_errno.h>
-#include <QpidError.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 char* file, const int line);
- std::string get_desc(apr_status_t status);
-
-#define CHECK_APR_SUCCESS(A) qpid::sys::check(A, __FILE__, __LINE__);
-
-}
-}
-
-// Inlined as it is called *a lot*
-void inline qpid::sys::check(apr_status_t status, const char* file, const int line){
- if (status != APR_SUCCESS){
- const int size = 50;
- char tmp[size];
- std::string msg(apr_strerror(status, tmp, size));
- throw qpid::QpidError(APR_ERROR + ((int) status), msg,
- qpid::SrcLine(file, line));
- }
-}
-
-
-
-
-#endif
diff --git a/qpid/cpp-0-9/lib/common/sys/apr/APRPool.cpp b/qpid/cpp-0-9/lib/common/sys/apr/APRPool.cpp
deleted file mode 100644
index e8b71f6e8a..0000000000
--- a/qpid/cpp-0-9/lib/common/sys/apr/APRPool.cpp
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- *
- * 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 "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/qpid/cpp-0-9/lib/common/sys/apr/APRPool.h b/qpid/cpp-0-9/lib/common/sys/apr/APRPool.h
deleted file mode 100644
index da7661fcfa..0000000000
--- a/qpid/cpp-0-9/lib/common/sys/apr/APRPool.h
+++ /dev/null
@@ -1,50 +0,0 @@
-#ifndef _APRPool_
-#define _APRPool_
-
-/*
- *
- * 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 <boost/noncopyable.hpp>
-#include <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/qpid/cpp-0-9/lib/common/sys/apr/APRSocket.cpp b/qpid/cpp-0-9/lib/common/sys/apr/APRSocket.cpp
deleted file mode 100644
index 96dbd132a1..0000000000
--- a/qpid/cpp-0-9/lib/common/sys/apr/APRSocket.cpp
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- *
- * 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 "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() const {
- return !closed;
-}
-
-uint8_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/qpid/cpp-0-9/lib/common/sys/apr/APRSocket.h b/qpid/cpp-0-9/lib/common/sys/apr/APRSocket.h
deleted file mode 100644
index a55dfc06b0..0000000000
--- a/qpid/cpp-0-9/lib/common/sys/apr/APRSocket.h
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- *
- * 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 _APRSocket_
-#define _APRSocket_
-
-#include <apr_network_io.h>
-#include <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() const;
- uint8_t read();
- ~APRSocket();
- };
-
-}
-}
-
-
-#endif
diff --git a/qpid/cpp-0-9/lib/common/sys/apr/LFProcessor.cpp b/qpid/cpp-0-9/lib/common/sys/apr/LFProcessor.cpp
deleted file mode 100644
index 2b6fc92623..0000000000
--- a/qpid/cpp-0-9/lib/common/sys/apr/LFProcessor.cpp
+++ /dev/null
@@ -1,179 +0,0 @@
-/*
- *
- * 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 <sstream>
-#include <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(std::exception e){
- std::cout << e.what() << 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/qpid/cpp-0-9/lib/common/sys/apr/LFProcessor.h b/qpid/cpp-0-9/lib/common/sys/apr/LFProcessor.h
deleted file mode 100644
index de90199472..0000000000
--- a/qpid/cpp-0-9/lib/common/sys/apr/LFProcessor.h
+++ /dev/null
@@ -1,121 +0,0 @@
-/*
- *
- * 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 _LFProcessor_
-#define _LFProcessor_
-
-#include <apr_poll.h>
-#include <iostream>
-#include <vector>
-#include <sys/Monitor.h>
-#include <sys/Runnable.h>
-#include <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/qpid/cpp-0-9/lib/common/sys/apr/LFSessionContext.cpp b/qpid/cpp-0-9/lib/common/sys/apr/LFSessionContext.cpp
deleted file mode 100644
index 503dfddbb7..0000000000
--- a/qpid/cpp-0-9/lib/common/sys/apr/LFSessionContext.cpp
+++ /dev/null
@@ -1,179 +0,0 @@
-/*
- *
- * 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 "LFSessionContext.h"
-#include "APRBase.h"
-#include <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(65536),
- out(65536),
- 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;
- try{
- while(frame.decode(in)){
- if(debug) log("RECV", &frame);
- handler->received(&frame);
- }
- }catch(QpidError error){
- std::cout << "Error [" << error.code << "] " << error.msg
- << " (" << error.loc.file << ":" << error.loc.line
- << ")" << std::endl;
- }
- }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(){
- Mutex::ScopedLock l(writeLock);
- closing = true;
- 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(ConnectionInputHandler* _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/qpid/cpp-0-9/lib/common/sys/apr/LFSessionContext.h b/qpid/cpp-0-9/lib/common/sys/apr/LFSessionContext.h
deleted file mode 100644
index 81cfc0efda..0000000000
--- a/qpid/cpp-0-9/lib/common/sys/apr/LFSessionContext.h
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- *
- * 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 _LFSessionContext_
-#define _LFSessionContext_
-
-#include <queue>
-
-#include <apr_network_io.h>
-#include <apr_poll.h>
-#include <apr_time.h>
-
-#include <AMQFrame.h>
-#include <Buffer.h>
-#include <sys/Monitor.h>
-#include <sys/ConnectionOutputHandler.h>
-#include <sys/ConnectionInputHandler.h>
-
-#include "APRSocket.h"
-#include "LFProcessor.h"
-
-namespace qpid {
-namespace sys {
-
-
-class LFSessionContext : public virtual qpid::sys::ConnectionOutputHandler
-{
- const bool debug;
- APRSocket socket;
- bool initiated;
-
- qpid::framing::Buffer in;
- qpid::framing::Buffer out;
-
- qpid::sys::ConnectionInputHandler* 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::ConnectionInputHandler* 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/qpid/cpp-0-9/lib/common/sys/apr/Socket.cpp b/qpid/cpp-0-9/lib/common/sys/apr/Socket.cpp
deleted file mode 100644
index bca4da6c96..0000000000
--- a/qpid/cpp-0-9/lib/common/sys/apr/Socket.cpp
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- *
- * 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 <sys/Socket.h>
-#include <apr/APRBase.h>
-#include <apr/APRPool.h>
-
-
-using namespace qpid::sys;
-
-Socket Socket::createTcp() {
- Socket s;
- CHECK_APR_SUCCESS(
- apr_socket_create(
- &s.socket, APR_INET, SOCK_STREAM, APR_PROTO_TCP,
- APRPool::get()));
- return s;
-}
-
-Socket::Socket(apr_socket_t* s) {
- socket = s;
-}
-
-void Socket::setTimeout(Time interval) {
- apr_socket_timeout_set(socket, interval/TIME_USEC);
-}
-
-void Socket::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,
- APRPool::get()));
- CHECK_APR_SUCCESS(apr_socket_connect(socket, address));
-}
-
-void Socket::close() {
- if (socket == 0) return;
- CHECK_APR_SUCCESS(apr_socket_close(socket));
- socket = 0;
-}
-
-ssize_t Socket::send(const void* data, size_t size)
-{
- apr_size_t sent = size;
- apr_status_t status =
- apr_socket_send(socket, reinterpret_cast<const char*>(data), &sent);
- if (APR_STATUS_IS_TIMEUP(status)) return SOCKET_TIMEOUT;
- if (APR_STATUS_IS_EOF(status)) return SOCKET_EOF;
- CHECK_APR_SUCCESS(status);
- return sent;
-}
-
-ssize_t Socket::recv(void* data, size_t size)
-{
- apr_size_t received = size;
- apr_status_t status =
- apr_socket_recv(socket, reinterpret_cast<char*>(data), &received);
- if (APR_STATUS_IS_TIMEUP(status))
- return SOCKET_TIMEOUT;
- if (APR_STATUS_IS_EOF(status))
- return SOCKET_EOF;
- CHECK_APR_SUCCESS(status);
- return received;
-}
-
-
diff --git a/qpid/cpp-0-9/lib/common/sys/apr/Thread.cpp b/qpid/cpp-0-9/lib/common/sys/apr/Thread.cpp
deleted file mode 100644
index 5c4799aa96..0000000000
--- a/qpid/cpp-0-9/lib/common/sys/apr/Thread.cpp
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- *
- * 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 <sys/Thread.h>
-
-using namespace qpid::sys;
-using qpid::sys::Runnable;
-
-void* APR_THREAD_FUNC Thread::runRunnable(apr_thread_t* thread, void *data) {
- reinterpret_cast<Runnable*>(data)->run();
- CHECK_APR_SUCCESS(apr_thread_exit(thread, APR_SUCCESS));
- return NULL;
-}
-
-
View Lorry Controller Status