diff options
Diffstat (limited to 'cpp/src/qpid/sys')
58 files changed, 1189 insertions, 2026 deletions
diff --git a/cpp/src/qpid/sys/AggregateOutput.cpp b/cpp/src/qpid/sys/AggregateOutput.cpp index fc95f46fb9..ebc5689ce5 100644 --- a/cpp/src/qpid/sys/AggregateOutput.cpp +++ b/cpp/src/qpid/sys/AggregateOutput.cpp @@ -32,8 +32,6 @@ void AggregateOutput::abort() { control.abort(); } void AggregateOutput::activateOutput() { control.activateOutput(); } -void AggregateOutput::giveReadCredit(int32_t credit) { control.giveReadCredit(credit); } - namespace { // Clear the busy flag and notify waiting threads in destructor. struct ScopedBusy { @@ -51,6 +49,7 @@ bool AggregateOutput::doOutput() { while (!tasks.empty()) { OutputTask* t=tasks.front(); tasks.pop_front(); + taskSet.erase(t); bool didOutput; { // Allow concurrent call to addOutputTask. @@ -59,7 +58,9 @@ bool AggregateOutput::doOutput() { didOutput = t->doOutput(); } if (didOutput) { - tasks.push_back(t); + if (taskSet.insert(t).second) { + tasks.push_back(t); + } return true; } } @@ -68,12 +69,15 @@ bool AggregateOutput::doOutput() { void AggregateOutput::addOutputTask(OutputTask* task) { Mutex::ScopedLock l(lock); - tasks.push_back(task); + if (taskSet.insert(task).second) { + tasks.push_back(task); + } } void AggregateOutput::removeOutputTask(OutputTask* task) { Mutex::ScopedLock l(lock); while (busy) lock.wait(); + taskSet.erase(task); tasks.erase(std::remove(tasks.begin(), tasks.end(), task), tasks.end()); } @@ -81,6 +85,7 @@ void AggregateOutput::removeAll() { Mutex::ScopedLock l(lock); while (busy) lock.wait(); + taskSet.clear(); tasks.clear(); } diff --git a/cpp/src/qpid/sys/AggregateOutput.h b/cpp/src/qpid/sys/AggregateOutput.h index d7c0ff29e3..e9dbd5a4cc 100644 --- a/cpp/src/qpid/sys/AggregateOutput.h +++ b/cpp/src/qpid/sys/AggregateOutput.h @@ -28,6 +28,7 @@ #include <algorithm> #include <deque> +#include <set> namespace qpid { namespace sys { @@ -44,9 +45,11 @@ namespace sys { class QPID_COMMON_CLASS_EXTERN AggregateOutput : public OutputTask, public OutputControl { typedef std::deque<OutputTask*> TaskList; + typedef std::set<OutputTask*> TaskSet; Monitor lock; TaskList tasks; + TaskSet taskSet; bool busy; OutputControl& control; @@ -56,7 +59,6 @@ class QPID_COMMON_CLASS_EXTERN AggregateOutput : public OutputTask, public Outpu // These may be called concurrently with any function. QPID_COMMON_EXTERN void abort(); QPID_COMMON_EXTERN void activateOutput(); - QPID_COMMON_EXTERN void giveReadCredit(int32_t); QPID_COMMON_EXTERN void addOutputTask(OutputTask* t); // These functions must not be called concurrently with each other. diff --git a/cpp/src/qpid/sys/AsynchIO.h b/cpp/src/qpid/sys/AsynchIO.h index b2eaaac9de..679665f8ad 100644 --- a/cpp/src/qpid/sys/AsynchIO.h +++ b/cpp/src/qpid/sys/AsynchIO.h @@ -21,9 +21,11 @@ * */ -#include "qpid/sys/IntegerTypes.h" #include "qpid/CommonImportExport.h" +#include "qpid/sys/IntegerTypes.h" +#include "qpid/sys/SecuritySettings.h" + #include <string.h> #include <boost/function.hpp> @@ -56,6 +58,7 @@ class AsynchConnector { public: typedef boost::function1<void, const Socket&> ConnectedCallback; typedef boost::function3<void, const Socket&, int, const std::string&> FailedCallback; + typedef boost::function1<void, AsynchConnector&> RequestCallback; // Call create() to allocate a new AsynchConnector object with the // specified poller, addressing, and callbacks. @@ -70,6 +73,7 @@ public: FailedCallback failCb); virtual void start(boost::shared_ptr<Poller> poller) = 0; virtual void stop() {}; + virtual void requestCallback(RequestCallback) = 0; protected: AsynchConnector() {} virtual ~AsynchConnector() {} @@ -155,11 +159,11 @@ public: virtual void notifyPendingWrite() = 0; virtual void queueWriteClose() = 0; virtual bool writeQueueEmpty() = 0; - virtual void startReading() = 0; - virtual void stopReading() = 0; virtual void requestCallback(RequestCallback) = 0; virtual BufferBase* getQueuedBuffer() = 0; + virtual SecuritySettings getSecuritySettings() = 0; + protected: // Derived class manages lifetime; must be constructed using the // static create() method. Deletes not allowed from outside. diff --git a/cpp/src/qpid/sys/AsynchIOHandler.cpp b/cpp/src/qpid/sys/AsynchIOHandler.cpp index 2e117a3fb7..cf08b482e6 100644 --- a/cpp/src/qpid/sys/AsynchIOHandler.cpp +++ b/cpp/src/qpid/sys/AsynchIOHandler.cpp @@ -51,15 +51,15 @@ struct ProtocolTimeoutTask : public sys::TimerTask { } }; -AsynchIOHandler::AsynchIOHandler(const std::string& id, ConnectionCodec::Factory* f) : +AsynchIOHandler::AsynchIOHandler(const std::string& id, ConnectionCodec::Factory* f, bool isClient0, bool nodict0) : identifier(id), aio(0), factory(f), codec(0), reads(0), readError(false), - isClient(false), - readCredit(InfiniteCredit) + isClient(isClient0), + nodict(nodict0) {} AsynchIOHandler::~AsynchIOHandler() { @@ -97,25 +97,20 @@ void AsynchIOHandler::abort() { if (!readError) { aio->requestCallback(boost::bind(&AsynchIOHandler::eof, this, _1)); } + aio->queueWriteClose(); } void AsynchIOHandler::activateOutput() { aio->notifyPendingWrite(); } -// Input side -void AsynchIOHandler::giveReadCredit(int32_t credit) { - // Check whether we started in the don't about credit state - if (readCredit.boolCompareAndSwap(InfiniteCredit, credit)) - return; - // TODO In theory should be able to use an atomic operation before taking the lock - // but in practice there seems to be an unexplained race in that case - ScopedLock<Mutex> l(creditLock); - if (readCredit.fetchAndAdd(credit) != 0) - return; - assert(readCredit.get() >= 0); - if (readCredit.get() != 0) - aio->startReading(); +namespace { + SecuritySettings getSecuritySettings(AsynchIO* aio, bool nodict) + { + SecuritySettings settings = aio->getSecuritySettings(); + settings.nodict = nodict; + return settings; + } } void AsynchIOHandler::readbuff(AsynchIO& , AsynchIO::BufferBase* buff) { @@ -123,26 +118,6 @@ void AsynchIOHandler::readbuff(AsynchIO& , AsynchIO::BufferBase* buff) { return; } - // Check here for read credit - if (readCredit.get() != InfiniteCredit) { - if (readCredit.get() == 0) { - // FIXME aconway 2009-10-01: Workaround to avoid "false wakeups". - // readbuff is sometimes called with no credit. - // This should be fixed somewhere else to avoid such calls. - aio->unread(buff); - return; - } - // TODO In theory should be able to use an atomic operation before taking the lock - // but in practice there seems to be an unexplained race in that case - ScopedLock<Mutex> l(creditLock); - if (--readCredit == 0) { - assert(readCredit.get() >= 0); - if (readCredit.get() == 0) { - aio->stopReading(); - } - } - } - ++reads; size_t decoded = 0; if (codec) { // Already initiated @@ -168,13 +143,16 @@ void AsynchIOHandler::readbuff(AsynchIO& , AsynchIO::BufferBase* buff) { QPID_LOG(debug, "RECV [" << identifier << "]: INIT(" << protocolInit << ")"); try { - codec = factory->create(protocolInit.getVersion(), *this, identifier, SecuritySettings()); + codec = factory->create(protocolInit.getVersion(), *this, identifier, getSecuritySettings(aio, nodict)); if (!codec) { //TODO: may still want to revise this... //send valid version header & close connection. write(framing::ProtocolInitiation(framing::highestProtocolVersion)); readError = true; aio->queueWriteClose(); + } else { + //read any further data that may already have been sent + decoded += codec->decode(buff->bytes+buff->dataStart+in.getPosition(), buff->dataCount-in.getPosition()); } } catch (const std::exception& e) { QPID_LOG(error, e.what()); @@ -223,7 +201,7 @@ void AsynchIOHandler::nobuffs(AsynchIO&) { void AsynchIOHandler::idle(AsynchIO&){ if (isClient && codec == 0) { - codec = factory->create(*this, identifier, SecuritySettings()); + codec = factory->create(*this, identifier, getSecuritySettings(aio, nodict)); write(framing::ProtocolInitiation(codec->getVersion())); // We've just sent the protocol negotiation so we can cancel the timeout for that // This is not ideal, because we've not received anything yet, but heartbeats will diff --git a/cpp/src/qpid/sys/AsynchIOHandler.h b/cpp/src/qpid/sys/AsynchIOHandler.h index fd0bc140e5..d93e24fd4c 100644 --- a/cpp/src/qpid/sys/AsynchIOHandler.h +++ b/cpp/src/qpid/sys/AsynchIOHandler.h @@ -51,24 +51,19 @@ class AsynchIOHandler : public OutputControl { uint32_t reads; bool readError; bool isClient; - AtomicValue<int32_t> readCredit; - static const int32_t InfiniteCredit = -1; - Mutex creditLock; + bool nodict; boost::intrusive_ptr<sys::TimerTask> timeoutTimerTask; void write(const framing::ProtocolInitiation&); public: - QPID_COMMON_EXTERN AsynchIOHandler(const std::string& id, qpid::sys::ConnectionCodec::Factory* f ); + QPID_COMMON_EXTERN AsynchIOHandler(const std::string& id, qpid::sys::ConnectionCodec::Factory* f, bool isClient, bool nodict); QPID_COMMON_EXTERN ~AsynchIOHandler(); QPID_COMMON_EXTERN void init(AsynchIO* a, Timer& timer, uint32_t maxTime); - QPID_COMMON_INLINE_EXTERN void setClient() { isClient = true; } - // Output side QPID_COMMON_EXTERN void abort(); QPID_COMMON_EXTERN void activateOutput(); - QPID_COMMON_EXTERN void giveReadCredit(int32_t credit); // Input side QPID_COMMON_EXTERN void readbuff(AsynchIO& aio, AsynchIOBufferBase* buff); diff --git a/cpp/src/qpid/sys/ClusterSafe.cpp b/cpp/src/qpid/sys/ClusterSafe.cpp deleted file mode 100644 index dd37615145..0000000000 --- a/cpp/src/qpid/sys/ClusterSafe.cpp +++ /dev/null @@ -1,66 +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 "ClusterSafe.h" -#include "qpid/log/Statement.h" -#include "qpid/sys/Thread.h" -#include <stdlib.h> - -namespace qpid { -namespace sys { - -namespace { -bool inCluster = false; -QPID_TSS bool inContext = false; -} - -bool isClusterSafe() { return !inCluster || inContext; } - -void assertClusterSafe() { - if (!isClusterSafe()) { - QPID_LOG(critical, "Modified cluster state outside of cluster context"); - ::abort(); - } -} - -ClusterSafeScope::ClusterSafeScope() { - save = inContext; - inContext = true; -} - -ClusterSafeScope::~ClusterSafeScope() { - assert(inContext); - inContext = save; -} - -ClusterUnsafeScope::ClusterUnsafeScope() { - save = inContext; - inContext = false; -} - -ClusterUnsafeScope::~ClusterUnsafeScope() { - assert(!inContext); - inContext = save; -} - -void enableClusterSafe() { inCluster = true; } - -}} // namespace qpid::sys diff --git a/cpp/src/qpid/sys/ClusterSafe.h b/cpp/src/qpid/sys/ClusterSafe.h deleted file mode 100644 index 27e4eb46a5..0000000000 --- a/cpp/src/qpid/sys/ClusterSafe.h +++ /dev/null @@ -1,87 +0,0 @@ -#ifndef QPID_SYS_CLUSTERSAFE_H -#define QPID_SYS_CLUSTERSAFE_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 "qpid/CommonImportExport.h" - -namespace qpid { -namespace sys { - -/** - * Assertion to add to code that modifies clustered state. - * - * In a non-clustered broker this is a no-op. - * - * In a clustered broker, checks that it is being called - * in a context where it is safe to modify clustered state. - * If not it aborts the process as this is a serious bug. - * - * This function is in the common library rather than the cluster - * library because it is called by code in the broker library. - */ -QPID_COMMON_EXTERN void assertClusterSafe(); - -/** - * In a non-clustered broker, returns true. - * - * In a clustered broker returns true if we are in a context where it - * is safe to modify cluster state. - * - * This function is in the common library rather than the cluster - * library because it is called by code in the broker library. - */ -QPID_COMMON_EXTERN bool isClusterSafe(); - -/** - * Mark a scope as cluster safe. Sets isClusterSafe in constructor and resets - * to previous value in destructor. - */ -class ClusterSafeScope { - public: - ClusterSafeScope(); - ~ClusterSafeScope(); - private: - bool save; -}; - -/** - * Mark a scope as cluster unsafe. Clears isClusterSafe in constructor and resets - * to previous value in destructor. - */ -class ClusterUnsafeScope { - public: - QPID_COMMON_EXTERN ClusterUnsafeScope(); - QPID_COMMON_EXTERN ~ClusterUnsafeScope(); - private: - bool save; -}; - -/** - * Enable cluster-safe assertions. By default they are no-ops. - * Called by cluster code. - */ -void enableClusterSafe(); - -}} // namespace qpid::sys - -#endif /*!QPID_SYS_CLUSTERSAFE_H*/ diff --git a/cpp/src/qpid/sys/Codec.h b/cpp/src/qpid/sys/Codec.h index ace721fbcc..e398403e47 100644 --- a/cpp/src/qpid/sys/Codec.h +++ b/cpp/src/qpid/sys/Codec.h @@ -42,7 +42,7 @@ class Codec /** Encode into buffer, return number of bytes encoded */ - virtual std::size_t encode(const char* buffer, std::size_t size) = 0; + virtual std::size_t encode(char* buffer, std::size_t size) = 0; /** Return true if we have data to encode */ virtual bool canEncode() = 0; diff --git a/cpp/src/qpid/sys/ConnectionOutputHandlerPtr.h b/cpp/src/qpid/sys/ConnectionOutputHandlerPtr.h index 95a08d15ae..53d56ad716 100644 --- a/cpp/src/qpid/sys/ConnectionOutputHandlerPtr.h +++ b/cpp/src/qpid/sys/ConnectionOutputHandlerPtr.h @@ -45,7 +45,6 @@ class ConnectionOutputHandlerPtr : public ConnectionOutputHandler size_t getBuffered() const { return next->getBuffered(); } void abort() { next->abort(); } void activateOutput() { next->activateOutput(); } - void giveReadCredit(int32_t credit) { next->giveReadCredit(credit); } void send(framing::AMQFrame& f) { next->send(f); } private: diff --git a/cpp/src/qpid/sys/FileSysDir.h b/cpp/src/qpid/sys/FileSysDir.h index ffe7823f0a..7432fe39c9 100755 --- a/cpp/src/qpid/sys/FileSysDir.h +++ b/cpp/src/qpid/sys/FileSysDir.h @@ -54,6 +54,15 @@ class FileSysDir void mkdir(void); + typedef void Callback(const std::string&); + + /** + * Call the Callback function for every regular file in the directory + * + * @param cb Callback function that receives the full path to the file + */ + void forEachFile(Callback cb) const; + std::string getPath () { return dirPath; } }; diff --git a/cpp/src/qpid/sys/OutputControl.h b/cpp/src/qpid/sys/OutputControl.h index eae99beb0f..0d801e9d16 100644 --- a/cpp/src/qpid/sys/OutputControl.h +++ b/cpp/src/qpid/sys/OutputControl.h @@ -1,3 +1,6 @@ +#ifndef QPID_SYS_OUTPUT_CONTROL_H +#define QPID_SYS_OUTPUT_CONTROL_H + /* * * Licensed to the Apache Software Foundation (ASF) under one @@ -21,9 +24,6 @@ #include "qpid/sys/IntegerTypes.h" -#ifndef _OutputControl_ -#define _OutputControl_ - namespace qpid { namespace sys { @@ -33,11 +33,10 @@ namespace sys { virtual ~OutputControl() {} virtual void abort() = 0; virtual void activateOutput() = 0; - virtual void giveReadCredit(int32_t credit) = 0; }; } } -#endif +#endif /*!QPID_SYS_OUTPUT_CONTROL_H*/ diff --git a/cpp/src/qpid/sys/ProtocolFactory.h b/cpp/src/qpid/sys/ProtocolFactory.h index 4d198a92da..236398c111 100644 --- a/cpp/src/qpid/sys/ProtocolFactory.h +++ b/cpp/src/qpid/sys/ProtocolFactory.h @@ -42,10 +42,10 @@ class ProtocolFactory : public qpid::SharedObject<ProtocolFactory> virtual void accept(boost::shared_ptr<Poller>, ConnectionCodec::Factory*) = 0; virtual void connect( boost::shared_ptr<Poller>, + const std::string& name, const std::string& host, const std::string& port, ConnectionCodec::Factory* codec, ConnectFailedCallback failed) = 0; - virtual bool supports(const std::string& /*capability*/) { return false; } }; inline ProtocolFactory::~ProtocolFactory() {} diff --git a/cpp/src/qpid/sys/RdmaIOPlugin.cpp b/cpp/src/qpid/sys/RdmaIOPlugin.cpp index b491d28d0a..51cc0ed109 100644 --- a/cpp/src/qpid/sys/RdmaIOPlugin.cpp +++ b/cpp/src/qpid/sys/RdmaIOPlugin.cpp @@ -23,6 +23,7 @@ #include "qpid/Plugin.h" #include "qpid/broker/Broker.h" +#include "qpid/broker/NameGenerator.h" #include "qpid/framing/AMQP_HighestVersion.h" #include "qpid/log/Statement.h" #include "qpid/sys/rdma/RdmaIO.h" @@ -67,7 +68,6 @@ class RdmaIOHandler : public OutputControl { void close(); void abort(); void activateOutput(); - void giveReadCredit(int32_t credit); void initProtocolOut(); // Input side @@ -83,7 +83,7 @@ class RdmaIOHandler : public OutputControl { }; RdmaIOHandler::RdmaIOHandler(Rdma::Connection::intrusive_ptr c, qpid::sys::ConnectionCodec::Factory* f) : - identifier(c->getFullName()), + identifier(broker::QPID_NAME_PREFIX+c->getFullName()), factory(f), codec(0), readError(false), @@ -199,10 +199,6 @@ void RdmaIOHandler::full(Rdma::AsynchIO&) { QPID_LOG(debug, "Rdma: buffer full [" << identifier << "]"); } -// TODO: Dummy implementation of read throttling -void RdmaIOHandler::giveReadCredit(int32_t) { -} - // The logic here is subtly different from TCP as RDMA is message oriented // so we define that an RDMA message is a frame - in this case there is no putting back // of any message remainder - there shouldn't be any. And what we read here can't be @@ -250,7 +246,7 @@ class RdmaIOProtocolFactory : public ProtocolFactory { public: RdmaIOProtocolFactory(int16_t port, int backlog); void accept(Poller::shared_ptr, ConnectionCodec::Factory*); - void connect(Poller::shared_ptr, const string& host, const std::string& port, ConnectionCodec::Factory*, ConnectFailedCallback); + void connect(Poller::shared_ptr, const std::string& name, const string& host, const std::string& port, ConnectionCodec::Factory*, ConnectFailedCallback); uint16_t getPort() const; @@ -371,6 +367,7 @@ void RdmaIOProtocolFactory::connected(Poller::shared_ptr poller, Rdma::Connectio void RdmaIOProtocolFactory::connect( Poller::shared_ptr poller, + const std::string& /*name*/, const std::string& host, const std::string& port, ConnectionCodec::Factory* f, ConnectFailedCallback failed) diff --git a/cpp/src/qpid/sys/SecurityLayer.h b/cpp/src/qpid/sys/SecurityLayer.h index 52bc40e352..317ada16de 100644 --- a/cpp/src/qpid/sys/SecurityLayer.h +++ b/cpp/src/qpid/sys/SecurityLayer.h @@ -33,8 +33,12 @@ namespace sys { class SecurityLayer : public Codec { public: + SecurityLayer(int ssf_) : ssf(ssf_) {} + int getSsf() const { return ssf; } virtual void init(Codec*) = 0; virtual ~SecurityLayer() {} + private: + int ssf; }; }} // namespace qpid::sys diff --git a/cpp/src/qpid/sys/SecuritySettings.h b/cpp/src/qpid/sys/SecuritySettings.h index bfcd08fd0f..d595cad660 100644 --- a/cpp/src/qpid/sys/SecuritySettings.h +++ b/cpp/src/qpid/sys/SecuritySettings.h @@ -21,6 +21,8 @@ * under the License. * */ +#include <string> + namespace qpid { namespace sys { diff --git a/cpp/src/qpid/sys/Socket.h b/cpp/src/qpid/sys/Socket.h index defec4879c..38183bd5fd 100644 --- a/cpp/src/qpid/sys/Socket.h +++ b/cpp/src/qpid/sys/Socket.h @@ -22,7 +22,6 @@ * */ -#include "qpid/sys/IOHandle.h" #include "qpid/sys/IntegerTypes.h" #include "qpid/CommonImportExport.h" #include <string> @@ -31,45 +30,43 @@ namespace qpid { namespace sys { class Duration; +class IOHandle; class SocketAddress; -class QPID_COMMON_CLASS_EXTERN Socket : public IOHandle +class Socket { public: - /** Create a socket wrapper for descriptor. */ - QPID_COMMON_EXTERN Socket(); + virtual ~Socket() {}; - /** Create a new Socket which is the same address family as this one */ - QPID_COMMON_EXTERN Socket* createSameTypeSocket() const; + virtual operator const IOHandle&() const = 0; /** Set socket non blocking */ - void setNonblocking() const; + virtual void setNonblocking() const = 0; - QPID_COMMON_EXTERN void setTcpNoDelay() const; + virtual void setTcpNoDelay() const = 0; - QPID_COMMON_EXTERN void connect(const std::string& host, const std::string& port) const; - QPID_COMMON_EXTERN void connect(const SocketAddress&) const; + virtual void connect(const SocketAddress&) const = 0; + virtual void finishConnect(const SocketAddress&) const = 0; - QPID_COMMON_EXTERN void close() const; + virtual void close() const = 0; /** Bind to a port and start listening. *@param port 0 means choose an available port. *@param backlog maximum number of pending connections. *@return The bound port. */ - QPID_COMMON_EXTERN int listen(const std::string& host = "", const std::string& port = "0", int backlog = 10) const; - QPID_COMMON_EXTERN int listen(const SocketAddress&, int backlog = 10) const; + virtual int listen(const SocketAddress&, int backlog = 10) const = 0; /** * Returns an address (host and port) for the remote end of the * socket */ - QPID_COMMON_EXTERN std::string getPeerAddress() const; + virtual std::string getPeerAddress() const = 0; /** * Returns an address (host and port) for the local end of the * socket */ - QPID_COMMON_EXTERN std::string getLocalAddress() const; + virtual std::string getLocalAddress() const = 0; /** * Returns the full address of the connection: local and remote host and port. @@ -80,31 +77,24 @@ public: * Returns the error code stored in the socket. This may be used * to determine the result of a non-blocking connect. */ - QPID_COMMON_EXTERN int getError() const; + virtual int getError() const = 0; /** Accept a connection from a socket that is already listening * and has an incoming connection */ - QPID_COMMON_EXTERN Socket* accept() const; + virtual Socket* accept() const = 0; - // TODO The following are raw operations, maybe they need better wrapping? - QPID_COMMON_EXTERN int read(void *buf, size_t count) const; - QPID_COMMON_EXTERN int write(const void *buf, size_t count) const; + virtual int read(void *buf, size_t count) const = 0; + virtual int write(const void *buf, size_t count) const = 0; -private: - /** Create socket */ - void createSocket(const SocketAddress&) const; - -public: - /** Construct socket with existing handle */ - Socket(IOHandlePrivate*); - -protected: - mutable std::string localname; - mutable std::string peername; - mutable bool nonblocking; - mutable bool nodelay; + /* Transport security related: */ + virtual int getKeyLen() const = 0; + virtual std::string getClientAuthId() const = 0; }; +/** Make the default socket for whatever platform we are executing on + */ +QPID_COMMON_EXTERN Socket* createSocket(); + }} #endif /*!_sys_Socket_h*/ diff --git a/cpp/src/qpid/sys/SocketAddress.h b/cpp/src/qpid/sys/SocketAddress.h index dcca109d94..a4da5cca79 100644 --- a/cpp/src/qpid/sys/SocketAddress.h +++ b/cpp/src/qpid/sys/SocketAddress.h @@ -44,11 +44,12 @@ public: QPID_COMMON_EXTERN bool nextAddress(); QPID_COMMON_EXTERN std::string asString(bool numeric=true) const; + QPID_COMMON_EXTERN std::string getHost() const; QPID_COMMON_EXTERN void setAddrInfoPort(uint16_t port); QPID_COMMON_EXTERN static std::string asString(::sockaddr const * const addr, size_t addrlen); QPID_COMMON_EXTERN static uint16_t getPort(::sockaddr const * const addr); - + private: std::string host; diff --git a/cpp/src/qpid/sys/SslPlugin.cpp b/cpp/src/qpid/sys/SslPlugin.cpp index 069e97758e..a40da24eb8 100644 --- a/cpp/src/qpid/sys/SslPlugin.cpp +++ b/cpp/src/qpid/sys/SslPlugin.cpp @@ -22,19 +22,19 @@ #include "qpid/sys/ProtocolFactory.h" #include "qpid/Plugin.h" -#include "qpid/sys/ssl/check.h" -#include "qpid/sys/ssl/util.h" -#include "qpid/sys/ssl/SslHandler.h" +#include "qpid/broker/Broker.h" +#include "qpid/broker/NameGenerator.h" +#include "qpid/log/Statement.h" #include "qpid/sys/AsynchIOHandler.h" #include "qpid/sys/AsynchIO.h" -#include "qpid/sys/ssl/SslIo.h" +#include "qpid/sys/ssl/util.h" #include "qpid/sys/ssl/SslSocket.h" -#include "qpid/broker/Broker.h" -#include "qpid/log/Statement.h" +#include "qpid/sys/SocketAddress.h" +#include "qpid/sys/SystemInfo.h" +#include "qpid/sys/Poller.h" #include <boost/bind.hpp> -#include <memory> - +#include <boost/ptr_container/ptr_vector.hpp> namespace qpid { namespace sys { @@ -64,38 +64,32 @@ struct SslServerOptions : ssl::SslOptions } }; -template <class T> -class SslProtocolFactoryTmpl : public ProtocolFactory { - private: - - typedef SslAcceptorTmpl<T> SslAcceptor; - +class SslProtocolFactory : public ProtocolFactory { + boost::ptr_vector<Socket> listeners; + boost::ptr_vector<AsynchAcceptor> acceptors; Timer& brokerTimer; uint32_t maxNegotiateTime; + uint16_t listeningPort; const bool tcpNoDelay; - T listener; - const uint16_t listeningPort; - std::auto_ptr<SslAcceptor> acceptor; bool nodict; public: - SslProtocolFactoryTmpl(const SslServerOptions&, int backlog, bool nodelay, Timer& timer, uint32_t maxTime); + SslProtocolFactory(const qpid::broker::Broker::Options& opts, const SslServerOptions& options, + Timer& timer); void accept(Poller::shared_ptr, ConnectionCodec::Factory*); - void connect(Poller::shared_ptr, const std::string& host, const std::string& port, + void connect(Poller::shared_ptr, const std::string& name, const std::string& host, const std::string& port, ConnectionCodec::Factory*, - boost::function2<void, int, std::string> failed); + ConnectFailedCallback); uint16_t getPort() const; - bool supports(const std::string& capability); private: - void established(Poller::shared_ptr, const Socket&, ConnectionCodec::Factory*, - bool isClient); + void establishedIncoming(Poller::shared_ptr, const Socket&, ConnectionCodec::Factory*); + void establishedOutgoing(Poller::shared_ptr, const Socket&, ConnectionCodec::Factory*, const std::string&); + void establishedCommon(AsynchIOHandler*, Poller::shared_ptr , const Socket&); + void connectFailed(const Socket&, int, const std::string&, ConnectFailedCallback); }; -typedef SslProtocolFactoryTmpl<SslSocket> SslProtocolFactory; -typedef SslProtocolFactoryTmpl<SslMuxSocket> SslMuxProtocolFactory; - // Static instance to initialise plugin static struct SslPlugin : public Plugin { @@ -124,7 +118,7 @@ static struct SslPlugin : public Plugin { } } } - + void initialize(Target& target) { QPID_LOG(trace, "Initialising SSL plugin"); broker::Broker* broker = dynamic_cast<broker::Broker*>(&target); @@ -139,19 +133,16 @@ static struct SslPlugin : public Plugin { const broker::Broker::Options& opts = broker->getOptions(); - ProtocolFactory::shared_ptr protocol(options.multiplex ? - static_cast<ProtocolFactory*>(new SslMuxProtocolFactory(options, - opts.connectionBacklog, - opts.tcpNoDelay, - broker->getTimer(), opts.maxNegotiateTime)) : - static_cast<ProtocolFactory*>(new SslProtocolFactory(options, - opts.connectionBacklog, - opts.tcpNoDelay, - broker->getTimer(), opts.maxNegotiateTime))); - QPID_LOG(notice, "Listening for " << - (options.multiplex ? "SSL or TCP" : "SSL") << - " connections on TCP port " << - protocol->getPort()); + ProtocolFactory::shared_ptr protocol( + static_cast<ProtocolFactory*>(new SslProtocolFactory(opts, options, broker->getTimer()))); + + if (protocol->getPort()!=0 ) { + QPID_LOG(notice, "Listening for " << + (options.multiplex ? "SSL or TCP" : "SSL") << + " connections on TCP/TCP6 port " << + protocol->getPort()); + } + broker->registerProtocolFactory("ssl", protocol); } catch (const std::exception& e) { QPID_LOG(error, "Failed to initialise SSL plugin: " << e.what()); @@ -161,99 +152,133 @@ static struct SslPlugin : public Plugin { } } sslPlugin; -template <class T> -SslProtocolFactoryTmpl<T>::SslProtocolFactoryTmpl(const SslServerOptions& options, int backlog, bool nodelay, Timer& timer, uint32_t maxTime) : +namespace { + // Expand list of Interfaces and addresses to a list of addresses + std::vector<std::string> expandInterfaces(const std::vector<std::string>& interfaces) { + std::vector<std::string> addresses; + // If there are no specific interfaces listed use a single "" to listen on every interface + if (interfaces.empty()) { + addresses.push_back(""); + return addresses; + } + for (unsigned i = 0; i < interfaces.size(); ++i) { + const std::string& interface = interfaces[i]; + if (!(SystemInfo::getInterfaceAddresses(interface, addresses))) { + // We don't have an interface of that name - + // Check for IPv6 ('[' ']') brackets and remove them + // then pass to be looked up directly + if (interface[0]=='[' && interface[interface.size()-1]==']') { + addresses.push_back(interface.substr(1, interface.size()-2)); + } else { + addresses.push_back(interface); + } + } + } + return addresses; + } +} + +SslProtocolFactory::SslProtocolFactory(const qpid::broker::Broker::Options& opts, const SslServerOptions& options, + Timer& timer) : brokerTimer(timer), - maxNegotiateTime(maxTime), - tcpNoDelay(nodelay), listeningPort(listener.listen(options.port, backlog, options.certName, options.clientAuth)), + maxNegotiateTime(opts.maxNegotiateTime), + tcpNoDelay(opts.tcpNoDelay), nodict(options.nodict) -{} - -void SslEstablished(Poller::shared_ptr poller, const qpid::sys::SslSocket& s, - ConnectionCodec::Factory* f, bool isClient, - Timer& timer, uint32_t maxTime, bool tcpNoDelay, bool nodict) { - qpid::sys::ssl::SslHandler* async = new qpid::sys::ssl::SslHandler(s.getFullAddress(), f, nodict); - - if (tcpNoDelay) { - s.setTcpNoDelay(tcpNoDelay); - QPID_LOG(info, "Set TCP_NODELAY on connection to " << s.getPeerAddress()); +{ + std::vector<std::string> addresses = expandInterfaces(opts.listenInterfaces); + if (addresses.empty()) { + // We specified some interfaces, but couldn't find addresses for them + QPID_LOG(warning, "SSL: No specified network interfaces found: Not Listening"); + listeningPort = 0; } - if (isClient) { - async->setClient(); + for (unsigned i = 0; i<addresses.size(); ++i) { + QPID_LOG(debug, "Using interface: " << addresses[i]); + SocketAddress sa(addresses[i], boost::lexical_cast<std::string>(options.port)); + + // We must have at least one resolved address + QPID_LOG(info, "Listening to: " << sa.asString()) + Socket* s = options.multiplex ? + new SslMuxSocket(options.certName, options.clientAuth) : + new SslSocket(options.certName, options.clientAuth); + uint16_t lport = s->listen(sa, opts.connectionBacklog); + QPID_LOG(debug, "Listened to: " << lport); + listeners.push_back(s); + + listeningPort = lport; + + // Try any other resolved addresses + while (sa.nextAddress()) { + // Hack to ensure that all listening connections are on the same port + sa.setAddrInfoPort(listeningPort); + QPID_LOG(info, "Listening to: " << sa.asString()) + Socket* s = options.multiplex ? + new SslMuxSocket(options.certName, options.clientAuth) : + new SslSocket(options.certName, options.clientAuth); + uint16_t lport = s->listen(sa, opts.connectionBacklog); + QPID_LOG(debug, "Listened to: " << lport); + listeners.push_back(s); + } } - - qpid::sys::ssl::SslIO* aio = new qpid::sys::ssl::SslIO(s, - boost::bind(&qpid::sys::ssl::SslHandler::readbuff, async, _1, _2), - boost::bind(&qpid::sys::ssl::SslHandler::eof, async, _1), - boost::bind(&qpid::sys::ssl::SslHandler::disconnect, async, _1), - boost::bind(&qpid::sys::ssl::SslHandler::closedSocket, async, _1, _2), - boost::bind(&qpid::sys::ssl::SslHandler::nobuffs, async, _1), - boost::bind(&qpid::sys::ssl::SslHandler::idle, async, _1)); - - async->init(aio,timer, maxTime); - aio->start(poller); -} - -template <> -void SslProtocolFactory::established(Poller::shared_ptr poller, const Socket& s, - ConnectionCodec::Factory* f, bool isClient) { - const SslSocket *sslSock = dynamic_cast<const SslSocket*>(&s); - - SslEstablished(poller, *sslSock, f, isClient, brokerTimer, maxNegotiateTime, tcpNoDelay, nodict); } -template <class T> -uint16_t SslProtocolFactoryTmpl<T>::getPort() const { - return listeningPort; // Immutable no need for lock. +void SslProtocolFactory::establishedIncoming(Poller::shared_ptr poller, const Socket& s, + ConnectionCodec::Factory* f) { + AsynchIOHandler* async = new AsynchIOHandler(broker::QPID_NAME_PREFIX+s.getFullAddress(), f, false, false); + establishedCommon(async, poller, s); } -template <class T> -void SslProtocolFactoryTmpl<T>::accept(Poller::shared_ptr poller, - ConnectionCodec::Factory* fact) { - acceptor.reset( - new SslAcceptor(listener, - boost::bind(&SslProtocolFactoryTmpl<T>::established, - this, poller, _1, fact, false))); - acceptor->start(poller); +void SslProtocolFactory::establishedOutgoing(Poller::shared_ptr poller, const Socket& s, + ConnectionCodec::Factory* f, const std::string& name) { + AsynchIOHandler* async = new AsynchIOHandler(name, f, true, false); + establishedCommon(async, poller, s); } -template <> -void SslMuxProtocolFactory::established(Poller::shared_ptr poller, const Socket& s, - ConnectionCodec::Factory* f, bool isClient) { - const SslSocket *sslSock = dynamic_cast<const SslSocket*>(&s); - - if (sslSock) { - SslEstablished(poller, *sslSock, f, isClient, brokerTimer, maxNegotiateTime, tcpNoDelay, nodict); - return; - } - - AsynchIOHandler* async = new AsynchIOHandler(s.getFullAddress(), f); - +void SslProtocolFactory::establishedCommon(AsynchIOHandler* async, Poller::shared_ptr poller, const Socket& s) { if (tcpNoDelay) { s.setTcpNoDelay(); QPID_LOG(info, "Set TCP_NODELAY on connection to " << s.getPeerAddress()); } - if (isClient) { - async->setClient(); - } - AsynchIO* aio = AsynchIO::create - (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)); + AsynchIO* aio = AsynchIO::create( + 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, brokerTimer, maxNegotiateTime); aio->start(poller); } -template <class T> -void SslProtocolFactoryTmpl<T>::connect( +uint16_t SslProtocolFactory::getPort() const { + return listeningPort; // Immutable no need for lock. +} + +void SslProtocolFactory::accept(Poller::shared_ptr poller, + ConnectionCodec::Factory* fact) { + for (unsigned i = 0; i<listeners.size(); ++i) { + acceptors.push_back( + AsynchAcceptor::create(listeners[i], + boost::bind(&SslProtocolFactory::establishedIncoming, this, poller, _1, fact))); + acceptors[i].start(poller); + } +} + +void SslProtocolFactory::connectFailed( + const Socket& s, int ec, const std::string& emsg, + ConnectFailedCallback failedCb) +{ + failedCb(ec, emsg); + s.close(); + delete &s; +} + +void SslProtocolFactory::connect( Poller::shared_ptr poller, + const std::string& name, const std::string& host, const std::string& port, ConnectionCodec::Factory* fact, ConnectFailedCallback failed) @@ -264,31 +289,23 @@ void SslProtocolFactoryTmpl<T>::connect( // shutdown. The allocated SslConnector frees itself when it // is no longer needed. - qpid::sys::ssl::SslSocket* socket = new qpid::sys::ssl::SslSocket(); - new SslConnector(*socket, poller, host, port, - boost::bind(&SslProtocolFactoryTmpl<T>::established, this, poller, _1, fact, true), - failed); -} - -namespace -{ -const std::string SSL = "ssl"; -} - -template <> -bool SslProtocolFactory::supports(const std::string& capability) -{ - std::string s = capability; - transform(s.begin(), s.end(), s.begin(), tolower); - return s == SSL; -} - -template <> -bool SslMuxProtocolFactory::supports(const std::string& capability) -{ - std::string s = capability; - transform(s.begin(), s.end(), s.begin(), tolower); - return s == SSL || s == "tcp"; + Socket* socket = new qpid::sys::ssl::SslSocket(); + try { + AsynchConnector* c = AsynchConnector::create( + *socket, + host, + port, + boost::bind(&SslProtocolFactory::establishedOutgoing, + this, poller, _1, fact, name), + boost::bind(&SslProtocolFactory::connectFailed, + this, _1, _2, _3, failed)); + c->start(poller); + } catch (std::exception&) { + // TODO: Design question - should we do the error callback and also throw? + int errCode = socket->getError(); + connectFailed(*socket, errCode, strError(errCode), failed); + throw; + } } }} // namespace qpid::sys diff --git a/cpp/src/qpid/sys/TCPIOPlugin.cpp b/cpp/src/qpid/sys/TCPIOPlugin.cpp index ed7cc3748d..1ef8708cd0 100644 --- a/cpp/src/qpid/sys/TCPIOPlugin.cpp +++ b/cpp/src/qpid/sys/TCPIOPlugin.cpp @@ -20,15 +20,17 @@ */ #include "qpid/sys/ProtocolFactory.h" -#include "qpid/sys/AsynchIOHandler.h" -#include "qpid/sys/AsynchIO.h" #include "qpid/Plugin.h" +#include "qpid/broker/Broker.h" +#include "qpid/broker/NameGenerator.h" +#include "qpid/log/Statement.h" +#include "qpid/sys/AsynchIOHandler.h" +#include "qpid/sys/AsynchIO.h" #include "qpid/sys/Socket.h" #include "qpid/sys/SocketAddress.h" +#include "qpid/sys/SystemInfo.h" #include "qpid/sys/Poller.h" -#include "qpid/broker/Broker.h" -#include "qpid/log/Statement.h" #include <boost/bind.hpp> #include <boost/ptr_container/ptr_vector.hpp> @@ -47,20 +49,19 @@ class AsynchIOProtocolFactory : public ProtocolFactory { const bool tcpNoDelay; public: - AsynchIOProtocolFactory(const std::string& host, const std::string& port, - int backlog, bool nodelay, - Timer& timer, uint32_t maxTime, - bool shouldListen); + AsynchIOProtocolFactory(const qpid::broker::Broker::Options& opts, Timer& timer, bool shouldListen); void accept(Poller::shared_ptr, ConnectionCodec::Factory*); - void connect(Poller::shared_ptr, const std::string& host, const std::string& port, + void connect(Poller::shared_ptr, const std::string& name, + const std::string& host, const std::string& port, ConnectionCodec::Factory*, ConnectFailedCallback); uint16_t getPort() const; private: - void established(Poller::shared_ptr, const Socket&, ConnectionCodec::Factory*, - bool isClient); + void establishedIncoming(Poller::shared_ptr, const Socket&, ConnectionCodec::Factory*); + void establishedOutgoing(Poller::shared_ptr, const Socket&, ConnectionCodec::Factory*, const std::string&); + void establishedCommon(AsynchIOHandler*, Poller::shared_ptr , const Socket&); void connectFailed(const Socket&, int, const std::string&, ConnectFailedCallback); }; @@ -93,14 +94,9 @@ static class TCPIOPlugin : public Plugin { bool shouldListen = !sslMultiplexEnabled(); ProtocolFactory::shared_ptr protocolt( - new AsynchIOProtocolFactory( - "", boost::lexical_cast<std::string>(opts.port), - opts.connectionBacklog, - opts.tcpNoDelay, - broker->getTimer(), opts.maxNegotiateTime, - shouldListen)); - - if (shouldListen) { + new AsynchIOProtocolFactory(opts, broker->getTimer(),shouldListen)); + + if (shouldListen && protocolt->getPort()!=0 ) { QPID_LOG(notice, "Listening on TCP/TCP6 port " << protocolt->getPort()); } @@ -109,54 +105,93 @@ static class TCPIOPlugin : public Plugin { } } tcpPlugin; -AsynchIOProtocolFactory::AsynchIOProtocolFactory(const std::string& host, const std::string& port, - int backlog, bool nodelay, - Timer& timer, uint32_t maxTime, - bool shouldListen) : +namespace { + // Expand list of Interfaces and addresses to a list of addresses + std::vector<std::string> expandInterfaces(const std::vector<std::string>& interfaces) { + std::vector<std::string> addresses; + // If there are no specific interfaces listed use a single "" to listen on every interface + if (interfaces.empty()) { + addresses.push_back(""); + return addresses; + } + for (unsigned i = 0; i < interfaces.size(); ++i) { + const std::string& interface = interfaces[i]; + if (!(SystemInfo::getInterfaceAddresses(interface, addresses))) { + // We don't have an interface of that name - + // Check for IPv6 ('[' ']') brackets and remove them + // then pass to be looked up directly + if (interface[0]=='[' && interface[interface.size()-1]==']') { + addresses.push_back(interface.substr(1, interface.size()-2)); + } else { + addresses.push_back(interface); + } + } + } + return addresses; + } +} + +AsynchIOProtocolFactory::AsynchIOProtocolFactory(const qpid::broker::Broker::Options& opts, Timer& timer, bool shouldListen) : brokerTimer(timer), - maxNegotiateTime(maxTime), - tcpNoDelay(nodelay) + maxNegotiateTime(opts.maxNegotiateTime), + tcpNoDelay(opts.tcpNoDelay) { if (!shouldListen) { - listeningPort = boost::lexical_cast<uint16_t>(port); + listeningPort = boost::lexical_cast<uint16_t>(opts.port); return; } - SocketAddress sa(host, port); - - // We must have at least one resolved address - QPID_LOG(info, "Listening to: " << sa.asString()) - Socket* s = new Socket; - uint16_t lport = s->listen(sa, backlog); - QPID_LOG(debug, "Listened to: " << lport); - listeners.push_back(s); + std::vector<std::string> addresses = expandInterfaces(opts.listenInterfaces); + if (addresses.empty()) { + // We specified some interfaces, but couldn't find addresses for them + QPID_LOG(warning, "TCP/TCP6: No specified network interfaces found: Not Listening"); + listeningPort = 0; + } - listeningPort = lport; + for (unsigned i = 0; i<addresses.size(); ++i) { + QPID_LOG(debug, "Using interface: " << addresses[i]); + SocketAddress sa(addresses[i], boost::lexical_cast<std::string>(opts.port)); - // Try any other resolved addresses - while (sa.nextAddress()) { - // Hack to ensure that all listening connections are on the same port - sa.setAddrInfoPort(listeningPort); + // We must have at least one resolved address QPID_LOG(info, "Listening to: " << sa.asString()) - Socket* s = new Socket; - uint16_t lport = s->listen(sa, backlog); + Socket* s = createSocket(); + uint16_t lport = s->listen(sa, opts.connectionBacklog); QPID_LOG(debug, "Listened to: " << lport); listeners.push_back(s); + + listeningPort = lport; + + // Try any other resolved addresses + while (sa.nextAddress()) { + // Hack to ensure that all listening connections are on the same port + sa.setAddrInfoPort(listeningPort); + QPID_LOG(info, "Listening to: " << sa.asString()) + Socket* s = createSocket(); + uint16_t lport = s->listen(sa, opts.connectionBacklog); + QPID_LOG(debug, "Listened to: " << lport); + listeners.push_back(s); + } } +} +void AsynchIOProtocolFactory::establishedIncoming(Poller::shared_ptr poller, const Socket& s, + ConnectionCodec::Factory* f) { + AsynchIOHandler* async = new AsynchIOHandler(broker::QPID_NAME_PREFIX+s.getFullAddress(), f, false, false); + establishedCommon(async, poller, s); } -void AsynchIOProtocolFactory::established(Poller::shared_ptr poller, const Socket& s, - ConnectionCodec::Factory* f, bool isClient) { - AsynchIOHandler* async = new AsynchIOHandler(s.getFullAddress(), f); +void AsynchIOProtocolFactory::establishedOutgoing(Poller::shared_ptr poller, const Socket& s, + ConnectionCodec::Factory* f, const std::string& name) { + AsynchIOHandler* async = new AsynchIOHandler(name, f, true, false); + establishedCommon(async, poller, s); +} +void AsynchIOProtocolFactory::establishedCommon(AsynchIOHandler* async, Poller::shared_ptr poller, const Socket& s) { if (tcpNoDelay) { s.setTcpNoDelay(); QPID_LOG(info, "Set TCP_NODELAY on connection to " << s.getPeerAddress()); } - if (isClient) - async->setClient(); AsynchIO* aio = AsynchIO::create (s, boost::bind(&AsynchIOHandler::readbuff, async, _1, _2), @@ -179,7 +214,7 @@ void AsynchIOProtocolFactory::accept(Poller::shared_ptr poller, for (unsigned i = 0; i<listeners.size(); ++i) { acceptors.push_back( AsynchAcceptor::create(listeners[i], - boost::bind(&AsynchIOProtocolFactory::established, this, poller, _1, fact, false))); + boost::bind(&AsynchIOProtocolFactory::establishedIncoming, this, poller, _1, fact))); acceptors[i].start(poller); } } @@ -195,6 +230,7 @@ void AsynchIOProtocolFactory::connectFailed( void AsynchIOProtocolFactory::connect( Poller::shared_ptr poller, + const std::string& name, const std::string& host, const std::string& port, ConnectionCodec::Factory* fact, ConnectFailedCallback failed) @@ -204,14 +240,14 @@ void AsynchIOProtocolFactory::connect( // upon connection failure or by the AsynchIO upon connection // shutdown. The allocated AsynchConnector frees itself when it // is no longer needed. - Socket* socket = new Socket(); + Socket* socket = createSocket(); try { AsynchConnector* c = AsynchConnector::create( *socket, host, port, - boost::bind(&AsynchIOProtocolFactory::established, - this, poller, _1, fact, true), + boost::bind(&AsynchIOProtocolFactory::establishedOutgoing, + this, poller, _1, fact, name), boost::bind(&AsynchIOProtocolFactory::connectFailed, this, _1, _2, _3, failed)); c->start(poller); diff --git a/cpp/src/qpid/sys/Timer.cpp b/cpp/src/qpid/sys/Timer.cpp index 973c6bd8b7..f8eef2c9ec 100644 --- a/cpp/src/qpid/sys/Timer.cpp +++ b/cpp/src/qpid/sys/Timer.cpp @@ -96,18 +96,13 @@ void TimerTask::cancel() { state = CANCELLED; } -void TimerTask::setFired() { - // Set nextFireTime to just before now, making readyToFire() true. - nextFireTime = AbsTime(sys::now(), Duration(-1)); -} - - +// TODO AStitcher 21/08/09 The threshholds for emitting warnings are a little arbitrary Timer::Timer() : active(false), late(50 * TIME_MSEC), overran(2 * TIME_MSEC), lateCancel(500 * TIME_MSEC), - warn(5 * TIME_SEC) + warn(60 * TIME_SEC) { start(); } @@ -133,7 +128,6 @@ public: } }; -// TODO AStitcher 21/08/09 The threshholds for emitting warnings are a little arbitrary void Timer::run() { Monitor::ScopedLock l(monitor); @@ -151,10 +145,6 @@ void Timer::run() { TimerTaskCallbackScope s(*t); if (s) { - { - Monitor::ScopedUnlock u(monitor); - drop(t); - } if (delay > lateCancel) { QPID_LOG(debug, t->name << " cancelled timer woken up " << delay / TIME_MSEC << "ms late"); @@ -171,8 +161,8 @@ void Timer::run() if (!tasks.empty()) { overrun = Duration(tasks.top()->nextFireTime, end); } - bool warningsEnabled; - QPID_LOG_TEST(warning, warningsEnabled); + bool warningsEnabled; // TimerWarning enabled + QPID_LOG_TEST(debug, warningsEnabled); // TimerWarning emitted at debug level if (warningsEnabled) { if (overrun > overran) { if (delay > overran) // if delay is significant to an overrun. @@ -235,9 +225,6 @@ void Timer::fire(boost::intrusive_ptr<TimerTask> t) { } } -// Provided for subclasses: called when a task is droped. -void Timer::drop(boost::intrusive_ptr<TimerTask>) {} - bool operator<(const intrusive_ptr<TimerTask>& a, const intrusive_ptr<TimerTask>& b) { diff --git a/cpp/src/qpid/sys/Timer.h b/cpp/src/qpid/sys/Timer.h index 5731b8d977..5045009609 100644 --- a/cpp/src/qpid/sys/Timer.h +++ b/cpp/src/qpid/sys/Timer.h @@ -67,10 +67,6 @@ class TimerTask : public RefCounted { std::string getName() const { return name; } - // Move the nextFireTime so readyToFire is true. - // Used by the cluster, where tasks are fired on cluster events, not on local time. - QPID_COMMON_EXTERN void setFired(); - protected: // Must be overridden with callback virtual void fire() = 0; @@ -99,7 +95,7 @@ class Timer : private Runnable { protected: QPID_COMMON_EXTERN virtual void fire(boost::intrusive_ptr<TimerTask> task); - QPID_COMMON_EXTERN virtual void drop(boost::intrusive_ptr<TimerTask> task); + // Allow derived classes to change the late/overran thresholds. Duration late; Duration overran; diff --git a/cpp/src/qpid/sys/TimerWarnings.cpp b/cpp/src/qpid/sys/TimerWarnings.cpp index 85e26da54a..00fb0d9db6 100644 --- a/cpp/src/qpid/sys/TimerWarnings.cpp +++ b/cpp/src/qpid/sys/TimerWarnings.cpp @@ -56,18 +56,18 @@ void TimerWarnings::log() { std::string task = i->first; TaskStats& stats = i->second; if (stats.lateDelay.count) - QPID_LOG(info, task << " task late " + QPID_LOG(debug, task << " task late " << stats.lateDelay.count << " times by " << stats.lateDelay.average()/TIME_MSEC << "ms on average."); if (stats.overranOverrun.count) - QPID_LOG(info, task << " task overran " + QPID_LOG(debug, task << " task overran " << stats.overranOverrun.count << " times by " << stats.overranOverrun.average()/TIME_MSEC << "ms (taking " << stats.overranTime.average() << "ns) on average."); if (stats.lateAndOverranOverrun.count) - QPID_LOG(info, task << " task late and overran " + QPID_LOG(debug, task << " task late and overran " << stats.lateAndOverranOverrun.count << " times: late " << stats.lateAndOverranDelay.average()/TIME_MSEC << "ms, overran " << stats.lateAndOverranOverrun.average()/TIME_MSEC << "ms (taking " diff --git a/cpp/src/qpid/sys/alloca.h b/cpp/src/qpid/sys/alloca.h deleted file mode 100644 index b3f59b7c3f..0000000000 --- a/cpp/src/qpid/sys/alloca.h +++ /dev/null @@ -1,42 +0,0 @@ -#ifndef QPID_SYS_ALLOCA_H -#define QPID_SYS_ALLOCA_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. - * - */ - -#if (defined(_WINDOWS) || defined (WIN32)) -# include <malloc.h> - -# if defined(_MSC_VER) -# ifdef alloc -# undef alloc -# endif -# define alloc _alloc -# ifdef alloca -# undef alloca -# endif -# define alloca _alloca -# endif -#endif -#if !defined _WINDOWS && !defined WIN32 -# include <alloca.h> -#endif - -#endif /*!QPID_SYS_ALLOCA_H*/ diff --git a/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.cpp b/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.cpp index 29b91f3e7a..79d9d08a59 100644 --- a/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.cpp +++ b/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.cpp @@ -29,8 +29,8 @@ namespace qpid { namespace sys { namespace cyrus { -CyrusSecurityLayer::CyrusSecurityLayer(sasl_conn_t* c, uint16_t maxFrameSize) : - conn(c), decrypted(0), decryptedSize(0), encrypted(0), encryptedSize(0), codec(0), maxInputSize(0), +CyrusSecurityLayer::CyrusSecurityLayer(sasl_conn_t* c, uint16_t maxFrameSize, int ssf) : + SecurityLayer(ssf), conn(c), decrypted(0), decryptedSize(0), encrypted(0), encryptedSize(0), codec(0), maxInputSize(0), decodeBuffer(maxFrameSize), encodeBuffer(maxFrameSize), encoded(0) { const void* value(0); @@ -68,7 +68,7 @@ size_t CyrusSecurityLayer::decode(const char* input, size_t size) return size; } -size_t CyrusSecurityLayer::encode(const char* buffer, size_t size) +size_t CyrusSecurityLayer::encode(char* buffer, size_t size) { size_t processed = 0;//records how many bytes have been written to buffer do { @@ -92,12 +92,12 @@ size_t CyrusSecurityLayer::encode(const char* buffer, size_t size) //can't fit all encrypted data in the buffer we've //been given, copy in what we can and hold on to the //rest until the next call - ::memcpy(const_cast<char*>(buffer + processed), encrypted, remaining); + ::memcpy(buffer + processed, encrypted, remaining); processed += remaining; encrypted += remaining; encryptedSize -= remaining; } else { - ::memcpy(const_cast<char*>(buffer + processed), encrypted, encryptedSize); + ::memcpy(buffer + processed, encrypted, encryptedSize); processed += encryptedSize; encrypted = 0; encryptedSize = 0; diff --git a/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.h b/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.h index 1645cf1a58..ae86ba5569 100644 --- a/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.h +++ b/cpp/src/qpid/sys/cyrus/CyrusSecurityLayer.h @@ -37,9 +37,9 @@ namespace cyrus { class CyrusSecurityLayer : public qpid::sys::SecurityLayer { public: - CyrusSecurityLayer(sasl_conn_t*, uint16_t maxFrameSize); + CyrusSecurityLayer(sasl_conn_t*, uint16_t maxFrameSize, int ssf); size_t decode(const char* buffer, size_t size); - size_t encode(const char* buffer, size_t size); + size_t encode(char* buffer, size_t size); bool canEncode(); void init(qpid::sys::Codec*); private: diff --git a/cpp/src/qpid/sys/epoll/EpollPoller.cpp b/cpp/src/qpid/sys/epoll/EpollPoller.cpp index c23403c66d..6fdf99637f 100644 --- a/cpp/src/qpid/sys/epoll/EpollPoller.cpp +++ b/cpp/src/qpid/sys/epoll/EpollPoller.cpp @@ -20,7 +20,6 @@ */ #include "qpid/sys/Poller.h" -#include "qpid/sys/IOHandle.h" #include "qpid/sys/Mutex.h" #include "qpid/sys/AtomicCount.h" #include "qpid/sys/DeletionManager.h" @@ -64,12 +63,12 @@ class PollerHandlePrivate { }; ::__uint32_t events; - const IOHandlePrivate* ioHandle; + const IOHandle* ioHandle; PollerHandle* pollerHandle; FDStat stat; Mutex lock; - PollerHandlePrivate(const IOHandlePrivate* h, PollerHandle* p) : + PollerHandlePrivate(const IOHandle* h, PollerHandle* p) : events(0), ioHandle(h), pollerHandle(p), @@ -77,7 +76,7 @@ class PollerHandlePrivate { } int fd() const { - return toFd(ioHandle); + return ioHandle->fd; } bool isActive() const { @@ -138,7 +137,7 @@ class PollerHandlePrivate { }; PollerHandle::PollerHandle(const IOHandle& h) : - impl(new PollerHandlePrivate(h.impl, this)) + impl(new PollerHandlePrivate(&h, this)) {} PollerHandle::~PollerHandle() { @@ -385,6 +384,7 @@ void PollerPrivate::resetMode(PollerHandlePrivate& eh) { int rc = ::epoll_ctl(epollFd, EPOLL_CTL_MOD, eh.fd(), &epe); // If something has closed the fd in the meantime try adding it back if (rc ==-1 && errno == ENOENT) { + eh.setIdle(); // Reset our handle as if starting from scratch rc = ::epoll_ctl(epollFd, EPOLL_CTL_ADD, eh.fd(), &epe); } QPID_POSIX_CHECK(rc); diff --git a/cpp/src/qpid/sys/posix/AsynchIO.cpp b/cpp/src/qpid/sys/posix/AsynchIO.cpp index 31355627cd..353a55f50c 100644 --- a/cpp/src/qpid/sys/posix/AsynchIO.cpp +++ b/cpp/src/qpid/sys/posix/AsynchIO.cpp @@ -143,6 +143,7 @@ class AsynchConnector : public qpid::sys::AsynchConnector, private: void connComplete(DispatchHandle& handle); + void requestedCall(RequestCallback rCb); private: ConnectedCallback connCallback; @@ -158,6 +159,7 @@ public: FailedCallback failCb); void start(Poller::shared_ptr poller); void stop(); + void requestCallback(RequestCallback rCb); }; AsynchConnector::AsynchConnector(const Socket& s, @@ -191,11 +193,30 @@ void AsynchConnector::stop() stopWatch(); } +void AsynchConnector::requestCallback(RequestCallback callback) { + // TODO creating a function object every time isn't all that + // efficient - if this becomes heavily used do something better (what?) + assert(callback); + DispatchHandle::call(boost::bind(&AsynchConnector::requestedCall, this, callback)); +} + +void AsynchConnector::requestedCall(RequestCallback callback) { + assert(callback); + callback(*this); +} + void AsynchConnector::connComplete(DispatchHandle& h) { int errCode = socket.getError(); if (errCode == 0) { h.stopWatch(); + try { + socket.finishConnect(sa); + } catch (const std::exception& e) { + failCallback(socket, 0, e.what()); + DispatchHandle::doDelete(); + return; + } connCallback(socket); } else { // Retry while we cause an immediate exception @@ -247,10 +268,9 @@ public: virtual void notifyPendingWrite(); virtual void queueWriteClose(); virtual bool writeQueueEmpty(); - virtual void startReading(); - virtual void stopReading(); virtual void requestCallback(RequestCallback); virtual BufferBase* getQueuedBuffer(); + virtual SecuritySettings getSecuritySettings(); private: ~AsynchIO(); @@ -282,13 +302,6 @@ private: * thread processing this handle. */ volatile bool writePending; - /** - * This records whether we've been reading is flow controlled: - * it's safe as a simple boolean as the only way to be stopped - * is in calls only allowed in the callback context, the only calls - * checking it are also in calls only allowed in callback context. - */ - volatile bool readingStopped; }; AsynchIO::AsynchIO(const Socket& s, @@ -307,8 +320,7 @@ AsynchIO::AsynchIO(const Socket& s, idleCallback(iCb), socket(s), queuedClose(false), - writePending(false), - readingStopped(false) { + writePending(false) { s.setNonblocking(); } @@ -344,7 +356,7 @@ void AsynchIO::queueReadBuffer(BufferBase* buff) { bool queueWasEmpty = bufferQueue.empty(); bufferQueue.push_back(buff); - if (queueWasEmpty && !readingStopped) + if (queueWasEmpty) DispatchHandle::rewatchRead(); } @@ -354,7 +366,7 @@ void AsynchIO::unread(BufferBase* buff) { bool queueWasEmpty = bufferQueue.empty(); bufferQueue.push_front(buff); - if (queueWasEmpty && !readingStopped) + if (queueWasEmpty) DispatchHandle::rewatchRead(); } @@ -386,17 +398,6 @@ bool AsynchIO::writeQueueEmpty() { return writeQueue.empty(); } -// This can happen outside the callback context -void AsynchIO::startReading() { - readingStopped = false; - DispatchHandle::rewatchRead(); -} - -void AsynchIO::stopReading() { - readingStopped = true; - DispatchHandle::unwatchRead(); -} - void AsynchIO::requestCallback(RequestCallback callback) { // TODO creating a function object every time isn't all that // efficient - if this becomes heavily used do something better (what?) @@ -429,11 +430,6 @@ AsynchIO::BufferBase* AsynchIO::getQueuedBuffer() { * to put it in and reading is not stopped by flow control. */ void AsynchIO::readable(DispatchHandle& h) { - if (readingStopped) { - // We have been flow controlled. - QPID_PROBE1(asynchio_read_flowcontrolled, &h); - return; - } AbsTime readStartTime = AbsTime::now(); size_t total = 0; int readCalls = 0; @@ -455,12 +451,6 @@ void AsynchIO::readable(DispatchHandle& h) { total += rc; readCallback(*this, buff); - if (readingStopped) { - // We have been flow controlled. - QPID_PROBE4(asynchio_read_finished_flowcontrolled, &h, duration, total, readCalls); - break; - } - if (rc != readCount) { // If we didn't fill the read buffer then time to stop reading QPID_PROBE4(asynchio_read_finished_done, &h, duration, total, readCalls); @@ -626,6 +616,13 @@ void AsynchIO::close(DispatchHandle& h) { } } +SecuritySettings AsynchIO::getSecuritySettings() { + SecuritySettings settings; + settings.ssf = socket.getKeyLen(); + settings.authid = socket.getClientAuthId(); + return settings; +} + } // namespace posix AsynchAcceptor* AsynchAcceptor::create(const Socket& s, diff --git a/cpp/src/qpid/sys/posix/Socket.cpp b/cpp/src/qpid/sys/posix/BSDSocket.cpp index 77ae1af60c..7c31b13ae9 100644 --- a/cpp/src/qpid/sys/posix/Socket.cpp +++ b/cpp/src/qpid/sys/posix/BSDSocket.cpp @@ -19,7 +19,7 @@ * */ -#include "qpid/sys/Socket.h" +#include "qpid/sys/posix/BSDSocket.h" #include "qpid/sys/SocketAddress.h" #include "qpid/sys/posix/check.h" @@ -67,25 +67,41 @@ uint16_t getLocalPort(int fd) } } -Socket::Socket() : - IOHandle(new IOHandlePrivate), +BSDSocket::BSDSocket() : + fd(-1), + handle(new IOHandle), nonblocking(false), nodelay(false) {} -Socket::Socket(IOHandlePrivate* h) : - IOHandle(h), +Socket* createSocket() +{ + return new BSDSocket; +} + +BSDSocket::BSDSocket(int fd0) : + fd(fd0), + handle(new IOHandle(fd)), nonblocking(false), nodelay(false) {} -void Socket::createSocket(const SocketAddress& sa) const +BSDSocket::~BSDSocket() +{} + +BSDSocket::operator const IOHandle&() const +{ + return *handle; +} + +void BSDSocket::createSocket(const SocketAddress& sa) const { - int& socket = impl->fd; - if (socket != -1) Socket::close(); + int& socket = fd; + if (socket != -1) BSDSocket::close(); int s = ::socket(getAddrInfo(sa).ai_family, getAddrInfo(sa).ai_socktype, 0); if (s < 0) throw QPID_POSIX_ERROR(errno); socket = s; + *handle = IOHandle(s); try { if (nonblocking) setNonblocking(); @@ -98,50 +114,31 @@ void Socket::createSocket(const SocketAddress& sa) const } catch (std::exception&) { ::close(s); socket = -1; + *handle = IOHandle(); throw; } } -Socket* Socket::createSameTypeSocket() const { - int& socket = impl->fd; - // Socket currently has no actual socket attached - if (socket == -1) - return new Socket; - - ::sockaddr_storage sa; - ::socklen_t salen = sizeof(sa); - QPID_POSIX_CHECK(::getsockname(socket, (::sockaddr*)&sa, &salen)); - int s = ::socket(sa.ss_family, SOCK_STREAM, 0); // Currently only work with SOCK_STREAM - if (s < 0) throw QPID_POSIX_ERROR(errno); - return new Socket(new IOHandlePrivate(s)); -} - -void Socket::setNonblocking() const { - int& socket = impl->fd; +void BSDSocket::setNonblocking() const { + int& socket = fd; nonblocking = true; if (socket != -1) { QPID_POSIX_CHECK(::fcntl(socket, F_SETFL, O_NONBLOCK)); } } -void Socket::setTcpNoDelay() const +void BSDSocket::setTcpNoDelay() const { - int& socket = impl->fd; + int& socket = fd; nodelay = true; if (socket != -1) { int flag = 1; - int result = ::setsockopt(impl->fd, IPPROTO_TCP, TCP_NODELAY, (char *)&flag, sizeof(flag)); + int result = ::setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (char *)&flag, sizeof(flag)); QPID_POSIX_CHECK(result); } } -void Socket::connect(const std::string& host, const std::string& port) const -{ - SocketAddress sa(host, port); - connect(sa); -} - -void Socket::connect(const SocketAddress& addr) const +void BSDSocket::connect(const SocketAddress& addr) const { // The display name for an outbound connection needs to be the name that was specified // for the address rather than a resolved IP address as we don't know which of @@ -154,7 +151,7 @@ void Socket::connect(const SocketAddress& addr) const createSocket(addr); - const int& socket = impl->fd; + const int& socket = fd; // TODO the correct thing to do here is loop on failure until you've used all the returned addresses if ((::connect(socket, getAddrInfo(addr).ai_addr, getAddrInfo(addr).ai_addrlen) < 0) && (errno != EINPROGRESS)) { @@ -165,11 +162,6 @@ void Socket::connect(const SocketAddress& addr) const // remote port (which is unoccupied) as the port to bind the local // end of the socket, resulting in a "circular" connection. // - // This seems like something the OS should prevent but I have - // confirmed that sporadic hangs in - // cluster_tests.LongTests.test_failover on RHEL5 are caused by - // such a circular connection. - // // Raise an error if we see such a connection, since we know there is // no listener on the peer address. // @@ -179,26 +171,25 @@ void Socket::connect(const SocketAddress& addr) const } } +void BSDSocket::finishConnect(const SocketAddress&) const +{ +} + void -Socket::close() const +BSDSocket::close() const { - int& socket = impl->fd; + int& socket = fd; if (socket == -1) return; if (::close(socket) < 0) throw QPID_POSIX_ERROR(errno); socket = -1; + *handle = IOHandle(); } -int Socket::listen(const std::string& host, const std::string& port, int backlog) const -{ - SocketAddress sa(host, port); - return listen(sa, backlog); -} - -int Socket::listen(const SocketAddress& sa, int backlog) const +int BSDSocket::listen(const SocketAddress& sa, int backlog) const { createSocket(sa); - const int& socket = impl->fd; + const int& socket = fd; int yes=1; QPID_POSIX_CHECK(::setsockopt(socket,SOL_SOCKET,SO_REUSEADDR,&yes,sizeof(yes))); @@ -210,11 +201,11 @@ int Socket::listen(const SocketAddress& sa, int backlog) const return getLocalPort(socket); } -Socket* Socket::accept() const +Socket* BSDSocket::accept() const { - int afd = ::accept(impl->fd, 0, 0); + int afd = ::accept(fd, 0, 0); if ( afd >= 0) { - Socket* s = new Socket(new IOHandlePrivate(afd)); + BSDSocket* s = new BSDSocket(afd); s->localname = localname; return s; } @@ -223,41 +214,51 @@ Socket* Socket::accept() const else throw QPID_POSIX_ERROR(errno); } -int Socket::read(void *buf, size_t count) const +int BSDSocket::read(void *buf, size_t count) const { - return ::read(impl->fd, buf, count); + return ::read(fd, buf, count); } -int Socket::write(const void *buf, size_t count) const +int BSDSocket::write(const void *buf, size_t count) const { - return ::write(impl->fd, buf, count); + return ::write(fd, buf, count); } -std::string Socket::getPeerAddress() const +std::string BSDSocket::getPeerAddress() const { if (peername.empty()) { - peername = getName(impl->fd, false); + peername = getName(fd, false); } return peername; } -std::string Socket::getLocalAddress() const +std::string BSDSocket::getLocalAddress() const { if (localname.empty()) { - localname = getName(impl->fd, true); + localname = getName(fd, true); } return localname; } -int Socket::getError() const +int BSDSocket::getError() const { int result; socklen_t rSize = sizeof (result); - if (::getsockopt(impl->fd, SOL_SOCKET, SO_ERROR, &result, &rSize) < 0) + if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &result, &rSize) < 0) throw QPID_POSIX_ERROR(errno); return result; } +int BSDSocket::getKeyLen() const +{ + return 0; +} + +std::string BSDSocket::getClientAuthId() const +{ + return std::string(); +} + }} // namespace qpid::sys diff --git a/cpp/src/qpid/sys/posix/BSDSocket.h b/cpp/src/qpid/sys/posix/BSDSocket.h new file mode 100644 index 0000000000..862d36c1b9 --- /dev/null +++ b/cpp/src/qpid/sys/posix/BSDSocket.h @@ -0,0 +1,113 @@ +#ifndef QPID_SYS_BSDSOCKET_H +#define QPID_SYS_BSDSOCKET_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 "qpid/sys/Socket.h" +#include "qpid/sys/IntegerTypes.h" +#include "qpid/CommonImportExport.h" +#include <string> + +#include <boost/scoped_ptr.hpp> + +namespace qpid { +namespace sys { + +class Duration; +class IOHandle; +class SocketAddress; + +namespace ssl { +class SslMuxSocket; +} + +class QPID_COMMON_CLASS_EXTERN BSDSocket : public Socket +{ +public: + /** Create a socket wrapper for descriptor. */ + QPID_COMMON_EXTERN BSDSocket(); + QPID_COMMON_EXTERN ~BSDSocket(); + + QPID_COMMON_EXTERN operator const IOHandle&() const; + + /** Set socket non blocking */ + QPID_COMMON_EXTERN virtual void setNonblocking() const; + + QPID_COMMON_EXTERN virtual void setTcpNoDelay() const; + + QPID_COMMON_EXTERN virtual void connect(const SocketAddress&) const; + QPID_COMMON_EXTERN virtual void finishConnect(const SocketAddress&) const; + + QPID_COMMON_EXTERN virtual void close() const; + + /** Bind to a port and start listening. + *@return The bound port number + */ + QPID_COMMON_EXTERN virtual int listen(const SocketAddress&, int backlog = 10) const; + + /** + * Returns an address (host and port) for the remote end of the + * socket + */ + QPID_COMMON_EXTERN std::string getPeerAddress() const; + /** + * Returns an address (host and port) for the local end of the + * socket + */ + QPID_COMMON_EXTERN std::string getLocalAddress() const; + + /** + * Returns the error code stored in the socket. This may be used + * to determine the result of a non-blocking connect. + */ + QPID_COMMON_EXTERN int getError() const; + + /** Accept a connection from a socket that is already listening + * and has an incoming connection + */ + QPID_COMMON_EXTERN virtual Socket* accept() const; + + // TODO The following are raw operations, maybe they need better wrapping? + QPID_COMMON_EXTERN virtual int read(void *buf, size_t count) const; + QPID_COMMON_EXTERN virtual int write(const void *buf, size_t count) const; + + QPID_COMMON_EXTERN int getKeyLen() const; + QPID_COMMON_EXTERN std::string getClientAuthId() const; + +protected: + /** Create socket */ + void createSocket(const SocketAddress&) const; + + mutable int fd; + mutable boost::scoped_ptr<IOHandle> handle; + mutable std::string localname; + mutable std::string peername; + mutable bool nonblocking; + mutable bool nodelay; + + /** Construct socket with existing handle */ + BSDSocket(int fd); + friend class qpid::sys::ssl::SslMuxSocket; // Needed for this constructor +}; + +}} +#endif /*!QPID_SYS_BSDSOCKET_H*/ diff --git a/cpp/src/qpid/sys/posix/FileSysDir.cpp b/cpp/src/qpid/sys/posix/FileSysDir.cpp index 22dc487e74..cec580164d 100755 --- a/cpp/src/qpid/sys/posix/FileSysDir.cpp +++ b/cpp/src/qpid/sys/posix/FileSysDir.cpp @@ -18,6 +18,7 @@ #include "qpid/sys/FileSysDir.h" #include "qpid/sys/StrError.h" +#include "qpid/log/Statement.h" #include "qpid/Exception.h" #include <sys/types.h> @@ -25,6 +26,8 @@ #include <fcntl.h> #include <cerrno> #include <unistd.h> +#include <dirent.h> +#include <stdlib.h> namespace qpid { namespace sys { @@ -51,4 +54,27 @@ void FileSysDir::mkdir(void) throw Exception ("Can't create directory: " + dirPath); } +void FileSysDir::forEachFile(Callback cb) const { + + ::dirent** namelist; + + int n = scandir(dirPath.c_str(), &namelist, 0, alphasort); + if (n == -1) throw Exception (strError(errno) + ": Can't scan directory: " + dirPath); + + for (int i = 0; i<n; ++i) { + std::string fullpath = dirPath + "/" + namelist[i]->d_name; + // Filter out non files/stat problems etc. + struct ::stat s; + // Can't throw here without leaking memory, so just do nothing with + // entries for which stat() fails. + if (!::stat(fullpath.c_str(), &s)) { + if (S_ISREG(s.st_mode)) { + cb(fullpath); + } + } + ::free(namelist[i]); + } + ::free(namelist); +} + }} // namespace qpid::sys diff --git a/cpp/src/qpid/sys/posix/IOHandle.cpp b/cpp/src/qpid/sys/posix/IOHandle.cpp index 9c049ee1de..d3f502a63c 100644 --- a/cpp/src/qpid/sys/posix/IOHandle.cpp +++ b/cpp/src/qpid/sys/posix/IOHandle.cpp @@ -19,26 +19,11 @@ * */ -#include "qpid/sys/IOHandle.h" - #include "qpid/sys/posix/PrivatePosix.h" namespace qpid { namespace sys { -int toFd(const IOHandlePrivate* h) -{ - return h->fd; -} - NullIOHandle DummyIOHandle; -IOHandle::IOHandle(IOHandlePrivate* h) : - impl(h) -{} - -IOHandle::~IOHandle() { - delete impl; -} - }} // namespace qpid::sys diff --git a/cpp/src/qpid/sys/posix/PollableCondition.cpp b/cpp/src/qpid/sys/posix/PollableCondition.cpp index abff8a5be8..aa129faf20 100644 --- a/cpp/src/qpid/sys/posix/PollableCondition.cpp +++ b/cpp/src/qpid/sys/posix/PollableCondition.cpp @@ -21,7 +21,6 @@ #include "qpid/sys/PollableCondition.h" #include "qpid/sys/DispatchHandle.h" -#include "qpid/sys/IOHandle.h" #include "qpid/sys/posix/PrivatePosix.h" #include "qpid/Exception.h" @@ -58,14 +57,14 @@ PollableConditionPrivate::PollableConditionPrivate( const sys::PollableCondition::Callback& cb, sys::PollableCondition& parent, const boost::shared_ptr<sys::Poller>& poller -) : IOHandle(new sys::IOHandlePrivate), cb(cb), parent(parent) +) : cb(cb), parent(parent) { int fds[2]; if (::pipe(fds) == -1) throw ErrnoException(QPID_MSG("Can't create PollableCondition")); - impl->fd = fds[0]; + fd = fds[0]; writeFd = fds[1]; - if (::fcntl(impl->fd, F_SETFL, O_NONBLOCK) == -1) + if (::fcntl(fd, F_SETFL, O_NONBLOCK) == -1) throw ErrnoException(QPID_MSG("Can't create PollableCondition")); if (::fcntl(writeFd, F_SETFL, O_NONBLOCK) == -1) throw ErrnoException(QPID_MSG("Can't create PollableCondition")); diff --git a/cpp/src/qpid/sys/posix/PosixPoller.cpp b/cpp/src/qpid/sys/posix/PosixPoller.cpp index eb0c3384d1..ae839b2e20 100644 --- a/cpp/src/qpid/sys/posix/PosixPoller.cpp +++ b/cpp/src/qpid/sys/posix/PosixPoller.cpp @@ -88,12 +88,12 @@ class PollerHandlePrivate { }; short events; - const IOHandlePrivate* ioHandle; + const IOHandle* ioHandle; PollerHandle* pollerHandle; FDStat stat; Mutex lock; - PollerHandlePrivate(const IOHandlePrivate* h, PollerHandle* p) : + PollerHandlePrivate(const IOHandle* h, PollerHandle* p) : events(0), ioHandle(h), pollerHandle(p), @@ -101,7 +101,7 @@ class PollerHandlePrivate { } int fd() const { - return toFd(ioHandle); + return ioHandle->fd; } bool isActive() const { @@ -162,7 +162,7 @@ class PollerHandlePrivate { }; PollerHandle::PollerHandle(const IOHandle& h) : - impl(new PollerHandlePrivate(h.impl, this)) + impl(new PollerHandlePrivate(&h, this)) {} PollerHandle::~PollerHandle() { diff --git a/cpp/src/qpid/sys/posix/SocketAddress.cpp b/cpp/src/qpid/sys/posix/SocketAddress.cpp index 344bd28669..cd23442226 100644 --- a/cpp/src/qpid/sys/posix/SocketAddress.cpp +++ b/cpp/src/qpid/sys/posix/SocketAddress.cpp @@ -102,6 +102,11 @@ std::string SocketAddress::asString(bool numeric) const return asString(ai.ai_addr, ai.ai_addrlen); } +std::string SocketAddress::getHost() const +{ + return host; +} + bool SocketAddress::nextAddress() { bool r = currentAddrInfo->ai_next != 0; if (r) diff --git a/cpp/src/qpid/sys/posix/SystemInfo.cpp b/cpp/src/qpid/sys/posix/SystemInfo.cpp index cfd2c64aee..ea7f521f2b 100755 --- a/cpp/src/qpid/sys/posix/SystemInfo.cpp +++ b/cpp/src/qpid/sys/posix/SystemInfo.cpp @@ -21,7 +21,6 @@ #include "qpid/log/Statement.h" #include "qpid/sys/SystemInfo.h" #include "qpid/sys/posix/check.h" -#include <set> #include <arpa/inet.h> #include <sys/ioctl.h> #include <sys/utsname.h> @@ -33,6 +32,7 @@ #include <iostream> #include <fstream> #include <sstream> +#include <map> #include <netdb.h> #include <string.h> @@ -77,84 +77,70 @@ inline bool isLoopback(const ::sockaddr* addr) { } } -void SystemInfo::getLocalIpAddresses (uint16_t port, - std::vector<Address> &addrList) { - ::ifaddrs* ifaddr = 0; - QPID_POSIX_CHECK(::getifaddrs(&ifaddr)); - for (::ifaddrs* ifap = ifaddr; ifap != 0; ifap = ifap->ifa_next) { - if (ifap->ifa_addr == 0) continue; - if (isLoopback(ifap->ifa_addr)) continue; - int family = ifap->ifa_addr->sa_family; - switch (family) { - case AF_INET6: { - // Ignore link local addresses as: - // * The scope id is illegal in URL syntax - // * Clients won't be able to use a link local address - // without adding their own (potentially different) scope id - sockaddr_in6* sa6 = (sockaddr_in6*)((void*)ifap->ifa_addr); - if (IN6_IS_ADDR_LINKLOCAL(&sa6->sin6_addr)) break; - // Fallthrough - } - case AF_INET: { - char dispName[NI_MAXHOST]; - int rc = ::getnameinfo( - ifap->ifa_addr, - (family == AF_INET) - ? sizeof(struct sockaddr_in) - : sizeof(struct sockaddr_in6), - dispName, sizeof(dispName), - 0, 0, NI_NUMERICHOST); - if (rc != 0) { - throw QPID_POSIX_ERROR(rc); - } - string addr(dispName); - addrList.push_back(Address(TCP, addr, port)); - break; - } - default: - continue; +namespace { + inline socklen_t sa_len(::sockaddr* sa) + { + switch (sa->sa_family) { + case AF_INET: + return sizeof(struct sockaddr_in); + case AF_INET6: + return sizeof(struct sockaddr_in6); + default: + return sizeof(struct sockaddr_storage); } } - ::freeifaddrs(ifaddr); - if (addrList.empty()) { - addrList.push_back(Address(TCP, LOOPBACK, port)); + inline bool isInetOrInet6(::sockaddr* sa) { + switch (sa->sa_family) { + case AF_INET: + case AF_INET6: + return true; + default: + return false; + } + } + typedef std::map<std::string, std::vector<std::string> > InterfaceInfo; + std::map<std::string, std::vector<std::string> > cachedInterfaces; + + void cacheInterfaceInfo() { + // Get interface info + ::ifaddrs* interfaceInfo; + QPID_POSIX_CHECK( ::getifaddrs(&interfaceInfo) ); + + char name[NI_MAXHOST]; + for (::ifaddrs* info = interfaceInfo; info != 0; info = info->ifa_next) { + + // Only use IPv4/IPv6 interfaces + if (!isInetOrInet6(info->ifa_addr)) continue; + + int rc=::getnameinfo(info->ifa_addr, sa_len(info->ifa_addr), + name, sizeof(name), 0, 0, + NI_NUMERICHOST); + if (rc >= 0) { + std::string address(name); + cachedInterfaces[info->ifa_name].push_back(address); + } else { + throw qpid::Exception(QPID_MSG(gai_strerror(rc))); + } + } + ::freeifaddrs(interfaceInfo); } } -namespace { -struct AddrInfo { - struct addrinfo* ptr; - AddrInfo(const std::string& host) : ptr(0) { - ::addrinfo hints; - ::memset(&hints, 0, sizeof(hints)); - hints.ai_family = AF_UNSPEC; // Allow both IPv4 and IPv6 - if (::getaddrinfo(host.c_str(), NULL, &hints, &ptr) != 0) - ptr = 0; - } - ~AddrInfo() { if (ptr) ::freeaddrinfo(ptr); } -}; +bool SystemInfo::getInterfaceAddresses(const std::string& interface, std::vector<std::string>& addresses) { + if ( cachedInterfaces.empty() ) cacheInterfaceInfo(); + InterfaceInfo::iterator i = cachedInterfaces.find(interface); + if ( i==cachedInterfaces.end() ) return false; + std::copy(i->second.begin(), i->second.end(), std::back_inserter(addresses)); + return true; } -bool SystemInfo::isLocalHost(const std::string& host) { - std::vector<Address> myAddrs; - getLocalIpAddresses(0, myAddrs); - std::set<string> localHosts; - for (std::vector<Address>::const_iterator i = myAddrs.begin(); i != myAddrs.end(); ++i) - localHosts.insert(i->host); - // Resolve host - AddrInfo ai(host); - if (!ai.ptr) return false; - for (struct addrinfo *res = ai.ptr; res != NULL; res = res->ai_next) { - if (isLoopback(res->ai_addr)) return true; - // Get string form of IP addr - char addr[NI_MAXHOST] = ""; - int error = ::getnameinfo(res->ai_addr, res->ai_addrlen, addr, NI_MAXHOST, NULL, 0, - NI_NUMERICHOST | NI_NUMERICSERV); - if (error) return false; - if (localHosts.find(addr) != localHosts.end()) return true; +void SystemInfo::getInterfaceNames(std::vector<std::string>& names ) { + if ( cachedInterfaces.empty() ) cacheInterfaceInfo(); + + for (InterfaceInfo::const_iterator i = cachedInterfaces.begin(); i!=cachedInterfaces.end(); ++i) { + names.push_back(i->first); } - return false; } void SystemInfo::getSystemId (std::string &osName, @@ -205,4 +191,11 @@ string SystemInfo::getProcessName() return value; } +// Always true. Only Windows has exception cases. +bool SystemInfo::threadSafeShutdown() +{ + return true; +} + + }} // namespace qpid::sys diff --git a/cpp/src/qpid/sys/rdma/rdma_wrap.cpp b/cpp/src/qpid/sys/rdma/rdma_wrap.cpp index efe454c5be..889ee9ff75 100644 --- a/cpp/src/qpid/sys/rdma/rdma_wrap.cpp +++ b/cpp/src/qpid/sys/rdma/rdma_wrap.cpp @@ -105,7 +105,7 @@ namespace Rdma { } QueuePair::QueuePair(boost::shared_ptr< ::rdma_cm_id > i) : - qpid::sys::IOHandle(new qpid::sys::IOHandlePrivate), + handle(new qpid::sys::IOHandle), pd(allocPd(i->verbs)), cchannel(mkCChannel(i->verbs)), scq(mkCq(i->verbs, DEFAULT_CQ_ENTRIES, 0, cchannel.get())), @@ -113,7 +113,7 @@ namespace Rdma { outstandingSendEvents(0), outstandingRecvEvents(0) { - impl->fd = cchannel->fd; + handle->fd = cchannel->fd; // Set cq context to this QueuePair object so we can find // ourselves again @@ -163,6 +163,11 @@ namespace Rdma { // The buffers vectors automatically deletes all the buffers we've allocated } + QueuePair::operator qpid::sys::IOHandle&() const + { + return *handle; + } + // Create buffers to use for writing void QueuePair::createSendBuffers(int sendBufferCount, int bufferSize, int reserved) { @@ -359,11 +364,11 @@ namespace Rdma { // Wrap the passed in rdma_cm_id with a Connection // this basically happens only on connection request Connection::Connection(::rdma_cm_id* i) : - qpid::sys::IOHandle(new qpid::sys::IOHandlePrivate), + handle(new qpid::sys::IOHandle), id(mkId(i)), context(0) { - impl->fd = id->channel->fd; + handle->fd = id->channel->fd; // Just overwrite the previous context as it will // have come from the listening connection @@ -372,12 +377,12 @@ namespace Rdma { } Connection::Connection() : - qpid::sys::IOHandle(new qpid::sys::IOHandlePrivate), + handle(new qpid::sys::IOHandle), channel(mkEChannel()), id(mkId(channel.get(), this, RDMA_PS_TCP)), context(0) { - impl->fd = channel->fd; + handle->fd = channel->fd; } Connection::~Connection() { @@ -385,6 +390,11 @@ namespace Rdma { id->context = 0; } + Connection::operator qpid::sys::IOHandle&() const + { + return *handle; + } + void Connection::ensureQueuePair() { assert(id.get()); diff --git a/cpp/src/qpid/sys/rdma/rdma_wrap.h b/cpp/src/qpid/sys/rdma/rdma_wrap.h index 8e3429027b..5f84793a5b 100644 --- a/cpp/src/qpid/sys/rdma/rdma_wrap.h +++ b/cpp/src/qpid/sys/rdma/rdma_wrap.h @@ -28,6 +28,7 @@ #include "qpid/sys/Mutex.h" #include <boost/shared_ptr.hpp> +#include <boost/scoped_ptr.hpp> #include <boost/intrusive_ptr.hpp> #include <boost/ptr_container/ptr_deque.hpp> @@ -116,9 +117,10 @@ namespace Rdma { // Wrapper for a queue pair - this has the functionality for // putting buffers on the receive queue and for sending buffers // to the other end of the connection. - class QueuePair : public qpid::sys::IOHandle, public qpid::RefCounted { + class QueuePair : public qpid::RefCounted { friend class Connection; + boost::scoped_ptr< qpid::sys::IOHandle > handle; boost::shared_ptr< ::ibv_pd > pd; boost::shared_ptr< ::ibv_mr > smr; boost::shared_ptr< ::ibv_mr > rmr; @@ -139,6 +141,8 @@ namespace Rdma { public: typedef boost::intrusive_ptr<QueuePair> intrusive_ptr; + operator qpid::sys::IOHandle&() const; + // Create a buffers to use for writing void createSendBuffers(int sendBufferCount, int dataSize, int headerSize); @@ -195,7 +199,8 @@ namespace Rdma { // registered buffers can't be shared between different connections // (this can only happen between connections on the same controller in any case, // so needs careful management if used) - class Connection : public qpid::sys::IOHandle, public qpid::RefCounted { + class Connection : public qpid::RefCounted { + boost::scoped_ptr< qpid::sys::IOHandle > handle; boost::shared_ptr< ::rdma_event_channel > channel; boost::shared_ptr< ::rdma_cm_id > id; QueuePair::intrusive_ptr qp; @@ -216,6 +221,8 @@ namespace Rdma { public: typedef boost::intrusive_ptr<Connection> intrusive_ptr; + operator qpid::sys::IOHandle&() const; + static intrusive_ptr make(); static intrusive_ptr find(::rdma_cm_id* i); diff --git a/cpp/src/qpid/sys/solaris/SystemInfo.cpp b/cpp/src/qpid/sys/solaris/SystemInfo.cpp index e5856f55e6..0e754e048b 100755 --- a/cpp/src/qpid/sys/solaris/SystemInfo.cpp +++ b/cpp/src/qpid/sys/solaris/SystemInfo.cpp @@ -60,31 +60,6 @@ bool SystemInfo::getLocalHostname(Address &address) { static const string LOCALHOST("127.0.0.1"); static const string TCP("tcp"); -void SystemInfo::getLocalIpAddresses(uint16_t port, - std::vector<Address> &addrList) { - int s = socket(PF_INET, SOCK_STREAM, 0); - for (int i=1;;i++) { - struct lifreq ifr; - ifr.lifr_index = i; - if (::ioctl(s, SIOCGIFADDR, &ifr) < 0) { - break; - } - struct sockaddr *sa = static_cast<struct sockaddr *>((void *) &ifr.lifr_addr); - if (sa->sa_family != AF_INET) { - // TODO: Url parsing currently can't cope with IPv6 addresses, defer for now - break; - } - struct sockaddr_in *sin = static_cast<struct sockaddr_in *>((void *)sa); - std::string addr(inet_ntoa(sin->sin_addr)); - if (addr != LOCALHOST) - addrList.push_back(Address(TCP, addr, port)); - } - if (addrList.empty()) { - addrList.push_back(Address(TCP, LOCALHOST, port)); - } - close (s); -} - void SystemInfo::getSystemId(std::string &osName, std::string &nodeName, std::string &release, @@ -126,4 +101,10 @@ string SystemInfo::getProcessName() return value; } +// Always true. Only Windows has exception cases. +bool SystemInfo::threadSafeShutdown() +{ + return true; +} + }} // namespace qpid::sys diff --git a/cpp/src/qpid/sys/ssl/SslHandler.cpp b/cpp/src/qpid/sys/ssl/SslHandler.cpp deleted file mode 100644 index eeb8c26a76..0000000000 --- a/cpp/src/qpid/sys/ssl/SslHandler.cpp +++ /dev/null @@ -1,217 +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 "qpid/sys/ssl/SslHandler.h" -#include "qpid/sys/ssl/SslIo.h" -#include "qpid/sys/ssl/SslSocket.h" -#include "qpid/sys/Timer.h" -#include "qpid/framing/AMQP_HighestVersion.h" -#include "qpid/framing/ProtocolInitiation.h" -#include "qpid/log/Statement.h" - -#include <boost/bind.hpp> - -namespace qpid { -namespace sys { -namespace ssl { - - -struct ProtocolTimeoutTask : public sys::TimerTask { - SslHandler& handler; - std::string id; - - ProtocolTimeoutTask(const std::string& i, const Duration& timeout, SslHandler& h) : - TimerTask(timeout, "ProtocolTimeout"), - handler(h), - id(i) - {} - - void fire() { - // If this fires it means that we didn't negotiate the connection in the timeout period - // Schedule closing the connection for the io thread - QPID_LOG(error, "Connection " << id << " No protocol received closing"); - handler.abort(); - } -}; - -SslHandler::SslHandler(std::string id, ConnectionCodec::Factory* f, bool _nodict) : - identifier(id), - aio(0), - factory(f), - codec(0), - readError(false), - isClient(false), - nodict(_nodict) -{} - -SslHandler::~SslHandler() { - if (codec) - codec->closed(); - if (timeoutTimerTask) - timeoutTimerTask->cancel(); - delete codec; -} - -void SslHandler::init(SslIO* a, Timer& timer, uint32_t maxTime) { - aio = a; - - // Start timer for this connection - timeoutTimerTask = new ProtocolTimeoutTask(identifier, maxTime*TIME_MSEC, *this); - timer.add(timeoutTimerTask); - - // Give connection some buffers to use - aio->createBuffers(); -} - -void SslHandler::write(const framing::ProtocolInitiation& data) -{ - QPID_LOG(debug, "SENT [" << identifier << "]: INIT(" << data << ")"); - SslIO::BufferBase* buff = aio->getQueuedBuffer(); - assert(buff); - framing::Buffer out(buff->bytes, buff->byteCount); - data.encode(out); - buff->dataCount = data.encodedSize(); - aio->queueWrite(buff); -} - -void SslHandler::abort() { - // Don't disconnect if we're already disconnecting - if (!readError) { - aio->requestCallback(boost::bind(&SslHandler::eof, this, _1)); - } -} -void SslHandler::activateOutput() { - aio->notifyPendingWrite(); -} - -void SslHandler::giveReadCredit(int32_t) { - // FIXME aconway 2008-12-05: not yet implemented. -} - -// Input side -void SslHandler::readbuff(SslIO& , SslIO::BufferBase* buff) { - if (readError) { - return; - } - size_t decoded = 0; - if (codec) { // Already initiated - try { - decoded = codec->decode(buff->bytes+buff->dataStart, buff->dataCount); - }catch(const std::exception& e){ - QPID_LOG(error, e.what()); - readError = true; - aio->queueWriteClose(); - } - }else{ - framing::Buffer in(buff->bytes+buff->dataStart, buff->dataCount); - framing::ProtocolInitiation protocolInit; - if (protocolInit.decode(in)) { - // We've just got the protocol negotiation so we can cancel the timeout for that - timeoutTimerTask->cancel(); - - decoded = in.getPosition(); - QPID_LOG(debug, "RECV [" << identifier << "]: INIT(" << protocolInit << ")"); - try { - codec = factory->create(protocolInit.getVersion(), *this, identifier, getSecuritySettings(aio)); - if (!codec) { - //TODO: may still want to revise this... - //send valid version header & close connection. - write(framing::ProtocolInitiation(framing::highestProtocolVersion)); - readError = true; - aio->queueWriteClose(); - } - } catch (const std::exception& e) { - QPID_LOG(error, e.what()); - readError = true; - aio->queueWriteClose(); - } - } - } - // TODO: unreading needs to go away, and when we can cope - // with multiple sub-buffers in the general buffer scheme, it will - if (decoded != size_t(buff->dataCount)) { - // Adjust buffer for used bytes and then "unread them" - buff->dataStart += decoded; - buff->dataCount -= decoded; - aio->unread(buff); - } else { - // Give whole buffer back to aio subsystem - aio->queueReadBuffer(buff); - } -} - -void SslHandler::eof(SslIO&) { - QPID_LOG(debug, "DISCONNECTED [" << identifier << "]"); - if (codec) codec->closed(); - aio->queueWriteClose(); -} - -void SslHandler::closedSocket(SslIO&, const SslSocket& s) { - // If we closed with data still to send log a warning - if (!aio->writeQueueEmpty()) { - QPID_LOG(warning, "CLOSING [" << identifier << "] unsent data (probably due to client disconnect)"); - } - delete &s; - aio->queueForDeletion(); - delete this; -} - -void SslHandler::disconnect(SslIO& a) { - // treat the same as eof - eof(a); -} - -// Notifications -void SslHandler::nobuffs(SslIO&) { -} - -void SslHandler::idle(SslIO&){ - if (isClient && codec == 0) { - codec = factory->create(*this, identifier, getSecuritySettings(aio)); - write(framing::ProtocolInitiation(codec->getVersion())); - // We've just sent the protocol negotiation so we can cancel the timeout for that - // This is not ideal, because we've not received anything yet, but heartbeats will - // be active soon - timeoutTimerTask->cancel(); - return; - } - if (codec == 0) return; - if (!codec->canEncode()) { - return; - } - SslIO::BufferBase* buff = aio->getQueuedBuffer(); - if (buff) { - size_t encoded=codec->encode(buff->bytes, buff->byteCount); - buff->dataCount = encoded; - aio->queueWrite(buff); - } - if (codec->isClosed()) - aio->queueWriteClose(); -} - -SecuritySettings SslHandler::getSecuritySettings(SslIO* aio) -{ - SecuritySettings settings = aio->getSecuritySettings(); - settings.nodict = nodict; - return settings; -} - - -}}} // namespace qpid::sys::ssl diff --git a/cpp/src/qpid/sys/ssl/SslHandler.h b/cpp/src/qpid/sys/ssl/SslHandler.h deleted file mode 100644 index 14814b0281..0000000000 --- a/cpp/src/qpid/sys/ssl/SslHandler.h +++ /dev/null @@ -1,85 +0,0 @@ -#ifndef QPID_SYS_SSL_SSLHANDLER_H -#define QPID_SYS_SSL_SSLHANDLER_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 "qpid/sys/ConnectionCodec.h" -#include "qpid/sys/OutputControl.h" - -#include <boost/intrusive_ptr.hpp> - -namespace qpid { - -namespace framing { - class ProtocolInitiation; -} - -namespace sys { - -class Timer; -class TimerTask; - -namespace ssl { - -class SslIO; -struct SslIOBufferBase; -class SslSocket; - -class SslHandler : public OutputControl { - std::string identifier; - SslIO* aio; - ConnectionCodec::Factory* factory; - ConnectionCodec* codec; - bool readError; - bool isClient; - bool nodict; - boost::intrusive_ptr<sys::TimerTask> timeoutTimerTask; - - void write(const framing::ProtocolInitiation&); - qpid::sys::SecuritySettings getSecuritySettings(SslIO* aio); - - public: - SslHandler(std::string id, ConnectionCodec::Factory* f, bool nodict); - ~SslHandler(); - void init(SslIO* a, Timer& timer, uint32_t maxTime); - - void setClient() { isClient = true; } - - // Output side - void abort(); - void activateOutput(); - void giveReadCredit(int32_t); - - // Input side - void readbuff(SslIO& aio, SslIOBufferBase* buff); - void eof(SslIO& aio); - void disconnect(SslIO& aio); - - // Notifications - void nobuffs(SslIO& aio); - void idle(SslIO& aio); - void closedSocket(SslIO& aio, const SslSocket& s); -}; - -}}} // namespace qpid::sys::ssl - -#endif /*!QPID_SYS_SSL_SSLHANDLER_H*/ diff --git a/cpp/src/qpid/sys/ssl/SslIo.cpp b/cpp/src/qpid/sys/ssl/SslIo.cpp deleted file mode 100644 index bbfb703170..0000000000 --- a/cpp/src/qpid/sys/ssl/SslIo.cpp +++ /dev/null @@ -1,470 +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 "qpid/sys/ssl/SslIo.h" -#include "qpid/sys/ssl/SslSocket.h" -#include "qpid/sys/ssl/check.h" - -#include "qpid/sys/Time.h" -#include "qpid/sys/posix/check.h" -#include "qpid/log/Statement.h" - -// TODO The basic algorithm here is not really POSIX specific and with a bit more abstraction -// could (should) be promoted to be platform portable -#include <unistd.h> -#include <sys/socket.h> -#include <signal.h> -#include <errno.h> -#include <string.h> - -#include <boost/bind.hpp> - -namespace qpid { -namespace sys { -namespace ssl { - -namespace { - -/* - * Make *process* not generate SIGPIPE when writing to closed - * pipe/socket (necessary as default action is to terminate process) - */ -void ignoreSigpipe() { - ::signal(SIGPIPE, SIG_IGN); -} - -/* - * We keep per thread state to avoid locking overhead. The assumption is that - * on average all the connections are serviced by all the threads so the state - * recorded in each thread is about the same. If this turns out not to be the - * case we could rebalance the info occasionally. - */ -__thread int threadReadTotal = 0; -__thread int threadReadCount = 0; -__thread int threadWriteTotal = 0; -__thread int threadWriteCount = 0; -__thread int64_t threadMaxIoTimeNs = 2 * 1000000; // start at 2ms -} - -/* - * Asynch Acceptor - */ - -template <class T> -SslAcceptorTmpl<T>::SslAcceptorTmpl(const T& s, Callback callback) : - acceptedCallback(callback), - handle(s, boost::bind(&SslAcceptorTmpl<T>::readable, this, _1), 0, 0), - socket(s) { - - s.setNonblocking(); - ignoreSigpipe(); -} - -template <class T> -SslAcceptorTmpl<T>::~SslAcceptorTmpl() -{ - handle.stopWatch(); -} - -template <class T> -void SslAcceptorTmpl<T>::start(Poller::shared_ptr poller) { - handle.startWatch(poller); -} - -/* - * We keep on accepting as long as there is something to accept - */ -template <class T> -void SslAcceptorTmpl<T>::readable(DispatchHandle& h) { - Socket* s; - do { - errno = 0; - // TODO: Currently we ignore the peers address, perhaps we should - // log it or use it for connection acceptance. - try { - s = socket.accept(); - if (s) { - acceptedCallback(*s); - } else { - break; - } - } catch (const std::exception& e) { - QPID_LOG(error, "Could not accept socket: " << e.what()); - } - } while (true); - - h.rewatch(); -} - -// Explicitly instantiate the templates we need -template class SslAcceptorTmpl<SslSocket>; -template class SslAcceptorTmpl<SslMuxSocket>; - -/* - * Asynch Connector - */ - -SslConnector::SslConnector(const SslSocket& s, - Poller::shared_ptr poller, - std::string hostname, - std::string port, - ConnectedCallback connCb, - FailedCallback failCb) : - DispatchHandle(s, - 0, - boost::bind(&SslConnector::connComplete, this, _1), - boost::bind(&SslConnector::connComplete, this, _1)), - connCallback(connCb), - failCallback(failCb), - socket(s) -{ - //TODO: would be better for connect to be performed on a - //non-blocking socket, but that doesn't work at present so connect - //blocks until complete - try { - socket.connect(hostname, port); - socket.setNonblocking(); - startWatch(poller); - } catch(std::exception& e) { - failure(-1, std::string(e.what())); - } -} - -void SslConnector::connComplete(DispatchHandle& h) -{ - int errCode = socket.getError(); - - h.stopWatch(); - if (errCode == 0) { - connCallback(socket); - DispatchHandle::doDelete(); - } else { - // TODO: This need to be fixed as strerror isn't thread safe - failure(errCode, std::string(::strerror(errCode))); - } -} - -void SslConnector::failure(int errCode, std::string message) -{ - if (failCallback) - failCallback(errCode, message); - - socket.close(); - delete &socket; - - DispatchHandle::doDelete(); -} - -/* - * Asynch reader/writer - */ -SslIO::SslIO(const SslSocket& s, - ReadCallback rCb, EofCallback eofCb, DisconnectCallback disCb, - ClosedCallback cCb, BuffersEmptyCallback eCb, IdleCallback iCb) : - - DispatchHandle(s, - boost::bind(&SslIO::readable, this, _1), - boost::bind(&SslIO::writeable, this, _1), - boost::bind(&SslIO::disconnected, this, _1)), - readCallback(rCb), - eofCallback(eofCb), - disCallback(disCb), - closedCallback(cCb), - emptyCallback(eCb), - idleCallback(iCb), - socket(s), - queuedClose(false), - writePending(false) { - - s.setNonblocking(); -} - -SslIO::~SslIO() { -} - -void SslIO::queueForDeletion() { - DispatchHandle::doDelete(); -} - -void SslIO::start(Poller::shared_ptr poller) { - DispatchHandle::startWatch(poller); -} - -void SslIO::createBuffers(uint32_t size) { - // Allocate all the buffer memory at once - bufferMemory.reset(new char[size*BufferCount]); - - // Create the Buffer structs in a vector - // And push into the buffer queue - buffers.reserve(BufferCount); - for (uint32_t i = 0; i < BufferCount; i++) { - buffers.push_back(BufferBase(&bufferMemory[i*size], size)); - queueReadBuffer(&buffers[i]); - } -} - -void SslIO::queueReadBuffer(BufferBase* buff) { - assert(buff); - buff->dataStart = 0; - buff->dataCount = 0; - bufferQueue.push_back(buff); - DispatchHandle::rewatchRead(); -} - -void SslIO::unread(BufferBase* buff) { - assert(buff); - if (buff->dataStart != 0) { - memmove(buff->bytes, buff->bytes+buff->dataStart, buff->dataCount); - buff->dataStart = 0; - } - bufferQueue.push_front(buff); - DispatchHandle::rewatchRead(); -} - -void SslIO::queueWrite(BufferBase* buff) { - assert(buff); - // If we've already closed the socket then throw the write away - if (queuedClose) { - bufferQueue.push_front(buff); - return; - } else { - writeQueue.push_front(buff); - } - writePending = false; - DispatchHandle::rewatchWrite(); -} - -void SslIO::notifyPendingWrite() { - writePending = true; - DispatchHandle::rewatchWrite(); -} - -void SslIO::queueWriteClose() { - queuedClose = true; - DispatchHandle::rewatchWrite(); -} - -void SslIO::requestCallback(RequestCallback callback) { - // TODO creating a function object every time isn't all that - // efficient - if this becomes heavily used do something better (what?) - assert(callback); - DispatchHandle::call(boost::bind(&SslIO::requestedCall, this, callback)); -} - -void SslIO::requestedCall(RequestCallback callback) { - assert(callback); - callback(*this); -} - -/** Return a queued buffer if there are enough - * to spare - */ -SslIO::BufferBase* SslIO::getQueuedBuffer() { - // Always keep at least one buffer (it might have data that was "unread" in it) - if (bufferQueue.size()<=1) - return 0; - BufferBase* buff = bufferQueue.back(); - assert(buff); - buff->dataStart = 0; - buff->dataCount = 0; - bufferQueue.pop_back(); - return buff; -} - -/* - * We keep on reading as long as we have something to read and a buffer to put - * it in - */ -void SslIO::readable(DispatchHandle& h) { - AbsTime readStartTime = AbsTime::now(); - do { - // (Try to) get a buffer - if (!bufferQueue.empty()) { - // Read into buffer - BufferBase* buff = bufferQueue.front(); - assert(buff); - bufferQueue.pop_front(); - errno = 0; - int readCount = buff->byteCount-buff->dataCount; - int rc = socket.read(buff->bytes + buff->dataCount, readCount); - if (rc > 0) { - buff->dataCount += rc; - threadReadTotal += rc; - - readCallback(*this, buff); - if (rc != readCount) { - // If we didn't fill the read buffer then time to stop reading - break; - } - - // Stop reading if we've overrun our timeslot - if (Duration(readStartTime, AbsTime::now()) > threadMaxIoTimeNs) { - break; - } - - } else { - // Put buffer back (at front so it doesn't interfere with unread buffers) - bufferQueue.push_front(buff); - assert(buff); - - // Eof or other side has gone away - if (rc == 0 || errno == ECONNRESET) { - eofCallback(*this); - h.unwatchRead(); - break; - } else if (errno == EAGAIN) { - // We have just put a buffer back so we know - // we can carry on watching for reads - break; - } else { - // Report error then just treat as a socket disconnect - QPID_LOG(error, "Error reading socket: " << getErrorString(PR_GetError())); - eofCallback(*this); - h.unwatchRead(); - break; - } - } - } else { - // Something to read but no buffer - if (emptyCallback) { - emptyCallback(*this); - } - // If we still have no buffers we can't do anything more - if (bufferQueue.empty()) { - h.unwatchRead(); - break; - } - - } - } while (true); - - ++threadReadCount; - return; -} - -/* - * We carry on writing whilst we have data to write and we can write - */ -void SslIO::writeable(DispatchHandle& h) { - AbsTime writeStartTime = AbsTime::now(); - do { - // See if we've got something to write - if (!writeQueue.empty()) { - // Write buffer - BufferBase* buff = writeQueue.back(); - writeQueue.pop_back(); - errno = 0; - assert(buff->dataStart+buff->dataCount <= buff->byteCount); - int rc = socket.write(buff->bytes+buff->dataStart, buff->dataCount); - if (rc >= 0) { - threadWriteTotal += rc; - - // If we didn't write full buffer put rest back - if (rc != buff->dataCount) { - buff->dataStart += rc; - buff->dataCount -= rc; - writeQueue.push_back(buff); - break; - } - - // Recycle the buffer - queueReadBuffer(buff); - - // Stop writing if we've overrun our timeslot - if (Duration(writeStartTime, AbsTime::now()) > threadMaxIoTimeNs) { - break; - } - } else { - // Put buffer back - writeQueue.push_back(buff); - if (errno == ECONNRESET || errno == EPIPE) { - // Just stop watching for write here - we'll get a - // disconnect callback soon enough - h.unwatchWrite(); - break; - } else if (errno == EAGAIN) { - // We have just put a buffer back so we know - // we can carry on watching for writes - break; - } else { - QPID_LOG(error, "Error writing to socket: " << getErrorString(PR_GetError())); - h.unwatchWrite(); - break; - } - } - } else { - // If we're waiting to close the socket then can do it now as there is nothing to write - if (queuedClose) { - close(h); - break; - } - // Fd is writable, but nothing to write - if (idleCallback) { - writePending = false; - idleCallback(*this); - } - // If we still have no buffers to write we can't do anything more - if (writeQueue.empty() && !writePending && !queuedClose) { - h.unwatchWrite(); - // The following handles the case where writePending is - // set to true after the test above; in this case its - // possible that the unwatchWrite overwrites the - // desired rewatchWrite so we correct that here - if (writePending) - h.rewatchWrite(); - break; - } - } - } while (true); - - ++threadWriteCount; - return; -} - -void SslIO::disconnected(DispatchHandle& h) { - // If we've already queued close do it instead of disconnected callback - if (queuedClose) { - close(h); - } else if (disCallback) { - disCallback(*this); - h.unwatch(); - } -} - -/* - * Close the socket and callback to say we've done it - */ -void SslIO::close(DispatchHandle& h) { - h.stopWatch(); - socket.close(); - if (closedCallback) { - closedCallback(*this, socket); - } -} - -SecuritySettings SslIO::getSecuritySettings() { - SecuritySettings settings; - settings.ssf = socket.getKeyLen(); - settings.authid = socket.getClientAuthId(); - return settings; -} - -}}} diff --git a/cpp/src/qpid/sys/ssl/SslIo.h b/cpp/src/qpid/sys/ssl/SslIo.h deleted file mode 100644 index f3112bfa65..0000000000 --- a/cpp/src/qpid/sys/ssl/SslIo.h +++ /dev/null @@ -1,193 +0,0 @@ -#ifndef _sys_ssl_SslIO -#define _sys_ssl_SslIO -/* - * - * 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 "qpid/sys/DispatchHandle.h" -#include "qpid/sys/SecuritySettings.h" - -#include <boost/function.hpp> -#include <boost/shared_array.hpp> -#include <deque> - -namespace qpid { -namespace sys { - -class Socket; - -namespace ssl { - -class SslSocket; - -/* - * Asynchronous ssl acceptor: accepts connections then does a callback - * with the accepted fd - */ -template <class T> -class SslAcceptorTmpl { -public: - typedef boost::function1<void, const Socket&> Callback; - -private: - Callback acceptedCallback; - qpid::sys::DispatchHandle handle; - const T& socket; - -public: - SslAcceptorTmpl(const T& s, Callback callback); - ~SslAcceptorTmpl(); - void start(qpid::sys::Poller::shared_ptr poller); - -private: - void readable(qpid::sys::DispatchHandle& handle); -}; - -/* - * Asynchronous ssl connector: starts the process of initiating a - * connection and invokes a callback when completed or failed. - */ -class SslConnector : private qpid::sys::DispatchHandle { -public: - typedef boost::function1<void, const SslSocket&> ConnectedCallback; - typedef boost::function2<void, int, std::string> FailedCallback; - -private: - ConnectedCallback connCallback; - FailedCallback failCallback; - const SslSocket& socket; - -public: - SslConnector(const SslSocket& socket, - Poller::shared_ptr poller, - std::string hostname, - std::string port, - ConnectedCallback connCb, - FailedCallback failCb = 0); - -private: - void connComplete(DispatchHandle& handle); - void failure(int, std::string); -}; - -struct SslIOBufferBase { - char* bytes; - int32_t byteCount; - int32_t dataStart; - int32_t dataCount; - - SslIOBufferBase(char* const b, const int32_t s) : - bytes(b), - byteCount(s), - dataStart(0), - dataCount(0) - {} - - virtual ~SslIOBufferBase() - {} -}; - -/* - * Asychronous reader/writer: - * Reader accepts buffers to read into; reads into the provided buffers - * and then does a callback with the buffer and amount read. Optionally it can callback - * when there is something to read but no buffer to read it into. - * - * Writer accepts a buffer and queues it for writing; can also be given - * a callback for when writing is "idle" (ie fd is writable, but nothing to write) - * - * The class is implemented in terms of DispatchHandle to allow it to be deleted by deleting - * the contained DispatchHandle - */ -class SslIO : private qpid::sys::DispatchHandle { -public: - typedef SslIOBufferBase BufferBase; - - typedef boost::function2<void, SslIO&, BufferBase*> ReadCallback; - typedef boost::function1<void, SslIO&> EofCallback; - typedef boost::function1<void, SslIO&> DisconnectCallback; - typedef boost::function2<void, SslIO&, const SslSocket&> ClosedCallback; - typedef boost::function1<void, SslIO&> BuffersEmptyCallback; - typedef boost::function1<void, SslIO&> IdleCallback; - typedef boost::function1<void, SslIO&> RequestCallback; - - SslIO(const SslSocket& s, - ReadCallback rCb, EofCallback eofCb, DisconnectCallback disCb, - ClosedCallback cCb = 0, BuffersEmptyCallback eCb = 0, IdleCallback iCb = 0); -private: - ReadCallback readCallback; - EofCallback eofCallback; - DisconnectCallback disCallback; - ClosedCallback closedCallback; - BuffersEmptyCallback emptyCallback; - IdleCallback idleCallback; - const SslSocket& socket; - std::deque<BufferBase*> bufferQueue; - std::deque<BufferBase*> writeQueue; - std::vector<BufferBase> buffers; - boost::shared_array<char> bufferMemory; - bool queuedClose; - /** - * This flag is used to detect and handle concurrency between - * calls to notifyPendingWrite() (which can be made from any thread) and - * the execution of the writeable() method (which is always on the - * thread processing this handle. - */ - volatile bool writePending; - -public: - /* - * Size of IO buffers - this is the maximum possible frame size + 1 - */ - const static uint32_t MaxBufferSize = 65536; - - /* - * Number of IO buffers allocated - I think the code can only use 2 - - * 1 for reading and 1 for writing, allocate 4 for safety - */ - const static uint32_t BufferCount = 4; - - void queueForDeletion(); - - void start(qpid::sys::Poller::shared_ptr poller); - void createBuffers(uint32_t size = MaxBufferSize); - void queueReadBuffer(BufferBase* buff); - void unread(BufferBase* buff); - void queueWrite(BufferBase* buff); - void notifyPendingWrite(); - void queueWriteClose(); - bool writeQueueEmpty() { return writeQueue.empty(); } - void requestCallback(RequestCallback); - BufferBase* getQueuedBuffer(); - - qpid::sys::SecuritySettings getSecuritySettings(); - -private: - ~SslIO(); - void readable(qpid::sys::DispatchHandle& handle); - void writeable(qpid::sys::DispatchHandle& handle); - void disconnected(qpid::sys::DispatchHandle& handle); - void requestedCall(RequestCallback); - void close(qpid::sys::DispatchHandle& handle); -}; - -}}} - -#endif // _sys_ssl_SslIO diff --git a/cpp/src/qpid/sys/ssl/SslSocket.cpp b/cpp/src/qpid/sys/ssl/SslSocket.cpp index 30234bb686..a328e49c13 100644 --- a/cpp/src/qpid/sys/ssl/SslSocket.cpp +++ b/cpp/src/qpid/sys/ssl/SslSocket.cpp @@ -20,6 +20,7 @@ */ #include "qpid/sys/ssl/SslSocket.h" +#include "qpid/sys/SocketAddress.h" #include "qpid/sys/ssl/check.h" #include "qpid/sys/ssl/util.h" #include "qpid/Exception.h" @@ -52,28 +53,6 @@ namespace sys { namespace ssl { namespace { -std::string getService(int fd, bool local) -{ - ::sockaddr_storage name; // big enough for any socket address - ::socklen_t namelen = sizeof(name); - - int result = -1; - if (local) { - result = ::getsockname(fd, (::sockaddr*)&name, &namelen); - } else { - result = ::getpeername(fd, (::sockaddr*)&name, &namelen); - } - - QPID_POSIX_CHECK(result); - - char servName[NI_MAXSERV]; - if (int rc=::getnameinfo((::sockaddr*)&name, namelen, 0, 0, - servName, sizeof(servName), - NI_NUMERICHOST | NI_NUMERICSERV) != 0) - throw QPID_POSIX_ERROR(rc); - return servName; -} - const std::string DOMAIN_SEPARATOR("@"); const std::string DC_SEPARATOR("."); const std::string DC("DC"); @@ -101,14 +80,18 @@ std::string getDomainFromSubject(std::string subject) } return domain; } - } -SslSocket::SslSocket() : socket(0), prototype(0) +SslSocket::SslSocket(const std::string& certName, bool clientAuth) : + nssSocket(0), certname(certName), prototype(0) { - impl->fd = ::socket (PF_INET, SOCK_STREAM, 0); - if (impl->fd < 0) throw QPID_POSIX_ERROR(errno); - socket = SSL_ImportFD(0, PR_ImportTCPSocket(impl->fd)); + //configure prototype socket: + prototype = SSL_ImportFD(0, PR_NewTCPSocket()); + + if (clientAuth) { + NSS_CHECK(SSL_OptionSet(prototype, SSL_REQUEST_CERTIFICATE, PR_TRUE)); + NSS_CHECK(SSL_OptionSet(prototype, SSL_REQUIRE_CERTIFICATE, PR_TRUE)); + } } /** @@ -116,25 +99,44 @@ SslSocket::SslSocket() : socket(0), prototype(0) * returned from accept. Because we use posix accept rather than * PR_Accept, we have to reset the handshake. */ -SslSocket::SslSocket(IOHandlePrivate* ioph, PRFileDesc* model) : Socket(ioph), socket(0), prototype(0) +SslSocket::SslSocket(int fd, PRFileDesc* model) : BSDSocket(fd), nssSocket(0), prototype(0) { - socket = SSL_ImportFD(model, PR_ImportTCPSocket(impl->fd)); - NSS_CHECK(SSL_ResetHandshake(socket, true)); + nssSocket = SSL_ImportFD(model, PR_ImportTCPSocket(fd)); + NSS_CHECK(SSL_ResetHandshake(nssSocket, PR_TRUE)); } void SslSocket::setNonblocking() const { + if (!nssSocket) { + BSDSocket::setNonblocking(); + return; + } PRSocketOptionData option; option.option = PR_SockOpt_Nonblocking; option.value.non_blocking = true; - PR_SetSocketOption(socket, &option); + PR_SetSocketOption(nssSocket, &option); } -void SslSocket::connect(const std::string& host, const std::string& port) const +void SslSocket::setTcpNoDelay() const { - std::stringstream namestream; - namestream << host << ":" << port; - connectname = namestream.str(); + if (!nssSocket) { + BSDSocket::setTcpNoDelay(); + return; + } + PRSocketOptionData option; + option.option = PR_SockOpt_NoDelay; + option.value.no_delay = true; + PR_SetSocketOption(nssSocket, &option); +} + +void SslSocket::connect(const SocketAddress& addr) const +{ + BSDSocket::connect(addr); +} + +void SslSocket::finishConnect(const SocketAddress& addr) const +{ + nssSocket = SSL_ImportFD(0, PR_ImportTCPSocket(fd)); void* arg; // Use the connection's cert-name if it has one; else use global cert-name @@ -145,75 +147,48 @@ void SslSocket::connect(const std::string& host, const std::string& port) const } else { arg = const_cast<char*>(SslOptions::global.certName.c_str()); } - NSS_CHECK(SSL_GetClientAuthDataHook(socket, NSS_GetClientAuthData, arg)); - NSS_CHECK(SSL_SetURL(socket, host.data())); - - char hostBuffer[PR_NETDB_BUF_SIZE]; - PRHostEnt hostEntry; - PR_CHECK(PR_GetHostByName(host.data(), hostBuffer, PR_NETDB_BUF_SIZE, &hostEntry)); - PRNetAddr address; - int value = PR_EnumerateHostEnt(0, &hostEntry, boost::lexical_cast<PRUint16>(port), &address); - if (value < 0) { - throw Exception(QPID_MSG("Error getting address for host: " << ErrorString())); - } else if (value == 0) { - throw Exception(QPID_MSG("Could not resolve address for host.")); - } - PR_CHECK(PR_Connect(socket, &address, PR_INTERVAL_NO_TIMEOUT)); - NSS_CHECK(SSL_ForceHandshake(socket)); + NSS_CHECK(SSL_GetClientAuthDataHook(nssSocket, NSS_GetClientAuthData, arg)); + + url = addr.getHost(); + NSS_CHECK(SSL_SetURL(nssSocket, url.data())); + + NSS_CHECK(SSL_ResetHandshake(nssSocket, PR_FALSE)); + NSS_CHECK(SSL_ForceHandshake(nssSocket)); } void SslSocket::close() const { - if (impl->fd > 0) { - PR_Close(socket); - impl->fd = -1; + if (!nssSocket) { + BSDSocket::close(); + return; + } + if (fd > 0) { + PR_Close(nssSocket); + fd = -1; } } -int SslSocket::listen(uint16_t port, int backlog, const std::string& certName, bool clientAuth) const +int SslSocket::listen(const SocketAddress& sa, int backlog) const { - //configure prototype socket: - prototype = SSL_ImportFD(0, PR_NewTCPSocket()); - if (clientAuth) { - NSS_CHECK(SSL_OptionSet(prototype, SSL_REQUEST_CERTIFICATE, PR_TRUE)); - NSS_CHECK(SSL_OptionSet(prototype, SSL_REQUIRE_CERTIFICATE, PR_TRUE)); - } - //get certificate and key (is this the correct way?) - CERTCertificate *cert = PK11_FindCertFromNickname(const_cast<char*>(certName.c_str()), 0); - if (!cert) throw Exception(QPID_MSG("Failed to load certificate '" << certName << "'")); + std::string cName( (certname == "") ? "localhost.localdomain" : certname); + CERTCertificate *cert = PK11_FindCertFromNickname(const_cast<char*>(cName.c_str()), 0); + if (!cert) throw Exception(QPID_MSG("Failed to load certificate '" << cName << "'")); SECKEYPrivateKey *key = PK11_FindKeyByAnyCert(cert, 0); if (!key) throw Exception(QPID_MSG("Failed to retrieve private key from certificate")); NSS_CHECK(SSL_ConfigSecureServer(prototype, cert, key, NSS_FindCertKEAType(cert))); SECKEY_DestroyPrivateKey(key); CERT_DestroyCertificate(cert); - //bind and listen - const int& socket = impl->fd; - int yes=1; - QPID_POSIX_CHECK(setsockopt(socket,SOL_SOCKET,SO_REUSEADDR,&yes,sizeof(yes))); - struct sockaddr_in name; - name.sin_family = AF_INET; - name.sin_port = htons(port); - name.sin_addr.s_addr = 0; - if (::bind(socket, (struct sockaddr*)&name, sizeof(name)) < 0) - throw Exception(QPID_MSG("Can't bind to port " << port << ": " << strError(errno))); - if (::listen(socket, backlog) < 0) - throw Exception(QPID_MSG("Can't listen on port " << port << ": " << strError(errno))); - - socklen_t namelen = sizeof(name); - if (::getsockname(socket, (struct sockaddr*)&name, &namelen) < 0) - throw QPID_POSIX_ERROR(errno); - - return ntohs(name.sin_port); + return BSDSocket::listen(sa, backlog); } -SslSocket* SslSocket::accept() const +Socket* SslSocket::accept() const { QPID_LOG(trace, "Accepting SSL connection."); - int afd = ::accept(impl->fd, 0, 0); + int afd = ::accept(fd, 0, 0); if ( afd >= 0) { - return new SslSocket(new IOHandlePrivate(afd), prototype); + return new SslSocket(afd, prototype); } else if (errno == EAGAIN) { return 0; } else { @@ -297,17 +272,22 @@ static bool isSslStream(int afd) { return isSSL2Handshake || isSSL3Handshake; } +SslMuxSocket::SslMuxSocket(const std::string& certName, bool clientAuth) : + SslSocket(certName, clientAuth) +{ +} + Socket* SslMuxSocket::accept() const { - int afd = ::accept(impl->fd, 0, 0); + int afd = ::accept(fd, 0, 0); if (afd >= 0) { QPID_LOG(trace, "Accepting connection with optional SSL wrapper."); if (isSslStream(afd)) { QPID_LOG(trace, "Accepted SSL connection."); - return new SslSocket(new IOHandlePrivate(afd), prototype); + return new SslSocket(afd, prototype); } else { QPID_LOG(trace, "Accepted Plaintext connection."); - return new Socket(new IOHandlePrivate(afd)); + return new BSDSocket(afd); } } else if (errno == EAGAIN) { return 0; @@ -318,32 +298,12 @@ Socket* SslMuxSocket::accept() const int SslSocket::read(void *buf, size_t count) const { - return PR_Read(socket, buf, count); + return PR_Read(nssSocket, buf, count); } int SslSocket::write(const void *buf, size_t count) const { - return PR_Write(socket, buf, count); -} - -uint16_t SslSocket::getLocalPort() const -{ - return std::atoi(getService(impl->fd, true).c_str()); -} - -uint16_t SslSocket::getRemotePort() const -{ - return atoi(getService(impl->fd, true).c_str()); -} - -void SslSocket::setTcpNoDelay(bool nodelay) const -{ - if (nodelay) { - PRSocketOptionData option; - option.option = PR_SockOpt_NoDelay; - option.value.no_delay = true; - PR_SetSocketOption(socket, &option); - } + return PR_Write(nssSocket, buf, count); } void SslSocket::setCertName(const std::string& name) @@ -359,7 +319,7 @@ int SslSocket::getKeyLen() const int keySize = 0; SECStatus rc; - rc = SSL_SecurityStatus( socket, + rc = SSL_SecurityStatus( nssSocket, &enabled, NULL, NULL, @@ -374,7 +334,7 @@ int SslSocket::getKeyLen() const std::string SslSocket::getClientAuthId() const { std::string authId; - CERTCertificate* cert = SSL_PeerCertificate(socket); + CERTCertificate* cert = SSL_PeerCertificate(nssSocket); if (cert) { authId = CERT_GetCommonName(&(cert->subject)); /* diff --git a/cpp/src/qpid/sys/ssl/SslSocket.h b/cpp/src/qpid/sys/ssl/SslSocket.h index eabadcbe23..fc97059cfd 100644 --- a/cpp/src/qpid/sys/ssl/SslSocket.h +++ b/cpp/src/qpid/sys/ssl/SslSocket.h @@ -23,7 +23,7 @@ */ #include "qpid/sys/IOHandle.h" -#include "qpid/sys/Socket.h" +#include "qpid/sys/posix/BSDSocket.h" #include <nspr.h> #include <string> @@ -37,55 +37,54 @@ class Duration; namespace ssl { -class SslSocket : public qpid::sys::Socket +class SslSocket : public qpid::sys::BSDSocket { public: - /** Create a socket wrapper for descriptor. */ - SslSocket(); + /** Create a socket wrapper for descriptor. + *@param certName name of certificate to use to identify the socket + */ + SslSocket(const std::string& certName = "", bool clientAuth = false); /** Set socket non blocking */ void setNonblocking() const; /** Set tcp-nodelay */ - void setTcpNoDelay(bool nodelay) const; + void setTcpNoDelay() const; /** Set SSL cert-name. Allows the cert-name to be set per * connection, overriding global cert-name settings from * NSSInit().*/ void setCertName(const std::string& certName); - void connect(const std::string& host, const std::string& port) const; + void connect(const SocketAddress&) const; + void finishConnect(const SocketAddress&) const; void close() const; /** Bind to a port and start listening. *@param port 0 means choose an available port. *@param backlog maximum number of pending connections. - *@param certName name of certificate to use to identify the server *@return The bound port. */ - int listen(uint16_t port = 0, int backlog = 10, const std::string& certName = "localhost.localdomain", bool clientAuth = false) const; + int listen(const SocketAddress&, int backlog = 10) const; /** * Accept a connection from a socket that is already listening * and has an incoming connection */ - SslSocket* accept() const; + virtual Socket* accept() const; // TODO The following are raw operations, maybe they need better wrapping? int read(void *buf, size_t count) const; int write(const void *buf, size_t count) const; - uint16_t getLocalPort() const; - uint16_t getRemotePort() const; - int getKeyLen() const; std::string getClientAuthId() const; protected: - mutable std::string connectname; - mutable PRFileDesc* socket; + mutable PRFileDesc* nssSocket; std::string certname; + mutable std::string url; /** * 'model' socket, with configuration to use when importing @@ -94,13 +93,14 @@ protected: */ mutable PRFileDesc* prototype; - SslSocket(IOHandlePrivate* ioph, PRFileDesc* model); - friend class SslMuxSocket; + SslSocket(int fd, PRFileDesc* model); + friend class SslMuxSocket; // Needed for this constructor }; class SslMuxSocket : public SslSocket { public: + SslMuxSocket(const std::string& certName = "", bool clientAuth = false); Socket* accept() const; }; diff --git a/cpp/src/qpid/sys/ssl/util.cpp b/cpp/src/qpid/sys/ssl/util.cpp index 3078e894df..de5d638b09 100644 --- a/cpp/src/qpid/sys/ssl/util.cpp +++ b/cpp/src/qpid/sys/ssl/util.cpp @@ -31,8 +31,6 @@ #include <iostream> #include <fstream> -#include <boost/filesystem/operations.hpp> -#include <boost/filesystem/path.hpp> namespace qpid { namespace sys { @@ -82,15 +80,14 @@ SslOptions SslOptions::global; char* readPasswordFromFile(PK11SlotInfo*, PRBool retry, void*) { const std::string& passwordFile = SslOptions::global.certPasswordFile; - if (retry || passwordFile.empty() || !boost::filesystem::exists(passwordFile)) { - return 0; - } else { - std::ifstream file(passwordFile.c_str()); - std::string password; - file >> password; - return PL_strdup(password.c_str()); - } -} + if (retry || passwordFile.empty()) return 0; + std::ifstream file(passwordFile.c_str()); + if (!file) return 0; + + std::string password; + file >> password; + return PL_strdup(password.c_str()); +} void initNSS(const SslOptions& options, bool server) { diff --git a/cpp/src/qpid/sys/windows/AsynchIO.cpp b/cpp/src/qpid/sys/windows/AsynchIO.cpp index 355acbe0e6..b36ee9f941 100644 --- a/cpp/src/qpid/sys/windows/AsynchIO.cpp +++ b/cpp/src/qpid/sys/windows/AsynchIO.cpp @@ -24,6 +24,8 @@ #include "qpid/sys/AsynchIO.h" #include "qpid/sys/Mutex.h" #include "qpid/sys/Socket.h" +#include "qpid/sys/windows/WinSocket.h" +#include "qpid/sys/SocketAddress.h" #include "qpid/sys/Poller.h" #include "qpid/sys/Thread.h" #include "qpid/sys/Time.h" @@ -50,8 +52,8 @@ namespace { * The function pointers for AcceptEx and ConnectEx need to be looked up * at run time. */ -const LPFN_ACCEPTEX lookUpAcceptEx(const qpid::sys::Socket& s) { - SOCKET h = toSocketHandle(s); +const LPFN_ACCEPTEX lookUpAcceptEx(const qpid::sys::IOHandle& io) { + SOCKET h = io.fd; GUID guidAcceptEx = WSAID_ACCEPTEX; DWORD dwBytes = 0; LPFN_ACCEPTEX fnAcceptEx; @@ -93,12 +95,14 @@ private: AsynchAcceptor::Callback acceptedCallback; const Socket& socket; + const SOCKET wSocket; const LPFN_ACCEPTEX fnAcceptEx; }; AsynchAcceptor::AsynchAcceptor(const Socket& s, Callback callback) : acceptedCallback(callback), socket(s), + wSocket(IOHandle(s).fd), fnAcceptEx(lookUpAcceptEx(s)) { s.setNonblocking(); @@ -121,8 +125,8 @@ void AsynchAcceptor::restart(void) { this, socket); BOOL status; - status = fnAcceptEx(toSocketHandle(socket), - toSocketHandle(*result->newSocket), + status = fnAcceptEx(wSocket, + IOHandle(*result->newSocket).fd, result->addressBuffer, 0, AsynchAcceptResult::SOCKADDRMAXLEN, @@ -133,16 +137,30 @@ void AsynchAcceptor::restart(void) { } +Socket* createSameTypeSocket(const Socket& sock) { + SOCKET socket = IOHandle(sock).fd; + // Socket currently has no actual socket attached + if (socket == INVALID_SOCKET) + return new WinSocket; + + ::sockaddr_storage sa; + ::socklen_t salen = sizeof(sa); + QPID_WINSOCK_CHECK(::getsockname(socket, (::sockaddr*)&sa, &salen)); + SOCKET s = ::socket(sa.ss_family, SOCK_STREAM, 0); // Currently only work with SOCK_STREAM + if (s == INVALID_SOCKET) throw QPID_WINDOWS_ERROR(WSAGetLastError()); + return new WinSocket(s); +} + AsynchAcceptResult::AsynchAcceptResult(AsynchAcceptor::Callback cb, AsynchAcceptor *acceptor, - const Socket& listener) + const Socket& lsocket) : callback(cb), acceptor(acceptor), - listener(toSocketHandle(listener)), - newSocket(listener.createSameTypeSocket()) { + listener(IOHandle(lsocket).fd), + newSocket(createSameTypeSocket(lsocket)) { } void AsynchAcceptResult::success(size_t /*bytesTransferred*/) { - ::setsockopt (toSocketHandle(*newSocket), + ::setsockopt (IOHandle(*newSocket).fd, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, (char*)&listener, @@ -180,6 +198,7 @@ public: ConnectedCallback connCb, FailedCallback failCb = 0); void start(Poller::shared_ptr poller); + void requestCallback(RequestCallback rCb); }; AsynchConnector::AsynchConnector(const Socket& sock, @@ -195,7 +214,7 @@ AsynchConnector::AsynchConnector(const Socket& sock, void AsynchConnector::start(Poller::shared_ptr) { try { - socket.connect(hostname, port); + socket.connect(SocketAddress(hostname, port)); socket.setNonblocking(); connCallback(socket); } catch(std::exception& e) { @@ -205,6 +224,13 @@ void AsynchConnector::start(Poller::shared_ptr) } } +// This can never be called in the current windows code as connect +// is blocking and requestCallback only makes sense if connect is +// non-blocking with the results returned via a poller callback. +void AsynchConnector::requestCallback(RequestCallback rCb) +{ +} + } // namespace windows AsynchAcceptor* AsynchAcceptor::create(const Socket& s, @@ -260,8 +286,6 @@ public: virtual void notifyPendingWrite(); virtual void queueWriteClose(); virtual bool writeQueueEmpty(); - virtual void startReading(); - virtual void stopReading(); virtual void requestCallback(RequestCallback); /** @@ -272,6 +296,8 @@ public: */ virtual BufferBase* getQueuedBuffer(); + virtual SecuritySettings getSecuritySettings(void); + private: ReadCallback readCallback; EofCallback eofCallback; @@ -319,6 +345,12 @@ private: void close(void); /** + * startReading initiates reading, readComplete() is + * called when the read completes. + */ + void startReading(); + + /** * readComplete is called when a read request is complete. * * @param result Results of the operation. @@ -362,7 +394,7 @@ class CallbackHandle : public IOHandle { public: CallbackHandle(AsynchIoResult::Completer completeCb, AsynchIO::RequestCallback reqCb = 0) : - IOHandle(new IOHandlePrivate (INVALID_SOCKET, completeCb, reqCb)) + IOHandle(INVALID_SOCKET, completeCb, reqCb) {} }; @@ -515,7 +547,7 @@ void AsynchIO::startReading() { DWORD bytesReceived = 0, flags = 0; InterlockedIncrement(&opsInProgress); readInProgress = true; - int status = WSARecv(toSocketHandle(socket), + int status = WSARecv(IOHandle(socket).fd, const_cast<LPWSABUF>(result->getWSABUF()), 1, &bytesReceived, &flags, @@ -537,15 +569,6 @@ void AsynchIO::startReading() { return; } -// stopReading was added to prevent a race condition with read-credit on Linux. -// It may or may not be required on windows. -// -// AsynchIOHandler::readbuff() calls stopReading() inside the same -// critical section that protects startReading() in -// AsynchIOHandler::giveReadCredit(). -// -void AsynchIO::stopReading() {} - // Queue the specified callback for invocation from an I/O thread. void AsynchIO::requestCallback(RequestCallback callback) { // This method is generally called from a processing thread; transfer @@ -613,7 +636,7 @@ void AsynchIO::startWrite(AsynchIO::BufferBase* buff) { buff, buff->dataCount); DWORD bytesSent = 0; - int status = WSASend(toSocketHandle(socket), + int status = WSASend(IOHandle(socket).fd, const_cast<LPWSABUF>(result->getWSABUF()), 1, &bytesSent, 0, @@ -639,6 +662,13 @@ void AsynchIO::close(void) { notifyClosed(); } +SecuritySettings AsynchIO::getSecuritySettings() { + SecuritySettings settings; + settings.ssf = socket.getKeyLen(); + settings.authid = socket.getClientAuthId(); + return settings; +} + void AsynchIO::readComplete(AsynchReadResult *result) { int status = result->getStatus(); size_t bytes = result->getTransferred(); @@ -683,7 +713,8 @@ void AsynchIO::writeComplete(AsynchWriteResult *result) { else { // An error... if it's a connection close, ignore it - it will be // noticed and handled on a read completion any moment now. - // What to do with real error??? Save the Buffer? + // What to do with real error??? Save the Buffer? TBD. + queueReadBuffer(buff); // All done; back to the pool } } diff --git a/cpp/src/qpid/sys/windows/FileSysDir.cpp b/cpp/src/qpid/sys/windows/FileSysDir.cpp index 88f1637d48..e090747715 100644 --- a/cpp/src/qpid/sys/windows/FileSysDir.cpp +++ b/cpp/src/qpid/sys/windows/FileSysDir.cpp @@ -24,6 +24,9 @@ #include <sys/stat.h> #include <direct.h> #include <errno.h> +#include <windows.h> +#include <strsafe.h> + namespace qpid { namespace sys { @@ -50,4 +53,36 @@ void FileSysDir::mkdir(void) throw Exception ("Can't create directory: " + dirPath); } +void FileSysDir::forEachFile(Callback cb) const { + + WIN32_FIND_DATAA findFileData; + char szDir[MAX_PATH]; + size_t dirPathLength; + HANDLE hFind = INVALID_HANDLE_VALUE; + + // create dirPath+"\*" in szDir + StringCchLength (dirPath.c_str(), MAX_PATH, &dirPathLength); + + if (dirPathLength > (MAX_PATH - 3)) { + throw Exception ("Directory path is too long: " + dirPath); + } + + StringCchCopy(szDir, MAX_PATH, dirPath.c_str()); + StringCchCat(szDir, MAX_PATH, TEXT("\\*")); + + // Special work for first file + hFind = FindFirstFileA(szDir, &findFileData); + if (INVALID_HANDLE_VALUE == hFind) { + return; + } + + // process everything that isn't a directory + do { + if (!(findFileData.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY)) { + std::string fileName(findFileData.cFileName); + cb(fileName); + } + } while (FindNextFile(hFind, &findFileData) != 0); +} + }} // namespace qpid::sys diff --git a/cpp/src/qpid/sys/windows/IOHandle.cpp b/cpp/src/qpid/sys/windows/IOHandle.cpp index 250737cb99..19a1c44875 100755 --- a/cpp/src/qpid/sys/windows/IOHandle.cpp +++ b/cpp/src/qpid/sys/windows/IOHandle.cpp @@ -19,24 +19,11 @@ * */ -#include "qpid/sys/IOHandle.h" #include "qpid/sys/windows/IoHandlePrivate.h" #include <windows.h> namespace qpid { namespace sys { -SOCKET toFd(const IOHandlePrivate* h) -{ - return h->fd; -} - -IOHandle::IOHandle(IOHandlePrivate* h) : - impl(h) -{} - -IOHandle::~IOHandle() { - delete impl; -} }} // namespace qpid::sys diff --git a/cpp/src/qpid/sys/windows/IoHandlePrivate.h b/cpp/src/qpid/sys/windows/IoHandlePrivate.h index 5943db5cc7..4529ad93ec 100755 --- a/cpp/src/qpid/sys/windows/IoHandlePrivate.h +++ b/cpp/src/qpid/sys/windows/IoHandlePrivate.h @@ -38,15 +38,14 @@ namespace sys { // completer from an I/O thread. If the callback mechanism is used, there // can be a RequestCallback set - this carries the callback object through // from AsynchIO::requestCallback() through to the I/O completion processing. -class IOHandlePrivate { - friend QPID_COMMON_EXTERN SOCKET toSocketHandle(const Socket& s); - static IOHandlePrivate* getImpl(const IOHandle& h); - +class IOHandle { public: - IOHandlePrivate(SOCKET f = INVALID_SOCKET, - windows::AsynchIoResult::Completer cb = 0, - AsynchIO::RequestCallback reqCallback = 0) : - fd(f), event(cb), cbRequest(reqCallback) + IOHandle(SOCKET f = INVALID_SOCKET, + windows::AsynchIoResult::Completer cb = 0, + AsynchIO::RequestCallback reqCallback = 0) : + fd(f), + event(cb), + cbRequest(reqCallback) {} SOCKET fd; @@ -54,8 +53,6 @@ public: AsynchIO::RequestCallback cbRequest; }; -QPID_COMMON_EXTERN SOCKET toSocketHandle(const Socket& s); - }} #endif /* _sys_windows_IoHandlePrivate_h */ diff --git a/cpp/src/qpid/sys/windows/IocpPoller.cpp b/cpp/src/qpid/sys/windows/IocpPoller.cpp index c81cef87b0..ecb33c5517 100755 --- a/cpp/src/qpid/sys/windows/IocpPoller.cpp +++ b/cpp/src/qpid/sys/windows/IocpPoller.cpp @@ -22,7 +22,7 @@ #include "qpid/sys/Poller.h" #include "qpid/sys/Mutex.h" #include "qpid/sys/Dispatcher.h" - +#include "qpid/sys/IOHandle.h" #include "qpid/sys/windows/AsynchIoResult.h" #include "qpid/sys/windows/IoHandlePrivate.h" #include "qpid/sys/windows/check.h" @@ -55,7 +55,7 @@ class PollerHandlePrivate { }; PollerHandle::PollerHandle(const IOHandle& h) : - impl(new PollerHandlePrivate(toSocketHandle(static_cast<const Socket&>(h)), h.impl->event, h.impl->cbRequest)) + impl(new PollerHandlePrivate(h.fd, h.event, h.cbRequest)) {} PollerHandle::~PollerHandle() { diff --git a/cpp/src/qpid/sys/windows/PollableCondition.cpp b/cpp/src/qpid/sys/windows/PollableCondition.cpp index bb637be0a6..3e2a5fb36c 100644 --- a/cpp/src/qpid/sys/windows/PollableCondition.cpp +++ b/cpp/src/qpid/sys/windows/PollableCondition.cpp @@ -52,14 +52,14 @@ private: PollableCondition& parent; boost::shared_ptr<sys::Poller> poller; LONG isSet; + LONG isDispatching; }; PollableConditionPrivate::PollableConditionPrivate(const sys::PollableCondition::Callback& cb, sys::PollableCondition& parent, const boost::shared_ptr<sys::Poller>& poller) - : IOHandle(new sys::IOHandlePrivate(INVALID_SOCKET, - boost::bind(&PollableConditionPrivate::dispatch, this, _1))), - cb(cb), parent(parent), poller(poller), isSet(0) + : IOHandle(INVALID_SOCKET, boost::bind(&PollableConditionPrivate::dispatch, this, _1)), + cb(cb), parent(parent), poller(poller), isSet(0), isDispatching(0) { } @@ -78,7 +78,12 @@ void PollableConditionPrivate::poke() void PollableConditionPrivate::dispatch(windows::AsynchIoResult *result) { delete result; // Poller::monitorHandle() allocates this + // If isDispatching is already set, just return. Else, enter. + if (::InterlockedCompareExchange(&isDispatching, 1, 0) == 1) + return; cb(parent); + LONG oops = ::InterlockedDecrement(&isDispatching); // Result must be 0 + assert(!oops); if (isSet) poke(); } diff --git a/cpp/src/qpid/sys/windows/QpidDllMain.h b/cpp/src/qpid/sys/windows/QpidDllMain.h new file mode 100644 index 0000000000..74eaf0256a --- /dev/null +++ b/cpp/src/qpid/sys/windows/QpidDllMain.h @@ -0,0 +1,72 @@ +/* + * 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 this file once in each DLL that relies on SystemInfo.h: + * threadSafeShutdown(). Note that Thread.cpp has a more elaborate + * DllMain, that also provides this functionality separately. + * + * Teardown is in the reverse order of the DLL dependencies used + * during the load phase. The calls to DllMain and the static + * destructors are from the same thread, so no locking is necessary + * and there is no downside to an invocation of DllMain by multiple + * Qpid DLLs. + */ + +#ifdef _DLL + +#include <qpid/ImportExport.h> +#include <windows.h> + +namespace qpid { +namespace sys { +namespace windows { + +QPID_IMPORT bool processExiting; +QPID_IMPORT bool libraryUnloading; + +}}} // namespace qpid::sys::SystemInfo + + +BOOL APIENTRY DllMain(HMODULE hm, DWORD reason, LPVOID reserved) { + switch (reason) { + case DLL_PROCESS_ATTACH: + case DLL_THREAD_ATTACH: + case DLL_THREAD_DETACH: + break; + + case DLL_PROCESS_DETACH: + // Remember how the process is terminating this DLL. + if (reserved != NULL) { + qpid::sys::windows::processExiting = true; + // Danger: all threading suspect, including indirect use of malloc or locks. + // Think twice before adding more functionality here. + return TRUE; + } + else { + qpid::sys::windows::libraryUnloading = true; + } + break; + } + return TRUE; +} + + +#endif diff --git a/cpp/src/qpid/sys/windows/SslAsynchIO.cpp b/cpp/src/qpid/sys/windows/SslAsynchIO.cpp index d263f00ab3..e48c799b29 100644 --- a/cpp/src/qpid/sys/windows/SslAsynchIO.cpp +++ b/cpp/src/qpid/sys/windows/SslAsynchIO.cpp @@ -209,18 +209,6 @@ bool SslAsynchIO::writeQueueEmpty() { return aio->writeQueueEmpty(); } -/* - * Initiate a read operation. AsynchIO::readComplete() will be - * called when the read is complete and data is available. - */ -void SslAsynchIO::startReading() { - aio->startReading(); -} - -void SslAsynchIO::stopReading() { - aio->stopReading(); -} - // Queue the specified callback for invocation from an I/O thread. void SslAsynchIO::requestCallback(RequestCallback callback) { aio->requestCallback(callback); @@ -241,11 +229,15 @@ AsynchIO::BufferBase* SslAsynchIO::getQueuedBuffer() { return sslBuff; } -unsigned int SslAsynchIO::getSslKeySize() { +SecuritySettings SslAsynchIO::getSecuritySettings() { SecPkgContext_KeyInfo info; memset(&info, 0, sizeof(info)); ::QueryContextAttributes(&ctxtHandle, SECPKG_ATTR_KEY_INFO, &info); - return info.KeySize; + + SecuritySettings settings; + settings.ssf = info.KeySize; + settings.authid = std::string(); + return settings; } void SslAsynchIO::negotiationDone() { diff --git a/cpp/src/qpid/sys/windows/SslAsynchIO.h b/cpp/src/qpid/sys/windows/SslAsynchIO.h index e9d9e8d629..2f6842b135 100644 --- a/cpp/src/qpid/sys/windows/SslAsynchIO.h +++ b/cpp/src/qpid/sys/windows/SslAsynchIO.h @@ -77,12 +77,9 @@ public: virtual void notifyPendingWrite(); virtual void queueWriteClose(); virtual bool writeQueueEmpty(); - virtual void startReading(); - virtual void stopReading(); virtual void requestCallback(RequestCallback); virtual BufferBase* getQueuedBuffer(); - - QPID_COMMON_EXTERN unsigned int getSslKeySize(); + virtual SecuritySettings getSecuritySettings(void); protected: CredHandle credHandle; diff --git a/cpp/src/qpid/sys/windows/SystemInfo.cpp b/cpp/src/qpid/sys/windows/SystemInfo.cpp index cef78dcc60..fb58d53b81 100755 --- a/cpp/src/qpid/sys/windows/SystemInfo.cpp +++ b/cpp/src/qpid/sys/windows/SystemInfo.cpp @@ -25,7 +25,8 @@ #include "qpid/sys/SystemInfo.h" #include "qpid/sys/IntegerTypes.h" -#include "qpid/Exception.h"
+#include "qpid/Exception.h" +#include "qpid/log/Statement.h" #include <assert.h> #include <winsock2.h> @@ -66,39 +67,10 @@ bool SystemInfo::getLocalHostname (Address &address) { static const std::string LOCALHOST("127.0.0.1"); static const std::string TCP("tcp"); -void SystemInfo::getLocalIpAddresses (uint16_t port, - std::vector<Address> &addrList) { - enum { MAX_URL_INTERFACES = 100 }; - - SOCKET s = socket (PF_INET, SOCK_STREAM, 0); - if (s != INVALID_SOCKET) { - INTERFACE_INFO interfaces[MAX_URL_INTERFACES]; - DWORD filledBytes = 0; - WSAIoctl (s, - SIO_GET_INTERFACE_LIST, - 0, - 0, - interfaces, - sizeof (interfaces), - &filledBytes, - 0, - 0); - unsigned int interfaceCount = filledBytes / sizeof (INTERFACE_INFO); - for (unsigned int i = 0; i < interfaceCount; ++i) { - if (interfaces[i].iiFlags & IFF_UP) { - std::string addr(inet_ntoa(interfaces[i].iiAddress.AddressIn.sin_addr)); - if (addr != LOCALHOST) - addrList.push_back(Address(TCP, addr, port)); - } - } - closesocket (s); - } -} - -bool SystemInfo::isLocalHost(const std::string& candidateHost) { - // FIXME aconway 2012-05-03: not implemented. - assert(0); - throw Exception("Not implemented: isLocalHost"); +// Null function which always fails to find an network interface name +bool SystemInfo::getInterfaceAddresses(const std::string&, std::vector<std::string>&) +{ + return false; } void SystemInfo::getSystemId (std::string &osName, @@ -208,4 +180,29 @@ std::string SystemInfo::getProcessName() return name; } + +#ifdef _DLL +namespace windows { +// set from one or more Qpid DLLs: i.e. in DllMain with DLL_PROCESS_DETACH +QPID_EXPORT bool processExiting = false; +QPID_EXPORT bool libraryUnloading = false; +} +#endif + +bool SystemInfo::threadSafeShutdown() +{ +#ifdef _DLL + if (!windows::processExiting && !windows::libraryUnloading) { + // called before exit() or FreeLibrary(), or by a DLL without + // a participating DllMain. + QPID_LOG(warning, "invalid query for shutdown state"); + throw qpid::Exception(QPID_MSG("Unable to determine shutdown state.")); + } + return !windows::processExiting; +#else + // Not a DLL: shutdown can only be by exit() or return from main(). + return false; +#endif +} + }} // namespace qpid::sys diff --git a/cpp/src/qpid/sys/windows/Thread.cpp b/cpp/src/qpid/sys/windows/Thread.cpp index 23b0033be4..b342c9da1d 100755 --- a/cpp/src/qpid/sys/windows/Thread.cpp +++ b/cpp/src/qpid/sys/windows/Thread.cpp @@ -27,6 +27,7 @@ #include "qpid/sys/Thread.h" #include "qpid/sys/Runnable.h" #include "qpid/sys/windows/check.h" +#include "qpid/sys/SystemInfo.h" #include <process.h> #include <windows.h> @@ -274,8 +275,17 @@ Thread Thread::current() { #ifdef _DLL +namespace qpid { +namespace sys { +namespace windows { + +extern bool processExiting; +extern bool libraryUnloading; + +}}} // namespace qpid::sys::SystemInfo + // DllMain: called possibly many times in a process lifetime if dll -// loaded and freed repeatedly . Be mindful of Windows loader lock +// loaded and freed repeatedly. Be mindful of Windows loader lock // and other DllMain restrictions. BOOL APIENTRY DllMain(HMODULE hm, DWORD reason, LPVOID reserved) { @@ -290,10 +300,12 @@ BOOL APIENTRY DllMain(HMODULE hm, DWORD reason, LPVOID reserved) { if (reserved != NULL) { // process exit(): threads are stopped arbitrarily and // possibly in an inconsistent state. Not even threadLock - // can be trusted. All static destructors have been - // called at this point and any resources this unit knows - // about will be released as part of process tear down by - // the OS. Accordingly, do nothing. + // can be trusted. All static destructors for this unit + // are pending and face the same unsafe environment. + // Any resources this unit knows about will be released as + // part of process tear down by the OS. Accordingly, skip + // any clean up tasks. + qpid::sys::windows::processExiting = true; return TRUE; } else { @@ -301,6 +313,7 @@ BOOL APIENTRY DllMain(HMODULE hm, DWORD reason, LPVOID reserved) { // encouraged to clean up to avoid leaks. Mostly we just // want any straggler threads to finish and notify // threadsDone as the last thing they do. + qpid::sys::windows::libraryUnloading = true; while (1) { { ScopedCriticalSection l(threadLock); diff --git a/cpp/src/qpid/sys/windows/Socket.cpp b/cpp/src/qpid/sys/windows/WinSocket.cpp index a4374260cc..b2d2d79c63 100644 --- a/cpp/src/qpid/sys/windows/Socket.cpp +++ b/cpp/src/qpid/sys/windows/WinSocket.cpp @@ -19,18 +19,12 @@ * */ -#include "qpid/sys/Socket.h" +#include "qpid/sys/windows/WinSocket.h" #include "qpid/sys/SocketAddress.h" #include "qpid/sys/windows/check.h" #include "qpid/sys/windows/IoHandlePrivate.h" - -// Ensure we get all of winsock2.h -#ifndef _WIN32_WINNT -#define _WIN32_WINNT 0x0501 -#endif - -#include <winsock2.h> +#include "qpid/sys/SystemInfo.h" namespace qpid { namespace sys { @@ -67,7 +61,8 @@ public: } ~WinSockSetup() { - WSACleanup(); + if (SystemInfo::threadSafeShutdown()) + WSACleanup(); } public: @@ -106,22 +101,32 @@ uint16_t getLocalPort(int fd) } } // namespace -Socket::Socket() : - IOHandle(new IOHandlePrivate), +WinSocket::WinSocket() : + handle(new IOHandle), nonblocking(false), nodelay(false) {} -Socket::Socket(IOHandlePrivate* h) : - IOHandle(h), +Socket* createSocket() +{ + return new WinSocket; +} + +WinSocket::WinSocket(SOCKET fd) : + handle(new IOHandle(fd)), nonblocking(false), nodelay(false) {} -void Socket::createSocket(const SocketAddress& sa) const +WinSocket::operator const IOHandle&() const { - SOCKET& socket = impl->fd; - if (socket != INVALID_SOCKET) Socket::close(); + return *handle; +} + +void WinSocket::createSocket(const SocketAddress& sa) const +{ + SOCKET& socket = handle->fd; + if (socket != INVALID_SOCKET) WinSocket::close(); SOCKET s = ::socket (getAddrInfo(sa).ai_family, getAddrInfo(sa).ai_socktype, @@ -139,39 +144,19 @@ void Socket::createSocket(const SocketAddress& sa) const } } -Socket* Socket::createSameTypeSocket() const { - SOCKET& socket = impl->fd; - // Socket currently has no actual socket attached - if (socket == INVALID_SOCKET) - return new Socket; - - ::sockaddr_storage sa; - ::socklen_t salen = sizeof(sa); - QPID_WINSOCK_CHECK(::getsockname(socket, (::sockaddr*)&sa, &salen)); - SOCKET s = ::socket(sa.ss_family, SOCK_STREAM, 0); // Currently only work with SOCK_STREAM - if (s == INVALID_SOCKET) throw QPID_WINDOWS_ERROR(WSAGetLastError()); - return new Socket(new IOHandlePrivate(s)); -} - -void Socket::setNonblocking() const { +void WinSocket::setNonblocking() const { u_long nonblock = 1; - QPID_WINSOCK_CHECK(ioctlsocket(impl->fd, FIONBIO, &nonblock)); -} - -void Socket::connect(const std::string& host, const std::string& port) const -{ - SocketAddress sa(host, port); - connect(sa); + QPID_WINSOCK_CHECK(ioctlsocket(handle->fd, FIONBIO, &nonblock)); } void -Socket::connect(const SocketAddress& addr) const +WinSocket::connect(const SocketAddress& addr) const { peername = addr.asString(false); createSocket(addr); - const SOCKET& socket = impl->fd; + const SOCKET& socket = handle->fd; int err; WSASetLastError(0); if ((::connect(socket, getAddrInfo(addr).ai_addr, getAddrInfo(addr).ai_addrlen) != 0) && @@ -180,44 +165,43 @@ Socket::connect(const SocketAddress& addr) const } void -Socket::close() const +WinSocket::finishConnect(const SocketAddress&) const { - SOCKET& socket = impl->fd; +} + +void +WinSocket::close() const +{ + SOCKET& socket = handle->fd; if (socket == INVALID_SOCKET) return; QPID_WINSOCK_CHECK(closesocket(socket)); socket = INVALID_SOCKET; } -int Socket::write(const void *buf, size_t count) const +int WinSocket::write(const void *buf, size_t count) const { - const SOCKET& socket = impl->fd; + const SOCKET& socket = handle->fd; int sent = ::send(socket, (const char *)buf, count, 0); if (sent == SOCKET_ERROR) return -1; return sent; } -int Socket::read(void *buf, size_t count) const +int WinSocket::read(void *buf, size_t count) const { - const SOCKET& socket = impl->fd; + const SOCKET& socket = handle->fd; int received = ::recv(socket, (char *)buf, count, 0); if (received == SOCKET_ERROR) return -1; return received; } -int Socket::listen(const std::string& host, const std::string& port, int backlog) const -{ - SocketAddress sa(host, port); - return listen(sa, backlog); -} - -int Socket::listen(const SocketAddress& addr, int backlog) const +int WinSocket::listen(const SocketAddress& addr, int backlog) const { createSocket(addr); - const SOCKET& socket = impl->fd; + const SOCKET& socket = handle->fd; BOOL yes=1; QPID_WINSOCK_CHECK(setsockopt(socket, SOL_SOCKET, SO_REUSEADDR, (char *)&yes, sizeof(yes))); @@ -229,48 +213,48 @@ int Socket::listen(const SocketAddress& addr, int backlog) const return getLocalPort(socket); } -Socket* Socket::accept() const +Socket* WinSocket::accept() const { - SOCKET afd = ::accept(impl->fd, 0, 0); + SOCKET afd = ::accept(handle->fd, 0, 0); if (afd != INVALID_SOCKET) - return new Socket(new IOHandlePrivate(afd)); + return new WinSocket(afd); else if (WSAGetLastError() == EAGAIN) return 0; else throw QPID_WINDOWS_ERROR(WSAGetLastError()); } -std::string Socket::getPeerAddress() const +std::string WinSocket::getPeerAddress() const { if (peername.empty()) { - peername = getName(impl->fd, false); + peername = getName(handle->fd, false); } return peername; } -std::string Socket::getLocalAddress() const +std::string WinSocket::getLocalAddress() const { if (localname.empty()) { - localname = getName(impl->fd, true); + localname = getName(handle->fd, true); } return localname; } -int Socket::getError() const +int WinSocket::getError() const { int result; socklen_t rSize = sizeof (result); - QPID_WINSOCK_CHECK(::getsockopt(impl->fd, SOL_SOCKET, SO_ERROR, (char *)&result, &rSize)); + QPID_WINSOCK_CHECK(::getsockopt(handle->fd, SOL_SOCKET, SO_ERROR, (char *)&result, &rSize)); return result; } -void Socket::setTcpNoDelay() const +void WinSocket::setTcpNoDelay() const { - SOCKET& socket = impl->fd; + SOCKET& socket = handle->fd; nodelay = true; if (socket != INVALID_SOCKET) { int flag = 1; - int result = setsockopt(impl->fd, + int result = setsockopt(handle->fd, IPPROTO_TCP, TCP_NODELAY, (char *)&flag, @@ -279,14 +263,14 @@ void Socket::setTcpNoDelay() const } } -inline IOHandlePrivate* IOHandlePrivate::getImpl(const qpid::sys::IOHandle &h) +int WinSocket::getKeyLen() const { - return h.impl; + return 0; } -SOCKET toSocketHandle(const Socket& s) +std::string WinSocket::getClientAuthId() const { - return IOHandlePrivate::getImpl(s)->fd; + return std::string(); } }} // namespace qpid::sys diff --git a/cpp/src/qpid/sys/windows/WinSocket.h b/cpp/src/qpid/sys/windows/WinSocket.h new file mode 100644 index 0000000000..bee6a58e7a --- /dev/null +++ b/cpp/src/qpid/sys/windows/WinSocket.h @@ -0,0 +1,118 @@ +#ifndef QPID_SYS_WINDOWS_BSDSOCKET_H +#define QPID_SYS_WINDOWS_BSDSOCKET_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 "qpid/sys/Socket.h" +#include "qpid/sys/IntegerTypes.h" +#include "qpid/CommonImportExport.h" +#include <string> + +#include <boost/scoped_ptr.hpp> + +// Ensure we get all of winsock2.h +#ifndef _WIN32_WINNT +#define _WIN32_WINNT 0x0501 +#endif + +#include <winsock2.h> + +namespace qpid { +namespace sys { + +namespace windows { +Socket* createSameTypeSocket(const Socket&); +} + +class Duration; +class IOHandle; +class SocketAddress; + +class QPID_COMMON_CLASS_EXTERN WinSocket : public Socket +{ +public: + /** Create a socket wrapper for descriptor. */ + QPID_COMMON_EXTERN WinSocket(); + + QPID_COMMON_EXTERN operator const IOHandle&() const; + + /** Set socket non blocking */ + QPID_COMMON_EXTERN virtual void setNonblocking() const; + + QPID_COMMON_EXTERN virtual void setTcpNoDelay() const; + + QPID_COMMON_EXTERN virtual void connect(const SocketAddress&) const; + QPID_COMMON_EXTERN virtual void finishConnect(const SocketAddress&) const; + + QPID_COMMON_EXTERN virtual void close() const; + + /** Bind to a port and start listening. + *@return The bound port number + */ + QPID_COMMON_EXTERN virtual int listen(const SocketAddress&, int backlog = 10) const; + + /** + * Returns an address (host and port) for the remote end of the + * socket + */ + QPID_COMMON_EXTERN std::string getPeerAddress() const; + /** + * Returns an address (host and port) for the local end of the + * socket + */ + QPID_COMMON_EXTERN std::string getLocalAddress() const; + + /** + * Returns the error code stored in the socket. This may be used + * to determine the result of a non-blocking connect. + */ + QPID_COMMON_EXTERN int getError() const; + + /** Accept a connection from a socket that is already listening + * and has an incoming connection + */ + QPID_COMMON_EXTERN virtual Socket* accept() const; + + // TODO The following are raw operations, maybe they need better wrapping? + QPID_COMMON_EXTERN virtual int read(void *buf, size_t count) const; + QPID_COMMON_EXTERN virtual int write(const void *buf, size_t count) const; + + QPID_COMMON_EXTERN int getKeyLen() const; + QPID_COMMON_EXTERN std::string getClientAuthId() const; + +protected: + /** Create socket */ + void createSocket(const SocketAddress&) const; + + mutable boost::scoped_ptr<IOHandle> handle; + mutable std::string localname; + mutable std::string peername; + mutable bool nonblocking; + mutable bool nodelay; + + /** Construct socket with existing handle */ + friend Socket* qpid::sys::windows::createSameTypeSocket(const Socket&); + WinSocket(SOCKET fd); +}; + +}} +#endif /*!QPID_SYS_WINDOWS_BSDSOCKET_H*/ |