Merged from trunk up to r796653

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/java-broker-0-10@821735 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 83 insertions, 39 deletions

diff --git a/qpid/cpp/src/qpid/broker/SemanticState.cpp b/qpid/cpp/src/qpid/broker/SemanticState.cpp
index d77bd9b7ca..37a9e9b4af 100644
--- a/qpid/cpp/src/qpid/broker/SemanticState.cpp
+++ b/qpid/cpp/src/qpid/broker/SemanticState.cpp
@@ -228,6 +228,8 @@ void SemanticState::record(const DeliveryRecord& delivery)
     unacked.push_back(delivery);
 }
 
+const std::string QPID_SYNC_FREQUENCY("qpid.sync_frequency");
+
 SemanticState::ConsumerImpl::ConsumerImpl(SemanticState* _parent, 
                                           const string& _name, 
                                           Queue::shared_ptr _queue, 
@@ -255,7 +257,7 @@ SemanticState::ConsumerImpl::ConsumerImpl(SemanticState* _parent,
     msgCredit(0), 
     byteCredit(0),
     notifyEnabled(true),
-    syncFrequency(_arguments.getAsInt("qpid.sync_frequency")),
+    syncFrequency(_arguments.getAsInt(QPID_SYNC_FREQUENCY)),
     deliveryCount(0)
 {}
 
diff --git a/qpid/cpp/src/qpid/broker/SessionAdapter.cpp b/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
index e52274597e..6f1d42249d 100644
--- a/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
+++ b/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
@@ -57,6 +57,8 @@ SessionAdapter::SessionAdapter(SemanticState& s) :
     dtxImpl(s)
 {}
 
+static const std::string TRUE("true");
+static const std::string FALSE("false");
 
 void SessionAdapter::ExchangeHandlerImpl::declare(const string& exchange, const string& type, 
                                                   const string& alternateExchange, 
@@ -67,8 +69,8 @@ void SessionAdapter::ExchangeHandlerImpl::declare(const string& exchange, const
         std::map<acl::Property, std::string> params;
         params.insert(make_pair(acl::PROP_TYPE, type));
         params.insert(make_pair(acl::PROP_ALTERNATE, alternateExchange));
-        params.insert(make_pair(acl::PROP_PASSIVE, std::string(passive ? "true" : "false") ));
-        params.insert(make_pair(acl::PROP_DURABLE, std::string(durable ? "true" : "false")));
+        params.insert(make_pair(acl::PROP_PASSIVE, std::string(passive ? TRUE : FALSE) ));
+        params.insert(make_pair(acl::PROP_DURABLE, std::string(durable ? TRUE : FALSE)));
         if (!acl->authorise(getConnection().getUserId(),acl::ACT_CREATE,acl::OBJ_EXCHANGE,exchange,&params) )
             throw NotAllowedException(QPID_MSG("ACL denied exhange declare request from " << getConnection().getUserId()));
     }
@@ -325,10 +327,10 @@ void SessionAdapter::QueueHandlerImpl::declare(const string& name, const string&
     if (acl) {
         std::map<acl::Property, std::string> params;
         params.insert(make_pair(acl::PROP_ALTERNATE, alternateExchange));
-        params.insert(make_pair(acl::PROP_PASSIVE, std::string(passive ? "true" : "false") ));
-        params.insert(make_pair(acl::PROP_DURABLE, std::string(durable ? "true" : "false")));
-        params.insert(make_pair(acl::PROP_EXCLUSIVE, std::string(exclusive ? "true" : "false")));
-        params.insert(make_pair(acl::PROP_AUTODELETE, std::string(autoDelete ? "true" : "false")));
+        params.insert(make_pair(acl::PROP_PASSIVE, std::string(passive ? TRUE : FALSE) ));
+        params.insert(make_pair(acl::PROP_DURABLE, std::string(durable ? TRUE : FALSE)));
+        params.insert(make_pair(acl::PROP_EXCLUSIVE, std::string(exclusive ? TRUE : FALSE)));
+        params.insert(make_pair(acl::PROP_AUTODELETE, std::string(autoDelete ? TRUE : FALSE)));
         if (!acl->authorise(getConnection().getUserId(),acl::ACT_CREATE,acl::OBJ_QUEUE,name,&params) )
             throw NotAllowedException(QPID_MSG("ACL denied queue create request from " << getConnection().getUserId()));
     }
diff --git a/qpid/cpp/src/qpid/cluster/Cpg.cpp b/qpid/cpp/src/qpid/cluster/Cpg.cpp
index f5e79012d9..0878ae94b9 100644
--- a/qpid/cpp/src/qpid/cluster/Cpg.cpp
+++ b/qpid/cpp/src/qpid/cluster/Cpg.cpp
@@ -29,6 +29,11 @@
 
 #include <unistd.h>
 
+// This is a macro instead of a function because we don't want to
+// evaluate the MSG argument unless there is an error.
+#define CPG_CHECK(RESULT, MSG) \
+    if ((RESULT) != CPG_OK) throw Exception(errorStr((RESULT), (MSG)))
+
 namespace qpid {
 namespace cluster {
 
@@ -36,7 +41,7 @@ using namespace std;
 
 Cpg* Cpg::cpgFromHandle(cpg_handle_t handle) {
     void* cpg=0;
-    check(cpg_context_get(handle, &cpg), "Cannot get CPG instance.");
+    CPG_CHECK(cpg_context_get(handle, &cpg), "Cannot get CPG instance.");
     if (!cpg) throw Exception("Cannot get CPG instance.");
     return reinterpret_cast<Cpg*>(cpg);
 }
@@ -66,7 +71,7 @@ void Cpg::globalConfigChange(
 
 int Cpg::getFd() {
     int fd;
-    check(cpg_fd_get(handle, &fd), "Cannot get CPG file descriptor");
+    CPG_CHECK(cpg_fd_get(handle, &fd), "Cannot get CPG file descriptor");
     return fd;
 }
 
@@ -84,8 +89,8 @@ Cpg::Cpg(Handler& h) : IOHandle(new sys::IOHandlePrivate), handler(h), isShutdow
         sys::sleep(5);
         err = cpg_initialize(&handle, &callbacks);
     }
-    check(err, "Failed to initialize CPG.");
-    check(cpg_context_set(handle, this), "Cannot set CPG context");
+    CPG_CHECK(err, "Failed to initialize CPG.");
+    CPG_CHECK(cpg_context_set(handle, this), "Cannot set CPG context");
     // Note: CPG is currently unix-specific. If CPG is ported to
     // windows then this needs to be refactored into
     // qpid::sys::<platform>
@@ -102,11 +107,11 @@ Cpg::~Cpg() {
 
 void Cpg::join(const std::string& name) {
     group = name;
-    check(cpg_join(handle, &group), cantJoinMsg(group));
+    CPG_CHECK(cpg_join(handle, &group), cantJoinMsg(group));
 }
     
 void Cpg::leave() {
-    check(cpg_leave(handle, &group), cantLeaveMsg(group));
+    CPG_CHECK(cpg_leave(handle, &group), cantLeaveMsg(group));
 }
 
 
@@ -115,14 +120,14 @@ void Cpg::leave() {
 bool Cpg::mcast(const iovec* iov, int iovLen) {
     // Check for flow control
     cpg_flow_control_state_t flowState;
-    check(cpg_flow_control_state_get(handle, &flowState), "Cannot get CPG flow control status.");
+    CPG_CHECK(cpg_flow_control_state_get(handle, &flowState), "Cannot get CPG flow control status.");
     if (flowState == CPG_FLOW_CONTROL_ENABLED)
         return false;
 
     cpg_error_t result;
     do {
         result = cpg_mcast_joined(handle, CPG_TYPE_AGREED, const_cast<iovec*>(iov), iovLen);
-        if (result != CPG_ERR_TRY_AGAIN) check(result, cantMcastMsg(group));
+        if (result != CPG_ERR_TRY_AGAIN) CPG_CHECK(result, cantMcastMsg(group));
     } while(result == CPG_ERR_TRY_AGAIN);
     return true;
 }
@@ -131,20 +136,20 @@ void Cpg::shutdown() {
     if (!isShutdown) {
         QPID_LOG(debug,"Shutting down CPG");
         isShutdown=true;
-        check(cpg_finalize(handle), "Error in shutdown of CPG");
+        CPG_CHECK(cpg_finalize(handle), "Error in shutdown of CPG");
     }
 }
 
 void Cpg::dispatchOne() {
-    check(cpg_dispatch(handle,CPG_DISPATCH_ONE), "Error in CPG dispatch");
+    CPG_CHECK(cpg_dispatch(handle,CPG_DISPATCH_ONE), "Error in CPG dispatch");
 }
 
 void Cpg::dispatchAll() {
-    check(cpg_dispatch(handle,CPG_DISPATCH_ALL), "Error in CPG dispatch");
+    CPG_CHECK(cpg_dispatch(handle,CPG_DISPATCH_ALL), "Error in CPG dispatch");
 }
 
 void Cpg::dispatchBlocking() {
-    check(cpg_dispatch(handle,CPG_DISPATCH_BLOCKING), "Error in CPG dispatch");
+    CPG_CHECK(cpg_dispatch(handle,CPG_DISPATCH_BLOCKING), "Error in CPG dispatch");
 }
 
 string Cpg::errorStr(cpg_error_t err, const std::string& msg) {
@@ -184,7 +189,7 @@ std::string Cpg::cantMcastMsg(const Name& group) {
 
 MemberId Cpg::self() const {
     unsigned int nodeid;
-    check(cpg_local_get(handle, &nodeid), "Cannot get local CPG identity");
+    CPG_CHECK(cpg_local_get(handle, &nodeid), "Cannot get local CPG identity");
     return MemberId(nodeid, getpid());
 }
 
diff --git a/qpid/cpp/src/qpid/cluster/Cpg.h b/qpid/cpp/src/qpid/cluster/Cpg.h
index ac27a09ae6..624721b560 100644
--- a/qpid/cpp/src/qpid/cluster/Cpg.h
+++ b/qpid/cpp/src/qpid/cluster/Cpg.h
@@ -120,10 +120,6 @@ class Cpg : public sys::IOHandle {
     static std::string cantLeaveMsg(const Name&);
     static std::string cantMcastMsg(const Name&);
 
-    static void check(cpg_error_t result, const std::string& msg) {
-        if (result != CPG_OK) throw Exception(errorStr(result, msg));
-    }
-
     static Cpg* cpgFromHandle(cpg_handle_t);
 
     static void globalDeliver(
diff --git a/qpid/cpp/src/qpid/sys/windows/PipeHandle.cpp b/qpid/cpp/src/qpid/sys/windows/PipeHandle.cpp
index e2cbff3908..062458ae5f 100755
--- a/qpid/cpp/src/qpid/sys/windows/PipeHandle.cpp
+++ b/qpid/cpp/src/qpid/sys/windows/PipeHandle.cpp
@@ -19,9 +19,6 @@
 
 #include "qpid/sys/PipeHandle.h"
 #include "qpid/sys/windows/check.h"
-#include <io.h>
-#include <fcntl.h>
-#include <errno.h>
 #include <winsock2.h>
 
 namespace qpid {
@@ -29,14 +26,53 @@ namespace sys {
 
 PipeHandle::PipeHandle(bool nonBlocking) {
 
-    int pair[2];
-    pair[0] = pair[1] = -1;
+    SOCKET listener, pair[2];
+    struct sockaddr_in addr;
+    int err;
+    int addrlen = sizeof(addr);
+    pair[0] = pair[1] = INVALID_SOCKET;
+    if ((listener = socket(AF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET)
+        throw QPID_WINDOWS_ERROR(WSAGetLastError());
 
-    if (_pipe(pair, 128, O_BINARY) == -1)
-        throw qpid::Exception(QPID_MSG("Creation of pipe failed"));
+    memset(&addr, 0, sizeof(addr));
+    addr.sin_family = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    addr.sin_port = 0;
 
-    writeFd = pair[0];
-    readFd = pair[1]; 
+    err = bind(listener, (const struct sockaddr*) &addr, sizeof(addr));
+    if (err == SOCKET_ERROR) {
+        err = WSAGetLastError();
+        closesocket(listener);
+        throw QPID_WINDOWS_ERROR(err);
+    }
+
+    err = getsockname(listener, (struct sockaddr*) &addr, &addrlen);
+    if (err == SOCKET_ERROR) {
+        err = WSAGetLastError();
+        closesocket(listener);
+        throw QPID_WINDOWS_ERROR(err);
+    }
+
+    try {
+        if (listen(listener, 1) == SOCKET_ERROR)
+            throw QPID_WINDOWS_ERROR(WSAGetLastError());
+        if ((pair[0] = socket(AF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET)
+            throw QPID_WINDOWS_ERROR(WSAGetLastError());
+        if (connect(pair[0], (const struct sockaddr*)&addr, sizeof(addr)) == SOCKET_ERROR)
+            throw QPID_WINDOWS_ERROR(WSAGetLastError());
+        if ((pair[1] = accept(listener, NULL, NULL)) == INVALID_SOCKET)
+            throw QPID_WINDOWS_ERROR(WSAGetLastError());
+
+        closesocket(listener);
+        writeFd = pair[0];
+        readFd = pair[1]; 
+    }
+    catch (...) {
+        closesocket(listener);
+        if (pair[0] != INVALID_SOCKET)
+            closesocket(pair[0]);
+        throw;
+    }
 
     // Set the socket to non-blocking
     if (nonBlocking) {
@@ -46,16 +82,16 @@ PipeHandle::PipeHandle(bool nonBlocking) {
 }
 
 PipeHandle::~PipeHandle() {
-    close(readFd);
-    close(writeFd);
+    closesocket(readFd);
+    closesocket(writeFd);
 }
 
 int PipeHandle::read(void* buf, size_t bufSize) {
-    return ::read(readFd, buf, bufSize);
+    return ::recv(readFd, (char *)buf, bufSize, 0);
 }
 
 int PipeHandle::write(const void* buf, size_t bufSize) {
-    return ::write(writeFd, buf, bufSize);
+    return ::send(writeFd, (const char *)buf, bufSize, 0);
 }
 
 int PipeHandle::getReadHandle() {
diff --git a/qpid/cpp/src/qpid/sys/windows/SystemInfo.cpp b/qpid/cpp/src/qpid/sys/windows/SystemInfo.cpp
index 3e2fcb1517..ea53fc199c 100755
--- a/qpid/cpp/src/qpid/sys/windows/SystemInfo.cpp
+++ b/qpid/cpp/src/qpid/sys/windows/SystemInfo.cpp
@@ -181,18 +181,21 @@ uint32_t SystemInfo::getParentProcessId()
 
 std::string SystemInfo::getProcessName()
 {
+    std::string name;
+
     // Only want info for the current process, so ask for something specific.
     // The module info won't be used here but it keeps the snapshot limited to
     // the current process so a search through all processes is not needed.
     HANDLE snap = CreateToolhelp32Snapshot(TH32CS_SNAPMODULE, 0);
     if (snap == INVALID_HANDLE_VALUE)
-        return 0;
+        return name;
     PROCESSENTRY32 entry;
     entry.dwSize = sizeof(entry);
     if (!Process32First(snap, &entry))
         entry.szExeFile[0] = '\0';
     CloseHandle(snap);
-    return std::string(entry.szExeFile);
+    name = entry.szExeFile;
+    return name;
 }
 
 }} // namespace qpid::sys