diff options
| author | Ted Ross <tross@apache.org> | 2009-12-18 16:11:39 +0000 |
|---|---|---|
| committer | Ted Ross <tross@apache.org> | 2009-12-18 16:11:39 +0000 |
| commit | 6bd740528732eb8da6f521be49999a37eb5f44ee (patch) | |
| tree | 14850edfc2282237a657df3fcb63bf2f4c1d850e | |
| parent | 9fc81d4e597ae271632d55c87fc6a61ed3f62a49 (diff) | |
| download | qpid-python-6bd740528732eb8da6f521be49999a37eb5f44ee.tar.gz | |
Rebased the qmfv2 branch to trunk version 892277.
git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/qmfv2@892292 13f79535-47bb-0310-9956-ffa450edef68
108 files changed, 5103 insertions, 1074 deletions
diff --git a/qpid/bin/release.sh b/qpid/bin/release.sh index 084e999362..b0b196fc8c 100755 --- a/qpid/bin/release.sh +++ b/qpid/bin/release.sh @@ -36,9 +36,11 @@ usage() echo "--all |-a : Generate all artefacts" echo "--source|-e : Generate the source artefact" echo "--cpp |-c : Generate the CPP artefacts" + echo "--dotnet|-d : Generate the dotnet artefacts" echo "--java |-j : Generate the java artefacts" echo "--ruby |-r : Generate the ruby artefacts" echo "--python|-p : Generate the python artefacts" + echo "--wcf |-w : Generate the WCF artefacts" echo "--source|-e : Generate the source artefact" echo "--sign |-s : Sign generated artefacts" echo "--upload|-u : Upload the artifacts directory to people.apache.org as qpid-\$VER" @@ -76,11 +78,15 @@ for arg in $* ; do JAVA="JAVA" RUBY="RUBY" PYTHON="PYTHON" + WCF="WCF" SOURCE="SOURCE" ;; --cpp|-c) CPP="CPP" ;; + --dotnet|-d) + DOTNET="DOTNET" + ;; --java|-j) JAVA="JAVA" ;; @@ -90,6 +96,9 @@ for arg in $* ; do --python|-p) PYTHON="PYTHON" ;; + --wcf|-w) + WCF="WCF" + ;; --source|-e) SOURCE="SOURCE" ;; @@ -130,12 +139,14 @@ echo REV:$REV echo VER:$VER # If nothing is specified then do it all -if [ -z "${CLEAN}${PREPARE}${CPP}${JAVA}${RUBY}${PYTHON}${SOURCE}${SIGN}${UPLOAD}" ] ; then +if [ -z "${CLEAN}${PREPARE}${CPP}${DOTNET}${JAVA}${RUBY}${PYTHON}${WCF}${SOURCE}${SIGN}${UPLOAD}" ] ; then PREPARE="PREPARE" CPP="CPP" + DOTNET="DOTNET" JAVA="JAVA" RUBY="RUBY" PYTHON="PYTHON" + WCF="WCF" SOURCE="SOURCE" SIGN="SIGN" @@ -181,6 +192,10 @@ if [ "PYTHON" == "$PYTHON" ] ; then tar -czf artifacts/qpid-python-${VER}.tar.gz qpid-${VER}/python qpid-${VER}/specs fi +if [ "WCF" == "$WCF" ] ; then + zip -rq artifacts/qpid-wcf-${VER}.zip qpid-${VER}/wcf +fi + if [ "CPP" == "$CPP" ] ; then pushd qpid-${VER}/cpp ./bootstrap @@ -204,6 +219,23 @@ if [ "JAVA" == "$JAVA" ] ; then cp qpid-${VER}/java/management/client/release/*.tar.gz artifacts/qpid-management-client-${VER}.tar.gz fi +if [ "DOTNET" == "$DOTNET" ] ; then + pushd qpid-${VER}/dotnet + cd Qpid.Common + ant + cd .. + ./build-nant-release mono-2.0 + + cd client-010/gentool + ant + cd .. + nant -t:mono-2.0 release-pkg + popd + + cp qpid-${VER}/dotnet/bin/mono-2.0/release/*.zip artifacts/qpid-dotnet-0-8-${VER}.zip + cp qpid-${VER}/dotnet/client-010/bin/mono-2.0/debug/*.zip artifacts/qpid-dotnet-0-10-${VER}.zip +fi + if [ "SIGN" == "$SIGN" ] ; then pushd artifacts sha1sum *.zip *.gz *.svnversion > SHA1SUM diff --git a/qpid/cpp/INSTALL b/qpid/cpp/INSTALL index 27e103d15b..d54d2affc3 100644 --- a/qpid/cpp/INSTALL +++ b/qpid/cpp/INSTALL @@ -91,6 +91,12 @@ the following must also be installed: * ruby-devel * python-devel * swig <http://www.swig.org> (1.3.35) + +UUID problems: +In some later Linux releases (such as Fedora 12), the uuid/uuid.h file has been +moved from e2fsprogs-devel into libuuid-devel. If you are using a newer Linux +release and run into a problem during configure in which uuid.h cannot be found, +look for and install the libuuid-devel package. 2.2. How to Install diff --git a/qpid/cpp/examples/CMakeLists.txt b/qpid/cpp/examples/CMakeLists.txt index 759a9f2020..7ecffed4bb 100644 --- a/qpid/cpp/examples/CMakeLists.txt +++ b/qpid/cpp/examples/CMakeLists.txt @@ -28,6 +28,9 @@ include_directories(${CMAKE_SOURCE_DIR}/include) # Shouldn't need this... but there are still client header inclusions # of Boost. When building examples at an install site, the Boost files # should be locatable aside from these settings. +# So set up to find the headers, find the libs at link time, but dynamically +# link them all and clear the CMake Boost library names to avoid adding them to +# the project files. include_directories( ${Boost_INCLUDE_DIR} ) link_directories( ${Boost_LIBRARY_DIRS} ) @@ -35,12 +38,12 @@ link_directories( ${Boost_LIBRARY_DIRS} ) # Linux needs to reference the boost libs, even though they should be # resolved via the Qpid libs. if (MSVC) - add_definitions( /D "NOMINMAX" /D "WIN32_LEAN_AND_MEAN" ) + add_definitions( /D "NOMINMAX" /D "WIN32_LEAN_AND_MEAN" /D "BOOST_ALL_DYN_LINK" ) # On Windows, prevent the accidental inclusion of Boost headers from # autolinking in the Boost libs. There should be no direct references to # Boost in the examples, and references via qpidclient/qpidcommon are # resolved in the Qpid libs. - add_definitions( /D "BOOST_ALL_DYN_LINK" /D "BOOST_ALL_NO_LIB" ) + add_definitions( /D "BOOST_ALL_NO_LIB" ) else (MSVC) set(_boost_libs_needed ${Boost_PROGRAM_OPTIONS_LIBRARY} ${Boost_FILESYSTEM_LIBRARY}) diff --git a/qpid/cpp/examples/messaging/CMakeLists.txt b/qpid/cpp/examples/messaging/CMakeLists.txt index 7c023e602a..b2b2bc3e43 100644 --- a/qpid/cpp/examples/messaging/CMakeLists.txt +++ b/qpid/cpp/examples/messaging/CMakeLists.txt @@ -17,10 +17,10 @@ # under the License. # -# disabling spout & drain under cmake until build issues on windows -# are resolved (QPID-2212): -#add_example(messaging drain ${Boost_PROGRAM_OPTIONS_LIBRARY}) -#add_example(messaging spout ${Boost_PROGRAM_OPTIONS_LIBRARY}) +# drain and spout have explicit Boost.program_options usage in them, so be +# sure that lib is linked in. +add_example(messaging drain ${Boost_PROGRAM_OPTIONS_LIBRARY}) +add_example(messaging spout ${Boost_PROGRAM_OPTIONS_LIBRARY}) add_example(messaging queue_receiver) add_example(messaging queue_sender) diff --git a/qpid/cpp/examples/messaging/client.cpp b/qpid/cpp/examples/messaging/client.cpp index d937127da6..49d5441bf0 100644 --- a/qpid/cpp/examples/messaging/client.cpp +++ b/qpid/cpp/examples/messaging/client.cpp @@ -46,7 +46,7 @@ int main(int argc, char** argv) { Sender sender = session.createSender("service_queue"); //create temp queue & receiver... - Address responseQueue("#response-queue; {create:always, type:queue, node-properties:{x-amqp0-10-auto-delete:true}}"); + Address responseQueue("#response-queue; {create:always, type:queue, node-properties:{ x-properties:{auto-delete:true}}}"); Receiver receiver = session.createReceiver(responseQueue); // Now send some messages ... diff --git a/qpid/cpp/examples/messaging/server.cpp b/qpid/cpp/examples/messaging/server.cpp index 024832f914..137323dd03 100644 --- a/qpid/cpp/examples/messaging/server.cpp +++ b/qpid/cpp/examples/messaging/server.cpp @@ -44,7 +44,7 @@ int main(int argc, char** argv) { try { Connection connection = Connection::open(url); Session session = connection.newSession(); - Receiver receiver = session.createReceiver("service_queue"); + Receiver receiver = session.createReceiver("service_queue; {create: always}"); while (true) { Message request = receiver.fetch(); diff --git a/qpid/cpp/include/qpid/client/SubscriptionSettings.h b/qpid/cpp/include/qpid/client/SubscriptionSettings.h index 4d64119d3a..b4cb302b56 100644 --- a/qpid/cpp/include/qpid/client/SubscriptionSettings.h +++ b/qpid/cpp/include/qpid/client/SubscriptionSettings.h @@ -49,13 +49,44 @@ struct SubscriptionSettings ) : flowControl(flow), acceptMode(accept), acquireMode(acquire), autoAck(autoAck_), completionMode(completion), exclusive(false) {} FlowControl flowControl; ///@< Flow control settings. @see FlowControl + /** + * The acceptMode determines whether the broker should expect + * delivery of messages to be acknowledged by the client + * indicating that it accepts them. A value of + * ACCEPT_MODE_EXPLICIT means that messages must be accepted + * (note: this may be done automatically by the library - see + * autoAck - or through an explicit call be the application - see + * Subscription::accept()) before they can be dequeued. A value of + * ACCEPT_MODE_NONE means that the broker can dequeue a message as + * soon as it is acquired. + */ AcceptMode acceptMode; ///@< ACCEPT_MODE_EXPLICIT or ACCEPT_MODE_NONE + /** + * The acquireMode determines whether messages are locked for the + * subscriber when delivered, and thus are not delivered to any + * other subscriber unless this subscriber releases them. + * + * The default is ACQUIRE_MODE_PRE_ACQUIRED meaning that the + * subscriber expects to have been given that message exclusively + * (i.e. the message will not be given to any other subscriber + * unless released explicitly or by this subscribers session + * failing without having accepted the message). + * + * Delivery of message in ACQUIRE_MODE_NOT_ACQUIRED mode means the + * message will still be available for other subscribers to + * receive. The application can if desired acquire a (set of) + * messages through an explicit acquire call - see + * Subscription::acquire(). + */ AcquireMode acquireMode; ///@< ACQUIRE_MODE_PRE_ACQUIRED or ACQUIRE_MODE_NOT_ACQUIRED - /** Automatically acknowledge (accept) batches of autoAck - * messages. 0 means no automatic acknowledgement. This has no - * effect for messsages received for a subscription with - * ACCEPT_MODE_NODE.*/ + /** + * Configures the frequency at which messages are automatically + * accepted (e.g. a value of 5 means that messages are accepted in + * batches of 5). A value of 0 means no automatic acknowledgement + * will occur and the application will itself be responsible for + * accepting messages. + */ unsigned int autoAck; /** * In windowing mode, completion of a message will cause the diff --git a/qpid/cpp/include/qpid/messaging/Variant.h b/qpid/cpp/include/qpid/messaging/Variant.h index c63138178b..de5cef4d67 100644 --- a/qpid/cpp/include/qpid/messaging/Variant.h +++ b/qpid/cpp/include/qpid/messaging/Variant.h @@ -149,6 +149,8 @@ class Variant QPID_CLIENT_EXTERN void setEncoding(const std::string&); QPID_CLIENT_EXTERN const std::string& getEncoding() const; + QPID_CLIENT_EXTERN bool isEqualTo(const Variant& a) const; + QPID_CLIENT_EXTERN void reset(); private: VariantImpl* impl; @@ -157,6 +159,7 @@ class Variant QPID_CLIENT_EXTERN std::ostream& operator<<(std::ostream& out, const Variant& value); QPID_CLIENT_EXTERN std::ostream& operator<<(std::ostream& out, const Variant::Map& map); QPID_CLIENT_EXTERN std::ostream& operator<<(std::ostream& out, const Variant::List& list); +QPID_CLIENT_EXTERN bool operator==(const Variant& a, const Variant& b); typedef Variant::Map VariantMap; diff --git a/qpid/cpp/src/qpid/sys/ExceptionHolder.h b/qpid/cpp/include/qpid/sys/ExceptionHolder.h index fecaa73eea..9eff1d64c7 100644 --- a/qpid/cpp/src/qpid/sys/ExceptionHolder.h +++ b/qpid/cpp/include/qpid/sys/ExceptionHolder.h @@ -22,7 +22,6 @@ * */ -#include "qpid/memory.h" #include <boost/shared_ptr.hpp> diff --git a/qpid/cpp/include/qpid/sys/windows/IntegerTypes.h b/qpid/cpp/include/qpid/sys/windows/IntegerTypes.h index 7b2c57ad8e..ece1a618e9 100755 --- a/qpid/cpp/include/qpid/sys/windows/IntegerTypes.h +++ b/qpid/cpp/include/qpid/sys/windows/IntegerTypes.h @@ -21,8 +21,6 @@ * */ -#include <BaseTsd.h> /* Windows system types */ - typedef unsigned char uint8_t; typedef char int8_t; typedef unsigned short uint16_t; @@ -33,8 +31,6 @@ typedef unsigned __int64 uint64_t; typedef __int64 int64_t; // Visual Studio doesn't define other common types, so set them up here too. -typedef int pid_t; -typedef SSIZE_T ssize_t; typedef unsigned int uint; #endif /*!QPID_SYS_WINDOWS_INTEGERTYPES_H*/ diff --git a/qpid/cpp/src/CMakeLists.txt b/qpid/cpp/src/CMakeLists.txt index 49ebe1e31e..899dd5568c 100644 --- a/qpid/cpp/src/CMakeLists.txt +++ b/qpid/cpp/src/CMakeLists.txt @@ -155,6 +155,7 @@ set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${COMPILER_FLAGS} ${WARNING_FLAGS}") # TODO: Not all these libs are needed everywhere: # Linux only uses filesystem program_options unit_test_framework # (which itself uses regex). +# Boost.system is sometimes needed; it's handled separately, below. find_package(Boost 1.33 REQUIRED COMPONENTS filesystem program_options date_time thread regex unit_test_framework) @@ -162,6 +163,16 @@ if(NOT Boost_FOUND) message(FATAL_ERROR "Boost C++ libraries not found. Please install or try setting BOOST_ROOT") endif(NOT Boost_FOUND) +# Boost.system was introduced at Boost 1.35; it's needed secondarily by other +# Boost libs Qpid needs, so be sure it's there. +if (NOT Boost_VERSION LESS 103500) + find_package(Boost COMPONENTS system) + if (NOT Boost_SYSTEM_LIBRARY) + set(Boost_SYSTEM_LIBRARY boost_system) + endif (NOT Boost_SYSTEM_LIBRARY) + +endif (NOT Boost_VERSION LESS 103500) + # Versions of cmake pre 2.6 don't set the Boost_*_LIBRARY variables correctly # these values are correct for Linux if (NOT Boost_PROGRAM_OPTIONS_LIBRARY) @@ -180,6 +191,10 @@ if (NOT Boost_REGEX_LIBRARY) set(Boost_REGEX_LIBRARY boost_regex) endif (NOT Boost_REGEX_LIBRARY) +if (NOT Boost_SYSTEM_LIBRARY) + set(Boost_SYSTEM_LIBRARY boost_system) +endif (NOT Boost_SYSTEM_LIBRARY) + # The Windows install also wants the Boost DLLs and headers that the release # is built with. The DLLs enable everything to run, and the headers ensure # that users building Qpid C++ client programs can compile (the C++ API @@ -412,17 +427,25 @@ if (CMAKE_SYSTEM_NAME STREQUAL Windows) qpid/sys/windows/Thread.cpp qpid/sys/windows/Time.cpp qpid/sys/windows/uuid.cpp + ${sslcommon_windows_SOURCES} ) set (qpidcommon_platform_LIBS - rpcrt4 ws2_32 + ${windows_ssl_libs} rpcrt4 ws2_32 ) set (qpidbroker_platform_SOURCES qpid/broker/windows/BrokerDefaults.cpp qpid/broker/windows/SaslAuthenticator.cpp + ${sslbroker_windows_SOURCES} + ) + set (qpidbroker_platform_LIBS + ${windows_ssl_libs} ) - set (qpidclient_platform_SOURCES qpid/client/windows/SaslFactory.cpp + ${sslclient_windows_SOURCES} + ) + set (qpidclient_platform_LIBS + ${windows_ssl_libs} ) set (qpidd_platform_SOURCES @@ -611,6 +634,7 @@ set (qpidclient_SOURCES qpid/client/SubscriptionImpl.cpp qpid/client/SubscriptionManager.cpp qpid/client/SubscriptionManagerImpl.cpp + qpid/client/TCPConnector.cpp qpid/messaging/Address.cpp qpid/messaging/Connection.cpp qpid/messaging/ConnectionImpl.h @@ -651,7 +675,7 @@ set (qpidclient_SOURCES ) add_library (qpidclient SHARED ${qpidclient_SOURCES}) -target_link_libraries (qpidclient qpidcommon) +target_link_libraries (qpidclient qpidcommon ${qpidclient_platform_LIBS}) set_target_properties (qpidclient PROPERTIES VERSION ${qpidc_version}) install (TARGETS qpidclient DESTINATION ${QPID_INSTALL_LIBDIR} @@ -669,6 +693,15 @@ if (WIN32) COMPONENT ${QPID_COMPONENT_CLIENT}) endif (WIN32) +if (WIN32) + set(AMQP_WCF_DIR ${qpid-cpp_SOURCE_DIR}/../wcf) + set(DTC_PLUGIN_SOURCE ${AMQP_WCF_DIR}/src/Apache/Qpid/DtcPlugin/DtcPlugin.cpp) + if (EXISTS ${DTC_PLUGIN_SOURCE}) + add_library (qpidxarm SHARED ${DTC_PLUGIN_SOURCE}) + target_link_libraries (qpidxarm qpidclient qpidcommon) + endif (EXISTS ${DTC_PLUGIN_SOURCE}) +endif (WIN32) + set (qpidbroker_SOURCES ${mgen_broker_cpp} ${qpidbroker_platform_SOURCES} @@ -737,7 +770,7 @@ set (qpidbroker_SOURCES qpid/sys/TCPIOPlugin.cpp ) add_library (qpidbroker SHARED ${qpidbroker_SOURCES}) -target_link_libraries (qpidbroker qpidcommon) +target_link_libraries (qpidbroker qpidcommon ${qpidbroker_platform_LIBS}) set_target_properties (qpidbroker PROPERTIES VERSION ${qpidc_version}) if (MSVC) set_target_properties (qpidbroker PROPERTIES COMPILE_FLAGS /wd4290) diff --git a/qpid/cpp/src/Makefile.am b/qpid/cpp/src/Makefile.am index 968bd7ca7a..4979aaf926 100644 --- a/qpid/cpp/src/Makefile.am +++ b/qpid/cpp/src/Makefile.am @@ -154,6 +154,7 @@ libqpidcommon_la_SOURCES += \ qpid/sys/posix/Fork.cpp \ qpid/sys/posix/StrError.cpp \ qpid/sys/posix/PollableCondition.cpp \ + qpid/sys/posix/PidFile.h \ qpid/sys/posix/PipeHandle.cpp \ qpid/log/posix/SinkOptions.h \ qpid/sys/posix/Fork.h @@ -444,7 +445,6 @@ libqpidcommon_la_SOURCES += \ qpid/sys/DispatchHandle.h \ qpid/sys/Dispatcher.cpp \ qpid/sys/Dispatcher.h \ - qpid/sys/ExceptionHolder.h \ qpid/sys/FileSysDir.h \ qpid/sys/Fork.h \ qpid/sys/LockFile.h \ @@ -691,6 +691,8 @@ libqpidclient_la_SOURCES = \ qpid/client/SubscriptionManager.cpp \ qpid/client/SubscriptionManagerImpl.cpp \ qpid/client/SubscriptionManagerImpl.h \ + qpid/client/TCPConnector.cpp \ + qpid/client/TCPConnector.h \ qpid/messaging/Address.cpp \ qpid/messaging/Connection.cpp \ qpid/messaging/ListContent.cpp \ @@ -789,6 +791,7 @@ nobase_include_HEADERS += \ ../include/qpid/management/ManagementEvent.h \ ../include/qpid/management/ManagementObject.h \ ../include/qpid/sys/Condition.h \ + ../include/qpid/sys/ExceptionHolder.h \ ../include/qpid/sys/IOHandle.h \ ../include/qpid/sys/IntegerTypes.h \ ../include/qpid/sys/Monitor.h \ diff --git a/qpid/cpp/src/cluster.cmake b/qpid/cpp/src/cluster.cmake index 910d75e768..6552c39f12 100644 --- a/qpid/cpp/src/cluster.cmake +++ b/qpid/cpp/src/cluster.cmake @@ -109,6 +109,7 @@ if (BUILD_CLUSTER) qpid/cluster/ExpiryPolicy.cpp qpid/cluster/FailoverExchange.cpp qpid/cluster/FailoverExchange.h + qpid/cluster/UpdateExchange.cpp qpid/cluster/UpdateExchange.h qpid/cluster/UpdateReceiver.h qpid/cluster/LockedConnectionMap.h @@ -133,7 +134,7 @@ if (BUILD_CLUSTER) ) add_library (cluster MODULE ${cluster_SOURCES}) - target_link_libraries (cluster ${LIBCPG} ${CMAN_LIB} qpidbroker qpidclient ${Boost_FILESYSTEM_LIBRARY}) + target_link_libraries (cluster ${LIBCPG} ${CMAN_LIB} qpidbroker qpidclient ${Boost_FILESYSTEM_LIBRARY} ${Boost_SYSTEM_LIBRARY}) set_target_properties (cluster PROPERTIES PREFIX "") # Create a second shared library for linking with test executables, diff --git a/qpid/cpp/src/cluster.mk b/qpid/cpp/src/cluster.mk index 9da6578df0..081889130e 100644 --- a/qpid/cpp/src/cluster.mk +++ b/qpid/cpp/src/cluster.mk @@ -69,6 +69,7 @@ cluster_la_SOURCES = \ qpid/cluster/FailoverExchange.cpp \ qpid/cluster/FailoverExchange.h \ qpid/cluster/UpdateExchange.h \ + qpid/cluster/UpdateExchange.cpp \ qpid/cluster/UpdateReceiver.h \ qpid/cluster/LockedConnectionMap.h \ qpid/cluster/Multicaster.cpp \ diff --git a/qpid/cpp/src/qpid/DataDir.cpp b/qpid/cpp/src/qpid/DataDir.cpp index 0ee30709af..ad732052ab 100644 --- a/qpid/cpp/src/qpid/DataDir.cpp +++ b/qpid/cpp/src/qpid/DataDir.cpp @@ -22,6 +22,7 @@ #include "qpid/DataDir.h" #include "qpid/log/Statement.h" #include "qpid/sys/FileSysDir.h" +#include "qpid/sys/LockFile.h" namespace qpid { diff --git a/qpid/cpp/src/qpid/DataDir.h b/qpid/cpp/src/qpid/DataDir.h index dfdd498cbc..828299f3ba 100644 --- a/qpid/cpp/src/qpid/DataDir.h +++ b/qpid/cpp/src/qpid/DataDir.h @@ -23,11 +23,14 @@ #include <string> #include <memory> -#include "qpid/sys/LockFile.h" #include "qpid/CommonImportExport.h" namespace qpid { + namespace sys { + class LockFile; + } + /** * DataDir class. */ diff --git a/qpid/cpp/src/qpid/broker/Broker.cpp b/qpid/cpp/src/qpid/broker/Broker.cpp index 3c67c429a0..849bf6d1f5 100644 --- a/qpid/cpp/src/qpid/broker/Broker.cpp +++ b/qpid/cpp/src/qpid/broker/Broker.cpp @@ -154,6 +154,7 @@ Broker::Broker(const Broker::Options& conf) : queueCleaner(queues, timer), queueEvents(poller,!conf.asyncQueueEvents), recovery(true), + clusterUpdatee(false), expiryPolicy(new ExpiryPolicy), connectionCounter(conf.maxConnections), getKnownBrokers(boost::bind(&Broker::getKnownBrokersImpl, this)) diff --git a/qpid/cpp/src/qpid/broker/Broker.h b/qpid/cpp/src/qpid/broker/Broker.h index 73d5860cb3..b85aa7d96c 100644 --- a/qpid/cpp/src/qpid/broker/Broker.h +++ b/qpid/cpp/src/qpid/broker/Broker.h @@ -168,8 +168,10 @@ public: std::vector<Url> getKnownBrokersImpl(); std::string federationTag; bool recovery; + bool clusterUpdatee; boost::intrusive_ptr<ExpiryPolicy> expiryPolicy; ConnectionCounter connectionCounter; + public: virtual ~Broker(); @@ -259,6 +261,9 @@ public: void setRecovery(bool set) { recovery = set; } bool getRecovery() const { return recovery; } + void setClusterUpdatee(bool set) { clusterUpdatee = set; } + bool isClusterUpdatee() const { return clusterUpdatee; } + management::ManagementAgent* getManagementAgent() { return managementAgent.get(); } ConnectionCounter& getConnectionCounter() {return connectionCounter;} diff --git a/qpid/cpp/src/qpid/broker/Daemon.cpp b/qpid/cpp/src/qpid/broker/Daemon.cpp index e1d400e01b..b30e5f18cb 100644 --- a/qpid/cpp/src/qpid/broker/Daemon.cpp +++ b/qpid/cpp/src/qpid/broker/Daemon.cpp @@ -15,10 +15,16 @@ * limitations under the License. * */ + +/* + * TODO: Note this is really a Posix specific implementation and so should be + * refactored together with windows/QpiddBroker into a more coherent daemon driver/ + * platform specific split + */ #include "qpid/broker/Daemon.h" #include "qpid/log/Statement.h" #include "qpid/Exception.h" -#include "qpid/sys/LockFile.h" +#include "qpid/sys/posix/PidFile.h" #include <errno.h> #include <fcntl.h> @@ -31,7 +37,7 @@ namespace qpid { namespace broker { using namespace std; -using qpid::sys::LockFile; +using qpid::sys::PidFile; Daemon::Daemon(std::string _pidDir) : pidDir(_pidDir) { struct stat s; @@ -176,7 +182,7 @@ uint16_t Daemon::wait(int timeout) { // parent waits for child. */ void Daemon::ready(uint16_t port) { // child lockFile = pidFile(pidDir, port); - LockFile lf(lockFile, true); + PidFile lf(lockFile, true); /* * Write the PID to the lockfile. @@ -200,7 +206,7 @@ void Daemon::ready(uint16_t port) { // child */ pid_t Daemon::getPid(string _pidDir, uint16_t port) { string name = pidFile(_pidDir, port); - LockFile lf(name, false); + PidFile lf(name, false); pid_t pid = lf.readPid(); if (kill(pid, 0) < 0 && errno != EPERM) { unlink(name.c_str()); diff --git a/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp b/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp index 0cfa2391fd..22ec5e86a0 100644 --- a/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp +++ b/qpid/cpp/src/qpid/broker/DeliveryRecord.cpp @@ -43,7 +43,7 @@ DeliveryRecord::DeliveryRecord(const QueuedMessage& _msg, acceptExpected(!accepted), cancelled(false), completed(false), - ended(accepted), + ended(accepted && acquired), windowing(_windowing), credit(msg.payload ? msg.payload->getRequiredCredit() : _credit) {} @@ -150,6 +150,10 @@ void DeliveryRecord::acquire(DeliveryIds& results) { if (queue->acquire(msg)) { acquired = true; results.push_back(id); + if (!acceptExpected) { + if (ended) { QPID_LOG(error, "Can't dequeue ended message"); } + else { queue->dequeue(0, msg); setEnded(); } + } } else { QPID_LOG(info, "Message already acquired " << id.getValue()); } diff --git a/qpid/cpp/src/qpid/broker/Exchange.cpp b/qpid/cpp/src/qpid/broker/Exchange.cpp index 4ebe126969..8efb9ac545 100644 --- a/qpid/cpp/src/qpid/broker/Exchange.cpp +++ b/qpid/cpp/src/qpid/broker/Exchange.cpp @@ -209,8 +209,10 @@ Exchange::shared_ptr Exchange::decode(ExchangeRegistry& exchanges, Buffer& buffe buffer.getShortString(name); bool durable(buffer.getOctet()); buffer.getShortString(type); - buffer.getShortString(altName); buffer.get(args); + // For backwards compatibility on restoring exchanges from before the alt-exchange update, perform check + if (buffer.available()) + buffer.getShortString(altName); try { Exchange::shared_ptr exch = exchanges.declare(name, type, durable, args).first; @@ -228,10 +230,10 @@ void Exchange::encode(Buffer& buffer) const buffer.putShortString(name); buffer.putOctet(durable); buffer.putShortString(getType()); - buffer.putShortString(alternate.get() ? alternate->getName() : string("")); if (args.isSet(qpidSequenceCounter)) args.setInt64(std::string(qpidSequenceCounter),sequenceNo); buffer.put(args); + buffer.putShortString(alternate.get() ? alternate->getName() : string("")); } uint32_t Exchange::encodedSize() const diff --git a/qpid/cpp/src/qpid/broker/Message.cpp b/qpid/cpp/src/qpid/broker/Message.cpp index e2799b0bff..47ca7a7ae8 100644 --- a/qpid/cpp/src/qpid/broker/Message.cpp +++ b/qpid/cpp/src/qpid/broker/Message.cpp @@ -425,19 +425,4 @@ framing::FieldTable& Message::getOrInsertHeaders() return getProperties<MessageProperties>()->getApplicationHeaders(); } - -void Message::setUpdateDestination(const std::string& d) -{ - updateDestination = d; -} - - -bool Message::isUpdateMessage() -{ - return updateDestination.size() && isA<MessageTransferBody>() - && getMethod<MessageTransferBody>()->getDestination() == updateDestination; -} - -std::string Message::updateDestination; - }} // namespace qpid::broker diff --git a/qpid/cpp/src/qpid/broker/Message.h b/qpid/cpp/src/qpid/broker/Message.h index 3894960c95..375fa9ce26 100644 --- a/qpid/cpp/src/qpid/broker/Message.h +++ b/qpid/cpp/src/qpid/broker/Message.h @@ -104,6 +104,10 @@ public: return frames.as<T>(); } + template <class T> T* getMethod() { + return frames.as<T>(); + } + template <class T> bool isA() const { return frames.isA<T>(); } @@ -157,9 +161,6 @@ public: void setDequeueCompleteCallback(MessageCallback& cb); void resetDequeueCompleteCallback(); - bool isUpdateMessage(); - static void setUpdateDestination(const std::string&); - private: typedef std::map<const Queue*,boost::intrusive_ptr<Message> > Replacement; @@ -190,7 +191,6 @@ public: MessageCallback* dequeueCallback; uint32_t requiredCredit; - static std::string updateDestination; }; }} diff --git a/qpid/cpp/src/qpid/broker/Queue.cpp b/qpid/cpp/src/qpid/broker/Queue.cpp index 780c254a56..f4231f2397 100644 --- a/qpid/cpp/src/qpid/broker/Queue.cpp +++ b/qpid/cpp/src/qpid/broker/Queue.cpp @@ -598,7 +598,7 @@ void Queue::push(boost::intrusive_ptr<Message>& msg, bool isRecovery){ string key = ft->getAsString(qpidVQMatchProperty); i = lvq.find(key); - if (i == lvq.end() || msg->isUpdateMessage()){ + if (i == lvq.end() || (broker && broker->isClusterUpdatee())) { messages.push_back(qm); listeners.populate(copy); lvq[key] = msg; @@ -940,11 +940,11 @@ void Queue::setPersistenceId(uint64_t _persistenceId) const void Queue::encode(Buffer& buffer) const { buffer.putShortString(name); - buffer.putShortString(alternateExchange.get() ? alternateExchange->getName() : std::string("")); buffer.put(settings); if (policy.get()) { buffer.put(*policy); } + buffer.putShortString(alternateExchange.get() ? alternateExchange->getName() : std::string("")); } uint32_t Queue::encodedSize() const @@ -959,15 +959,17 @@ Queue::shared_ptr Queue::decode ( QueueRegistry& queues, Buffer& buffer, bool re { string name; buffer.getShortString(name); - string altExch; - buffer.getShortString(altExch); std::pair<Queue::shared_ptr, bool> result = queues.declare(name, true); - result.first->alternateExchangeName.assign(altExch); buffer.get(result.first->settings); result.first->configure(result.first->settings, recovering ); if (result.first->policy.get() && buffer.available() >= result.first->policy->encodedSize()) { buffer.get ( *(result.first->policy) ); } + if (buffer.available()) { + string altExch; + buffer.getShortString(altExch); + result.first->alternateExchangeName.assign(altExch); + } return result.first; } diff --git a/qpid/cpp/src/qpid/broker/RecoverableMessage.h b/qpid/cpp/src/qpid/broker/RecoverableMessage.h index f755fdf727..c98857ceb0 100644 --- a/qpid/cpp/src/qpid/broker/RecoverableMessage.h +++ b/qpid/cpp/src/qpid/broker/RecoverableMessage.h @@ -37,6 +37,7 @@ class RecoverableMessage public: typedef boost::shared_ptr<RecoverableMessage> shared_ptr; virtual void setPersistenceId(uint64_t id) = 0; + virtual void setRedelivered() = 0; /** * Used by store to determine whether to load content on recovery * or let message load its own content as and when it requires it. diff --git a/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp b/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp index 0369fcd71d..12ac2d2bfd 100644 --- a/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp +++ b/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp @@ -48,6 +48,7 @@ public: RecoverableMessageImpl(const intrusive_ptr<Message>& _msg, uint64_t _stagingThreshold); ~RecoverableMessageImpl() {}; void setPersistenceId(uint64_t id); + void setRedelivered(); bool loadContent(uint64_t available); void decodeContent(framing::Buffer& buffer); void recover(Queue::shared_ptr queue); @@ -187,6 +188,11 @@ void RecoverableMessageImpl::setPersistenceId(uint64_t id) msg->setPersistenceId(id); } +void RecoverableMessageImpl::setRedelivered() +{ + msg->redeliver(); +} + void RecoverableQueueImpl::recover(RecoverableMessage::shared_ptr msg) { dynamic_pointer_cast<RecoverableMessageImpl>(msg)->recover(queue); diff --git a/qpid/cpp/src/qpid/broker/SemanticState.cpp b/qpid/cpp/src/qpid/broker/SemanticState.cpp index 9680ada936..e9b6aad967 100644 --- a/qpid/cpp/src/qpid/broker/SemanticState.cpp +++ b/qpid/cpp/src/qpid/broker/SemanticState.cpp @@ -70,7 +70,8 @@ SemanticState::SemanticState(DeliveryAdapter& da, SessionContext& ss) tagGenerator("sgen"), dtxSelected(false), authMsg(getSession().getBroker().getOptions().auth && !getSession().getConnection().isFederationLink()), - userID(getSession().getConnection().getUserId()) + userID(getSession().getConnection().getUserId()), + defaultRealm(getSession().getBroker().getOptions().realm) { acl = getSession().getBroker().getAcl(); } @@ -311,7 +312,7 @@ bool SemanticState::ConsumerImpl::deliver(QueuedMessage& msg) bool sync = syncFrequency && ++deliveryCount >= syncFrequency; if (sync) deliveryCount = 0;//reset parent->deliver(record, sync); - if (!ackExpected) record.setEnded();//allows message to be released now its been delivered + if (!ackExpected && acquire) record.setEnded();//allows message to be released now its been delivered if (windowing || ackExpected || !acquire) { parent->record(record); } @@ -429,7 +430,7 @@ void SemanticState::route(intrusive_ptr<Message> msg, Deliverable& strategy) { std::string id = msg->hasProperties<MessageProperties>() ? msg->getProperties<MessageProperties>()->getUserId() : nullstring; - if (authMsg && !id.empty() && id != userID ) + if (authMsg && !id.empty() && id != userID && id.append("@").append(defaultRealm) != userID) { QPID_LOG(debug, "authorised user id : " << userID << " but user id in message declared as " << id); throw UnauthorizedAccessException(QPID_MSG("authorised user id : " << userID << " but user id in message declared as " << id)); @@ -438,7 +439,7 @@ void SemanticState::route(intrusive_ptr<Message> msg, Deliverable& strategy) { if (acl && acl->doTransferAcl()) { if (!acl->authorise(getSession().getConnection().getUserId(),acl::ACT_PUBLISH,acl::OBJ_EXCHANGE,exchangeName, msg->getRoutingKey() )) - throw NotAllowedException(QPID_MSG(getSession().getConnection().getUserId() << " cannot publish to " << + throw NotAllowedException(QPID_MSG(userID << " cannot publish to " << exchangeName << " with routing-key " << msg->getRoutingKey())); } diff --git a/qpid/cpp/src/qpid/broker/SemanticState.h b/qpid/cpp/src/qpid/broker/SemanticState.h index 99f793c1fc..e5e3f909f1 100644 --- a/qpid/cpp/src/qpid/broker/SemanticState.h +++ b/qpid/cpp/src/qpid/broker/SemanticState.h @@ -156,6 +156,7 @@ class SemanticState : private boost::noncopyable { AclModule* acl; const bool authMsg; const string userID; + const string defaultRealm; void route(boost::intrusive_ptr<Message> msg, Deliverable& strategy); void checkDtxTimeout(); diff --git a/qpid/cpp/src/qpid/client/Connector.cpp b/qpid/cpp/src/qpid/client/Connector.cpp index ad60c9d7e1..2c4feffdcf 100644 --- a/qpid/cpp/src/qpid/client/Connector.cpp +++ b/qpid/cpp/src/qpid/client/Connector.cpp @@ -18,9 +18,9 @@ * under the License. * */ + #include "qpid/client/Connector.h" -#include "qpid/client/Bounds.h" #include "qpid/client/ConnectionImpl.h" #include "qpid/client/ConnectionSettings.h" #include "qpid/log/Statement.h" @@ -35,10 +35,8 @@ #include <iostream> #include <map> -#include <deque> #include <boost/bind.hpp> #include <boost/format.hpp> -#include <boost/weak_ptr.hpp> namespace qpid { namespace client { @@ -81,353 +79,4 @@ void Connector::activateSecurityLayer(std::auto_ptr<qpid::sys::SecurityLayer>) { } -class TCPConnector : public Connector, public sys::Codec, private sys::Runnable -{ - typedef std::deque<framing::AMQFrame> Frames; - struct Buff; - - const uint16_t maxFrameSize; - - sys::Mutex lock; - Frames frames; // Outgoing frame queue - size_t lastEof; // Position after last EOF in frames - uint64_t currentSize; - Bounds* bounds; - - framing::ProtocolVersion version; - bool initiated; - bool closed; - bool joined; - - sys::ShutdownHandler* shutdownHandler; - framing::InputHandler* input; - framing::InitiationHandler* initialiser; - framing::OutputHandler* output; - - sys::Thread receiver; - - sys::Socket socket; - - sys::AsynchIO* aio; - std::string identifier; - boost::shared_ptr<sys::Poller> poller; - std::auto_ptr<qpid::sys::SecurityLayer> securityLayer; - - ~TCPConnector(); - - void run(); - void handleClosed(); - bool closeInternal(); - - void connected(const Socket&); - void connectFailed(const std::string& msg); - bool readbuff(qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*); - void writebuff(qpid::sys::AsynchIO&); - void writeDataBlock(const framing::AMQDataBlock& data); - void eof(qpid::sys::AsynchIO&); - - boost::weak_ptr<ConnectionImpl> impl; - - void connect(const std::string& host, int port); - void close(); - void send(framing::AMQFrame& frame); - void abort(); - - void setInputHandler(framing::InputHandler* handler); - void setShutdownHandler(sys::ShutdownHandler* handler); - sys::ShutdownHandler* getShutdownHandler() const; - framing::OutputHandler* getOutputHandler(); - const std::string& getIdentifier() const; - void activateSecurityLayer(std::auto_ptr<qpid::sys::SecurityLayer>); - - size_t decode(const char* buffer, size_t size); - size_t encode(const char* buffer, size_t size); - bool canEncode(); - -public: - TCPConnector(framing::ProtocolVersion pVersion, - const ConnectionSettings&, - ConnectionImpl*); - unsigned int getSSF() { return 0; } -}; - -// Static constructor which registers connector here -namespace { - Connector* create(framing::ProtocolVersion v, const ConnectionSettings& s, ConnectionImpl* c) { - return new TCPConnector(v, s, c); - } - - struct StaticInit { - StaticInit() { - Connector::registerFactory("tcp", &create); - }; - } init; -} - -struct TCPConnector::Buff : public AsynchIO::BufferBase { - Buff(size_t size) : AsynchIO::BufferBase(new char[size], size) {} - ~Buff() { delete [] bytes;} -}; - -TCPConnector::TCPConnector(ProtocolVersion ver, - const ConnectionSettings& settings, - ConnectionImpl* cimpl) - : maxFrameSize(settings.maxFrameSize), - lastEof(0), - currentSize(0), - bounds(cimpl), - version(ver), - initiated(false), - closed(true), - joined(true), - shutdownHandler(0), - aio(0), - impl(cimpl->shared_from_this()) -{ - QPID_LOG(debug, "TCPConnector created for " << version.toString()); - settings.configureSocket(socket); -} - -TCPConnector::~TCPConnector() { - close(); -} - -void TCPConnector::connect(const std::string& host, int port){ - Mutex::ScopedLock l(lock); - assert(closed); - assert(joined); - poller = Poller::shared_ptr(new Poller); - AsynchConnector::create(socket, - poller, - host, port, - boost::bind(&TCPConnector::connected, this, _1), - boost::bind(&TCPConnector::connectFailed, this, _3)); - closed = false; - joined = false; - receiver = Thread(this); -} - -void TCPConnector::connected(const Socket&) { - aio = AsynchIO::create(socket, - boost::bind(&TCPConnector::readbuff, this, _1, _2), - boost::bind(&TCPConnector::eof, this, _1), - boost::bind(&TCPConnector::eof, this, _1), - 0, // closed - 0, // nobuffs - boost::bind(&TCPConnector::writebuff, this, _1)); - for (int i = 0; i < 32; i++) { - aio->queueReadBuffer(new Buff(maxFrameSize)); - } - aio->start(poller); - - identifier = str(format("[%1% %2%]") % socket.getLocalPort() % socket.getPeerAddress()); - ProtocolInitiation init(version); - writeDataBlock(init); -} - -void TCPConnector::connectFailed(const std::string& msg) { - QPID_LOG(warning, "Connecting failed: " << msg); - closed = true; - poller->shutdown(); - closeInternal(); - if (shutdownHandler) - shutdownHandler->shutdown(); -} - -bool TCPConnector::closeInternal() { - bool ret; - { - Mutex::ScopedLock l(lock); - ret = !closed; - if (!closed) { - closed = true; - aio->queueForDeletion(); - poller->shutdown(); - } - if (joined || receiver.id() == Thread::current().id()) { - return ret; - } - joined = true; - } - receiver.join(); - return ret; -} - -void TCPConnector::close() { - closeInternal(); -} - -void TCPConnector::abort() { - // Can't abort a closed connection - if (!closed) { - if (aio) { - // Established connection - aio->requestCallback(boost::bind(&TCPConnector::eof, this, _1)); - } else { - // We're still connecting - connectFailed("Connection timedout"); - } - } -} - -void TCPConnector::setInputHandler(InputHandler* handler){ - input = handler; -} - -void TCPConnector::setShutdownHandler(ShutdownHandler* handler){ - shutdownHandler = handler; -} - -OutputHandler* TCPConnector::getOutputHandler() { - return this; -} - -sys::ShutdownHandler* TCPConnector::getShutdownHandler() const { - return shutdownHandler; -} - -const std::string& TCPConnector::getIdentifier() const { - return identifier; -} - -void TCPConnector::send(AMQFrame& frame) { - Mutex::ScopedLock l(lock); - frames.push_back(frame); - //only ask to write if this is the end of a frameset or if we - //already have a buffers worth of data - currentSize += frame.encodedSize(); - bool notifyWrite = false; - if (frame.getEof()) { - lastEof = frames.size(); - notifyWrite = true; - } else { - notifyWrite = (currentSize >= maxFrameSize); - } - if (notifyWrite && !closed) aio->notifyPendingWrite(); -} - -void TCPConnector::handleClosed() { - if (closeInternal() && shutdownHandler) - shutdownHandler->shutdown(); -} - -void TCPConnector::writebuff(AsynchIO& /*aio*/) -{ - Codec* codec = securityLayer.get() ? (Codec*) securityLayer.get() : (Codec*) this; - if (codec->canEncode()) { - std::auto_ptr<AsynchIO::BufferBase> buffer = std::auto_ptr<AsynchIO::BufferBase>(aio->getQueuedBuffer()); - if (!buffer.get()) buffer = std::auto_ptr<AsynchIO::BufferBase>(new Buff(maxFrameSize)); - - size_t encoded = codec->encode(buffer->bytes, buffer->byteCount); - - buffer->dataStart = 0; - buffer->dataCount = encoded; - aio->queueWrite(buffer.release()); - } -} - -// Called in IO thread. -bool TCPConnector::canEncode() -{ - Mutex::ScopedLock l(lock); - //have at least one full frameset or a whole buffers worth of data - return lastEof || currentSize >= maxFrameSize; -} - -// Called in IO thread. -size_t TCPConnector::encode(const char* buffer, size_t size) -{ - framing::Buffer out(const_cast<char*>(buffer), size); - size_t bytesWritten(0); - { - Mutex::ScopedLock l(lock); - while (!frames.empty() && out.available() >= frames.front().encodedSize() ) { - frames.front().encode(out); - QPID_LOG(trace, "SENT " << identifier << ": " << frames.front()); - frames.pop_front(); - if (lastEof) --lastEof; - } - bytesWritten = size - out.available(); - currentSize -= bytesWritten; - } - if (bounds) bounds->reduce(bytesWritten); - return bytesWritten; -} - -bool TCPConnector::readbuff(AsynchIO& aio, AsynchIO::BufferBase* buff) -{ - Codec* codec = securityLayer.get() ? (Codec*) securityLayer.get() : (Codec*) this; - int32_t decoded = codec->decode(buff->bytes+buff->dataStart, buff->dataCount); - // TODO: unreading needs to go away, and when we can cope - // with multiple sub-buffers in the general buffer scheme, it will - if (decoded < 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); - } - return true; -} - -size_t TCPConnector::decode(const char* buffer, size_t size) -{ - framing::Buffer in(const_cast<char*>(buffer), size); - if (!initiated) { - framing::ProtocolInitiation protocolInit; - if (protocolInit.decode(in)) { - QPID_LOG(debug, "RECV " << identifier << " INIT(" << protocolInit << ")"); - if(!(protocolInit==version)){ - throw Exception(QPID_MSG("Unsupported version: " << protocolInit - << " supported version " << version)); - } - } - initiated = true; - } - AMQFrame frame; - while(frame.decode(in)){ - QPID_LOG(trace, "RECV " << identifier << ": " << frame); - input->received(frame); - } - return size - in.available(); -} - -void TCPConnector::writeDataBlock(const AMQDataBlock& data) { - AsynchIO::BufferBase* buff = new Buff(maxFrameSize); - framing::Buffer out(buff->bytes, buff->byteCount); - data.encode(out); - buff->dataCount = data.encodedSize(); - aio->queueWrite(buff); -} - -void TCPConnector::eof(AsynchIO&) { - handleClosed(); -} - -void TCPConnector::run() { - // Keep the connection impl in memory until run() completes. - boost::shared_ptr<ConnectionImpl> protect = impl.lock(); - assert(protect); - try { - Dispatcher d(poller); - - d.run(); - } catch (const std::exception& e) { - QPID_LOG(error, QPID_MSG("FAIL " << identifier << ": " << e.what())); - handleClosed(); - } - try { - socket.close(); - } catch (const std::exception&) {} -} - -void TCPConnector::activateSecurityLayer(std::auto_ptr<qpid::sys::SecurityLayer> sl) -{ - securityLayer = sl; - securityLayer->init(this); -} - - }} // namespace qpid::client diff --git a/qpid/cpp/src/qpid/client/MessageReplayTracker.cpp b/qpid/cpp/src/qpid/client/MessageReplayTracker.cpp index 0782f6e63d..079fb1167a 100644 --- a/qpid/cpp/src/qpid/client/MessageReplayTracker.cpp +++ b/qpid/cpp/src/qpid/client/MessageReplayTracker.cpp @@ -43,9 +43,7 @@ void MessageReplayTracker::init(AsyncSession s) void MessageReplayTracker::replay(AsyncSession s) { session = s; - std::list<ReplayRecord> copy; - buffer.swap(copy); - std::for_each(copy.begin(), copy.end(), boost::bind(&ReplayRecord::send, _1, boost::ref(*this))); + std::for_each(buffer.begin(), buffer.end(), boost::bind(&ReplayRecord::send, _1, boost::ref(*this))); session.flush(); count = 0; } diff --git a/qpid/cpp/src/qpid/client/TCPConnector.cpp b/qpid/cpp/src/qpid/client/TCPConnector.cpp new file mode 100644 index 0000000000..1a6e51d54d --- /dev/null +++ b/qpid/cpp/src/qpid/client/TCPConnector.cpp @@ -0,0 +1,327 @@ +/* + * + * 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/client/TCPConnector.h" + +#include "qpid/client/ConnectionImpl.h" +#include "qpid/client/ConnectionSettings.h" +#include "qpid/log/Statement.h" +#include "qpid/sys/Codec.h" +#include "qpid/sys/Time.h" +#include "qpid/framing/AMQFrame.h" +#include "qpid/sys/AsynchIO.h" +#include "qpid/sys/Dispatcher.h" +#include "qpid/sys/Poller.h" +#include "qpid/sys/SecurityLayer.h" +#include "qpid/Msg.h" + +#include <iostream> +#include <boost/bind.hpp> +#include <boost/format.hpp> + +namespace qpid { +namespace client { + +using namespace qpid::sys; +using namespace qpid::framing; +using boost::format; +using boost::str; + +// Static constructor which registers connector here +namespace { + Connector* create(framing::ProtocolVersion v, const ConnectionSettings& s, ConnectionImpl* c) { + return new TCPConnector(v, s, c); + } + + struct StaticInit { + StaticInit() { + Connector::registerFactory("tcp", &create); + }; + } init; +} + +struct TCPConnector::Buff : public AsynchIO::BufferBase { + Buff(size_t size) : AsynchIO::BufferBase(new char[size], size) {} + ~Buff() { delete [] bytes;} +}; + +TCPConnector::TCPConnector(ProtocolVersion ver, + const ConnectionSettings& settings, + ConnectionImpl* cimpl) + : maxFrameSize(settings.maxFrameSize), + lastEof(0), + currentSize(0), + bounds(cimpl), + version(ver), + initiated(false), + closed(true), + joined(true), + shutdownHandler(0), + aio(0), + impl(cimpl->shared_from_this()) +{ + QPID_LOG(debug, "TCPConnector created for " << version.toString()); + settings.configureSocket(socket); +} + +TCPConnector::~TCPConnector() { + close(); +} + +void TCPConnector::connect(const std::string& host, int port) { + Mutex::ScopedLock l(lock); + assert(closed); + assert(joined); + poller = Poller::shared_ptr(new Poller); + AsynchConnector::create(socket, + poller, + host, port, + boost::bind(&TCPConnector::connected, this, _1), + boost::bind(&TCPConnector::connectFailed, this, _3)); + closed = false; + joined = false; + receiver = Thread(this); +} + +void TCPConnector::connected(const Socket&) { + aio = AsynchIO::create(socket, + boost::bind(&TCPConnector::readbuff, this, _1, _2), + boost::bind(&TCPConnector::eof, this, _1), + boost::bind(&TCPConnector::eof, this, _1), + 0, // closed + 0, // nobuffs + boost::bind(&TCPConnector::writebuff, this, _1)); + for (int i = 0; i < 32; i++) { + aio->queueReadBuffer(new Buff(maxFrameSize)); + } + aio->start(poller); + + identifier = str(format("[%1% %2%]") % socket.getLocalPort() % socket.getPeerAddress()); + ProtocolInitiation init(version); + writeDataBlock(init); +} + +void TCPConnector::connectFailed(const std::string& msg) { + QPID_LOG(warning, "Connecting failed: " << msg); + closed = true; + poller->shutdown(); + closeInternal(); + if (shutdownHandler) + shutdownHandler->shutdown(); +} + +bool TCPConnector::closeInternal() { + bool ret; + { + Mutex::ScopedLock l(lock); + ret = !closed; + if (!closed) { + closed = true; + aio->queueForDeletion(); + poller->shutdown(); + } + if (joined || receiver.id() == Thread::current().id()) { + return ret; + } + joined = true; + } + receiver.join(); + return ret; +} + +void TCPConnector::close() { + closeInternal(); +} + +void TCPConnector::abort() { + // Can't abort a closed connection + if (!closed) { + if (aio) { + // Established connection + aio->requestCallback(boost::bind(&TCPConnector::eof, this, _1)); + } else { + // We're still connecting + connectFailed("Connection timedout"); + } + } +} + +void TCPConnector::setInputHandler(InputHandler* handler){ + input = handler; +} + +void TCPConnector::setShutdownHandler(ShutdownHandler* handler){ + shutdownHandler = handler; +} + +OutputHandler* TCPConnector::getOutputHandler() { + return this; +} + +sys::ShutdownHandler* TCPConnector::getShutdownHandler() const { + return shutdownHandler; +} + +const std::string& TCPConnector::getIdentifier() const { + return identifier; +} + +void TCPConnector::send(AMQFrame& frame) { + Mutex::ScopedLock l(lock); + frames.push_back(frame); + //only ask to write if this is the end of a frameset or if we + //already have a buffers worth of data + currentSize += frame.encodedSize(); + bool notifyWrite = false; + if (frame.getEof()) { + lastEof = frames.size(); + notifyWrite = true; + } else { + notifyWrite = (currentSize >= maxFrameSize); + } + if (notifyWrite && !closed) aio->notifyPendingWrite(); +} + +void TCPConnector::handleClosed() { + if (closeInternal() && shutdownHandler) + shutdownHandler->shutdown(); +} + +void TCPConnector::writebuff(AsynchIO& /*aio*/) +{ + Codec* codec = securityLayer.get() ? (Codec*) securityLayer.get() : (Codec*) this; + if (codec->canEncode()) { + std::auto_ptr<AsynchIO::BufferBase> buffer = std::auto_ptr<AsynchIO::BufferBase>(aio->getQueuedBuffer()); + if (!buffer.get()) buffer = std::auto_ptr<AsynchIO::BufferBase>(new Buff(maxFrameSize)); + + size_t encoded = codec->encode(buffer->bytes, buffer->byteCount); + + buffer->dataStart = 0; + buffer->dataCount = encoded; + aio->queueWrite(buffer.release()); + } +} + +// Called in IO thread. +bool TCPConnector::canEncode() +{ + Mutex::ScopedLock l(lock); + //have at least one full frameset or a whole buffers worth of data + return lastEof || currentSize >= maxFrameSize; +} + +// Called in IO thread. +size_t TCPConnector::encode(const char* buffer, size_t size) +{ + framing::Buffer out(const_cast<char*>(buffer), size); + size_t bytesWritten(0); + { + Mutex::ScopedLock l(lock); + while (!frames.empty() && out.available() >= frames.front().encodedSize() ) { + frames.front().encode(out); + QPID_LOG(trace, "SENT " << identifier << ": " << frames.front()); + frames.pop_front(); + if (lastEof) --lastEof; + } + bytesWritten = size - out.available(); + currentSize -= bytesWritten; + } + if (bounds) bounds->reduce(bytesWritten); + return bytesWritten; +} + +bool TCPConnector::readbuff(AsynchIO& aio, AsynchIO::BufferBase* buff) +{ + Codec* codec = securityLayer.get() ? (Codec*) securityLayer.get() : (Codec*) this; + int32_t decoded = codec->decode(buff->bytes+buff->dataStart, buff->dataCount); + // TODO: unreading needs to go away, and when we can cope + // with multiple sub-buffers in the general buffer scheme, it will + if (decoded < 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); + } + return true; +} + +size_t TCPConnector::decode(const char* buffer, size_t size) +{ + framing::Buffer in(const_cast<char*>(buffer), size); + if (!initiated) { + framing::ProtocolInitiation protocolInit; + if (protocolInit.decode(in)) { + QPID_LOG(debug, "RECV " << identifier << " INIT(" << protocolInit << ")"); + if(!(protocolInit==version)){ + throw Exception(QPID_MSG("Unsupported version: " << protocolInit + << " supported version " << version)); + } + } + initiated = true; + } + AMQFrame frame; + while(frame.decode(in)){ + QPID_LOG(trace, "RECV " << identifier << ": " << frame); + input->received(frame); + } + return size - in.available(); +} + +void TCPConnector::writeDataBlock(const AMQDataBlock& data) { + AsynchIO::BufferBase* buff = new Buff(maxFrameSize); + framing::Buffer out(buff->bytes, buff->byteCount); + data.encode(out); + buff->dataCount = data.encodedSize(); + aio->queueWrite(buff); +} + +void TCPConnector::eof(AsynchIO&) { + handleClosed(); +} + +void TCPConnector::run() { + // Keep the connection impl in memory until run() completes. + boost::shared_ptr<ConnectionImpl> protect = impl.lock(); + assert(protect); + try { + Dispatcher d(poller); + + d.run(); + } catch (const std::exception& e) { + QPID_LOG(error, QPID_MSG("FAIL " << identifier << ": " << e.what())); + handleClosed(); + } + try { + socket.close(); + } catch (const std::exception&) {} +} + +void TCPConnector::activateSecurityLayer(std::auto_ptr<qpid::sys::SecurityLayer> sl) +{ + securityLayer = sl; + securityLayer->init(this); +} + + +}} // namespace qpid::client diff --git a/qpid/cpp/src/qpid/client/TCPConnector.h b/qpid/cpp/src/qpid/client/TCPConnector.h new file mode 100644 index 0000000000..6dc07d1f5d --- /dev/null +++ b/qpid/cpp/src/qpid/client/TCPConnector.h @@ -0,0 +1,117 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ + +#ifndef _TCPConnector_ +#define _TCPConnector_ + +#include "Connector.h" +#include "qpid/client/Bounds.h" +#include "qpid/framing/AMQFrame.h" +#include "qpid/sys/AsynchIO.h" +#include "qpid/sys/Codec.h" +#include "qpid/sys/IntegerTypes.h" +#include "qpid/sys/Mutex.h" +#include "qpid/sys/Runnable.h" +#include "qpid/sys/SecurityLayer.h" +#include "qpid/sys/Socket.h" +#include "qpid/sys/Thread.h" + +#include <boost/shared_ptr.hpp> +#include <boost/weak_ptr.hpp> +#include <deque> +#include <string> + +namespace qpid { +namespace client { + +class TCPConnector : public Connector, public sys::Codec, private sys::Runnable +{ + typedef std::deque<framing::AMQFrame> Frames; + struct Buff; + + const uint16_t maxFrameSize; + + sys::Mutex lock; + Frames frames; // Outgoing frame queue + size_t lastEof; // Position after last EOF in frames + uint64_t currentSize; + Bounds* bounds; + + framing::ProtocolVersion version; + bool initiated; + bool closed; + bool joined; + + sys::ShutdownHandler* shutdownHandler; + framing::InputHandler* input; + framing::InitiationHandler* initialiser; + framing::OutputHandler* output; + + sys::Thread receiver; + + sys::Socket socket; + + sys::AsynchIO* aio; + std::string identifier; + boost::shared_ptr<sys::Poller> poller; + std::auto_ptr<qpid::sys::SecurityLayer> securityLayer; + + ~TCPConnector(); + + void run(); + void handleClosed(); + bool closeInternal(); + + virtual void connected(const qpid::sys::Socket&); + void connectFailed(const std::string& msg); + bool readbuff(qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*); + void writebuff(qpid::sys::AsynchIO&); + void writeDataBlock(const framing::AMQDataBlock& data); + void eof(qpid::sys::AsynchIO&); + + boost::weak_ptr<ConnectionImpl> impl; + + void connect(const std::string& host, int port); + void close(); + void send(framing::AMQFrame& frame); + void abort(); + + void setInputHandler(framing::InputHandler* handler); + void setShutdownHandler(sys::ShutdownHandler* handler); + sys::ShutdownHandler* getShutdownHandler() const; + framing::OutputHandler* getOutputHandler(); + const std::string& getIdentifier() const; + void activateSecurityLayer(std::auto_ptr<qpid::sys::SecurityLayer>); + + size_t decode(const char* buffer, size_t size); + size_t encode(const char* buffer, size_t size); + bool canEncode(); + +public: + TCPConnector(framing::ProtocolVersion pVersion, + const ConnectionSettings&, + ConnectionImpl*); + unsigned int getSSF() { return 0; } +}; + +}} // namespace qpid::client + +#endif /* _TCPConnector_ */ diff --git a/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp b/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp index 2ce29a3ae4..b70e67d12f 100644 --- a/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp +++ b/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp @@ -46,7 +46,6 @@ namespace amqp0_10 { using qpid::Exception; using qpid::messaging::Address; -using qpid::messaging::Filter; using qpid::messaging::InvalidAddress; using qpid::messaging::Variant; using qpid::framing::ExchangeBoundResult; @@ -617,7 +616,7 @@ void Queue::checkCreate(qpid::client::AsyncSession& session, CheckMode mode) } else { try { sync(session).queueDeclare(arg::queue=name, arg::passive=true); - } catch (const qpid::framing::NotFoundException& e) { + } catch (const qpid::framing::NotFoundException& /*e*/) { throw InvalidAddress((boost::format("Queue %1% does not exist") % name).str()); } catch (const std::exception& e) { throw InvalidAddress(e.what()); @@ -716,7 +715,7 @@ void Exchange::checkCreate(qpid::client::AsyncSession& session, CheckMode mode) } else { try { sync(session).exchangeDeclare(arg::exchange=name, arg::passive=true); - } catch (const qpid::framing::NotFoundException& e) { + } catch (const qpid::framing::NotFoundException& /*e*/) { throw InvalidAddress((boost::format("Exchange %1% does not exist") % name).str()); } catch (const std::exception& e) { throw InvalidAddress(e.what()); diff --git a/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.h b/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.h index 46d3f96243..01c8c51595 100644 --- a/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.h +++ b/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.h @@ -31,8 +31,7 @@ class ReplyTo; } namespace messaging { -struct Address; -struct Filter; +class Address; } namespace client { diff --git a/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp b/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp index bb47288e88..c7dff05dd7 100644 --- a/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp +++ b/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp @@ -39,7 +39,6 @@ #include <boost/function.hpp> #include <boost/intrusive_ptr.hpp> -using qpid::messaging::Filter; using qpid::messaging::KeyError; using qpid::messaging::MessageImplAccess; using qpid::messaging::Sender; diff --git a/qpid/cpp/src/qpid/cluster/Cluster.cpp b/qpid/cpp/src/qpid/cluster/Cluster.cpp index f877720350..d049001eb0 100644 --- a/qpid/cpp/src/qpid/cluster/Cluster.cpp +++ b/qpid/cpp/src/qpid/cluster/Cluster.cpp @@ -619,6 +619,7 @@ void Cluster::initMapCompleted(Lock& l) { if (initMap.isUpdateNeeded()) { // Joining established cluster. broker.setRecovery(false); // Ditch my current store. + broker.setClusterUpdatee(true); state = JOINER; } else { // I can go ready. @@ -813,6 +814,7 @@ void Cluster::checkUpdateIn(Lock& l) { memberUpdate(l); mcast.mcastControl(ClusterReadyBody(ProtocolVersion(), myUrl.str()), self); state = CATCHUP; + broker.setClusterUpdatee(false); discarding = false; // ok to set, we're stalled for update. QPID_LOG(notice, *this << " update complete, starting catch-up."); deliverEventQueue.start(); diff --git a/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp b/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp index 7f0a2752b0..e4aee6730b 100644 --- a/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp +++ b/qpid/cpp/src/qpid/cluster/ClusterPlugin.cpp @@ -139,7 +139,6 @@ struct ClusterPlugin : public Plugin { broker->setConnectionFactory( boost::shared_ptr<sys::ConnectionCodec::Factory>( new ConnectionCodec::Factory(broker->getConnectionFactory(), *cluster))); - broker::Message::setUpdateDestination(UpdateClient::UPDATE); ManagementAgent* mgmt = broker->getManagementAgent(); if (mgmt) { std::auto_ptr<IdAllocator> allocator(new UpdateClientIdAllocator()); diff --git a/qpid/cpp/src/qpid/cluster/Cpg.cpp b/qpid/cpp/src/qpid/cluster/Cpg.cpp index 5efa91f11c..3ae0c970c7 100644 --- a/qpid/cpp/src/qpid/cluster/Cpg.cpp +++ b/qpid/cpp/src/qpid/cluster/Cpg.cpp @@ -39,6 +39,8 @@ namespace cluster { using namespace std; + + Cpg* Cpg::cpgFromHandle(cpg_handle_t handle) { void* cpg=0; CPG_CHECK(cpg_context_get(handle, &cpg), "Cannot get CPG instance."); @@ -46,6 +48,28 @@ Cpg* Cpg::cpgFromHandle(cpg_handle_t handle) { return reinterpret_cast<Cpg*>(cpg); } +// Applies the same retry-logic to all cpg calls that need it. +void Cpg::callCpg ( CpgOp & c ) { + cpg_error_t result; + unsigned int snooze = 10; + for ( unsigned int nth_try = 0; nth_try < cpgRetries; ++ nth_try ) { + if ( CPG_OK == (result = c.op(handle, & group))) { + QPID_LOG(info, c.opName << " successful."); + break; + } + else if ( result == CPG_ERR_TRY_AGAIN ) { + QPID_LOG(info, "Retrying " << c.opName ); + sys::usleep ( snooze ); + snooze *= 10; + snooze = (snooze <= maxCpgRetrySleep) ? snooze : maxCpgRetrySleep; + } + else break; // Don't retry unless CPG tells us to. + } + + if ( result != CPG_OK ) + CPG_CHECK(result, c.msg(group)); +} + // Global callback functions. void Cpg::globalDeliver ( cpg_handle_t handle, @@ -129,11 +153,11 @@ Cpg::~Cpg() { void Cpg::join(const std::string& name) { group = name; - CPG_CHECK(cpg_join(handle, &group), cantJoinMsg(group)); + callCpg ( cpgJoinOp ); } void Cpg::leave() { - CPG_CHECK(cpg_leave(handle, &group), cantLeaveMsg(group)); + callCpg ( cpgLeaveOp ); } @@ -158,7 +182,8 @@ void Cpg::shutdown() { if (!isShutdown) { QPID_LOG(debug,"Shutting down CPG"); isShutdown=true; - CPG_CHECK(cpg_finalize(handle), "Error in shutdown of CPG"); + + callCpg ( cpgFinalizeOp ); } } @@ -201,6 +226,10 @@ std::string Cpg::cantJoinMsg(const Name& group) { return "Cannot join CPG group "+group.str(); } +std::string Cpg::cantFinalizeMsg(const Name& group) { + return "Cannot finalize CPG group "+group.str(); +} + std::string Cpg::cantLeaveMsg(const Name& group) { return "Cannot leave CPG group "+group.str(); } diff --git a/qpid/cpp/src/qpid/cluster/Cpg.h b/qpid/cpp/src/qpid/cluster/Cpg.h index cffbf0bdb3..6b81c602bd 100644 --- a/qpid/cpp/src/qpid/cluster/Cpg.h +++ b/qpid/cpp/src/qpid/cluster/Cpg.h @@ -39,6 +39,7 @@ namespace cluster { * On error all functions throw Cpg::Exception. * */ + class Cpg : public sys::IOHandle { public: struct Exception : public ::qpid::Exception { @@ -114,10 +115,73 @@ class Cpg : public sys::IOHandle { int getFd(); private: + + // Maximum number of retries for cog functions that can tell + // us to "try again later". + static const unsigned int cpgRetries = 5; + + // Don't let sleep-time between cpg retries to go above 0.1 second. + static const unsigned int maxCpgRetrySleep = 100000; + + + // Base class for the Cpg operations that need retry capability. + struct CpgOp { + std::string opName; + + CpgOp ( std::string opName ) + : opName(opName) { } + + virtual cpg_error_t op ( cpg_handle_t handle, struct cpg_name * ) = 0; + virtual std::string msg(const Name&) = 0; + virtual ~CpgOp ( ) { } + }; + + + struct CpgJoinOp : public CpgOp { + CpgJoinOp ( ) + : CpgOp ( std::string("cpg_join") ) { } + + cpg_error_t op(cpg_handle_t handle, struct cpg_name * group) { + return cpg_join ( handle, group ); + } + + std::string msg(const Name& name) { return cantJoinMsg(name); } + }; + + struct CpgLeaveOp : public CpgOp { + CpgLeaveOp ( ) + : CpgOp ( std::string("cpg_leave") ) { } + + cpg_error_t op(cpg_handle_t handle, struct cpg_name * group) { + return cpg_leave ( handle, group ); + } + + std::string msg(const Name& name) { return cantLeaveMsg(name); } + }; + + struct CpgFinalizeOp : public CpgOp { + CpgFinalizeOp ( ) + : CpgOp ( std::string("cpg_finalize") ) { } + + cpg_error_t op(cpg_handle_t handle, struct cpg_name *) { + return cpg_finalize ( handle ); + } + + std::string msg(const Name& name) { return cantFinalizeMsg(name); } + }; + + // This fn standardizes retry policy across all Cpg ops that need it. + void callCpg ( CpgOp & ); + + CpgJoinOp cpgJoinOp; + CpgLeaveOp cpgLeaveOp; + CpgFinalizeOp cpgFinalizeOp; + static std::string errorStr(cpg_error_t err, const std::string& msg); static std::string cantJoinMsg(const Name&); static std::string cantLeaveMsg(const Name&); static std::string cantMcastMsg(const Name&); + static std::string cantFinalizeMsg(const Name&); static Cpg* cpgFromHandle(cpg_handle_t); diff --git a/qpid/cpp/src/qpid/cluster/UpdateClient.cpp b/qpid/cpp/src/qpid/cluster/UpdateClient.cpp index f263577fd3..279284da2c 100644 --- a/qpid/cpp/src/qpid/cluster/UpdateClient.cpp +++ b/qpid/cpp/src/qpid/cluster/UpdateClient.cpp @@ -217,10 +217,11 @@ class MessageUpdater { // Disable client code that clears the delivery-properties.exchange sb.get()->setDoClearDeliveryPropertiesExchange(false); framing::MessageTransferBody transfer( - framing::ProtocolVersion(), UpdateClient::UPDATE, message::ACCEPT_MODE_NONE, - message::ACQUIRE_MODE_PRE_ACQUIRED); + *message.payload->getFrames().as<framing::MessageTransferBody>()); + transfer.setDestination(UpdateClient::UPDATE); - sb.get()->send(transfer, message.payload->getFrames(), !message.payload->isContentReleased()); + sb.get()->send(transfer, message.payload->getFrames(), + !message.payload->isContentReleased()); if (message.payload->isContentReleased()){ uint16_t maxFrameSize = sb.get()->getConnection()->getNegotiatedSettings().maxFrameSize; uint16_t maxContentSize = maxFrameSize - AMQFrame::frameOverhead(); diff --git a/qpid/cpp/src/qpid/cluster/UpdateExchange.cpp b/qpid/cpp/src/qpid/cluster/UpdateExchange.cpp new file mode 100644 index 0000000000..11937f296f --- /dev/null +++ b/qpid/cpp/src/qpid/cluster/UpdateExchange.cpp @@ -0,0 +1,47 @@ +/* + * + * 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/framing/MessageTransferBody.h" +#include "qpid/broker/Message.h" +#include "UpdateExchange.h" + +namespace qpid { +namespace cluster { + +using framing::MessageTransferBody; +using framing::DeliveryProperties; + +UpdateExchange::UpdateExchange(management::Manageable* parent) + : broker::Exchange(UpdateClient::UPDATE, parent), + broker::FanOutExchange(UpdateClient::UPDATE, parent) {} + + +void UpdateExchange::setProperties(const boost::intrusive_ptr<broker::Message>& msg) { + MessageTransferBody* transfer = msg->getMethod<MessageTransferBody>(); + assert(transfer); + const DeliveryProperties* props = msg->getProperties<DeliveryProperties>(); + assert(props); + if (props->hasExchange()) + transfer->setDestination(props->getExchange()); + else + transfer->clearDestinationFlag(); +} + +}} // namespace qpid::cluster diff --git a/qpid/cpp/src/qpid/cluster/UpdateExchange.h b/qpid/cpp/src/qpid/cluster/UpdateExchange.h index 00a92c7f1e..9d7d9ee5fc 100644 --- a/qpid/cpp/src/qpid/cluster/UpdateExchange.h +++ b/qpid/cpp/src/qpid/cluster/UpdateExchange.h @@ -30,14 +30,14 @@ namespace qpid { namespace cluster { /** - * A keyless exchange (like fanout exchange) that does not modify delivery-properties.exchange - * on messages. + * A keyless exchange (like fanout exchange) that does not modify + * delivery-properties.exchange but copies it to the MessageTransfer. */ class UpdateExchange : public broker::FanOutExchange { public: - UpdateExchange(management::Manageable* parent) : broker::Exchange(UpdateClient::UPDATE, parent), broker::FanOutExchange(UpdateClient::UPDATE, parent) {} - void setProperties(const boost::intrusive_ptr<broker::Message>&) {} + UpdateExchange(management::Manageable* parent); + void setProperties(const boost::intrusive_ptr<broker::Message>&); }; }} // namespace qpid::cluster diff --git a/qpid/cpp/src/qpid/cluster/WatchDogPlugin.cpp b/qpid/cpp/src/qpid/cluster/WatchDogPlugin.cpp index 43dba5e09b..fc2b830dac 100644 --- a/qpid/cpp/src/qpid/cluster/WatchDogPlugin.cpp +++ b/qpid/cpp/src/qpid/cluster/WatchDogPlugin.cpp @@ -113,7 +113,7 @@ struct WatchDogPlugin : public qpid::Plugin, public qpid::sys::Fork { protected: void child() { // Child of fork - const char* watchdog = ::getenv("QPID_WATCHDOG_EXE"); // For use in tests + const char* watchdog = ::getenv("QPID_WATCHDOG_EXEC"); // For use in tests if (!watchdog) watchdog=QPID_EXEC_DIR "/qpidd_watchdog"; std::string interval = boost::lexical_cast<std::string>(settings.interval); ::execl(watchdog, watchdog, interval.c_str(), NULL); diff --git a/qpid/cpp/src/qpid/framing/FrameSet.cpp b/qpid/cpp/src/qpid/framing/FrameSet.cpp index f50035e49a..c03dd39458 100644 --- a/qpid/cpp/src/qpid/framing/FrameSet.cpp +++ b/qpid/cpp/src/qpid/framing/FrameSet.cpp @@ -52,6 +52,11 @@ const AMQMethodBody* FrameSet::getMethod() const return parts.empty() ? 0 : parts[0].getMethod(); } +AMQMethodBody* FrameSet::getMethod() +{ + return parts.empty() ? 0 : parts[0].getMethod(); +} + const AMQHeaderBody* FrameSet::getHeaders() const { return parts.size() < 2 ? 0 : parts[1].castBody<AMQHeaderBody>(); diff --git a/qpid/cpp/src/qpid/framing/FrameSet.h b/qpid/cpp/src/qpid/framing/FrameSet.h index e3e8727600..398a709353 100644 --- a/qpid/cpp/src/qpid/framing/FrameSet.h +++ b/qpid/cpp/src/qpid/framing/FrameSet.h @@ -57,6 +57,7 @@ public: bool isContentBearing() const; QPID_COMMON_EXTERN const AMQMethodBody* getMethod() const; + QPID_COMMON_EXTERN AMQMethodBody* getMethod(); QPID_COMMON_EXTERN const AMQHeaderBody* getHeaders() const; QPID_COMMON_EXTERN AMQHeaderBody* getHeaders(); @@ -70,6 +71,11 @@ public: return (method && method->isA<T>()) ? dynamic_cast<const T*>(method) : 0; } + template <class T> T* as() { + AMQMethodBody* method = getMethod(); + return (method && method->isA<T>()) ? dynamic_cast<T*>(method) : 0; + } + template <class T> const T* getHeaderProperties() const { const AMQHeaderBody* header = getHeaders(); return header ? header->get<T>() : 0; diff --git a/qpid/cpp/src/qpid/messaging/Message.cpp b/qpid/cpp/src/qpid/messaging/Message.cpp index fb4e800eaa..deb40b6aa3 100644 --- a/qpid/cpp/src/qpid/messaging/Message.cpp +++ b/qpid/cpp/src/qpid/messaging/Message.cpp @@ -27,7 +27,7 @@ namespace messaging { Message::Message(const std::string& bytes) : impl(new MessageImpl(bytes)) {} Message::Message(const char* bytes, size_t count) : impl(new MessageImpl(bytes, count)) {} -Message::Message(const Message& m) : impl(new MessageImpl(m.getContent())) {} +Message::Message(const Message& m) : impl(new MessageImpl(*m.impl)) {} Message::~Message() { delete impl; } Message& Message::operator=(const Message& m) { *impl = *m.impl; return *this; } diff --git a/qpid/cpp/src/qpid/messaging/Variant.cpp b/qpid/cpp/src/qpid/messaging/Variant.cpp index 71f9fbe646..9c2f92f47a 100644 --- a/qpid/cpp/src/qpid/messaging/Variant.cpp +++ b/qpid/cpp/src/qpid/messaging/Variant.cpp @@ -21,11 +21,10 @@ #include "qpid/messaging/Variant.h" #include <boost/format.hpp> #include <boost/lexical_cast.hpp> +#include <algorithm> +#include <sstream> namespace qpid { -namespace client { -} - namespace messaging { InvalidConversion::InvalidConversion(const std::string& msg) : Exception(msg) {} @@ -81,6 +80,9 @@ class VariantImpl void setEncoding(const std::string&); const std::string& getEncoding() const; + bool isEqualTo(VariantImpl&) const; + bool isEquivalentTo(VariantImpl&) const; + static VariantImpl* create(const Variant&); private: const VariantType type; @@ -170,6 +172,18 @@ bool toBool(const std::string& s) throw InvalidConversion(QPID_MSG("Cannot convert " << s << " to bool")); } +template <class T> std::string toString(const T& t) +{ + std::stringstream out; + out << t; + return out.str(); +} + +template <class T> bool equal(const T& a, const T& b) +{ + return a.size() == b.size() && std::equal(a.begin(), a.end(), b.begin()); +} + } bool VariantImpl::asBool() const @@ -298,10 +312,37 @@ std::string VariantImpl::asString() const case VAR_DOUBLE: return boost::lexical_cast<std::string>(value.d); case VAR_FLOAT: return boost::lexical_cast<std::string>(value.f); case VAR_STRING: return *reinterpret_cast<std::string*>(value.v); + case VAR_LIST: return toString(asList()); + case VAR_MAP: return toString(asMap()); default: throw InvalidConversion(QPID_MSG("Cannot convert from " << getTypeName(type) << " to " << getTypeName(VAR_STRING))); } } +bool VariantImpl::isEqualTo(VariantImpl& other) const +{ + if (type == other.type) { + switch(type) { + case VAR_VOID: return true; + case VAR_BOOL: return value.b == other.value.b; + case VAR_UINT8: return value.ui8 == other.value.ui8; + case VAR_UINT16: return value.ui16 == other.value.ui16; + case VAR_UINT32: return value.ui32 == other.value.ui32; + case VAR_UINT64: return value.ui64 == other.value.ui64; + case VAR_INT8: return value.i8 == other.value.i8; + case VAR_INT16: return value.i16 == other.value.i16; + case VAR_INT32: return value.i32 == other.value.i32; + case VAR_INT64: return value.i64 == other.value.i64; + case VAR_DOUBLE: return value.d == other.value.d; + case VAR_FLOAT: return value.f == other.value.f; + case VAR_STRING: return *reinterpret_cast<std::string*>(value.v) + == *reinterpret_cast<std::string*>(other.value.v); + case VAR_LIST: return equal(asList(), other.asList()); + case VAR_MAP: return equal(asMap(), other.asMap()); + } + } + return false; +} + const Variant::Map& VariantImpl::asMap() const { switch(type) { @@ -605,4 +646,14 @@ std::ostream& operator<<(std::ostream& out, const Variant& value) return out; } +bool operator==(const Variant& a, const Variant& b) +{ + return a.isEqualTo(b); +} + +bool Variant::isEqualTo(const Variant& other) const +{ + return impl->isEqualTo(*other.impl); +} + }} // namespace qpid::messaging diff --git a/qpid/cpp/src/qpid/sys/LockFile.h b/qpid/cpp/src/qpid/sys/LockFile.h index 1f0a9e13b3..14a76cbf3e 100644 --- a/qpid/cpp/src/qpid/sys/LockFile.h +++ b/qpid/cpp/src/qpid/sys/LockFile.h @@ -29,7 +29,7 @@ namespace qpid { namespace sys { -class LockFilePrivate; +class LockFilePrivate; /** * @class LockFile @@ -43,34 +43,17 @@ class LockFilePrivate; */ class LockFile : private boost::noncopyable { - boost::shared_ptr<LockFilePrivate> impl; - std::string path; bool created; + boost::shared_ptr<LockFilePrivate> impl; + +protected: + int read(void*, size_t) const; + int write(void*, size_t) const; public: QPID_COMMON_EXTERN LockFile(const std::string& path_, bool create); QPID_COMMON_EXTERN ~LockFile(); - - /** - * Read the process ID from the lock file. This method assumes that - * if there is a process ID in the file, it was written there by - * writePid(); thus, it's at the start of the file. - * - * Throws an exception if there is an error reading the file. - * - * @returns The stored process ID. No validity check is done on it. - */ - QPID_COMMON_EXTERN pid_t readPid(void) const; - - /** - * Write the current process's ID to the lock file. It's written at - * the start of the file and will overwrite any other content that - * may be in the file. - * - * Throws an exception if the write fails. - */ - QPID_COMMON_EXTERN void writePid(void); }; }} /* namespace qpid::sys */ diff --git a/qpid/cpp/src/qpid/sys/SocketAddress.h b/qpid/cpp/src/qpid/sys/SocketAddress.h index fcb9c81d43..27b9642f2c 100644 --- a/qpid/cpp/src/qpid/sys/SocketAddress.h +++ b/qpid/cpp/src/qpid/sys/SocketAddress.h @@ -37,6 +37,8 @@ class SocketAddress { public: /** Create a SocketAddress from hostname and port*/ QPID_COMMON_EXTERN SocketAddress(const std::string& host, const std::string& port); + QPID_COMMON_EXTERN SocketAddress(const SocketAddress&); + QPID_COMMON_EXTERN SocketAddress& operator=(const SocketAddress&); QPID_COMMON_EXTERN ~SocketAddress(); std::string asString() const; @@ -44,7 +46,7 @@ public: private: std::string host; std::string port; - ::addrinfo* addrInfo; + mutable ::addrinfo* addrInfo; }; }} diff --git a/qpid/cpp/src/qpid/sys/posix/LockFile.cpp b/qpid/cpp/src/qpid/sys/posix/LockFile.cpp index 4900252984..1862ff6ac9 100755 --- a/qpid/cpp/src/qpid/sys/posix/LockFile.cpp +++ b/qpid/cpp/src/qpid/sys/posix/LockFile.cpp @@ -17,6 +17,7 @@ */ #include "qpid/sys/LockFile.h" +#include "qpid/sys/posix/PidFile.h" #include <string> #include <unistd.h> @@ -31,6 +32,7 @@ namespace sys { class LockFilePrivate { friend class LockFile; + friend class PidFile; int fd; @@ -64,27 +66,43 @@ LockFile::~LockFile() { } } -pid_t LockFile::readPid(void) const { +int LockFile::read(void* bytes, size_t len) const { if (!impl) - throw Exception("Lock file not open"); + throw Exception("Lock file not open: " + path); - pid_t pid; - int desired_read = sizeof(pid_t); - if (desired_read > ::read(impl->fd, &pid, desired_read) ) { - throw Exception("Cannot read lock file " + path); + ssize_t rc = ::read(impl->fd, bytes, len); + if ((ssize_t)len > rc) { + throw Exception("Cannot read lock file: " + path); } - return pid; + return rc; } -void LockFile::writePid(void) { +int LockFile::write(void* bytes, size_t len) const { if (!impl) - throw Exception("Lock file not open"); + throw Exception("Lock file not open: " + path); + + ssize_t rc = ::write(impl->fd, bytes, len); + if ((ssize_t)len > rc) { + throw Exception("Cannot write lock file: " + path); + } + return rc; +} + +PidFile::PidFile(const std::string& path_, bool create): + LockFile(path_, create) +{} +pid_t PidFile::readPid(void) const { + pid_t pid; + int desired_read = sizeof(pid_t); + read(&pid, desired_read); + return pid; +} + +void PidFile::writePid(void) { pid_t pid = getpid(); int desired_write = sizeof(pid_t); - if (desired_write > ::write(impl->fd, &pid, desired_write)) { - throw Exception("Cannot write lock file " + path); - } + write(&pid, desired_write); } }} /* namespace qpid::sys */ diff --git a/qpid/cpp/src/qpid/sys/posix/PidFile.h b/qpid/cpp/src/qpid/sys/posix/PidFile.h new file mode 100644 index 0000000000..fb19d407f4 --- /dev/null +++ b/qpid/cpp/src/qpid/sys/posix/PidFile.h @@ -0,0 +1,62 @@ +#ifndef _sys_PidFile_h +#define _sys_PidFile_h + +/* + * + * Copyright (c) 2008 The Apache Software Foundation + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +#include "qpid/sys/LockFile.h" + +#include "qpid/CommonImportExport.h" +#include "qpid/sys/IntegerTypes.h" + +#include <boost/noncopyable.hpp> +#include <boost/shared_ptr.hpp> +#include <string> + +namespace qpid { +namespace sys { + +class PidFile : public LockFile +{ +public: + QPID_COMMON_EXTERN PidFile(const std::string& path_, bool create); + + /** + * Read the process ID from the lock file. This method assumes that + * if there is a process ID in the file, it was written there by + * writePid(); thus, it's at the start of the file. + * + * Throws an exception if there is an error reading the file. + * + * @returns The stored process ID. No validity check is done on it. + */ + QPID_COMMON_EXTERN pid_t readPid(void) const; + + /** + * Write the current process's ID to the lock file. It's written at + * the start of the file and will overwrite any other content that + * may be in the file. + * + * Throws an exception if the write fails. + */ + QPID_COMMON_EXTERN void writePid(void); +}; + +}} /* namespace qpid::sys */ + +#endif /*!_sys_PidFile_h*/ diff --git a/qpid/cpp/src/qpid/sys/posix/SocketAddress.cpp b/qpid/cpp/src/qpid/sys/posix/SocketAddress.cpp index fe8812299c..cb44f8e41f 100644 --- a/qpid/cpp/src/qpid/sys/posix/SocketAddress.cpp +++ b/qpid/cpp/src/qpid/sys/posix/SocketAddress.cpp @@ -35,27 +35,34 @@ SocketAddress::SocketAddress(const std::string& host0, const std::string& port0) port(port0), addrInfo(0) { - ::addrinfo hints; - ::memset(&hints, 0, sizeof(hints)); - hints.ai_family = AF_INET; // In order to allow AF_INET6 we'd have to change createTcp() as well - hints.ai_socktype = SOCK_STREAM; +} - const char* node = 0; - if (host.empty()) { - hints.ai_flags |= AI_PASSIVE; - } else { - node = host.c_str(); - } - const char* service = port.empty() ? "0" : port.c_str(); +SocketAddress::SocketAddress(const SocketAddress& sa) : + host(sa.host), + port(sa.port), + addrInfo(0) +{ +} + +SocketAddress& SocketAddress::operator=(const SocketAddress& sa) +{ + if (&sa != this) { + host = sa.host; + port = sa.port; - int n = ::getaddrinfo(node, service, &hints, &addrInfo); - if (n != 0) - throw Exception(QPID_MSG("Cannot resolve " << host << ": " << ::gai_strerror(n))); + if (addrInfo) { + ::freeaddrinfo(addrInfo); + addrInfo = 0; + } + } + return *this; } SocketAddress::~SocketAddress() { - ::freeaddrinfo(addrInfo); + if (addrInfo) { + ::freeaddrinfo(addrInfo); + } } std::string SocketAddress::asString() const @@ -65,6 +72,25 @@ std::string SocketAddress::asString() const const ::addrinfo& getAddrInfo(const SocketAddress& sa) { + if (!sa.addrInfo) { + ::addrinfo hints; + ::memset(&hints, 0, sizeof(hints)); + hints.ai_family = AF_INET; // In order to allow AF_INET6 we'd have to change createTcp() as well + hints.ai_socktype = SOCK_STREAM; + + const char* node = 0; + if (sa.host.empty()) { + hints.ai_flags |= AI_PASSIVE; + } else { + node = sa.host.c_str(); + } + const char* service = sa.port.empty() ? "0" : sa.port.c_str(); + + int n = ::getaddrinfo(node, service, &hints, &sa.addrInfo); + if (n != 0) + throw Exception(QPID_MSG("Cannot resolve " << sa.host << ": " << ::gai_strerror(n))); + } + return *sa.addrInfo; } diff --git a/qpid/cpp/src/qpid/sys/windows/LockFile.cpp b/qpid/cpp/src/qpid/sys/windows/LockFile.cpp index e9079b6094..048c2d5b18 100755 --- a/qpid/cpp/src/qpid/sys/windows/LockFile.cpp +++ b/qpid/cpp/src/qpid/sys/windows/LockFile.cpp @@ -36,7 +36,7 @@ public: LockFile::LockFile(const std::string& path_, bool create) : path(path_), created(create) { - HANDLE h = CreateFile(path.c_str(), + HANDLE h = ::CreateFile(path.c_str(), create ? (GENERIC_READ|GENERIC_WRITE) : GENERIC_READ, FILE_SHARE_READ|FILE_SHARE_WRITE|FILE_SHARE_DELETE, 0, /* Default security */ @@ -44,41 +44,21 @@ LockFile::LockFile(const std::string& path_, bool create) FILE_FLAG_DELETE_ON_CLOSE, /* Delete file when closed */ NULL); if (h == INVALID_HANDLE_VALUE) - throw qpid::Exception(path + qpid::sys::strError(GetLastError())); + throw qpid::Exception(path + ": " + qpid::sys::strError(GetLastError())); + + // Lock up to 4Gb + if (!::LockFile(h, 0, 0, 0xffffffff, 0)) + throw qpid::Exception(path + ": " + qpid::sys::strError(GetLastError())); impl.reset(new LockFilePrivate(h)); } LockFile::~LockFile() { if (impl) { if (impl->fd != INVALID_HANDLE_VALUE) { - CloseHandle(impl->fd); + ::UnlockFile(impl->fd, 0, 0, 0xffffffff, 0); + ::CloseHandle(impl->fd); } } } -pid_t LockFile::readPid(void) const { - if (!impl) - throw Exception("Lock file not open"); - - pid_t pid; - DWORD desired_read = sizeof(pid_t); - DWORD actual_read = 0; - if (!ReadFile(impl->fd, &pid, desired_read, &actual_read, 0)) { - throw Exception("Cannot read lock file " + path); - } - return pid; -} - -void LockFile::writePid(void) { - if (!impl) - throw Exception("Lock file not open"); - - pid_t pid = GetCurrentProcessId(); - DWORD desired_write = sizeof(pid_t); - DWORD written = 0; - if (!WriteFile(impl->fd, &pid, desired_write, &written, 0)) { - throw Exception("Cannot write lock file " + path); - } -} - }} /* namespace qpid::sys */ diff --git a/qpid/cpp/src/tests/CMakeLists.txt b/qpid/cpp/src/tests/CMakeLists.txt index 43f92b2d2d..469d777dce 100644 --- a/qpid/cpp/src/tests/CMakeLists.txt +++ b/qpid/cpp/src/tests/CMakeLists.txt @@ -140,6 +140,7 @@ set(unit_tests_to_build ClientMessageTest PollableCondition Variant + ClientMessage ${xml_tests} CACHE STRING "Which unit tests to build" ) diff --git a/qpid/cpp/src/tests/ClientMessage.cpp b/qpid/cpp/src/tests/ClientMessage.cpp new file mode 100644 index 0000000000..ab1cf9d102 --- /dev/null +++ b/qpid/cpp/src/tests/ClientMessage.cpp @@ -0,0 +1,46 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +#include <iostream> +#include "qpid/messaging/Message.h" + +#include "unit_test.h" + +using namespace qpid::messaging; + +namespace qpid { +namespace tests { + +QPID_AUTO_TEST_SUITE(ClientMessageSuite) + +QPID_AUTO_TEST_CASE(testCopyConstructor) +{ + Message m("my-data"); + m.setSubject("my-subject"); + m.getHeaders()["a"] = "ABC"; + Message c(m); + BOOST_CHECK_EQUAL(m.getContent(), c.getContent()); + BOOST_CHECK_EQUAL(m.getSubject(), c.getSubject()); + BOOST_CHECK_EQUAL(m.getHeaders()["a"], c.getHeaders()["a"]); +} + +QPID_AUTO_TEST_SUITE_END() + +}} // namespace qpid::tests diff --git a/qpid/cpp/src/tests/Makefile.am b/qpid/cpp/src/tests/Makefile.am index 013efa8c05..9a6e67ca2e 100644 --- a/qpid/cpp/src/tests/Makefile.am +++ b/qpid/cpp/src/tests/Makefile.am @@ -116,7 +116,8 @@ unit_test_SOURCES= unit_test.cpp unit_test.h \ ClientMessageTest.cpp \ PollableCondition.cpp \ Variant.cpp \ - Address.cpp + Address.cpp \ + ClientMessage.cpp if HAVE_XML unit_test_SOURCES+= XmlClientSessionTest.cpp diff --git a/qpid/cpp/src/tests/SocketProxy.h b/qpid/cpp/src/tests/SocketProxy.h index df243cb42a..4582dc36fd 100644 --- a/qpid/cpp/src/tests/SocketProxy.h +++ b/qpid/cpp/src/tests/SocketProxy.h @@ -141,12 +141,12 @@ class SocketProxy : private qpid::sys::Runnable } // Something is set; relay data as needed until something closes if (FD_ISSET(server->getFd(), &socks)) { - ssize_t n = server->read(buffer, sizeof(buffer)); + int n = server->read(buffer, sizeof(buffer)); throwIf(n <= 0, "SocketProxy: server disconnected"); if (!dropServer) client.write(buffer, n); } if (FD_ISSET(client.getFd(), &socks)) { - ssize_t n = client.read(buffer, sizeof(buffer)); + int n = client.read(buffer, sizeof(buffer)); throwIf(n <= 0, "SocketProxy: client disconnected"); if (!dropServer) server->write(buffer, n); } diff --git a/qpid/cpp/src/tests/Variant.cpp b/qpid/cpp/src/tests/Variant.cpp index 2d68bb842c..21005779f0 100644 --- a/qpid/cpp/src/tests/Variant.cpp +++ b/qpid/cpp/src/tests/Variant.cpp @@ -157,6 +157,22 @@ QPID_AUTO_TEST_CASE(testMap) BOOST_CHECK_EQUAL(std::string("now it's a string"), value.asMap()["my-key"].asString()); } +QPID_AUTO_TEST_CASE(testIsEqualTo) +{ + BOOST_CHECK_EQUAL(Variant("abc"), Variant("abc")); + BOOST_CHECK_EQUAL(Variant(1234), Variant(1234)); + + Variant a = Variant::Map(); + a.asMap()["colour"] = "red"; + a.asMap()["pi"] = 3.14f; + a.asMap()["my-key"] = 1234; + Variant b = Variant::Map(); + b.asMap()["colour"] = "red"; + b.asMap()["pi"] = 3.14f; + b.asMap()["my-key"] = 1234; + BOOST_CHECK_EQUAL(a, b); +} + QPID_AUTO_TEST_SUITE_END() }} // namespace qpid::tests diff --git a/qpid/cpp/src/tests/cluster.cmake b/qpid/cpp/src/tests/cluster.cmake index cc7d154e7d..9084bfb85b 100644 --- a/qpid/cpp/src/tests/cluster.cmake +++ b/qpid/cpp/src/tests/cluster.cmake @@ -35,7 +35,7 @@ set (cluster_test_SOURCES InitialStatusMap StoreStatus ) -add_executable (cluster_test cluster_test ${cluster_test_SOURCES} ${platform_test_additions}) +add_executable (cluster_test ${cluster_test_SOURCES} ${platform_test_additions}) target_link_libraries (cluster_test ${qpid_test_boost_libs} qpidclient qpidbroker cluster_shared) remember_location(cluster_test) diff --git a/qpid/cpp/src/tests/cluster_tests.py b/qpid/cpp/src/tests/cluster_tests.py index e57a826000..178271e977 100755 --- a/qpid/cpp/src/tests/cluster_tests.py +++ b/qpid/cpp/src/tests/cluster_tests.py @@ -57,6 +57,34 @@ class ShortTests(BrokerTest): self.assertEqual("y", m.content) s2.connection.close() + def test_store_direct_update_match(self): + """Verify that brokers stores an identical message whether they receive it + direct from clients or during an update, no header or other differences""" + cluster = self.cluster(0, args=["--load-module", self.test_store_lib]) + cluster.start(args=["--test-store-dump", "direct.dump"]) + # Try messages with various headers + cluster[0].send_message("q", Message(durable=True, content="foobar", + subject="subject", + reply_to="reply_to", + properties={"n":10})) + # Try messages of different sizes + for size in range(0,10000,100): + cluster[0].send_message("q", Message(content="x"*size, durable=True)) + # Try sending via named exchange + c = cluster[0].connect_old() + s = c.session(str(qpid.datatypes.uuid4())) + s.exchange_bind(exchange="amq.direct", binding_key="foo", queue="q") + props = s.delivery_properties(routing_key="foo", delivery_mode=2) + s.message_transfer( + destination="amq.direct", + message=qpid.datatypes.Message(props, "content")) + + # Now update a new member and compare their dumps. + cluster.start(args=["--test-store-dump", "updatee.dump"]) + assert file("direct.dump").read() == file("updatee.dump").read() + os.remove("direct.dump") + os.remove("updatee.dump") + class LongTests(BrokerTest): """Tests that can run for a long time if -DDURATION=<minutes> is set""" def duration(self): diff --git a/qpid/cpp/src/tests/test_store.cpp b/qpid/cpp/src/tests/test_store.cpp index c675c6daa3..257e77b6b4 100644 --- a/qpid/cpp/src/tests/test_store.cpp +++ b/qpid/cpp/src/tests/test_store.cpp @@ -34,11 +34,14 @@ #include "qpid/broker/NullMessageStore.h" #include "qpid/broker/Broker.h" +#include "qpid/framing/AMQFrame.h" #include "qpid/log/Statement.h" #include "qpid/Plugin.h" #include "qpid/Options.h" #include <boost/cast.hpp> #include <boost/lexical_cast.hpp> +#include <memory> +#include <fstream> using namespace qpid; using namespace broker; @@ -51,10 +54,13 @@ namespace tests { struct TestStoreOptions : public Options { string name; + string dump; TestStoreOptions() : Options("Test Store Options") { addOptions() - ("test-store-name", optValue(name, "NAME"), "Name to identify test store instance."); + ("test-store-name", optValue(name, "NAME"), "Name of test store instance.") + ("test-store-dump", optValue(dump, "FILE"), "File to dump enqueued messages.") + ; } }; @@ -71,19 +77,38 @@ struct Completer : public Runnable { class TestStore : public NullMessageStore { public: - TestStore(const string& name_, Broker& broker_) : name(name_), broker(broker_) {} + TestStore(const TestStoreOptions& opts, Broker& broker_) + : options(opts), name(opts.name), broker(broker_) + { + QPID_LOG(info, "TestStore name=" << name << " dump=" << options.dump); + if (!options.dump.empty()) + dump.reset(new ofstream(options.dump.c_str())); + } ~TestStore() { for_each(threads.begin(), threads.end(), boost::bind(&Thread::join, _1)); } + virtual bool isNull() const { return false; } + void enqueue(TransactionContext* , - const boost::intrusive_ptr<PersistableMessage>& msg, + const boost::intrusive_ptr<PersistableMessage>& pmsg, const PersistableQueue& ) { - string data = boost::polymorphic_downcast<Message*>(msg.get())->getFrames().getContent(); + Message* msg = dynamic_cast<Message*>(pmsg.get()); + assert(msg); + + // Dump the message if there is a dump file. + if (dump.get()) { + msg->getFrames().getMethod()->print(*dump); + *dump << endl << " "; + msg->getFrames().getHeaders()->print(*dump); + *dump << endl << " "; + *dump << msg->getFrames().getContentSize() << endl; + } // Check the message for special instructions. + string data = msg->getFrames().getContent(); size_t i = string::npos; size_t j = string::npos; if (strncmp(data.c_str(), TEST_STORE_DO.c_str(), strlen(TEST_STORE_DO.c_str())) == 0 @@ -119,9 +144,11 @@ class TestStore : public NullMessageStore { private: static const string TEST_STORE_DO, EXCEPTION, EXIT_PROCESS, ASYNC; + TestStoreOptions options; string name; Broker& broker; vector<Thread> threads; + std::auto_ptr<ofstream> dump; }; const string TestStore::TEST_STORE_DO = "TEST_STORE_DO: "; @@ -139,7 +166,7 @@ struct TestStorePlugin : public Plugin { { Broker* broker = dynamic_cast<Broker*>(&target); if (!broker) return; - boost::shared_ptr<MessageStore> p(new TestStore(options.name, *broker)); + boost::shared_ptr<MessageStore> p(new TestStore(options, *broker)); broker->setStore (p); } diff --git a/qpid/cpp/src/tests/test_watchdog b/qpid/cpp/src/tests/test_watchdog index 2e9967ee24..5afdaa5e24 100755 --- a/qpid/cpp/src/tests/test_watchdog +++ b/qpid/cpp/src/tests/test_watchdog @@ -21,9 +21,10 @@ # Tests for the watchdog plug-in +source ./test_env.sh # Start a broker with watchdog, freeze it with kill -STOP, verify that it is killed. -PORT=`$QPIDD_EXEC -dp0 --no-data-dir --auth=no --no-module-dir --load-module $WATCHDOG_LIB --log-to-file=qpidd_watchdog.log --watchdog-interval 1` -PID=`$QPIDD_EXEC --no-module-dir -cp $PORT` +PORT=`$QPIDD_EXEC -dp0 --no-data-dir --auth=no --no-module-dir --load-module $WATCHDOG_LIB --log-to-file=qpidd_watchdog.log --watchdog-interval 1` || exit 1 +PID=`$QPIDD_EXEC --no-module-dir -cp $PORT` || exit 1 kill -STOP $PID sleep 2 diff --git a/qpid/cpp/src/windows/QpiddBroker.cpp b/qpid/cpp/src/windows/QpiddBroker.cpp index fc4f9f8a92..15380dda0b 100644 --- a/qpid/cpp/src/windows/QpiddBroker.cpp +++ b/qpid/cpp/src/windows/QpiddBroker.cpp @@ -34,7 +34,6 @@ const char *QPIDD_MODULE_DIR = "."; #include "qpid/Options.h" #include "qpid/Plugin.h" #include "qpid/sys/IntegerTypes.h" -#include "qpid/sys/LockFile.h" #include "qpid/sys/windows/check.h" #include "qpid/broker/Broker.h" @@ -59,74 +58,68 @@ namespace { const std::string TCP = "tcp"; -std::string brokerPidFile(std::string piddir, uint16_t port) -{ - std::ostringstream path; - path << piddir << "\\broker_" << port << ".pid"; - return path.str(); -} - // ShutdownEvent maintains an event that can be used to ask the broker // to stop. Analogous to sending SIGTERM/SIGINT to the posix broker. // The signal() method signals the event. class ShutdownEvent { public: - ShutdownEvent(pid_t other = 0); + ShutdownEvent(int port); ~ShutdownEvent(); + void create(); + void open(); void signal(); + private: + std::string eventName; + protected: - std::string eventName(pid_t pid); HANDLE event; }; class ShutdownHandler : public ShutdownEvent, public qpid::sys::Runnable { public: - ShutdownHandler(const boost::intrusive_ptr<Broker>& b) - : ShutdownEvent() { broker = b; } + ShutdownHandler(int port, const boost::intrusive_ptr<Broker>& b) + : ShutdownEvent(port) { broker = b; } private: virtual void run(); // Inherited from Runnable boost::intrusive_ptr<Broker> broker; }; -ShutdownEvent::ShutdownEvent(pid_t other) : event(NULL) { - // If given a pid, open an event assumedly created by that pid. If there's - // no pid, create a new event using the current process id. - if (other == 0) { - std::string name = eventName(GetCurrentProcessId()); - // Auto-reset event in case multiple processes try to signal a - // broker that doesn't respond for some reason. Initially not signaled. - event = CreateEvent(NULL, false, false, name.c_str()); - } - else { - std::string name = eventName(other); - event = OpenEvent(EVENT_MODIFY_STATE, false, name.c_str()); - } +ShutdownEvent::ShutdownEvent(int port) : event(NULL) { + std::ostringstream name; + name << "qpidd_" << port << std::ends; + eventName = name.str(); +} + +void ShutdownEvent::create() { + // Auto-reset event in case multiple processes try to signal a + // broker that doesn't respond for some reason. Initially not signaled. + event = ::CreateEvent(NULL, false, false, eventName.c_str()); + QPID_WINDOWS_CHECK_NULL(event); +} + +void ShutdownEvent::open() { + // TODO: Might need to search Global\\ name if unadorned name fails + event = ::OpenEvent(EVENT_MODIFY_STATE, false, eventName.c_str()); QPID_WINDOWS_CHECK_NULL(event); } ShutdownEvent::~ShutdownEvent() { - CloseHandle(event); + ::CloseHandle(event); event = NULL; } void ShutdownEvent::signal() { - QPID_WINDOWS_CHECK_NOT(SetEvent(event), 0); -} - -std::string ShutdownEvent::eventName(pid_t pid) { - std::ostringstream name; - name << "qpidd_" << pid << std::ends; - return name.str(); + QPID_WINDOWS_CHECK_NOT(::SetEvent(event), 0); } void ShutdownHandler::run() { if (event == NULL) return; - WaitForSingleObject(event, INFINITE); + ::WaitForSingleObject(event, INFINITE); if (broker.get()) { broker->shutdown(); broker = 0; // Release the broker reference @@ -134,19 +127,89 @@ void ShutdownHandler::run() { } // Console control handler to properly handle ctl-c. +int ourPort; BOOL CtrlHandler(DWORD ctl) { - ShutdownEvent shutter; // no pid specified == shut me down + ShutdownEvent shutter(ourPort); // We have to have set up the port before interrupting + shutter.open(); shutter.signal(); return ((ctl == CTRL_C_EVENT || ctl == CTRL_CLOSE_EVENT) ? TRUE : FALSE); } +template <typename T> +class NamedSharedMemory { + std::string name; + HANDLE memory; + T* data; + +public: + NamedSharedMemory(const std::string&); + ~NamedSharedMemory(); + + T& create(); + T& get(); +}; + +template <typename T> +NamedSharedMemory<T>::NamedSharedMemory(const std::string& n) : + name(n), + memory(NULL), + data(0) +{}; + +template <typename T> +NamedSharedMemory<T>::~NamedSharedMemory() { + if (data) + ::UnmapViewOfFile(data); + if (memory != NULL) + ::CloseHandle(memory); +}; + +template <typename T> +T& NamedSharedMemory<T>::create() { + assert(memory == NULL); + + // Create named shared memory file + memory = ::CreateFileMapping(INVALID_HANDLE_VALUE, NULL, PAGE_READWRITE, 0, sizeof(T), name.c_str()); + QPID_WINDOWS_CHECK_NULL(memory); + + // Map file into memory + data = static_cast<T*>(::MapViewOfFile(memory, FILE_MAP_WRITE, 0, 0, 0)); + QPID_WINDOWS_CHECK_NULL(data); + + return *data; +} + +template <typename T> +T& NamedSharedMemory<T>::get() { + if (memory == NULL) { + // TODO: Might need to search Global\\ name if unadorned name fails + memory = ::OpenFileMapping(FILE_MAP_WRITE, FALSE, name.c_str()); + QPID_WINDOWS_CHECK_NULL(memory); + + data = static_cast<T*>(::MapViewOfFile(memory, FILE_MAP_WRITE, 0, 0, 0)); + QPID_WINDOWS_CHECK_NULL(data); + } + + return *data; +} + +std::string brokerInfoName(uint16_t port) +{ + std::ostringstream path; + path << "qpidd_info_" << port; + return path.str(); +} + +struct BrokerInfo { + DWORD pid; +}; + } struct ProcessControlOptions : public qpid::Options { bool quit; bool check; - std::string piddir; //std::string transport; No transport options yet - TCP is it. ProcessControlOptions() @@ -154,18 +217,9 @@ struct ProcessControlOptions : public qpid::Options { quit(false), check(false) //, transport(TCP) { - const DWORD pathLen = MAX_PATH + 1; - char tempDir[pathLen]; - if (GetTempPath(pathLen, tempDir) == 0) - piddir = "C:\\WINDOWS\\TEMP\\"; - else - piddir = tempDir; - piddir += "qpidd"; - // Only have TCP for now, so don't need this... // ("transport", optValue(transport, "TRANSPORT"), "The transport for which to return the port") addOptions() - ("pid-dir", qpid::optValue(piddir, "DIR"), "Directory where port-specific PID file is stored") ("check,c", qpid::optValue(check), "Prints the broker's process ID to stdout and returns 0 if the broker is running, otherwise returns 1") ("quit,q", qpid::optValue(quit), "Tells the broker to shut down"); } @@ -207,57 +261,50 @@ int QpiddBroker::execute (QpiddOptions *options) { if (myOptions->control.check || myOptions->control.quit) { // Relies on port number being set via --port or QPID_PORT env variable. - qpid::sys::LockFile getPid (brokerPidFile(myOptions->control.piddir, - options->broker.port), - false); - pid_t pid = getPid.readPid(); + NamedSharedMemory<BrokerInfo> info(brokerInfoName(options->broker.port)); + int pid = info.get().pid; if (pid < 0) return 1; if (myOptions->control.check) std::cout << pid << std::endl; if (myOptions->control.quit) { - ShutdownEvent shutter(pid); - HANDLE brokerHandle = OpenProcess(SYNCHRONIZE, false, pid); - QPID_WINDOWS_CHECK_NULL(brokerHandle); + ShutdownEvent shutter(options->broker.port); + shutter.open(); shutter.signal(); - WaitForSingleObject(brokerHandle, INFINITE); - CloseHandle(brokerHandle); + HANDLE brokerHandle = ::OpenProcess(SYNCHRONIZE, false, pid); + QPID_WINDOWS_CHECK_NULL(brokerHandle); + ::WaitForSingleObject(brokerHandle, INFINITE); + ::CloseHandle(brokerHandle); } return 0; } boost::intrusive_ptr<Broker> brokerPtr(new Broker(options->broker)); - // Make sure the pid directory exists, creating if needed. LockFile - // will throw an exception that makes little sense if it can't create - // the file. - if (!CreateDirectory(myOptions->control.piddir.c_str(), 0)) { - DWORD err = GetLastError(); - if (err != ERROR_ALREADY_EXISTS) - throw qpid::Exception(QPID_MSG("Can't create pid-dir " + - myOptions->control.piddir + - ": " + - qpid::sys::strError(err))); - } // Need the correct port number to use in the pid file name. if (options->broker.port == 0) options->broker.port = brokerPtr->getPort(""); - qpid::sys::LockFile myPid(brokerPidFile(myOptions->control.piddir, - options->broker.port), - true); - myPid.writePid(); + + BrokerInfo info; + info.pid = ::GetCurrentProcessId(); + + NamedSharedMemory<BrokerInfo> sharedInfo(brokerInfoName(options->broker.port)); + sharedInfo.create() = info; // Allow the broker to receive a shutdown request via a qpidd --quit // command. Note that when the broker is run as a service this operation // should not be allowed. - - ShutdownHandler waitShut(brokerPtr); + ourPort = options->broker.port; + ShutdownHandler waitShut(ourPort, brokerPtr); + waitShut.create(); qpid::sys::Thread waitThr(waitShut); // Wait for shutdown event - SetConsoleCtrlHandler((PHANDLER_ROUTINE)CtrlHandler, TRUE); + ::SetConsoleCtrlHandler((PHANDLER_ROUTINE)CtrlHandler, TRUE); brokerPtr->accept(); std::cout << options->broker.port << std::endl; brokerPtr->run(); waitShut.signal(); // In case we shut down some other way waitThr.join(); + + // CloseHandle(h); return 0; } diff --git a/qpid/java/broker/etc/log4j.xml b/qpid/java/broker/etc/log4j.xml index 484b203df8..d5c3537cc9 100644 --- a/qpid/java/broker/etc/log4j.xml +++ b/qpid/java/broker/etc/log4j.xml @@ -22,30 +22,43 @@ <log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/" debug="null" threshold="null"> <appender class="org.apache.log4j.QpidCompositeRollingAppender" name="ArchivingFileAppender"> - <!-- Ensure that logs allways have the dateFormat set--> - <param name="StaticLogFileName" value="false"/> - <param name="File" value="${QPID_WORK}/log/${logprefix}qpid${logsuffix}.log"/> - <param name="Append" value="false"/> + <!-- Ensure that logs allways have the dateFormat set Default: TRUE--> + <param name="StaticLogFileName" value="true"/> + <param name="file" value="${QPID_WORK}/log/${logprefix}qpid${logsuffix}.log"/> + <!-- Style of rolling to use, by: + File Size (1), + Date(2), + Both(3) - DEFAULT + When Date (or Both) is enabled then the value of DatePattern will determine + when the new file is made. e.g. a DatePattern of "'.'yyyy-MM-dd-HH-mm" + which includes minutes will cause a new backup file to be made every minute. + --> + <param name="RollingStyle" value="1"/> <!-- Change the direction so newer files have bigger numbers --> - <!-- So log.1 is written then log.2 etc This prevents a lot of file renames at log rollover --> - <param name="CountDirection" value="1"/> - <!-- Use default 10MB --> - <!--param name="MaxFileSize" value="100000"/--> + <!-- + negative means backups become <latest>,.0,.1,2,...,n + 0 means backup name is date stampted and follow Positive number if DataPattern clashes. + Positive means backup becomes <lastest,n,n-1,n-2,..0 + + Default is negative. + --> + <param name="CountDirection" value="0"/> + <!-- Use default 1MB --> + <param name="MaxFileSize" value="1MB"/> <param name="DatePattern" value="'.'yyyy-MM-dd-HH-mm"/> - <!-- Unlimited number of backups --> + <!-- Unlimited number of backups : Default: 0, no backups, -1 infinite --> <param name="MaxSizeRollBackups" value="-1"/> - <!-- Compress(gzip) the backup files--> + <!-- Compress(gzip) the backup files default:FALSE--> <param name="CompressBackupFiles" value="true"/> - <!-- Compress the backup files using a second thread --> + <!-- Compress the backup files using a second thread DEFAULT: FALSE--> <param name="CompressAsync" value="true"/> - <!-- Start at zero numbered files--> - <param name="ZeroBased" value="true"/> - <!-- Backup Location --> - <param name="backupFilesToPath" value="${QPID_WORK}/backup/log"/> + <!-- Backup Location : Default same dir as log file --> + <param name="backupFilesToPath" value="${QPID_WORK}/backup/log"/> + <layout class="org.apache.log4j.PatternLayout"> <param name="ConversionPattern" value="%d %-5p [%t] %C{2} (%F:%L) - %m%n"/> - </layout> + </layout>` </appender> <appender class="org.apache.log4j.FileAppender" name="FileAppender"> diff --git a/qpid/java/broker/src/main/java/org/apache/log4j/QpidCompositeRollingAppender.java b/qpid/java/broker/src/main/java/org/apache/log4j/QpidCompositeRollingAppender.java index 7e0c4defe1..be52b82789 100644 --- a/qpid/java/broker/src/main/java/org/apache/log4j/QpidCompositeRollingAppender.java +++ b/qpid/java/broker/src/main/java/org/apache/log4j/QpidCompositeRollingAppender.java @@ -157,6 +157,7 @@ public class QpidCompositeRollingAppender extends FileAppender protected String backupFilesToPath = null; private final ConcurrentLinkedQueue<CompressJob> _compress = new ConcurrentLinkedQueue<CompressJob>(); private AtomicBoolean _compressing = new AtomicBoolean(false); + private static final String COMPRESS_EXTENSION = ".gz"; /** The default constructor does nothing. */ public QpidCompositeRollingAppender() @@ -370,10 +371,6 @@ public class QpidCompositeRollingAppender extends FileAppender if (!staticLogFileName) { scheduledFilename = fileName = fileName.trim() + sdf.format(now); - if (countDirection > 0) - { - scheduledFilename = fileName = fileName + '.' + (++curSizeRollBackups); - } } super.setFile(fileName, append, bufferedIO, bufferSize); @@ -513,75 +510,10 @@ public class QpidCompositeRollingAppender extends FileAppender */ protected void existingInit() { - - if (zeroBased) - { - curSizeRollBackups = -1; - } - curTimeRollBackups = 0; // part A starts here - String filter; - if (staticLogFileName || !rollDate) - { - filter = baseFileName + ".*"; - } - else - { - filter = scheduledFilename + ".*"; - } - - File f = new File(baseFileName); - f = f.getParentFile(); - if (f == null) - { - f = new File("."); - } - - LogLog.debug("Searching for existing files in: " + f); - String[] files = f.list(); - - if (files != null) - { - for (int i = 0; i < files.length; i++) - { - if (!files[i].startsWith(baseFileName)) - { - continue; - } - - int index = files[i].lastIndexOf("."); - - if (staticLogFileName) - { - int endLength = files[i].length() - index; - if ((baseFileName.length() + endLength) != files[i].length()) - { - // file is probably scheduledFilename + .x so I don't care - continue; - } - } - - try - { - int backup = Integer.parseInt(files[i].substring(index + 1, files[i].length())); - LogLog.debug("From file: " + files[i] + " -> " + backup); - if (backup > curSizeRollBackups) - { - curSizeRollBackups = backup; - } - } - catch (Exception e) - { - // this happens when file.log -> file.log.yyyy-mm-dd which is normal - // when staticLogFileName == false - LogLog.debug("Encountered a backup file not ending in .x " + files[i]); - } - } - } - - LogLog.debug("curSizeRollBackups starts at: " + curSizeRollBackups); + // This is now down at first log when curSizeRollBackup==0 see rollFile // part A ends here // part B not yet implemented @@ -663,55 +595,11 @@ public class QpidCompositeRollingAppender extends FileAppender this.closeFile(); // keep windows happy. - // delete the old stuff here - - if (staticLogFileName) - { - /* Compute filename, but only if datePattern is specified */ - if (datePattern == null) - { - errorHandler.error("Missing DatePattern option in rollOver()."); - return; - } - - // is the new file name equivalent to the 'current' one - // something has gone wrong if we hit this -- we should only - // roll over if the new file will be different from the old - String dateFormat = sdf.format(now); - if (scheduledFilename.equals(fileName + dateFormat)) - { - errorHandler.error("Compare " + scheduledFilename + " : " + fileName + dateFormat); - - return; - } - - // close current file, and rename it to datedFilename - this.closeFile(); - - // we may have to roll over a large number of backups here - String from, to; - for (int i = 1; i <= curSizeRollBackups; i++) - { - from = fileName + '.' + i; - to = scheduledFilename + '.' + i; - rollFile(from, to, false); - } - - rollFile(fileName, scheduledFilename, compress); - } - else - { - if (compress) - { - compress(fileName); - } - } + rollFile(); try { - // This will also close the file. This is OK since multiple - // close operations are safe. curSizeRollBackups = 0; // We're cleared out the old date and are ready for the new // new scheduled name @@ -735,55 +623,46 @@ public class QpidCompositeRollingAppender extends FileAppender { if (compress) { - LogLog.debug("Attempting to compress file with same output name."); + LogLog.error("Attempting to compress file with same output name."); } return; } File target = new File(to); - if (target.exists()) - { - LogLog.debug("deleting existing target file: " + target); - target.delete(); - } File file = new File(from); - if (compress) + // Perform Roll by renaming + if (!file.getPath().equals(target.getPath())) { - compress(file, target); + file.renameTo(target); } - else + + // Compress file after it has been moved out the way... this is safe + // as it will gain a .gz ending and we can then safely delete this file + // as it will not be the statically named value. + if (compress) { - if (!file.getPath().equals(target.getPath())) - { - file.renameTo(target); - } + compress(target); } LogLog.debug(from + " -> " + to); } - protected void compress(String file) - { - File f = new File(file); - compress(f, f); - } - - private void compress(File from, File target) + private void compress(File target) { if (compressAsync) { synchronized (_compress) { - _compress.offer(new CompressJob(from, target)); + _compress.offer(new CompressJob(target, target)); } startCompression(); } else { - doCompress(from, target); + doCompress(target, target); } } @@ -796,9 +675,9 @@ public class QpidCompositeRollingAppender extends FileAppender } /** Delete's the specified file if it exists */ - protected static void deleteFile(String fileName) + protected void deleteFile(String fileName) { - File file = new File(fileName); + File file = compress ? new File(fileName + COMPRESS_EXTENSION) : new File(fileName); if (file.exists()) { file.delete(); @@ -837,69 +716,277 @@ public class QpidCompositeRollingAppender extends FileAppender // If maxBackups <= 0, then there is no file renaming to be done. if (maxSizeRollBackups != 0) { + rollFile(); + } - if (countDirection < 0) + try + { + // This will also close the file. This is OK since multiple + // close operations are safe. + this.setFile(baseFileName, false); + } + catch (IOException e) + { + LogLog.error("setFile(" + fileName + ", false) call failed.", e); + } + } + + /** + * Perform file Rollover ensuring the countDirection is applied along with + * the other options + */ + private void rollFile() + { + LogLog.debug("CD="+countDirection+",start"); + if (countDirection < 0) + { + // If we haven't rolled yet then validate we have the right value + // for curSizeRollBackups + if (curSizeRollBackups == 0) { - // Delete the oldest file, to keep Windows happy. - if (curSizeRollBackups == maxSizeRollBackups) + //Validate curSizeRollBackups + curSizeRollBackups = countFileIndex(fileName); + // decrement to offset the later increment + curSizeRollBackups--; + } + + // If we are not keeping an infinite set of backups the delete oldest + if (maxSizeRollBackups > 0) + { + LogLog.debug("CD=-1,curSizeRollBackups:"+curSizeRollBackups); + LogLog.debug("CD=-1,maxSizeRollBackups:"+maxSizeRollBackups); + + // Delete the oldest file. + // curSizeRollBackups is never -1 so infinite backups are ok here + if ((curSizeRollBackups - maxSizeRollBackups) >= 0) { - deleteFile(fileName + '.' + maxSizeRollBackups); + //The oldest file is the one with the largest number + // as the 0 is always fileName + // which moves to fileName.1 etc. + LogLog.debug("CD=-1,deleteFile:"+curSizeRollBackups); + deleteFile(fileName + '.' + curSizeRollBackups); + // decrement to offset the later increment curSizeRollBackups--; } + } + // Map {(maxBackupIndex - 1), ..., 2, 1} to {maxBackupIndex, ..., 3, 2} + for (int i = curSizeRollBackups; i >= 1; i--) + { + String oldName = (fileName + "." + i); + String newName = (fileName + '.' + (i + 1)); - // Map {(maxBackupIndex - 1), ..., 2, 1} to {maxBackupIndex, ..., 3, 2} - for (int i = curSizeRollBackups; i >= 1; i--) + // Ensure that when compressing we rename the compressed archives + if (compress) { - rollFile((fileName + "." + i), (fileName + '.' + (i + 1)), false); + rollFile(oldName + COMPRESS_EXTENSION, newName + COMPRESS_EXTENSION, false); } + else + { + rollFile(oldName, newName, false); + } + } - curSizeRollBackups++; - // Rename fileName to fileName.1 - rollFile(fileName, fileName + ".1", compress); + curSizeRollBackups++; + // Rename fileName to fileName.1 + rollFile(fileName, fileName + ".1", compress); - } // REMOVE This code branching for Alexander Cerna's request - else if (countDirection == 0) + } // REMOVE This code branching for Alexander Cerna's request + else if (countDirection == 0) + { + // rollFile based on date pattern + now.setTime(System.currentTimeMillis()); + String newFile = fileName + sdf.format(now); + + // If we haven't rolled yet then validate we have the right value + // for curSizeRollBackups + if (curSizeRollBackups == 0) { - // rollFile based on date pattern - curSizeRollBackups++; - now.setTime(System.currentTimeMillis()); - scheduledFilename = fileName + sdf.format(now); - rollFile(fileName, scheduledFilename, compress); + //Validate curSizeRollBackups + curSizeRollBackups = countFileIndex(newFile); + // to balance the increment just coming up. as the count returns + // the next free number not the last used. + curSizeRollBackups--; } - else - { // countDirection > 0 - if ((curSizeRollBackups >= maxSizeRollBackups) && (maxSizeRollBackups > 0)) + + // If we are not keeping an infinite set of backups the delete oldest + if (maxSizeRollBackups > 0) + { + // Don't prune older files if they exist just go for the last + // one based on our maxSizeRollBackups. This means we may have + // more files left on disk that maxSizeRollBackups if this value + // is adjusted between runs but that is an acceptable state. + // Otherwise we would have to check on startup that we didn't + // have more than maxSizeRollBackups and prune then. + + if (((curSizeRollBackups - maxSizeRollBackups) >= 0)) { - // delete the first and keep counting up. + LogLog.debug("CD=0,curSizeRollBackups:"+curSizeRollBackups); + LogLog.debug("CD=0,maxSizeRollBackups:"+maxSizeRollBackups); + + // delete the first and keep counting up. int oldestFileIndex = curSizeRollBackups - maxSizeRollBackups + 1; - deleteFile(fileName + '.' + oldestFileIndex); + LogLog.debug("CD=0,deleteFile:"+oldestFileIndex); + deleteFile(newFile + '.' + oldestFileIndex); } + } + + + String finalName = newFile; + + curSizeRollBackups++; + + // Add rollSize if it is > 0 + if (curSizeRollBackups > 0 ) + { + finalName = newFile + '.' + curSizeRollBackups; + + } - if (staticLogFileName) + rollFile(fileName, finalName, compress); + } + else + { // countDirection > 0 + // If we haven't rolled yet then validate we have the right value + // for curSizeRollBackups + if (curSizeRollBackups == 0) + { + //Validate curSizeRollBackups + curSizeRollBackups = countFileIndex(fileName); + // to balance the increment just coming up. as the count returns + // the next free number not the last used. + curSizeRollBackups--; + } + + // If we are not keeping an infinite set of backups the delete oldest + if (maxSizeRollBackups > 0) + { + LogLog.debug("CD=1,curSizeRollBackups:"+curSizeRollBackups); + LogLog.debug("CD=1,maxSizeRollBackups:"+maxSizeRollBackups); + + // Don't prune older files if they exist just go for the last + // one based on our maxSizeRollBackups. This means we may have + // more files left on disk that maxSizeRollBackups if this value + // is adjusted between runs but that is an acceptable state. + // Otherwise we would have to check on startup that we didn't + // have more than maxSizeRollBackups and prune then. + + if (((curSizeRollBackups - maxSizeRollBackups) >= 0)) { - curSizeRollBackups++; - rollFile(fileName, fileName + '.' + curSizeRollBackups, compress); + // delete the first and keep counting up. + int oldestFileIndex = curSizeRollBackups - maxSizeRollBackups + 1; + LogLog.debug("CD=1,deleteFile:"+oldestFileIndex); + deleteFile(fileName + '.' + oldestFileIndex); } - else + } + + + curSizeRollBackups++; + + rollFile(fileName, fileName + '.' + curSizeRollBackups, compress); + + } + LogLog.debug("CD="+countDirection+",done"); + } + + + private int countFileIndex(String fileName) + { + return countFileIndex(fileName, true); + } + /** + * Use filename as a base name and find what count number we are up to by + * looking at the files in this format: + * + * <filename>.<count>[COMPRESS_EXTENSION] + * + * If a count value of 1 cannot be found then a directory listing is + * performed to try and identify if there is a valid value for <count>. + * + * + * @param fileName the basefilename to use + * @param checkBackupLocation should backupFilesToPath location be checked for existing backups + * @return int the next free index + */ + private int countFileIndex(String fileName, boolean checkBackupLocation) + { + String testFileName; + + // It is possible for index 1..n to be missing leaving n+1..n+1+m logs + // in this scenario we should still return n+1+m+1 + int index=1; + + testFileName = fileName + "." + index; + + // Bail out early if there is a problem with the file + if (new File(testFileName) == null + || new File(testFileName + COMPRESS_EXTENSION) == null) + + { + return index; + } + + // Check that we do not have the 1..n missing scenario + if (!(new File(testFileName).exists() + || new File(testFileName + COMPRESS_EXTENSION).exists())) + + { + int max=0; + String prunedFileName = new File(fileName).getName(); + + // Look through all files to find next index + if (new File(fileName).getParentFile() != null) + { + for (File file : new File(fileName).getParentFile().listFiles()) { - if (compress) + String name = file.getName(); + + if (name.startsWith(prunedFileName) && !name.equals(prunedFileName)) { - compress(fileName); + String parsedCount = name.substring(prunedFileName.length() + 1); + + if (parsedCount.endsWith(COMPRESS_EXTENSION)) + { + parsedCount = parsedCount.substring(0, parsedCount.indexOf(COMPRESS_EXTENSION)); + } + + try + { + max = Integer.parseInt(parsedCount); + + // if we got a good value then update our index value. + if (max > index) + { + // +1 as we want to return the next free value. + index = max + 1; + } + } + catch (NumberFormatException nfe) + { + //ignore it assume file doesn't exist. + } } } } + + // Update testFileName + testFileName = fileName + "." + index; } - try + + while (new File(testFileName).exists() + || new File(testFileName + COMPRESS_EXTENSION).exists()) { - // This will also close the file. This is OK since multiple - // close operations are safe. - this.setFile(baseFileName, false); + index++; + testFileName = fileName + "." + index; } - catch (IOException e) + + if (checkBackupLocation && index == 1 && backupFilesToPath != null) { - LogLog.error("setFile(" + fileName + ", false) call failed.", e); + LogLog.debug("Trying backup location:"+backupFilesToPath + System.getProperty("file.separator") + fileName); + return countFileIndex(backupFilesToPath + System.getProperty("file.separator") + new File(fileName).getName(), false); } + + return index; } protected synchronized void doCompress(File from, File to) @@ -907,11 +994,11 @@ public class QpidCompositeRollingAppender extends FileAppender String toFile; if (backupFilesToPath == null) { - toFile = to.getPath() + ".gz"; + toFile = to.getPath() + COMPRESS_EXTENSION; } else { - toFile = backupFilesToPath + System.getProperty("file.separator") + to.getName() + ".gz"; + toFile = backupFilesToPath + System.getProperty("file.separator") + to.getName() + COMPRESS_EXTENSION; } File target = new File(toFile); diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java index 3d5d99f0b0..825aa9795e 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java @@ -1689,15 +1689,28 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener while (queueListIterator.advance()) { QueueEntry node = queueListIterator.getNode(); - if (!node.isDeleted() && node.expired() && node.acquire()) + // Only process nodes that are not currently deleted + if (!node.isDeleted()) { - dequeueEntry(node); - } - else - { - if(_managedObject!=null) + // If the node has exired then aquire it + if (node.expired() && node.acquire()) + { + // Then dequeue it. + dequeueEntry(node); + } + else { - _managedObject.checkForNotification(node.getMessage()); + if (_managedObject != null) + { + // There is a chance that the node could be deleted by + // the time the check actually occurs. So verify we + // can actually get the message to perform the check. + ServerMessage msg = node.getMessage(); + if (msg != null) + { + _managedObject.checkForNotification(msg); + } + } } } } diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java index 6826dcf324..78deeeb164 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java @@ -268,10 +268,11 @@ public class VirtualHostImpl implements Accessable, VirtualHost { q.checkMessageStatus(); } - catch (AMQException e) + catch (Exception e) { _logger.error("Exception in housekeeping for queue: " + q.getName().toString(), e); - throw new RuntimeException(e); + //Don't throw exceptions as this will stop the + // house keeping task from running. } } } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractBytesMessage.java b/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractBytesMessage.java index 234212c301..535b55ced5 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractBytesMessage.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractBytesMessage.java @@ -85,11 +85,18 @@ public abstract class AbstractBytesMessage extends AbstractJMSMessage } public String toBodyString() throws JMSException - { - checkReadable(); + { try { - return Functions.str(_data.buf(), 100); + if (_data != null) + { + return Functions.str(_data.buf(), 100,0); + } + else + { + return ""; + } + } catch (Exception e) { diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/util/Functions.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/util/Functions.java index c220694b50..9f1c0ca9eb 100644 --- a/qpid/java/common/src/main/java/org/apache/qpid/transport/util/Functions.java +++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/util/Functions.java @@ -57,12 +57,17 @@ public class Functions public static final String str(ByteBuffer buf, int limit) { + return str(buf, limit,buf.position()); + } + + public static final String str(ByteBuffer buf, int limit,int start) + { StringBuilder str = new StringBuilder(); str.append('"'); for (int i = 0; i < min(buf.remaining(), limit); i++) { - byte c = buf.get(buf.position() + i); + byte c = buf.get(start + i); if (c > 31 && c < 127 && c != '\\') { diff --git a/qpid/java/management/tools/qpid-cli/src/org/apache/qpid/commands/Commandviewcontent.java b/qpid/java/management/tools/qpid-cli/src/org/apache/qpid/commands/Commandviewcontent.java index a0843b91fa..0c8a4b62b4 100644 --- a/qpid/java/management/tools/qpid-cli/src/org/apache/qpid/commands/Commandviewcontent.java +++ b/qpid/java/management/tools/qpid-cli/src/org/apache/qpid/commands/Commandviewcontent.java @@ -39,10 +39,7 @@ public class Commandviewcontent extends CommandImpl public static final String COMMAND_NAME = "viewcontent"; - private String object; - private String name; - private String vhost; - private int number = 0; + private long number = 0; private QueueObject objname; private MBeanServerConnection mbsc; private String method1; @@ -61,7 +58,7 @@ public class Commandviewcontent extends CommandImpl { Set set = null; Object temp[] = { null }; - objname.setQueryString(this.object, this.name, this.vhost); + objname.setQueryString(getObject(), getName(), getVirtualhost()); set = objname.returnObjects(); String temp_header = "", header = "", message_data = "", encoding = null; @@ -233,8 +230,7 @@ public class Commandviewcontent extends CommandImpl private void setnumber(String number) { - Integer i = new Integer(number); - this.number = i.intValue(); + this.number = Long.valueOf(number); } private static String removeSpaces(String s) @@ -246,7 +242,7 @@ public class Commandviewcontent extends CommandImpl return t; } - public int getnumber() + public long getnumber() { return this.number; } diff --git a/qpid/java/management/tools/qpid-cli/src/org/apache/qpid/commands/objects/ObjectNames.java b/qpid/java/management/tools/qpid-cli/src/org/apache/qpid/commands/objects/ObjectNames.java index 7791b84a4f..1432f18018 100644 --- a/qpid/java/management/tools/qpid-cli/src/org/apache/qpid/commands/objects/ObjectNames.java +++ b/qpid/java/management/tools/qpid-cli/src/org/apache/qpid/commands/objects/ObjectNames.java @@ -21,8 +21,10 @@ package org.apache.qpid.commands.objects; +import java.util.HashMap; import java.util.Iterator; import java.util.List; +import java.util.Map; import java.util.Set; import javax.management.Attribute; @@ -31,6 +33,8 @@ import javax.management.MBeanInfo; import javax.management.MBeanServerConnection; import javax.management.ObjectName; +import org.apache.qpid.management.common.mbeans.ManagedQueue; + public class ObjectNames { public String querystring = null; @@ -38,6 +42,16 @@ public class ObjectNames public Set<ObjectName> set = null; public String attributes = ""; public String attributevalues = "";// = null; + + private static final Map<String,Integer> QUEUE_ATTRIBUTES = new HashMap<String,Integer>(); + static + { + QUEUE_ATTRIBUTES.put(ManagedQueue.ATTR_NAME, 1); + QUEUE_ATTRIBUTES.put(ManagedQueue.ATTR_DURABLE, 2); + QUEUE_ATTRIBUTES.put(ManagedQueue.ATTR_ACTIVE_CONSUMER_COUNT, 3); + QUEUE_ATTRIBUTES.put(ManagedQueue.ATTR_MSG_COUNT, 4); + QUEUE_ATTRIBUTES.put(ManagedQueue.ATTR_RCVD_MSG_COUNT, 5); + } public ObjectNames(MBeanServerConnection mbsc) { @@ -318,14 +332,12 @@ public class ObjectNames Iterator it = set.iterator(); String line = ""; int iterator = 0; - int attr_count = 0; String temp1 = ""; String temp2 = ""; try { do { - attr_count = 0; ObjectName temp_object = null; if (it.hasNext()) { @@ -349,7 +361,13 @@ public class ObjectNames for (MBeanAttributeInfo attr : attr_info) { Object toWrite = null; - attr_count++; + Integer attr_count = QUEUE_ATTRIBUTES.get(attr.getName()); + + if(attr_count == null) + { + continue; + } + try { toWrite = mbsc.getAttribute(temp_object, attr.getName()); @@ -358,7 +376,7 @@ public class ObjectNames switch (attr_count) { case 1: - case 3: + case 2: temp1 = attr.getName(); while (temp1.length() < 10) temp1 = " " + temp1; @@ -368,7 +386,7 @@ public class ObjectNames temp2 = " " + temp2; attributevalues = attributevalues + temp2 + "|"; break; - case 6: + case 3: temp1 = attr.getName(); while (temp1.length() < 20) temp1 = " " + temp1; @@ -378,7 +396,7 @@ public class ObjectNames temp2 = " " + temp2; attributevalues = attributevalues + temp2 + "|"; break; - case 7: + case 4: temp1 = attr.getName(); while (temp1.length() < 13) temp1 = " " + temp1; @@ -388,7 +406,7 @@ public class ObjectNames temp2 = " " + temp2; attributevalues = attributevalues + temp2 + "|"; break; - case 9: + case 5: temp1 = attr.getName(); while (temp1.length() < 20) temp1 = " " + temp1; @@ -402,29 +420,10 @@ public class ObjectNames } else if (output.compareToIgnoreCase("csv") == 0) { - switch (attr_count) - { - case 1: - case 3: - case 6: - temp1 = attr.getName(); - attributes = attributes + temp1 + seperator; - temp2 = toWrite.toString(); - attributevalues = attributevalues + temp2 + seperator; - break; - case 7: - temp1 = attr.getName(); - attributes = attributes + temp1 + seperator; - temp2 = toWrite.toString(); - attributevalues = attributevalues + temp2 + seperator; - break; - case 9: - temp1 = attr.getName(); - attributes = attributes + temp1 + seperator; - temp2 = toWrite.toString(); - attributevalues = attributevalues + temp2 + seperator; - break; - } + temp1 = attr.getName(); + attributes = attributes + temp1 + seperator; + temp2 = toWrite.toString(); + attributevalues = attributevalues + temp2 + seperator; } else { diff --git a/qpid/java/management/tools/qpid-cli/src/org/apache/qpid/commands/objects/QueueObject.java b/qpid/java/management/tools/qpid-cli/src/org/apache/qpid/commands/objects/QueueObject.java index 7c82682c27..fcf0464ced 100644 --- a/qpid/java/management/tools/qpid-cli/src/org/apache/qpid/commands/objects/QueueObject.java +++ b/qpid/java/management/tools/qpid-cli/src/org/apache/qpid/commands/objects/QueueObject.java @@ -22,10 +22,10 @@ package org.apache.qpid.commands.objects; import javax.management.MBeanServerConnection; -import javax.management.MBeanAttributeInfo; -import javax.management.MBeanInfo; import javax.management.ObjectName; +import org.apache.qpid.management.common.mbeans.ManagedQueue; + public class QueueObject extends ObjectNames { public QueueObject(MBeanServerConnection mbsc) @@ -47,33 +47,12 @@ public class QueueObject extends ObjectNames public int getmessagecount(ObjectName queue) { - int attr_count = 0; - String value; - Integer depth = null; - + Number depth = null; + try { - MBeanInfo bean_info; - bean_info = mbsc.getMBeanInfo(queue); - MBeanAttributeInfo[] attr_info = bean_info.getAttributes(); - if (attr_info == null) - return 0; - else - { - for (MBeanAttributeInfo attr : attr_info) - { - Object toWrite = null; - attr_count++; - toWrite = mbsc.getAttribute(queue, attr.getName()); - if (attr_count == 7) - { - value = toWrite.toString(); - depth = new Integer(value); - } - } - - } - + depth = (Number) mbsc.getAttribute(queue, ManagedQueue.ATTR_MSG_COUNT); + } catch (Exception ex) { diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/ProducerFlowControlTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/ProducerFlowControlTest.java index ecb2f7d559..d139f8d8b4 100644 --- a/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/ProducerFlowControlTest.java +++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/ProducerFlowControlTest.java @@ -407,6 +407,11 @@ public class ProducerFlowControlTest extends AbstractTestLogging consumerConnection.start(); consumer.receive(); + + //perform a synchronous op on the connection + ((AMQSession) consumerSession).declareExchange( + new AMQShortString("amq.direct"), new AMQShortString("direct"), false); + assertFalse("Queue should not be overfull", queueMBean.isFlowOverfull()); consumer.receive(); diff --git a/qpid/java/test-profiles/Excludes b/qpid/java/test-profiles/Excludes index 47ae6d09ba..e816222d33 100644 --- a/qpid/java/test-profiles/Excludes +++ b/qpid/java/test-profiles/Excludes @@ -29,4 +29,10 @@ org.apache.qpid.test.unit.ack.AcknowledgeAfterFailoverTest#testClientAck org.apache.qpid.test.unit.ack.AcknowledgeAfterFailoverOnMessageTest#* org.apache.qpid.test.unit.ack.AcknowledgeAfterFailoverTest#* - +// The following tests exibit random failures and are temporarily excluded. +// QPID-2224 Random Test failures - The test needs to be fixed. +MINANetworkDriverTest#* +// QPID-2225 Random Test failures against the in-vm broker. +FailoverBeforeConsumingRecover#* +// QPID-2262 Random test failures - The test needs to be fixed. +SimpleACLTest#testClientPublishInvalidQueueSuccess diff --git a/qpid/java/test-profiles/cpp.testprofile b/qpid/java/test-profiles/cpp.testprofile index 0f7a718a8b..04407accc3 100644 --- a/qpid/java/test-profiles/cpp.testprofile +++ b/qpid/java/test-profiles/cpp.testprofile @@ -15,5 +15,5 @@ broker.args= broker=${broker.executable} -p @PORT --data-dir ${build.data}/@PORT -t --auth no --no-module-dir ${broker.modules} ${broker.args} -profile.excludes=010PrefetchExcludes 010TransientExcludes -test.excludes=Excludes 010Excludes ${profile}.excludes ${profile.excludes} cpp.excludes +profile.excludes=CPPPrefetchExcludes CPPTransientExcludes +test.excludes=Excludes CPPExcludes ${profile}.excludes ${profile.excludes} cpp.excludes diff --git a/qpid/java/testkit/bin/setenv.sh b/qpid/java/testkit/bin/setenv.sh index 0449ecffc7..71bb219dcf 100644 --- a/qpid/java/testkit/bin/setenv.sh +++ b/qpid/java/testkit/bin/setenv.sh @@ -17,6 +17,10 @@ # under the License. # +# If QPIDD_EXEC ..etc is not set, it will first check to see +# if this is run from a qpid svn check out, if not it will look +# for installed rpms. + abs_path() { D=`dirname "$1"` @@ -25,24 +29,45 @@ abs_path() } # Environment for python tests -test -d ../../../python || { echo "WARNING: skipping test, no python directory."; exit 0; } -PYTHON_DIR=../../../python -PYTHONPATH=$PYTHON_DIR:$PYTHON_DIR/qpid + +if [ -d ../../../python ] ; then + PYTHON_DIR=../../../python + PYTHONPATH=$PYTHON_DIR:$PYTHON_DIR/qpid +elif [ -z `echo $PYTHONPATH | awk '$0 ~ /qpid/'` ]; then + echo "WARNING: skipping test, no qpid python scripts found ."; exit 0; +fi + if [ "$QPIDD_EXEC" = "" ] ; then - test -x ../../../cpp/src/qpidd || { echo "WARNING: skipping test, QPIDD_EXEC not set and qpidd not found."; exit 0; } - QPIDD_EXEC=`abs_path "../../../cpp/src/qpidd"` - #QPIDD_EXEC=/opt/workspace/qpid/trunk/qpid/cpp/src/.libs/lt-qpidd + if [ -x ../../../cpp/src/qpidd ]; then + QPIDD_EXEC=`abs_path "../../../cpp/src/qpidd"` + elif [ -n "$(which qpidd)" ] ; then + QPIDD_EXEC=$(which qpidd) + else + echo "WARNING: skipping test, QPIDD_EXEC not set and qpidd not found."; exit 0; + fi fi if [ "$CLUSTER_LIB" = "" ] ; then - test -x ../../../cpp/src/.libs/cluster.so || { echo "WARNING: skipping test, CLUSTER_LIB not set and cluster.so not found."; exit 0; } - CLUSTER_LIB=`abs_path "../../../cpp/src/.libs/cluster.so"` - #CLUSTER_LIB=/opt/workspace/qpid/trunk/qpid/cpp/src/.libs/cluster.so + if [ -x ../../../cpp/src/.libs/cluster.so ]; then + CLUSTER_LIB=`abs_path "../../../cpp/src/.libs/cluster.so"` + elif [ -e /usr/lib64/qpid/daemon/cluster.so ] ; then + CLUSTER_LIB="/usr/lib64/qpid/daemon/cluster.so" + elif [ -e /usr/lib/qpid/daemon/cluster.so ] ; then + CLUSTER_LIB="/usr/lib/qpid/daemon/cluster.so" + else + echo "WARNING: skipping test, CLUSTER_LIB not set and cluster.so not found."; exit 0; + fi fi if [ "$QP_CP" = "" ] ; then - QP_JAR_PATH=`abs_path "../../build/lib/"` + if [ -d ../../build/lib/ ]; then + QP_JAR_PATH=`abs_path "../../build/lib/"` + elif [ -d /usr/share/java/qpid-deps ]; then + QP_JAR_PATH=`abs_path "/usr/share/java"` + else + "WARNING: skipping test, QP_CP not set and the Qpid jars are not present."; exit 0; + fi QP_CP=`find $QP_JAR_PATH -name '*.jar' | tr '\n' ':'` fi @@ -50,5 +75,4 @@ if [ "$OUTDIR" = "" ] ; then OUTDIR=`abs_path "../output"` fi - export PYTHONPATH PYTHON_DIR QPIDD_EXEC CLUSTER_LIB QP_CP OUTDIR diff --git a/qpid/python/Makefile b/qpid/python/Makefile index b5e48653d0..cc493a155e 100644 --- a/qpid/python/Makefile +++ b/qpid/python/Makefile @@ -50,7 +50,7 @@ build: $(TARGETS) doc: @mkdir -p $(BUILD) - epydoc qpid.messaging -o $(BUILD)/doc --no-private --no-sourcecode --include-log + PYTHONPATH=. epydoc qpid.messaging -o $(BUILD)/doc --no-private --no-sourcecode --include-log epydoc qmf.qmfCommon qmf.qmfConsole -o $(BUILD)/doc/qmf --no-private --no-sourcecode --include-log install: build diff --git a/qpid/python/qpid/address.py b/qpid/python/qpid/address.py index 5c675b8782..6228ac757b 100644 --- a/qpid/python/qpid/address.py +++ b/qpid/python/qpid/address.py @@ -24,6 +24,8 @@ l = Lexicon() LBRACE = l.define("LBRACE", r"\{") RBRACE = l.define("RBRACE", r"\}") +LBRACK = l.define("LBRACK", r"\[") +RBRACK = l.define("RBRACK", r"\]") COLON = l.define("COLON", r":") SEMI = l.define("SEMI", r";") SLASH = l.define("SLASH", r"/") @@ -128,8 +130,30 @@ class AddressParser(Parser): return tok2obj(self.eat()) elif self.matches(LBRACE): return self.map() + elif self.matches(LBRACK): + return self.list() else: - raise ParseError(self.next(), NUMBER, STRING, ID, LBRACE) + raise ParseError(self.next(), NUMBER, STRING, ID, LBRACE, LBRACK) + + def list(self): + self.eat(LBRACK) + + result = [] + + while True: + if self.matches(RBRACK): + break + else: + result.append(self.value()) + if self.matches(COMMA): + self.eat(COMMA) + elif self.matches(RBRACK): + break + else: + raise ParseError(self.next(), COMMA, RBRACK) + + self.eat(RBRACK) + return result def parse(addr): return AddressParser(lex(addr)).parse() diff --git a/qpid/python/qpid/brokertest.py b/qpid/python/qpid/brokertest.py index 9fa79a220b..83d6c44d84 100644 --- a/qpid/python/qpid/brokertest.py +++ b/qpid/python/qpid/brokertest.py @@ -259,15 +259,15 @@ class Cluster: self.args += [ "--load-module", BrokerTest.cluster_lib ] self.start_n(count, expect=expect, wait=wait) - def start(self, name=None, expect=EXPECT_RUNNING, wait=True): + def start(self, name=None, expect=EXPECT_RUNNING, wait=True, args=[]): """Add a broker to the cluster. Returns the index of the new broker.""" if not name: name="%s-%d" % (self.name, len(self._brokers)) log.debug("Cluster %s starting member %s" % (self.name, name)) - self._brokers.append(self.test.broker(self.args, name, expect, wait)) + self._brokers.append(self.test.broker(self.args+args, name, expect, wait)) return self._brokers[-1] - def start_n(self, count, expect=EXPECT_RUNNING, wait=True): - for i in range(count): self.start(expect=expect, wait=wait) + def start_n(self, count, expect=EXPECT_RUNNING, wait=True, args=[]): + for i in range(count): self.start(expect=expect, wait=wait, args=args) # Behave like a list of brokers. def __len__(self): return len(self._brokers) @@ -289,6 +289,7 @@ class BrokerTest(TestCase): receiver_exec = os.getenv("RECEIVER_EXEC") sender_exec = os.getenv("SENDER_EXEC") store_lib = os.getenv("STORE_LIB") + test_store_lib = os.getenv("TEST_STORE_LIB") rootdir = os.getcwd() def configure(self, config): self.config=config diff --git a/qpid/python/qpid/driver.py b/qpid/python/qpid/driver.py index 3cc69921bb..2851c3aad3 100644 --- a/qpid/python/qpid/driver.py +++ b/qpid/python/qpid/driver.py @@ -605,6 +605,7 @@ class Driver: cmd = ExchangeDeclare(exchange=name, durable=durable) elif type == "queue": cmd = QueueDeclare(queue=name, durable=durable) + bindings = xprops.pop("bindings", []) else: return ("unrecognized type, must be topic or queue: %s" % type,) @@ -621,9 +622,20 @@ class Driver: else: subtype = None + cmds = [cmd] + if type == "queue": + for b in bindings: + try: + n, s, o = address.parse(b) + except address.ParseError, e: + return (e,) + cmds.append(ExchangeBind(name, n, s, o)) + + for c in cmds[:-1]: + sst.write_cmd(c) def do_action(): action(type, subtype) - sst.write_cmd(cmd, do_action) + sst.write_cmd(cmds[-1], do_action) def delete(self, sst, name, action): def do_delete(er, qr): @@ -840,6 +852,7 @@ class Driver: msg.reply_to = reply_to2addr(mp.reply_to) msg.correlation_id = mp.correlation_id msg.durable = dp.delivery_mode == delivery_mode.persistent + msg.redelivered = dp.redelivered msg.properties = mp.application_headers msg.content_type = mp.content_type msg._transfer_id = xfr.id diff --git a/qpid/python/qpid/messaging.py b/qpid/python/qpid/messaging.py index 5dd8b22ee3..4f2c190ce2 100644 --- a/qpid/python/qpid/messaging.py +++ b/qpid/python/qpid/messaging.py @@ -787,6 +787,7 @@ class Message: self.reply_to = reply_to self.correlation_id = correlation_id self.durable = durable + self.redelivered = False if properties is None: self.properties = {} else: diff --git a/qpid/python/qpid/tests/address.py b/qpid/python/qpid/tests/address.py index f772425e42..7c101eee5e 100644 --- a/qpid/python/qpid/tests/address.py +++ b/qpid/python/qpid/tests/address.py @@ -145,7 +145,7 @@ class AddressTests(ParserBase, Test): def testBadOptions3(self): self.invalid("name/subject; { key:", - "expecting (NUMBER, STRING, ID, LBRACE), got EOF " + "expecting (NUMBER, STRING, ID, LBRACE, LBRACK), got EOF " "line:1,20:name/subject; { key:") def testBadOptions4(self): @@ -167,3 +167,33 @@ class AddressTests(ParserBase, Test): self.invalid("name/subject; { key: value } asdf", "expecting EOF, got ID('asdf') " "line:1,29:name/subject; { key: value } asdf") + + def testList1(self): + self.valid("name/subject; { key: [] }", "name", "subject", {"key": []}) + + def testList2(self): + self.valid("name/subject; { key: ['one'] }", "name", "subject", {"key": ['one']}) + + def testList3(self): + self.valid("name/subject; { key: [1, 2, 3] }", "name", "subject", + {"key": [1, 2, 3]}) + + def testList4(self): + self.valid("name/subject; { key: [1, [2, 3], 4] }", "name", "subject", + {"key": [1, [2, 3], 4]}) + + def testBadList1(self): + self.invalid("name/subject; { key: [ }", "expecting (NUMBER, STRING, ID, LBRACE, LBRACK), " + "got RBRACE('}') line:1,23:name/subject; { key: [ }") + + def testBadList2(self): + self.invalid("name/subject; { key: [ 1 }", "expecting (COMMA, RBRACK), " + "got RBRACE('}') line:1,25:name/subject; { key: [ 1 }") + + def testBadList3(self): + self.invalid("name/subject; { key: [ 1 2 }", "expecting (COMMA, RBRACK), " + "got NUMBER('2') line:1,25:name/subject; { key: [ 1 2 }") + + def testBadList4(self): + self.invalid("name/subject; { key: [ 1 2 ] }", "expecting (COMMA, RBRACK), " + "got NUMBER('2') line:1,25:name/subject; { key: [ 1 2 ] }") diff --git a/qpid/python/qpid/tests/messaging.py b/qpid/python/qpid/tests/messaging.py index cbfbc5a56d..f2a270192e 100644 --- a/qpid/python/qpid/tests/messaging.py +++ b/qpid/python/qpid/tests/messaging.py @@ -642,6 +642,28 @@ class AddressTests(Base): # XXX: need to figure out close after error self.conn._remove_session(self.ssn) + def testBindings(self): + snd = self.ssn.sender(""" +test-bindings-queue; { + create: always, + delete: always, + node-properties: { + x-properties: { + bindings: ["amq.topic/a.#", "amq.direct/b", "amq.topic/c.*"] + } + } +} +""") + snd.send("one") + snd_a = self.ssn.sender("amq.topic/a.foo") + snd_b = self.ssn.sender("amq.direct/b") + snd_c = self.ssn.sender("amq.topic/c.bar") + snd_a.send("two") + snd_b.send("three") + snd_c.send("four") + rcv = self.ssn.receiver("test-bindings-queue") + self.drain(rcv, expected=["one", "two", "three", "four"]) + NOSUCH_Q = "this-queue-should-not-exist" UNPARSEABLE_ADDR = "name/subject; {bad options" UNLEXABLE_ADDR = "\0x0\0x1\0x2\0x3" diff --git a/qpid/python/todo.txt b/qpid/python/todo.txt index 54c8f84484..7b3ca90638 100644 --- a/qpid/python/todo.txt +++ b/qpid/python/todo.txt @@ -44,12 +44,6 @@ Connection: - connected: F, NR - - start: F, NR, ND - + see comment on Receiver.start - - - stop: F, NR, ND - + see comment on Receiver.start - - close: F, NR, ND + currently there is no distinction between a "close" that does a complete handshake with the remote broker, and a "close" @@ -90,16 +84,7 @@ Session: - rollback: F, NR - - start: F, NR, ND - + see comment on Receiver.start - - - stop: F, NR, ND - + see comment on Receiver.start - - - fetch: NF - + this method if added would permit the listener functionality - to be removed as it would become trivial to replicate via user - code + - next_receiver: F, NR - close: F, ND + see comment on Connection.close @@ -146,18 +131,6 @@ Receiver: + changing capacity/prefetch to issue credit on ack rather than fetch return - - start: F, NR, ND - + when a receiver is "stopped", it is treated as if the receiver - has zero capacity even if the capacity field is set to a non - zero value, this is a mild convenience since you can stop and - then start a receiver without having to remember what the - capacity was before, however it is unclear if this is really - worthwhile since stop/start is entirely equivalent to setting - capacity to a zero/non-zero value - - - stop: F, NR, ND - + see comment on Receiver.start - - sync/wait: NF - close: F, NR @@ -185,10 +158,10 @@ Message: Address: - syntax: F, NR - - subject related changes, e.g. allowing patterns on both ends: NF, ND - - creating/deleting queues/exchanges - + need to handle cleanup of temp queues/topics: NF - + passthrough options for creating exchanges/queues: NF + - subject related changes, e.g. allowing patterns on both ends: NF + - creating/deleting queues/exchanges F, NR + + need to handle cleanup of temp queues/topics: F, NR + + passthrough options for creating exchanges/queues: F, NR - integration with java: NF - queue browsing: NF - temporary queues: NF diff --git a/qpid/wcf/QpidWcf.sln b/qpid/wcf/QpidWcf.sln index 54d8dd7b0b..9f83cf553a 100644 --- a/qpid/wcf/QpidWcf.sln +++ b/qpid/wcf/QpidWcf.sln @@ -34,79 +34,44 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Channel", "src\Apache\Qpid\ EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
- Debug|Any CPU = Debug|Any CPU
- Debug|Mixed Platforms = Debug|Mixed Platforms
Debug|Win32 = Debug|Win32
Debug|x64 = Debug|x64
- Debug|x86 = Debug|x86
- Release|Any CPU = Release|Any CPU
- Release|Mixed Platforms = Release|Mixed Platforms
Release|Win32 = Release|Win32
Release|x64 = Release|x64
- Release|x86 = Release|x86
EndGlobalSection
GlobalSection(ProjectConfigurationPlatforms) = postSolution
- {820BFC34-A40F-46BA-B86B-05334854CA17}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
- {820BFC34-A40F-46BA-B86B-05334854CA17}.Debug|Any CPU.Build.0 = Debug|Any CPU
- {820BFC34-A40F-46BA-B86B-05334854CA17}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU
- {820BFC34-A40F-46BA-B86B-05334854CA17}.Debug|Mixed Platforms.Build.0 = Debug|Any CPU
{820BFC34-A40F-46BA-B86B-05334854CA17}.Debug|Win32.ActiveCfg = Debug|Any CPU
+ {820BFC34-A40F-46BA-B86B-05334854CA17}.Debug|Win32.Build.0 = Debug|Any CPU
{820BFC34-A40F-46BA-B86B-05334854CA17}.Debug|x64.ActiveCfg = Debug|Any CPU
- {820BFC34-A40F-46BA-B86B-05334854CA17}.Debug|x86.ActiveCfg = Debug|Any CPU
- {820BFC34-A40F-46BA-B86B-05334854CA17}.Release|Any CPU.ActiveCfg = Release|Any CPU
- {820BFC34-A40F-46BA-B86B-05334854CA17}.Release|Any CPU.Build.0 = Release|Any CPU
- {820BFC34-A40F-46BA-B86B-05334854CA17}.Release|Mixed Platforms.ActiveCfg = Release|Any CPU
- {820BFC34-A40F-46BA-B86B-05334854CA17}.Release|Mixed Platforms.Build.0 = Release|Any CPU
+ {820BFC34-A40F-46BA-B86B-05334854CA17}.Debug|x64.Build.0 = Debug|Any CPU
{820BFC34-A40F-46BA-B86B-05334854CA17}.Release|Win32.ActiveCfg = Release|Any CPU
+ {820BFC34-A40F-46BA-B86B-05334854CA17}.Release|Win32.Build.0 = Release|Any CPU
{820BFC34-A40F-46BA-B86B-05334854CA17}.Release|x64.ActiveCfg = Release|Any CPU
- {820BFC34-A40F-46BA-B86B-05334854CA17}.Release|x86.ActiveCfg = Release|Any CPU
- {C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Debug|Any CPU.ActiveCfg = Debug|Win32
- {C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Debug|Any CPU.Build.0 = Debug|Win32
- {C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Debug|Mixed Platforms.ActiveCfg = Debug|Win32
- {C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Debug|Mixed Platforms.Build.0 = Debug|Win32
+ {820BFC34-A40F-46BA-B86B-05334854CA17}.Release|x64.Build.0 = Release|Any CPU
{C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Debug|Win32.ActiveCfg = Debug|Win32
{C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Debug|Win32.Build.0 = Debug|Win32
- {C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Debug|x64.ActiveCfg = Debug|Win32
- {C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Debug|x86.ActiveCfg = Debug|Win32
- {C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Release|Any CPU.ActiveCfg = Release|Win32
- {C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Release|Mixed Platforms.ActiveCfg = Release|Win32
- {C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Release|Mixed Platforms.Build.0 = Release|Win32
+ {C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Debug|x64.ActiveCfg = Debug|x64
+ {C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Debug|x64.Build.0 = Debug|x64
{C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Release|Win32.ActiveCfg = Release|Win32
{C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Release|Win32.Build.0 = Release|Win32
- {C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Release|x64.ActiveCfg = Release|Win32
- {C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Release|x86.ActiveCfg = Release|Win32
- {E2D8C779-E417-40BA-BEE1-EE034268482F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
- {E2D8C779-E417-40BA-BEE1-EE034268482F}.Debug|Any CPU.Build.0 = Debug|Any CPU
- {E2D8C779-E417-40BA-BEE1-EE034268482F}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU
- {E2D8C779-E417-40BA-BEE1-EE034268482F}.Debug|Mixed Platforms.Build.0 = Debug|Any CPU
+ {C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Release|x64.ActiveCfg = Release|x64
+ {C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}.Release|x64.Build.0 = Release|x64
{E2D8C779-E417-40BA-BEE1-EE034268482F}.Debug|Win32.ActiveCfg = Debug|Any CPU
+ {E2D8C779-E417-40BA-BEE1-EE034268482F}.Debug|Win32.Build.0 = Debug|Any CPU
{E2D8C779-E417-40BA-BEE1-EE034268482F}.Debug|x64.ActiveCfg = Debug|Any CPU
{E2D8C779-E417-40BA-BEE1-EE034268482F}.Debug|x64.Build.0 = Debug|Any CPU
- {E2D8C779-E417-40BA-BEE1-EE034268482F}.Debug|x86.ActiveCfg = Debug|Any CPU
- {E2D8C779-E417-40BA-BEE1-EE034268482F}.Debug|x86.Build.0 = Debug|Any CPU
- {E2D8C779-E417-40BA-BEE1-EE034268482F}.Release|Any CPU.ActiveCfg = Release|Any CPU
- {E2D8C779-E417-40BA-BEE1-EE034268482F}.Release|Any CPU.Build.0 = Release|Any CPU
- {E2D8C779-E417-40BA-BEE1-EE034268482F}.Release|Mixed Platforms.ActiveCfg = Release|Any CPU
- {E2D8C779-E417-40BA-BEE1-EE034268482F}.Release|Mixed Platforms.Build.0 = Release|Any CPU
{E2D8C779-E417-40BA-BEE1-EE034268482F}.Release|Win32.ActiveCfg = Release|Any CPU
+ {E2D8C779-E417-40BA-BEE1-EE034268482F}.Release|Win32.Build.0 = Release|Any CPU
{E2D8C779-E417-40BA-BEE1-EE034268482F}.Release|x64.ActiveCfg = Release|Any CPU
{E2D8C779-E417-40BA-BEE1-EE034268482F}.Release|x64.Build.0 = Release|Any CPU
- {E2D8C779-E417-40BA-BEE1-EE034268482F}.Release|x86.ActiveCfg = Release|Any CPU
- {E2D8C779-E417-40BA-BEE1-EE034268482F}.Release|x86.Build.0 = Release|Any CPU
- {8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
- {8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Debug|Any CPU.Build.0 = Debug|Any CPU
- {8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU
- {8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Debug|Mixed Platforms.Build.0 = Debug|Any CPU
{8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Debug|Win32.ActiveCfg = Debug|Any CPU
+ {8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Debug|Win32.Build.0 = Debug|Any CPU
{8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Debug|x64.ActiveCfg = Debug|Any CPU
- {8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Debug|x86.ActiveCfg = Debug|Any CPU
- {8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Release|Any CPU.ActiveCfg = Release|Any CPU
- {8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Release|Any CPU.Build.0 = Release|Any CPU
- {8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Release|Mixed Platforms.ActiveCfg = Release|Any CPU
- {8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Release|Mixed Platforms.Build.0 = Release|Any CPU
+ {8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Debug|x64.Build.0 = Debug|Any CPU
{8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Release|Win32.ActiveCfg = Release|Any CPU
+ {8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Release|Win32.Build.0 = Release|Any CPU
{8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Release|x64.ActiveCfg = Release|Any CPU
- {8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Release|x86.ActiveCfg = Release|Any CPU
+ {8AABAB30-7D1E-4539-B7D1-05450262BAD2}.Release|x64.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
diff --git a/qpid/wcf/src/Apache/Qpid/AmqpTypes/AmqpTypes.csproj b/qpid/wcf/src/Apache/Qpid/AmqpTypes/AmqpTypes.csproj index cca131b98a..2a544cf6d0 100644 --- a/qpid/wcf/src/Apache/Qpid/AmqpTypes/AmqpTypes.csproj +++ b/qpid/wcf/src/Apache/Qpid/AmqpTypes/AmqpTypes.csproj @@ -153,6 +153,6 @@ cd "$(ProjectDir)obj\$(ConfigurationName)" del $(AssemblyName).dll
del $(AssemblyName).pdb
cd "$(ProjectDir)"
-CreateNetModule.bat</PostBuildEvent>
+CreateNetModule.bat $(ConfigurationName)</PostBuildEvent>
</PropertyGroup>
</Project>
\ No newline at end of file diff --git a/qpid/wcf/src/Apache/Qpid/AmqpTypes/CreateNetModule.bat b/qpid/wcf/src/Apache/Qpid/AmqpTypes/CreateNetModule.bat index ddbe1407a7..d67e2119f9 100755 --- a/qpid/wcf/src/Apache/Qpid/AmqpTypes/CreateNetModule.bat +++ b/qpid/wcf/src/Apache/Qpid/AmqpTypes/CreateNetModule.bat @@ -16,4 +16,18 @@ REM KIND, either express or implied. See the License for the REM specific language governing permissions and limitations REM under the License. -%systemroot%\Microsoft.NET\Framework\v3.5\Csc.exe /noconfig /nowarn:1701,1702 /errorreport:prompt /warn:4 /define:DEBUG;TRACE /reference:"%programfiles%\Reference Assemblies\Microsoft\Framework\v3.5\System.Core.dll" /reference:"%programfiles%\Reference Assemblies\Microsoft\Framework\v3.5\System.Data.DataSetExtensions.dll" /reference:%systemroot%\Microsoft.NET\Framework\v2.0.50727\System.Data.dll /reference:%systemroot%\Microsoft.NET\Framework\v2.0.50727\System.dll /reference:%systemroot%\Microsoft.NET\Framework\v2.0.50727\System.Xml.dll /reference:"%programfiles%\Reference Assemblies\Microsoft\Framework\v3.5\System.Xml.Linq.dll" /debug+ /debug:full /filealign:512 /optimize- /out:obj\Debug\Apache.Qpid.AmqpTypes.netmodule /target:module *.cs
\ No newline at end of file +if %1 == Release goto release + +echo generating Debug netmodule + +%systemroot%\Microsoft.NET\Framework\v3.5\Csc.exe /noconfig /nowarn:1701,1702 /errorreport:prompt /warn:4 /define:DEBUG;TRACE /reference:"%programfiles%\Reference Assemblies\Microsoft\Framework\v3.5\System.Core.dll" /reference:"%programfiles%\Reference Assemblies\Microsoft\Framework\v3.5\System.Data.DataSetExtensions.dll" /reference:%systemroot%\Microsoft.NET\Framework\v2.0.50727\System.Data.dll /reference:%systemroot%\Microsoft.NET\Framework\v2.0.50727\System.dll /reference:%systemroot%\Microsoft.NET\Framework\v2.0.50727\System.Xml.dll /reference:"%programfiles%\Reference Assemblies\Microsoft\Framework\v3.5\System.Xml.Linq.dll" /debug+ /debug:full /filealign:512 /optimize- /out:obj\Debug\Apache.Qpid.AmqpTypes.netmodule /target:module *.cs + +goto end + +:release + +echo generating Release netmodule + +%systemroot%\Microsoft.NET\Framework\v3.5\Csc.exe /noconfig /nowarn:1701,1702 /errorreport:prompt /warn:4 /define:TRACE /reference:"%programfiles%\Reference Assemblies\Microsoft\Framework\v3.5\System.Core.dll" /reference:"%programfiles%\Reference Assemblies\Microsoft\Framework\v3.5\System.Data.DataSetExtensions.dll" /reference:C:\Windows\Microsoft.NET\Framework\v2.0.50727\System.Data.dll /reference:C:\Windows\Microsoft.NET\Framework\v2.0.50727\System.dll /reference:C:\Windows\Microsoft.NET\Framework\v2.0.50727\System.Xml.dll /reference:"%programfiles%\Reference Assemblies\Microsoft\Framework\v3.5\System.Xml.Linq.dll" /debug:pdbonly /filealign:512 /keyfile:..\..\..\wcfnet.snk /optimize+ /out:obj\Release\Apache.Qpid.AmqpTypes.netmodule /target:module *.cs + +:end
\ No newline at end of file diff --git a/qpid/wcf/src/Apache/Qpid/Channel/AmqpChannelFactory.cs b/qpid/wcf/src/Apache/Qpid/Channel/AmqpChannelFactory.cs index 542f1a00a8..5012c76d7e 100644 --- a/qpid/wcf/src/Apache/Qpid/Channel/AmqpChannelFactory.cs +++ b/qpid/wcf/src/Apache/Qpid/Channel/AmqpChannelFactory.cs @@ -32,7 +32,8 @@ namespace Apache.Qpid.Channel AmqpChannelProperties channelProperties; long maxBufferPoolSize; bool shared; - int prefetchLimit; + int prefetchLimit; + List<AmqpTransportChannel> openChannels; internal AmqpChannelFactory(AmqpTransportBindingElement bindingElement, BindingContext context) : base(context.Binding) @@ -45,7 +46,7 @@ namespace Apache.Qpid.Channel Collection<MessageEncodingBindingElement> messageEncoderBindingElements = context.BindingParameters.FindAll<MessageEncodingBindingElement>(); - if(messageEncoderBindingElements.Count > 1) + if (messageEncoderBindingElements.Count > 1) { throw new InvalidOperationException("More than one MessageEncodingBindingElement was found in the BindingParameters of the BindingContext"); } @@ -57,6 +58,8 @@ namespace Apache.Qpid.Channel { this.messageEncoderFactory = new TextMessageEncodingBindingElement().CreateMessageEncoderFactory(); } + + openChannels = new List<AmqpTransportChannel>(); } @@ -93,8 +96,42 @@ namespace Apache.Qpid.Channel protected override TChannel OnCreateChannel(EndpointAddress remoteAddress, Uri via) { - return (TChannel)(object) new AmqpTransportChannel(this, this.channelProperties, remoteAddress, this.messageEncoderFactory.Encoder, this.maxBufferPoolSize, this.shared, this.prefetchLimit); + AmqpTransportChannel channel = new AmqpTransportChannel(this, this.channelProperties, remoteAddress, this.messageEncoderFactory.Encoder, this.maxBufferPoolSize, this.shared, this.prefetchLimit); + lock (openChannels) + { + channel.Closed += new EventHandler(channel_Closed); + openChannels.Add(channel); + } + return (TChannel)(object) channel; } + void channel_Closed(object sender, EventArgs e) + { + if (this.State != CommunicationState.Opened) + { + return; + } + + lock (openChannels) + { + AmqpTransportChannel channel = (AmqpTransportChannel)sender; + if (openChannels.Contains(channel)) + { + openChannels.Remove(channel); + } + } + } + + protected override void OnClose(TimeSpan timeout) + { + base.OnClose(timeout); + lock (openChannels) + { + foreach (AmqpTransportChannel channel in openChannels) + { + channel.Close(timeout); + } + } + } } } diff --git a/qpid/wcf/src/Apache/Qpid/Channel/AmqpChannelListener.cs b/qpid/wcf/src/Apache/Qpid/Channel/AmqpChannelListener.cs index 8894b68584..3d7801e7c6 100644 --- a/qpid/wcf/src/Apache/Qpid/Channel/AmqpChannelListener.cs +++ b/qpid/wcf/src/Apache/Qpid/Channel/AmqpChannelListener.cs @@ -130,15 +130,21 @@ namespace Apache.Qpid.Channel protected override IInputChannel OnAcceptChannel(TimeSpan timeout) { + if (this.IsDisposed) + { + return null; + } + if (amqpTransportChannel == null) { + // TODO: add timeout processing amqpTransportChannel = new AmqpTransportChannel(this, this.channelProperties, new EndpointAddress(uri), messageEncoderFactory.Encoder, maxBufferPoolSize, this.shared, this.prefetchLimit); - return (IInputChannel)(object) amqpTransportChannel; + return (IInputChannel)(object)amqpTransportChannel; } - // TODO: remove "max one channel" restriction, add timeout processing + // Singleton channel. Subsequent Accepts wait until the listener is closed acceptWaitEvent.WaitOne(); return null; } @@ -155,7 +161,11 @@ namespace Apache.Qpid.Channel protected override void OnClose(TimeSpan timeout) { - // TODO: (+ OnAbort) + if (amqpTransportChannel != null) + { + amqpTransportChannel.Close(); + } + acceptWaitEvent.Set(); } protected override IAsyncResult OnBeginClose(TimeSpan timeout, AsyncCallback callback, object state) @@ -170,7 +180,9 @@ namespace Apache.Qpid.Channel protected override void OnAbort() { - // TODO: + if (amqpTransportChannel != null) + amqpTransportChannel.Abort(); + acceptWaitEvent.Set(); } } } diff --git a/qpid/wcf/src/Apache/Qpid/Channel/AmqpTransportBindingElement.cs b/qpid/wcf/src/Apache/Qpid/Channel/AmqpTransportBindingElement.cs index 08c565af18..7993252309 100644 --- a/qpid/wcf/src/Apache/Qpid/Channel/AmqpTransportBindingElement.cs +++ b/qpid/wcf/src/Apache/Qpid/Channel/AmqpTransportBindingElement.cs @@ -25,7 +25,7 @@ namespace Apache.Qpid.Channel using System.ServiceModel.Description; using Apache.Qpid.AmqpTypes; - public class AmqpTransportBindingElement : TransportBindingElement + public class AmqpTransportBindingElement : TransportBindingElement, ITransactedBindingElement { AmqpChannelProperties channelProperties; bool shared; @@ -112,6 +112,11 @@ namespace Apache.Qpid.Channel set { this.shared = value; } } + public bool TransactedReceiveEnabled + { + get { return true; } + } + public TransferMode TransferMode { get { return this.channelProperties.TransferMode; } diff --git a/qpid/wcf/src/Apache/Qpid/Channel/AmqpTransportChannel.cs b/qpid/wcf/src/Apache/Qpid/Channel/AmqpTransportChannel.cs index 5924142046..a6f6ee6800 100644 --- a/qpid/wcf/src/Apache/Qpid/Channel/AmqpTransportChannel.cs +++ b/qpid/wcf/src/Apache/Qpid/Channel/AmqpTransportChannel.cs @@ -278,7 +278,6 @@ namespace Apache.Qpid.Channel public bool TryReceive(TimeSpan timeout, out Message message) { - this.ThrowIfDisposedOrNotOpen(); AmqpMessage amqpMessage; message = null; @@ -379,6 +378,7 @@ namespace Apache.Qpid.Channel protected override void OnAbort() { //// TODO: check for network-less qpid teardown or launch special thread + this.CloseEndPoint(); this.Cleanup(); } diff --git a/qpid/wcf/src/Apache/Qpid/Channel/Channel.csproj b/qpid/wcf/src/Apache/Qpid/Channel/Channel.csproj index 7484bc38ac..ac90fb7d64 100644 --- a/qpid/wcf/src/Apache/Qpid/Channel/Channel.csproj +++ b/qpid/wcf/src/Apache/Qpid/Channel/Channel.csproj @@ -90,6 +90,7 @@ under the License. <Reference Include="System.ServiceModel">
<RequiredTargetFramework>3.0</RequiredTargetFramework>
</Reference>
+ <Reference Include="System.Transactions" />
<Reference Include="System.XML" />
</ItemGroup>
<ItemGroup>
diff --git a/qpid/wcf/src/Apache/Qpid/DtcPlugin/DtcPlugin.cpp b/qpid/wcf/src/Apache/Qpid/DtcPlugin/DtcPlugin.cpp new file mode 100644 index 0000000000..f9d8bd8521 --- /dev/null +++ b/qpid/wcf/src/Apache/Qpid/DtcPlugin/DtcPlugin.cpp @@ -0,0 +1,664 @@ +/*
+* 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.
+*/
+
+
+//
+// This module provides the backend recovery driver for Windows resource managers based on
+// the IDtcToXaHelperSinglePipe interface. The dll is loaded (LoadLibrary) directly into DTC
+// itself and runs at a different protection level from the resource manager instance, which
+// runs inside the application.
+//
+// The DTC dynamically loads this file, calls GetXaSwitch() to access the XA interface
+// implementation and unloads the dll when done.
+//
+// This DTC plugin is only called for registration and recovery. Each time the application
+// registers the Qpid resource manager with DTC, the plugin is loaded and a successful
+// connection via xa_open is confirmed before completing registration and saving the DSN
+// connection string in the DTC log for possible recovery. On recovery, the DSN is re-used to
+// restablish a new connection with the broker and perform recovery.
+//
+// Because this plugin is not involved in coordinating any active transactions it only needs to
+// partially implement the XA interface.
+//
+// For the same reason, the locking strategy is simple. A single global lock is used.
+// Whenever networking activity is about to take place, the lock is relinquished and retaken
+// soon thereafter.
+
+
+#include <windows.h>
+#include <transact.h>
+#include <xolehlp.h>
+#include <txdtc.h>
+#include <xa.h>
+
+#include "qpid/client/AsyncSession.h"
+#include "qpid/client/Connection.h"
+
+
+#include <map>
+#include <iostream>
+#include <fstream>
+
+namespace Apache {
+namespace Qpid {
+namespace DtcPlugin {
+
+using namespace qpid::client;
+using namespace qpid::framing;
+using namespace qpid::framing::dtx;
+
+class ResourceManager
+{
+private:
+ Connection qpidConnection;
+ Session qpidSession;
+ bool active;
+ std::string host;
+ int port;
+ int rmid;
+ std::vector<qpid::framing::Xid> inDoubtXids;
+ // current scan position, or -1 if no scan
+ int cursor;
+public:
+ ResourceManager(int id, std::string h, int p) : rmid(id), host(h), port(p), active(false), cursor(-1) {}
+ ~ResourceManager() {}
+ INT open();
+ INT close();
+ INT commit(XID *xid);
+ INT rollback(XID *xid);
+ INT recover(XID *xids, long count, long flags);
+};
+
+
+CRITICAL_SECTION rmLock;
+
+std::map<int, ResourceManager*> rmMap;
+HMODULE thisDll = NULL;
+bool memLocked = false;
+
+#define QPIDHMCHARS 512
+
+void pinDll() {
+ if (!memLocked) {
+ char thisDllName[QPIDHMCHARS];
+ HMODULE ignore;
+
+ DWORD nc = GetModuleFileName(thisDll, thisDllName, QPIDHMCHARS);
+ if ((nc > 0) && (nc < QPIDHMCHARS)) {
+ memLocked = GetModuleHandleEx(GET_MODULE_HANDLE_EX_FLAG_PIN, thisDllName, &ignore);
+ }
+ }
+}
+
+
+void XaToQpid(XID &winXid, Xid &qpidXid) {
+ // convert from XA defined structure XID to the Qpid framing structure
+ qpidXid.setFormat((uint32_t) winXid.formatID);
+ int bqualPos = 0;
+ if (winXid.gtrid_length > 0) {
+ qpidXid.setGlobalId(std::string(winXid.data, winXid.gtrid_length));
+ bqualPos = winXid.gtrid_length;
+ }
+ if (winXid.bqual_length > 0) {
+ qpidXid.setBranchId(std::string(winXid.data + bqualPos, winXid.bqual_length));
+ }
+}
+
+
+// this function assumes that the qpidXid has already been validated for the memory copy
+
+void QpidToXa(Xid &qpidXid, XID &winXid) {
+ // convert from the Qpid framing structure to the XA defined structure XID
+ winXid.formatID = qpidXid.getFormat();
+
+ const std::string& global_s = qpidXid.getGlobalId();
+ size_t gl = global_s.size();
+ winXid.gtrid_length = (long) gl;
+ if (gl > 0)
+ global_s.copy(winXid.data, gl);
+
+ const std::string branch_s = qpidXid.getBranchId();
+ size_t bl = branch_s.size();
+ winXid.bqual_length = (long) bl;
+ if (bl > 0)
+ branch_s.copy(winXid.data + gl, bl);
+}
+
+
+/* parse string from AmqpConnection.h
+
+ this info will eventually include authentication tokens
+
+ dataSourceName = String::Format("{0}.{1}..AMQP.{2}.{3}", port, host,
+ System::Diagnostics::Process::GetCurrentProcess()->Id,
+ AppDomain::CurrentDomain->Id);
+*/
+
+bool parseDsn (const char *dsn, std::string& host, int& port) {
+ if (dsn == NULL)
+ return false;
+
+ size_t len = strnlen(dsn, 1025);
+ if (len > 1024)
+ return false;
+
+ int firstDot = 0;
+ for (int i = 0; i < len; i++)
+ if (dsn[i] == '.') {
+ firstDot = i;
+ break;
+ }
+ if (!firstDot)
+ return false;
+
+ // look for 2 dots side by side to indicate end of the host
+ int doubleDot = 0;
+ for (int i = firstDot + 1; i < (len - 1); i++)
+ if ((dsn[i] == '.') && (dsn[i+1] == '.')) {
+ doubleDot = i;
+ break;
+ }
+ if (!doubleDot)
+ return false;
+
+ port = 0;
+ for (int i = 0; i < firstDot; i++) {
+ char c = dsn[i];
+ if ((c < '0') || (c > '9'))
+ return false;
+ port = (10 * port) + (c - '0');
+ }
+
+ host.assign(dsn + firstDot + 1, (doubleDot - firstDot) - 1);
+ return true;
+}
+
+
+INT ResourceManager::open() {
+ INT rv = XAER_RMERR; // placeholder until we successfully connect to resource
+ active = true;
+ LeaveCriticalSection(&rmLock);
+
+ try {
+ qpidConnection.open(host, port);
+ qpidSession = qpidConnection.newSession();
+ rv = XA_OK;
+/*
+TODO: logging
+ } catch (const qpid::Exception& error) {
+ // log it
+ } catch (const std::exception& e2) {
+ // log it
+*/
+ } catch (...) {
+ // TODO: log it
+ }
+
+ EnterCriticalSection(&rmLock);
+ active = false;
+ return rv;
+}
+
+
+INT ResourceManager::close() {
+ // should never be called when already sending other commands to broker
+ if (active)
+ return XAER_PROTO;
+
+ INT rv = XAER_RMERR; // placeholder until we successfully close resource
+ active = true;
+ LeaveCriticalSection(&rmLock);
+ try {
+ if (qpidSession.isValid()) {
+ qpidSession.close();
+ }
+ if (qpidConnection.isOpen()) {
+ qpidConnection.close();
+ }
+ } catch (...) {
+ // TODO: log it
+ }
+
+ EnterCriticalSection(&rmLock);
+ active = false;
+
+ if (!qpidConnection.isOpen()) {
+ rv = XA_OK;
+ }
+ return rv;
+}
+
+
+INT ResourceManager::commit(XID *xid) {
+ if (active)
+ return XAER_PROTO;
+
+ INT rv = XAER_RMFAIL;
+ active = true;
+ LeaveCriticalSection(&rmLock);
+
+ try {
+ qpid::framing::Xid qpidXid;
+ XaToQpid(*xid, qpidXid);
+
+ XaResult xaResult = qpidSession.dtxCommit(qpidXid, false, true);
+ if (xaResult.hasStatus()) {
+ uint16_t status = xaResult.getStatus();
+ switch ((XaStatus) status) {
+ case XA_STATUS_XA_OK:
+ case XA_STATUS_XA_RDONLY:
+ case XA_STATUS_XA_HEURCOM:
+ rv = XA_OK;
+ break;
+
+ default:
+ // commit failed and a retry won't fix
+ rv = XAER_RMERR;
+ break;
+ }
+
+ }
+ } catch (...) {
+ // TODO: log it
+ }
+
+ EnterCriticalSection(&rmLock);
+ active = false;
+ return rv;
+}
+
+
+INT ResourceManager::rollback(XID *xid) {
+ if (active)
+ return XAER_PROTO;
+
+ INT rv = XAER_RMFAIL;
+ active = true;
+ LeaveCriticalSection(&rmLock);
+
+ try {
+ qpid::framing::Xid qpidXid;
+ XaToQpid(*xid, qpidXid);
+
+ XaResult xaResult = qpidSession.dtxRollback(qpidXid, true);
+ if (xaResult.hasStatus()) {
+ uint16_t status = xaResult.getStatus();
+ switch ((XaStatus) status) {
+ case XA_STATUS_XA_OK:
+ case XA_STATUS_XA_HEURRB:
+ rv = XA_OK;
+ break;
+
+ default:
+ // RM internal error
+ rv = XA_RBPROTO;
+ break;
+ }
+ }
+ } catch (...) {
+ // TODO: log it
+ }
+
+ EnterCriticalSection(&rmLock);
+ active = false;
+ return rv;
+}
+
+
+INT ResourceManager::recover(XID *xids, long count, long flags) {
+ if (active)
+ return XAER_PROTO;
+
+ if ((xids == NULL) && (count != 0))
+ return XAER_INVAL;
+
+ if (count < 0)
+ return XAER_INVAL;
+
+ if (!(flags & TMSTARTRSCAN) && (cursor == -1))
+ // no existing scan and no scan requested
+ return XAER_INVAL;
+
+ INT status = XA_OK;
+
+ if (flags & TMSTARTRSCAN) {
+ // start a fresh scan
+ cursor = -1;
+ inDoubtXids.clear();
+ active = true;
+ LeaveCriticalSection(&rmLock);
+
+ try {
+ // status if we can't talk to the broker
+ status = XAER_RMFAIL;
+ std::vector<std::string> wireFormatXids;
+
+ DtxRecoverResult dtxrr = qpidSession.dtxRecover(true);
+
+ // status if we can't process the xids
+ status = XAER_RMERR;
+ dtxrr.getInDoubt().collect(wireFormatXids);
+ size_t nXids = wireFormatXids.size();
+
+ if (nXids > 0) {
+ StructHelper decoder;
+ Xid qpidXid;
+ for (int i = 0; i < nXids; i++) {
+ decoder.decode (qpidXid, wireFormatXids[i]);
+ inDoubtXids.push_back(qpidXid);
+ }
+
+ // if we got here the decoder validated the Xids
+ status = XA_OK;
+
+ // make sure none are too big, just in case
+
+ for (int i = 0; i < nXids; i++) {
+ Xid& xid = inDoubtXids[i];
+ size_t l1 = xid.hasGlobalId() ? xid.getGlobalId().size() : 0;
+ size_t l2 = xid.hasBranchId() ? xid.getBranchId().size() : 0;
+ if ((l1 > MAXGTRIDSIZE) || (l2 > MAXBQUALSIZE) ||
+ ((l1 + l2) > XIDDATASIZE)) {
+ status = XAER_RMERR;
+ break;
+ }
+ }
+ }
+ else {
+ // nXids == 0, the previously cleared inDoubtXids is correctly populated
+ status = XA_OK;
+ }
+
+ if (status == XA_OK)
+ cursor = 0;
+ } catch (...) {
+ // TODO: log it
+ }
+
+ EnterCriticalSection(&rmLock);
+ active = false;
+ }
+ else {
+ // TMSTARTRSCAN not set, is there an existing scan to work from?
+ if (cursor == -1)
+ return XAER_INVAL;
+ }
+
+ if (status != XA_OK)
+ return status;
+
+ INT actualCount = count;
+ if (count > 0) {
+ int nAvailable = (int) inDoubtXids.size() - cursor;
+ if (nAvailable < count)
+ actualCount = nAvailable;
+
+ for (int i = 0; i < actualCount; i++) {
+ Xid& qpidXid = inDoubtXids[i + cursor];
+ QpidToXa(qpidXid, xids[i]);
+ }
+ }
+
+ if (flags & TMENDRSCAN) {
+ cursor = -1;
+ inDoubtXids.clear();
+ }
+
+ return actualCount;
+}
+
+
+// Call with lock held
+
+ResourceManager* findRm(int rmid) {
+ if (rmMap.find(rmid) == rmMap.end()) {
+ return NULL;
+ }
+ return rmMap[rmid];
+}
+
+
+INT __cdecl xa_open (char *xa_info, int rmid, long flags) {
+ if (flags & TMASYNC)
+ return XAER_ASYNC;
+
+ INT rv = XAER_RMERR;
+ EnterCriticalSection(&rmLock);
+
+ ResourceManager* rmp = findRm(rmid);
+ if (rmp != NULL) {
+ // error: already in use
+ rv = XAER_PROTO;
+ }
+ else {
+ std::string brokerHost;
+ int brokerPort;
+ if (parseDsn(xa_info, brokerHost, brokerPort)) {
+
+ try {
+ rmp = new ResourceManager(rmid, brokerHost, brokerPort);
+
+ rv = rmp->open();
+ if (rv != XA_OK) {
+ delete (rmp);
+ }
+ else {
+ rmMap[rmid] = rmp;
+ }
+ } catch (...) {}
+ }
+ else {
+ rv = XAER_INVAL;
+ }
+ }
+
+ LeaveCriticalSection(&rmLock);
+ return rv;
+}
+
+
+INT __cdecl xa_close (char *xa_info, int rmid, long flags) {
+ if (flags & TMASYNC)
+ return XAER_ASYNC;
+
+ INT rv = XAER_RMERR;
+
+ EnterCriticalSection(&rmLock);
+ ResourceManager* rmp = findRm(rmid);
+
+ if (rmp == NULL) {
+ // can close multiple times
+ rv = XA_OK;
+ }
+ else {
+ rv = rmp->close();
+ rmMap.erase(rmid);
+ try {
+ delete (rmp);
+ } catch (...) {
+ // TODO: log it
+ }
+ }
+
+ LeaveCriticalSection(&rmLock);
+ return rv;
+}
+
+
+INT __cdecl xa_commit (XID *xid, int rmid, long flags) {
+ if (flags & TMASYNC)
+ return XAER_ASYNC;
+
+ INT rv = XAER_RMFAIL;
+
+ EnterCriticalSection(&rmLock);
+ ResourceManager* rmp = findRm(rmid);
+
+ if (rmp == NULL) {
+ rv = XAER_INVAL;
+ }
+ else {
+ rv = rmp->commit(xid);
+ }
+
+ LeaveCriticalSection(&rmLock);
+ return rv;
+}
+
+
+INT __cdecl xa_rollback (XID *xid, int rmid, long flags) {
+ if (flags & TMASYNC)
+ return XAER_ASYNC;
+
+ INT rv = XAER_RMFAIL;
+
+ EnterCriticalSection(&rmLock);
+ ResourceManager* rmp = findRm(rmid);
+
+ if (rmp == NULL) {
+ rv = XAER_INVAL;
+ }
+ else {
+ rv = rmp->rollback(xid);
+ }
+
+ LeaveCriticalSection(&rmLock);
+ return rv;
+}
+
+
+INT __cdecl xa_recover (XID *xids, long count, int rmid, long flags) {
+ INT rv = XAER_RMFAIL;
+
+ EnterCriticalSection(&rmLock);
+ ResourceManager* rmp = findRm(rmid);
+
+ if (rmp == NULL) {
+ rv = XAER_PROTO;
+ }
+ else {
+ rv = rmp->recover(xids, count, flags);
+ }
+
+ LeaveCriticalSection(&rmLock);
+ return rv;
+}
+
+
+INT __cdecl xa_start (XID *xid, int rmid, long flags) {
+ // not used in recovery
+ return XAER_PROTO;
+}
+
+
+INT __cdecl xa_end (XID *xid, int rmid, long flags) {
+ // not used in recovery
+ return XAER_PROTO;
+}
+
+
+INT __cdecl xa_prepare (XID *xid, int rmid, long flags) {
+ // not used in recovery
+ return XAER_PROTO;
+}
+
+
+INT __cdecl xa_forget (XID *xid, int rmid, long flags) {
+ // not used in recovery
+ return XAER_PROTO;
+}
+
+
+INT __cdecl xa_complete (int *handle, int *retval, int rmid, long flags) {
+ // not used in recovery
+ return XAER_PROTO;
+}
+
+
+
+xa_switch_t xaSwitch;
+
+HRESULT __cdecl GetQpidXaSwitch (DWORD XaSwitchFlags, xa_switch_t ** ppXaSwitch)
+{
+ // needed for now due to implicit use of FreeLibrary in WSACleanup() in qpid/cpp/src/qpid/sys/windows/Socket.cpp
+ pinDll();
+
+ if (xaSwitch.xa_open_entry != xa_open) {
+
+ xaSwitch.xa_open_entry = xa_open;
+ xaSwitch.xa_close_entry = xa_close;
+ xaSwitch.xa_start_entry = xa_start;
+ xaSwitch.xa_end_entry = xa_end;
+ xaSwitch.xa_prepare_entry = xa_prepare;
+ xaSwitch.xa_commit_entry = xa_commit;
+ xaSwitch.xa_rollback_entry = xa_rollback;
+ xaSwitch.xa_recover_entry = xa_recover;
+ xaSwitch.xa_forget_entry = xa_forget;
+ xaSwitch.xa_complete_entry = xa_complete;
+
+ strcpy_s(xaSwitch.name, RMNAMESZ, "qpidxarm");
+ xaSwitch.flags = TMNOMIGRATE;
+ xaSwitch.version = 0;
+ }
+ *ppXaSwitch = &xaSwitch;
+ return S_OK;
+}
+
+
+
+
+}}} // namespace Apache::Qpid::DtcPlugin
+
+
+// GetXaSwitch
+
+extern "C" {
+
+ __declspec(dllexport) HRESULT __cdecl GetXaSwitch (DWORD XaSwitchFlags, xa_switch_t ** ppXaSwitch)
+ {
+ return Apache::Qpid::DtcPlugin::GetQpidXaSwitch (XaSwitchFlags, ppXaSwitch);
+ }
+}
+
+
+// dllmain
+
+BOOL APIENTRY DllMain( HMODULE hModule,
+ DWORD ul_reason_for_call,
+ LPVOID lpReserved)
+{
+
+ switch (ul_reason_for_call)
+ {
+ case DLL_PROCESS_ATTACH:
+ InitializeCriticalSection(&Apache::Qpid::DtcPlugin::rmLock);
+ Apache::Qpid::DtcPlugin::thisDll = hModule;
+ break;
+
+ case DLL_PROCESS_DETACH:
+ DeleteCriticalSection(&Apache::Qpid::DtcPlugin::rmLock);
+ break;
+
+ case DLL_THREAD_ATTACH:
+ case DLL_THREAD_DETACH:
+ break;
+ }
+ return TRUE;
+}
+
diff --git a/qpid/wcf/src/Apache/Qpid/Interop/AmqpConnection.cpp b/qpid/wcf/src/Apache/Qpid/Interop/AmqpConnection.cpp index 02d6c7ab18..c3afdf2280 100644 --- a/qpid/wcf/src/Apache/Qpid/Interop/AmqpConnection.cpp +++ b/qpid/wcf/src/Apache/Qpid/Interop/AmqpConnection.cpp @@ -19,6 +19,7 @@ #include <windows.h> #include <msclr\lock.h> +#include <oletx2xa.h> #include "qpid/client/AsyncSession.h" #include "qpid/client/SubscriptionManager.h" @@ -31,6 +32,8 @@ #include "AmqpSession.h" #include "QpidMarshal.h" #include "QpidException.h" +#include "DtxResourceManager.h" +#include "XaTransaction.h" namespace Apache { namespace Qpid { @@ -66,6 +69,9 @@ AmqpConnection::AmqpConnection(String^ server, int port) : success = true; const ConnectionSettings& settings = connectionp->getNegotiatedSettings(); this->maxFrameSize = settings.maxFrameSize; + this->host = server; + this->port = port; + this->isOpen = true; } catch (const qpid::Exception& error) { String^ errmsg = gcnew String(error.what()); openException = gcnew QpidException(errmsg); @@ -80,6 +86,12 @@ AmqpConnection::AmqpConnection(String^ server, int port) : } } +AmqpConnection^ AmqpConnection::Clone() { + if (disposed) + throw gcnew ObjectDisposedException("AmqpConnection.Clone"); + return gcnew AmqpConnection (this->host, this->port); +} + void AmqpConnection::Cleanup() { { @@ -91,6 +103,7 @@ void AmqpConnection::Cleanup() try { // let the child sessions clean up + for each(AmqpSession^ s in sessions) { s->ConnectionClosed(); } @@ -98,6 +111,7 @@ void AmqpConnection::Cleanup() finally { if (connectionp != NULL) { + isOpen = false; connectionp->close(); delete connectionp; connectionp = NULL; diff --git a/qpid/wcf/src/Apache/Qpid/Interop/AmqpConnection.h b/qpid/wcf/src/Apache/Qpid/Interop/AmqpConnection.h index 2641391e82..6533185fa1 100644 --- a/qpid/wcf/src/Apache/Qpid/Interop/AmqpConnection.h +++ b/qpid/wcf/src/Apache/Qpid/Interop/AmqpConnection.h @@ -28,6 +28,7 @@ using namespace std; using namespace qpid::client; ref class AmqpSession; +ref class DtxResourceManager; public delegate void ConnectionIdleEventHandler(Object^ sender, EventArgs^ eventArgs); @@ -35,21 +36,44 @@ public ref class AmqpConnection { private: Connection* connectionp; - void Cleanup(); + String^ host; + int port; bool disposed; Collections::Generic::List<AmqpSession^>^ sessions; bool isOpen; int busyCount; int maxFrameSize; + DtxResourceManager^ dtxResourceManager; + void Cleanup(); + // unique string used for distributed transactions + String^ dataSourceName; internal: void NotifyBusy(); void NotifyIdle(); + AmqpConnection^ Clone(); property int MaxFrameSize { int get () { return maxFrameSize; } } + property DtxResourceManager^ CachedResourceManager { + DtxResourceManager^ get () { return dtxResourceManager; } + void set (DtxResourceManager^ value) { dtxResourceManager = value; } + } + + property String^ DataSourceName { + // Note: any change to this format has to be reflected in the DTC plugin's xa_open() + String^ get() { + if (dataSourceName == nullptr) { + dataSourceName = String::Format("{0}.{1}..AMQP.{2}.{3}", port, host, + System::Diagnostics::Process::GetCurrentProcess()->Id, + AppDomain::CurrentDomain->Id); + } + return dataSourceName; + } + } + public: AmqpConnection(System::String^ server, int port); ~AmqpConnection(); diff --git a/qpid/wcf/src/Apache/Qpid/Interop/AmqpSession.cpp b/qpid/wcf/src/Apache/Qpid/Interop/AmqpSession.cpp index 425a592509..d2adb41205 100644 --- a/qpid/wcf/src/Apache/Qpid/Interop/AmqpSession.cpp +++ b/qpid/wcf/src/Apache/Qpid/Interop/AmqpSession.cpp @@ -19,6 +19,7 @@ #include <windows.h> #include <msclr\lock.h> +#include <oletx2xa.h> #include "qpid/client/AsyncSession.h" #include "qpid/client/SubscriptionManager.h" @@ -28,6 +29,7 @@ #include "qpid/client/Message.h" #include "qpid/framing/MessageTransferBody.h" #include "qpid/client/Future.h" +#include "qpid/framing/Xid.h" #include "AmqpConnection.h" #include "AmqpSession.h" @@ -37,6 +39,8 @@ #include "OutputLink.h" #include "QpidMarshal.h" #include "QpidException.h" +#include "XaTransaction.h" +#include "DtxResourceManager.h" namespace Apache { namespace Qpid { @@ -44,6 +48,7 @@ namespace Interop { using namespace System; using namespace System::Runtime::InteropServices; +using namespace System::Transactions; using namespace msclr; using namespace qpid::client; @@ -55,15 +60,20 @@ AmqpSession::AmqpSession(AmqpConnection^ conn, qpid::client::Connection* qpidCon sessionp(NULL), sessionImplp(NULL), subs_mgrp(NULL), - openCount(0) + helperRunning(false), + openCount(0), + syncCount(0), + closing(false), + dtxEnabled(false) { bool success = false; - try { sessionp = new qpid::client::AsyncSession; *sessionp = qpidConnectionp->newSession(); subs_mgrp = new SubscriptionManager (*sessionp); waiters = gcnew Collections::Generic::List<CompletionWaiter^>(); + sessionLock = waiters; // waiters convenient and not publicly visible + openCloseLock = gcnew Object(); success = true; } finally { if (!success) { @@ -77,29 +87,36 @@ AmqpSession::AmqpSession(AmqpConnection^ conn, qpid::client::Connection* qpidCon void AmqpSession::Cleanup() { + bool connected = connection->IsOpen; + if (subs_mgrp != NULL) { - subs_mgrp->stop(); + if (connected) + subs_mgrp->stop(); delete subs_mgrp; subs_mgrp = NULL; } - if (localQueuep != NULL) { - delete localQueuep; - localQueuep = NULL; - } - if (sessionp != NULL) { - sessionp->close(); + if (connected) { + sessionp->close(); + } delete sessionp; sessionp = NULL; sessionImplp = NULL; } +} - if (connectionp != NULL) { - connectionp->close(); - delete connectionp; - connectionp = NULL; - } + +static qpid::framing::Xid& getXid(XaTransaction^ xaTx) +{ + return *((qpid::framing::Xid *)xaTx->XidHandle.ToPointer()); +} + + +void AmqpSession::CheckOpen() +{ + if (closing) + throw gcnew ObjectDisposedException("AmqpSession"); } @@ -107,7 +124,41 @@ void AmqpSession::Cleanup() void AmqpSession::ConnectionClosed() { - lock l(waiters); + lock l(sessionLock); + + if (closing) + return; + + closing = true; + + if (connection->IsOpen) { + // send closing handshakes... + + if (dtxEnabled) { + // session may close before all its transactions complete, at least force the phase 0 flush + if (pendingTransactions->Count > 0) { + array<XaTransaction^>^ txArray = pendingTransactions->ToArray(); + l.release(); + for each (XaTransaction^ xaTx in txArray) { + //xaTx->SessionClosing(this); + xaTx->WaitForCompletion(); + } + l.acquire(); + } + } + + WaitLastSync (%l); + // Assert pendingTransactions->Count == 0 + + if (openXaTransaction != nullptr) { + // send final dtxend + sessionp->dtxEnd(getXid(openXaTransaction), false, true, false); + openXaTransaction = nullptr; + openSystemTransaction = nullptr; + // this operation will complete by the time Cleanup() returns + } + } + Cleanup(); } @@ -119,10 +170,14 @@ InputLink^ AmqpSession::CreateInputLink(System::String^ sourceQueue) InputLink^ AmqpSession::CreateInputLink(System::String^ sourceQueue, bool exclusive, bool temporary, System::String^ filterKey, System::String^ exchange) { + lock ocl(openCloseLock); + lock l(sessionLock); + CheckOpen(); + InputLink^ link = gcnew InputLink (this, sourceQueue, sessionp, subs_mgrp, exclusive, temporary, filterKey, exchange); { - lock l(waiters); if (openCount == 0) { + l.release(); connection->NotifyBusy(); } openCount++; @@ -132,9 +187,11 @@ InputLink^ AmqpSession::CreateInputLink(System::String^ sourceQueue, bool exclus OutputLink^ AmqpSession::CreateOutputLink(System::String^ targetQueue) { - OutputLink^ link = gcnew OutputLink (this, targetQueue); + lock ocl(openCloseLock); + lock l(sessionLock); + CheckOpen(); - lock l(waiters); + OutputLink^ link = gcnew OutputLink (this, targetQueue); if (sessionImplp == NULL) { // not needed unless sending messages @@ -144,6 +201,7 @@ OutputLink^ AmqpSession::CreateOutputLink(System::String^ targetQueue) } if (openCount == 0) { + l.release(); connection->NotifyBusy(); } openCount++; @@ -155,7 +213,7 @@ OutputLink^ AmqpSession::CreateOutputLink(System::String^ targetQueue) // called whenever a child InputLink or OutputLink is closed or finalized void AmqpSession::NotifyClosed() { - lock l(waiters); + lock ocl(openCloseLock); openCount--; if (openCount == 0) { connection->NotifyIdle(); @@ -165,9 +223,14 @@ void AmqpSession::NotifyClosed() CompletionWaiter^ AmqpSession::SendMessage (System::String^ queue, MessageBodyStream ^mbody, TimeSpan timeout, bool async, AsyncCallback^ callback, Object^ state) { - lock l(waiters); - if (sessionp == NULL) - throw gcnew ObjectDisposedException("Send"); + lock l(sessionLock); + + // delimit with session dtx commands depending on the transaction context + UpdateTransactionState(%l); + + CheckOpen(); + + bool syncPending = false; // create an AMQP message.transfer command to use with the partial frameset from the MessageBodyStream @@ -191,26 +254,34 @@ CompletionWaiter^ AmqpSession::SendMessage (System::String^ queue, MessageBodySt // waiter is responsible for releasing the Future native resource futurep = NULL; addWaiter(waiter); - } - - l.release(); - - if (waiter != nullptr) return waiter; + } // synchronous send with no timeout: no need to involve the asyncHelper thread + IncrementSyncs(); + syncPending = true; + l.release(); internalWaitForCompletion((IntPtr) futurep); } finally { + if (syncPending) { + if (!l.is_locked()) + l.acquire(); + DecrementSyncs(); + } if (futurep != NULL) delete (futurep); } return nullptr; } + void AmqpSession::Bind(System::String^ queue, System::String^ exchange, System::String^ filterKey) { + lock l(sessionLock); + CheckOpen(); + sessionp->exchangeBind(arg::queue=QpidMarshal::ToNative(queue), arg::exchange=QpidMarshal::ToNative(exchange), arg::bindingKey=QpidMarshal::ToNative(filterKey)); @@ -220,14 +291,8 @@ void AmqpSession::Bind(System::String^ queue, System::String^ exchange, System:: void AmqpSession::internalWaitForCompletion(IntPtr fp) { - lock l(waiters); - if (sessionp == NULL) - throw gcnew ObjectDisposedException("AmqpSession"); + Debug::Assert(syncCount > 0, "sync counter mismatch"); - // increment the smart pointer count to sessionImplp to guard agains async close - Session sessionCopy(*sessionp); - - l.release(); // Qpid native lib call to wait for the command completion ((Future *)fp.ToPointer())->wait(*sessionImplp); } @@ -235,6 +300,7 @@ void AmqpSession::internalWaitForCompletion(IntPtr fp) // call with lock held void AmqpSession::addWaiter(CompletionWaiter^ waiter) { + IncrementSyncs(); waiters->Add(waiter); if (!helperRunning) { helperRunning = true; @@ -247,13 +313,14 @@ void AmqpSession::removeWaiter(CompletionWaiter^ waiter) { // a waiter can be removed from anywhere in the list if timed out - lock l(waiters); + lock l(sessionLock); int idx = waiters->IndexOf(waiter); if (idx == -1) { // TODO: assert or log } else { waiters->RemoveAt(idx); + DecrementSyncs(); } } @@ -262,7 +329,7 @@ void AmqpSession::removeWaiter(CompletionWaiter^ waiter) void AmqpSession::asyncHelper(Object ^unused) { - lock l(waiters); + lock l(sessionLock); while (true) { if (waiters->Count == 0) { @@ -279,27 +346,267 @@ void AmqpSession::asyncHelper(Object ^unused) } } -bool AmqpSession::MessageStop(Completion &comp, std::string &name) +bool AmqpSession::MessageStop(std::string &name) { - lock l(waiters); + lock l(sessionLock); - if (sessionp == NULL) + if (closing) return false; - comp = sessionp->messageStop(name, true); + sessionp->messageStop(name, true); return true; } void AmqpSession::AcceptAndComplete(SequenceSet& transfers) { - lock l(waiters); + lock l(sessionLock); + + // delimit with session dtx commands depending on the transaction context + UpdateTransactionState(%l); - if (sessionp == NULL) - throw gcnew ObjectDisposedException("Accept"); + CheckOpen(); sessionp->markCompleted(transfers, false); sessionp->messageAccept(transfers, false); } +// call with session lock held + +void AmqpSession::UpdateTransactionState(lock^ slock) +{ + Transaction^ currentTx = Transaction::Current; + if ((currentTx == nullptr) && !dtxEnabled) { + // no transaction scope and no previous dtx work to monitor + return; + } + + if (currentTx == openSystemTransaction) { + // no change + return; + } + + if (!dtxEnabled) { + // AMQP requires that this be the first dtx-related command on the session + sessionp->dtxSelect(false); + dtxEnabled = true; + pendingTransactions = gcnew Collections::Generic::List<XaTransaction^>(); + } + + bool notify = false; // unless the System.Transaction is no longer active + XaTransaction^ oldXaTx = openXaTransaction; + if (openSystemTransaction != nullptr) { + // The application may start a new transaction before the phase0 on rollback + try { + if (openSystemTransaction->TransactionInformation->Status != TransactionStatus::Active) { + notify = true; + } + } catch (System::ObjectDisposedException^) { + notify = true; + } + } + + slock->release(); + // only use stack variables until lock re-acquired + + if (notify) { + // will do call back to all enlisted sessions. call with session lock released. + // If NotifyPhase0() wins the race to start phase 0, openXaTransaction will be null + oldXaTx->NotifyPhase0(); + } + + XaTransaction^ newXaTx = nullptr; + if (currentTx != nullptr) { + // This must be called with locks released. The DTC and System.Transactions methods that + // will be called hold locks that interfere with the ITransactionResourceAsync callbacks. + newXaTx = DtxResourceManager::GetXaTransaction(this, currentTx); + } + + slock->acquire(); + + if (closing) + return; + + if (openSystemTransaction != nullptr) { + // some other transaction has the dtx window open + // close the XID window, suspend = true... in case it is used again + sessionp->dtxEnd(getXid(openXaTransaction), false, true, false); + openSystemTransaction = nullptr; + openXaTransaction = nullptr; + } + + + // Call enlist with session lock held. The XaTransaction will call DtxStart before returning. + if (newXaTx != nullptr) { + if (!pendingTransactions->Contains(newXaTx)) { + pendingTransactions->Add(newXaTx); + } + + newXaTx->Enlist(this); + } + + openXaTransaction = newXaTx; + openSystemTransaction = currentTx; +} + + +typedef TypedResult<qpid::framing::XaResult> XaResultCompletion; + + +// send the required closing dtx.End before Phase 1 + +IntPtr AmqpSession::BeginPhase0Flush(XaTransaction ^xaTx) { + + lock l(sessionLock); + IntPtr completionp = IntPtr::Zero; + try { + if (sessionp != NULL) { + + // proceed even if "closing == true", the phase 0 is part of the transition from closing to closed + + if (xaTx != openXaTransaction) { + // a different transaction (or none) is in scope, so xaTx was previously suspended. + // must re-open it to close it properly + if (openXaTransaction != nullptr) { + // suspend the session's current pending transaction + // it wil be reopened in a future enlistment or phase 0 flush. + sessionp->dtxEnd(getXid(openXaTransaction), false, true, false); + } + // resuming + sessionp->dtxStart(getXid(xaTx), false, true, false); + } + + // the closing (i.e. non-suspended) dtxEnd happens here (exactly once for a given transaction) + // set the sync bit since phase0 is a precondition to prepare or abort + completionp = (IntPtr) new XaResultCompletion(sessionp->dtxEnd(getXid(xaTx), false, false, true)); + IncrementSyncs(); + } + } + catch (System::Exception^ ) { + // all the caller wants to know is if completionp is non-null + } + + openXaTransaction = nullptr; + openSystemTransaction = nullptr; + return completionp; +} + + +void AmqpSession::EndPhase0Flush(XaTransaction ^xaTx, IntPtr intptr) { + XaResultCompletion *completionp = (XaResultCompletion *) intptr.ToPointer(); + lock l(sessionLock); + + if (completionp != NULL) { + try { + l.release(); + completionp->wait(); + pendingTransactions->Remove(xaTx); + } + catch (System::Exception^) { + // connection closed or network drop + } + finally { + l.acquire(); + DecrementSyncs(); + delete completionp; + } + } +} + + +IntPtr AmqpSession::DtxStart(IntPtr ip, bool join, bool resume) { + // called with session lock held (as a callback from the Enlist()) + // The XaTransaction knows if this should be the originating dtxStart, or a join/resume + IntPtr rv = IntPtr::Zero; + qpid::framing::Xid* xidp = (qpid::framing::Xid *) ip.ToPointer(); + if (join || resume) { + sessionp->dtxStart(*xidp, join, resume, false); + } + else { + // The XaTransaction needs to track when the first dtxStart completes to safely request a join + IncrementSyncs(); // caller must use ReleaseCompletion() for corresponding DecrementSyncs + rv = (IntPtr) new XaResultCompletion(sessionp->dtxStart(*xidp, join, resume, false)); + } + + return rv; +} + + +IntPtr AmqpSession::DtxPrepare(IntPtr ip) { + qpid::framing::Xid* xidp = (qpid::framing::Xid *) ip.ToPointer(); + lock l(sessionLock); + + if (closing) + return IntPtr::Zero; + + IncrementSyncs(); // caller must use ReleaseCompletion() for corresponding DecrementSyncs + return (IntPtr) new XaResultCompletion(sessionp->dtxPrepare(*xidp, true)); +} + + +IntPtr AmqpSession::DtxCommit(IntPtr ip, bool onePhase) { + qpid::framing::Xid* xidp = (qpid::framing::Xid *) ip.ToPointer(); + lock l(sessionLock); + + if (closing) + return IntPtr::Zero; + + IncrementSyncs(); // caller must use ReleaseCompletion() for corresponding DecrementSyncs + return (IntPtr) new XaResultCompletion(sessionp->dtxCommit(*xidp, onePhase, true)); +} + + +IntPtr AmqpSession::DtxRollback(IntPtr ip) { + qpid::framing::Xid* xidp = (qpid::framing::Xid *) ip.ToPointer(); + lock l(sessionLock); + if (closing) + return IntPtr::Zero; + + IncrementSyncs(); // caller must use ReleaseCompletion() for corresponding DecrementSyncs + + return (IntPtr) new XaResultCompletion(sessionp->dtxRollback(*xidp, true)); +} + + +//call with lock held +void AmqpSession::IncrementSyncs() { + syncCount++; +} + + +//call with lock held +void AmqpSession::DecrementSyncs() { + syncCount--; + Debug::Assert(syncCount >= 0, "sync counter underrun"); + if (syncCount == 0) { + if (closeWaitHandle != nullptr) { + // now OK to move from closing to closed + closeWaitHandle->Set(); + } + } +} + + +// call with lock held +void AmqpSession::WaitLastSync(lock ^l) { + if (syncCount == 0) + return; + if (AppDomain::CurrentDomain->IsFinalizingForUnload()) { + // a wait would be a hang. No more syncs coming + return; + } + if (closeWaitHandle == nullptr) + closeWaitHandle = gcnew ManualResetEvent(false); + l->release(); + closeWaitHandle->WaitOne(); + l->acquire(); +} + + +void AmqpSession::ReleaseCompletion(IntPtr completion) { + lock l(sessionLock); + DecrementSyncs(); + delete completion.ToPointer(); +} + }}} // namespace Apache::Qpid::Cli diff --git a/qpid/wcf/src/Apache/Qpid/Interop/AmqpSession.h b/qpid/wcf/src/Apache/Qpid/Interop/AmqpSession.h index 8306cdf720..88ffd18dcc 100644 --- a/qpid/wcf/src/Apache/Qpid/Interop/AmqpSession.h +++ b/qpid/wcf/src/Apache/Qpid/Interop/AmqpSession.h @@ -29,29 +29,50 @@ namespace Interop { using namespace System; using namespace System::Runtime::InteropServices; +using namespace System::Transactions; +using namespace System::Diagnostics; + using namespace qpid::client; using namespace std; ref class InputLink; ref class OutputLink; +ref class XaTransaction; public ref class AmqpSession { private: + Object^ sessionLock; + Object^ openCloseLock; AmqpConnection^ connection; - Connection* connectionp; AsyncSession* sessionp; SessionImpl* sessionImplp; SubscriptionManager* subs_mgrp; - LocalQueue* localQueuep; Collections::Generic::List<CompletionWaiter^>^ waiters; bool helperRunning; + + // number of active InputLinks and OutputLinks int openCount; + // the number of async commands sent to the broker that need completion confirmation + int syncCount; + + bool closing; + ManualResetEvent^ closeWaitHandle; + bool dtxEnabled; + Transaction^ openSystemTransaction; + XaTransaction^ openXaTransaction; + Collections::Generic::List<XaTransaction^>^ pendingTransactions; + void Cleanup(); + void CheckOpen(); void asyncHelper(Object ^); void addWaiter(CompletionWaiter^ waiter); + void UpdateTransactionState(msclr::lock^ sessionLock); + void IncrementSyncs(); + void DecrementSyncs(); + void WaitLastSync(msclr::lock^ l); public: OutputLink^ CreateOutputLink(System::String^ targetQueue); @@ -66,16 +87,21 @@ internal: void NotifyClosed(); CompletionWaiter^ SendMessage (System::String^ queue, MessageBodyStream ^mbody, TimeSpan timeout, bool async, AsyncCallback^ callback, Object^ state); void ConnectionClosed(); - void internalWaitForCompletion(IntPtr Future); + void internalWaitForCompletion(IntPtr future); void removeWaiter(CompletionWaiter^ waiter); - bool MessageStop(Completion &comp, std::string &name); + bool MessageStop(std::string &name); void AcceptAndComplete(SequenceSet& transfers); + IntPtr BeginPhase0Flush(XaTransaction^); + void EndPhase0Flush(XaTransaction^, IntPtr); + IntPtr DtxStart(IntPtr xidp, bool, bool); + IntPtr DtxPrepare(IntPtr xidp); + IntPtr DtxCommit(IntPtr xidp, bool onePhase); + IntPtr DtxRollback(IntPtr xidp); + void ReleaseCompletion(IntPtr completion); property AmqpConnection^ Connection { AmqpConnection^ get () { return connection; } } - - }; }}} // namespace Apache::Qpid::Interop diff --git a/qpid/wcf/src/Apache/Qpid/Interop/DtxResourceManager.cpp b/qpid/wcf/src/Apache/Qpid/Interop/DtxResourceManager.cpp new file mode 100644 index 0000000000..6ea31f8401 --- /dev/null +++ b/qpid/wcf/src/Apache/Qpid/Interop/DtxResourceManager.cpp @@ -0,0 +1,285 @@ +/*
+* 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 <windows.h>
+#include <msclr\lock.h>
+#include <transact.h>
+#include <xolehlp.h>
+#include <txdtc.h>
+#include <oletx2xa.h>
+#include <iostream>
+#include <fstream>
+
+#include "qpid/client/AsyncSession.h"
+#include "qpid/client/SubscriptionManager.h"
+#include "qpid/client/Connection.h"
+#include "qpid/client/Message.h"
+#include "qpid/client/MessageListener.h"
+#include "qpid/framing/FrameSet.h"
+
+#include "AmqpConnection.h"
+#include "AmqpSession.h"
+#include "DtxResourceManager.h"
+#include "XaTransaction.h"
+#include "QpidException.h"
+#include "QpidMarshal.h"
+
+namespace Apache {
+namespace Qpid {
+namespace Interop {
+
+using namespace System;
+using namespace System::Runtime::InteropServices;
+using namespace System::Transactions;
+using namespace msclr;
+
+
+/*
+ * There is one DtxResourceManager per broker and per application process.
+ *
+ * Each RM manages a collection of active XaTransaction objects. Participating AmqpSessions enlist
+ * (or re-enlist) with an XaTransaction indexed by the corresponding System.Transaction object. The
+ * RM maintains its own AmqpSession for sending 2PC commnds (dtxPrepare, dtxCommit etc.). The
+ * XaTransaction object works through the lifecycle of the Transaction, including prompting the
+ * enlisted sessions to send their delimiting dtxEnd commands.
+ *
+ * A separate DtcPlugin.cpp file provides the recovery logic when needed in a library named
+ * qpidxarm.dll. The MSDTC maintans recovery info in its log and tracks when there may be
+ * transactions in doubt. See the documentation for IDtcToXaHelperSinglePipe.
+ *
+ * To enable transaction support:
+ * DTC requires a registry key to find the plugin
+ * [HKEY_LOCAL_MACHINE\SOFTWARE\Microsoft\MSDTC\XADLL] qpidxarm.dll -> [path to qpidxarm.dll]
+ * DTC needs to be configured for XA
+ * cmdprompt -> dcomcnfg -> Component services -> My Computer -> DTC -> Local DTC -> right click properties -> Security -> Enable XA Transactions
+ *
+ */
+
+// TODO: provide shutdown mechanism, perhaps callback from Connection Idle for enlisted connections.
+// But note that a new RM registration with the DTC is very expensive.
+
+
+DtxResourceManager::DtxResourceManager(AmqpConnection^ appConnection) {
+ dtcComp = NULL;
+ xaHelperp = NULL;
+ rmCookie = 0;
+ doubtCount = 0;
+ tmDown = false;
+ AmqpConnection^ clonedCon = appConnection->Clone();
+ dtxControlSession = clonedCon->CreateSession();
+ dataSourceName = clonedCon->DataSourceName;
+ transactionMap = gcnew Collections::Generic::Dictionary<Transaction^, XaTransaction^>();
+
+ HRESULT hr;
+
+ try {
+ // instead of pinning this instance, just use tmp stack variables for small stuff
+ IUnknown* tmp = NULL;
+ // request the default DTC
+ hr = DtcGetTransactionManager(NULL, NULL, IID_IUnknown, 0, 0, 0, (void **)&tmp);
+ if (hr != S_OK)
+ throw gcnew QpidException("connection failure to DTC service");
+ dtcComp = tmp;
+
+ IDtcToXaHelperSinglePipe *tmp2 = NULL;
+ hr = ((IUnknown *)dtcComp)->QueryInterface(IID_IDtcToXaHelperSinglePipe, (void**) &tmp2);
+ if (hr != S_OK)
+ throw gcnew QpidException("DTC XA unavailable");
+ xaHelperp = tmp2;
+
+ std::string native_dsn = QpidMarshal::ToNative(dataSourceName);
+ DWORD tmp3;
+
+ // This call doesn't return until the DTC has opened and closed a connection to the broker
+ // and written a recovery entry in its log.
+ hr = ((IDtcToXaHelperSinglePipe *) xaHelperp)->XARMCreate(const_cast<char *>(native_dsn.c_str()), "qpidxarm.dll", &tmp3);
+ if (hr != S_OK) {
+ switch (hr) {
+ case E_FAIL:
+ throw gcnew QpidException("Resource Manager DLL configuration error");
+ case E_INVALIDARG:
+ throw gcnew QpidException("Resource Manager internal error");
+ case E_OUTOFMEMORY:
+ throw gcnew QpidException("Resource Manager out of memory");
+ case E_UNEXPECTED:
+ throw gcnew QpidException("Resource Manager internal failure");
+ case XACT_E_TMNOTAVAILABLE:
+ case XACT_E_CONNECTION_DOWN:
+ throw gcnew QpidException("MSDTC unavailable");
+
+ default:
+ throw gcnew QpidException("Resource Manager Registration failed");
+ }
+ }
+
+ rmCookie = tmp3;
+ }
+ finally {
+ if (rmCookie == 0) {
+ // undo partial construction
+ Cleanup();
+ }
+ }
+}
+
+
+DtxResourceManager::!DtxResourceManager() {
+ Cleanup();
+}
+
+
+DtxResourceManager::~DtxResourceManager() {
+ GC::SuppressFinalize(this);
+ Cleanup();
+}
+
+
+// Called when the DTC COM proxy sends TMDOWN to a pending XaTransaction
+// called once for each outstanding tx
+
+void DtxResourceManager::TmDown() {
+ // this block is the only place where both locks are held
+ lock l1(transactionMap);
+ lock l2(resourceManagerMap);
+ if (tmDown)
+ return;
+
+ tmDown = true;
+ resourceManagerMap->Remove(this->dataSourceName);
+ // defer cleanup until last TmDown notification received
+}
+
+
+
+void DtxResourceManager::Cleanup() {
+ for each (Collections::Generic::KeyValuePair<Transaction^, XaTransaction^> kvp in transactionMap) {
+ XaTransaction^ xaTr = kvp.Value;
+ xaTr->ChildFinalize();
+ }
+
+ try {
+ if (rmCookie != 0) {
+ // implies no recovery needed
+ bool cleanSession = (doubtCount == 0) && (transactionMap->Count == 0);
+ ((IDtcToXaHelperSinglePipe *)xaHelperp)->ReleaseRMCookie(rmCookie, cleanSession);
+ rmCookie = 0;
+ }
+
+
+ if (xaHelperp != NULL) {
+ ((IDtcToXaHelperSinglePipe *) xaHelperp)->Release();
+ xaHelperp = NULL;
+ }
+
+ if (dtcComp != NULL) {
+ ((IUnknown *) dtcComp)->Release();
+ dtcComp = NULL;
+ }
+
+ if (dtxControlSession != nullptr) {
+ dtxControlSession->Connection->Close();
+ }
+
+ }
+ catch (Exception^) {}
+}
+
+
+XaTransaction^ DtxResourceManager::GetXaTransaction(AmqpSession^ appSession, Transaction^ transaction) {
+ // find or create the RM instance associated with the session's broker
+ AmqpConnection^ connection = appSession->Connection;
+ DtxResourceManager^ instance = connection->CachedResourceManager;
+
+ // try cached rm first
+ if (instance != nullptr) {
+ XaTransaction^ xaTx = instance->InternalGetXaTransaction(appSession, transaction);
+ if (xaTx != nullptr)
+ return xaTx;
+ else {
+ // cached version no longer available, force new rm creation
+ connection->CachedResourceManager = nullptr;
+ }
+ }
+
+ lock l(resourceManagerMap);
+ String^ dsn = connection->DataSourceName;
+ if (!resourceManagerMap->TryGetValue(dsn, instance)) {
+ instance = gcnew DtxResourceManager(connection->Clone());
+ resourceManagerMap->Add(dsn, instance);
+ connection->CachedResourceManager = instance;
+ }
+ l.release();
+
+ return instance->InternalGetXaTransaction(appSession, transaction);
+}
+
+
+XaTransaction^ DtxResourceManager::InternalGetXaTransaction(AmqpSession^ appSession, Transaction^ transaction) {
+ // find or create the tx proxy instance associated with the DTC transaction
+ lock l(transactionMap);
+ if (tmDown)
+ return nullptr;
+
+ XaTransaction^ xaTransaction = nullptr;
+ if (!transactionMap->TryGetValue(transaction, xaTransaction)) {
+ xaTransaction = gcnew XaTransaction(transaction, (IDtcToXaHelperSinglePipe *) xaHelperp, rmCookie, this);
+ transactionMap->Add(transaction, xaTransaction);
+ }
+
+ return xaTransaction;
+}
+
+void DtxResourceManager::Complete(Transaction ^tx) {
+ lock l(transactionMap);
+ transactionMap->Remove(tx);
+
+ if (tmDown && (transactionMap->Count == 0)) {
+ // no more activity on this instance
+ GC::SuppressFinalize(this);
+ Cleanup();
+ }
+}
+
+
+void DtxResourceManager::IncrementDoubt() {
+ Interlocked::Increment(doubtCount);
+}
+
+
+void DtxResourceManager::DecrementDoubt() {
+ Interlocked::Decrement(doubtCount);
+}
+
+
+#ifdef QPID_RECOVERY_TEST_HOOK
+void DtxResourceManager::ForceRecovery(Transaction ^tx) {
+ lock l(resourceManagerMap);
+ for each (Collections::Generic::KeyValuePair<System::String^, DtxResourceManager^> kvp in resourceManagerMap) {
+
+ Collections::Generic::Dictionary<Transaction^, XaTransaction^>^ txmap = kvp.Value->transactionMap;
+ XaTransaction^ xaTransaction = nullptr;
+ lock l2(txmap);
+ if (txmap->TryGetValue(tx, xaTransaction)) {
+ xaTransaction->ForceRecovery();
+ }
+ }
+}
+#endif
+
+}}} // namespace Apache::Qpid::Interop
diff --git a/qpid/wcf/src/Apache/Qpid/Interop/DtxResourceManager.h b/qpid/wcf/src/Apache/Qpid/Interop/DtxResourceManager.h new file mode 100644 index 0000000000..7df491eec2 --- /dev/null +++ b/qpid/wcf/src/Apache/Qpid/Interop/DtxResourceManager.h @@ -0,0 +1,76 @@ +/*
+* 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.
+*/
+
+#pragma once
+
+namespace Apache {
+namespace Qpid {
+namespace Interop {
+
+using namespace System;
+using namespace System::Threading;
+using namespace System::Transactions;
+
+ref class XaTransaction;
+
+public ref class DtxResourceManager
+{
+private:
+ // Receive() or WaitForMessage()
+ AmqpSession^ dtxControlSession;
+ String^ dataSourceName;
+ bool consumed;
+ DWORD rmCookie;
+ void* xaHelperp;
+ void* dtcComp;
+ int doubtCount;
+ DtxResourceManager(AmqpConnection^);
+ XaTransaction^ InternalGetXaTransaction (AmqpSession^ session, Transaction^ transaction);
+ bool tmDown;
+
+ // The active transactions
+ Collections::Generic::Dictionary<Transaction^, XaTransaction^>^ transactionMap;
+
+ // one resource manager per AMQP broker per process
+ static Collections::Generic::Dictionary<System::String^, DtxResourceManager^>^ resourceManagerMap =
+ gcnew Collections::Generic::Dictionary<System::String^, DtxResourceManager^>();
+
+ void Cleanup();
+ ~DtxResourceManager();
+ !DtxResourceManager();
+
+internal:
+ static XaTransaction^ GetXaTransaction (AmqpSession^ session, Transaction^ transaction);
+ void Complete(Transaction ^tx);
+ void TmDown();
+
+ property AmqpSession^ DtxControlSession {
+ AmqpSession^ get () { return dtxControlSession; }
+ }
+
+ void IncrementDoubt();
+ void DecrementDoubt();
+
+#ifdef QPID_RECOVERY_TEST_HOOK
+public:
+ static void ForceRecovery(Transaction ^tx);
+#endif
+};
+
+}}} // namespace Apache::Qpid::Interop
diff --git a/qpid/wcf/src/Apache/Qpid/Interop/InputLink.cpp b/qpid/wcf/src/Apache/Qpid/Interop/InputLink.cpp index e12151d943..3245cd3540 100644 --- a/qpid/wcf/src/Apache/Qpid/Interop/InputLink.cpp +++ b/qpid/wcf/src/Apache/Qpid/Interop/InputLink.cpp @@ -86,6 +86,8 @@ InputLink::InputLink(AmqpSession^ session, System::String^ sourceQueue, System::Exception^ linkException = nullptr; waiters = gcnew Collections::Generic::List<MessageWaiter^>(); + linkLock = waiters; // private and available + subscriptionLock = gcnew Object(); try { std::string qname = QpidMarshal::ToNative(sourceQueue); @@ -120,10 +122,13 @@ InputLink::InputLink(AmqpSession^ session, System::String^ sourceQueue, } } +// called with lock held void InputLink::ReleaseNative() { // involves talking to the Broker unless the connection is broken - if (subscriptionp != NULL) { + + if ((subscriptionp != NULL) && !finalizing) { + // TODO: find boost time error on cleanup when in finalizer thread try { subscriptionp->cancel(); } @@ -134,20 +139,31 @@ void InputLink::ReleaseNative() } // free native mem (or smart pointers) that we own - if (subscriptionp != NULL) + if (subscriptionp != NULL) { delete subscriptionp; - if (queuePtrp != NULL) + subscriptionp = NULL; + } + if (queuePtrp != NULL) { delete queuePtrp; - if (localQueuep != NULL) - delete localQueuep; - if (dequeuedFrameSetpp != NULL) + queuePtrp = NULL; + } + if (localQueuep != NULL) { + if (!finalizing) { + // TODO: find boost time error on cleanup when in finalizer thread + delete localQueuep; + localQueuep = NULL; + } + } + if (dequeuedFrameSetpp != NULL) { delete dequeuedFrameSetpp; + dequeuedFrameSetpp = NULL; + } } void InputLink::Cleanup() { { - lock l(waiters); + lock l(linkLock); if (disposed) return; @@ -162,6 +178,9 @@ void InputLink::Cleanup() if (queuePtrp != NULL) (*queuePtrp)->close(); + // wait for any sync operations on the subscription to complete before ReleaseNative + lock l2(subscriptionLock); + try {} finally { @@ -179,6 +198,7 @@ InputLink::~InputLink() InputLink::!InputLink() { + finalizing = true; Cleanup(); } @@ -204,7 +224,7 @@ bool InputLink::haveMessage() IntPtr InputLink::nextLocalMessage() { - lock l(waiters); + lock l(linkLock); if (disposed) return (IntPtr) NULL; @@ -250,7 +270,7 @@ IntPtr InputLink::nextLocalMessage() void InputLink::unblockWaiter() { // to be followed by resetQueue() below - lock l(waiters); + lock l(linkLock); if (disposed) return; (*queuePtrp)->close(); @@ -264,7 +284,7 @@ void InputLink::unblockWaiter() void InputLink::resetQueue() { - lock l(waiters); + lock l(linkLock); if (disposed) return; if ((*queuePtrp)->isClosed()) { @@ -282,7 +302,7 @@ bool InputLink::internalWaitForMessage() bool received = false; QpidFrameSetPtr* frameSetpp = NULL; try { - lock l(waiters); + lock l(linkLock); if (disposed) return false; if (haveMessage()) @@ -348,7 +368,7 @@ void InputLink::addWaiter(MessageWaiter^ waiter) void InputLink::removeWaiter(MessageWaiter^ waiter) { // a waiter can be removed from anywhere in the list if timed out - lock l(waiters); + lock l(linkLock); int idx = waiters->IndexOf(waiter); if (idx == -1) { // TODO: assert or log @@ -388,7 +408,7 @@ void InputLink::removeWaiter(MessageWaiter^ waiter) { void InputLink::asyncHelper() { - lock l(waiters); + lock l(linkLock); while (true) { if (disposed && (waiters->Count == 0)) { @@ -419,14 +439,14 @@ void InputLink::asyncHelper() void InputLink::sync() { - // for the timeout thread - lock l(waiters); + // used by the MessageWaiter timeout thread to not run before fully initialized + lock l(linkLock); } void InputLink::PrefetchLimit::set(int value) { - lock l(waiters); + lock l(linkLock); prefetchLimit = value; int delta = 0; @@ -475,31 +495,32 @@ void InputLink::AdjustCredit() void InputLink::SyncCredit(Object ^unused) { - lock l(waiters); + lock l(linkLock); try { if (disposed) return; - Completion comp; - if (!amqpSession->MessageStop(comp, subscriptionp->getName())) { + if (!amqpSession->MessageStop(subscriptionp->getName())) { // connection closed return; } - // get a private scoped copy to use outside the lock - Subscription s(*subscriptionp); - l.release(); // use setFlowControl to re-enable credit flow on the broker. - // previously used comp.wait() here, but setFlowControl is a sync operation - s.setFlowControl(s.getSettings().flowControl); + // setFlowControl is a sync operation + { + lock l2(subscriptionLock); + if (subscriptionp != NULL) { + subscriptionp->setFlowControl(subscriptionp->getSettings().flowControl); + } + } l.acquire(); if (disposed) return; - // let existing waiters use up any + // let existing waiters use up any messages that arrived. // local queue size can only decrease until more credit is issued while (true) { if ((waiters->Count > 0) && ((*queuePtrp)->size() > 0)) { @@ -700,7 +721,7 @@ AmqpMessage^ InputLink::createAmqpMessage(IntPtr msgp) bool InputLink::TryReceive(TimeSpan timeout, [Out] AmqpMessage^% amqpMessage) { - lock l(waiters); + lock l(linkLock); if (waiters->Count == 0) { // see if there is a message already available without blocking @@ -740,7 +761,7 @@ IAsyncResult^ InputLink::BeginTryReceive(TimeSpan timeout, AsyncCallback^ callba //TODO: if haveMessage() complete synchronously - lock l(waiters); + lock l(linkLock); MessageWaiter^ waiter = gcnew MessageWaiter(this, timeout, true, true, callback, state); addWaiter(waiter); return waiter; @@ -779,7 +800,10 @@ bool InputLink::EndTryReceive(IAsyncResult^ result, [Out] AmqpMessage^% amqpMess bool InputLink::WaitForMessage(TimeSpan timeout) { - lock l(waiters); + lock l(linkLock); + + if (disposed) + return false; if (waiters->Count == 0) { // see if there is a message already available without blocking @@ -799,12 +823,12 @@ bool InputLink::WaitForMessage(TimeSpan timeout) return false; } - return true; + return haveMessage(); } IAsyncResult^ InputLink::BeginWaitForMessage(TimeSpan timeout, AsyncCallback^ callback, Object^ state) { - lock l(waiters); + lock l(linkLock); // Same as for BeginTryReceive, except consuming = false MessageWaiter^ waiter = gcnew MessageWaiter(this, timeout, false, true, callback, state); @@ -822,7 +846,7 @@ bool InputLink::EndWaitForMessage(IAsyncResult^ result) return false; } - return true; + return haveMessage(); } diff --git a/qpid/wcf/src/Apache/Qpid/Interop/InputLink.h b/qpid/wcf/src/Apache/Qpid/Interop/InputLink.h index f59a03a8c3..2f96b91944 100644 --- a/qpid/wcf/src/Apache/Qpid/Interop/InputLink.h +++ b/qpid/wcf/src/Apache/Qpid/Interop/InputLink.h @@ -45,6 +45,8 @@ private: Collections::Generic::List<MessageWaiter^>^ waiters; bool disposed; bool finalizing; + Object^ linkLock; + Object^ subscriptionLock; QpidFrameSetPtr* dequeuedFrameSetpp; ManualResetEvent^ asyncHelperWaitHandle; // number of messages to buffer locally for future consumption diff --git a/qpid/wcf/src/Apache/Qpid/Interop/Interop.vcproj b/qpid/wcf/src/Apache/Qpid/Interop/Interop.vcproj index b662be9d54..2056c97d57 100644 --- a/qpid/wcf/src/Apache/Qpid/Interop/Interop.vcproj +++ b/qpid/wcf/src/Apache/Qpid/Interop/Interop.vcproj @@ -32,6 +32,9 @@ <Platform
Name="Win32"
/>
+ <Platform
+ Name="x64"
+ />
</Platforms>
<ToolFiles>
</ToolFiles>
@@ -63,10 +66,10 @@ />
<Tool
Name="VCCLCompilerTool"
- AdditionalOptions="/FU Debug\Apache.Qpid.AmqpTypes.netmodule"
+ AdditionalOptions=" /Zm1000 /wd4244 /wd4800 /wd4355 /FU $(ConfigurationName)\Apache.Qpid.AmqpTypes.netmodule"
Optimization="0"
AdditionalIncludeDirectories=""$(QPID_BUILD_ROOT)\include";"$(QPID_BUILD_ROOT)\src";..\..\..\..\..\cpp\include;..\..\..\..\..\cpp\src;"$(BOOST_ROOT)""
- PreprocessorDefinitions="WIN32;_DEBUG;_CRT_NONSTDC_NO_WARNINGS;WIN32_LEAN_AND_MEAN;NOMINMAX;_SCL_SECURE_NO_WARNINGS;BOOST_ALL_DYN_LINK"
+ PreprocessorDefinitions="WIN32,_WINDOWS,_DEBUG,BOOST_ALL_DYN_LINK,_CRT_NONSTDC_NO_WARNINGS,NOMINMAX,WIN32_LEAN_AND_MEAN,_SCL_SECURE_NO_WARNINGS,HAVE_CONFIG_H"
RuntimeLibrary="3"
UsePrecompiledHeader="0"
WarningLevel="3"
@@ -83,8 +86,8 @@ />
<Tool
Name="VCLinkerTool"
- AdditionalOptions="$(QPID_BUILD_ROOT)\src\Debug\qpidcommond.lib $(QPID_BUILD_ROOT)\src\Debug\qpidclientd.lib Debug\Apache.Qpid.AmqpTypes.netmodule"
- AdditionalDependencies="$(NoInherit)"
+ AdditionalOptions=" /STACK:10000000 /machine:I386 $(ConfigurationName)\Apache.Qpid.AmqpTypes.netmodule"
+ AdditionalDependencies="$(NOINHERIT) kernel32.lib user32.lib gdi32.lib winspool.lib shell32.lib ole32.lib oleaut32.lib uuid.lib comdlg32.lib advapi32.lib xolehlp.lib $(QPID_BUILD_ROOT)\src\Debug\qpidclientd.lib $(QPID_BUILD_ROOT)\src\Debug\qpidcommond.lib rpcrt4.lib ws2_32.lib "
OutputFile="$(OutDir)\Apache.Qpid.Interop.dll"
LinkIncremental="2"
AdditionalLibraryDirectories=""$(BOOST_ROOT)\lib""
@@ -117,7 +120,7 @@ </Configuration>
<Configuration
Name="Release|Win32"
- OutputDirectory="$(SolutionDir)$(ConfigurationName)"
+ OutputDirectory="$(ProjectDir)$(ConfigurationName)"
IntermediateDirectory="$(ConfigurationName)"
ConfigurationType="2"
CharacterSet="1"
@@ -126,6 +129,7 @@ >
<Tool
Name="VCPreBuildEventTool"
+ CommandLine="copy ..\AmqpTypes\obj\$(ConfigurationName)\Apache.Qpid.AmqpTypes.netmodule $(ConfigurationName)"
/>
<Tool
Name="VCCustomBuildTool"
@@ -141,9 +145,12 @@ />
<Tool
Name="VCCLCompilerTool"
- PreprocessorDefinitions="WIN32;NDEBUG"
+ AdditionalOptions=" /Zm1000 /wd4244 /wd4800 /wd4355 /FU $(ConfigurationName)\Apache.Qpid.AmqpTypes.netmodule"
+ AdditionalIncludeDirectories=""$(QPID_BUILD_ROOT)\include";"$(QPID_BUILD_ROOT)\src";..\..\..\..\..\cpp\include;..\..\..\..\..\cpp\src;"$(BOOST_ROOT)""
+ PreprocessorDefinitions="WIN32,_WINDOWS,NDEBUG,BOOST_ALL_DYN_LINK,_CRT_NONSTDC_NO_WARNINGS,NOMINMAX,WIN32_LEAN_AND_MEAN,_SCL_SECURE_NO_WARNINGS,HAVE_CONFIG_H"
+ Optimization="2"
RuntimeLibrary="2"
- UsePrecompiledHeader="2"
+ UsePrecompiledHeader="0"
WarningLevel="3"
DebugInformationFormat="3"
/>
@@ -158,9 +165,12 @@ />
<Tool
Name="VCLinkerTool"
- AdditionalDependencies="$(NoInherit)"
- LinkIncremental="1"
- GenerateDebugInformation="true"
+ AdditionalOptions=" /STACK:10000000 /machine:I386 $(ConfigurationName)\Apache.Qpid.AmqpTypes.netmodule"
+ AdditionalDependencies="$(NOINHERIT) kernel32.lib user32.lib gdi32.lib winspool.lib shell32.lib ole32.lib oleaut32.lib uuid.lib comdlg32.lib advapi32.lib xolehlp.lib $(QPID_BUILD_ROOT)\src\Release\qpidclient.lib $(QPID_BUILD_ROOT)\src\Release\qpidcommon.lib rpcrt4.lib ws2_32.lib "
+ LinkIncremental="2"
+ OutputFile="$(OutDir)\Apache.Qpid.Interop.dll"
+ AdditionalLibraryDirectories=""$(BOOST_ROOT)\lib""
+ KeyFile="$(SolutionDir)\src\wcfnet.snk"
TargetMachine="1"
/>
<Tool
@@ -185,6 +195,165 @@ Name="VCPostBuildEventTool"
/>
</Configuration>
+ <Configuration
+ Name="Debug|x64"
+ OutputDirectory="$(ProjectDir)$(PlatformName)\$(ConfigurationName)"
+ IntermediateDirectory="$(PlatformName)\$(ConfigurationName)"
+ ConfigurationType="2"
+ CharacterSet="1"
+ ManagedExtensions="1"
+ WholeProgramOptimization="1"
+ >
+ <Tool
+ Name="VCPreBuildEventTool"
+ CommandLine="copy ..\AmqpTypes\obj\$(ConfigurationName)\Apache.Qpid.AmqpTypes.netmodule $(PlatformName)\$(ConfigurationName)"
+ />
+ <Tool
+ Name="VCCustomBuildTool"
+ />
+ <Tool
+ Name="VCXMLDataGeneratorTool"
+ />
+ <Tool
+ Name="VCWebServiceProxyGeneratorTool"
+ />
+ <Tool
+ Name="VCMIDLTool"
+ TargetEnvironment="3"
+ />
+ <Tool
+ Name="VCCLCompilerTool"
+ AdditionalOptions=" /Zm1000 /wd4244 /wd4800 /wd4355 /FU $(PlatformName)\$(ConfigurationName)\Apache.Qpid.AmqpTypes.netmodule"
+ Optimization="0"
+ AdditionalIncludeDirectories=""$(QPID_BUILD_ROOT)\include";"$(QPID_BUILD_ROOT)\src";..\..\..\..\..\cpp\include;..\..\..\..\..\cpp\src;"$(BOOST_ROOT)""
+ PreprocessorDefinitions="WIN32,_WINDOWS,_DEBUG,BOOST_ALL_DYN_LINK,_CRT_NONSTDC_NO_WARNINGS,NOMINMAX,WIN32_LEAN_AND_MEAN,_SCL_SECURE_NO_WARNINGS,HAVE_CONFIG_H"
+ RuntimeLibrary="3"
+ UsePrecompiledHeader="0"
+ WarningLevel="3"
+ DebugInformationFormat="3"
+ />
+ <Tool
+ Name="VCManagedResourceCompilerTool"
+ />
+ <Tool
+ Name="VCResourceCompilerTool"
+ />
+ <Tool
+ Name="VCPreLinkEventTool"
+ />
+ <Tool
+ Name="VCLinkerTool"
+ AdditionalOptions=" /STACK:10000000 /machine:x64 /debug $(PlatformName)\$(ConfigurationName)\Apache.Qpid.AmqpTypes.netmodule"
+ AdditionalDependencies="$(NoInherit) kernel32.lib user32.lib gdi32.lib winspool.lib shell32.lib ole32.lib oleaut32.lib uuid.lib comdlg32.lib advapi32.lib rpcrt4.lib ws2_32.lib xolehlp.lib $(QPID_BUILD_ROOT)\src\Debug\qpidcommond.lib $(QPID_BUILD_ROOT)\src\Debug\qpidclientd.lib"
+ OutputFile="$(OutDir)\Apache.Qpid.Interop.dll"
+ LinkIncremental="2"
+ AdditionalLibraryDirectories=""$(BOOST_ROOT)\lib""
+ GenerateDebugInformation="true"
+ AssemblyDebug="1"
+ TargetMachine="17"
+ KeyFile="$(SolutionDir)\src\wcfnet.snk"
+ />
+ <Tool
+ Name="VCALinkTool"
+ />
+ <Tool
+ Name="VCManifestTool"
+ />
+ <Tool
+ Name="VCXDCMakeTool"
+ />
+ <Tool
+ Name="VCBscMakeTool"
+ />
+ <Tool
+ Name="VCFxCopTool"
+ />
+ <Tool
+ Name="VCAppVerifierTool"
+ />
+ <Tool
+ Name="VCPostBuildEventTool"
+ />
+ </Configuration>
+ <Configuration
+ Name="Release|x64"
+ OutputDirectory="$(ProjectDir)$(PlatformName)\$(ConfigurationName)"
+ IntermediateDirectory="$(PlatformName)\$(ConfigurationName)"
+ ConfigurationType="2"
+ CharacterSet="1"
+ ManagedExtensions="1"
+ WholeProgramOptimization="1"
+ >
+ <Tool
+ Name="VCPreBuildEventTool"
+ CommandLine="copy ..\AmqpTypes\obj\$(ConfigurationName)\Apache.Qpid.AmqpTypes.netmodule $(PlatformName)\$(ConfigurationName)"
+ />
+ <Tool
+ Name="VCCustomBuildTool"
+ />
+ <Tool
+ Name="VCXMLDataGeneratorTool"
+ />
+ <Tool
+ Name="VCWebServiceProxyGeneratorTool"
+ />
+ <Tool
+ Name="VCMIDLTool"
+ TargetEnvironment="3"
+ />
+ <Tool
+ Name="VCCLCompilerTool"
+ AdditionalOptions=" /Zm1000 /wd4244 /wd4800 /wd4355 /FU $(PlatformName)\$(ConfigurationName)\Apache.Qpid.AmqpTypes.netmodule"
+ AdditionalIncludeDirectories=""$(QPID_BUILD_ROOT)\include";"$(QPID_BUILD_ROOT)\src";..\..\..\..\..\cpp\include;..\..\..\..\..\cpp\src;"$(BOOST_ROOT)""
+ PreprocessorDefinitions="WIN32,_WINDOWS,NDEBUG,BOOST_ALL_DYN_LINK,_CRT_NONSTDC_NO_WARNINGS,NOMINMAX,WIN32_LEAN_AND_MEAN,_SCL_SECURE_NO_WARNINGS,HAVE_CONFIG_H"
+ InlineFunctionExpansion="2"
+ Optimization="2"
+ RuntimeLibrary="2"
+ UsePrecompiledHeader="0"
+ WarningLevel="3"
+ DebugInformationFormat="3"
+ />
+ <Tool
+ Name="VCManagedResourceCompilerTool"
+ />
+ <Tool
+ Name="VCResourceCompilerTool"
+ />
+ <Tool
+ Name="VCPreLinkEventTool"
+ />
+ <Tool
+ Name="VCLinkerTool"
+ AdditionalOptions=" /STACK:10000000 /machine:x64 $(PlatformName)\$(ConfigurationName)\Apache.Qpid.AmqpTypes.netmodule"
+ AdditionalDependencies="$(NoInherit) kernel32.lib user32.lib gdi32.lib winspool.lib shell32.lib ole32.lib oleaut32.lib uuid.lib comdlg32.lib advapi32.lib xolehlp.lib $(QPID_BUILD_ROOT)\src\Release\qpidclient.lib $(QPID_BUILD_ROOT)\src\Release\qpidcommon.lib rpcrt4.lib ws2_32.lib"
+ LinkIncremental="2"
+ OutputFile="$(OutDir)\Apache.Qpid.Interop.dll"
+ AdditionalLibraryDirectories=""$(BOOST_ROOT)\lib""
+ KeyFile="$(SolutionDir)\src\wcfnet.snk"
+ TargetMachine="17"
+ />
+ <Tool
+ Name="VCALinkTool"
+ />
+ <Tool
+ Name="VCManifestTool"
+ />
+ <Tool
+ Name="VCXDCMakeTool"
+ />
+ <Tool
+ Name="VCBscMakeTool"
+ />
+ <Tool
+ Name="VCFxCopTool"
+ />
+ <Tool
+ Name="VCAppVerifierTool"
+ />
+ <Tool
+ Name="VCPostBuildEventTool"
+ />
+ </Configuration>
</Configurations>
<References>
<AssemblyReference
@@ -207,6 +376,11 @@ AssemblyName="System.Runtime.Serialization, Version=3.0.0.0, PublicKeyToken=b77a5c561934e089, processorArchitecture=MSIL"
MinFrameworkVersion="196608"
/>
+ <AssemblyReference
+ RelativePath="System.Transactions.dll"
+ AssemblyName="System.Transactions, Version=2.0.0.0, PublicKeyToken=b77a5c561934e089, processorArchitecture=x86"
+ MinFrameworkVersion="131072"
+ />
</References>
<Files>
<Filter
@@ -250,6 +424,14 @@ RelativePath=".\OutputLink.cpp"
>
</File>
+ <File
+ RelativePath=".\DtxResourceManager.cpp"
+ >
+ </File>
+ <File
+ RelativePath=".\XaTransaction.cpp"
+ >
+ </File>
</Filter>
<Filter
Name="Header Files"
@@ -296,6 +478,14 @@ RelativePath=".\QpidMarshal.h"
>
</File>
+ <File
+ RelativePath=".\DtxResourceManager.h"
+ >
+ </File>
+ <File
+ RelativePath=".\XaTransaction.h"
+ >
+ </File>
</Filter>
</Files>
<Globals>
diff --git a/qpid/wcf/src/Apache/Qpid/Interop/XaTransaction.cpp b/qpid/wcf/src/Apache/Qpid/Interop/XaTransaction.cpp new file mode 100644 index 0000000000..23743316ff --- /dev/null +++ b/qpid/wcf/src/Apache/Qpid/Interop/XaTransaction.cpp @@ -0,0 +1,525 @@ +/*
+* 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 <windows.h>
+#include <msclr\lock.h>
+#include <transact.h>
+#include <xolehlp.h>
+#include <txdtc.h>
+#include <oletx2xa.h>
+#include <iostream>
+#include <fstream>
+
+#include "qpid/client/AsyncSession.h"
+#include "qpid/client/SubscriptionManager.h"
+#include "qpid/client/Connection.h"
+#include "qpid/framing/FrameSet.h"
+#include "qpid/framing/Xid.h"
+
+#include "QpidException.h"
+#include "AmqpConnection.h"
+#include "AmqpSession.h"
+#include "DtxResourceManager.h"
+#include "XaTransaction.h"
+
+namespace Apache {
+namespace Qpid {
+namespace Interop {
+
+using namespace System;
+using namespace System::Runtime::InteropServices;
+using namespace System::Transactions;
+using namespace msclr;
+
+using namespace qpid::framing::dtx;
+
+// ------------------------------------------------------------------------
+// Start of a pure native code section
+#pragma unmanaged
+// ------------------------------------------------------------------------
+
+// This is the native COM object the DTC expects to talk to for coordination.
+// There is exactly one native instance of this for each managed XaTransaction object.
+
+
+class DtcCallbackHandler : public ITransactionResourceAsync
+{
+private:
+ long useCount;
+ DtcCallbackFp managedCallback;
+public:
+ ITransactionEnlistmentAsync *txHandle;
+ DtcCallbackHandler(DtcCallbackFp cbp) : managedCallback(cbp), useCount(0) {}
+ ~DtcCallbackHandler() {}
+ virtual HRESULT __stdcall PrepareRequest(BOOL unused, DWORD grfrm, BOOL unused2, BOOL singlePhase);
+ virtual HRESULT __stdcall CommitRequest(DWORD grfrm, XACTUOW *unused);
+ virtual HRESULT __stdcall AbortRequest(BOID *unused, BOOL unused2, XACTUOW *unused3);
+
+ virtual HRESULT __stdcall TMDown();
+ virtual HRESULT __stdcall DtcCallbackHandler::QueryInterface (REFIID riid, void **ppvObject);
+ virtual ULONG __stdcall DtcCallbackHandler::AddRef();
+ virtual ULONG __stdcall DtcCallbackHandler::Release();
+ void __stdcall AbortRequestDone();
+};
+
+
+HRESULT DtcCallbackHandler::PrepareRequest(BOOL unused, DWORD grfrm, BOOL unused2, BOOL singlePhase)
+{
+ if (singlePhase) {
+ return managedCallback(DTC_SINGLE_PHASE) ? S_OK : E_FAIL;
+ }
+
+ return managedCallback(DTC_PREPARE) ? S_OK : E_FAIL;
+}
+
+
+HRESULT DtcCallbackHandler::CommitRequest(DWORD grfrm, XACTUOW *unused)
+{
+ return managedCallback(DTC_COMMIT) ? S_OK : E_FAIL;
+}
+
+HRESULT DtcCallbackHandler::AbortRequest(BOID *unused, BOOL unused2, XACTUOW *unused3)
+{
+ return managedCallback(DTC_ABORT) ? S_OK : E_FAIL;
+}
+
+
+HRESULT DtcCallbackHandler::TMDown()
+{
+ return managedCallback(DTC_TMDOWN) ? S_OK : E_FAIL;
+}
+
+
+HRESULT DtcCallbackHandler::QueryInterface (REFIID riid, void **ppvObject)
+{
+ *ppvObject = NULL;
+
+ if ((riid == IID_IUnknown) || (riid == IID_IResourceManagerSink))
+ *ppvObject = this;
+ else
+ return ResultFromScode(E_NOINTERFACE);
+
+ this->AddRef();
+ return S_OK;
+}
+
+
+ULONG DtcCallbackHandler::AddRef()
+{
+ return InterlockedIncrement(&useCount);
+}
+
+
+ULONG DtcCallbackHandler::Release()
+{
+ long uc = InterlockedDecrement(&useCount);
+
+ if (uc)
+ return uc;
+
+ delete this;
+ return 0;
+}
+
+
+// ------------------------------------------------------------------------
+// End of pure native code section
+#pragma managed
+// ------------------------------------------------------------------------
+
+#ifdef QPID_RECOVERY_TEST_HOOK
+void XaTransaction::ForceRecovery() {
+ debugFailMode = true;
+}
+#endif
+
+// ------------------------------------------------------------------------
+// ------------------------------------------------------------------------
+
+
+XaTransaction::XaTransaction(Transaction^ t, IDtcToXaHelperSinglePipe *xaHelperp, DWORD rmCookie, DtxResourceManager^ rm) {
+ bool success = false;
+ xidp = NULL;
+ commandCompletionp = NULL;
+ firstDtxStartCompletionp = NULL;
+ nativeHandler = NULL;
+ resourceManager = rm;
+ controlSession = rm->DtxControlSession;
+ active = true;
+ preparing = false;
+ systemTransaction = t;
+ IntPtr comTxp = IntPtr::Zero;
+ completionHandle = gcnew ManualResetEvent(false);
+
+ try {
+ enlistedSessions = gcnew Collections::Generic::List<AmqpSession^>();
+
+ // take a System.Transactions.Transaction and obtain
+ // the corresponding DTC COM object.
+ IDtcTransaction^ dtcTransaction = TransactionInterop::GetDtcTransaction(t);
+ comTxp = Marshal::GetIUnknownForObject(dtcTransaction);
+ XID winXid;
+ HRESULT hr = xaHelperp->ConvertTridToXID((DWORD *)comTxp.ToPointer(), rmCookie, &winXid);
+ if (hr != S_OK)
+ throw gcnew QpidException("get XA XID");
+
+ // Convert the X/Open format to the internal Qpid format
+ xidp = new qpid::framing::Xid();
+ xidp->setFormat((uint32_t) winXid.formatID);
+ int bqualPos = 0;
+ if (winXid.gtrid_length > 0) {
+ xidp->setGlobalId(std::string(winXid.data, winXid.gtrid_length));
+ bqualPos = winXid.gtrid_length;
+ }
+ if (winXid.bqual_length > 0) {
+ xidp->setBranchId(std::string(winXid.data + bqualPos, winXid.bqual_length));
+ }
+
+ // create the callback chain: DTC proxy -> DtcCallbackHandler -> this
+ inboundDelegate = gcnew DtcCallbackDelegate(this, &XaTransaction::DtcCallback);
+ IntPtr ip = Marshal::GetFunctionPointerForDelegate(inboundDelegate);
+ nativeHandler = new DtcCallbackHandler(static_cast<DtcCallbackFp>(ip.ToPointer()));
+ // add myself for later smart pointer destruction
+ nativeHandler->AddRef();
+
+ hr = xaHelperp->EnlistWithRM(rmCookie, (ITransaction *)comTxp.ToPointer(), nativeHandler, &(nativeHandler->txHandle));
+
+ if (hr != S_OK)
+ throw gcnew QpidException("Enlist");
+
+ success = true;
+ }
+ finally {
+ if (!success)
+ Cleanup();
+ if (comTxp != IntPtr::Zero)
+ ((IUnknown *) comTxp.ToPointer())->Release();
+ }
+}
+
+
+void XaTransaction::Cleanup() {
+ if (firstDtxStartCompletionp != NULL) {
+ try {
+ firstEnlistedSession->ReleaseCompletion((IntPtr) firstDtxStartCompletionp);
+ }
+ catch (...) {
+ // TODO: log it?
+ }
+
+ firstDtxStartCompletionp = NULL;
+ }
+
+ if (nativeHandler != NULL) {
+ nativeHandler->Release();
+ nativeHandler = NULL;
+ }
+ if (xidp != NULL) {
+ delete xidp;
+ xidp = NULL;
+ }
+}
+
+
+XaTransaction^ XaTransaction::Enlist (AmqpSession ^session) {
+ lock l(enlistedSessions);
+ if (!active)
+ throw gcnew QpidException("transaction enlistment internal error");
+ if (!enlistedSessions->Contains(session)) {
+ enlistedSessions->Add(session);
+ if (firstEnlistedSession == nullptr) {
+ firstEnlistedSession = session;
+ IntPtr intptr = session->DtxStart((IntPtr) xidp, false, false);
+ firstDtxStartCompletionp = (TypedResult<qpid::framing::XaResult> *) intptr.ToPointer();
+ }
+ else {
+ // the broker must see the dtxStart as a join operation, and it must arrive
+ // at the broker after the first dtx start
+ if (firstDtxStartCompletionp != NULL)
+ firstDtxStartCompletionp->wait();
+ session->DtxStart((IntPtr) xidp, true, false);
+ }
+ }
+ else {
+ // already started once, so resume is true
+ session->DtxStart((IntPtr) xidp, false, true);
+ }
+ return this;
+}
+
+
+void XaTransaction::SessionClosing(AmqpSession^ session) {
+ lock l(enlistedSessions);
+ if (!enlistedSessions->Contains(session))
+ return;
+
+ enlistedSessions->Remove(session);
+ if (!active) {
+ // Phase0Flush already done on all sessions
+ l.release();
+ return;
+ }
+
+ IntPtr completion = session->BeginPhase0Flush(this);
+ session->EndPhase0Flush(this, completion);
+
+ if (session == firstEnlistedSession) {
+ // if we just completed the dtxEnd, we know the dtxStart completed before that
+ if (firstDtxStartCompletionp != NULL) {
+ firstEnlistedSession->ReleaseCompletion((IntPtr) firstDtxStartCompletionp);
+ firstDtxStartCompletionp = NULL;
+ }
+ }
+}
+
+
+void XaTransaction::Phase0Flush() {
+ // let each session delimit their transactional work with an AMQP dtx.end protocol frame
+ lock l(enlistedSessions);
+ if (!active)
+ return;
+
+ active = false; // no more enlistments
+ int scount = enlistedSessions->Count;
+
+ if (scount > 0) {
+ array<IntPtr> ^completions = gcnew array<IntPtr>(scount);
+ for (int i = 0; i < scount; i++) {
+
+ // TODO: skip phase0 flush for rollback case
+
+ completions[i] = enlistedSessions[i]->BeginPhase0Flush(this);
+ }
+
+ for (int i = 0; i < scount; i++) {
+ // without each session.sync(), session commands are queued up in the right order,
+ // but on their separate outbound channels, and destined for receipt at separate Broker inbound
+ // channels. It is not clear how to be sure Phase 0 dtx.End is processed in the
+ // correct order before commit on the broker without the sync.
+ enlistedSessions[i]->EndPhase0Flush(this, completions[i]);
+ }
+ }
+
+ // since all dtxEnds have completed, we know all starts have too
+ if (firstDtxStartCompletionp != NULL) {
+ try {
+ firstEnlistedSession->ReleaseCompletion((IntPtr) firstDtxStartCompletionp);
+ }
+ catch (...) {
+ // TODO: log it?
+ }
+
+ firstDtxStartCompletionp = NULL;
+ }
+}
+
+
+bool XaTransaction::DtcCallback (DtcCallbackType callback) {
+ // called by the DTC proxy thread. Be brief and don't block (but Phase0Flush?)
+
+ if (AppDomain::CurrentDomain->IsFinalizingForUnload())
+ return false;
+
+ IntPtr intptr = IntPtr::Zero;
+ currentCommand = callback;
+
+ try {
+ switch (callback) {
+ case DTC_PREPARE:
+ Phase0Flush();
+ try {
+ intptr = controlSession->DtxPrepare((IntPtr) xidp);
+ preparing = true;
+ resourceManager->IncrementDoubt();
+ }
+ catch (System::Exception^ ) {
+ // intptr remains nullptr
+ }
+ commandCompletionp = (TypedResult<qpid::framing::XaResult> *) intptr.ToPointer();
+ ThreadPool::QueueUserWorkItem(gcnew WaitCallback(this, &XaTransaction::AsyncCompleter));
+ break;
+
+ case DTC_COMMIT:
+#ifdef QPID_RECOVERY_TEST_HOOK
+ if (debugFailMode){ return; }
+#endif
+ // no phase 0 required. always preceded by a prepare
+ try {
+ intptr = controlSession->DtxCommit((IntPtr) xidp, false);
+ }
+ catch (System::Exception^ ) {
+ // intptr remains nullptr
+ }
+ commandCompletionp = (TypedResult<qpid::framing::XaResult> *) intptr.ToPointer();
+ ThreadPool::QueueUserWorkItem(gcnew WaitCallback(this, &XaTransaction::AsyncCompleter));
+ break;
+
+ case DTC_ABORT:
+ Phase0Flush();
+#ifdef QPID_RECOVERY_TEST_HOOK
+ if (debugFailMode){ return; }
+#endif
+ try {
+ intptr = controlSession->DtxRollback((IntPtr) xidp);
+ }
+ catch (System::Exception^ ) {
+ // intptr remains nullptr
+ }
+ commandCompletionp = (TypedResult<qpid::framing::XaResult> *) intptr.ToPointer();
+ ThreadPool::QueueUserWorkItem(gcnew WaitCallback(this, &XaTransaction::AsyncCompleter));
+ break;
+
+ case DTC_SINGLE_PHASE:
+ Phase0Flush();
+ try {
+ intptr = controlSession->DtxCommit((IntPtr) xidp, true);
+ }
+ catch (System::Exception^ ) {
+ // intptr remains nullptr
+ }
+ commandCompletionp = (TypedResult<qpid::framing::XaResult> *) intptr.ToPointer();
+ ThreadPool::QueueUserWorkItem(gcnew WaitCallback(this, &XaTransaction::AsyncCompleter));
+ break;
+
+ case DTC_TMDOWN:
+ commandCompletionp = NULL;
+ ThreadPool::QueueUserWorkItem(gcnew WaitCallback(this, &XaTransaction::AsyncCompleter));
+ break;
+ }
+ return true;
+ }
+ catch (System::Exception^ e) {
+ // TODO: log it
+ Console::WriteLine("Unexpected DtcCallback exception: {0}", e->ToString());
+ }
+ catch (...) {
+ // TODO: log it
+ }
+ return false;
+}
+
+
+// this handles the case where the application regains control for
+// a new transaction before we are notified (abort/rollback
+// optimization in DTC).
+
+void XaTransaction::NotifyPhase0() {
+ if (active)
+ Phase0Flush();
+}
+
+
+void XaTransaction::AsyncCompleter(Object ^unused) {
+ bool success = false;
+
+ if (commandCompletionp != NULL) {
+ try {
+ // waits for the AMQP broker's response and returns the decoded content
+ XaResult& xaResult = commandCompletionp->get();
+ if (xaResult.hasStatus()) {
+ if (xaResult.getStatus() == XaStatus::XA_STATUS_XA_OK) {
+ success = true;
+ }
+ }
+ }
+ catch (...) {
+ // TODO: log it?
+ }
+ try {
+ controlSession->ReleaseCompletion((IntPtr) commandCompletionp);
+ }
+ catch (...) {
+ // TODO: log it?
+ }
+
+ commandCompletionp = NULL;
+ }
+
+ ITransactionEnlistmentAsync *dtcTxHandle = nativeHandler->txHandle;
+
+ HRESULT hr = success ? S_OK : E_FAIL;
+
+ switch (currentCommand) {
+ case DTC_PREPARE:
+ dtcTxHandle->PrepareRequestDone(hr, NULL, NULL);
+ break;
+
+ case DTC_COMMIT:
+ dtcTxHandle->CommitRequestDone(hr);
+ if (success)
+ resourceManager->DecrementDoubt();
+ Complete();
+ break;
+
+ case DTC_ABORT:
+ dtcTxHandle->AbortRequestDone(hr);
+ if (success) {
+ if (preparing) {
+ preparing = false;
+ resourceManager->DecrementDoubt();
+ }
+ }
+ Complete();
+ break;
+
+ case DTC_SINGLE_PHASE:
+ if (success)
+ hr = XACT_S_SINGLEPHASE;
+ dtcTxHandle->PrepareRequestDone(hr, NULL, NULL);
+ Complete();
+ break;
+
+ case DTC_TMDOWN:
+ // Stop the RM from accepting new enlistments
+ resourceManager->TmDown();
+ Complete();
+ break;
+ }
+}
+
+
+void XaTransaction::Complete() {
+ Cleanup();
+ resourceManager->Complete(systemTransaction);
+ completionHandle->Set();
+}
+
+
+void XaTransaction::WaitForCompletion() {
+ completionHandle->WaitOne();
+}
+
+
+ /*
+void XaTransaction::WaitForFlush() {
+ isFlushedHandle->WaitOne();
+}
+ */
+
+// called from DtxResourceManager Finalize
+
+void XaTransaction::ChildFinalize() {
+ lock l(enlistedSessions);
+ Phase0Flush();
+ Cleanup();
+}
+
+
+
+}}} // namespace Apache::Qpid::Interop
diff --git a/qpid/wcf/src/Apache/Qpid/Interop/XaTransaction.h b/qpid/wcf/src/Apache/Qpid/Interop/XaTransaction.h new file mode 100644 index 0000000000..8ff9f99893 --- /dev/null +++ b/qpid/wcf/src/Apache/Qpid/Interop/XaTransaction.h @@ -0,0 +1,96 @@ +/*
+* 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.
+*/
+
+#pragma once
+
+namespace Apache {
+namespace Qpid {
+namespace Interop {
+
+using namespace System;
+using namespace System::Threading;
+using namespace System::Transactions;
+
+enum DtcCallbackType{
+ DTC_PREPARE,
+ DTC_COMMIT,
+ DTC_ABORT,
+ DTC_SINGLE_PHASE,
+ DTC_TMDOWN
+};
+
+
+ref class DtxResourceManager;
+class DtcCallbackHandler;
+
+// Function pointer declaratiom for managed space delegate
+typedef bool (__stdcall *DtcCallbackFp)(DtcCallbackType);
+
+// and the delegate with the same signature
+public delegate bool DtcCallbackDelegate(DtcCallbackType);
+
+
+
+public ref class XaTransaction
+{
+private:
+ bool active;
+ DtxResourceManager^ resourceManager;
+ Transaction^ systemTransaction;
+ AmqpSession^ controlSession;
+ Collections::Generic::List<AmqpSession^>^ enlistedSessions;
+ qpid::framing::Xid* xidp;
+ DtcCallbackHandler* nativeHandler;
+ bool preparing;
+ DtcCallbackDelegate^ inboundDelegate;
+ // the Qpid async result of the AMQP dtx prepare/commit commands
+ TypedResult<qpid::framing::XaResult>* commandCompletionp;
+ // the Qpid async result of the first session to do dtx start
+ TypedResult<qpid::framing::XaResult>* firstDtxStartCompletionp;
+ ManualResetEvent^ completionHandle;
+
+ AmqpSession^ firstEnlistedSession;
+ DtcCallbackType currentCommand;
+ void AsyncCompleter(Object ^);
+ void Phase0Flush();
+ void Cleanup();
+ void Complete();
+
+internal:
+ XaTransaction(Transaction^ t, IDtcToXaHelperSinglePipe *pXaHelper, DWORD rmCookie, DtxResourceManager^ rm);
+ XaTransaction^ Enlist (AmqpSession ^session);
+ bool DtcCallback (DtcCallbackType callback);
+ void NotifyPhase0();
+ void ChildFinalize();
+ void SessionClosing(AmqpSession^ session);
+ void WaitForCompletion();
+
+ property IntPtr XidHandle {
+ IntPtr get () { return (IntPtr) xidp; }
+ }
+
+#ifdef QPID_RECOVERY_TEST_HOOK
+ void ForceRecovery();
+ bool debugFailMode;
+#endif
+
+};
+
+}}} // namespace Apache::Qpid::Interop
+
diff --git a/qpid/wcf/test/Apache/Qpid/Test/Channel/Functional/FunctionalTests.csproj b/qpid/wcf/test/Apache/Qpid/Test/Channel/Functional/FunctionalTests.csproj index b7f4ed5d0a..d01cd99ff5 100644 --- a/qpid/wcf/test/Apache/Qpid/Test/Channel/Functional/FunctionalTests.csproj +++ b/qpid/wcf/test/Apache/Qpid/Test/Channel/Functional/FunctionalTests.csproj @@ -49,7 +49,9 @@ under the License. <WarningLevel>4</WarningLevel>
</PropertyGroup>
<ItemGroup>
- <Reference Include="nunit.framework, Version=2.5.2.9222, Culture=neutral, PublicKeyToken=96d09a1eb7f44a77, processorArchitecture=MSIL" />
+ <Reference Include="nunit.framework, Version=2.5.2.9222, Culture=neutral, PublicKeyToken=96d09a1eb7f44a77, processorArchitecture=MSIL">
+ <SpecificVersion>False</SpecificVersion>
+ </Reference>
<Reference Include="System" />
<Reference Include="System.Core">
<RequiredTargetFramework>3.5</RequiredTargetFramework>
diff --git a/qpid/wcf/test/Apache/Qpid/Test/Channel/WcfPerftest/RawBodyUtility.cs b/qpid/wcf/test/Apache/Qpid/Test/Channel/WcfPerftest/RawBodyUtility.cs new file mode 100644 index 0000000000..55a01c790c --- /dev/null +++ b/qpid/wcf/test/Apache/Qpid/Test/Channel/WcfPerftest/RawBodyUtility.cs @@ -0,0 +1,161 @@ +/*
+* 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.
+*/
+
+namespace Apache.Qpid.Test.Channel.WcfPerftest
+{
+ using System;
+ using System.Collections;
+ using System.IO;
+ using System.ServiceModel;
+ using System.ServiceModel.Channels;
+ using System.ServiceModel.Description;
+ using System.Threading;
+ using System.Text;
+ using System.Xml;
+ using Apache.Qpid.Channel;
+
+
+ /// <summary>
+ /// A sample interface for populating and extracting message body content.
+ /// Just enough methods to handle basic Interop text and raw byte messages.
+ /// </summary>
+
+
+ public interface IRawBodyUtility
+ {
+ Message CreateMessage(byte[] body, int offset, int len);
+ Message CreateMessage(byte[] body);
+ byte[] GetBytes(Message m, byte[] recyclableBuffer);
+
+ Message CreateMessage(string body);
+ string GetText(Message m);
+ }
+
+ // an implementation of IRawBodyUtility that expects a RawMessageEncoder based channel
+
+ public class RawEncoderUtility : IRawBodyUtility
+ {
+ public Message CreateMessage(byte[] body, int offset, int count)
+ {
+ return Message.CreateMessage(MessageVersion.None, "", new RawEncoderBodyWriter(body, offset, count));
+ }
+
+ public Message CreateMessage(byte[] body)
+ {
+ return CreateMessage(body, 0, body.Length);
+ }
+
+ public byte[] GetBytes(Message message, byte[] recyclableBuffer)
+ {
+ XmlDictionaryReader reader = message.GetReaderAtBodyContents();
+ int length;
+
+ while (!reader.HasValue)
+ {
+ reader.Read();
+ if (reader.EOF)
+ {
+ throw new InvalidDataException("empty XmlDictionaryReader");
+ }
+ }
+
+ if (reader.TryGetBase64ContentLength(out length))
+ {
+ byte[] bytes = null;
+ if (recyclableBuffer != null)
+ {
+ if (recyclableBuffer.Length == length)
+ {
+ // reuse
+ bytes = recyclableBuffer;
+ }
+ }
+
+ if (bytes == null)
+ {
+ bytes = new byte[length];
+ }
+
+ // this is the single copy mechanism from native to managed space with no intervening
+ // buffers. One could also write a method GetBytes(msg, myBuf, offset)...
+ reader.ReadContentAsBase64(bytes, 0, length);
+ reader.Close();
+ return bytes;
+ }
+ else
+ {
+ // uses whatever default buffering mechanism is used by the base XmlDictionaryReader class
+ return reader.ReadContentAsBase64();
+ }
+ }
+
+ public Message CreateMessage(string body)
+ {
+ return Message.CreateMessage(MessageVersion.None, "", new RawEncoderBodyWriter(body));
+ }
+
+ public string GetText(Message message)
+ {
+ byte[] rawBuffer = GetBytes(message, null);
+ return Encoding.UTF8.GetString(rawBuffer, 0, rawBuffer.Length);
+ }
+
+ internal class RawEncoderBodyWriter : BodyWriter
+ {
+ // works only with the Raw Encoder; the "body" is either a single string or byte[] segment
+ String bodyAsString;
+ byte[] bodyAsBytes;
+ int offset;
+ int count;
+
+ public RawEncoderBodyWriter(string body)
+ : base(false) // isBuffered
+ {
+ this.bodyAsString = body;
+ }
+
+ public RawEncoderBodyWriter(byte[] body, int offset, int count)
+ : base(false) // isBuffered
+ {
+ this.bodyAsBytes = body;
+ this.offset = offset;
+ this.count = count;
+ }
+
+ protected override void OnWriteBodyContents(System.Xml.XmlDictionaryWriter writer)
+ {
+ // TODO: RawMessageEncoder.StreamElementName should be public.
+ writer.WriteStartElement("Binary"); // the expected Raw encoder "<Binary>" virtual xml tag
+
+ if (bodyAsString != null)
+ {
+ byte[] buf = Encoding.UTF8.GetBytes(bodyAsString);
+ writer.WriteBase64(buf, 0, buf.Length);
+ }
+ else
+ {
+ writer.WriteBase64(this.bodyAsBytes, this.offset, this.count);
+ }
+
+ writer.WriteEndElement();
+ }
+ }
+ }
+
+}
diff --git a/qpid/wcf/test/Apache/Qpid/Test/Channel/WcfPerftest/WcfPerftest.cs b/qpid/wcf/test/Apache/Qpid/Test/Channel/WcfPerftest/WcfPerftest.cs new file mode 100644 index 0000000000..992d6e9bd2 --- /dev/null +++ b/qpid/wcf/test/Apache/Qpid/Test/Channel/WcfPerftest/WcfPerftest.cs @@ -0,0 +1,661 @@ +/*
+* 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.
+*/
+
+namespace Apache.Qpid.Test.Channel.WcfPerftest
+{
+ using System;
+ using System.Collections;
+ using System.Collections.Generic;
+ using System.ComponentModel;
+ using System.Configuration;
+ using System.Diagnostics;
+ using System.IO;
+ using System.ServiceModel;
+ using System.ServiceModel.Channels;
+ using System.ServiceModel.Description;
+ using System.Threading;
+ using System.Transactions;
+ using System.Text;
+ using System.Xml;
+ using Apache.Qpid.AmqpTypes;
+ using Apache.Qpid.Channel;
+
+ // this program implements a subset of the functionality in qpid\cpp\src\tests\perftest.cpp
+
+ // for a given broker, create reader and writer channels to queues/exchanges
+ // lazilly creates binding and channel factories
+
+ public class QueueChannelFactory
+ {
+ private static AmqpBinding brokerBinding;
+ private static IChannelFactory<IInputChannel> readerFactory;
+ private static IChannelFactory<IOutputChannel> writerFactory;
+ private static string brokerAddr = "127.0.0.1";
+ private static int brokerPort = 5672;
+
+ public static void SetBroker(string addr, int port)
+ {
+ brokerAddr = addr;
+ brokerPort = port;
+ }
+
+ private static void InitializeBinding()
+ {
+ AmqpBinaryBinding binding = new AmqpBinaryBinding();
+ binding.BrokerHost = brokerAddr;
+ binding.BrokerPort = brokerPort;
+ binding.TransferMode = TransferMode.Streamed;
+ binding.PrefetchLimit = 5000;
+ binding.Shared = true;
+ brokerBinding = binding;
+ }
+
+ public static IInputChannel CreateReaderChannel(string queueName)
+ {
+ lock (typeof(QueueChannelFactory))
+ {
+ if (brokerBinding == null)
+ {
+ InitializeBinding();
+ }
+
+ if (readerFactory == null)
+ {
+ readerFactory = brokerBinding.BuildChannelFactory<IInputChannel>();
+ readerFactory.Open();
+ }
+
+ IInputChannel channel = readerFactory.CreateChannel(new EndpointAddress(
+ new Uri("amqp:" + queueName)));
+ channel.Open();
+
+ return channel;
+ }
+ }
+
+ public static IOutputChannel CreateWriterChannel(string exchangeName, string routingKey)
+ {
+ lock (typeof(QueueChannelFactory))
+ {
+ if (brokerBinding == null)
+ {
+ InitializeBinding();
+ }
+
+ if (writerFactory == null)
+ {
+ writerFactory = brokerBinding.BuildChannelFactory<IOutputChannel>();
+ writerFactory.Open();
+ }
+
+ IOutputChannel channel = writerFactory.CreateChannel(new EndpointAddress(
+ "amqp:" + exchangeName +
+ "?routingkey=" + routingKey));
+ channel.Open();
+
+ return channel;
+ }
+ }
+ }
+
+ public enum ClientType
+ {
+ Publisher,
+ Subscriber,
+ InteropDemo
+ }
+
+ public class Options
+ {
+ public string broker;
+ public int port;
+ public UInt64 messageCount;
+ public int messageSize;
+ public ClientType type;
+ public string baseName;
+ public int subTxSize;
+ public int pubTxSize;
+ public bool durable;
+
+ public Options()
+ {
+ this.broker = "127.0.0.1";
+ this.port = 5672;
+ this.messageCount = 500000;
+ this.messageSize = 1024;
+ this.type = ClientType.InteropDemo; // default: once as pub and once as sub
+ this.baseName = "perftest";
+ this.pubTxSize = 0;
+ this.subTxSize = 0;
+ this.durable = false;
+ }
+
+ public void Parse(string[] args)
+ {
+ int argCount = args.Length;
+ int current = 0;
+ bool typeSelected = false;
+
+ while (current < argCount)
+ {
+ string arg = args[current];
+ if (arg == "--publish")
+ {
+ if (typeSelected)
+ throw new ArgumentException("too many roles");
+
+ this.type = ClientType.Publisher;
+ typeSelected = true;
+ }
+ else if (arg == "--subscribe")
+ {
+ if (typeSelected)
+ throw new ArgumentException("too many roles");
+
+ this.type = ClientType.Subscriber;
+ typeSelected = true;
+ }
+ else if (arg == "--size")
+ {
+ arg = args[++current];
+ int i = int.Parse(arg);
+ if (i > 0)
+ {
+ this.messageSize = i;
+ }
+ }
+ else if (arg == "--count")
+ {
+ arg = args[++current];
+ UInt64 i = UInt64.Parse(arg);
+ if (i > 0)
+ {
+ this.messageCount = i;
+ }
+ }
+ else if (arg == "--broker")
+ {
+ this.broker = args[++current];
+ }
+ else if (arg == "--port")
+ {
+ arg = args[++current];
+ int i = int.Parse(arg);
+ if (i > 0)
+ {
+ this.port = i;
+ }
+ }
+ else if (arg == "--base-name")
+ {
+ this.baseName = args[++current];
+ }
+
+ else if (arg == "--tx")
+ {
+ arg = args[++current];
+ int i = int.Parse(arg);
+ if (i > 0)
+ {
+ this.subTxSize = i;
+ this.pubTxSize = i;
+ }
+ }
+
+ else if (arg == "--durable")
+ {
+ arg = args[++current];
+ if (arg.Equals("yes"))
+ {
+ this.durable = true;
+ }
+ }
+
+ current++;
+ }
+ }
+ }
+
+
+ public class Client
+ {
+ protected Options opts;
+
+ public static void Expect(string actual, string expect)
+ {
+ if (expect != actual)
+ {
+ throw new Exception("Expecting " + expect + " but received " + actual);
+ }
+ }
+
+ public static void Close(IChannel channel)
+ {
+ if (channel == null)
+ {
+ return;
+ }
+
+ try
+ {
+ channel.Close();
+ }
+ catch (Exception e)
+ {
+ Console.WriteLine("channel close exception {0}", e);
+ }
+ }
+
+ public string Fqn(string name)
+ {
+ return opts.baseName + '_' + name;
+ }
+ }
+
+
+ public class WcfPerftest
+ {
+ static void WarmUpTransactionSubsystem(Options opts)
+ {
+ // see if any use of transactions is expected
+ if ((opts.type == ClientType.Publisher) && (opts.pubTxSize == 0))
+ return;
+
+ if ((opts.type == ClientType.Subscriber) && (opts.subTxSize == 0))
+ return;
+
+ if (opts.type == ClientType.InteropDemo)
+ {
+ if ((opts.subTxSize == 0) && (opts.pubTxSize == 0))
+ return;
+ }
+
+ Console.WriteLine("Initializing transactions");
+ IRawBodyUtility bodyUtil = new RawEncoderUtility();
+
+ // send a transacted message to nowhere to force the initial registration with MSDTC
+ IOutputChannel channel = QueueChannelFactory.CreateWriterChannel("", Guid.NewGuid().ToString());
+ Message msg = bodyUtil.CreateMessage("sacrificial transacted message from WcfPerftest");
+ using (TransactionScope ts = new TransactionScope())
+ {
+ channel.Send(msg);
+ // abort/rollback
+ ts.Dispose();
+ }
+ channel.Close();
+ Console.WriteLine("transaction resource manager ready");
+ }
+
+ static void InteropDemo(Options opts)
+ {
+ string perftest_cpp_exe = "perftest.exe";
+ string commonArgs = String.Format(" --count {0} --size {1}", opts.messageCount, opts.messageSize);
+
+ if (opts.durable)
+ {
+ commonArgs += " --durable yes";
+ }
+
+ Console.WriteLine("===== WCF Subscriber and C++ Publisher =====");
+
+ Process setup = new Process();
+ setup.StartInfo.FileName = perftest_cpp_exe;
+ setup.StartInfo.UseShellExecute = false;
+ setup.StartInfo.Arguments = "--setup" + commonArgs;
+ try
+ {
+ setup.Start();
+ }
+ catch (Win32Exception win32e)
+ {
+ Console.WriteLine("Cannot execute {0}: PATH not set?", perftest_cpp_exe);
+ Console.WriteLine(" Error: {0}", win32e.Message);
+ return;
+ }
+ setup.WaitForExit();
+
+ Process control = new Process();
+ control.StartInfo.FileName = perftest_cpp_exe;
+ control.StartInfo.UseShellExecute = false;
+ control.StartInfo.Arguments = "--control" + commonArgs;
+ control.Start();
+
+ Process publish = new Process();
+ publish.StartInfo.FileName = perftest_cpp_exe;
+ publish.StartInfo.UseShellExecute = false;
+ publish.StartInfo.Arguments = "--publish" + commonArgs;
+ publish.Start();
+
+ SubscribeThread subscribeWcf = new SubscribeThread(opts.baseName + "0", opts);
+ Thread subThread = new Thread(subscribeWcf.Run);
+ subThread.Start();
+
+ subThread.Join();
+ publish.WaitForExit();
+ control.WaitForExit();
+
+ Console.WriteLine();
+ Console.WriteLine("===== WCF Publisher and C++ Subscriber =====");
+
+ setup = new Process();
+ setup.StartInfo.FileName = perftest_cpp_exe;
+ setup.StartInfo.UseShellExecute = false;
+ setup.StartInfo.Arguments = "--setup" + commonArgs;
+ setup.Start();
+ setup.WaitForExit();
+
+ control = new Process();
+ control.StartInfo.FileName = perftest_cpp_exe;
+ control.StartInfo.UseShellExecute = false;
+ control.StartInfo.Arguments = "--control" + commonArgs;
+ control.Start();
+
+ PublishThread pub = new PublishThread(opts.baseName + "0", "", opts);
+ Thread pubThread = new Thread(pub.Run);
+ pubThread.Start();
+
+ Process subscribeCpp = new Process();
+ subscribeCpp.StartInfo.FileName = perftest_cpp_exe;
+ subscribeCpp.StartInfo.UseShellExecute = false;
+ subscribeCpp.StartInfo.Arguments = "--subscribe" + commonArgs;
+ subscribeCpp.Start();
+
+ subscribeCpp.WaitForExit();
+ pubThread.Join();
+ control.WaitForExit();
+ }
+
+ static void Main(string[] mainArgs)
+ {
+ Options opts = new Options();
+ opts.Parse(mainArgs);
+ QueueChannelFactory.SetBroker(opts.broker, opts.port);
+
+ WarmUpTransactionSubsystem(opts);
+
+ if (opts.type == ClientType.Publisher)
+ {
+ PublishThread pub = new PublishThread(opts.baseName + "0", "", opts);
+ Thread pubThread = new Thread(pub.Run);
+ pubThread.Start();
+ pubThread.Join();
+ }
+ else if (opts.type == ClientType.Subscriber)
+ {
+ SubscribeThread sub = new SubscribeThread(opts.baseName + "0", opts);
+ Thread subThread = new Thread(sub.Run);
+ subThread.Start();
+ subThread.Join();
+ }
+ else
+ {
+ InteropDemo(opts);
+ }
+
+ if (System.Diagnostics.Debugger.IsAttached)
+ {
+ Console.WriteLine("Hit return to continue...");
+ Console.ReadLine();
+ }
+ }
+ }
+
+ public class PublishThread : Client
+ {
+ string destination; // exchange/queue
+ string routingKey;
+ int msgSize;
+ UInt64 msgCount;
+ IOutputChannel publishQueue;
+
+ public PublishThread(string key, string q, Options opts)
+ {
+ this.routingKey = key;
+ this.destination = q;
+ this.msgSize = opts.messageSize;
+ this.msgCount = opts.messageCount;
+ this.opts = opts;
+ }
+
+ static void StampSequenceNo(byte[] data, UInt64 n)
+ {
+ int wordLen = IntPtr.Size; // mimic size_t in C++
+
+ if (data.Length < wordLen)
+ throw new ArgumentException("message size");
+ for (int i = 0; i < wordLen; i++)
+ {
+ data[i] = (byte) (n & 0xff);
+ n >>= 8;
+ }
+ }
+
+ public void Run()
+ {
+ IRawBodyUtility bodyUtil = new RawEncoderUtility();
+
+ IInputChannel startQueue = null;
+ IOutputChannel doneQueue = null;
+ UInt64 batchSize = (UInt64)opts.pubTxSize;
+ bool txPending = false;
+ AmqpProperties amqpProperties = null;
+
+ if (opts.durable)
+ {
+ amqpProperties = new AmqpProperties();
+ amqpProperties.Durable = true;
+ }
+
+ try
+ {
+ publishQueue = QueueChannelFactory.CreateWriterChannel(this.destination, this.routingKey);
+ doneQueue = QueueChannelFactory.CreateWriterChannel("", this.Fqn("pub_done"));
+ startQueue = QueueChannelFactory.CreateReaderChannel(this.Fqn("pub_start"));
+
+ // wait for our start signal
+ Message msg;
+ msg = startQueue.Receive(TimeSpan.MaxValue);
+ Expect(bodyUtil.GetText(msg), "start");
+ msg.Close();
+
+ Stopwatch stopwatch = new Stopwatch();
+ AsyncCallback sendCallback = new AsyncCallback(this.AsyncSendCB);
+
+ byte[] data = new byte[this.msgSize];
+ IAsyncResult sendResult = null;
+
+ Console.WriteLine("sending {0}", this.msgCount);
+ stopwatch.Start();
+
+ if (batchSize > 0)
+ {
+ Transaction.Current = new CommittableTransaction();
+ }
+
+ for (UInt64 i = 0; i < this.msgCount; i++)
+ {
+ StampSequenceNo(data, i);
+ msg = bodyUtil.CreateMessage(data);
+ if (amqpProperties != null)
+ {
+ msg.Properties.Add("AmqpProperties", amqpProperties);
+ }
+
+ sendResult = publishQueue.BeginSend(msg, TimeSpan.MaxValue, sendCallback, msg);
+
+ if (batchSize > 0)
+ {
+ txPending = true;
+ if (((i + 1) % batchSize) == 0)
+ {
+ ((CommittableTransaction)Transaction.Current).Commit();
+ txPending = false;
+ Transaction.Current = new CommittableTransaction();
+ }
+ }
+ }
+
+ if (txPending)
+ {
+ ((CommittableTransaction)Transaction.Current).Commit();
+ }
+
+ Transaction.Current = null;
+
+ sendResult.AsyncWaitHandle.WaitOne();
+ stopwatch.Stop();
+
+ double mps = (msgCount / stopwatch.Elapsed.TotalSeconds);
+
+ msg = bodyUtil.CreateMessage(String.Format("{0:0.##}", mps));
+ doneQueue.Send(msg, TimeSpan.MaxValue);
+ msg.Close();
+ }
+ finally
+ {
+ Close((IChannel)doneQueue);
+ Close((IChannel)publishQueue);
+ Close(startQueue);
+ }
+ }
+
+ void AsyncSendCB(IAsyncResult result)
+ {
+ publishQueue.EndSend(result);
+ ((Message)result.AsyncState).Close();
+ }
+ }
+
+ public class SubscribeThread : Client
+ {
+ string queue;
+ int msgSize;
+ UInt64 msgCount;
+ IInputChannel subscribeQueue;
+
+ public SubscribeThread(string q, Options opts)
+ {
+ this.queue = q;
+ this.msgSize = opts.messageSize;
+ this.msgCount = opts.messageCount;
+ this.opts = opts;
+ }
+
+ static UInt64 GetSequenceNumber(byte[] data)
+ {
+ int wordLen = IntPtr.Size; // mimic size_t in C++
+
+ if (data.Length < wordLen)
+ throw new ArgumentException("message size");
+ UInt64 n = 0;
+ for (int i = (wordLen - 1); i >= 0; i--)
+ {
+ n = (256 * n) + data[i];
+ }
+ return n;
+ }
+
+ public void Run()
+ {
+ IRawBodyUtility bodyUtil = new RawEncoderUtility();
+
+ IOutputChannel readyQueue = null;
+ IOutputChannel doneQueue = null;
+ UInt64 batchSize = (UInt64)opts.subTxSize;
+ bool txPending = false;
+ byte[] data = null;
+
+ try
+ {
+ this.subscribeQueue = QueueChannelFactory.CreateReaderChannel(this.queue);
+ readyQueue = QueueChannelFactory.CreateWriterChannel("", this.Fqn("sub_ready"));
+ doneQueue = QueueChannelFactory.CreateWriterChannel("", this.Fqn("sub_done"));
+
+ Message msg = bodyUtil.CreateMessage("ready");
+ readyQueue.Send(msg, TimeSpan.MaxValue);
+ msg.Close();
+
+
+ Stopwatch stopwatch = new Stopwatch();
+ stopwatch.Start();
+
+ Console.WriteLine("receiving {0}", this.msgCount);
+ UInt64 expect = 0;
+
+ if (batchSize > 0)
+ {
+ Transaction.Current = new CommittableTransaction();
+ }
+
+ for (UInt64 i = 0; i < this.msgCount; i++)
+ {
+ msg = subscribeQueue.Receive(TimeSpan.MaxValue);
+
+ data = bodyUtil.GetBytes(msg, data);
+ msg.Close();
+ if (data.Length != this.msgSize)
+ {
+ throw new Exception("subscribe message size mismatch");
+ }
+
+ UInt64 n = GetSequenceNumber(data);
+ if (n != expect)
+ {
+ throw new Exception(String.Format("message sequence error. expected {0} got {1}", expect, n));
+ }
+ expect = n + 1;
+
+ if (batchSize > 0)
+ {
+ txPending = true;
+ if (((i + 1) % batchSize) == 0)
+ {
+ ((CommittableTransaction)Transaction.Current).Commit();
+ txPending = false;
+ Transaction.Current = new CommittableTransaction();
+ }
+ }
+ }
+
+ if (txPending)
+ {
+ ((CommittableTransaction)Transaction.Current).Commit();
+ }
+
+ Transaction.Current = null;
+
+ stopwatch.Stop();
+
+ double mps = (msgCount / stopwatch.Elapsed.TotalSeconds);
+
+ msg = bodyUtil.CreateMessage(String.Format("{0:0.##}", mps));
+ doneQueue.Send(msg, TimeSpan.MaxValue);
+ msg.Close();
+
+ subscribeQueue.Close();
+ }
+ finally
+ {
+ Close((IChannel)doneQueue);
+ Close((IChannel)this.subscribeQueue);
+ Close(readyQueue);
+ }
+ }
+ }
+}
diff --git a/qpid/wcf/test/Apache/Qpid/Test/Channel/WcfPerftest/WcfPerftest.csproj b/qpid/wcf/test/Apache/Qpid/Test/Channel/WcfPerftest/WcfPerftest.csproj new file mode 100644 index 0000000000..44ef998a24 --- /dev/null +++ b/qpid/wcf/test/Apache/Qpid/Test/Channel/WcfPerftest/WcfPerftest.csproj @@ -0,0 +1,83 @@ +<?xml version="1.0" encoding="utf-8"?>
+<!--
+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.
+-->
+<Project ToolsVersion="3.5" DefaultTargets="Build" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
+ <PropertyGroup>
+ <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
+ <Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform>
+ <ProductVersion>9.0.21022</ProductVersion>
+ <SchemaVersion>2.0</SchemaVersion>
+ <ProjectGuid>{D0F8FDE4-7AC6-4CFF-986A-50D06F7FD733}</ProjectGuid>
+ <OutputType>Exe</OutputType>
+ <RootNamespace>Apache.Qpid.Test.Channel.WcfPerftest</RootNamespace>
+ <AssemblyName>WcfPerftest</AssemblyName>
+ <TargetFrameworkVersion>v3.5</TargetFrameworkVersion>
+ <FileAlignment>512</FileAlignment>
+ </PropertyGroup>
+ <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
+ <DebugSymbols>true</DebugSymbols>
+ <DebugType>full</DebugType>
+ <Optimize>false</Optimize>
+ <OutputPath>bin\Debug\</OutputPath>
+ <DefineConstants>DEBUG;TRACE</DefineConstants>
+ <ErrorReport>prompt</ErrorReport>
+ <WarningLevel>4</WarningLevel>
+ </PropertyGroup>
+ <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
+ <DebugType>pdbonly</DebugType>
+ <Optimize>true</Optimize>
+ <OutputPath>bin\Release\</OutputPath>
+ <DefineConstants>TRACE</DefineConstants>
+ <ErrorReport>prompt</ErrorReport>
+ <WarningLevel>4</WarningLevel>
+ </PropertyGroup>
+ <ItemGroup>
+ <Reference Include="System" />
+ <Reference Include="System.Runtime.Serialization">
+ <RequiredTargetFramework>3.0</RequiredTargetFramework>
+ </Reference>
+ <Reference Include="System.ServiceModel">
+ <RequiredTargetFramework>3.0</RequiredTargetFramework>
+ </Reference>
+ <Reference Include="System.Transactions" />
+ <Reference Include="System.XML" />
+ </ItemGroup>
+ <ItemGroup>
+ <Compile Include="WcfPerftest.cs" />
+ <Compile Include="RawBodyUtility.cs" />
+ </ItemGroup>
+ <ItemGroup>
+ <ProjectReference Include="..\..\..\..\..\..\src\Apache\Qpid\Channel\Channel.csproj">
+ <Project>{8AABAB30-7D1E-4539-B7D1-05450262BAD2}</Project>
+ <Name>Channel</Name>
+ </ProjectReference>
+ <ProjectReference Include="..\..\..\..\..\..\src\Apache\Qpid\Interop\Interop.vcproj">
+ <Project>{C9B6AC75-6332-47A4-B82B-0C20E0AF2D34}</Project>
+ <Name>Interop</Name>
+ </ProjectReference>
+ </ItemGroup>
+ <Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
+ <!-- To modify your build process, add your task inside one of the targets below and uncomment it.
+ Other similar extension points exist, see Microsoft.Common.targets.
+ <Target Name="BeforeBuild">
+ </Target>
+ <Target Name="AfterBuild">
+ </Target>
+ -->
+</Project>
|
