You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kg...@apache.org on 2012/11/05 17:52:14 UTC
svn commit: r1405875 - in /qpid/proton/branches/kgiusti-msvc2010: ./ bin/
examples/ proton-c/ proton-c/bindings/php/ proton-c/bindings/python/
proton-c/bindings/ruby/ proton-c/examples/ proton-c/include/proton/
proton-c/src/ proton-c/src/dispatcher/ pr...
Author: kgiusti
Date: Mon Nov 5 16:52:12 2012
New Revision: 1405875
URL: http://svn.apache.org/viewvc?rev=1405875&view=rev
Log:
NO-JIRA: merge in latest from trunk
Added:
qpid/proton/branches/kgiusti-msvc2010/proton-c/examples/
- copied from r1405867, qpid/proton/trunk/proton-c/examples/
qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/ssl_db/bad-server-certificate.pem
- copied unchanged from r1405867, qpid/proton/trunk/tests/proton_tests/ssl_db/bad-server-certificate.pem
qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/ssl_db/bad-server-private-key.pem
- copied unchanged from r1405867, qpid/proton/trunk/tests/proton_tests/ssl_db/bad-server-private-key.pem
Removed:
qpid/proton/branches/kgiusti-msvc2010/examples/
Modified:
qpid/proton/branches/kgiusti-msvc2010/ (props changed)
qpid/proton/branches/kgiusti-msvc2010/bin/release.sh
qpid/proton/branches/kgiusti-msvc2010/proton-c/CMakeLists.txt
qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/php/CMakeLists.txt
qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/python/CMakeLists.txt
qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/python/proton.py
qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/ruby/CMakeLists.txt
qpid/proton/branches/kgiusti-msvc2010/proton-c/include/proton/cproton.i
qpid/proton/branches/kgiusti-msvc2010/proton-c/include/proton/driver.h
qpid/proton/branches/kgiusti-msvc2010/proton-c/include/proton/messenger.h
qpid/proton/branches/kgiusti-msvc2010/proton-c/soversion.cmake
qpid/proton/branches/kgiusti-msvc2010/proton-c/src/dispatcher/dispatcher.c
qpid/proton/branches/kgiusti-msvc2010/proton-c/src/driver.c
qpid/proton/branches/kgiusti-msvc2010/proton-c/src/engine/engine-internal.h
qpid/proton/branches/kgiusti-msvc2010/proton-c/src/engine/engine.c
qpid/proton/branches/kgiusti-msvc2010/proton-c/src/messenger.c
qpid/proton/branches/kgiusti-msvc2010/proton-c/src/ssl/openssl.c
qpid/proton/branches/kgiusti-msvc2010/proton-c/src/ssl/ssl-internal.h
qpid/proton/branches/kgiusti-msvc2010/proton-c/src/ssl/ssl_stub.c
qpid/proton/branches/kgiusti-msvc2010/proton-c/src/util.h
qpid/proton/branches/kgiusti-msvc2010/proton-j/src/main/java/org/apache/qpid/proton/engine/Link.java
qpid/proton/branches/kgiusti-msvc2010/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/LinkImpl.java
qpid/proton/branches/kgiusti-msvc2010/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java
qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/messenger.py
qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/sasl.py
qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/ssl.py
qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/ssl_db/README.txt
Propchange: qpid/proton/branches/kgiusti-msvc2010/
------------------------------------------------------------------------------
Merged /qpid/proton/trunk:r1404579-1405867
Modified: qpid/proton/branches/kgiusti-msvc2010/bin/release.sh
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/bin/release.sh?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/bin/release.sh (original)
+++ qpid/proton/branches/kgiusti-msvc2010/bin/release.sh Mon Nov 5 16:52:12 2012
@@ -86,7 +86,6 @@ mkdir -p "${WORKDIR}"
cd ${WORKDIR}
svn export -qr ${REVISION} ${URL}/${BRANCH}/proton-c ${rootname}
svn export -qr ${REVISION} ${URL}/${BRANCH}/tests ${rootname}/tests
- svn export -qr ${REVISION} ${URL}/${BRANCH}/examples ${rootname}/examples
cat <<EOF > ${rootname}/SVN_INFO
Repo: ${URL}
@@ -97,7 +96,6 @@ EOF
##
## Remove content not for release
##
- rm -rf ${rootname}/examples/broker
rm -rf ${rootname}/examples/mailbox
echo "Generating Archive: ${CURRDIR}/${rootname}.tar.gz"
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/CMakeLists.txt
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/CMakeLists.txt?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/CMakeLists.txt (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/CMakeLists.txt Mon Nov 5 16:52:12 2012
@@ -21,16 +21,11 @@ cmake_minimum_required (VERSION 2.6)
project (Proton C)
set (PN_VERSION_MAJOR 0)
-set (PN_VERSION_MINOR 1)
+set (PN_VERSION_MINOR 2)
set (PN_VERSION "${PN_VERSION_MAJOR}.${PN_VERSION_MINOR}")
include(soversion.cmake)
-# Disable rpath
-if (DEFINED PROTON_DISABLE_RPATH)
- set (CMAKE_SKIP_RPATH ${PROTON_DISABLE_RPATH})
-endif()
-
# In rpm builds the build sets some variables:
# CMAKE_INSTALL_PREFIX - this is a standard cmake variable
# INCLUDE_INSTALL_DIR
@@ -107,19 +102,26 @@ if (SWIG_FOUND)
endif (SWIG_FOUND)
add_subdirectory(docs/api)
+add_subdirectory(examples/messenger/c)
# Should really be finding the uuid library appropriate for the platform
# in lieu of doing this set the library name directly.
set (UUID_LIB uuid)
+set (COMPILE_WARNING_FLAGS "-Wall -Werror -pedantic-errors")
+set (COMPILE_LANGUAGE_FLAGS "-std=c99")
+set (COMPILE_PLATFORM_FLAGS "-std=gnu99")
+# Set any additional platform specific C compiler flags
+if (CMAKE_COMPILER_IS_GNUCC)
+ set (PN_EXTRA_C_FLAGS "-Wall -Werror -pedantic-errors -std=c99 -g -fPIC")
+endif (CMAKE_COMPILER_IS_GNUCC)
+
set (qpid-proton-platform
src/driver.c
${pn_driver_ssl_impl}
)
-add_library (
- qpid-proton SHARED
-
+set (qpid-proton-core
src/util.c
src/error.c
src/buffer.c
@@ -138,36 +140,48 @@ add_library (
src/messenger.c
- ${qpid-proton-platform}
-
${PROJECT_BINARY_DIR}/encodings.h
${PROJECT_BINARY_DIR}/protocol.h
)
-target_link_libraries (qpid-proton ${UUID_LIB} ${SSL_LIB})
-add_executable (proton src/proton.c)
-target_link_libraries (proton qpid-proton)
+set_source_files_properties (
+ ${qpid-proton-core}
+ PROPERTIES
+ COMPILE_FLAGS "${COMPILE_WARNING_FLAGS} ${COMPILE_LANGUAGE_FLAGS}"
+)
-add_executable (proton-dump src/proton-dump.c)
-target_link_libraries (proton-dump qpid-proton)
+set_source_files_properties (
+ ${qpid-proton-platform}
+ PROPERTIES
+ COMPILE_FLAGS "${COMPILE_WARNING_FLAGS} ${COMPILE_PLATFORM_FLAGS}"
+)
-# Set any additional platform specific C compiler flags
-if (CMAKE_COMPILER_IS_GNUCC)
- set (PN_EXTRA_C_FLAGS "-Wall -Werror -pedantic-errors -std=c99 -g -fPIC")
-endif (CMAKE_COMPILER_IS_GNUCC)
+add_library (
+ qpid-proton SHARED
+
+ ${qpid-proton-core}
+ ${qpid-proton-platform}
+)
+
+target_link_libraries (qpid-proton ${UUID_LIB} ${SSL_LIB})
set_target_properties (
qpid-proton
PROPERTIES
- COMPILE_FLAGS "${PN_EXTRA_C_FLAGS}"
- VERSION ${PN_VERSION}
- SOVERSION ${PN_SOVERSION}
+ VERSION "${PN_LIB_SOMAJOR}.${PN_LIB_SOMINOR}"
+ SOVERSION "${PN_LIB_SOMAJOR}"
)
+add_executable (proton src/proton.c)
+target_link_libraries (proton qpid-proton)
+
+add_executable (proton-dump src/proton-dump.c)
+target_link_libraries (proton-dump qpid-proton)
+
set_target_properties (
- proton proton-dump
- PROPERTIES
- COMPILE_FLAGS "${PN_EXTRA_C_FLAGS}"
+ proton proton-dump
+ PROPERTIES
+ COMPILE_FLAGS "${COMPILE_WARNING_FLAGS} ${COMPILE_LANGUAGE_FLAGS}"
)
# Install executables and libraries
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/php/CMakeLists.txt
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/php/CMakeLists.txt?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/php/CMakeLists.txt (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/php/CMakeLists.txt Mon Nov 5 16:52:12 2012
@@ -72,7 +72,7 @@ set(PHP_EXT_DIR ${PHP_EXT_DIR_DEFAULT} C
set(PHP_INI_DIR ${PHP_INI_DIR_DEFAULT} CACHE PATH "Directory scanned for PHP ini files.")
set(PHP_INCLUDE_DIR ${PHP_INCLUDE_DIR_DEFAULT} CACHE PATH "PHP include directory.")
-install(FILES ${CMAKE_CURRENT_BINARY_DIR}/cproton.so
+install(TARGETS cproton
DESTINATION ${PHP_EXT_DIR}
COMPONENT PHP)
install(FILES ${CMAKE_CURRENT_BINARY_DIR}/cproton.php
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/python/CMakeLists.txt
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/python/CMakeLists.txt?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/python/CMakeLists.txt (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/python/CMakeLists.txt Mon Nov 5 16:52:12 2012
@@ -18,8 +18,8 @@
#
set(CMAKE_SWIG_FLAGS "-threads")
-swig_add_module(cproton python python.i)
include_directories (${PYTHON_INCLUDE_PATH})
+swig_add_module(cproton python python.i)
swig_link_libraries(cproton ${BINDING_DEPS} ${PYTHON_LIBRARIES})
find_package(PythonInterp REQUIRED)
@@ -58,7 +58,6 @@ install(FILES ${CMAKE_CURRENT_BINARY_DIR
${CMAKE_CURRENT_SOURCE_DIR}/proton.pyo
DESTINATION ${PYTHON_SITEARCH_PACKAGES}
COMPONENT Python)
-install(FILES ${CMAKE_CURRENT_BINARY_DIR}/_cproton.so
- RENAME _cproton.so
+install(TARGETS _cproton
DESTINATION ${PYTHON_SITEARCH_PACKAGES}
COMPONENT Python)
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/python/proton.py
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/python/proton.py?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/python/proton.py (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/python/proton.py Mon Nov 5 16:52:12 2012
@@ -33,6 +33,14 @@ The proton APIs consist of the following
from cproton import *
import uuid
+class Constant(object):
+
+ def __init__(self, name):
+ self.name = name
+
+ def __repr__(self):
+ return self.name
+
class ProtonException(Exception):
"""
The root of the proton exception hierarchy. All proton exception
@@ -66,6 +74,29 @@ EXCEPTIONS = {
PN_TIMEOUT: Timeout
}
+PENDING = Constant("PENDING")
+ACCEPTED = Constant("ACCEPTED")
+REJECTED = Constant("REJECTED")
+
+STATUSES = {
+ PN_STATUS_ACCEPTED: ACCEPTED,
+ PN_STATUS_REJECTED: REJECTED,
+ PN_STATUS_PENDING: PENDING,
+ PN_STATUS_UNKNOWN: None
+ }
+
+AUTOMATIC = Constant("AUTOMATIC")
+MANUAL = Constant("MANUAL")
+
+_ACCEPT_MODE2CONST = {
+ PN_ACCEPT_MODE_AUTO: AUTOMATIC,
+ PN_ACCEPT_MODE_MANUAL: MANUAL
+ }
+_CONST2ACCEPT_MODE = {
+ AUTOMATIC: PN_ACCEPT_MODE_AUTO,
+ MANUAL: PN_ACCEPT_MODE_MANUAL
+ }
+
class Messenger(object):
"""
The L{Messenger} class defines a high level interface for sending
@@ -226,6 +257,43 @@ The timeout property contains the defaul
operations performed by the L{Messenger}.
""")
+ def _get_accept_mode(self):
+ return _ACCEPT_MODE2CONST(pn_messenger_get_accept_mode(self._mng))
+
+ def _set_accept_mode(self, mode):
+ mode = _CONST2ACCEPT_MODE[mode]
+ self._check(pn_messenger_set_accept_mode(self._mng, mode))
+
+ accept_mode = property(_get_accept_mode, _set_accept_mode,
+ doc="""
+The accept mode for the messenger. Can be set to AUTOMATIC or MANUAL.
+The default is AUTOMATIC.
+""")
+
+ def _get_incoming_window(self):
+ return pn_messenger_get_incoming_window(self._mng)
+
+ def _set_incoming_window(self, window):
+ self._check(pn_messenger_set_incoming_window(self._mng, window))
+
+ incoming_window = property(_get_incoming_window, _set_incoming_window,
+ doc="""
+The incoming tracking window for the messenger. The messenger will
+track the status of this many incoming deliveries. Defaults to zero.
+""")
+
+ def _get_outgoing_window(self):
+ return pn_messenger_get_outgoing_window(self._mng)
+
+ def _set_outgoing_window(self, window):
+ self._check(pn_messenger_set_outgoing_window(self._mng, window))
+
+ outgoing_window = property(_get_outgoing_window, _set_outgoing_window,
+ doc="""
+The outgoing tracking window for the messenger. The messenger will
+track the status of this many outgoing deliveries. Defaults to zero.
+""")
+
def start(self):
"""
Transitions the L{Messenger} to an active state. A L{Messenger} is
@@ -261,7 +329,9 @@ operations performed by the L{Messenger}
@type source: string
@param source: the source of messages to subscribe to
"""
- self._check(pn_messenger_subscribe(self._mng, source))
+ sub_impl = pn_messenger_subscribe(self._mng, source)
+ if not sub_impl:
+ self._check(PN_ERR)
def put(self, message):
"""
@@ -278,6 +348,19 @@ operations performed by the L{Messenger}
"""
message._pre_encode()
self._check(pn_messenger_put(self._mng, message._msg))
+ return pn_messenger_outgoing_tracker(self._mng)
+
+ def status(self, tracker):
+ disp = pn_messenger_status(self._mng, tracker);
+ return STATUSES.get(disp, disp)
+
+ def settle(self, tracker=None):
+ if tracker is None:
+ tracker = pn_messenger_outgoing_tracker(self._mng)
+ flags = PN_CUMULATIVE
+ else:
+ flags = 0
+ self._check(pn_messenger_settle(self._mng, tracker, flags))
def send(self):
"""
@@ -295,7 +378,7 @@ operations performed by the L{Messenger}
"""
self._check(pn_messenger_recv(self._mng, n))
- def get(self, message):
+ def get(self, message=None):
"""
Moves the message from the head of the incoming message queue into
the supplied message object. Any content in the message will be
@@ -304,8 +387,30 @@ operations performed by the L{Messenger}
@type message: Message
@param message: the destination message object
"""
- self._check(pn_messenger_get(self._mng, message._msg))
- message._post_decode()
+ if message is None:
+ impl = None
+ else:
+ impl = message._msg
+ self._check(pn_messenger_get(self._mng, impl))
+ if message is not None:
+ message._post_decode()
+ return pn_messenger_incoming_tracker(self._mng)
+
+ def accept(self, tracker=None):
+ if tracker is None:
+ tracker = pn_messenger_incoming_tracker(self._mng)
+ flags = PN_CUMULATIVE
+ else:
+ flags = 0
+ self._check(pn_messenger_accept(self._mng, tracker, flags))
+
+ def reject(self, tracker=None):
+ if tracker is None:
+ tracker = pn_messenger_incoming_tracker(self._mng)
+ flags = PN_CUMULATIVE
+ else:
+ flags = 0
+ self._check(pn_messenger_reject(self._mng, tracker, flags))
@property
def outgoing(self):
@@ -721,14 +826,6 @@ class Described(object):
else:
return False
-class Constant(object):
-
- def __init__(self, name):
- self.name = name
-
- def __repr__(self):
- return self.name
-
UNDESCRIBED = Constant("UNDESCRIBED")
class Array(object):
@@ -2152,4 +2249,5 @@ __all__ = ["Messenger", "Message", "Prot
"Session", "Link", "Terminus", "Sender", "Receiver", "Delivery",
"Transport", "TransportException", "SASL", "SSL", "Described",
"Array", "symbol", "char", "timestamp", "ulong", "UNDESCRIBED",
- "SSLUnavailable", "PN_SESSION_WINDOW"]
+ "SSLUnavailable", "PN_SESSION_WINDOW", "AUTOMATIC", "MANUAL",
+ "PENDING", "ACCEPTED", "REJECTED"]
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/ruby/CMakeLists.txt
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/ruby/CMakeLists.txt?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/ruby/CMakeLists.txt (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/bindings/ruby/CMakeLists.txt Mon Nov 5 16:52:12 2012
@@ -20,10 +20,7 @@
include_directories (${RUBY_INCLUDE_PATH})
swig_add_module(cproton-ruby ruby ruby.i)
swig_link_libraries(cproton-ruby ${BINDING_DEPS} ${RUBY_LIBRARY})
-set_target_properties(cproton-ruby PROPERTIES PREFIX "")
-add_custom_command(TARGET cproton-ruby
- POST_BUILD
- COMMAND cmake -E rename cproton-ruby.so cproton.so)
+set_target_properties(cproton-ruby PROPERTIES PREFIX "" OUTPUT_NAME "cproton")
execute_process(COMMAND ${RUBY_EXECUTABLE} -r rbconfig -e "puts RbConfig::CONFIG['prefix']"
OUTPUT_VARIABLE RUBY_PREFIX
@@ -31,7 +28,6 @@ execute_process(COMMAND ${RUBY_EXECUTABL
string(REPLACE "${RUBY_PREFIX}/" "" RUBY_PFX_ARCH_DIR ${RUBY_ARCH_DIR})
-install(FILES ${CMAKE_CURRENT_BINARY_DIR}/cproton.so
- RENAME cproton.so
+install(TARGETS cproton-ruby
DESTINATION ${RUBY_PFX_ARCH_DIR}
COMPONENT Ruby)
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/include/proton/cproton.i
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/include/proton/cproton.i?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/include/proton/cproton.i (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/include/proton/cproton.i Mon Nov 5 16:52:12 2012
@@ -1289,7 +1289,6 @@ typedef long long int int64_t;
{
require:
messenger != NULL;
- msg != NULL;
}
%contract pn_messenger_outgoing(pn_messenger_t *messenger)
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/include/proton/driver.h
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/include/proton/driver.h?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/include/proton/driver.h (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/include/proton/driver.h Mon Nov 5 16:52:12 2012
@@ -189,6 +189,8 @@ pn_connector_t *pn_listener_accept(pn_li
*/
void *pn_listener_context(pn_listener_t *listener);
+void pn_listener_set_context(pn_listener_t *listener, void *context);
+
/** Close the socket used by the listener.
*
* @param[in] listener the listener whose socket will be closed.
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/include/proton/messenger.h
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/include/proton/messenger.h?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/include/proton/messenger.h (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/include/proton/messenger.h Mon Nov 5 16:52:12 2012
@@ -34,6 +34,20 @@ extern "C" {
*/
typedef struct pn_messenger_t pn_messenger_t; /**< Messenger*/
+typedef struct pn_subscription_t pn_subscription_t; /**< Subscription*/
+typedef int64_t pn_tracker_t;
+
+typedef enum {
+ PN_ACCEPT_MODE_AUTO,
+ PN_ACCEPT_MODE_MANUAL
+} pn_accept_mode_t;
+
+typedef enum {
+ PN_STATUS_UNKNOWN = 0,
+ PN_STATUS_PENDING = 1,
+ PN_STATUS_ACCEPTED = 2,
+ PN_STATUS_REJECTED = 3
+} pn_status_t;
/** Construct a new Messenger with the given name. The name is global.
* If a NULL name is supplied, a UUID based name will be chosen.
@@ -159,6 +173,18 @@ int pn_messenger_errno(pn_messenger_t *m
*/
const char *pn_messenger_error(pn_messenger_t *messenger);
+pn_accept_mode_t pn_messenger_get_accept_mode(pn_messenger_t *messenger);
+
+int pn_messenger_set_accept_mode(pn_messenger_t *messenger, pn_accept_mode_t mode);
+
+int pn_messenger_get_outgoing_window(pn_messenger_t *messenger);
+
+int pn_messenger_set_outgoing_window(pn_messenger_t *messenger, int window);
+
+int pn_messenger_get_incoming_window(pn_messenger_t *messenger);
+
+int pn_messenger_set_incoming_window(pn_messenger_t *messenger, int window);
+
/** Starts a messenger. A messenger cannot send or recv messages until
* it is started.
*
@@ -184,10 +210,13 @@ int pn_messenger_stop(pn_messenger_t *me
* @param[in] messenger the messenger to subscribe
* @param[in] source
*
- * @return an error code or zero on success
- * @see error.h
+ * @return a subscription
*/
-int pn_messenger_subscribe(pn_messenger_t *messenger, const char *source);
+pn_subscription_t *pn_messenger_subscribe(pn_messenger_t *messenger, const char *source);
+
+void *pn_subscription_get_context(pn_subscription_t *sub);
+
+void pn_subscription_set_context(pn_subscription_t *sub, void *context);
/** Puts a message on the outgoing message queue for a messenger.
*
@@ -199,8 +228,30 @@ int pn_messenger_subscribe(pn_messenger_
*/
int pn_messenger_put(pn_messenger_t *messenger, pn_message_t *msg);
+/** Gets the last known remote state of the delivery associated with
+ * the given tracker.
+ *
+ * @param[in] messenger the messenger
+ * @param[in] tracker the tracker identify the delivery
+ *
+ * @return a status code for the delivery
+ */
+pn_status_t pn_messenger_status(pn_messenger_t *messenger, pn_tracker_t tracker);
+
+int pn_messenger_settle(pn_messenger_t *messenger, pn_tracker_t tracker, int flags);
+
+/** Gets the tracker for the message most recently provided to
+ * pn_messenger_put.
+ *
+ * @param[in] messenger the messenger
+ *
+ * @return a pn_tracker_t or an undefined value if pn_messenger_get
+ * has never been called for the given messenger
+ */
+pn_tracker_t pn_messenger_outgoing_tracker(pn_messenger_t *messenger);
+
/** Sends any messages in the outgoing message queue for a messenger.
- * This will blocks until the messages have been sent.
+ * This will block until the messages have been sent.
*
* @param[in] messenger the messager
*
@@ -232,6 +283,42 @@ int pn_messenger_recv(pn_messenger_t *me
*/
int pn_messenger_get(pn_messenger_t *messenger, pn_message_t *msg);
+/** Gets the tracker for the message most recently fetched by
+ * pn_messenger_get.
+ *
+ * @param[in] messenger the messenger
+ *
+ * @return a pn_tracker_t or an undefined value if pn_messenger_get
+ * has never been called for the given messenger
+ */
+pn_tracker_t pn_messenger_incoming_tracker(pn_messenger_t *messenger);
+
+pn_subscription_t *pn_messenger_incoming_subscription(pn_messenger_t *messenger);
+
+#define PN_CUMULATIVE (0x1)
+
+/** Accepts the incoming messages identified by the tracker. Use the
+ * PN_CUMULATIVE flag to accept everything prior to the supplied
+ * tracker.
+ *
+ * @param[in] messenger the messenger
+ *
+ * @return an error code or zero on success
+ * @see error.h
+ */
+int pn_messenger_accept(pn_messenger_t *messenger, pn_tracker_t tracker, int flags);
+
+/** Rejects the incoming messages identified by the tracker. Use the
+ * PN_CUMULATIVE flag to reject everything prior to the supplied
+ * tracker.
+ *
+ * @param[in] messenger the messenger
+ *
+ * @return an error code or zero on success
+ * @see error.h
+ */
+int pn_messenger_reject(pn_messenger_t *messenger, pn_tracker_t tracker, int flags);
+
/** Returns the number of messages in the outgoing message queue of a messenger.
*
* @param[in] the messenger
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/soversion.cmake
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/soversion.cmake?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/soversion.cmake (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/soversion.cmake Mon Nov 5 16:52:12 2012
@@ -1 +1,2 @@
-set (PN_SOVERSION "1.0.0")
\ No newline at end of file
+set (PN_LIB_SOMAJOR 1)
+set (PN_LIB_SOMINOR "0.0")
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/src/dispatcher/dispatcher.c
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/src/dispatcher/dispatcher.c?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/src/dispatcher/dispatcher.c (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/src/dispatcher/dispatcher.c Mon Nov 5 16:52:12 2012
@@ -64,6 +64,7 @@ void pn_dispatcher_free(pn_dispatcher_t
pn_buffer_free(disp->input);
pn_data_free(disp->args);
pn_data_free(disp->output_args);
+ pn_buffer_free(disp->frame);
free(disp->output);
free(disp);
}
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/src/driver.c
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/src/driver.c?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/src/driver.c (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/src/driver.c Mon Nov 5 16:52:12 2012
@@ -19,8 +19,7 @@
*
*/
-#define _POSIX_C_SOURCE 1
-
+#include <assert.h>
#include <poll.h>
#include <stdio.h>
#include <time.h>
@@ -210,6 +209,12 @@ void *pn_listener_context(pn_listener_t
return l ? l->context : NULL;
}
+void pn_listener_set_context(pn_listener_t *listener, void *context)
+{
+ assert(listener);
+ listener->context = context;
+}
+
static void pn_configure_sock(int sock) {
// this would be nice, but doesn't appear to exist on linux
/*
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/src/engine/engine-internal.h
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/src/engine/engine-internal.h?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/src/engine/engine-internal.h (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/src/engine/engine-internal.h Mon Nov 5 16:52:12 2012
@@ -82,6 +82,13 @@ typedef struct {
size_t link_capacity;
pn_link_state_t **handles;
size_t handle_capacity;
+
+ uint64_t disp_code;
+ bool disp_settled;
+ bool disp_type;
+ pn_sequence_t disp_first;
+ pn_sequence_t disp_last;
+ bool disp;
} pn_session_state_t;
#define SCRATCH (1024)
@@ -216,20 +223,6 @@ struct pn_delivery_t {
void pn_link_dump(pn_link_t *link);
-#define PN_ENSURE(ARRAY, CAPACITY, COUNT) \
- while ((CAPACITY) < (COUNT)) { \
- (CAPACITY) = (CAPACITY) ? 2 * (CAPACITY) : 16; \
- (ARRAY) = realloc((ARRAY), (CAPACITY) * sizeof (*(ARRAY))); \
- } \
-
-#define PN_ENSUREZ(ARRAY, CAPACITY, COUNT) \
- { \
- size_t _old_capacity = (CAPACITY); \
- PN_ENSURE((ARRAY), (CAPACITY), (COUNT)); \
- memset((ARRAY) + _old_capacity, 0, \
- sizeof(*(ARRAY))*((CAPACITY) - _old_capacity)); \
- }
-
void pn_dump(pn_connection_t *conn);
void pn_transport_sasl_init(pn_transport_t *transport);
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/src/engine/engine.c
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/src/engine/engine.c?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/src/engine/engine.c (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/src/engine/engine.c Mon Nov 5 16:52:12 2012
@@ -1151,12 +1151,20 @@ int pn_link_unsettled(pn_link_t *link)
pn_delivery_t *pn_unsettled_head(pn_link_t *link)
{
- return link->unsettled_head;
+ pn_delivery_t *d = link->unsettled_head;
+ while (d && d->local_settled) {
+ d = d->unsettled_next;
+ }
+ return d;
}
pn_delivery_t *pn_unsettled_next(pn_delivery_t *delivery)
{
- return delivery->unsettled_next;
+ pn_delivery_t *d = delivery->unsettled_next;
+ while (d && d->local_settled) {
+ d = d->unsettled_next;
+ }
+ return d;
}
bool pn_is_current(pn_delivery_t *delivery)
@@ -1661,10 +1669,12 @@ int pn_do_disposition(pn_dispatcher_t *d
if (id < lwm) continue;
pn_delivery_state_t *state = pn_delivery_buffer_get(deliveries, id - lwm);
pn_delivery_t *delivery = state->delivery;
- delivery->remote_state = dispo;
- delivery->remote_settled = settled;
- delivery->updated = true;
- pn_work_update(transport->connection, delivery);
+ if (delivery) {
+ delivery->remote_state = dispo;
+ delivery->remote_settled = settled;
+ delivery->updated = true;
+ pn_work_update(transport->connection, delivery);
+ }
}
return 0;
@@ -1982,10 +1992,30 @@ int pn_process_flow_receiver(pn_transpor
return 0;
}
+int pn_flush_disp(pn_transport_t *transport, pn_session_state_t *ssn_state)
+{
+ uint64_t code = ssn_state->disp_code;
+ bool settled = ssn_state->disp_settled;
+ if (ssn_state->disp) {
+ int err = pn_post_frame(transport->disp, ssn_state->local_channel, "DL[oIIo?DL[]]", DISPOSITION,
+ ssn_state->disp_type, ssn_state->disp_first, ssn_state->disp_last,
+ settled, (bool)code, code);
+ if (err) return err;
+ ssn_state->disp_type = 0;
+ ssn_state->disp_code = 0;
+ ssn_state->disp_settled = 0;
+ ssn_state->disp_first = 0;
+ ssn_state->disp_last = 0;
+ ssn_state->disp = false;
+ }
+ return 0;
+}
+
int pn_post_disp(pn_transport_t *transport, pn_delivery_t *delivery)
{
pn_link_t *link = delivery->link;
pn_session_state_t *ssn_state = pn_session_get_state(transport, link->session);
+ pn_modified(transport->connection, &link->session->endpoint);
// XXX: check for null state
pn_delivery_state_t *state = delivery->transport_context;
uint64_t code;
@@ -1996,16 +2026,42 @@ int pn_post_disp(pn_transport_t *transpo
case PN_RELEASED:
code = RELEASED;
break;
+ case PN_REJECTED:
+ code = REJECTED;
+ break;
//TODO: rejected and modified (both take extra data which may need to be passed through somehow) e.g. change from enum to discriminated union?
default:
code = 0;
}
- if (code || delivery->local_settled)
- return pn_post_frame(transport->disp, ssn_state->local_channel, "DL[oIIo?DL[]]", DISPOSITION,
- link->endpoint.type == RECEIVER, state->id, state->id,
- delivery->local_settled,
- (bool)code, code);
+ if (!code && !delivery->local_settled) {
+ return 0;
+ }
+
+ if (ssn_state->disp && code == ssn_state->disp_code &&
+ delivery->local_settled == ssn_state->disp_settled &&
+ ssn_state->disp_type == (link->endpoint.type == RECEIVER)) {
+ if (state->id == ssn_state->disp_first - 1) {
+ ssn_state->disp_first = state->id;
+ return 0;
+ } else if (state->id == ssn_state->disp_last + 1) {
+ ssn_state->disp_last = state->id;
+ return 0;
+ }
+ }
+
+ if (ssn_state->disp) {
+ int err = pn_flush_disp(transport, ssn_state);
+ if (err) return err;
+ }
+
+ ssn_state->disp_type = (link->endpoint.type == RECEIVER);
+ ssn_state->disp_code = code;
+ ssn_state->disp_settled = delivery->local_settled;
+ ssn_state->disp_first = state->id;
+ ssn_state->disp_last = state->id;
+ ssn_state->disp = true;
+
return 0;
}
@@ -2092,6 +2148,10 @@ int pn_process_tpwork(pn_transport_t *tr
pn_delivery_t *delivery = conn->tpwork_head;
while (delivery)
{
+ if (!delivery->transport_context && transport->disp->available > 0) {
+ break;
+ }
+
pn_link_t *link = delivery->link;
if (pn_link_is_sender(link)) {
int err = pn_process_tpwork_sender(transport, delivery);
@@ -2112,6 +2172,21 @@ int pn_process_tpwork(pn_transport_t *tr
return 0;
}
+int pn_process_flush_disp(pn_transport_t *transport, pn_endpoint_t *endpoint)
+{
+ if (endpoint->type == SESSION) {
+ pn_session_t *session = (pn_session_t *) endpoint;
+ pn_session_state_t *state = pn_session_get_state(transport, session);
+ if ((int16_t) state->local_channel >= 0 && !transport->close_sent)
+ {
+ int err = pn_flush_disp(transport, state);
+ if (err) return err;
+ }
+ }
+
+ return 0;
+}
+
int pn_process_flow_sender(pn_transport_t *transport, pn_endpoint_t *endpoint)
{
if (endpoint->type == SENDER && endpoint->state & PN_LOCAL_ACTIVE)
@@ -2251,6 +2326,8 @@ int pn_process(pn_transport_t *transport
if ((err = pn_phase(transport, pn_process_tpwork))) return err;
if ((err = pn_phase(transport, pn_process_tpwork))) return err;
+ if ((err = pn_phase(transport, pn_process_flush_disp))) return err;
+
if ((err = pn_phase(transport, pn_process_flow_sender))) return err;
if ((err = pn_phase(transport, pn_process_link_teardown))) return err;
if ((err = pn_phase(transport, pn_process_ssn_teardown))) return err;
@@ -2292,7 +2369,7 @@ static ssize_t pn_output_write_sasl(pn_t
ssize_t n = pn_sasl_output(sasl, bytes, size);
if (n == PN_EOS) {
transport->process_output = pn_output_write_amqp_header;
- return 0;
+ return transport->process_output(transport, bytes, size);
} else {
return n;
}
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/src/messenger.c
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/src/messenger.c?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/src/messenger.c (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/src/messenger.c Mon Nov 5 16:52:12 2012
@@ -24,6 +24,7 @@
#include <proton/util.h>
#include <proton/ssl.h>
#include <proton/buffer.h>
+#include <assert.h>
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
@@ -31,6 +32,15 @@
#include <uuid/uuid.h>
#include "util.h"
+typedef struct {
+ size_t capacity;
+ int window;
+ pn_sequence_t lwm;
+ pn_sequence_t mwm;
+ pn_sequence_t hwm;
+ pn_delivery_t **deliveries;
+} pn_queue_t;
+
struct pn_messenger_t {
char *name;
char *certificate;
@@ -42,10 +52,181 @@ struct pn_messenger_t {
int credit;
int distributed;
uint64_t next_tag;
+ pn_queue_t outgoing;
+ pn_queue_t incoming;
+ pn_accept_mode_t accept_mode;
+ pn_subscription_t *subscriptions;
+ size_t sub_capacity;
+ size_t sub_count;
+ pn_subscription_t *incoming_subscription;
pn_buffer_t *buffer;
pn_error_t *error;
};
+struct pn_subscription_t {
+ char *scheme;
+ void *context;
+};
+
+void pn_queue_init(pn_queue_t *queue)
+{
+ queue->capacity = 1024;
+ queue->window = 0;
+ queue->lwm = 0;
+ queue->mwm = 0;
+ queue->hwm = 0;
+ queue->deliveries = calloc(queue->capacity, sizeof(pn_delivery_t *));
+}
+
+void pn_queue_tini(pn_queue_t *queue)
+{
+ free(queue->deliveries);
+}
+
+bool pn_queue_contains(pn_queue_t *queue, pn_sequence_t id)
+{
+ return (id - queue->lwm >= 0) && (queue->hwm - id > 0);
+}
+
+pn_delivery_t *pn_queue_get(pn_queue_t *queue, pn_sequence_t id)
+{
+ if (pn_queue_contains(queue, id)) {
+ size_t offset = id - queue->lwm;
+ assert(offset >= 0 && offset < queue->capacity);
+ return queue->deliveries[offset];
+ } else {
+ return NULL;
+ }
+}
+
+void pn_queue_gc(pn_queue_t *queue)
+{
+ size_t count = queue->hwm - queue->lwm;
+ size_t delta = 0;
+
+ while (delta < count && !queue->deliveries[delta]) {
+ delta++;
+ }
+
+ memmove(queue->deliveries, queue->deliveries + delta, (count - delta)*sizeof(pn_delivery_t *));
+ queue->lwm += delta;
+ if (queue->mwm - queue->lwm < 0) {
+ queue->mwm = queue->lwm;
+ }
+}
+
+void pn_incref(pn_connection_t *conn)
+{
+ intptr_t refcount = (intptr_t) pn_connection_get_context(conn);
+ pn_connection_set_context(conn, (void *) (refcount + 1));
+}
+
+void pn_decref(pn_connection_t *conn)
+{
+ intptr_t refcount = (intptr_t) pn_connection_get_context(conn);
+ pn_connection_set_context(conn, (void *) (refcount - 1));
+ if (refcount == 1) {
+ pn_connection_free(conn);
+ }
+}
+
+void pn_queue_del(pn_queue_t *queue, pn_delivery_t *delivery)
+{
+ pn_sequence_t id = (pn_sequence_t) (intptr_t) pn_delivery_get_context(delivery);
+ if (pn_queue_contains(queue, id)) {
+ size_t offset = id - queue->lwm;
+ assert(offset >= 0 && offset < queue->capacity);
+ queue->deliveries[offset] = NULL;
+ pn_delivery_set_context(delivery, NULL);
+ pn_connection_t *conn =
+ pn_session_connection(pn_link_session(pn_delivery_link(delivery)));
+ pn_decref(conn);
+ }
+}
+
+pn_sequence_t pn_queue_add(pn_queue_t *queue, pn_delivery_t *delivery)
+{
+ pn_sequence_t id = queue->hwm++;
+ size_t offset = id - queue->lwm;
+ PN_ENSUREZ(queue->deliveries, queue->capacity, offset + 1);
+ assert(offset >= 0 && offset < queue->capacity);
+ queue->deliveries[offset] = delivery;
+ pn_delivery_set_context(delivery, (void *) (intptr_t) id);
+ pn_connection_t *conn =
+ pn_session_connection(pn_link_session(pn_delivery_link(delivery)));
+ pn_incref(conn);
+ return id;
+}
+
+void pn_queue_slide(pn_queue_t *queue)
+{
+ if (queue->window >= 0) {
+ while (queue->hwm - queue->mwm > queue->window) {
+ pn_delivery_t *d = pn_queue_get(queue, queue->lwm);
+ if (d) {
+ pn_delivery_settle(d);
+ pn_queue_del(queue, d);
+ } else {
+ pn_queue_gc(queue);
+ }
+ }
+ }
+ pn_queue_gc(queue);
+}
+
+int pn_queue_update(pn_queue_t *queue, pn_sequence_t id, pn_status_t status,
+ int flags, bool settle, bool match)
+{
+ if (!pn_queue_contains(queue, id)) {
+ return 0;
+ }
+
+ size_t start;
+ if (PN_CUMULATIVE | flags) {
+ start = queue->lwm;
+ } else {
+ start = id;
+ }
+
+ for (pn_sequence_t i = start; i <= id; i++) {
+ pn_delivery_t *d = pn_queue_get(queue, i);
+ if (d) {
+ if (match) {
+ pn_delivery_update(d, pn_delivery_remote_state(d));
+ } else {
+ switch (status) {
+ case PN_STATUS_ACCEPTED:
+ pn_delivery_update(d, PN_ACCEPTED);
+ break;
+ case PN_STATUS_REJECTED:
+ pn_delivery_update(d, PN_REJECTED);
+ break;
+ default:
+ break;
+ }
+ }
+ if (pn_delivery_local_state(d) && i - queue->mwm < 0) {
+ queue->mwm = i;
+ }
+ if (settle) {
+ pn_delivery_settle(d);
+ pn_queue_del(queue, d);
+ }
+ }
+ }
+
+ pn_queue_slide(queue);
+
+ return 0;
+}
+
+#define OUTGOING (0x0000000000000000)
+#define INCOMING (0x1000000000000000)
+
+#define pn_tracker(direction, sequence) ((direction) | (sequence))
+#define pn_tracker_direction(tracker) ((tracker) & (0x1000000000000000))
+#define pn_tracker_sequence(tracker) ((pn_sequence_t) ((tracker) & (0x00000000FFFFFFFF)))
+
char *build_name(const char *name)
{
if (name) {
@@ -74,6 +255,13 @@ pn_messenger_t *pn_messenger(const char
m->credit = 0;
m->distributed = 0;
m->next_tag = 0;
+ pn_queue_init(&m->outgoing);
+ pn_queue_init(&m->incoming);
+ m->accept_mode = PN_ACCEPT_MODE_AUTO;
+ m->subscriptions = NULL;
+ m->sub_capacity = 0;
+ m->sub_count = 0;
+ m->incoming_subscription = NULL;
m->buffer = pn_buffer(1024);
m->error = pn_error();
}
@@ -157,6 +345,12 @@ void pn_messenger_free(pn_messenger_t *m
pn_driver_free(messenger->driver);
pn_buffer_free(messenger->buffer);
pn_error_free(messenger->error);
+ pn_queue_tini(&messenger->incoming);
+ pn_queue_tini(&messenger->outgoing);
+ for (int i = 0; i < messenger->sub_count; i++) {
+ free(messenger->subscriptions[i].scheme);
+ }
+ free(messenger->subscriptions);
free(messenger);
}
}
@@ -209,6 +403,22 @@ void pn_messenger_endpoints(pn_messenger
pn_connection_open(conn);
}
+ pn_delivery_t *d = pn_work_head(conn);
+ while (d) {
+ pn_link_t *link = pn_delivery_link(d);
+ if (pn_delivery_updated(d) && pn_link_is_sender(link)) {
+ pn_delivery_update(d, pn_delivery_remote_state(d));
+ }
+ pn_delivery_clear(d);
+ d = pn_work_next(d);
+ }
+
+ pn_queue_slide(&messenger->outgoing);
+
+ if (pn_work_head(conn)) {
+ return;
+ }
+
pn_session_t *ssn = pn_session_head(conn, PN_LOCAL_UNINIT);
while (ssn) {
pn_session_open(ssn);
@@ -282,7 +492,8 @@ int pn_messenger_tsync(pn_messenger_t *m
pn_listener_t *l;
while ((l = pn_driver_listener(messenger->driver))) {
- char *scheme = pn_listener_context(l);
+ pn_subscription_t *sub = pn_listener_context(l);
+ char *scheme = sub->scheme;
pn_connector_t *c = pn_listener_accept(l);
pn_transport_t *t = pn_connector_transport(c);
pn_ssl_t *ssl = pn_ssl(t);
@@ -300,6 +511,7 @@ int pn_messenger_tsync(pn_messenger_t *m
pn_sasl_server(sasl);
pn_sasl_done(sasl, PN_SASL_OK);
pn_connection_t *conn = pn_connection();
+ pn_incref(conn);
pn_connection_set_container(conn, messenger->name);
pn_connector_set_connection(c, conn);
}
@@ -312,7 +524,7 @@ int pn_messenger_tsync(pn_messenger_t *m
if (pn_connector_closed(c)) {
pn_connector_free(c);
pn_messenger_reclaim(messenger, conn);
- pn_connection_free(conn);
+ pn_decref(conn);
pn_messenger_flow(messenger);
} else {
pn_connector_process(c);
@@ -365,7 +577,6 @@ int pn_messenger_stop(pn_messenger_t *me
pn_listener_close(l);
pn_listener_t *prev = l;
l = pn_listener_next(l);
- free(pn_listener_context(prev));
pn_listener_free(prev);
}
@@ -446,6 +657,7 @@ pn_connection_t *pn_messenger_resolve(pn
pn_sasl_client(sasl);
}
pn_connection_t *connection = pn_connection();
+ pn_incref(connection);
pn_connection_set_container(connection, messenger->name);
pn_connection_set_hostname(connection, domain);
pn_connection_open(connection);
@@ -454,6 +666,27 @@ pn_connection_t *pn_messenger_resolve(pn
return connection;
}
+pn_subscription_t *pn_subscription(pn_messenger_t *messenger, const char *scheme)
+{
+ PN_ENSURE(messenger->subscriptions, messenger->sub_capacity, messenger->sub_count + 1);
+ pn_subscription_t *sub = messenger->subscriptions + messenger->sub_count++;
+ sub->scheme = pn_strdup(scheme);
+ sub->context = NULL;
+ return sub;
+}
+
+void *pn_subscription_get_context(pn_subscription_t *sub)
+{
+ assert(sub);
+ return sub->context;
+}
+
+void pn_subscription_set_context(pn_subscription_t *sub, void *context)
+{
+ assert(sub);
+ sub->context = context;
+}
+
pn_link_t *pn_messenger_link(pn_messenger_t *messenger, const char *address, bool sender)
{
char copy[(address ? strlen(address) : 0) + 1];
@@ -482,12 +715,11 @@ pn_link_t *pn_messenger_link(pn_messenge
pn_session_open(ssn);
link = sender ? pn_sender(ssn, "sender-xxx") : pn_receiver(ssn, "receiver-xxx");
// XXX
- if (sender) {
- pn_terminus_set_address(pn_link_target(link), name);
- pn_terminus_set_address(pn_link_source(link), name);
- } else {
- pn_terminus_set_address(pn_link_target(link), name);
- pn_terminus_set_address(pn_link_source(link), name);
+ pn_terminus_set_address(pn_link_target(link), name);
+ pn_terminus_set_address(pn_link_source(link), name);
+ if (!sender) {
+ pn_subscription_t *sub = pn_subscription(messenger, NULL);
+ pn_link_set_context(link, sub);
}
pn_link_open(link);
return link;
@@ -503,7 +735,7 @@ pn_link_t *pn_messenger_target(pn_messen
return pn_messenger_link(messenger, target, true);
}
-int pn_messenger_subscribe(pn_messenger_t *messenger, const char *source)
+pn_subscription_t *pn_messenger_subscribe(pn_messenger_t *messenger, const char *source)
{
char copy[strlen(source) + 1];
strcpy(copy, source);
@@ -519,27 +751,76 @@ int pn_messenger_subscribe(pn_messenger_
if (host[0] == '~') {
pn_listener_t *lnr = pn_listener(messenger->driver, host + 1,
- port ? port : default_port(scheme),
- pn_strdup(scheme));
+ port ? port : default_port(scheme), NULL);
if (lnr) {
- return 0;
+ pn_subscription_t *sub = pn_subscription(messenger, scheme);
+ pn_listener_set_context(lnr, sub);
+ return sub;
} else {
- return pn_error_format(messenger->error, PN_ERR,
- "unable to subscribe to source: %s (%s)", source,
- pn_driver_error(messenger->driver));
+ pn_error_format(messenger->error, PN_ERR,
+ "unable to subscribe to source: %s (%s)", source,
+ pn_driver_error(messenger->driver));
+ return NULL;
}
} else {
pn_link_t *src = pn_messenger_source(messenger, source);
if (src) {
- return 0;
+ pn_subscription_t *sub = pn_link_get_context(src);
+ return sub;
} else {
- return pn_error_format(messenger->error, PN_ERR,
- "unable to subscribe to source: %s (%s)", source,
- pn_driver_error(messenger->driver));
+ pn_error_format(messenger->error, PN_ERR,
+ "unable to subscribe to source: %s (%s)", source,
+ pn_driver_error(messenger->driver));
+ return NULL;
}
}
}
+pn_accept_mode_t pn_messenger_get_accept_mode(pn_messenger_t *messenger)
+{
+ return messenger->accept_mode;
+}
+
+int pn_messenger_set_accept_mode(pn_messenger_t *messenger, pn_accept_mode_t mode)
+{
+ messenger->accept_mode = mode;
+ return 0;
+}
+
+int pn_messenger_get_outgoing_window(pn_messenger_t *messenger)
+{
+ return messenger->outgoing.window;
+}
+
+int pn_messenger_set_outgoing_window(pn_messenger_t *messenger, int window)
+{
+ if (window >= PN_SESSION_WINDOW) {
+ return pn_error_format(messenger->error, PN_ARG_ERR,
+ "specified window (%i) exceeds max (%i)",
+ window, PN_SESSION_WINDOW);
+ }
+
+ messenger->outgoing.window = window;
+ return 0;
+}
+
+int pn_messenger_get_incoming_window(pn_messenger_t *messenger)
+{
+ return messenger->incoming.window;
+}
+
+int pn_messenger_set_incoming_window(pn_messenger_t *messenger, int window)
+{
+ if (window >= PN_SESSION_WINDOW) {
+ return pn_error_format(messenger->error, PN_ARG_ERR,
+ "specified window (%i) exceeds max (%i)",
+ window, PN_SESSION_WINDOW);
+ }
+
+ messenger->incoming.window = window;
+ return 0;
+}
+
static void outward_munge(pn_messenger_t *mng, pn_message_t *msg)
{
const char *address = pn_message_get_reply_to(msg);
@@ -586,15 +867,17 @@ int pn_messenger_put(pn_messenger_t *mes
char tag[8];
void *ptr = &tag;
uint64_t next = messenger->next_tag++;
- *((uint32_t *) ptr) = next;
- pn_delivery(sender, pn_dtag(tag, 8));
+ *((uint64_t *) ptr) = next;
+ pn_delivery_t *d = pn_delivery(sender, pn_dtag(tag, 8));
ssize_t n = pn_link_send(sender, encoded, size);
if (n < 0) {
return pn_error_format(messenger->error, n, "send error: %s",
pn_error_text(pn_link_error(sender)));
} else {
pn_link_advance(sender);
- pn_messenger_tsync(messenger, false_pred, 0);
+ pn_queue_add(&messenger->outgoing, d);
+ // XXX: doing this every time is slow, need to be smarter
+ //pn_messenger_tsync(messenger, false_pred, 0);
return 0;
}
}
@@ -603,6 +886,56 @@ int pn_messenger_put(pn_messenger_t *mes
return PN_ERR;
}
+pn_tracker_t pn_messenger_outgoing_tracker(pn_messenger_t *messenger)
+{
+ return pn_tracker(OUTGOING, messenger->outgoing.hwm - 1);
+}
+
+pn_queue_t *pn_tracker_queue(pn_messenger_t *messenger, pn_tracker_t tracker)
+{
+ if (pn_tracker_direction(tracker) == OUTGOING) {
+ return &messenger->outgoing;
+ } else {
+ return &messenger->incoming;
+ }
+}
+
+pn_status_t disp2status(pn_disposition_t disp)
+{
+ if (!disp) return PN_STATUS_PENDING;
+
+ switch (disp) {
+ case PN_ACCEPTED:
+ return PN_STATUS_ACCEPTED;
+ case PN_REJECTED:
+ return PN_STATUS_REJECTED;
+ default:
+ assert(0);
+ }
+}
+
+pn_status_t pn_messenger_status(pn_messenger_t *messenger, pn_tracker_t tracker)
+{
+ pn_queue_t *queue = pn_tracker_queue(messenger, tracker);
+ pn_delivery_t *d = pn_queue_get(queue, pn_tracker_sequence(tracker));
+ if (d) {
+ if (pn_delivery_remote_state(d))
+ return disp2status(pn_delivery_remote_state(d));
+ else if (pn_delivery_settled(d))
+ return disp2status(pn_delivery_local_state(d));
+ else
+ return PN_STATUS_PENDING;
+ } else {
+ return PN_STATUS_UNKNOWN;
+ }
+}
+
+int pn_messenger_settle(pn_messenger_t *messenger, pn_tracker_t tracker, int flags)
+{
+ pn_queue_t *queue = pn_tracker_queue(messenger, tracker);
+ return pn_queue_update(queue, pn_tracker_sequence(tracker), 0, flags, true, true);
+}
+
bool pn_messenger_sent(pn_messenger_t *messenger)
{
pn_connector_t *ctor = pn_connector_head(messenger->driver);
@@ -612,11 +945,13 @@ bool pn_messenger_sent(pn_messenger_t *m
pn_link_t *link = pn_link_head(conn, PN_LOCAL_ACTIVE);
while (link) {
if (pn_link_is_sender(link)) {
+ if (pn_link_queued(link)) {
+ return false;
+ }
+
pn_delivery_t *d = pn_unsettled_head(link);
while (d) {
- if (pn_delivery_remote_state(d) || pn_delivery_settled(d)) {
- pn_delivery_settle(d);
- } else {
+ if (!pn_delivery_remote_state(d) && !pn_delivery_settled(d)) {
return false;
}
d = pn_unsettled_next(d);
@@ -677,8 +1012,9 @@ int pn_messenger_get(pn_messenger_t *mes
pn_delivery_t *d = pn_work_head(conn);
while (d) {
- if (pn_delivery_readable(d)) {
+ if (pn_delivery_readable(d) && !pn_delivery_partial(d)) {
pn_link_t *l = pn_delivery_link(d);
+ pn_subscription_t *sub = pn_link_get_context(l);
size_t pending = pn_delivery_pending(d);
pn_buffer_t *buf = messenger->buffer;
int err = pn_buffer_ensure(buf, pending + 1);
@@ -689,22 +1025,35 @@ int pn_messenger_get(pn_messenger_t *mes
return pn_error_format(messenger->error, n, "didn't receive pending bytes: %zi", n);
}
n = pn_link_recv(l, encoded + pending, 1);
- pn_delivery_update(d, PN_ACCEPTED);
- pn_delivery_settle(d);
+ pn_link_advance(l);
messenger->distributed--;
if (n != PN_EOS) {
return pn_error_format(messenger->error, n, "PN_EOS expected");
}
+ pn_queue_add(&messenger->incoming, d);
+ messenger->incoming_subscription = sub;
if (msg) {
int err = pn_message_decode(msg, encoded, pending);
if (err) {
return pn_error_format(messenger->error, err, "error decoding message: %s",
pn_message_error(msg));
} else {
- return 0;
+ if (messenger->accept_mode == PN_ACCEPT_MODE_AUTO) {
+ return pn_messenger_accept(messenger,
+ pn_messenger_incoming_tracker(messenger),
+ 0);
+ } else {
+ return 0;
+ }
}
} else {
- return 0;
+ if (messenger->accept_mode == PN_ACCEPT_MODE_AUTO) {
+ return pn_messenger_accept(messenger,
+ pn_messenger_incoming_tracker(messenger),
+ 0);
+ } else {
+ return 0;
+ }
}
}
d = pn_work_next(d);
@@ -718,6 +1067,39 @@ int pn_messenger_get(pn_messenger_t *mes
return PN_EOS;
}
+pn_tracker_t pn_messenger_incoming_tracker(pn_messenger_t *messenger)
+{
+ return pn_tracker(INCOMING, messenger->incoming.hwm - 1);
+}
+
+pn_subscription_t *pn_messenger_incoming_subscription(pn_messenger_t *messenger)
+{
+ assert(messenger);
+ return messenger->incoming_subscription;
+}
+
+int pn_messenger_accept(pn_messenger_t *messenger, pn_tracker_t tracker, int flags)
+{
+ if (pn_tracker_direction(tracker) != INCOMING) {
+ return pn_error_format(messenger->error, PN_ARG_ERR,
+ "invalid tracker, incoming tracker required");
+ }
+
+ return pn_queue_update(&messenger->incoming, pn_tracker_sequence(tracker),
+ PN_ACCEPTED, flags, false, false);
+}
+
+int pn_messenger_reject(pn_messenger_t *messenger, pn_tracker_t tracker, int flags)
+{
+ if (pn_tracker_direction(tracker) != INCOMING) {
+ return pn_error_format(messenger->error, PN_ARG_ERR,
+ "invalid tracker, incoming tracker required");
+ }
+
+ return pn_queue_update(&messenger->incoming, pn_tracker_sequence(tracker),
+ PN_REJECTED, flags, false, false);
+}
+
int pn_messenger_queued(pn_messenger_t *messenger, bool sender)
{
if (!messenger) return 0;
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/src/ssl/openssl.c
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/src/ssl/openssl.c?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/src/ssl/openssl.c (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/src/ssl/openssl.c Mon Nov 5 16:52:12 2012
@@ -19,8 +19,6 @@
*
*/
-#define _POSIX_C_SOURCE 1
-
#include <proton/ssl.h>
#include "./ssl-internal.h"
#include <proton/engine.h>
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/src/ssl/ssl-internal.h
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/src/ssl/ssl-internal.h?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/src/ssl/ssl-internal.h (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/src/ssl/ssl-internal.h Mon Nov 5 16:52:12 2012
@@ -21,10 +21,6 @@
*
*/
-#ifndef _POSIX_C_SOURCE
-#define _POSIX_C_SOURCE 1
-#endif
-
#include <proton/driver.h>
/** @file
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/src/ssl/ssl_stub.c
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/src/ssl/ssl_stub.c?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/src/ssl/ssl_stub.c (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/src/ssl/ssl_stub.c Mon Nov 5 16:52:12 2012
@@ -19,8 +19,6 @@
*
*/
-#define _POSIX_C_SOURCE 1
-
#include <proton/ssl.h>
#include "ssl-internal.h"
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-c/src/util.h
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-c/src/util.h?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-c/src/util.h (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-c/src/util.h Mon Nov 5 16:52:12 2012
@@ -95,4 +95,18 @@ char *pn_strndup(const char *src, size_t
#define pn_min(X,Y) ((X) > (Y) ? (Y) : (X))
#define pn_max(X,Y) ((X) < (Y) ? (Y) : (X))
+#define PN_ENSURE(ARRAY, CAPACITY, COUNT) \
+ while ((CAPACITY) < (COUNT)) { \
+ (CAPACITY) = (CAPACITY) ? 2 * (CAPACITY) : 16; \
+ (ARRAY) = realloc((ARRAY), (CAPACITY) * sizeof (*(ARRAY))); \
+ } \
+
+#define PN_ENSUREZ(ARRAY, CAPACITY, COUNT) \
+ { \
+ size_t _old_capacity = (CAPACITY); \
+ PN_ENSURE((ARRAY), (CAPACITY), (COUNT)); \
+ memset((ARRAY) + _old_capacity, 0, \
+ sizeof(*(ARRAY))*((CAPACITY) - _old_capacity)); \
+ }
+
#endif /* util.h */
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-j/src/main/java/org/apache/qpid/proton/engine/Link.java
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-j/src/main/java/org/apache/qpid/proton/engine/Link.java?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-j/src/main/java/org/apache/qpid/proton/engine/Link.java (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-j/src/main/java/org/apache/qpid/proton/engine/Link.java Mon Nov 5 16:52:12 2012
@@ -34,6 +34,13 @@ public interface Link extends Endpoint
{
/**
+ * Returns the name of the link
+ *
+ * @return the link name
+ */
+ String getName();
+
+ /**
*
* @param tag a tag for the delivery
* @param offset
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/LinkImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/LinkImpl.java?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/LinkImpl.java (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/LinkImpl.java Mon Nov 5 16:52:12 2012
@@ -75,7 +75,7 @@ public abstract class LinkImpl extends E
modified();
}
- String getName()
+ public String getName()
{
return _name;
}
Modified: qpid/proton/branches/kgiusti-msvc2010/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java (original)
+++ qpid/proton/branches/kgiusti-msvc2010/proton-j/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java Mon Nov 5 16:52:12 2012
@@ -305,9 +305,9 @@ public class TransportImpl extends Endpo
{
TransportSender transportLink = sender.getTransportLink();
TransportSession transportSession = sender.getSession().getTransportSession();
- int credits = sender.getCredit();
- sender.setCredit(0);
- transportLink.setDeliveryCount(transportLink.getDeliveryCount().add(UnsignedInteger.valueOf(credits)));
+ UnsignedInteger credits = transportLink.getLinkCredit();
+ transportLink.setLinkCredit(UnsignedInteger.valueOf(0));
+ transportLink.setDeliveryCount(transportLink.getDeliveryCount().add(credits));
transportLink.setLinkCredit(UnsignedInteger.ZERO);
Flow flow = new Flow();
Modified: qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/messenger.py
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/messenger.py?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/messenger.py (original)
+++ qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/messenger.py Mon Nov 5 16:52:12 2012
@@ -29,11 +29,14 @@ class Test(common.Test):
self.server.start()
self.server.subscribe("amqp://~0.0.0.0:12345")
self.thread = Thread(name="server-thread", target=self.run)
+ self.thread.daemon = True
self.running = True
- self.thread.start()
self.client = Messenger("client")
self.client.timeout=1000
+
+ def start(self):
+ self.thread.start()
self.client.start()
def teardown(self):
@@ -48,6 +51,8 @@ class Test(common.Test):
self.client = None
self.server = None
+REJECT_ME = "*REJECT-ME*"
+
class MessengerTest(Test):
def run(self):
@@ -57,15 +62,21 @@ class MessengerTest(Test):
self.server.recv(10)
while self.server.incoming:
self.server.get(msg)
- if msg.reply_to:
- msg.address = msg.reply_to
- self.server.put(msg)
+ if msg.body == REJECT_ME:
+ self.server.reject()
+ else:
+ self.server.accept()
+ if msg.reply_to:
+ msg.address = msg.reply_to
+ self.server.put(msg)
+ self.server.settle()
except Timeout:
print "server timed out"
self.server.stop()
self.running = False
def _testSendReceive(self, size=None):
+ self.start()
msg = Message()
msg.address="amqp://0.0.0.0:12345"
msg.subject="Hello World!"
@@ -109,6 +120,7 @@ class MessengerTest(Test):
self._testSendReceive(1024*1024)
def testSendBogus(self):
+ self.start()
msg = Message()
msg.address="totally-bogus-address"
try:
@@ -116,3 +128,96 @@ class MessengerTest(Test):
except MessengerException, exc:
err = str(exc)
assert "unable to send to address: totally-bogus-address (" in err, err
+
+ def testOutgoingWindow(self):
+ self.server.accept_mode = MANUAL
+ self.start()
+ msg = Message()
+ msg.address="amqp://0.0.0.0:12345"
+ msg.subject="Hello World!"
+
+ trackers = []
+ for i in range(10):
+ trackers.append(self.client.put(msg))
+
+ self.client.send()
+
+ for t in trackers:
+ assert self.client.status(t) is None
+
+ self.client.outgoing_window = 5
+
+ trackers = []
+ for i in range(10):
+ trackers.append(self.client.put(msg))
+
+ for t in trackers:
+ assert self.client.status(t) is PENDING
+
+ self.client.send()
+
+ count = 0
+ for t in trackers:
+ count += 1
+ if count > 5:
+ assert self.client.status(t) is ACCEPTED
+ else:
+ assert self.client.status(t) is None
+
+ def testReject(self):
+ self.server.accept_mode = MANUAL
+ self.start()
+ msg = Message()
+ msg.address="amqp://0.0.0.0:12345"
+ msg.subject="Hello World!"
+
+ self.client.outgoing_window = 10
+ trackers = []
+ rejected = []
+ for i in range(10):
+ if i == 5:
+ msg.body = REJECT_ME
+ else:
+ msg.body = "Yay!"
+ trackers.append(self.client.put(msg))
+ if msg.body == REJECT_ME:
+ rejected.append(trackers[-1])
+
+ self.client.send()
+
+ for t in trackers:
+ if t in rejected:
+ assert self.client.status(t) is REJECTED
+ else:
+ assert self.client.status(t) is ACCEPTED
+
+ def testIncomingWindow(self):
+ self.server.accept_mode = MANUAL
+ self.server.outgoing_window = 10
+ self.start()
+ msg = Message()
+ msg.address="amqp://0.0.0.0:12345"
+ msg.subject="Hello World!"
+
+ self.client.outgoing_window = 10
+ trackers = []
+ for i in range(10):
+ trackers.append(self.client.put(msg))
+
+ self.client.send()
+
+ for t in trackers:
+ assert self.client.status(t) is ACCEPTED
+
+ self.client.incoming_window = 10
+
+ remaining = 10
+
+ trackers = []
+ while remaining:
+ self.client.recv(remaining)
+ while self.client.incoming:
+ trackers.append(self.client.get())
+ remaining -= 1
+ for t in trackers:
+ assert self.client.status(t) is ACCEPTED
Modified: qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/sasl.py
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/sasl.py?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/sasl.py (original)
+++ qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/sasl.py Mon Nov 5 16:52:12 2012
@@ -84,3 +84,25 @@ class SaslTest(Test):
self.s1.plain("secundus", "trustno1")
self.pump()
assert self.s2.recv() == "\x00secundus\x00trustno1"
+
+ def testPipelined2(self):
+ self.s1.mechanisms("ANONYMOUS")
+ self.s1.client()
+
+ out1 = self.t1.output(1024)
+ n = self.t2.input(out1)
+ assert n == len(out1)
+
+ self.s2.mechanisms("ANONYMOUS")
+ self.s2.server()
+ self.s2.done(SASL.OK)
+ c2 = Connection()
+ c2.open()
+ self.t2.bind(c2)
+
+ out2 = self.t2.output(1024)
+ n = self.t1.input(out2)
+ assert n == len(out2)
+
+ out1 = self.t1.output(1024)
+ assert len(out1) > 0
Modified: qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/ssl.py
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/ssl.py?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/ssl.py (original)
+++ qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/ssl.py Mon Nov 5 16:52:12 2012
@@ -209,4 +209,28 @@ class SslTest(common.Test):
server_conn.close()
self._pump()
+ def test_bad_server_certificate(self):
+ """ A server with a self-signed certificate that is not trusted by the
+ client. The client should reject the server.
+ """
+ self.server.set_credentials(self._testpath("bad-server-certificate.pem"),
+ self._testpath("bad-server-private-key.pem"),
+ "server-password")
+ self.server.set_peer_authentication( SSL.ANONYMOUS_PEER )
+
+ self.client.set_trusted_ca_db(self._testpath("ca-certificate.pem"))
+ self.client.set_peer_authentication( SSL.VERIFY_PEER )
+
+ client_conn = Connection()
+ self.t_client.bind(client_conn)
+ server_conn = Connection()
+ self.t_server.bind(server_conn)
+ client_conn.open()
+ server_conn.open()
+ try:
+ self._pump()
+ assert False, "Client failed to reject bad certificate."
+ except TransportException, e:
+ pass
+
Modified: qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/ssl_db/README.txt
URL: http://svn.apache.org/viewvc/qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/ssl_db/README.txt?rev=1405875&r1=1405874&r2=1405875&view=diff
==============================================================================
--- qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/ssl_db/README.txt (original)
+++ qpid/proton/branches/kgiusti-msvc2010/tests/proton_tests/ssl_db/README.txt Mon Nov 5 16:52:12 2012
@@ -16,6 +16,7 @@ the CA.
server-private-key.pem - encrypted key used to create server-certificate.pem. Password is
"server-password"
+bad-server-certificate.pem, bad-server-private-key.pem - a certificate/key that is not trusted by the client, for negative test.
These certificates have been created using the OpenSSL tool.
@@ -32,4 +33,5 @@ The following commands were used to crea
openssl req -newkey rsa:2048 -keyout client-private-key.pem -passout pass:client-password -out client-request.pem -subj "/O=Client/CN=127.0.0.1"
openssl x509 -req -in client-request.pem -CA ca-certificate.pem -CAkey ca-private-key.pem -CAcreateserial -passin pass:ca-password -days 99999 -out client-certificate.pem
-
+# Create a "bad" certificate - not signed by a trusted authority
+ openssl req -x509 -newkey rsa:2048 -keyout bad-server-private-key.pem -passout pass:server-password -out bad-server-certificate.pem -days 99999 -subj "/O=Not Trusted, Inc/CN=127.0.0.1"
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org