You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2017/11/07 19:26:48 UTC
[06/10] qpid-proton git commit: PROTON-1064: [ruby] Wrap
pn_connection_driver_t as ConnectionDriver
PROTON-1064: [ruby] Wrap pn_connection_driver_t as ConnectionDriver
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/9bb1baad
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/9bb1baad
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/9bb1baad
Branch: refs/heads/master
Commit: 9bb1baad263e094427e863e3455b8bc7dbd1f949
Parents: b3d1b07
Author: Alan Conway <ac...@redhat.com>
Authored: Mon Sep 18 22:31:48 2017 -0400
Committer: Alan Conway <ac...@redhat.com>
Committed: Tue Nov 7 13:31:51 2017 -0500
----------------------------------------------------------------------
proton-c/bindings/ruby/CMakeLists.txt | 6 +-
proton-c/bindings/ruby/cproton.i | 677 +++++++++++++++++++
proton-c/bindings/ruby/lib/core/connection.rb | 2 +-
.../bindings/ruby/lib/core/connection_driver.rb | 182 +++++
proton-c/bindings/ruby/lib/core/endpoint.rb | 14 +-
proton-c/bindings/ruby/lib/qpid_proton.rb | 1 +
proton-c/bindings/ruby/ruby.i | 640 ------------------
.../ruby/tests/test_connection_driver.rb | 70 ++
proton-c/include/proton/cproton.i | 13 +-
9 files changed, 956 insertions(+), 649 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/9bb1baad/proton-c/bindings/ruby/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/proton-c/bindings/ruby/CMakeLists.txt b/proton-c/bindings/ruby/CMakeLists.txt
index 93364e2..3397b43 100644
--- a/proton-c/bindings/ruby/CMakeLists.txt
+++ b/proton-c/bindings/ruby/CMakeLists.txt
@@ -27,7 +27,7 @@ list(APPEND SWIG_MODULE_cproton-ruby_EXTRA_DEPS
${PROTON_HEADERS}
)
include_directories (${RUBY_INCLUDE_PATH})
-swig_add_library(cproton-ruby LANGUAGE ruby SOURCES ruby.i)
+swig_add_library(cproton-ruby LANGUAGE ruby SOURCES cproton.i)
swig_link_libraries(cproton-ruby ${BINDING_DEPS} ${RUBY_LIBRARY})
# set a compiler macro to relay the Ruby version to the extension.
@@ -55,11 +55,11 @@ if (GEM_EXE)
add_custom_command(
OUTPUT ${bin}/qpid_proton-${PN_VERSION}.gem
COMMAND ${CMAKE_COMMAND} -E copy_directory ${src} ${bin}/gem
- COMMAND ${CMAKE_COMMAND} -E copy ${bin}/rubyRUBY_wrap.c ${bin}/gem/ext/cproton/cproton.c
+ COMMAND ${CMAKE_COMMAND} -E copy ${bin}/cprotonRUBY_wrap.c ${bin}/gem/ext/cproton/cproton.c
COMMAND ${GEM_EXE} build qpid_proton.gemspec
COMMAND ${CMAKE_COMMAND} -E copy ${bin}/gem/qpid_proton-${PN_VERSION}.gem ${bin}
WORKING_DIRECTORY ${bin}/gem
- DEPENDS ${RUBY_SRC} ${src}/LICENSE ${src}/TODO ${src}/ChangeLog cproton-ruby ${bin}/rubyRUBY_wrap.c
+ DEPENDS ${RUBY_SRC} ${src}/LICENSE ${src}/TODO ${src}/ChangeLog cproton-ruby ${bin}/cprotonRUBY_wrap.c
)
add_custom_target(ruby-gem ALL DEPENDS ${bin}/qpid_proton-${PN_VERSION}.gem )
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/9bb1baad/proton-c/bindings/ruby/cproton.i
----------------------------------------------------------------------
diff --git a/proton-c/bindings/ruby/cproton.i b/proton-c/bindings/ruby/cproton.i
new file mode 100644
index 0000000..107f5d8
--- /dev/null
+++ b/proton-c/bindings/ruby/cproton.i
@@ -0,0 +1,677 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+%module cproton
+
+%{
+#include <proton/connection_driver.h>
+#include <proton/engine.h>
+#include <proton/handlers.h>
+#include <proton/message.h>
+#include <proton/messenger.h>
+#include <proton/reactor.h>
+#include <proton/sasl.h>
+#include <proton/ssl.h>
+#include <proton/types.h>
+#include <proton/url.h>
+%}
+
+/*
+NOTE: According to ccache-swig man page: "Known problems are using
+preprocessor directives within %inline blocks and the use of ’#pragma SWIG’."
+This includes using macros in an %inline section.
+
+Keep preprocessor directives and macro expansions in the normal header section.
+*/
+
+%include <cstring.i>
+
+%cstring_output_withsize(char *OUTPUT, size_t *OUTPUT_SIZE)
+%cstring_output_allocate_size(char **ALLOC_OUTPUT, size_t *ALLOC_SIZE, free(*$1));
+%cstring_output_maxsize(char *OUTPUT, size_t MAX_OUTPUT_SIZE)
+
+%{
+#if !defined(RSTRING_LEN)
+# define RSTRING_LEN(x) (RSTRING(x)->len)
+# define RSTRING_PTR(x) (RSTRING(x)->ptr)
+#endif
+%}
+
+%typemap(in) pn_bytes_t {
+ if ($input == Qnil) {
+ $1.start = NULL;
+ $1.size = 0;
+ } else {
+ $1.start = RSTRING_PTR($input);
+ if (!$1.start) {
+ $1.size = 0;
+ }
+ $1.size = RSTRING_LEN($input);
+ }
+}
+
+%typemap(out) pn_bytes_t {
+ $result = rb_str_new($1.start, $1.size);
+}
+
+%typemap(in) pn_atom_t
+{
+ if ($input == Qnil)
+ {
+ $1.type = PN_NULL;
+ }
+ else
+ {
+ switch(TYPE($input))
+ {
+ case T_TRUE:
+ $1.type = PN_BOOL;
+ $1.u.as_bool = true;
+ break;
+
+ case T_FALSE:
+ $1.type = PN_BOOL;
+ $1.u.as_bool = false;
+ break;
+
+ case T_FLOAT:
+ $1.type = PN_FLOAT;
+ $1.u.as_float = NUM2DBL($input);
+ break;
+
+ case T_STRING:
+ $1.type = PN_STRING;
+ $1.u.as_bytes.start = RSTRING_PTR($input);
+ if ($1.u.as_bytes.start)
+ {
+ $1.u.as_bytes.size = RSTRING_LEN($input);
+ }
+ else
+ {
+ $1.u.as_bytes.size = 0;
+ }
+ break;
+
+ case T_FIXNUM:
+ $1.type = PN_INT;
+ $1.u.as_int = FIX2LONG($input);
+ break;
+
+ case T_BIGNUM:
+ $1.type = PN_LONG;
+ $1.u.as_long = NUM2LL($input);
+ break;
+
+ }
+ }
+}
+
+%typemap(out) pn_atom_t
+{
+ switch($1.type)
+ {
+ case PN_NULL:
+ $result = Qnil;
+ break;
+
+ case PN_BOOL:
+ $result = $1.u.as_bool ? Qtrue : Qfalse;
+ break;
+
+ case PN_BYTE:
+ $result = INT2NUM($1.u.as_byte);
+ break;
+
+ case PN_UBYTE:
+ $result = UINT2NUM($1.u.as_ubyte);
+ break;
+
+ case PN_SHORT:
+ $result = INT2NUM($1.u.as_short);
+ break;
+
+ case PN_USHORT:
+ $result = UINT2NUM($1.u.as_ushort);
+ break;
+
+ case PN_INT:
+ $result = INT2NUM($1.u.as_int);
+ break;
+
+ case PN_UINT:
+ $result = UINT2NUM($1.u.as_uint);
+ break;
+
+ case PN_LONG:
+ $result = LL2NUM($1.u.as_long);
+ break;
+
+ case PN_ULONG:
+ $result = ULL2NUM($1.u.as_ulong);
+ break;
+
+ case PN_FLOAT:
+ $result = rb_float_new($1.u.as_float);
+ break;
+
+ case PN_DOUBLE:
+ $result = rb_float_new($1.u.as_double);
+ break;
+
+ case PN_STRING:
+ $result = rb_str_new($1.u.as_bytes.start, $1.u.as_bytes.size);
+ break;
+
+ default:
+ break;
+ }
+}
+
+%typemap (in) pn_decimal32_t
+{
+ $1 = FIX2UINT($input);
+}
+
+%typemap (out) pn_decimal32_t
+{
+ $result = ULL2NUM($1);
+}
+
+%typemap (in) pn_decimal64_t
+{
+ $1 = NUM2ULL($input);
+}
+
+%typemap (out) pn_decimal64_t
+{
+ $result = ULL2NUM($1);
+}
+
+%typemap (in) pn_decimal128_t
+{
+ int index;
+
+ for(index = 0; index < 16; index++)
+ {
+ VALUE element = rb_ary_entry($input, index);
+ $1.bytes[16 - (index + 1)] = FIX2INT(element);
+ }
+}
+
+%typemap (out) pn_decimal128_t
+{
+ int index;
+
+ $result = rb_ary_new2(16);
+ for(index = 0; index < 16; index++)
+ {
+ rb_ary_store($result, 16 - (index + 1), CHR2FIX($1.bytes[index]));
+ }
+}
+
+%typemap (in) pn_uuid_t
+{
+ int index;
+
+ for(index = 0; index < 16; index++)
+ {
+ VALUE element = rb_ary_entry($input, index);
+ $1.bytes[16 - (index + 1)] = FIX2INT(element);
+ }
+}
+
+%typemap (out) pn_uuid_t
+{
+ int index;
+
+ $result = rb_ary_new2(16);
+ for(index = 0; index < 16; index++)
+ {
+ rb_ary_store($result, 16 - (index + 1), CHR2FIX($1.bytes[index]));
+ }
+}
+
+int pn_message_encode(pn_message_t *msg, char *OUTPUT, size_t *OUTPUT_SIZE);
+%ignore pn_message_encode;
+
+ssize_t pn_link_send(pn_link_t *transport, char *STRING, size_t LENGTH);
+%ignore pn_link_send;
+
+%rename(pn_link_recv) wrap_pn_link_recv;
+%inline %{
+ int wrap_pn_link_recv(pn_link_t *link, char *OUTPUT, size_t *OUTPUT_SIZE) {
+ ssize_t sz = pn_link_recv(link, OUTPUT, *OUTPUT_SIZE);
+ if (sz >= 0) {
+ *OUTPUT_SIZE = sz;
+ } else {
+ *OUTPUT_SIZE = 0;
+ }
+ return sz;
+ }
+%}
+%ignore pn_link_recv;
+
+ssize_t pn_transport_input(pn_transport_t *transport, char *STRING, size_t LENGTH);
+%ignore pn_transport_input;
+
+%rename(pn_transport_output) wrap_pn_transport_output;
+%inline %{
+ int wrap_pn_transport_output(pn_transport_t *transport, char *OUTPUT, size_t *OUTPUT_SIZE) {
+ ssize_t sz = pn_transport_output(transport, OUTPUT, *OUTPUT_SIZE);
+ if (sz >= 0) {
+ *OUTPUT_SIZE = sz;
+ } else {
+ *OUTPUT_SIZE = 0;
+ }
+ return sz;
+ }
+%}
+%ignore pn_transport_output;
+
+%rename(pn_transport_peek) wrap_pn_transport_peek;
+%inline %{
+ int wrap_pn_transport_peek(pn_transport_t *transport, char *OUTPUT, size_t *OUTPUT_SIZE) {
+ ssize_t sz = pn_transport_peek(transport, OUTPUT, *OUTPUT_SIZE);
+ if(sz >= 0) {
+ *OUTPUT_SIZE = sz;
+ } else {
+ *OUTPUT_SIZE = 0;
+ }
+ return sz;
+ }
+%}
+%ignore pn_transport_peek;
+
+%rename(pn_delivery) wrap_pn_delivery;
+%inline %{
+ pn_delivery_t *wrap_pn_delivery(pn_link_t *link, char *STRING, size_t LENGTH) {
+ return pn_delivery(link, pn_dtag(STRING, LENGTH));
+ }
+%}
+%ignore pn_delivery;
+
+// Suppress "Warning(451): Setting a const char * variable may leak memory." on pn_delivery_tag_t
+%warnfilter(451) pn_delivery_tag_t;
+%rename(pn_delivery_tag) wrap_pn_delivery_tag;
+%inline %{
+ void wrap_pn_delivery_tag(pn_delivery_t *delivery, char **ALLOC_OUTPUT, size_t *ALLOC_SIZE) {
+ pn_delivery_tag_t tag = pn_delivery_tag(delivery);
+ *ALLOC_OUTPUT = malloc(tag.size);
+ *ALLOC_SIZE = tag.size;
+ memcpy(*ALLOC_OUTPUT, tag.start, tag.size);
+ }
+%}
+%ignore pn_delivery_tag;
+
+bool pn_ssl_get_cipher_name(pn_ssl_t *ssl, char *OUTPUT, size_t MAX_OUTPUT_SIZE);
+%ignore pn_ssl_get_cipher_name;
+
+bool pn_ssl_get_protocol_name(pn_ssl_t *ssl, char *OUTPUT, size_t MAX_OUTPUT_SIZE);
+%ignore pn_ssl_get_protocol_name;
+
+%inline %{
+#if defined(RUBY20) || defined(RUBY21)
+
+ typedef void *non_blocking_return_t;
+#define RB_BLOCKING_CALL rb_thread_call_without_gvl
+
+#elif defined(RUBY19)
+
+ typedef VALUE non_blocking_return_t;
+#define RB_BLOCKING_CALL rb_thread_blocking_region
+
+#endif
+ %}
+
+%rename(pn_messenger_send) wrap_pn_messenger_send;
+%rename(pn_messenger_recv) wrap_pn_messenger_recv;
+%rename(pn_messenger_work) wrap_pn_messenger_work;
+
+%inline %{
+
+#if defined(RB_BLOCKING_CALL)
+
+ static non_blocking_return_t pn_messenger_send_no_gvl(void *args) {
+ VALUE result = Qnil;
+ pn_messenger_t *messenger = (pn_messenger_t *)((void **)args)[0];
+ int *limit = (int *)((void **)args)[1];
+
+ int rc = pn_messenger_send(messenger, *limit);
+
+ result = INT2NUM(rc);
+ return (non_blocking_return_t )result;
+ }
+
+ static non_blocking_return_t pn_messenger_recv_no_gvl(void *args) {
+ VALUE result = Qnil;
+ pn_messenger_t *messenger = (pn_messenger_t *)((void **)args)[0];
+ int *limit = (int *)((void **)args)[1];
+
+ int rc = pn_messenger_recv(messenger, *limit);
+
+ result = INT2NUM(rc);
+ return (non_blocking_return_t )result;
+ }
+
+ static non_blocking_return_t pn_messenger_work_no_gvl(void *args) {
+ VALUE result = Qnil;
+ pn_messenger_t *messenger = (pn_messenger_t *)((void **)args)[0];
+ int *timeout = (int *)((void **)args)[1];
+
+ int rc = pn_messenger_work(messenger, *timeout);
+
+ result = INT2NUM(rc);
+ return (non_blocking_return_t )result;
+ }
+
+#endif
+
+ int wrap_pn_messenger_send(pn_messenger_t *messenger, int limit) {
+ int result = 0;
+
+#if defined(RB_BLOCKING_CALL)
+
+ // only release the gil if we're blocking
+ if(pn_messenger_is_blocking(messenger)) {
+ VALUE rc;
+ void* args[2];
+
+ args[0] = messenger;
+ args[1] = &limit;
+
+ rc = RB_BLOCKING_CALL(pn_messenger_send_no_gvl,
+ &args, RUBY_UBF_PROCESS, NULL);
+
+ if(RTEST(rc))
+ {
+ result = FIX2INT(rc);
+ }
+ }
+
+#else // !defined(RB_BLOCKING_CALL)
+ result = pn_messenger_send(messenger, limit);
+#endif // defined(RB_BLOCKING_CALL)
+
+ return result;
+ }
+
+ int wrap_pn_messenger_recv(pn_messenger_t *messenger, int limit) {
+ int result = 0;
+
+#if defined(RB_BLOCKING_CALL)
+ // only release the gil if we're blocking
+ if(pn_messenger_is_blocking(messenger)) {
+ VALUE rc;
+ void* args[2];
+
+ args[0] = messenger;
+ args[1] = &limit;
+
+ rc = RB_BLOCKING_CALL(pn_messenger_recv_no_gvl,
+ &args, RUBY_UBF_PROCESS, NULL);
+
+ if(RTEST(rc))
+ {
+ result = FIX2INT(rc);
+ }
+
+ } else {
+ result = pn_messenger_recv(messenger, limit);
+ }
+#else // !defined(RB_BLOCKING_CALL)
+ result = pn_messenger_recv(messenger, limit);
+#endif // defined(RB_BLOCKING_CALL)
+
+ return result;
+ }
+
+ int wrap_pn_messenger_work(pn_messenger_t *messenger, int timeout) {
+ int result = 0;
+
+#if defined(RB_BLOCKING_CALL)
+ // only release the gil if we're blocking
+ if(timeout) {
+ VALUE rc;
+ void* args[2];
+
+ args[0] = messenger;
+ args[1] = &timeout;
+
+ rc = RB_BLOCKING_CALL(pn_messenger_work_no_gvl,
+ &args, RUBY_UBF_PROCESS, NULL);
+
+ if(RTEST(rc))
+ {
+ result = FIX2INT(rc);
+ }
+ } else {
+ result = pn_messenger_work(messenger, timeout);
+ }
+#else
+ result = pn_messenger_work(messenger, timeout);
+#endif
+
+ return result;
+ }
+
+%}
+
+%ignore pn_messenger_send;
+%ignore pn_messenger_recv;
+%ignore pn_messenger_work;
+
+%{
+typedef struct Pn_rbkey_t {
+ void *registry;
+ char *method;
+ char *key_value;
+} Pn_rbkey_t;
+
+void Pn_rbkey_initialize(void *vp_rbkey) {
+ Pn_rbkey_t *rbkey = (Pn_rbkey_t*)vp_rbkey;
+ assert(rbkey);
+ rbkey->registry = NULL;
+ rbkey->method = NULL;
+ rbkey->key_value = NULL;
+}
+
+void Pn_rbkey_finalize(void *vp_rbkey) {
+ Pn_rbkey_t *rbkey = (Pn_rbkey_t*)vp_rbkey;
+ if(rbkey && rbkey->registry && rbkey->method && rbkey->key_value) {
+ rb_funcall((VALUE )rbkey->registry, rb_intern(rbkey->method), 1, rb_str_new2(rbkey->key_value));
+ }
+ if(rbkey->key_value) {
+ free(rbkey->key_value);
+ rbkey->key_value = NULL;
+ }
+}
+
+/* NOTE: no macro or preprocessor definitions in %inline sections */
+#define CID_Pn_rbkey CID_pn_void
+#define Pn_rbkey_inspect NULL
+#define Pn_rbkey_compare NULL
+#define Pn_rbkey_hashcode NULL
+
+pn_class_t* Pn_rbkey__class(void) {
+ static pn_class_t clazz = PN_CLASS(Pn_rbkey);
+ return &clazz;
+}
+
+Pn_rbkey_t *Pn_rbkey_new(void) {
+ return (Pn_rbkey_t *) pn_class_new(Pn_rbkey__class(), sizeof(Pn_rbkey_t));
+}
+%}
+
+pn_class_t* Pn_rbkey__class(void);
+Pn_rbkey_t *Pn_rbkey_new(void);
+
+%inline %{
+
+Pn_rbkey_t *Pn_rbkey_new(void);
+
+void Pn_rbkey_set_registry(Pn_rbkey_t *rbkey, void *registry) {
+ assert(rbkey);
+ rbkey->registry = registry;
+}
+
+void *Pn_rbkey_get_registry(Pn_rbkey_t *rbkey) {
+ assert(rbkey);
+ return rbkey->registry;
+}
+
+void Pn_rbkey_set_method(Pn_rbkey_t *rbkey, char *method) {
+ assert(rbkey);
+ rbkey->method = method;
+}
+
+char *Pn_rbkey_get_method(Pn_rbkey_t *rbkey) {
+ assert(rbkey);
+ return rbkey->method;
+}
+
+void Pn_rbkey_set_key_value(Pn_rbkey_t *rbkey, char *key_value) {
+ assert(rbkey);
+ rbkey->key_value = malloc(strlen(key_value) + 1);
+ strncpy(rbkey->key_value, key_value, strlen(key_value) + 1);
+}
+
+char *Pn_rbkey_get_key_value(Pn_rbkey_t *rbkey) {
+ assert(rbkey);
+ return rbkey->key_value;
+}
+
+Pn_rbkey_t *pni_void2rbkey(void *object) {
+ return (Pn_rbkey_t *)object;
+}
+
+VALUE pn_void2rb(void *object) {
+ return (VALUE )object;
+}
+
+void *pn_rb2void(VALUE object) {
+ return (void *)object;
+}
+
+VALUE pni_address_of(void *object) {
+ return ULL2NUM((unsigned long )object);
+}
+
+%}
+
+//%rename(pn_collector_put) wrap_pn_collector_put;
+//%inline %{
+// pn_event_t *wrap_pn_collector_put(pn_collector_t *collector, void *context,
+// pn_event_type_t type) {
+// return pn_collector_put(collector, PN_RBREF, context, type);
+// }
+// %}
+//%ignore pn_collector_put;
+
+int pn_ssl_get_peer_hostname(pn_ssl_t *ssl, char *OUTPUT, size_t *OUTPUT_SIZE);
+%ignore pn_ssl_get_peer_hostname;
+
+%inline %{
+
+ VALUE pni_ruby_get_proton_module() {
+ VALUE mQpid = rb_define_module("Qpid");
+ return rb_define_module_under(mQpid, "Proton");
+ }
+
+ void pni_ruby_add_to_registry(VALUE key, VALUE value) {
+ VALUE result = rb_funcall(pni_ruby_get_proton_module(), rb_intern("add_to_registry"), 2, key, value);
+ }
+
+ VALUE pni_ruby_get_from_registry(VALUE key) {
+ return rb_funcall(pni_ruby_get_proton_module(), rb_intern("get_from_registry"), 1, key);
+ }
+
+ void pni_ruby_delete_from_registry(VALUE stored_key) {
+ rb_funcall(pni_ruby_get_proton_module(), rb_intern("delete_from_registry"), 1, stored_key);
+ }
+
+ typedef struct {
+ VALUE handler_key;
+ } Pni_rbhandler_t;
+
+ static Pni_rbhandler_t *pni_rbhandler(pn_handler_t *handler) {
+ return (Pni_rbhandler_t *) pn_handler_mem(handler);
+ }
+
+ static void pni_rbdispatch(pn_handler_t *handler, pn_event_t *event, pn_event_type_t type) {
+ Pni_rbhandler_t *rbh = pni_rbhandler(handler);
+ VALUE rbhandler = pni_ruby_get_from_registry(rbh->handler_key);
+
+ rb_funcall(rbhandler, rb_intern("dispatch"), 2, SWIG_NewPointerObj(event, SWIGTYPE_p_pn_event_t, 0), INT2FIX(type));
+ }
+
+ static void pni_rbhandler_finalize(pn_handler_t *handler) {
+ Pni_rbhandler_t *rbh = pni_rbhandler(handler);
+ pni_ruby_delete_from_registry(rbh->handler_key);
+ }
+
+ pn_handler_t *pn_rbhandler(VALUE handler) {
+ pn_handler_t *chandler = pn_handler_new(pni_rbdispatch, sizeof(Pni_rbhandler_t), pni_rbhandler_finalize);
+ Pni_rbhandler_t *rhy = pni_rbhandler(chandler);
+
+ VALUE ruby_key = rb_class_new_instance(0, NULL, rb_cObject);
+ pni_ruby_add_to_registry(ruby_key, handler);
+
+ rhy->handler_key = ruby_key;
+
+ return chandler;
+ }
+
+
+ /* Helpers for working with pn_connection_driver_t */
+
+ size_t pni_connection_driver_read_size(pn_connection_driver_t* d) {
+ return pn_connection_driver_read_buffer(d).size;
+ }
+
+ size_t pni_connection_driver_write_size(pn_connection_driver_t* d) {
+ return pn_connection_driver_write_buffer(d).size;
+ }
+
+ pn_connection_t *pni_connection_driver_connection(pn_connection_driver_t* d) {
+ return d->connection;
+ }
+
+ pn_transport_t *pni_connection_driver_transport(pn_connection_driver_t* d) {
+ return d->transport;
+ }
+
+ size_t pni_connection_driver_read_copy(pn_connection_driver_t* d, char *STRING, size_t LENGTH ) {
+ pn_rwbytes_t rbuf = pn_connection_driver_read_buffer(d);
+ size_t n = LENGTH < rbuf.size ? LENGTH : rbuf.size;
+ memcpy(rbuf.start, STRING, n);
+ pn_connection_driver_read_done(d, n);
+ return n;
+ }
+
+ pn_connection_driver_t *pni_connection_driver() {
+ pn_connection_driver_t *d = (pn_connection_driver_t*)malloc(sizeof(*d));
+ if (pn_connection_driver_init(d, NULL, NULL) != 0) {
+ free(d);
+ return NULL;
+ }
+ return d;
+ }
+
+%}
+
+%include "proton/cproton.i"
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/9bb1baad/proton-c/bindings/ruby/lib/core/connection.rb
----------------------------------------------------------------------
diff --git a/proton-c/bindings/ruby/lib/core/connection.rb b/proton-c/bindings/ruby/lib/core/connection.rb
index ef785b2..6caf589 100644
--- a/proton-c/bindings/ruby/lib/core/connection.rb
+++ b/proton-c/bindings/ruby/lib/core/connection.rb
@@ -196,7 +196,7 @@ module Qpid::Proton
# Open the local end of the connection.
#
# @option options [String] :container_id Unique AMQP container ID, defaults to a UUID
- # @option [String] :link_prefix Prefix for generated link names, default is container_id
+ # @option options [String] :link_prefix Prefix for generated link names, default is container_id
#
def open(options={})
object_to_data(@offered_capabilities, Cproton.pn_connection_offered_capabilities(@impl))
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/9bb1baad/proton-c/bindings/ruby/lib/core/connection_driver.rb
----------------------------------------------------------------------
diff --git a/proton-c/bindings/ruby/lib/core/connection_driver.rb b/proton-c/bindings/ruby/lib/core/connection_driver.rb
new file mode 100644
index 0000000..b5b38ac
--- /dev/null
+++ b/proton-c/bindings/ruby/lib/core/connection_driver.rb
@@ -0,0 +1,182 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+require 'socket'
+
+module Qpid
+ module Proton
+
+ # Associate an AMQP {Connection} with an {IO} and a {MessagingHandler}
+ #
+ # - Read AMQP binary data from the {IO} (#read, #process)
+ # - Call on_* methods on the {MessagingHandler} for AMQP events (#dispatch, #process)
+ # - Write AMQP binary data to the {IO} (#write, #process)
+ #
+ # Thread safety: The {ConnectionDriver} is not thread safe but separate
+ # {ConnectionDriver} instances can be processed concurrently. The
+ # {Container} handles multiple connections concurrently in multiple threads.
+ #
+ class ConnectionDriver
+
+ # Create a {Connection} and associate it with +io+ and +handler+
+ #
+ # @param io [#read_nonblock, #write_nonblock] An {IO} or {IO}-like object that responds
+ # to #read_nonblock and #write_nonblock.
+ # @param handler [MessagingHandler] The handler to be invoked for AMQP events
+ #
+ def initialize io, handler=nil
+ @impl = Cproton.pni_connection_driver or raise RuntimeError, "cannot create connection driver"
+ @io = io
+ @handler = handler || Handler::MessagingHandler.new # Default handler for default behaviour
+ @rbuf = "" # String to re-use as read buffer
+ end
+
+ # @return [MessagingHandler]
+ attr_reader :handler
+
+ # @return [Connection]
+ def connection() Connection.wrap(Cproton.pni_connection_driver_connection(@impl)); end
+
+ # @return [Transport]
+ def transport() Transport.wrap(Cproton.pni_connection_driver_transport(@impl)); end
+
+ # @return [IO] Allows ConnectionDriver to be passed directly to {IO#select}
+ def to_io() @io; end
+
+ # @return [Bool] True if the driver can read more data
+ def can_read?() Cproton.pni_connection_driver_read_size(@impl) > 0; end
+
+ # @return [Bool] True if the driver has data to write
+ def can_write?() Cproton.pni_connection_driver_write_size(@impl) > 0; end
+
+ # True if read and write sides of the IO are closed. Note this does not imply
+ # {#finished?} since there may still be events to dispatch.
+ def closed?
+ Cproton.pn_connection_driver_read_closed(@impl) &&
+ Cproton.pn_connection_driver_read_closed(@impl)
+ end
+
+ # True if the ConnectionDriver has nothing left to do: {#closed?} and
+ # there are no more events to dispatch.
+ def finished?() Cproton.pn_connection_driver_finished(@impl); end
+
+ # Dispatch available events, call the relevant on_* methods on the {#handler}.
+ def dispatch(extra_handlers = nil)
+ extra_handlers ||= []
+ while event = Event::Event.wrap(Cproton.pn_connection_driver_next_event(@impl))
+ event.dispatch(@handler)
+ extra_handlers.each { |h| event.dispatch h }
+ end
+ end
+
+ # Read from IO without blocking.
+ # IO errors are not raised, they are passed to {#handler}.on_transport_error by {#dispatch}
+ def read
+ size = Cproton.pni_connection_driver_read_size(@impl)
+ return if size <= 0
+ @io.read_nonblock(size, @rbuf) # Use the same string rbuf for reading each time
+ Cproton.pni_connection_driver_read_copy(@impl, @rbuf) unless @rbuf.empty?
+ rescue Errno::EWOULDBLOCK, Errno::EAGAIN, Errno::EINTR
+ # Try again later.
+ rescue EOFError # EOF is not an error
+ Cproton.pn_connection_driver_read_close(@impl)
+ rescue IOError => e # IOError is passed to the transport
+ error "read: #{e}"
+ Cproton.pn_connection_driver_read_close(@impl)
+ end
+
+ # Write to IO without blocking.
+ # IO errors are not raised, they are passed to {#handler}.on_transport_error by {#dispatch}
+ def write
+ n = @io.write_nonblock(Cproton.pn_connection_driver_write_buffer(@impl))
+ Cproton.pn_connection_driver_write_done(@impl, n) if n > 0
+ rescue Errno::EWOULDBLOCK, Errno::EAGAIN, Errno::EINTR
+ # Try again later.
+ rescue IOError => e
+ error "write: #{e}"
+ Cproton.pn_connection_driver_write_close(@impl)
+ end
+
+ # Generate timed events and IO, for example idle-timeout and heart-beat events.
+ # May generate events for {#dispatch} and change the readable/writeable state.
+ #
+ # @param [Time] now the current time, defaults to {Time#now}.
+ #
+ # @return [Time] time of the next scheduled event, or nil if there are no
+ # scheduled events. If non-nil, tick() must be called again no later than
+ # this time.
+ def tick(now=Time.now)
+ transport = Cproton.pni_connection_driver_transport(@impl)
+ ms = Cproton.pn_transport_tick(transport, (now.to_r * 1000).to_i)
+ return ms.zero? ? nil : Time.at(ms.to_r / 1000);
+ end
+
+ # Do read, tick, write and dispatch without blocking.
+ # @param [Bool] io_readable true if the IO might be readable
+ # @param [Bool] io_writable true if the IO might be writeable
+ # @param [Time] now the current time
+ # @return [Time] Latest time to call {#process} again for scheduled events,
+ # or nil if there are no scheduled events
+ def process(io_readable=true, io_writable=true, now=Time.now)
+ read if io_readable
+ next_tick = tick(now)
+ if io_writable
+ dispatch
+ write
+ end
+ dispatch
+ return next_tick
+ end
+
+ # Close the read side of the IO with optional error.
+ # @param e [#to_s] Non-nil error will call {#handler}.on_transport_error on next {#dispatch}
+ def close_read(e=nil)
+ @io.close_read
+ error(e)
+ Cproton.pn_connection_driver_read_close(@impl)
+ end
+
+ # Close the write side of the IO with optional error
+ # @param e [#to_s] Non-nil error will call {#handler}.on_transport_error on next {#dispatch}
+ def close_write(e=nil)
+ @io.close_write
+ error(e)
+ Cproton.pn_connection_driver_write_close(@impl)
+ end
+
+ # Close both sides of the IO with optional error
+ # @param e [#to_s] Non-nil error will call {#handler}.on_transport_error on next {#dispatch}
+ def close(e=nil)
+ if !closed?
+ close_read(e)
+ close_write(e)
+ end
+ end
+
+ def to_s
+ transport = Cproton.pni_connection_driver_tranport(@impl)
+ return "#<#{self.class.name}[#{transport}]:#{@io}>"
+ end
+
+ private
+
+ def error(e)
+ Cproton.pn_connection_driver_errorf(@impl, "proton:io", "%s", e.to_s) if e
+ end
+ end
+ end
+end
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/9bb1baad/proton-c/bindings/ruby/lib/core/endpoint.rb
----------------------------------------------------------------------
diff --git a/proton-c/bindings/ruby/lib/core/endpoint.rb b/proton-c/bindings/ruby/lib/core/endpoint.rb
index f3ddbcb..7c6f0a3 100644
--- a/proton-c/bindings/ruby/lib/core/endpoint.rb
+++ b/proton-c/bindings/ruby/lib/core/endpoint.rb
@@ -82,6 +82,12 @@ module Qpid::Proton
self.connection.transport
end
+ # @return [Bool] true if endpoint has sent and received a CLOSE frame
+ def closed?() check_state(LOCAL_CLOSED | REMOTE_CLOSED); end
+
+ # @return [Bool] true if endpoint has sent and received an OPEN frame
+ def open?() check_state(LOCAL_ACTIVE | REMOTE_ACTIVE); end
+
def local_uninit?
check_state(LOCAL_UNINIT)
end
@@ -106,10 +112,6 @@ module Qpid::Proton
check_state(REMOTE_CLOSED)
end
- def check_state(state_mask)
- !(self.state & state_mask).zero?
- end
-
def handler
reactor = Qpid::Proton::Reactor::Reactor.wrap(Cproton.pn_object_reactor(@impl))
if reactor.nil?
@@ -135,6 +137,10 @@ module Qpid::Proton
Cproton.pn_decref(impl)
end
+ private
+
+ def check_state(mask) (self.state & mask) == mask; end
+
end
end
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/9bb1baad/proton-c/bindings/ruby/lib/qpid_proton.rb
----------------------------------------------------------------------
diff --git a/proton-c/bindings/ruby/lib/qpid_proton.rb b/proton-c/bindings/ruby/lib/qpid_proton.rb
index 0180291..dae9b5f 100644
--- a/proton-c/bindings/ruby/lib/qpid_proton.rb
+++ b/proton-c/bindings/ruby/lib/qpid_proton.rb
@@ -78,6 +78,7 @@ require "core/ssl"
require "core/transport"
require "core/base_handler"
require "core/url"
+require "core/connection_driver"
# Messenger API classes
require "messenger/subscription"
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/9bb1baad/proton-c/bindings/ruby/ruby.i
----------------------------------------------------------------------
diff --git a/proton-c/bindings/ruby/ruby.i b/proton-c/bindings/ruby/ruby.i
deleted file mode 100644
index d5979f3..0000000
--- a/proton-c/bindings/ruby/ruby.i
+++ /dev/null
@@ -1,640 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-%module cproton
-
-%{
-#include <proton/engine.h>
-#include <proton/message.h>
-#include <proton/sasl.h>
-#include <proton/messenger.h>
-#include <proton/ssl.h>
-#include <proton/types.h>
-#include <proton/url.h>
-#include <proton/reactor.h>
-#include <proton/handlers.h>
-%}
-
-/*
-NOTE: According to ccache-swig man page: "Known problems are using
-preprocessor directives within %inline blocks and the use of ’#pragma SWIG’."
-This includes using macros in an %inline section.
-
-Keep preprocessor directives and macro expansions in the normal header section.
-*/
-
-%include <cstring.i>
-
-%cstring_output_withsize(char *OUTPUT, size_t *OUTPUT_SIZE)
-%cstring_output_allocate_size(char **ALLOC_OUTPUT, size_t *ALLOC_SIZE, free(*$1));
-%cstring_output_maxsize(char *OUTPUT, size_t MAX_OUTPUT_SIZE)
-
-%{
-#if !defined(RSTRING_LEN)
-# define RSTRING_LEN(x) (RSTRING(X)->len)
-# define RSTRING_PTR(x) (RSTRING(x)->ptr)
-#endif
-%}
-
-%typemap(in) pn_bytes_t {
- if ($input == Qnil) {
- $1.start = NULL;
- $1.size = 0;
- } else {
- $1.start = RSTRING_PTR($input);
- if (!$1.start) {
- $1.size = 0;
- }
- $1.size = RSTRING_LEN($input);
- }
-}
-
-%typemap(out) pn_bytes_t {
- $result = rb_str_new($1.start, $1.size);
-}
-
-%typemap(in) pn_atom_t
-{
- if ($input == Qnil)
- {
- $1.type = PN_NULL;
- }
- else
- {
- switch(TYPE($input))
- {
- case T_TRUE:
- $1.type = PN_BOOL;
- $1.u.as_bool = true;
- break;
-
- case T_FALSE:
- $1.type = PN_BOOL;
- $1.u.as_bool = false;
- break;
-
- case T_FLOAT:
- $1.type = PN_FLOAT;
- $1.u.as_float = NUM2DBL($input);
- break;
-
- case T_STRING:
- $1.type = PN_STRING;
- $1.u.as_bytes.start = RSTRING_PTR($input);
- if ($1.u.as_bytes.start)
- {
- $1.u.as_bytes.size = RSTRING_LEN($input);
- }
- else
- {
- $1.u.as_bytes.size = 0;
- }
- break;
-
- case T_FIXNUM:
- $1.type = PN_INT;
- $1.u.as_int = FIX2LONG($input);
- break;
-
- case T_BIGNUM:
- $1.type = PN_LONG;
- $1.u.as_long = NUM2LL($input);
- break;
-
- }
- }
-}
-
-%typemap(out) pn_atom_t
-{
- switch($1.type)
- {
- case PN_NULL:
- $result = Qnil;
- break;
-
- case PN_BOOL:
- $result = $1.u.as_bool ? Qtrue : Qfalse;
- break;
-
- case PN_BYTE:
- $result = INT2NUM($1.u.as_byte);
- break;
-
- case PN_UBYTE:
- $result = UINT2NUM($1.u.as_ubyte);
- break;
-
- case PN_SHORT:
- $result = INT2NUM($1.u.as_short);
- break;
-
- case PN_USHORT:
- $result = UINT2NUM($1.u.as_ushort);
- break;
-
- case PN_INT:
- $result = INT2NUM($1.u.as_int);
- break;
-
- case PN_UINT:
- $result = UINT2NUM($1.u.as_uint);
- break;
-
- case PN_LONG:
- $result = LL2NUM($1.u.as_long);
- break;
-
- case PN_ULONG:
- $result = ULL2NUM($1.u.as_ulong);
- break;
-
- case PN_FLOAT:
- $result = rb_float_new($1.u.as_float);
- break;
-
- case PN_DOUBLE:
- $result = rb_float_new($1.u.as_double);
- break;
-
- case PN_STRING:
- $result = rb_str_new($1.u.as_bytes.start, $1.u.as_bytes.size);
- break;
-
- default:
- break;
- }
-}
-
-%typemap (in) pn_decimal32_t
-{
- $1 = FIX2UINT($input);
-}
-
-%typemap (out) pn_decimal32_t
-{
- $result = ULL2NUM($1);
-}
-
-%typemap (in) pn_decimal64_t
-{
- $1 = NUM2ULL($input);
-}
-
-%typemap (out) pn_decimal64_t
-{
- $result = ULL2NUM($1);
-}
-
-%typemap (in) pn_decimal128_t
-{
- int index;
-
- for(index = 0; index < 16; index++)
- {
- VALUE element = rb_ary_entry($input, index);
- $1.bytes[16 - (index + 1)] = FIX2INT(element);
- }
-}
-
-%typemap (out) pn_decimal128_t
-{
- int index;
-
- $result = rb_ary_new2(16);
- for(index = 0; index < 16; index++)
- {
- rb_ary_store($result, 16 - (index + 1), CHR2FIX($1.bytes[index]));
- }
-}
-
-%typemap (in) pn_uuid_t
-{
- int index;
-
- for(index = 0; index < 16; index++)
- {
- VALUE element = rb_ary_entry($input, index);
- $1.bytes[16 - (index + 1)] = FIX2INT(element);
- }
-}
-
-%typemap (out) pn_uuid_t
-{
- int index;
-
- $result = rb_ary_new2(16);
- for(index = 0; index < 16; index++)
- {
- rb_ary_store($result, 16 - (index + 1), CHR2FIX($1.bytes[index]));
- }
-}
-
-int pn_message_encode(pn_message_t *msg, char *OUTPUT, size_t *OUTPUT_SIZE);
-%ignore pn_message_encode;
-
-ssize_t pn_link_send(pn_link_t *transport, char *STRING, size_t LENGTH);
-%ignore pn_link_send;
-
-%rename(pn_link_recv) wrap_pn_link_recv;
-%inline %{
- int wrap_pn_link_recv(pn_link_t *link, char *OUTPUT, size_t *OUTPUT_SIZE) {
- ssize_t sz = pn_link_recv(link, OUTPUT, *OUTPUT_SIZE);
- if (sz >= 0) {
- *OUTPUT_SIZE = sz;
- } else {
- *OUTPUT_SIZE = 0;
- }
- return sz;
- }
-%}
-%ignore pn_link_recv;
-
-ssize_t pn_transport_input(pn_transport_t *transport, char *STRING, size_t LENGTH);
-%ignore pn_transport_input;
-
-%rename(pn_transport_output) wrap_pn_transport_output;
-%inline %{
- int wrap_pn_transport_output(pn_transport_t *transport, char *OUTPUT, size_t *OUTPUT_SIZE) {
- ssize_t sz = pn_transport_output(transport, OUTPUT, *OUTPUT_SIZE);
- if (sz >= 0) {
- *OUTPUT_SIZE = sz;
- } else {
- *OUTPUT_SIZE = 0;
- }
- return sz;
- }
-%}
-%ignore pn_transport_output;
-
-%rename(pn_transport_peek) wrap_pn_transport_peek;
-%inline %{
- int wrap_pn_transport_peek(pn_transport_t *transport, char *OUTPUT, size_t *OUTPUT_SIZE) {
- ssize_t sz = pn_transport_peek(transport, OUTPUT, *OUTPUT_SIZE);
- if(sz >= 0) {
- *OUTPUT_SIZE = sz;
- } else {
- *OUTPUT_SIZE = 0;
- }
- return sz;
- }
-%}
-%ignore pn_transport_peek;
-
-%rename(pn_delivery) wrap_pn_delivery;
-%inline %{
- pn_delivery_t *wrap_pn_delivery(pn_link_t *link, char *STRING, size_t LENGTH) {
- return pn_delivery(link, pn_dtag(STRING, LENGTH));
- }
-%}
-%ignore pn_delivery;
-
-// Suppress "Warning(451): Setting a const char * variable may leak memory." on pn_delivery_tag_t
-%warnfilter(451) pn_delivery_tag_t;
-%rename(pn_delivery_tag) wrap_pn_delivery_tag;
-%inline %{
- void wrap_pn_delivery_tag(pn_delivery_t *delivery, char **ALLOC_OUTPUT, size_t *ALLOC_SIZE) {
- pn_delivery_tag_t tag = pn_delivery_tag(delivery);
- *ALLOC_OUTPUT = malloc(tag.size);
- *ALLOC_SIZE = tag.size;
- memcpy(*ALLOC_OUTPUT, tag.start, tag.size);
- }
-%}
-%ignore pn_delivery_tag;
-
-bool pn_ssl_get_cipher_name(pn_ssl_t *ssl, char *OUTPUT, size_t MAX_OUTPUT_SIZE);
-%ignore pn_ssl_get_cipher_name;
-
-bool pn_ssl_get_protocol_name(pn_ssl_t *ssl, char *OUTPUT, size_t MAX_OUTPUT_SIZE);
-%ignore pn_ssl_get_protocol_name;
-
-%inline %{
-#if defined(RUBY20) || defined(RUBY21)
-
- typedef void *non_blocking_return_t;
-#define RB_BLOCKING_CALL rb_thread_call_without_gvl
-
-#elif defined(RUBY19)
-
- typedef VALUE non_blocking_return_t;
-#define RB_BLOCKING_CALL rb_thread_blocking_region
-
-#endif
- %}
-
-%rename(pn_messenger_send) wrap_pn_messenger_send;
-%rename(pn_messenger_recv) wrap_pn_messenger_recv;
-%rename(pn_messenger_work) wrap_pn_messenger_work;
-
-%inline %{
-
-#if defined(RB_BLOCKING_CALL)
-
- static non_blocking_return_t pn_messenger_send_no_gvl(void *args) {
- VALUE result = Qnil;
- pn_messenger_t *messenger = (pn_messenger_t *)((void **)args)[0];
- int *limit = (int *)((void **)args)[1];
-
- int rc = pn_messenger_send(messenger, *limit);
-
- result = INT2NUM(rc);
- return (non_blocking_return_t )result;
- }
-
- static non_blocking_return_t pn_messenger_recv_no_gvl(void *args) {
- VALUE result = Qnil;
- pn_messenger_t *messenger = (pn_messenger_t *)((void **)args)[0];
- int *limit = (int *)((void **)args)[1];
-
- int rc = pn_messenger_recv(messenger, *limit);
-
- result = INT2NUM(rc);
- return (non_blocking_return_t )result;
- }
-
- static non_blocking_return_t pn_messenger_work_no_gvl(void *args) {
- VALUE result = Qnil;
- pn_messenger_t *messenger = (pn_messenger_t *)((void **)args)[0];
- int *timeout = (int *)((void **)args)[1];
-
- int rc = pn_messenger_work(messenger, *timeout);
-
- result = INT2NUM(rc);
- return (non_blocking_return_t )result;
- }
-
-#endif
-
- int wrap_pn_messenger_send(pn_messenger_t *messenger, int limit) {
- int result = 0;
-
-#if defined(RB_BLOCKING_CALL)
-
- // only release the gil if we're blocking
- if(pn_messenger_is_blocking(messenger)) {
- VALUE rc;
- void* args[2];
-
- args[0] = messenger;
- args[1] = &limit;
-
- rc = RB_BLOCKING_CALL(pn_messenger_send_no_gvl,
- &args, RUBY_UBF_PROCESS, NULL);
-
- if(RTEST(rc))
- {
- result = FIX2INT(rc);
- }
- }
-
-#else // !defined(RB_BLOCKING_CALL)
- result = pn_messenger_send(messenger, limit);
-#endif // defined(RB_BLOCKING_CALL)
-
- return result;
- }
-
- int wrap_pn_messenger_recv(pn_messenger_t *messenger, int limit) {
- int result = 0;
-
-#if defined(RB_BLOCKING_CALL)
- // only release the gil if we're blocking
- if(pn_messenger_is_blocking(messenger)) {
- VALUE rc;
- void* args[2];
-
- args[0] = messenger;
- args[1] = &limit;
-
- rc = RB_BLOCKING_CALL(pn_messenger_recv_no_gvl,
- &args, RUBY_UBF_PROCESS, NULL);
-
- if(RTEST(rc))
- {
- result = FIX2INT(rc);
- }
-
- } else {
- result = pn_messenger_recv(messenger, limit);
- }
-#else // !defined(RB_BLOCKING_CALL)
- result = pn_messenger_recv(messenger, limit);
-#endif // defined(RB_BLOCKING_CALL)
-
- return result;
- }
-
- int wrap_pn_messenger_work(pn_messenger_t *messenger, int timeout) {
- int result = 0;
-
-#if defined(RB_BLOCKING_CALL)
- // only release the gil if we're blocking
- if(timeout) {
- VALUE rc;
- void* args[2];
-
- args[0] = messenger;
- args[1] = &timeout;
-
- rc = RB_BLOCKING_CALL(pn_messenger_work_no_gvl,
- &args, RUBY_UBF_PROCESS, NULL);
-
- if(RTEST(rc))
- {
- result = FIX2INT(rc);
- }
- } else {
- result = pn_messenger_work(messenger, timeout);
- }
-#else
- result = pn_messenger_work(messenger, timeout);
-#endif
-
- return result;
- }
-
-%}
-
-%ignore pn_messenger_send;
-%ignore pn_messenger_recv;
-%ignore pn_messenger_work;
-
-%{
-typedef struct Pn_rbkey_t {
- void *registry;
- char *method;
- char *key_value;
-} Pn_rbkey_t;
-
-void Pn_rbkey_initialize(void *vp_rbkey) {
- Pn_rbkey_t *rbkey = (Pn_rbkey_t*)vp_rbkey;
- assert(rbkey);
- rbkey->registry = NULL;
- rbkey->method = NULL;
- rbkey->key_value = NULL;
-}
-
-void Pn_rbkey_finalize(void *vp_rbkey) {
- Pn_rbkey_t *rbkey = (Pn_rbkey_t*)vp_rbkey;
- if(rbkey && rbkey->registry && rbkey->method && rbkey->key_value) {
- rb_funcall((VALUE )rbkey->registry, rb_intern(rbkey->method), 1, rb_str_new2(rbkey->key_value));
- }
- if(rbkey->key_value) {
- free(rbkey->key_value);
- rbkey->key_value = NULL;
- }
-}
-
-/* NOTE: no macro or preprocessor definitions in %inline sections */
-#define CID_Pn_rbkey CID_pn_void
-#define Pn_rbkey_inspect NULL
-#define Pn_rbkey_compare NULL
-#define Pn_rbkey_hashcode NULL
-
-pn_class_t* Pn_rbkey__class(void) {
- static pn_class_t clazz = PN_CLASS(Pn_rbkey);
- return &clazz;
-}
-
-Pn_rbkey_t *Pn_rbkey_new(void) {
- return (Pn_rbkey_t *) pn_class_new(Pn_rbkey__class(), sizeof(Pn_rbkey_t));
-}
-%}
-
-pn_class_t* Pn_rbkey__class(void);
-Pn_rbkey_t *Pn_rbkey_new(void);
-
-%inline %{
-
-Pn_rbkey_t *Pn_rbkey_new(void);
-
-void Pn_rbkey_set_registry(Pn_rbkey_t *rbkey, void *registry) {
- assert(rbkey);
- rbkey->registry = registry;
-}
-
-void *Pn_rbkey_get_registry(Pn_rbkey_t *rbkey) {
- assert(rbkey);
- return rbkey->registry;
-}
-
-void Pn_rbkey_set_method(Pn_rbkey_t *rbkey, char *method) {
- assert(rbkey);
- rbkey->method = method;
-}
-
-char *Pn_rbkey_get_method(Pn_rbkey_t *rbkey) {
- assert(rbkey);
- return rbkey->method;
-}
-
-void Pn_rbkey_set_key_value(Pn_rbkey_t *rbkey, char *key_value) {
- assert(rbkey);
- rbkey->key_value = malloc(strlen(key_value) + 1);
- strncpy(rbkey->key_value, key_value, strlen(key_value) + 1);
-}
-
-char *Pn_rbkey_get_key_value(Pn_rbkey_t *rbkey) {
- assert(rbkey);
- return rbkey->key_value;
-}
-
-Pn_rbkey_t *pni_void2rbkey(void *object) {
- return (Pn_rbkey_t *)object;
-}
-
-VALUE pn_void2rb(void *object) {
- return (VALUE )object;
-}
-
-void *pn_rb2void(VALUE object) {
- return (void *)object;
-}
-
-VALUE pni_address_of(void *object) {
- return ULL2NUM((unsigned long )object);
-}
-
-%}
-
-//%rename(pn_collector_put) wrap_pn_collector_put;
-//%inline %{
-// pn_event_t *wrap_pn_collector_put(pn_collector_t *collector, void *context,
-// pn_event_type_t type) {
-// return pn_collector_put(collector, PN_RBREF, context, type);
-// }
-// %}
-//%ignore pn_collector_put;
-
-int pn_ssl_get_peer_hostname(pn_ssl_t *ssl, char *OUTPUT, size_t *OUTPUT_SIZE);
-%ignore pn_ssl_get_peer_hostname;
-
-%inline %{
-
- VALUE pni_ruby_get_proton_module() {
- VALUE mQpid = rb_define_module("Qpid");
- return rb_define_module_under(mQpid, "Proton");
- }
-
- void pni_ruby_add_to_registry(VALUE key, VALUE value) {
- VALUE result = rb_funcall(pni_ruby_get_proton_module(), rb_intern("add_to_registry"), 2, key, value);
- }
-
- VALUE pni_ruby_get_from_registry(VALUE key) {
- return rb_funcall(pni_ruby_get_proton_module(), rb_intern("get_from_registry"), 1, key);
- }
-
- void pni_ruby_delete_from_registry(VALUE stored_key) {
- rb_funcall(pni_ruby_get_proton_module(), rb_intern("delete_from_registry"), 1, stored_key);
- }
-
- typedef struct {
- VALUE handler_key;
- } Pni_rbhandler_t;
-
- static Pni_rbhandler_t *pni_rbhandler(pn_handler_t *handler) {
- return (Pni_rbhandler_t *) pn_handler_mem(handler);
- }
-
- static void pni_rbdispatch(pn_handler_t *handler, pn_event_t *event, pn_event_type_t type) {
- Pni_rbhandler_t *rbh = pni_rbhandler(handler);
- VALUE rbhandler = pni_ruby_get_from_registry(rbh->handler_key);
-
- rb_funcall(rbhandler, rb_intern("dispatch"), 2, SWIG_NewPointerObj(event, SWIGTYPE_p_pn_event_t, 0), INT2FIX(type));
- }
-
- static void pni_rbhandler_finalize(pn_handler_t *handler) {
- Pni_rbhandler_t *rbh = pni_rbhandler(handler);
- pni_ruby_delete_from_registry(rbh->handler_key);
- }
-
- pn_handler_t *pn_rbhandler(VALUE handler) {
- pn_handler_t *chandler = pn_handler_new(pni_rbdispatch, sizeof(Pni_rbhandler_t), pni_rbhandler_finalize);
- Pni_rbhandler_t *rhy = pni_rbhandler(chandler);
-
- VALUE ruby_key = rb_class_new_instance(0, NULL, rb_cObject);
- pni_ruby_add_to_registry(ruby_key, handler);
-
- rhy->handler_key = ruby_key;
-
- return chandler;
- }
-
-%}
-
-%include "proton/cproton.i"
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/9bb1baad/proton-c/bindings/ruby/tests/test_connection_driver.rb
----------------------------------------------------------------------
diff --git a/proton-c/bindings/ruby/tests/test_connection_driver.rb b/proton-c/bindings/ruby/tests/test_connection_driver.rb
new file mode 100644
index 0000000..2ddc8ef
--- /dev/null
+++ b/proton-c/bindings/ruby/tests/test_connection_driver.rb
@@ -0,0 +1,70 @@
+#--
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#++
+
+require 'test_tools'
+
+include Qpid::Proton
+
+class ConnectionDriverTest < Minitest::Test
+
+ def setup
+ @sockets = Socket.pair(:LOCAL, :STREAM, 0)
+ end
+
+ def test_send_recv
+ send_class = Class.new(MessagingHandler) do
+ attr_reader :accepted
+ def on_sendable(event) event.sender.send Message.new("foo"); end
+ def on_accepted(event) event.connection.close; @accepted = true; end
+ end
+
+ recv_class = Class.new(MessagingHandler) do
+ attr_reader :message
+ def on_link_opened(event) event.link.flow(1); event.link.open; end
+ def on_message(event) @message = event.message; event.connection.close; end
+ end
+
+ sender = ConnectionDriver.new(@sockets[0], send_class.new)
+ sender.connection.open();
+ sender.connection.open_sender()
+
+ receiver = ConnectionDriver.new(@sockets[1], recv_class.new)
+ drivers = [sender, receiver]
+ until drivers.all? { |d| d.finished? }
+ rd = drivers.select {|d| d.can_read? }
+ wr = drivers.select {|d| d.can_write? }
+ rs, ws = IO.select(rd, wr)
+ ws.each { |d| d.write; d.dispatch }
+ rs.each { |d| d.read; d.dispatch }
+ end
+ assert_equal(receiver.handler.message.body, "foo")
+ assert(sender.handler.accepted)
+ end
+
+ def test_idle
+ idle_class = Class.new(MessagingHandler) do
+ def on_connection_bound(event) event.transport.idle_timeout = 10; end
+ end
+ drivers = [ConnectionDriver.new(@sockets[0], idle_class.new), ConnectionDriver.new(@sockets[1])]
+ drivers[0].connection.open()
+ now = Time.now
+ drivers.each { |d| d.process(true, true, now) } until drivers[0].connection.open?
+ assert_equal(10, drivers[0].transport.idle_timeout)
+ assert_in_delta(10, (drivers[0].tick(now) - now)*1000, 1)
+ end
+end
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/9bb1baad/proton-c/include/proton/cproton.i
----------------------------------------------------------------------
diff --git a/proton-c/include/proton/cproton.i b/proton-c/include/proton/cproton.i
index 931437e..5f375de 100644
--- a/proton-c/include/proton/cproton.i
+++ b/proton-c/include/proton/cproton.i
@@ -1053,7 +1053,18 @@ typedef unsigned long int uintptr_t;
pn_selectable_t *pn_cast_pn_selectable(void *x) { return (pn_selectable_t *) x; }
%}
-%include "proton/url.h"
+/* Connection driver */
+%{
+#include <proton/connection_driver.h>
+%}
+/* Don't wrap the pn_connection_driver_t struct, just the functions */
+%ignore pn_connection_driver_t;
+%ignore pn_connection_driver_verrorf;
+%ignore pn_connection_driver_logf;
+%ignore pn_connection_driver_vlogf;
+%include "proton/connection_driver.h"
+
+%include "proton/url.h"
%include "proton/reactor.h"
%include "proton/handlers.h"
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org