summaryrefslogtreecommitdiff
path: root/cpp/src
diff options
context:
space:
mode:
Diffstat (limited to 'cpp/src')
-rw-r--r--cpp/src/Makefile.am5
-rw-r--r--cpp/src/qpid/sys/AsynchIOHandler.cpp (renamed from cpp/src/qpid/sys/AsynchIOAcceptor.cpp)156
-rw-r--r--cpp/src/qpid/sys/AsynchIOHandler.h70
-rw-r--r--cpp/src/qpid/sys/TCPIOPlugin.cpp109
4 files changed, 198 insertions, 142 deletions
diff --git a/cpp/src/Makefile.am b/cpp/src/Makefile.am
index 40e094de5d..09fd75d5c1 100644
--- a/cpp/src/Makefile.am
+++ b/cpp/src/Makefile.am
@@ -172,7 +172,7 @@ libqpidcommon_la_SOURCES = \
qpid/Plugin.cpp \
qpid/Url.cpp \
qpid/sys/AggregateOutput.cpp \
- qpid/sys/AsynchIOAcceptor.cpp \
+ qpid/sys/AsynchIOHandler.cpp \
qpid/sys/Dispatcher.cpp \
qpid/sys/Runnable.cpp \
qpid/sys/SystemInfo.cpp \
@@ -259,7 +259,8 @@ libqpidbroker_la_SOURCES = \
qpid/management/Manageable.cpp \
qpid/management/ManagementAgent.cpp \
qpid/management/ManagementExchange.cpp \
- qpid/management/ManagementObject.cpp
+ qpid/management/ManagementObject.cpp \
+ qpid/sys/TCPIOPlugin.cpp
libqpidclient_la_LIBADD = libqpidcommon.la
libqpidclient_la_SOURCES = \
diff --git a/cpp/src/qpid/sys/AsynchIOAcceptor.cpp b/cpp/src/qpid/sys/AsynchIOHandler.cpp
index 5133fde183..ca2bd7c93c 100644
--- a/cpp/src/qpid/sys/AsynchIOAcceptor.cpp
+++ b/cpp/src/qpid/sys/AsynchIOHandler.cpp
@@ -19,56 +19,14 @@
*
*/
-#include "Acceptor.h"
-
-#include "Socket.h"
-#include "AsynchIO.h"
-#include "Mutex.h"
-#include "Thread.h"
-
-#include "qpid/sys/ConnectionOutputHandler.h"
+#include "AsynchIOHandler.h"
#include "qpid/framing/AMQP_HighestVersion.h"
-#include "qpid/framing/reply_exceptions.h"
#include "qpid/framing/ProtocolInitiation.h"
#include "qpid/log/Statement.h"
-#include <boost/bind.hpp>
-#include <boost/assert.hpp>
-#include <queue>
-#include <vector>
-#include <memory>
-#include <ostream>
-
namespace qpid {
namespace sys {
-class AsynchIOAcceptor : public Acceptor {
- Socket listener;
- const uint16_t listeningPort;
- std::auto_ptr<AsynchAcceptor> acceptor;
-
- public:
- AsynchIOAcceptor(int16_t port, int backlog);
- void run(Poller::shared_ptr, ConnectionCodec::Factory*);
- void connect(Poller::shared_ptr, const std::string& host, int16_t port, ConnectionCodec::Factory*);
-
- uint16_t getPort() const;
- std::string getHost() const;
-
- private:
- void accepted(Poller::shared_ptr, const Socket&, ConnectionCodec::Factory*);
-};
-
-Acceptor::shared_ptr Acceptor::create(int16_t port, int backlog)
-{
- return Acceptor::shared_ptr(new AsynchIOAcceptor(port, backlog));
-}
-
-AsynchIOAcceptor::AsynchIOAcceptor(int16_t port, int backlog) :
- listeningPort(listener.listen(port, backlog)),
- acceptor(0)
-{}
-
// Buffer definition
struct Buff : public AsynchIO::BufferBase {
Buff() :
@@ -78,110 +36,28 @@ struct Buff : public AsynchIO::BufferBase {
{ delete [] bytes;}
};
-class AsynchIOHandler : public OutputControl {
- std::string identifier;
- AsynchIO* aio;
- ConnectionCodec::Factory* factory;
- ConnectionCodec* codec;
- bool readError;
- bool isClient;
-
- void write(const framing::ProtocolInitiation&);
-
- public:
- AsynchIOHandler(std::string id, ConnectionCodec::Factory* f) :
- identifier(id),
- aio(0),
- factory(f),
- codec(0),
- readError(false),
- isClient(false)
- {}
-
- ~AsynchIOHandler() {
- if (codec)
- codec->closed();
- delete codec;
- }
-
- void setClient() { isClient = true; }
-
- void init(AsynchIO* a) {
- aio = a;
- }
-
- // Output side
- void close();
- void activateOutput();
-
- // Input side
- void readbuff(AsynchIO& aio, AsynchIO::BufferBase* buff);
- void eof(AsynchIO& aio);
- void disconnect(AsynchIO& aio);
-
- // Notifications
- void nobuffs(AsynchIO& aio);
- void idle(AsynchIO& aio);
- void closedSocket(AsynchIO& aio, const Socket& s);
-};
-
-void AsynchIOAcceptor::accepted(Poller::shared_ptr poller, const Socket& s, ConnectionCodec::Factory* f) {
- AsynchIOHandler* async = new AsynchIOHandler(s.getPeerAddress(), f);
- AsynchIO* aio = new AsynchIO(s,
- boost::bind(&AsynchIOHandler::readbuff, async, _1, _2),
- boost::bind(&AsynchIOHandler::eof, async, _1),
- boost::bind(&AsynchIOHandler::disconnect, async, _1),
- boost::bind(&AsynchIOHandler::closedSocket, async, _1, _2),
- boost::bind(&AsynchIOHandler::nobuffs, async, _1),
- boost::bind(&AsynchIOHandler::idle, async, _1));
- async->init(aio);
-
- // Give connection some buffers to use
- for (int i = 0; i < 4; i++) {
- aio->queueReadBuffer(new Buff);
- }
- aio->start(poller);
-}
-
-
-uint16_t AsynchIOAcceptor::getPort() const {
- return listeningPort; // Immutable no need for lock.
-}
+AsynchIOHandler::AsynchIOHandler(std::string id, ConnectionCodec::Factory* f) :
+ identifier(id),
+ aio(0),
+ factory(f),
+ codec(0),
+ readError(false),
+ isClient(false)
+{}
-std::string AsynchIOAcceptor::getHost() const {
- return listener.getSockname();
+AsynchIOHandler::~AsynchIOHandler() {
+ if (codec)
+ codec->closed();
+ delete codec;
}
-void AsynchIOAcceptor::run(Poller::shared_ptr poller, ConnectionCodec::Factory* fact) {
- acceptor.reset(
- new AsynchAcceptor(listener,
- boost::bind(&AsynchIOAcceptor::accepted, this, poller, _1, fact)));
- acceptor->start(poller);
-}
-
-void AsynchIOAcceptor::connect(
- Poller::shared_ptr poller,
- const std::string& host, int16_t port,
- ConnectionCodec::Factory* f)
-{
- Socket* socket = new Socket();//Should be deleted by handle when socket closes
- socket->connect(host, port);
- AsynchIOHandler* async = new AsynchIOHandler(socket->getPeerAddress(), f);
- async->setClient();
- AsynchIO* aio = new AsynchIO(*socket,
- boost::bind(&AsynchIOHandler::readbuff, async, _1, _2),
- boost::bind(&AsynchIOHandler::eof, async, _1),
- boost::bind(&AsynchIOHandler::disconnect, async, _1),
- boost::bind(&AsynchIOHandler::closedSocket, async, _1, _2),
- boost::bind(&AsynchIOHandler::nobuffs, async, _1),
- boost::bind(&AsynchIOHandler::idle, async, _1));
- async->init(aio);
+void AsynchIOHandler::init(AsynchIO* a, int numBuffs) {
+ aio = a;
// Give connection some buffers to use
- for (int i = 0; i < 4; i++) {
+ for (int i = 0; i < numBuffs; i++) {
aio->queueReadBuffer(new Buff);
}
- aio->start(poller);
}
void AsynchIOHandler::write(const framing::ProtocolInitiation& data)
diff --git a/cpp/src/qpid/sys/AsynchIOHandler.h b/cpp/src/qpid/sys/AsynchIOHandler.h
new file mode 100644
index 0000000000..530613367a
--- /dev/null
+++ b/cpp/src/qpid/sys/AsynchIOHandler.h
@@ -0,0 +1,70 @@
+#ifndef _sys_AsynchIOHandler_h
+#define _sys_AsynchIOHandler_h
+/*
+ *
+ * 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 "OutputControl.h"
+#include "ConnectionCodec.h"
+#include "AsynchIO.h"
+
+namespace qpid {
+
+namespace framing {
+ class ProtocolInitiation;
+}
+
+namespace sys {
+
+class AsynchIOHandler : public OutputControl {
+ std::string identifier;
+ AsynchIO* aio;
+ ConnectionCodec::Factory* factory;
+ ConnectionCodec* codec;
+ bool readError;
+ bool isClient;
+
+ void write(const framing::ProtocolInitiation&);
+
+ public:
+ AsynchIOHandler(std::string id, ConnectionCodec::Factory* f);
+ ~AsynchIOHandler();
+ void init(AsynchIO* a, int numBuffs);
+
+ void setClient() { isClient = true; }
+
+ // Output side
+ void close();
+ void activateOutput();
+
+ // Input side
+ void readbuff(AsynchIO& aio, AsynchIO::BufferBase* buff);
+ void eof(AsynchIO& aio);
+ void disconnect(AsynchIO& aio);
+
+ // Notifications
+ void nobuffs(AsynchIO& aio);
+ void idle(AsynchIO& aio);
+ void closedSocket(AsynchIO& aio, const Socket& s);
+};
+
+}} // namespace qpid::sys
+
+#endif // _sys_AsynchIOHandler_h
diff --git a/cpp/src/qpid/sys/TCPIOPlugin.cpp b/cpp/src/qpid/sys/TCPIOPlugin.cpp
new file mode 100644
index 0000000000..779319b7e4
--- /dev/null
+++ b/cpp/src/qpid/sys/TCPIOPlugin.cpp
@@ -0,0 +1,109 @@
+/*
+ *
+ * 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 "Acceptor.h"
+
+#include "AsynchIOHandler.h"
+#include "AsynchIO.h"
+
+#include <boost/bind.hpp>
+#include <memory>
+
+namespace qpid {
+namespace sys {
+
+class AsynchIOAcceptor : public Acceptor {
+ Socket listener;
+ const uint16_t listeningPort;
+ std::auto_ptr<AsynchAcceptor> acceptor;
+
+ public:
+ AsynchIOAcceptor(int16_t port, int backlog);
+ void run(Poller::shared_ptr, ConnectionCodec::Factory*);
+ void connect(Poller::shared_ptr, const std::string& host, int16_t port, ConnectionCodec::Factory*);
+
+ uint16_t getPort() const;
+ std::string getHost() const;
+
+ private:
+ void accepted(Poller::shared_ptr, const Socket&, ConnectionCodec::Factory*);
+};
+
+Acceptor::shared_ptr Acceptor::create(int16_t port, int backlog)
+{
+ return Acceptor::shared_ptr(new AsynchIOAcceptor(port, backlog));
+}
+
+AsynchIOAcceptor::AsynchIOAcceptor(int16_t port, int backlog) :
+ listeningPort(listener.listen(port, backlog)),
+ acceptor(0)
+{}
+
+void AsynchIOAcceptor::accepted(Poller::shared_ptr poller, const Socket& s, ConnectionCodec::Factory* f) {
+ AsynchIOHandler* async = new AsynchIOHandler(s.getPeerAddress(), f);
+ AsynchIO* aio = new AsynchIO(s,
+ boost::bind(&AsynchIOHandler::readbuff, async, _1, _2),
+ boost::bind(&AsynchIOHandler::eof, async, _1),
+ boost::bind(&AsynchIOHandler::disconnect, async, _1),
+ boost::bind(&AsynchIOHandler::closedSocket, async, _1, _2),
+ boost::bind(&AsynchIOHandler::nobuffs, async, _1),
+ boost::bind(&AsynchIOHandler::idle, async, _1));
+ async->init(aio, 4);
+ aio->start(poller);
+}
+
+
+uint16_t AsynchIOAcceptor::getPort() const {
+ return listeningPort; // Immutable no need for lock.
+}
+
+std::string AsynchIOAcceptor::getHost() const {
+ return listener.getSockname();
+}
+
+void AsynchIOAcceptor::run(Poller::shared_ptr poller, ConnectionCodec::Factory* fact) {
+ acceptor.reset(
+ new AsynchAcceptor(listener,
+ boost::bind(&AsynchIOAcceptor::accepted, this, poller, _1, fact)));
+ acceptor->start(poller);
+}
+
+void AsynchIOAcceptor::connect(
+ Poller::shared_ptr poller,
+ const std::string& host, int16_t port,
+ ConnectionCodec::Factory* f)
+{
+ Socket* socket = new Socket();//Should be deleted by handle when socket closes
+ socket->connect(host, port);
+ AsynchIOHandler* async = new AsynchIOHandler(socket->getPeerAddress(), f);
+ async->setClient();
+ AsynchIO* aio = new AsynchIO(*socket,
+ boost::bind(&AsynchIOHandler::readbuff, async, _1, _2),
+ boost::bind(&AsynchIOHandler::eof, async, _1),
+ boost::bind(&AsynchIOHandler::disconnect, async, _1),
+ boost::bind(&AsynchIOHandler::closedSocket, async, _1, _2),
+ boost::bind(&AsynchIOHandler::nobuffs, async, _1),
+ boost::bind(&AsynchIOHandler::idle, async, _1));
+ async->init(aio, 4);
+ aio->start(poller);
+}
+
+}} // namespace qpid::sys
View Lorry Controller Status