You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kw...@apache.org on 2013/01/30 16:32:51 UTC
svn commit: r1440470 - in /qpid/proton/branches/jni-binding: ./
proton-c/bindings/php/ proton-c/bindings/python/ proton-c/src/codec/
proton-c/src/engine/ proton-c/src/sasl/ proton-c/src/ssl/
proton-j/proton-api/src/main/resources/ proton-j/proton/src/m...
Author: kwall
Date: Wed Jan 30 15:32:51 2013
New Revision: 1440470
URL: http://svn.apache.org/viewvc?rev=1440470&view=rev
Log:
NO-JIRA: Merged latest from trunk (at rev r1440441) to this branch jni-binding (previously merged to 1437537)
with commands:
svn merge -r1437537:HEAD https://svn.apache.org/repos/asf/qpid/proton/trunk
--- Merging r1437538 through r1440441 into '.':
svn merge -r1437537:HEAD https://svn.apache.org/repos/asf/qpid/proton/trunk/tests/proton_tests/common.py tests/python/proton_tests/common.py
svn merge -r1437537:HEAD https://svn.apache.org/repos/asf/qpid/proton/trunk/tests/proton_tests/engine.py tests/python/proton_tests/engine.py
svn merge -r1437537:HEAD https://svn.apache.org/repos/asf/qpid/proton/trunk/proton-j/proton/src/main/scripts/proton.py proton-j/proton-api/src/main/resources/proton.py
Modified:
qpid/proton/branches/jni-binding/ (props changed)
qpid/proton/branches/jni-binding/proton-c/bindings/php/php.i
qpid/proton/branches/jni-binding/proton-c/bindings/python/proton.py
qpid/proton/branches/jni-binding/proton-c/bindings/python/python.i
qpid/proton/branches/jni-binding/proton-c/src/codec/codec.c
qpid/proton/branches/jni-binding/proton-c/src/engine/engine-internal.h
qpid/proton/branches/jni-binding/proton-c/src/engine/engine.c
qpid/proton/branches/jni-binding/proton-c/src/sasl/sasl.c
qpid/proton/branches/jni-binding/proton-c/src/ssl/openssl.c
qpid/proton/branches/jni-binding/proton-c/src/ssl/ssl-internal.h
qpid/proton/branches/jni-binding/proton-j/proton-api/src/main/resources/proton.py (contents, props changed)
qpid/proton/branches/jni-binding/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java
qpid/proton/branches/jni-binding/tests/python/proton_tests/common.py (contents, props changed)
qpid/proton/branches/jni-binding/tests/python/proton_tests/engine.py (contents, props changed)
qpid/proton/branches/jni-binding/tests/python/proton_tests/ssl.py (props changed)
Propchange: qpid/proton/branches/jni-binding/
------------------------------------------------------------------------------
Merged /qpid/proton/trunk:r1437538-1440456
Modified: qpid/proton/branches/jni-binding/proton-c/bindings/php/php.i
URL: http://svn.apache.org/viewvc/qpid/proton/branches/jni-binding/proton-c/bindings/php/php.i?rev=1440470&r1=1440469&r2=1440470&view=diff
==============================================================================
--- qpid/proton/branches/jni-binding/proton-c/bindings/php/php.i (original)
+++ qpid/proton/branches/jni-binding/proton-c/bindings/php/php.i Wed Jan 30 15:32:51 2013
@@ -278,6 +278,18 @@ pn_listener_t *pn_listener_fd(pn_driver_
}
%ignore pn_listener_context;
+%rename(pn_listener_set_context) wrap_pn_listener_set_context;
+%inline {
+ void wrap_pn_listener_set_context(pn_listener_t *l, void *PHP_CONTEXT) {
+ zval *old = pn_listener_context(l);
+ if (old) {
+ zval_ptr_dtor(&old); // drop the reference taken on input
+ }
+ pn_listener_set_context(l, PHP_CONTEXT);
+ }
+}
+%ignore pn_listener_set_context;
+
%rename(pn_listener_free) wrap_pn_listener_free;
%inline %{
void wrap_pn_listener_free(pn_listener_t *l) {
Modified: qpid/proton/branches/jni-binding/proton-c/bindings/python/proton.py
URL: http://svn.apache.org/viewvc/qpid/proton/branches/jni-binding/proton-c/bindings/python/proton.py?rev=1440470&r1=1440469&r2=1440470&view=diff
==============================================================================
--- qpid/proton/branches/jni-binding/proton-c/bindings/python/proton.py (original)
+++ qpid/proton/branches/jni-binding/proton-c/bindings/python/proton.py Wed Jan 30 15:32:51 2013
@@ -32,6 +32,10 @@ The proton APIs consist of the following
from cproton import *
import uuid
+try:
+ bytes()
+except NameError:
+ bytes = str
LANGUAGE = "C"
@@ -1831,6 +1835,10 @@ class Connection(Endpoint):
def work_head(self):
return wrap_delivery(pn_work_head(self._conn))
+ @property
+ def error(self):
+ return pn_error_code(pn_connection_error(self._conn))
+
class SessionException(ProtonException):
pass
@@ -1868,6 +1876,9 @@ class Session(Endpoint):
self._update_cond()
pn_session_close(self._ssn)
+ def next(self, mask):
+ return wrap_session(pn_session_next(self._ssn, mask))
+
@property
def state(self):
return pn_session_state(self._ssn)
@@ -1981,6 +1992,15 @@ class Link(Endpoint):
def next(self, mask):
return wrap_link(pn_link_next(self._link, mask))
+ @property
+ def is_sender(self):
+ return pn_link_is_sender(self._link)
+
+ @property
+ def is_receiver(self):
+ return pn_link_is_receiver(self._link)
+
+
class Terminus(object):
UNSPECIFIED = PN_UNSPECIFIED
@@ -2138,6 +2158,11 @@ class Delivery(object):
def work_next(self):
return wrap_delivery(pn_work_next(self._dlv))
+ @property
+ def link(self):
+ return wrap_link(pn_delivery_link(self._dlv))
+
+
class TransportException(ProtonException):
pass
@@ -2147,12 +2172,17 @@ class Transport(object):
TRACE_FRM = PN_TRACE_FRM
TRACE_RAW = PN_TRACE_RAW
- def __init__(self):
- self._trans = pn_transport()
+ def __init__(self, _trans=None):
+ if not _trans:
+ self._trans = pn_transport()
+ else:
+ self._shared_trans = True
+ self._trans = _trans
def __del__(self):
if hasattr(self, "_trans"):
- pn_transport_free(self._trans)
+ if not hasattr(self, "_shared_trans"):
+ pn_transport_free(self._trans)
del self._trans
def _check(self, err):
@@ -2283,6 +2313,17 @@ class SASL(object):
def done(self, outcome):
pn_sasl_done(self._sasl, outcome)
+ STATE_CONF = PN_SASL_CONF
+ STATE_IDLE = PN_SASL_IDLE
+ STATE_STEP = PN_SASL_STEP
+ STATE_PASS = PN_SASL_PASS
+ STATE_FAIL = PN_SASL_FAIL
+
+ @property
+ def state(self):
+ return pn_sasl_state(self._sasl)
+
+
class SSLException(TransportException):
pass
@@ -2384,6 +2425,127 @@ class SSLSessionDetails(object):
def get_session_id(self):
return self._session_id
+
+###
+# Driver
+###
+
+class DriverException(ProtonException):
+ """
+ The DriverException class is the root of the driver exception hierarchy.
+ """
+ pass
+
+
+def wrap_connector(cxtr):
+ if not cxtr: return None
+ ctx = pn_connector_context(cxtr)
+ if ctx: return ctx
+ wrapper = Connector(_cxtr=cxtr)
+ pn_connector_set_context(cxtr, wrapper)
+ return wrapper
+
+class Connector(object):
+ def __init__(self, _cxtr):
+ self._cxtr = _cxtr
+
+ def next(self):
+ return wrap_connector(pn_connector_next(self._cxtr))
+
+ def process(self):
+ pn_connector_process(self._cxtr)
+
+ def listener(self):
+ return wrap_listener(pn_connector_listener(self._cxtr))
+
+ def sasl(self):
+ ## seems easier just to grab the SASL associated with the transport:
+ trans = self.transport
+ if trans:
+ return SASL(self.transport)
+ return None
+
+ @property
+ def transport(self):
+ trans = pn_connector_transport(self._cxtr)
+ if trans:
+ return Transport(trans)
+ return None
+
+ def close(self):
+ return pn_connector_close(self._cxtr)
+
+ @property
+ def closed(self):
+ return pn_connector_closed(self._cxtr)
+
+ def _get_connection(self):
+ return wrap_connection(pn_connector_connection(self._cxtr))
+
+ def _set_connection(self, conn):
+ pn_connector_set_connection(self._cxtr, conn._conn)
+
+ connection = property(_get_connection, _set_connection,
+ doc="""
+Associate a Connection with this Connector.
+""")
+
+def wrap_listener(lsnr):
+ if not lsnr: return None
+ ctx = pn_listener_context(lsnr)
+ if ctx: return ctx
+ wrapper = Listener(_lsnr=lsnr)
+ pn_listener_set_context(lsnr, wrapper)
+ return wrapper
+
+class Listener(object):
+ def __init__(self, _lsnr=None):
+ self._lsnr = _lsnr
+
+ def next(self):
+ return wrap_listener(pn_listener_next(self._lsnr))
+
+ def accept(self):
+ cxtr = pn_listener_accept(self._lsnr)
+ return wrap_connector(cxtr)
+
+ def close(self):
+ pn_listener_close(self._lsnr)
+
+class Driver(object):
+ def __init__(self):
+ self._driver = pn_driver()
+
+ def __del__(self):
+ if hasattr(self, "_driver"):
+ pn_driver_free(self._driver)
+ del self._driver
+
+ def wait(self, timeout):
+ return pn_driver_wait(self._driver, timeout)
+
+ def wakeup(self):
+ return pn_driver_wakeup(self._driver)
+
+ def listener(self, host, port):
+ return wrap_listener(pn_listener(self._driver, host, port, None))
+
+ def pending_listener(self):
+ return wrap_listener(pn_driver_listener(self._driver))
+
+ def head_listener(self):
+ return wrap_listener(pn_listener_head(self._driver))
+
+ def connector(self, host, port):
+ return wrap_connector(pn_connector(self._driver, host, port, None))
+
+ def head_connector(self):
+ return wrap_connector(pn_connector_head(self._driver))
+
+ def pending_connector(self):
+ return wrap_connector(pn_driver_connector(self._driver))
+
+
__all__ = [
"LANGUAGE",
"PN_SESSION_WINDOW",
@@ -2396,11 +2558,15 @@ __all__ = [
"Array",
"Condition",
"Connection",
+ "Connector",
"Data",
"Delivery",
"Described",
+ "Driver",
+ "DriverException",
"Endpoint",
"Link",
+ "Listener",
"Message",
"MessageException",
"Messenger",
Modified: qpid/proton/branches/jni-binding/proton-c/bindings/python/python.i
URL: http://svn.apache.org/viewvc/qpid/proton/branches/jni-binding/proton-c/bindings/python/python.i?rev=1440470&r1=1440469&r2=1440470&view=diff
==============================================================================
--- qpid/proton/branches/jni-binding/proton-c/bindings/python/python.i (original)
+++ qpid/proton/branches/jni-binding/proton-c/bindings/python/python.i Wed Jan 30 15:32:51 2013
@@ -193,6 +193,16 @@ ssize_t pn_transport_input(pn_transport_
}
%ignore pn_listener_context;
+%rename(pn_listener_set_context) wrap_pn_listener_set_context;
+%inline {
+ void wrap_pn_listener_set_context(pn_listener_t *l, PyObject *context) {
+ Py_XDECREF((PyObject *)pn_listener_context(l));
+ Py_XINCREF(context);
+ pn_listener_set_context(l, context);
+ }
+}
+%ignore pn_listener_set_context;
+
%rename(pn_listener_free) wrap_pn_listener_free;
%inline %{
void wrap_pn_listener_free(pn_listener_t *l) {
Modified: qpid/proton/branches/jni-binding/proton-c/src/codec/codec.c
URL: http://svn.apache.org/viewvc/qpid/proton/branches/jni-binding/proton-c/src/codec/codec.c?rev=1440470&r1=1440469&r2=1440470&view=diff
==============================================================================
--- qpid/proton/branches/jni-binding/proton-c/src/codec/codec.c (original)
+++ qpid/proton/branches/jni-binding/proton-c/src/codec/codec.c Wed Jan 30 15:32:51 2013
@@ -1353,10 +1353,10 @@ int pn_data_vscan(pn_data_t *data, const
while (*fmt) {
char code = *(fmt++);
- bool found;
+ bool found = false;
pn_type_t type;
- bool scanned;
+ bool scanned = false;
bool suspend = resume_count > 0;
switch (code) {
Modified: qpid/proton/branches/jni-binding/proton-c/src/engine/engine-internal.h
URL: http://svn.apache.org/viewvc/qpid/proton/branches/jni-binding/proton-c/src/engine/engine-internal.h?rev=1440470&r1=1440469&r2=1440470&view=diff
==============================================================================
--- qpid/proton/branches/jni-binding/proton-c/src/engine/engine-internal.h (original)
+++ qpid/proton/branches/jni-binding/proton-c/src/engine/engine-internal.h Wed Jan 30 15:32:51 2013
@@ -107,10 +107,15 @@ typedef struct {
#include <proton/sasl.h>
#include <proton/ssl.h>
+typedef struct pn_io_layer_t {
+ void *context;
+ struct pn_io_layer_t *next;
+ ssize_t (*process_input)(struct pn_io_layer_t *io_layer, const char *, size_t);
+ ssize_t (*process_output)(struct pn_io_layer_t *io_layer, char *, size_t);
+ pn_timestamp_t (*process_tick)(struct pn_io_layer_t *io_layer, pn_timestamp_t);
+} pn_io_layer_t;
+
struct pn_transport_t {
- ssize_t (*process_input)(pn_transport_t *, const char *, size_t);
- ssize_t (*process_output)(pn_transport_t *, char *, size_t);
- pn_timestamp_t (*process_tick)(pn_transport_t *, pn_timestamp_t);
size_t header_count;
pn_sasl_t *sasl;
pn_ssl_t *ssl;
@@ -128,6 +133,12 @@ struct pn_transport_t {
uint32_t remote_max_frame;
pn_condition_t remote_condition;
+#define PN_IO_SSL 0
+#define PN_IO_SASL 1
+#define PN_IO_AMQP 2
+#define PN_IO_LAYER_CT (PN_IO_AMQP+1)
+ pn_io_layer_t io_layers[PN_IO_LAYER_CT];
+
/* dead remote detection */
pn_millis_t local_idle_timeout;
pn_timestamp_t dead_remote_deadline;
@@ -253,4 +264,8 @@ void pn_link_dump(pn_link_t *link);
void pn_dump(pn_connection_t *conn);
void pn_transport_sasl_init(pn_transport_t *transport);
+ssize_t pn_io_layer_input_passthru(pn_io_layer_t *, const char *, size_t );
+ssize_t pn_io_layer_output_passthru(pn_io_layer_t *, char *, size_t );
+pn_timestamp_t pn_io_layer_tick_passthru(pn_io_layer_t *, pn_timestamp_t);
+
#endif /* engine-internal.h */
Modified: qpid/proton/branches/jni-binding/proton-c/src/engine/engine.c
URL: http://svn.apache.org/viewvc/qpid/proton/branches/jni-binding/proton-c/src/engine/engine.c?rev=1440470&r1=1440469&r2=1440470&view=diff
==============================================================================
--- qpid/proton/branches/jni-binding/proton-c/src/engine/engine.c (original)
+++ qpid/proton/branches/jni-binding/proton-c/src/engine/engine.c Wed Jan 30 15:32:51 2013
@@ -728,26 +728,36 @@ int pn_do_detach(pn_dispatcher_t *disp);
int pn_do_end(pn_dispatcher_t *disp);
int pn_do_close(pn_dispatcher_t *disp);
-static ssize_t pn_input_read_sasl_header(pn_transport_t *transport, const char *bytes, size_t available);
-static ssize_t pn_input_read_sasl(pn_transport_t *transport, const char *bytes, size_t available);
-static ssize_t pn_input_read_amqp_header(pn_transport_t *transport, const char *bytes, size_t available);
-static ssize_t pn_input_read_amqp(pn_transport_t *transport, const char *bytes, size_t available);
-static ssize_t pn_output_write_sasl_header(pn_transport_t *transport, char *bytes, size_t available);
-static ssize_t pn_output_write_sasl(pn_transport_t *transport, char *bytes, size_t available);
-static ssize_t pn_output_write_amqp_header(pn_transport_t *transport, char *bytes, size_t available);
-static ssize_t pn_output_write_amqp(pn_transport_t *transport, char *bytes, size_t available);
-static pn_timestamp_t pn_process_tick(pn_transport_t *transport, pn_timestamp_t now);
+static ssize_t pn_input_read_amqp_header(pn_io_layer_t *io_layer, const char *bytes, size_t available);
+static ssize_t pn_input_read_amqp(pn_io_layer_t *io_layer, const char *bytes, size_t available);
+static ssize_t pn_output_write_amqp_header(pn_io_layer_t *io_layer, char *bytes, size_t available);
+static ssize_t pn_output_write_amqp(pn_io_layer_t *io_layer, char *bytes, size_t available);
+static pn_timestamp_t pn_tick_amqp(pn_io_layer_t *io_layer, pn_timestamp_t now);
void pn_transport_init(pn_transport_t *transport)
{
- transport->process_input = pn_input_read_amqp_header;
- transport->process_output = pn_output_write_amqp_header;
- transport->process_tick = NULL;
transport->header_count = 0;
transport->sasl = NULL;
transport->ssl = NULL;
transport->disp = pn_dispatcher(0, transport);
+ pn_io_layer_t *io_layer = transport->io_layers;
+ while (io_layer != &transport->io_layers[PN_IO_AMQP]) {
+ io_layer->context = NULL;
+ io_layer->next = io_layer + 1;
+ io_layer->process_input = pn_io_layer_input_passthru;
+ io_layer->process_output = pn_io_layer_output_passthru;
+ io_layer->process_tick = pn_io_layer_tick_passthru;
+ ++io_layer;
+ }
+
+ pn_io_layer_t *amqp = &transport->io_layers[PN_IO_AMQP];
+ amqp->context = transport;
+ amqp->process_input = pn_input_read_amqp_header;
+ amqp->process_output = pn_output_write_amqp_header;
+ amqp->process_tick = pn_io_layer_tick_passthru;
+ amqp->next = NULL;
+
pn_dispatcher_action(transport->disp, OPEN, "OPEN", pn_do_open);
pn_dispatcher_action(transport->disp, BEGIN, "BEGIN", pn_do_begin);
pn_dispatcher_action(transport->disp, ATTACH, "ATTACH", pn_do_attach);
@@ -827,12 +837,6 @@ pn_transport_t *pn_transport()
return transport;
}
-void pn_transport_sasl_init(pn_transport_t *transport)
-{
- transport->process_input = pn_input_read_sasl_header;
- transport->process_output = pn_output_write_sasl_header;
-}
-
int pn_transport_bind(pn_transport_t *transport, pn_connection_t *connection)
{
if (!transport) return PN_ARG_ERR;
@@ -1422,7 +1426,7 @@ int pn_do_open(pn_dispatcher_t *disp)
transport->disp->halt = true;
}
if (transport->remote_idle_timeout)
- transport->process_tick = pn_process_tick; // enable timeouts
+ transport->io_layers[PN_IO_AMQP].process_tick = pn_tick_amqp; // enable timeouts
transport->open_rcvd = true;
return 0;
}
@@ -1812,15 +1816,12 @@ ssize_t pn_transport_input(pn_transport_
{
if (!transport) return PN_ARG_ERR;
+ pn_io_layer_t *io_layer = transport->io_layers;
size_t consumed = 0;
- const bool use_ssl = transport->ssl != NULL;
while (true) {
ssize_t n;
- if (use_ssl)
- n = pn_ssl_input( transport->ssl, bytes + consumed, available - consumed);
- else
- n = transport->process_input(transport, bytes + consumed, available - consumed);
+ n = io_layer->process_input( io_layer, bytes + consumed, available - consumed);
if (n > 0) {
consumed += n;
if (consumed >= available) {
@@ -1844,11 +1845,9 @@ ssize_t pn_transport_input(pn_transport_
return consumed;
}
-#define SASL_HEADER ("AMQP\x03\x01\x00\x00")
-
static ssize_t pn_input_read_header(pn_transport_t *transport, const char *bytes, size_t available,
const char *header, size_t size, const char *protocol,
- ssize_t (*next)(pn_transport_t *, const char *, size_t))
+ ssize_t (*next)(pn_io_layer_t *, const char *, size_t))
{
const char *point = header + transport->header_count;
int delta = pn_min(available, size - transport->header_count);
@@ -1861,7 +1860,7 @@ static ssize_t pn_input_read_header(pn_t
transport->header_count += delta;
if (transport->header_count == size) {
transport->header_count = 0;
- transport->process_input = next;
+ transport->io_layers[PN_IO_AMQP].process_input = next;
if (transport->disp->trace & PN_TRACE_FRM)
fprintf(stderr, " <- %s\n", protocol);
@@ -1870,33 +1869,18 @@ static ssize_t pn_input_read_header(pn_t
}
}
-static ssize_t pn_input_read_sasl_header(pn_transport_t *transport, const char *bytes, size_t available)
-{
- return pn_input_read_header(transport, bytes, available, SASL_HEADER, 8, "SASL", pn_input_read_sasl);
-}
-
-static ssize_t pn_input_read_sasl(pn_transport_t *transport, const char *bytes, size_t available)
-{
- pn_sasl_t *sasl = transport->sasl;
- ssize_t n = pn_sasl_input(sasl, bytes, available);
- if (n == PN_EOS) {
- transport->process_input = pn_input_read_amqp_header;
- return transport->process_input(transport, bytes, available);
- } else {
- return n;
- }
-}
-
#define AMQP_HEADER ("AMQP\x00\x01\x00\x00")
-static ssize_t pn_input_read_amqp_header(pn_transport_t *transport, const char *bytes, size_t available)
+static ssize_t pn_input_read_amqp_header(pn_io_layer_t *io_layer, const char *bytes, size_t available)
{
+ pn_transport_t *transport = (pn_transport_t *)io_layer->context;
return pn_input_read_header(transport, bytes, available, AMQP_HEADER, 8,
"AMQP", pn_input_read_amqp);
}
-static ssize_t pn_input_read_amqp(pn_transport_t *transport, const char *bytes, size_t available)
+static ssize_t pn_input_read_amqp(pn_io_layer_t *io_layer, const char *bytes, size_t available)
{
+ pn_transport_t *transport = (pn_transport_t *)io_layer->context;
if (transport->close_rcvd) {
if (available > 0) {
pn_do_error(transport, "amqp:connection:framing-error", "data after close");
@@ -1923,9 +1907,10 @@ static ssize_t pn_input_read_amqp(pn_tra
}
/* process AMQP related timer events */
-static pn_timestamp_t pn_process_tick(pn_transport_t *transport, pn_timestamp_t now)
+static pn_timestamp_t pn_tick_amqp(pn_io_layer_t *io_layer, pn_timestamp_t now)
{
pn_timestamp_t timeout = 0;
+ pn_transport_t *transport = (pn_transport_t *)io_layer->context;
if (transport->local_idle_timeout) {
if (transport->dead_remote_deadline == 0 ||
@@ -2492,45 +2477,29 @@ static ssize_t pn_output_write_header(pn
char *bytes, size_t size,
const char *header, size_t hdrsize,
const char *protocol,
- ssize_t (*next)(pn_transport_t *, char *, size_t))
+ ssize_t (*next)(pn_io_layer_t *, char *, size_t))
{
if (transport->disp->trace & PN_TRACE_FRM)
fprintf(stderr, " -> %s\n", protocol);
if (size >= hdrsize) {
memmove(bytes, header, hdrsize);
- transport->process_output = next;
+ transport->io_layers[PN_IO_AMQP].process_output = next;
return hdrsize;
} else {
return pn_error_format(transport->error, PN_UNDERFLOW, "underflow writing %s header", protocol);
}
}
-static ssize_t pn_output_write_sasl_header(pn_transport_t *transport, char *bytes, size_t size)
-{
- return pn_output_write_header(transport, bytes, size, SASL_HEADER, 8, "SASL",
- pn_output_write_sasl);
-}
-
-static ssize_t pn_output_write_sasl(pn_transport_t *transport, char *bytes, size_t size)
-{
- pn_sasl_t *sasl = transport->sasl;
- ssize_t n = pn_sasl_output(sasl, bytes, size);
- if (n == PN_EOS) {
- transport->process_output = pn_output_write_amqp_header;
- return transport->process_output(transport, bytes, size);
- } else {
- return n;
- }
-}
-
-static ssize_t pn_output_write_amqp_header(pn_transport_t *transport, char *bytes, size_t size)
+static ssize_t pn_output_write_amqp_header(pn_io_layer_t *io_layer, char *bytes, size_t size)
{
+ pn_transport_t *transport = (pn_transport_t *)io_layer->context;
return pn_output_write_header(transport, bytes, size, AMQP_HEADER, 8, "AMQP",
pn_output_write_amqp);
}
-static ssize_t pn_output_write_amqp(pn_transport_t *transport, char *bytes, size_t size)
+static ssize_t pn_output_write_amqp(pn_io_layer_t *io_layer, char *bytes, size_t size)
{
+ pn_transport_t *transport = (pn_transport_t *)io_layer->context;
if (!transport->connection) {
return 0;
}
@@ -2553,32 +2522,26 @@ ssize_t pn_transport_output(pn_transport
{
if (!transport) return PN_ARG_ERR;
+ pn_io_layer_t *io_layer = transport->io_layers;
size_t total = 0;
- const bool use_ssl = transport->ssl != NULL;
-
while (size - total > 0) {
ssize_t n;
- if (use_ssl)
- n = pn_ssl_output( transport->ssl, bytes + total, size - total);
- else
- n = transport->process_output(transport, bytes + total, size - total);
+ n = io_layer->process_output( io_layer, bytes + total, size - total);
if (n > 0) {
total += n;
} else if (n == 0) {
break;
- } else if (n == PN_EOS) {
- if (total > 0) {
- break;
- } else {
- if (transport->disp->trace & (PN_TRACE_RAW | PN_TRACE_FRM))
+ } else {
+ if (total > 0)
+ break; // return what was output
+ if (transport->disp->trace & (PN_TRACE_RAW | PN_TRACE_FRM)) {
+ if (n == PN_EOS)
pn_dispatcher_trace(transport->disp, 0, "-> EOS\n");
- return PN_EOS;
+ else
+ pn_dispatcher_trace(transport->disp, 0, "-> EOS (%zi) %s\n", n,
+ pn_error_text(transport->error));
}
- } else {
- if (transport->disp->trace & (PN_TRACE_RAW | PN_TRACE_FRM))
- pn_dispatcher_trace(transport->disp, 0, "-> EOS (%zi) %s\n", n,
- pn_error_text(transport->error));
return n;
}
}
@@ -2619,7 +2582,7 @@ pn_millis_t pn_transport_get_idle_timeou
void pn_transport_set_idle_timeout(pn_transport_t *transport, pn_millis_t timeout)
{
transport->local_idle_timeout = timeout;
- transport->process_tick = pn_process_tick;
+ transport->io_layers[PN_IO_AMQP].process_tick = pn_tick_amqp;
}
pn_millis_t pn_transport_get_remote_idle_timeout(pn_transport_t *transport)
@@ -2687,9 +2650,8 @@ void pn_link_drain(pn_link_t *receiver,
pn_timestamp_t pn_transport_tick(pn_transport_t *transport, pn_timestamp_t now)
{
- if (transport && transport->process_tick)
- return transport->process_tick( transport, now );
- return 0;
+ pn_io_layer_t *io_layer = transport->io_layers;
+ return io_layer->process_tick( io_layer, now );
}
uint64_t pn_transport_get_frames_output(const pn_transport_t *transport)
@@ -2906,3 +2868,30 @@ int pn_condition_redirect_port(pn_condit
pn_data_rewind(data);
return port;
}
+
+/** Pass through input handler */
+ssize_t pn_io_layer_input_passthru(pn_io_layer_t *io_layer, const char *data, size_t available)
+{
+ pn_io_layer_t *next = io_layer->next;
+ if (next)
+ return next->process_input( next, data, available );
+ return PN_EOS;
+}
+
+/** Pass through output handler */
+ssize_t pn_io_layer_output_passthru(pn_io_layer_t *io_layer, char *bytes, size_t size)
+{
+ pn_io_layer_t *next = io_layer->next;
+ if (next)
+ return next->process_output( next, bytes, size );
+ return PN_EOS;
+}
+
+/** Pass through tick handler */
+pn_timestamp_t pn_io_layer_tick_passthru(pn_io_layer_t *io_layer, pn_timestamp_t now)
+{
+ pn_io_layer_t *next = io_layer->next;
+ if (next)
+ return next->process_tick( next, now );
+ return 0;
+}
Modified: qpid/proton/branches/jni-binding/proton-c/src/sasl/sasl.c
URL: http://svn.apache.org/viewvc/qpid/proton/branches/jni-binding/proton-c/src/sasl/sasl.c?rev=1440470&r1=1440469&r2=1440470&view=diff
==============================================================================
--- qpid/proton/branches/jni-binding/proton-c/src/sasl/sasl.c (original)
+++ qpid/proton/branches/jni-binding/proton-c/src/sasl/sasl.c Wed Jan 30 15:32:51 2013
@@ -34,6 +34,9 @@
#define SCRATCH (1024)
struct pn_sasl_t {
+ pn_transport_t *transport;
+ pn_io_layer_t *io_layer;
+ size_t header_count;
pn_dispatcher_t *disp;
bool client;
bool configured;
@@ -49,6 +52,11 @@ struct pn_sasl_t {
char scratch[SCRATCH];
};
+static ssize_t pn_input_read_sasl_header(pn_io_layer_t *io_layer, const char *bytes, size_t available);
+static ssize_t pn_input_read_sasl(pn_io_layer_t *io_layer, const char *bytes, size_t available);
+static ssize_t pn_output_write_sasl_header(pn_io_layer_t *io_layer, char *bytes, size_t available);
+static ssize_t pn_output_write_sasl(pn_io_layer_t *io_layer, char *bytes, size_t available);
+
int pn_do_init(pn_dispatcher_t *disp);
int pn_do_mechanisms(pn_dispatcher_t *disp);
int pn_do_challenge(pn_dispatcher_t *disp);
@@ -81,8 +89,14 @@ pn_sasl_t *pn_sasl(pn_transport_t *trans
sasl->rcvd_done = false;
transport->sasl = sasl;
+ sasl->transport = transport;
+ sasl->io_layer = &transport->io_layers[PN_IO_SASL];
+ sasl->io_layer->context = sasl;
+ sasl->io_layer->process_input = pn_input_read_sasl_header;
+ sasl->io_layer->process_output = pn_output_write_sasl_header;
+ sasl->io_layer->process_tick = pn_io_layer_tick_passthru;
- pn_transport_sasl_init(transport);
+ sasl->header_count = 0;
}
return transport->sasl;
@@ -300,7 +314,7 @@ void pn_sasl_process(pn_sasl_t *sasl)
}
}
-ssize_t pn_sasl_input(pn_sasl_t *sasl, char *bytes, size_t available)
+ssize_t pn_sasl_input(pn_sasl_t *sasl, const char *bytes, size_t available)
{
ssize_t n = pn_dispatcher_input(sasl->disp, bytes, available);
if (n < 0) return n;
@@ -391,3 +405,66 @@ int pn_do_outcome(pn_dispatcher_t *disp)
disp->halt = true;
return 0;
}
+
+#define SASL_HEADER ("AMQP\x03\x01\x00\x00")
+#define SASL_HEADER_LEN 8
+
+static ssize_t pn_input_read_sasl_header(pn_io_layer_t *io_layer, const char *bytes, size_t available)
+{
+ pn_sasl_t *sasl = (pn_sasl_t *)io_layer->context;
+ const char *point = SASL_HEADER + sasl->header_count;
+ int delta = pn_min(available, SASL_HEADER_LEN - sasl->header_count);
+ if (!available || memcmp(bytes, point, delta)) {
+ char quoted[1024];
+ pn_quote_data(quoted, 1024, bytes, available);
+ return pn_error_format(sasl->transport->error, PN_ERR,
+ "%s header mismatch: '%s'", "SASL", quoted);
+ } else {
+ sasl->header_count += delta;
+ if (sasl->header_count == SASL_HEADER_LEN) {
+ sasl->io_layer->process_input = pn_input_read_sasl;
+ if (sasl->disp->trace & PN_TRACE_FRM)
+ fprintf(stderr, " <- %s\n", "SASL");
+ }
+ return delta;
+ }
+}
+
+static ssize_t pn_input_read_sasl(pn_io_layer_t *io_layer, const char *bytes, size_t available)
+{
+ pn_sasl_t *sasl = (pn_sasl_t *)io_layer->context;
+ ssize_t n = pn_sasl_input(sasl, bytes, available);
+ if (n == PN_EOS) {
+ sasl->io_layer->process_input = pn_io_layer_input_passthru;
+ pn_io_layer_t *io_next = sasl->io_layer->next;
+ return io_next->process_input( io_next, bytes, available );
+ }
+ return n;
+}
+
+static ssize_t pn_output_write_sasl_header(pn_io_layer_t *io_layer, char *bytes, size_t size)
+{
+ pn_sasl_t *sasl = (pn_sasl_t *)io_layer->context;
+ if (sasl->disp->trace & PN_TRACE_FRM)
+ fprintf(stderr, " -> %s\n", "SASL");
+ if (size >= SASL_HEADER_LEN) {
+ memmove(bytes, SASL_HEADER, SASL_HEADER_LEN);
+ sasl->io_layer->process_output = pn_output_write_sasl;
+ return SASL_HEADER_LEN;
+ } else {
+ return pn_error_format(sasl->transport->error, PN_UNDERFLOW, "underflow writing %s header", "SASL");
+ }
+}
+
+static ssize_t pn_output_write_sasl(pn_io_layer_t *io_layer, char *bytes, size_t size)
+{
+ pn_sasl_t *sasl = (pn_sasl_t *)io_layer->context;
+ ssize_t n = pn_sasl_output(sasl, bytes, size);
+ if (n == PN_EOS) {
+ sasl->io_layer->process_output = pn_io_layer_output_passthru;
+ pn_io_layer_t *io_next = sasl->io_layer->next;
+ return io_next->process_output( io_next, bytes, size );
+ }
+ return n;
+}
+
Modified: qpid/proton/branches/jni-binding/proton-c/src/ssl/openssl.c
URL: http://svn.apache.org/viewvc/qpid/proton/branches/jni-binding/proton-c/src/ssl/openssl.c?rev=1440470&r1=1440469&r2=1440470&view=diff
==============================================================================
--- qpid/proton/branches/jni-binding/proton-c/src/ssl/openssl.c (original)
+++ qpid/proton/branches/jni-binding/proton-c/src/ssl/openssl.c Wed Jan 30 15:32:51 2013
@@ -72,6 +72,7 @@ struct pn_ssl_domain_t {
struct pn_ssl_t {
pn_transport_t *transport;
+ pn_io_layer_t *io_layer;
pn_ssl_domain_t *domain;
const char *session_id;
const char *peer_hostname;
@@ -95,10 +96,6 @@ struct pn_ssl_t {
char inbuf[APP_BUF_SIZE];
size_t in_count;
- // process cleartext i/o "above" the SSL layer
- ssize_t (*process_input)(pn_transport_t *, const char *, size_t);
- ssize_t (*process_output)(pn_transport_t *, char *, size_t);
-
pn_trace_t trace;
};
@@ -117,14 +114,12 @@ struct pn_ssl_session_t {
/* */
static int keyfile_pw_cb(char *buf, int size, int rwflag, void *userdata);
-static ssize_t process_input_ssl( pn_transport_t *transport, const char *input_data, size_t len);
-static ssize_t process_output_ssl( pn_transport_t *transport, char *input_data, size_t len);
-static ssize_t process_input_cleartext(pn_transport_t *transport, const char *input_data, size_t len);
-static ssize_t process_output_cleartext(pn_transport_t *transport, char *buffer, size_t max_len);
-static ssize_t process_input_unknown(pn_transport_t *transport, const char *input_data, size_t len);
-static ssize_t process_output_unknown(pn_transport_t *transport, char *input_data, size_t len);
-static ssize_t process_input_done(pn_transport_t *transport, const char *input_data, size_t len);
-static ssize_t process_output_done(pn_transport_t *transport, char *input_data, size_t len);
+static ssize_t process_input_ssl( pn_io_layer_t *io_layer, const char *input_data, size_t len);
+static ssize_t process_output_ssl( pn_io_layer_t *io_layer, char *input_data, size_t len);
+static ssize_t process_input_unknown( pn_io_layer_t *io_layer, const char *input_data, size_t len);
+static ssize_t process_output_unknown( pn_io_layer_t *io_layer, char *input_data, size_t len);
+static ssize_t process_input_done(pn_io_layer_t *io_layer, const char *input_data, size_t len);
+static ssize_t process_output_done(pn_io_layer_t *io_layer, char *input_data, size_t len);
static connection_mode_t check_for_ssl_connection( const char *data, size_t len );
static int init_ssl_socket( pn_ssl_t * );
static void release_ssl_socket( pn_ssl_t * );
@@ -643,11 +638,11 @@ int pn_ssl_init( pn_ssl_t *ssl, pn_ssl_d
ssl->domain = domain;
domain->ref_count++;
if (domain->allow_unsecured) {
- ssl->process_input = process_input_unknown;
- ssl->process_output = process_output_unknown;
+ ssl->io_layer->process_input = process_input_unknown;
+ ssl->io_layer->process_output = process_output_unknown;
} else {
- ssl->process_input = process_input_ssl;
- ssl->process_output = process_output_ssl;
+ ssl->io_layer->process_input = process_input_ssl;
+ ssl->io_layer->process_output = process_output_ssl;
}
if (session_id && domain->mode == PN_SSL_MODE_CLIENT)
@@ -712,19 +707,6 @@ void pn_ssl_free( pn_ssl_t *ssl)
free(ssl);
}
-// move data received from the network into the SSL layer
-ssize_t pn_ssl_input(pn_ssl_t *ssl, const char *bytes, size_t available)
-{
- return ssl->process_input( ssl->transport, bytes, available );
-}
-
-// pull output from the SSL layer and move into network output buffers
-ssize_t pn_ssl_output(pn_ssl_t *ssl, char *buffer, size_t max_size)
-{
- return ssl->process_output( ssl->transport, buffer, max_size );
-}
-
-
pn_ssl_t *pn_ssl(pn_transport_t *transport)
{
if (!transport) return NULL;
@@ -735,6 +717,12 @@ pn_ssl_t *pn_ssl(pn_transport_t *transpo
ssl->transport = transport;
transport->ssl = ssl;
+ ssl->io_layer = &transport->io_layers[PN_IO_SSL];
+ ssl->io_layer->context = ssl;
+ ssl->io_layer->process_input = pn_io_layer_input_passthru;
+ ssl->io_layer->process_output = pn_io_layer_output_passthru;
+ ssl->io_layer->process_tick = pn_io_layer_tick_passthru;
+
ssl->trace = (transport->disp) ? transport->disp->trace : PN_TRACE_OFF;
return ssl;
@@ -780,8 +768,8 @@ static int start_ssl_shutdown( pn_ssl_t
static int setup_ssl_connection( pn_ssl_t *ssl )
{
_log( ssl, "SSL connection detected.\n");
- ssl->process_input = process_input_ssl;
- ssl->process_output = process_output_ssl;
+ ssl->io_layer->process_input = process_input_ssl;
+ ssl->io_layer->process_output = process_output_ssl;
return 0;
}
@@ -790,10 +778,9 @@ static int setup_ssl_connection( pn_ssl_
// take data from the network, and pass it into SSL. Attempt to read decrypted data from
// SSL socket and pass it to the application.
-static ssize_t process_input_ssl( pn_transport_t *transport, const char *input_data, size_t available)
+static ssize_t process_input_ssl( pn_io_layer_t *io_layer, const char *input_data, size_t available)
{
- pn_ssl_t *ssl = transport->ssl;
- if (!ssl) return PN_ERR;
+ pn_ssl_t *ssl = (pn_ssl_t *)io_layer->context;
if (ssl->ssl == NULL && init_ssl_socket(ssl)) return PN_ERR;
_log( ssl, "process_input_ssl( data size=%d )\n",available );
@@ -866,7 +853,8 @@ static ssize_t process_input_ssl( pn_tra
if (!ssl->app_input_closed) {
char *data = ssl->inbuf;
if (ssl->in_count > 0 || ssl->ssl_closed) { /* if ssl_closed, send 0 count */
- ssize_t consumed = transport->process_input(transport, data, ssl->in_count);
+ pn_io_layer_t *io_next = ssl->io_layer->next;
+ ssize_t consumed = io_next->process_input( io_next, data, ssl->in_count);
if (consumed > 0) {
ssl->in_count -= consumed;
data += consumed;
@@ -906,15 +894,15 @@ static ssize_t process_input_ssl( pn_tra
//}
if (ssl->app_input_closed && (SSL_get_shutdown(ssl->ssl) & SSL_SENT_SHUTDOWN) ) {
consumed = ssl->app_input_closed;
- ssl->process_input = process_input_done;
+ ssl->io_layer->process_input = process_input_done;
}
_log(ssl, "process_input_ssl() returning %d\n", (int) consumed);
return consumed;
}
-static ssize_t process_output_ssl( pn_transport_t *transport, char *buffer, size_t max_len)
+static ssize_t process_output_ssl( pn_io_layer_t *io_layer, char *buffer, size_t max_len)
{
- pn_ssl_t *ssl = transport->ssl;
+ pn_ssl_t *ssl = (pn_ssl_t *)io_layer->context;
if (!ssl) return PN_ERR;
if (ssl->ssl == NULL && init_ssl_socket(ssl)) return PN_ERR;
@@ -926,7 +914,8 @@ static ssize_t process_output_ssl( pn_tr
// first, get any pending application output, if possible
if (!ssl->app_output_closed && ssl->out_count < APP_BUF_SIZE) {
- ssize_t app_bytes = transport->process_output(transport, &ssl->outbuf[ssl->out_count], APP_BUF_SIZE - ssl->out_count);
+ pn_io_layer_t *io_next = ssl->io_layer->next;
+ ssize_t app_bytes = io_next->process_output( io_next, &ssl->outbuf[ssl->out_count], APP_BUF_SIZE - ssl->out_count);
if (app_bytes > 0) {
ssl->out_count += app_bytes;
work_pending = true;
@@ -1018,7 +1007,7 @@ static ssize_t process_output_ssl( pn_tr
//}
if (written == 0 && (SSL_get_shutdown(ssl->ssl) & SSL_SENT_SHUTDOWN) && BIO_pending(ssl->bio_net_io) == 0) {
written = ssl->app_output_closed ? ssl->app_output_closed : PN_EOS;
- ssl->process_output = process_output_done;
+ ssl->io_layer->process_output = process_output_done;
}
_log(ssl, "process_output_ssl() returning %d\n", (int) written);
return written;
@@ -1100,48 +1089,33 @@ static void release_ssl_socket( pn_ssl_t
}
-//////// CLEARTEXT CONNECTIONS
-
-static ssize_t process_input_cleartext(pn_transport_t *transport, const char *input_data, size_t len)
-{
- // just write directly to layer "above" SSL
- return transport->process_input( transport, input_data, len );
-}
-
-static ssize_t process_output_cleartext(pn_transport_t *transport, char *buffer, size_t max_len)
-{
- // just read directly from the layer "above" SSL
- return transport->process_output( transport, buffer, max_len );
-}
-
-
-
static int setup_cleartext_connection( pn_ssl_t *ssl )
{
_log( ssl, "Cleartext connection detected.\n");
- ssl->process_input = process_input_cleartext;
- ssl->process_output = process_output_cleartext;
+ ssl->io_layer->process_input = pn_io_layer_input_passthru;
+ ssl->io_layer->process_output = pn_io_layer_output_passthru;
return 0;
}
// until we determine if the client is using SSL or not:
-static ssize_t process_input_unknown(pn_transport_t *transport, const char *input_data, size_t len)
+static ssize_t process_input_unknown(pn_io_layer_t *io_layer, const char *input_data, size_t len)
{
+ pn_ssl_t *ssl = (pn_ssl_t *)io_layer->context;
switch (check_for_ssl_connection( input_data, len )) {
case SSL_CONNECTION:
- setup_ssl_connection( transport->ssl );
- return transport->ssl->process_input( transport, input_data, len );
+ setup_ssl_connection( ssl );
+ return ssl->io_layer->process_input( ssl->io_layer, input_data, len );
case CLEAR_CONNECTION:
- setup_cleartext_connection( transport->ssl );
- return transport->ssl->process_input( transport, input_data, len );
+ setup_cleartext_connection( ssl );
+ return ssl->io_layer->process_input( ssl->io_layer, input_data, len );
default:
return 0;
}
}
-static ssize_t process_output_unknown(pn_transport_t *transport, char *input_data, size_t len)
+static ssize_t process_output_unknown(pn_io_layer_t *io_layer, char *input_data, size_t len)
{
// do not do output until we know if SSL is used or not
return 0;
@@ -1253,11 +1227,11 @@ int pn_ssl_get_peer_hostname( pn_ssl_t *
return 0;
}
-static ssize_t process_input_done(pn_transport_t *transport, const char *input_data, size_t len)
+static ssize_t process_input_done(pn_io_layer_t *io_layer, const char *input_data, size_t len)
{
return PN_EOS;
}
-static ssize_t process_output_done(pn_transport_t *transport, char *input_data, size_t len)
+static ssize_t process_output_done(pn_io_layer_t *io_layer, char *input_data, size_t len)
{
return PN_EOS;
}
Modified: qpid/proton/branches/jni-binding/proton-c/src/ssl/ssl-internal.h
URL: http://svn.apache.org/viewvc/qpid/proton/branches/jni-binding/proton-c/src/ssl/ssl-internal.h?rev=1440470&r1=1440469&r2=1440470&view=diff
==============================================================================
--- qpid/proton/branches/jni-binding/proton-c/src/ssl/ssl-internal.h (original)
+++ qpid/proton/branches/jni-binding/proton-c/src/ssl/ssl-internal.h Wed Jan 30 15:32:51 2013
@@ -33,13 +33,6 @@
// release the SSL context
void pn_ssl_free( pn_ssl_t *ssl);
-// move data received from the network into the SSL layer
-ssize_t pn_ssl_input(pn_ssl_t *ssl, const char *bytes, size_t available);
-
-// copy data generated by SSL into the network's output buffers
-ssize_t pn_ssl_output(pn_ssl_t *ssl, char *buffer, size_t max_size);
-
-
void pn_ssl_trace(pn_ssl_t *ssl, pn_trace_t trace);
#endif /* ssl-internal.h */
Modified: qpid/proton/branches/jni-binding/proton-j/proton-api/src/main/resources/proton.py
URL: http://svn.apache.org/viewvc/qpid/proton/branches/jni-binding/proton-j/proton-api/src/main/resources/proton.py?rev=1440470&r1=1440469&r2=1440470&view=diff
==============================================================================
--- qpid/proton/branches/jni-binding/proton-j/proton-api/src/main/resources/proton.py (original)
+++ qpid/proton/branches/jni-binding/proton-j/proton-api/src/main/resources/proton.py Wed Jan 30 15:32:51 2013
@@ -904,6 +904,20 @@ class SSL(object):
peer_hostname = property(_get_peer_hostname, _set_peer_hostname)
+
+class Driver(object):
+ """ Proton-c platform abstraction - not needed."""
+ def __init__(self, *args, **kwargs):
+ raise Skipped()
+class Connector(object):
+ """ Proton-c platform abstraction - not needed."""
+ def __init__(self, *args, **kwargs):
+ raise Skipped()
+class Listener(object):
+ """ Proton-c platform abstraction - not needed."""
+ def __init__(self, *args, **kwargs):
+ raise Skipped()
+
__all__ = [
"ACCEPTED",
"LANGUAGE",
@@ -913,10 +927,13 @@ __all__ = [
"PN_SESSION_WINDOW",
"Condition",
"Connection",
+ "Connector",
"Data",
"Delivery",
+ "Driver",
"Endpoint",
"Link",
+ "Listener",
"Message",
"MessageException",
"Messenger",
Propchange: qpid/proton/branches/jni-binding/proton-j/proton-api/src/main/resources/proton.py
------------------------------------------------------------------------------
--- svn:mergeinfo (added)
+++ svn:mergeinfo Wed Jan 30 15:32:51 2013
@@ -0,0 +1,5 @@
+/qpid/proton/branches/kgiusti-msvc2010/proton-j/proton-api/src/main/resources/proton.py:1404579-1405897
+/qpid/proton/branches/kgiusti-proton-136/proton-j/proton-api/src/main/resources/proton.py:1417075-1423611
+/qpid/proton/branches/openssl/proton-j/proton-api/src/main/resources/proton.py:1387623-1387788
+/qpid/proton/trunk/proton-j/proton/src/main/scripts/proton.py:1437538-1440460
+/qpid/proton/trunk/proton-j/proton-api/src/main/resources/proton.py:1421251-1440456
Modified: qpid/proton/branches/jni-binding/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java
URL: http://svn.apache.org/viewvc/qpid/proton/branches/jni-binding/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java?rev=1440470&r1=1440469&r2=1440470&view=diff
==============================================================================
--- qpid/proton/branches/jni-binding/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java (original)
+++ qpid/proton/branches/jni-binding/proton-j/proton/src/main/java/org/apache/qpid/proton/engine/impl/TransportImpl.java Wed Jan 30 15:32:51 2013
@@ -315,6 +315,8 @@ public class TransportImpl extends Endpo
Detach detach = new Detach();
detach.setHandle(localHandle);
+ // TODO - need an API for detaching rather than closing the link
+ detach.setClosed(true);
EndpointError localError = link.getLocalError();
if( localError !=null ) {
Modified: qpid/proton/branches/jni-binding/tests/python/proton_tests/common.py
URL: http://svn.apache.org/viewvc/qpid/proton/branches/jni-binding/tests/python/proton_tests/common.py?rev=1440470&r1=1440469&r2=1440470&view=diff
==============================================================================
--- qpid/proton/branches/jni-binding/tests/python/proton_tests/common.py (original)
+++ qpid/proton/branches/jni-binding/tests/python/proton_tests/common.py Wed Jan 30 15:32:51 2013
@@ -17,6 +17,10 @@
# under the License.
#
+from random import randint
+from threading import Thread
+from proton import Driver, Connection, SASL, Endpoint, Delivery
+
class Test:
def __init__(self, name):
@@ -24,3 +28,199 @@ class Test:
class Skipped(Exception):
skipped = True
+
+
+class TestServer(object):
+ """ Base class for creating test-specific message servers.
+ """
+ def __init__(self, **kwargs):
+ self.args = kwargs
+ self.driver = Driver()
+ self.host = "127.0.0.1"
+ self.port = 0
+ if "host" in kwargs:
+ self.host = kwargs["host"]
+ if "port" in kwargs:
+ self.port = kwargs["port"]
+ self.driver_timeout = -1
+ self.credit_batch = 10
+ self.thread = Thread(name="server-thread", target=self.run)
+ self.thread.daemon = True
+ self.running = True
+
+ def start(self):
+ retry = 0
+ if self.port == 0:
+ self.port = str(randint(49152, 65535))
+ retry = 10
+ self.listener = self.driver.listener(self.host, self.port)
+ while not self.listener and retry > 0:
+ retry -= 1
+ self.port = str(randint(49152, 65535))
+ self.listener = self.driver.listener(self.host, self.port)
+ assert self.listener, "No free port for server to listen on!"
+ self.thread.start()
+
+ def stop(self):
+ self.running = False
+ self.driver.wakeup()
+ self.thread.join()
+ if self.listener:
+ self.listener.close()
+ cxtr = self.driver.head_connector()
+ while cxtr:
+ if not cxtr.closed:
+ cxtr.close()
+ cxtr = cxtr.next()
+
+ # Note: all following methods all run under the thread:
+
+ def run(self):
+ while self.running:
+ self.driver.wait(self.driver_timeout)
+ self.process_listeners()
+ self.process_connectors()
+
+
+ def process_listeners(self):
+ """ Service each pending listener
+ """
+ l = self.driver.pending_listener()
+ while l:
+ cxtr = l.accept()
+ assert(cxtr)
+ self.init_connector(cxtr)
+ l = self.driver.pending_listener()
+
+ def init_connector(self, cxtr):
+ """ Initialize a newly accepted connector
+ """
+ sasl = cxtr.sasl()
+ sasl.mechanisms("ANONYMOUS")
+ sasl.server()
+ cxtr.connection = Connection()
+ if "idle_timeout" in self.args:
+ cxtr.transport.idle_timeout = self.args["idle_timeout"]
+
+ def process_connectors(self):
+ """ Service each pending connector
+ """
+ cxtr = self.driver.pending_connector()
+ while cxtr:
+ self.process_connector(cxtr)
+ cxtr = self.driver.pending_connector()
+
+ def process_connector(self, cxtr):
+ """ Process a pending connector
+ """
+ if not cxtr.closed:
+ cxtr.process()
+ sasl = cxtr.sasl()
+ if sasl.state != SASL.STATE_PASS:
+ self.authenticate_connector(cxtr)
+ else:
+ conn = cxtr.connection
+ if conn:
+ self.service_connection(conn)
+ cxtr.process()
+
+ def authenticate_connector(self, cxtr):
+ """ Deal with a connector that has not passed SASL
+ """
+ # by default, just permit anyone
+ sasl = cxtr.sasl()
+ if sasl.state == SASL.STATE_STEP:
+ sasl.done(SASL.OK)
+
+ def service_connection(self, conn):
+ """ Process a Connection
+ """
+ if conn.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_UNINIT:
+ conn.open()
+
+ # open all pending sessions
+ ssn = conn.session_head(Endpoint.LOCAL_UNINIT)
+ while ssn:
+ self.init_session(ssn)
+ ssn.open()
+ ssn = ssn.next(Endpoint.LOCAL_UNINIT)
+
+ # configure and open any pending links
+ link = conn.link_head(Endpoint.LOCAL_UNINIT)
+ while link:
+ self.init_link(link)
+ link.open()
+ link = link.next(Endpoint.LOCAL_UNINIT);
+
+ ## Step 2: Now drain all the pending deliveries from the connection's
+ ## work queue and process them
+
+ delivery = conn.work_head
+ while delivery:
+ self.process_delivery(delivery)
+ delivery = conn.work_head
+
+ ## Step 3: Clean up any links or sessions that have been closed by the
+ ## remote. If the connection has been closed remotely, clean that up
+ ## also.
+
+ # teardown any terminating links
+ link = conn.link_head(Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED)
+ while link:
+ link.close()
+ link = link.next(Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED)
+
+ # teardown any terminating sessions
+ ssn = conn.session_head(Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED)
+ while ssn:
+ ssn.close(ssn)
+ ssn = ssn.next(Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED)
+
+ if conn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED:
+ conn.close()
+
+ def init_session(self, ssn):
+ """ Test-specific Session initialization
+ """
+ pass
+
+ def init_link(self, link):
+ """ Test-specific Link initialization
+ """
+ pass
+
+ def process_delivery(self, delivery):
+ """ Test-specific Delivery processing.
+ """
+ pass
+
+
+class TestServerDrain(TestServer):
+ """ A primitive test server that accepts connections and simply discards any
+ messages sent to it.
+ """
+ def __init__(self, **kwargs):
+ TestServer.__init__(self, **kwargs)
+
+ def init_link(self, link):
+ """ Test-specific Link initialization
+ """
+ if link.is_receiver:
+ link.flow(self.credit_batch)
+
+ def process_delivery(self, delivery):
+ """ Just drop any incomming messages
+ """
+ link = delivery.link
+ if delivery.readable: # inbound data available
+ m = link.recv(1024)
+ while m:
+ #print("Dropping msg...%s" % str(m))
+ m = link.recv(1024)
+ delivery.update(Delivery.ACCEPTED)
+ delivery.settle()
+ else:
+ link.advance()
+ if link.credit == 0:
+ link.flow(self.credit_batch)
+
Propchange: qpid/proton/branches/jni-binding/tests/python/proton_tests/common.py
------------------------------------------------------------------------------
--- svn:mergeinfo (added)
+++ svn:mergeinfo Wed Jan 30 15:32:51 2013
@@ -0,0 +1,5 @@
+/qpid/proton/branches/kgiusti-msvc2010/tests/python/proton_tests/common.py:1404579-1405897
+/qpid/proton/branches/kgiusti-proton-136/tests/python/proton_tests/common.py:1417075-1423611
+/qpid/proton/branches/openssl/tests/python/proton_tests/common.py:1387623-1387788
+/qpid/proton/trunk/tests/proton_tests/common.py:1437538-1440458
+/qpid/proton/trunk/tests/python/proton_tests/common.py:1421251-1440456
Modified: qpid/proton/branches/jni-binding/tests/python/proton_tests/engine.py
URL: http://svn.apache.org/viewvc/qpid/proton/branches/jni-binding/tests/python/proton_tests/engine.py?rev=1440470&r1=1440469&r2=1440470&view=diff
==============================================================================
--- qpid/proton/branches/jni-binding/tests/python/proton_tests/engine.py (original)
+++ qpid/proton/branches/jni-binding/tests/python/proton_tests/engine.py Wed Jan 30 15:32:51 2013
@@ -18,6 +18,7 @@
#
import os, common
+from time import time, sleep
from proton import *
# future test areas
@@ -1342,3 +1343,107 @@ class PipelineTest(Test):
d.settle()
assert rcv.queued == 0, rcv.queued
+
+class ServerTest(Test):
+
+ def testKeepalive(self):
+ """ Verify that idle frames are sent to keep a Connection alive
+ """
+ idle_timeout_secs = 1
+ self.server = common.TestServerDrain()
+ self.server.start()
+ self.driver = Driver()
+ self.cxtr = self.driver.connector(self.server.host, self.server.port)
+ self.cxtr.transport.idle_timeout = idle_timeout_secs * 1000 #msecs
+ self.cxtr.sasl().mechanisms("ANONYMOUS")
+ self.cxtr.sasl().client()
+ self.conn = Connection()
+ self.cxtr.connection = self.conn
+ self.conn.open()
+ #self.session = self.conn.session()
+ #self.session.open()
+ #self.link = self.session.sender("test-sender")
+ #self.link.open()
+
+ # wait for the connection to come up
+
+ timeout = time() + 10
+ while self.conn.state != (Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE) \
+ and time() <= timeout:
+ self.cxtr.process()
+ self.driver.wait(1)
+ self.cxtr.process()
+
+ assert self.conn.state == (Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE), "Connection failed"
+
+ # wait up to 3x the idle timeout
+ old_count = self.cxtr.transport.frames_input
+ duration = 3 * idle_timeout_secs
+ timeout = time() + duration
+ while time() <= timeout:
+ self.cxtr.process()
+ self.driver.wait(10 * duration * 1000)
+ self.cxtr.process()
+
+ assert self.conn.state == (Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE), "Connection terminated"
+ assert self.cxtr.transport.frames_input > old_count, "No idle frames received"
+
+ self.server.stop()
+
+ def testIdleTimeout(self):
+ """ Verify that a Connection is terminated properly when Idle frames do not
+ arrive in a timely manner.
+ """
+ idle_timeout_secs = 1
+ self.server = common.TestServerDrain(idle_timeout=idle_timeout_secs * 1000)
+ self.server.start()
+ self.driver = Driver()
+ self.cxtr = self.driver.connector(self.server.host, self.server.port)
+ self.cxtr.sasl().mechanisms("ANONYMOUS")
+ self.cxtr.sasl().client()
+ self.conn = Connection()
+ self.cxtr.connection = self.conn
+ self.conn.open()
+
+ # wait for the connection to come up
+
+ timeout = time() + 10
+ while self.conn.state != (Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE) \
+ and time() <= timeout:
+ self.cxtr.process()
+ self.driver.wait(10 * 1000)
+ self.cxtr.process()
+
+ assert self.conn.state == (Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE), "Connection failed"
+
+ # verify the connection stays up even if we don't explicitly send stuff
+ # wait up to 3x the idle timeout
+ old_count = self.cxtr.transport.frames_output
+ duration = 3 * idle_timeout_secs
+ timeout = time() + duration
+ while time() <= timeout:
+ self.cxtr.process()
+ self.driver.wait(10 * duration * 1000)
+ self.cxtr.process()
+
+ assert self.conn.state == (Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE), "Connection terminated"
+ assert self.cxtr.transport.frames_output > old_count, "No idle frames sent"
+
+ # now wait to explicitly cause the other side to expire:
+
+ sleep(idle_timeout_secs * 3)
+
+ # and check that the remote killed the connection:
+
+ timeout = time() + 10
+ while (self.conn.state & Endpoint.REMOTE_ACTIVE) and time() <= timeout:
+ self.cxtr.process()
+ self.driver.wait(10*1000)
+ self.cxtr.process()
+
+ assert self.conn.state & Endpoint.REMOTE_CLOSED, "Connection failed to close"
+
+ self.server.stop()
+
+
+
Propchange: qpid/proton/branches/jni-binding/tests/python/proton_tests/engine.py
------------------------------------------------------------------------------
--- svn:mergeinfo (added)
+++ svn:mergeinfo Wed Jan 30 15:32:51 2013
@@ -0,0 +1,5 @@
+/qpid/proton/branches/kgiusti-msvc2010/tests/python/proton_tests/engine.py:1404579-1405897
+/qpid/proton/branches/kgiusti-proton-136/tests/python/proton_tests/engine.py:1417075-1423611
+/qpid/proton/branches/openssl/tests/python/proton_tests/engine.py:1387623-1387788
+/qpid/proton/trunk/tests/proton_tests/engine.py:1437538-1440458
+/qpid/proton/trunk/tests/python/proton_tests/engine.py:1421251-1440456
Propchange: qpid/proton/branches/jni-binding/tests/python/proton_tests/ssl.py
------------------------------------------------------------------------------
Merged /qpid/proton/trunk/tests/python/proton_tests/ssl.py:r1437538-1440456
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org