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 2018/07/03 22:13:25 UTC
[36/89] [abbrv] [partial] qpid-proton git commit: PROTON-1728:
Reorganize the source tree
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/37136940/c/src/messenger/messenger.c
----------------------------------------------------------------------
diff --git a/c/src/messenger/messenger.c b/c/src/messenger/messenger.c
new file mode 100644
index 0000000..39c00c6
--- /dev/null
+++ b/c/src/messenger/messenger.c
@@ -0,0 +1,2439 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include <proton/messenger.h>
+
+#include <proton/connection.h>
+#include <proton/delivery.h>
+#include <proton/event.h>
+#include <proton/object.h>
+#include <proton/sasl.h>
+#include <proton/session.h>
+#include <proton/url.h>
+
+#include <assert.h>
+#include <ctype.h>
+#include <stdlib.h>
+#include <string.h>
+#include <stdio.h>
+
+#include "core/log_private.h"
+#include "core/util.h"
+#include "platform/platform.h" // pn_i_getpid, pn_i_now, pni_snprintf
+#include "platform/platform_fmt.h"
+#include "store.h"
+#include "subscription.h"
+#include "transform.h"
+
+#include "reactor/io.h"
+#include "reactor/selectable.h"
+#include "reactor/selector.h"
+
+typedef struct pn_link_ctx_t pn_link_ctx_t;
+
+typedef struct {
+ pn_string_t *text;
+ bool passive;
+ char *scheme;
+ char *user;
+ char *pass;
+ char *host;
+ char *port;
+ char *name;
+} pn_address_t;
+
+// algorithm for granting credit to receivers
+typedef enum {
+ // pn_messenger_recv( X ), where:
+ LINK_CREDIT_EXPLICIT, // X > 0
+ LINK_CREDIT_AUTO, // X == -1
+ LINK_CREDIT_MANUAL // X == -2
+} pn_link_credit_mode_t;
+
+struct pn_messenger_t {
+ pn_address_t address;
+ char *name;
+ char *certificate;
+ char *private_key;
+ char *password;
+ char *trusted_certificates;
+ pn_io_t *io;
+ pn_list_t *pending; // pending selectables
+ pn_selectable_t *interruptor;
+ pn_socket_t ctrl[2];
+ pn_list_t *listeners;
+ pn_list_t *connections;
+ pn_selector_t *selector;
+ pn_collector_t *collector;
+ pn_list_t *credited;
+ pn_list_t *blocked;
+ pn_timestamp_t next_drain;
+ uint64_t next_tag;
+ pni_store_t *outgoing;
+ pni_store_t *incoming;
+ pn_list_t *subscriptions;
+ pn_subscription_t *incoming_subscription;
+ pn_error_t *error;
+ pn_transform_t *routes;
+ pn_transform_t *rewrites;
+ pn_tracker_t outgoing_tracker;
+ pn_tracker_t incoming_tracker;
+ pn_string_t *original;
+ pn_string_t *rewritten;
+ pn_string_t *domain;
+ int timeout;
+ int send_threshold;
+ pn_link_credit_mode_t credit_mode;
+ int credit_batch; // when LINK_CREDIT_AUTO
+ int credit; // available
+ int distributed; // credit
+ int receivers; // # receiver links
+ int draining; // # links in drain state
+ int connection_error;
+ int flags;
+ int snd_settle_mode; /* pn_snd_settle_mode_t or -1 for unset */
+ pn_rcv_settle_mode_t rcv_settle_mode;
+ pn_tracer_t tracer;
+ pn_ssl_verify_mode_t ssl_peer_authentication_mode;
+ bool blocking;
+ bool passive;
+ bool interrupted;
+ bool worked;
+};
+
+#define CTX_HEAD \
+ pn_messenger_t *messenger; \
+ pn_selectable_t *selectable; \
+ bool pending;
+
+typedef struct pn_ctx_t {
+ CTX_HEAD
+} pn_ctx_t;
+
+typedef struct {
+ CTX_HEAD
+ char *host;
+ char *port;
+ pn_subscription_t *subscription;
+ pn_ssl_domain_t *domain;
+} pn_listener_ctx_t;
+
+typedef struct {
+ CTX_HEAD
+ pn_connection_t *connection;
+ char *address;
+ char *scheme;
+ char *user;
+ char *pass;
+ char *host;
+ char *port;
+ pn_listener_ctx_t *listener;
+} pn_connection_ctx_t;
+
+static pn_connection_ctx_t *pni_context(pn_selectable_t *sel)
+{
+ assert(sel);
+ pn_connection_ctx_t *ctx = (pn_connection_ctx_t *) pni_selectable_get_context(sel);
+ assert(ctx);
+ return ctx;
+}
+
+static pn_transport_t *pni_transport(pn_selectable_t *sel)
+{
+ return pn_connection_transport(pni_context(sel)->connection);
+}
+
+static ssize_t pni_connection_capacity(pn_selectable_t *sel)
+{
+ pn_transport_t *transport = pni_transport(sel);
+ ssize_t capacity = pn_transport_capacity(transport);
+ if (capacity < 0) {
+ if (pn_transport_closed(transport)) {
+ pn_selectable_terminate(sel);
+ }
+ }
+ return capacity;
+}
+
+bool pn_messenger_flow(pn_messenger_t *messenger);
+
+static ssize_t pni_connection_pending(pn_selectable_t *sel)
+{
+ pn_connection_ctx_t *ctx = pni_context(sel);
+ pn_messenger_flow(ctx->messenger);
+ pn_transport_t *transport = pni_transport(sel);
+ ssize_t pending = pn_transport_pending(transport);
+ if (pending < 0) {
+ if (pn_transport_closed(transport)) {
+ pn_selectable_terminate(sel);
+ }
+ }
+ return pending;
+}
+
+static pn_timestamp_t pni_connection_deadline(pn_selectable_t *sel)
+{
+ pn_connection_ctx_t *ctx = pni_context(sel);
+ return ctx->messenger->next_drain;
+}
+
+static void pni_connection_update(pn_selectable_t *sel) {
+ ssize_t c = pni_connection_capacity(sel);
+ pn_selectable_set_reading(sel, c > 0);
+ ssize_t p = pni_connection_pending(sel);
+ pn_selectable_set_writing(sel, p > 0);
+ pn_selectable_set_deadline(sel, pni_connection_deadline(sel));
+ if (c < 0 && p < 0) {
+ pn_selectable_terminate(sel);
+ }
+}
+
+#include <errno.h>
+
+static void pn_error_report(const char *pfx, const char *error)
+{
+ pn_logf("%s ERROR %s", pfx, error);
+}
+
+void pni_modified(pn_ctx_t *ctx)
+{
+ pn_messenger_t *m = ctx->messenger;
+ pn_selectable_t *sel = ctx->selectable;
+ if (pn_selectable_is_registered(sel) && !ctx->pending) {
+ pn_list_add(m->pending, sel);
+ ctx->pending = true;
+ }
+}
+
+void pni_conn_modified(pn_connection_ctx_t *ctx)
+{
+ pni_connection_update(ctx->selectable);
+ pni_modified((pn_ctx_t *) ctx);
+}
+
+void pni_lnr_modified(pn_listener_ctx_t *lnr)
+{
+ pni_modified((pn_ctx_t *) lnr);
+}
+
+int pn_messenger_process_events(pn_messenger_t *messenger);
+
+static void pni_connection_error(pn_selectable_t *sel)
+{
+ pn_transport_t *transport = pni_transport(sel);
+ pn_transport_close_tail(transport);
+ pn_transport_close_head(transport);
+}
+
+static void pni_connection_readable(pn_selectable_t *sel)
+{
+ pn_connection_ctx_t *context = pni_context(sel);
+ pn_messenger_t *messenger = context->messenger;
+ pn_connection_t *connection = context->connection;
+ pn_transport_t *transport = pni_transport(sel);
+ ssize_t capacity = pn_transport_capacity(transport);
+ if (capacity > 0) {
+ ssize_t n = pn_recv(messenger->io, pn_selectable_get_fd(sel),
+ pn_transport_tail(transport), capacity);
+ if (n <= 0) {
+ if (n == 0 || !pn_wouldblock(messenger->io)) {
+ pn_transport_close_tail(transport);
+ if (!(pn_connection_state(connection) & PN_REMOTE_CLOSED)) {
+ pn_error_report("CONNECTION", "connection aborted (remote)");
+ }
+ }
+ } else {
+ int err = pn_transport_process(transport, (size_t)n);
+ if (err)
+ pn_error_copy(messenger->error, pn_transport_error(transport));
+ }
+ }
+
+ pn_messenger_process_events(messenger);
+ pn_messenger_flow(messenger);
+ messenger->worked = true;
+ pni_conn_modified(context);
+}
+
+static void pni_connection_writable(pn_selectable_t *sel)
+{
+ pn_connection_ctx_t *context = pni_context(sel);
+ pn_messenger_t *messenger = context->messenger;
+ pn_transport_t *transport = pni_transport(sel);
+ ssize_t pending = pn_transport_pending(transport);
+ if (pending > 0) {
+ ssize_t n = pn_send(messenger->io, pn_selectable_get_fd(sel),
+ pn_transport_head(transport), pending);
+ if (n < 0) {
+ if (!pn_wouldblock(messenger->io)) {
+ pn_transport_close_head(transport);
+ }
+ } else {
+ pn_transport_pop(transport, n);
+ }
+ }
+
+ pn_messenger_process_events(messenger);
+ pn_messenger_flow(messenger);
+ messenger->worked = true;
+ pni_conn_modified(context);
+}
+
+static void pni_connection_expired(pn_selectable_t *sel)
+{
+ pn_connection_ctx_t *ctx = pni_context(sel);
+ pn_messenger_flow(ctx->messenger);
+ ctx->messenger->worked = true;
+ pni_conn_modified(ctx);
+}
+
+static void pni_messenger_reclaim(pn_messenger_t *messenger, pn_connection_t *conn);
+
+static void pni_connection_finalize(pn_selectable_t *sel)
+{
+ pn_connection_ctx_t *ctx = (pn_connection_ctx_t *) pni_selectable_get_context(sel);
+ pn_socket_t fd = pn_selectable_get_fd(sel);
+ pn_close(ctx->messenger->io, fd);
+ pn_list_remove(ctx->messenger->pending, sel);
+ pni_messenger_reclaim(ctx->messenger, ctx->connection);
+}
+
+pn_connection_t *pn_messenger_connection(pn_messenger_t *messenger,
+ pn_socket_t sock,
+ const char *scheme,
+ char *user,
+ char *pass,
+ char *host,
+ char *port,
+ pn_listener_ctx_t *lnr);
+
+static void pni_listener_readable(pn_selectable_t *sel)
+{
+ pn_listener_ctx_t *ctx = (pn_listener_ctx_t *) pni_selectable_get_context(sel);
+ pn_subscription_t *sub = ctx->subscription;
+ const char *scheme = pn_subscription_scheme(sub);
+ char name[1024];
+ pn_socket_t sock = pn_accept(ctx->messenger->io, pn_selectable_get_fd(sel), name, 1024);
+
+ pn_transport_t *t = pn_transport();
+ pn_transport_set_server(t);
+ if (ctx->messenger->flags & PN_FLAGS_ALLOW_INSECURE_MECHS) {
+ pn_sasl_t *s = pn_sasl(t);
+ pn_sasl_set_allow_insecure_mechs(s, true);
+ }
+ pn_ssl_t *ssl = pn_ssl(t);
+ pn_ssl_init(ssl, ctx->domain, NULL);
+
+ pn_connection_t *conn = pn_messenger_connection(ctx->messenger, sock, scheme, NULL, NULL, NULL, NULL, ctx);
+ pn_transport_bind(t, conn);
+ pn_decref(t);
+ pni_conn_modified((pn_connection_ctx_t *) pn_connection_get_context(conn));
+}
+
+static void pn_listener_ctx_free(pn_messenger_t *messenger, pn_listener_ctx_t *ctx);
+
+static void pni_listener_finalize(pn_selectable_t *sel)
+{
+ pn_listener_ctx_t *lnr = (pn_listener_ctx_t *) pni_selectable_get_context(sel);
+ pn_messenger_t *messenger = lnr->messenger;
+ pn_close(messenger->io, pn_selectable_get_fd(sel));
+ pn_list_remove(messenger->pending, sel);
+ pn_listener_ctx_free(messenger, lnr);
+}
+
+static bool pn_streq(const char *a, const char *b)
+{
+ return a == b || (a && b && !strcmp(a, b));
+}
+
+static const char *default_port(const char *scheme)
+{
+ if (scheme && pn_streq(scheme, "amqps"))
+ return "5671";
+ else
+ return "5672";
+}
+
+static pn_listener_ctx_t *pn_listener_ctx(pn_messenger_t *messenger,
+ const char *scheme,
+ const char *host,
+ const char *port)
+{
+ pn_socket_t socket = pn_listen(messenger->io, host, port ? port : default_port(scheme));
+ if (socket == PN_INVALID_SOCKET) {
+ pn_error_copy(messenger->error, pn_io_error(messenger->io));
+ pn_error_format(messenger->error, PN_ERR, "CONNECTION ERROR (%s:%s): %s\n",
+ messenger->address.host, messenger->address.port,
+ pn_error_text(messenger->error));
+
+ return NULL;
+ }
+
+ pn_listener_ctx_t *ctx = (pn_listener_ctx_t *) pn_class_new(PN_OBJECT, sizeof(pn_listener_ctx_t));
+ ctx->messenger = messenger;
+ ctx->domain = pn_ssl_domain(PN_SSL_MODE_SERVER);
+ if (messenger->certificate) {
+ int err = pn_ssl_domain_set_credentials(ctx->domain, messenger->certificate,
+ messenger->private_key,
+ messenger->password);
+
+ if (err) {
+ pn_error_format(messenger->error, PN_ERR, "invalid credentials");
+ pn_ssl_domain_free(ctx->domain);
+ pn_free(ctx);
+ pn_close(messenger->io, socket);
+ return NULL;
+ }
+ }
+
+ if (!(scheme && !strcmp(scheme, "amqps"))) {
+ pn_ssl_domain_allow_unsecured_client(ctx->domain);
+ }
+
+ pn_subscription_t *sub = pn_subscription(messenger, scheme, host, port);
+ ctx->subscription = sub;
+ ctx->host = pn_strdup(host);
+ ctx->port = pn_strdup(port);
+
+ pn_selectable_t *selectable = pn_selectable();
+ pn_selectable_set_reading(selectable, true);
+ pn_selectable_on_readable(selectable, pni_listener_readable);
+ pn_selectable_on_release(selectable, pn_selectable_free);
+ pn_selectable_on_finalize(selectable, pni_listener_finalize);
+ pn_selectable_set_fd(selectable, socket);
+ pni_selectable_set_context(selectable, ctx);
+ pn_list_add(messenger->pending, selectable);
+ ctx->selectable = selectable;
+ ctx->pending = true;
+
+ pn_list_add(messenger->listeners, ctx);
+ return ctx;
+}
+
+static void pn_listener_ctx_free(pn_messenger_t *messenger, pn_listener_ctx_t *ctx)
+{
+ pn_list_remove(messenger->listeners, ctx);
+ // XXX: subscriptions are freed when the messenger is freed pn_subscription_free(ctx->subscription);
+ free(ctx->host);
+ free(ctx->port);
+ pn_ssl_domain_free(ctx->domain);
+ pn_free(ctx);
+}
+
+static pn_connection_ctx_t *pn_connection_ctx(pn_messenger_t *messenger,
+ pn_connection_t *conn,
+ pn_socket_t sock,
+ const char *scheme,
+ const char *user,
+ const char *pass,
+ const char *host,
+ const char *port,
+ pn_listener_ctx_t *lnr)
+{
+ pn_connection_ctx_t *ctx = (pn_connection_ctx_t *) pn_connection_get_context(conn);
+ assert(!ctx);
+ ctx = (pn_connection_ctx_t *) malloc(sizeof(pn_connection_ctx_t));
+ ctx->messenger = messenger;
+ ctx->connection = conn;
+ pn_selectable_t *sel = pn_selectable();
+ ctx->selectable = sel;
+ pn_selectable_on_error(sel, pni_connection_error);
+ pn_selectable_on_readable(sel, pni_connection_readable);
+ pn_selectable_on_writable(sel, pni_connection_writable);
+ pn_selectable_on_expired(sel, pni_connection_expired);
+ pn_selectable_on_release(sel, pn_selectable_free);
+ pn_selectable_on_finalize(sel, pni_connection_finalize);
+ pn_selectable_set_fd(ctx->selectable, sock);
+ pni_selectable_set_context(ctx->selectable, ctx);
+ pn_list_add(messenger->pending, ctx->selectable);
+ ctx->pending = true;
+ ctx->scheme = pn_strdup(scheme);
+ ctx->user = pn_strdup(user);
+ ctx->pass = pn_strdup(pass);
+ ctx->host = pn_strdup(host);
+ ctx->port = pn_strdup(port);
+ ctx->listener = lnr;
+ pn_connection_set_context(conn, ctx);
+ return ctx;
+}
+
+static void pn_connection_ctx_free(pn_connection_t *conn)
+{
+ pn_connection_ctx_t *ctx = (pn_connection_ctx_t *) pn_connection_get_context(conn);
+ if (ctx) {
+ pni_selectable_set_context(ctx->selectable, NULL);
+ free(ctx->scheme);
+ free(ctx->user);
+ free(ctx->pass);
+ free(ctx->host);
+ free(ctx->port);
+ free(ctx);
+ pn_connection_set_context(conn, NULL);
+ }
+}
+
+#define OUTGOING (0x0000000000000000)
+#define INCOMING (0x1000000000000000)
+
+#define pn_tracker(direction, sequence) ((direction) | (sequence))
+#define pn_tracker_direction(tracker) ((tracker) & (0x1000000000000000))
+#define pn_tracker_sequence(tracker) ((pn_sequence_t) ((tracker) & (0x00000000FFFFFFFF)))
+
+
+static char *build_name(const char *name)
+{
+ static bool seeded = false;
+ // UUID standard format: 8-4-4-4-12 (36 chars, 32 alphanumeric and 4 hypens)
+ static const char *uuid_fmt = "%02X%02X%02X%02X-%02X%02X-%02X%02X-%02X%02X-%02X%02X%02X%02X%02X%02X";
+
+ int count = 0;
+ char *generated;
+ uint8_t bytes[16];
+ unsigned int r = 0;
+
+ if (name) {
+ return pn_strdup(name);
+ }
+
+ if (!seeded) {
+ int pid = pn_i_getpid();
+ int nowish = (int)pn_i_now();
+ // the lower bits of time are the most random, shift pid to push some
+ // randomness into the higher order bits
+ srand(nowish ^ (pid<<16));
+ seeded = true;
+ }
+
+ while (count < 16) {
+ if (!r) {
+ r = (unsigned int) rand();
+ }
+
+ bytes[count] = r & 0xFF;
+ r >>= 8;
+ count++;
+ }
+
+ // From RFC4122, the version bits are set to 0100
+ bytes[6] = (bytes[6] & 0x0F) | 0x40;
+
+ // From RFC4122, the top two bits of byte 8 get set to 01
+ bytes[8] = (bytes[8] & 0x3F) | 0x80;
+
+ generated = (char *) malloc(37*sizeof(char));
+ sprintf(generated, uuid_fmt,
+ bytes[0], bytes[1], bytes[2], bytes[3],
+ bytes[4], bytes[5], bytes[6], bytes[7],
+ bytes[8], bytes[9], bytes[10], bytes[11],
+ bytes[12], bytes[13], bytes[14], bytes[15]);
+ return generated;
+}
+
+struct pn_link_ctx_t {
+ pn_subscription_t *subscription;
+};
+
+// compute the maximum amount of credit each receiving link is
+// entitled to. The actual credit given to the link depends on what
+// amount of credit is actually available.
+static int per_link_credit( pn_messenger_t *messenger )
+{
+ if (messenger->receivers == 0) return 0;
+ int total = messenger->credit + messenger->distributed;
+ return pn_max(total/messenger->receivers, 1);
+}
+
+static void link_ctx_setup( pn_messenger_t *messenger,
+ pn_connection_t *connection,
+ pn_link_t *link )
+{
+ if (pn_link_is_receiver(link)) {
+ messenger->receivers++;
+ pn_link_ctx_t *ctx = (pn_link_ctx_t *) calloc(1, sizeof(pn_link_ctx_t));
+ assert( ctx );
+ assert( !pn_link_get_context(link) );
+ pn_link_set_context( link, ctx );
+ pn_list_add(messenger->blocked, link);
+ }
+}
+
+static void link_ctx_release( pn_messenger_t *messenger, pn_link_t *link )
+{
+ if (pn_link_is_receiver(link)) {
+ pn_link_ctx_t *ctx = (pn_link_ctx_t *) pn_link_get_context( link );
+ if (!ctx) return;
+ assert( messenger->receivers > 0 );
+ messenger->receivers--;
+ if (pn_link_get_drain(link)) {
+ pn_link_set_drain(link, false);
+ assert( messenger->draining > 0 );
+ messenger->draining--;
+ }
+ pn_list_remove(messenger->credited, link);
+ pn_list_remove(messenger->blocked, link);
+ pn_link_set_context( link, NULL );
+ free( ctx );
+ }
+}
+
+static void pni_interruptor_readable(pn_selectable_t *sel)
+{
+ pn_messenger_t *messenger = (pn_messenger_t *) pni_selectable_get_context(sel);
+ char buf[1024];
+ pn_read(messenger->io, pn_selectable_get_fd(sel), buf, 1024);
+ messenger->interrupted = true;
+}
+
+static void pni_interruptor_finalize(pn_selectable_t *sel)
+{
+ pn_messenger_t *messenger = (pn_messenger_t *) pni_selectable_get_context(sel);
+ messenger->interruptor = NULL;
+}
+
+pn_messenger_t *pn_messenger(const char *name)
+{
+ pn_messenger_t *m = (pn_messenger_t *) malloc(sizeof(pn_messenger_t));
+
+ if (m) {
+ m->name = build_name(name);
+ m->certificate = NULL;
+ m->private_key = NULL;
+ m->password = NULL;
+ m->trusted_certificates = NULL;
+ m->timeout = -1;
+ m->blocking = true;
+ m->passive = false;
+ m->io = pn_io();
+ m->pending = pn_list(PN_WEAKREF, 0);
+ m->interruptor = pn_selectable();
+ pn_selectable_set_reading(m->interruptor, true);
+ pn_selectable_on_readable(m->interruptor, pni_interruptor_readable);
+ pn_selectable_on_release(m->interruptor, pn_selectable_free);
+ pn_selectable_on_finalize(m->interruptor, pni_interruptor_finalize);
+ pn_list_add(m->pending, m->interruptor);
+ m->interrupted = false;
+ // Explicitly initialise pipe file descriptors to invalid values in case pipe
+ // fails, if we don't do this m->ctrl[0] could default to 0 - which is stdin.
+ m->ctrl[0] = -1;
+ m->ctrl[1] = -1;
+ pn_pipe(m->io, m->ctrl);
+ pn_selectable_set_fd(m->interruptor, m->ctrl[0]);
+ pni_selectable_set_context(m->interruptor, m);
+ m->listeners = pn_list(PN_WEAKREF, 0);
+ m->connections = pn_list(PN_WEAKREF, 0);
+ m->selector = pn_io_selector(m->io);
+ m->collector = pn_collector();
+ m->credit_mode = LINK_CREDIT_EXPLICIT;
+ m->credit_batch = 1024;
+ m->credit = 0;
+ m->distributed = 0;
+ m->receivers = 0;
+ m->draining = 0;
+ m->credited = pn_list(PN_WEAKREF, 0);
+ m->blocked = pn_list(PN_WEAKREF, 0);
+ m->next_drain = 0;
+ m->next_tag = 0;
+ m->outgoing = pni_store();
+ m->incoming = pni_store();
+ m->subscriptions = pn_list(PN_OBJECT, 0);
+ m->incoming_subscription = NULL;
+ m->error = pn_error();
+ m->routes = pn_transform();
+ m->rewrites = pn_transform();
+ m->outgoing_tracker = 0;
+ m->incoming_tracker = 0;
+ m->address.text = pn_string(NULL);
+ m->original = pn_string(NULL);
+ m->rewritten = pn_string(NULL);
+ m->domain = pn_string(NULL);
+ m->connection_error = 0;
+ m->flags = PN_FLAGS_ALLOW_INSECURE_MECHS; // TODO: Change this back to 0 for the Proton 0.11 release
+ m->snd_settle_mode = -1; /* Default depends on sender/receiver */
+ m->rcv_settle_mode = PN_RCV_FIRST;
+ m->tracer = NULL;
+ m->ssl_peer_authentication_mode = PN_SSL_VERIFY_PEER_NAME;
+ }
+
+ return m;
+}
+
+int pni_messenger_add_subscription(pn_messenger_t *messenger, pn_subscription_t *subscription)
+{
+ return pn_list_add(messenger->subscriptions, subscription);
+}
+
+
+const char *pn_messenger_name(pn_messenger_t *messenger)
+{
+ return messenger->name;
+}
+
+int pn_messenger_set_certificate(pn_messenger_t *messenger, const char *certificate)
+{
+ if (messenger->certificate) free(messenger->certificate);
+ messenger->certificate = pn_strdup(certificate);
+ return 0;
+}
+
+const char *pn_messenger_get_certificate(pn_messenger_t *messenger)
+{
+ return messenger->certificate;
+}
+
+int pn_messenger_set_private_key(pn_messenger_t *messenger, const char *private_key)
+{
+ if (messenger->private_key) free(messenger->private_key);
+ messenger->private_key = pn_strdup(private_key);
+ return 0;
+}
+
+const char *pn_messenger_get_private_key(pn_messenger_t *messenger)
+{
+ return messenger->private_key;
+}
+
+int pn_messenger_set_password(pn_messenger_t *messenger, const char *password)
+{
+ if (messenger->password) free(messenger->password);
+ messenger->password = pn_strdup(password);
+ return 0;
+}
+
+const char *pn_messenger_get_password(pn_messenger_t *messenger)
+{
+ return messenger->password;
+}
+
+int pn_messenger_set_trusted_certificates(pn_messenger_t *messenger, const char *trusted_certificates)
+{
+ if (messenger->trusted_certificates) free(messenger->trusted_certificates);
+ messenger->trusted_certificates = pn_strdup(trusted_certificates);
+ return 0;
+}
+
+const char *pn_messenger_get_trusted_certificates(pn_messenger_t *messenger)
+{
+ return messenger->trusted_certificates;
+}
+
+int pn_messenger_set_timeout(pn_messenger_t *messenger, int timeout)
+{
+ if (!messenger) return PN_ARG_ERR;
+ messenger->timeout = timeout;
+ return 0;
+}
+
+int pn_messenger_get_timeout(pn_messenger_t *messenger)
+{
+ return messenger ? messenger->timeout : 0;
+}
+
+bool pn_messenger_is_blocking(pn_messenger_t *messenger)
+{
+ assert(messenger);
+ return messenger->blocking;
+}
+
+int pn_messenger_set_blocking(pn_messenger_t *messenger, bool blocking)
+{
+ messenger->blocking = blocking;
+ return 0;
+}
+
+bool pn_messenger_is_passive(pn_messenger_t *messenger)
+{
+ assert(messenger);
+ return messenger->passive;
+}
+
+int pn_messenger_set_passive(pn_messenger_t *messenger, bool passive)
+{
+ messenger->passive = passive;
+ return 0;
+}
+
+pn_selectable_t *pn_messenger_selectable(pn_messenger_t *messenger)
+{
+ assert(messenger);
+ pn_messenger_process_events(messenger);
+ pn_list_t *p = messenger->pending;
+ size_t n = pn_list_size(p);
+ if (n) {
+ pn_selectable_t *s = (pn_selectable_t *) pn_list_get(p, n - 1);
+ pn_list_del(p, n-1, 1);
+ // this is a total hack, messenger has selectables whose context
+ // are the messenger itself and whose context share a common
+ // prefix that is described by pn_ctx_t
+ void *c = pni_selectable_get_context(s);
+ if (c != messenger) {
+ pn_ctx_t *ctx = (pn_ctx_t *) c;
+ ctx->pending = false;
+ }
+ return s;
+ } else {
+ return NULL;
+ }
+}
+
+static void pni_reclaim(pn_messenger_t *messenger)
+{
+ while (pn_list_size(messenger->listeners)) {
+ pn_listener_ctx_t *l = (pn_listener_ctx_t *) pn_list_get(messenger->listeners, 0);
+ pn_listener_ctx_free(messenger, l);
+ }
+
+ while (pn_list_size(messenger->connections)) {
+ pn_connection_t *c = (pn_connection_t *) pn_list_get(messenger->connections, 0);
+ pni_messenger_reclaim(messenger, c);
+ }
+}
+
+void pn_messenger_free(pn_messenger_t *messenger)
+{
+ if (messenger) {
+ pn_free(messenger->domain);
+ pn_free(messenger->rewritten);
+ pn_free(messenger->original);
+ pn_free(messenger->address.text);
+ free(messenger->name);
+ free(messenger->certificate);
+ free(messenger->private_key);
+ free(messenger->password);
+ free(messenger->trusted_certificates);
+ pni_reclaim(messenger);
+ pn_free(messenger->pending);
+ pn_selectable_free(messenger->interruptor);
+ pn_close(messenger->io, messenger->ctrl[0]);
+ pn_close(messenger->io, messenger->ctrl[1]);
+ pn_free(messenger->listeners);
+ pn_free(messenger->connections);
+ pn_selector_free(messenger->selector);
+ pn_collector_free(messenger->collector);
+ pn_error_free(messenger->error);
+ pni_store_free(messenger->incoming);
+ pni_store_free(messenger->outgoing);
+ pn_free(messenger->subscriptions);
+ pn_free(messenger->rewrites);
+ pn_free(messenger->routes);
+ pn_free(messenger->credited);
+ pn_free(messenger->blocked);
+ pn_free(messenger->io);
+ free(messenger);
+ }
+}
+
+int pn_messenger_errno(pn_messenger_t *messenger)
+{
+ if (messenger) {
+ return pn_error_code(messenger->error);
+ } else {
+ return PN_ARG_ERR;
+ }
+}
+
+pn_error_t *pn_messenger_error(pn_messenger_t *messenger)
+{
+ assert(messenger);
+ return messenger->error;
+}
+
+// Run the credit scheduler, grant flow as needed. Return True if
+// credit allocation for any link has changed.
+bool pn_messenger_flow(pn_messenger_t *messenger)
+{
+ bool updated = false;
+ if (messenger->receivers == 0) {
+ messenger->next_drain = 0;
+ return updated;
+ }
+
+ if (messenger->credit_mode == LINK_CREDIT_AUTO) {
+ // replenish, but limit the max total messages buffered
+ const int max = messenger->receivers * messenger->credit_batch;
+ const int used = messenger->distributed + pn_messenger_incoming(messenger);
+ if (max > used)
+ messenger->credit = max - used;
+ } else if (messenger->credit_mode == LINK_CREDIT_MANUAL) {
+ messenger->next_drain = 0;
+ return false;
+ }
+
+ const int batch = per_link_credit(messenger);
+ while (messenger->credit > 0 && pn_list_size(messenger->blocked)) {
+ pn_link_t *link = (pn_link_t *) pn_list_get(messenger->blocked, 0);
+ pn_list_del(messenger->blocked, 0, 1);
+
+ const int more = pn_min( messenger->credit, batch );
+ messenger->distributed += more;
+ messenger->credit -= more;
+ pn_link_flow(link, more);
+ pn_list_add(messenger->credited, link);
+ updated = true;
+ }
+
+ if (!pn_list_size(messenger->blocked)) {
+ messenger->next_drain = 0;
+ } else {
+ // not enough credit for all links
+ if (!messenger->draining) {
+ pn_logf("%s: let's drain", messenger->name);
+ if (messenger->next_drain == 0) {
+ messenger->next_drain = pn_i_now() + 250;
+ pn_logf("%s: initializing next_drain", messenger->name);
+ } else if (messenger->next_drain <= pn_i_now()) {
+ // initiate drain, free up at most enough to satisfy blocked
+ messenger->next_drain = 0;
+ int needed = pn_list_size(messenger->blocked) * batch;
+ for (size_t i = 0; i < pn_list_size(messenger->credited); i++) {
+ pn_link_t *link = (pn_link_t *) pn_list_get(messenger->credited, i);
+ if (!pn_link_get_drain(link)) {
+ pn_link_set_drain(link, true);
+ needed -= pn_link_remote_credit(link);
+ messenger->draining++;
+ updated = true;
+ }
+
+ if (needed <= 0) {
+ break;
+ }
+ }
+ } else {
+ pn_logf("%s: delaying", messenger->name);
+ }
+ }
+ }
+ return updated;
+}
+
+static int pn_transport_config(pn_messenger_t *messenger,
+ pn_connection_t *connection)
+{
+ pn_connection_ctx_t *ctx = (pn_connection_ctx_t *) pn_connection_get_context(connection);
+ pn_transport_t *transport = pn_connection_transport(connection);
+ if (messenger->tracer)
+ pn_transport_set_tracer(transport, messenger->tracer);
+ if (ctx->scheme && !strcmp(ctx->scheme, "amqps")) {
+ pn_ssl_domain_t *d = pn_ssl_domain(PN_SSL_MODE_CLIENT);
+ if (messenger->certificate) {
+ int err = pn_ssl_domain_set_credentials( d, messenger->certificate,
+ messenger->private_key,
+ messenger->password);
+ if (err) {
+ pn_ssl_domain_free(d);
+ pn_error_report("CONNECTION", "invalid credentials");
+ return err;
+ }
+ }
+ if (messenger->trusted_certificates) {
+ int err = pn_ssl_domain_set_trusted_ca_db(d, messenger->trusted_certificates);
+ if (err) {
+ pn_ssl_domain_free(d);
+ pn_error_report("CONNECTION", "invalid certificate db");
+ return err;
+ }
+ err = pn_ssl_domain_set_peer_authentication(
+ d, messenger->ssl_peer_authentication_mode, NULL);
+ if (err) {
+ pn_ssl_domain_free(d);
+ pn_error_report("CONNECTION", "error configuring ssl to verify peer");
+ }
+ } else {
+ int err = pn_ssl_domain_set_peer_authentication(d, PN_SSL_ANONYMOUS_PEER, NULL);
+ if (err) {
+ pn_ssl_domain_free(d);
+ pn_error_report("CONNECTION", "error configuring ssl for anonymous peer");
+ return err;
+ }
+ }
+ pn_ssl_t *ssl = pn_ssl(transport);
+ pn_ssl_init(ssl, d, NULL);
+ pn_ssl_domain_free( d );
+ }
+
+ return 0;
+}
+
+static void pn_condition_report(const char *pfx, pn_condition_t *condition)
+{
+ if (pn_condition_is_redirect(condition)) {
+ pn_logf("%s NOTICE (%s) redirecting to %s:%i",
+ pfx,
+ pn_condition_get_name(condition),
+ pn_condition_redirect_host(condition),
+ pn_condition_redirect_port(condition));
+ } else if (pn_condition_is_set(condition)) {
+ char error[1024];
+ pni_snprintf(error, 1024, "(%s) %s",
+ pn_condition_get_name(condition),
+ pn_condition_get_description(condition));
+ pn_error_report(pfx, error);
+ }
+}
+
+int pni_pump_in(pn_messenger_t *messenger, const char *address, pn_link_t *receiver)
+{
+ pn_delivery_t *d = pn_link_current(receiver);
+ if (!pn_delivery_readable(d) || pn_delivery_partial(d)) {
+ return 0;
+ }
+
+ pni_entry_t *entry = pni_store_put(messenger->incoming, address);
+ pn_buffer_t *buf = pni_entry_bytes(entry);
+ pni_entry_set_delivery(entry, d);
+
+ pn_link_ctx_t *ctx = (pn_link_ctx_t *) pn_link_get_context( receiver );
+ pni_entry_set_context(entry, ctx ? ctx->subscription : NULL);
+
+ size_t pending = pn_delivery_pending(d);
+ int err = pn_buffer_ensure(buf, pending + 1);
+ if (err) return pn_error_format(messenger->error, err, "get: error growing buffer");
+ char *encoded = pn_buffer_memory(buf).start;
+ ssize_t n = pn_link_recv(receiver, encoded, pending);
+ if (n != (ssize_t) pending) {
+ return pn_error_format(messenger->error, n,
+ "didn't receive pending bytes: %" PN_ZI " %" PN_ZI,
+ n, pending);
+ }
+ n = pn_link_recv(receiver, encoded + pending, 1);
+ pn_link_advance(receiver);
+
+ pn_link_t *link = receiver;
+
+ if (messenger->credit_mode != LINK_CREDIT_MANUAL) {
+ // account for the used credit
+ assert(ctx);
+ assert(messenger->distributed);
+ messenger->distributed--;
+
+ // replenish if low (< 20% maximum batch) and credit available
+ if (!pn_link_get_drain(link) && pn_list_size(messenger->blocked) == 0 &&
+ messenger->credit > 0) {
+ const int max = per_link_credit(messenger);
+ const int lo_thresh = (int)(max * 0.2 + 0.5);
+ if (pn_link_remote_credit(link) < lo_thresh) {
+ const int more =
+ pn_min(messenger->credit, max - pn_link_remote_credit(link));
+ messenger->credit -= more;
+ messenger->distributed += more;
+ pn_link_flow(link, more);
+ }
+ }
+ // check if blocked
+ if (pn_list_index(messenger->blocked, link) < 0 &&
+ pn_link_remote_credit(link) == 0) {
+ pn_list_remove(messenger->credited, link);
+ if (pn_link_get_drain(link)) {
+ pn_link_set_drain(link, false);
+ assert(messenger->draining > 0);
+ messenger->draining--;
+ }
+ pn_list_add(messenger->blocked, link);
+ }
+ }
+
+ if (n != PN_EOS) {
+ return pn_error_format(messenger->error, n, "PN_EOS expected");
+ }
+ pn_buffer_append(buf, encoded, pending); // XXX
+
+ return 0;
+}
+
+void pni_messenger_reclaim_link(pn_messenger_t *messenger, pn_link_t *link)
+{
+ if (pn_link_is_receiver(link) && pn_link_credit(link) > 0) {
+ int credit = pn_link_credit(link);
+ messenger->credit += credit;
+ messenger->distributed -= credit;
+ }
+
+ pn_delivery_t *d = pn_unsettled_head(link);
+ while (d) {
+ pni_entry_t *e = (pni_entry_t *) pn_delivery_get_context(d);
+ if (e) {
+ pni_entry_set_delivery(e, NULL);
+ if (pn_delivery_buffered(d)) {
+ pni_entry_set_status(e, PN_STATUS_ABORTED);
+ }
+ }
+ d = pn_unsettled_next(d);
+ }
+
+ link_ctx_release(messenger, link);
+}
+
+void pni_messenger_reclaim(pn_messenger_t *messenger, pn_connection_t *conn)
+{
+ if (!conn) return;
+
+ pn_link_t *link = pn_link_head(conn, 0);
+ while (link) {
+ pni_messenger_reclaim_link(messenger, link);
+ link = pn_link_next(link, 0);
+ }
+
+ pn_list_remove(messenger->connections, conn);
+ pn_connection_ctx_free(conn);
+ pn_transport_free(pn_connection_transport(conn));
+ pn_connection_free(conn);
+}
+
+pn_connection_t *pn_messenger_connection(pn_messenger_t *messenger,
+ pn_socket_t sock,
+ const char *scheme,
+ char *user,
+ char *pass,
+ char *host,
+ char *port,
+ pn_listener_ctx_t *lnr)
+{
+ pn_connection_t *connection = pn_connection();
+ if (!connection) return NULL;
+ pn_connection_collect(connection, messenger->collector);
+ pn_connection_ctx(messenger, connection, sock, scheme, user, pass, host, port, lnr);
+
+ pn_connection_set_container(connection, messenger->name);
+ pn_connection_set_hostname(connection, host);
+ pn_connection_set_user(connection, user);
+ pn_connection_set_password(connection, pass);
+
+ pn_list_add(messenger->connections, connection);
+
+ return connection;
+}
+
+void pn_messenger_process_connection(pn_messenger_t *messenger, pn_event_t *event)
+{
+ pn_connection_t *conn = pn_event_connection(event);
+ pn_connection_ctx_t *ctx = (pn_connection_ctx_t *) pn_connection_get_context(conn);
+
+ if (pn_connection_state(conn) & PN_LOCAL_UNINIT) {
+ pn_connection_open(conn);
+ }
+
+ if (pn_connection_state(conn) == (PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED)) {
+ pn_condition_t *condition = pn_connection_remote_condition(conn);
+ pn_condition_report("CONNECTION", condition);
+ pn_connection_close(conn);
+ if (pn_condition_is_redirect(condition)) {
+ const char *host = pn_condition_redirect_host(condition);
+ char buf[1024];
+ sprintf(buf, "%i", pn_condition_redirect_port(condition));
+
+ pn_close(messenger->io, pn_selectable_get_fd(ctx->selectable));
+ pn_socket_t sock = pn_connect(messenger->io, host, buf);
+ pn_selectable_set_fd(ctx->selectable, sock);
+ pn_transport_unbind(pn_connection_transport(conn));
+ pn_connection_reset(conn);
+ pn_transport_t *t = pn_transport();
+ if (messenger->flags & PN_FLAGS_ALLOW_INSECURE_MECHS &&
+ messenger->address.user && messenger->address.pass) {
+ pn_sasl_t *s = pn_sasl(t);
+ pn_sasl_set_allow_insecure_mechs(s, true);
+ }
+ pn_transport_bind(t, conn);
+ pn_decref(t);
+ pn_transport_config(messenger, conn);
+ }
+ }
+}
+
+void pn_messenger_process_session(pn_messenger_t *messenger, pn_event_t *event)
+{
+ pn_session_t *ssn = pn_event_session(event);
+
+ if (pn_session_state(ssn) & PN_LOCAL_UNINIT) {
+ pn_session_open(ssn);
+ }
+
+ if (pn_session_state(ssn) == (PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED)) {
+ pn_session_close(ssn);
+ }
+}
+
+void pn_messenger_process_link(pn_messenger_t *messenger, pn_event_t *event)
+{
+ pn_link_t *link = pn_event_link(event);
+ pn_connection_t *conn = pn_event_connection(event);
+ pn_connection_ctx_t *ctx = (pn_connection_ctx_t *) pn_connection_get_context(conn);
+
+ if (pn_link_state(link) & PN_LOCAL_UNINIT) {
+ pn_terminus_copy(pn_link_source(link), pn_link_remote_source(link));
+ pn_terminus_copy(pn_link_target(link), pn_link_remote_target(link));
+ link_ctx_setup( messenger, conn, link );
+ pn_link_open(link);
+ if (pn_link_is_receiver(link)) {
+ pn_listener_ctx_t *lnr = ctx->listener;
+ ((pn_link_ctx_t *)pn_link_get_context(link))->subscription = lnr ? lnr->subscription : NULL;
+ }
+ }
+
+ if (pn_link_state(link) & PN_REMOTE_ACTIVE) {
+ pn_link_ctx_t *ctx = (pn_link_ctx_t *) pn_link_get_context(link);
+ if (ctx) {
+ const char *addr = pn_terminus_get_address(pn_link_remote_source(link));
+ if (ctx->subscription) {
+ pni_subscription_set_address(ctx->subscription, addr);
+ }
+ }
+ }
+
+ if (pn_link_state(link) & PN_REMOTE_CLOSED) {
+ if (PN_LOCAL_ACTIVE & pn_link_state(link)) {
+ pn_condition_report("LINK", pn_link_remote_condition(link));
+ pn_link_close(link);
+ pni_messenger_reclaim_link(messenger, link);
+ pn_link_free(link);
+ }
+ }
+}
+
+int pni_pump_out(pn_messenger_t *messenger, const char *address, pn_link_t *sender);
+
+void pn_messenger_process_flow(pn_messenger_t *messenger, pn_event_t *event)
+{
+ pn_link_t *link = pn_event_link(event);
+
+ if (pn_link_is_sender(link)) {
+ pni_pump_out(messenger, pn_terminus_get_address(pn_link_target(link)), link);
+ } else {
+ // account for any credit left over after draining links has completed
+ if (pn_link_get_drain(link)) {
+ if (!pn_link_draining(link)) {
+ // drain completed!
+ int drained = pn_link_drained(link);
+ messenger->distributed -= drained;
+ messenger->credit += drained;
+ pn_link_set_drain(link, false);
+ messenger->draining--;
+ pn_list_remove(messenger->credited, link);
+ pn_list_add(messenger->blocked, link);
+ }
+ }
+ }
+}
+
+void pn_messenger_process_delivery(pn_messenger_t *messenger, pn_event_t *event)
+{
+ pn_delivery_t *d = pn_event_delivery(event);
+ pn_link_t *link = pn_event_link(event);
+ if (pn_delivery_updated(d)) {
+ if (pn_link_is_sender(link)) {
+ pn_delivery_update(d, pn_delivery_remote_state(d));
+ }
+ pni_entry_t *e = (pni_entry_t *) pn_delivery_get_context(d);
+ if (e) pni_entry_updated(e);
+ }
+ pn_delivery_clear(d);
+ if (pn_delivery_readable(d)) {
+ int err = pni_pump_in(messenger, pn_terminus_get_address(pn_link_source(link)), link);
+ if (err) {
+ pn_logf("%s", pn_error_text(messenger->error));
+ }
+ }
+}
+
+void pn_messenger_process_transport(pn_messenger_t *messenger, pn_event_t *event)
+{
+ pn_connection_t *conn = pn_event_connection(event);
+ pn_connection_ctx_t *ctx = (pn_connection_ctx_t *) pn_connection_get_context(conn);
+ if (ctx) {
+ pni_conn_modified(ctx);
+ }
+}
+
+int pn_messenger_process_events(pn_messenger_t *messenger)
+{
+ int processed = 0;
+ pn_event_t *event;
+ while ((event = pn_collector_peek(messenger->collector))) {
+ processed++;
+ switch (pn_event_type(event)) {
+ case PN_CONNECTION_INIT:
+ pn_logf("connection created: %p", (void *) pn_event_connection(event));
+ break;
+ case PN_SESSION_INIT:
+ pn_logf("session created: %p", (void *) pn_event_session(event));
+ break;
+ case PN_LINK_INIT:
+ pn_logf("link created: %p", (void *) pn_event_link(event));
+ break;
+ case PN_CONNECTION_REMOTE_OPEN:
+ case PN_CONNECTION_REMOTE_CLOSE:
+ case PN_CONNECTION_LOCAL_OPEN:
+ case PN_CONNECTION_LOCAL_CLOSE:
+ pn_messenger_process_connection(messenger, event);
+ break;
+ case PN_SESSION_REMOTE_OPEN:
+ case PN_SESSION_REMOTE_CLOSE:
+ case PN_SESSION_LOCAL_OPEN:
+ case PN_SESSION_LOCAL_CLOSE:
+ pn_messenger_process_session(messenger, event);
+ break;
+ case PN_LINK_REMOTE_OPEN:
+ case PN_LINK_REMOTE_CLOSE:
+ case PN_LINK_REMOTE_DETACH:
+ case PN_LINK_LOCAL_OPEN:
+ case PN_LINK_LOCAL_CLOSE:
+ case PN_LINK_LOCAL_DETACH:
+ pn_messenger_process_link(messenger, event);
+ break;
+ case PN_LINK_FLOW:
+ pn_messenger_process_flow(messenger, event);
+ break;
+ case PN_DELIVERY:
+ pn_messenger_process_delivery(messenger, event);
+ break;
+ case PN_TRANSPORT:
+ case PN_TRANSPORT_ERROR:
+ case PN_TRANSPORT_HEAD_CLOSED:
+ case PN_TRANSPORT_TAIL_CLOSED:
+ case PN_TRANSPORT_CLOSED:
+ pn_messenger_process_transport(messenger, event);
+ break;
+ case PN_EVENT_NONE:
+ break;
+ case PN_CONNECTION_BOUND:
+ break;
+ case PN_CONNECTION_UNBOUND:
+ break;
+ case PN_CONNECTION_FINAL:
+ break;
+ case PN_SESSION_FINAL:
+ break;
+ case PN_LINK_FINAL:
+ break;
+ default:
+ break;
+ }
+ pn_collector_pop(messenger->collector);
+ }
+
+ return processed;
+}
+
+/**
+ * Function to invoke AMQP related timer events, such as a heartbeat to prevent
+ * remote_idle timeout events
+ */
+static void pni_messenger_tick(pn_messenger_t *messenger)
+{
+ for (size_t i = 0; i < pn_list_size(messenger->connections); i++) {
+ pn_connection_t *connection =
+ (pn_connection_t *)pn_list_get(messenger->connections, i);
+ pn_transport_t *transport = pn_connection_transport(connection);
+ if (transport) {
+ pn_transport_tick(transport, pn_i_now());
+
+ // if there is pending data, such as an empty heartbeat frame, call
+ // process events. This should kick off the chain of selectables for
+ // reading/writing.
+ ssize_t pending = pn_transport_pending(transport);
+ if (pending > 0) {
+ pn_connection_ctx_t *cctx =
+ (pn_connection_ctx_t *)pn_connection_get_context(connection);
+ pn_messenger_process_events(messenger);
+ pn_messenger_flow(messenger);
+ pni_conn_modified(pni_context(cctx->selectable));
+ }
+ }
+ }
+}
+
+int pn_messenger_process(pn_messenger_t *messenger)
+{
+ bool doMessengerTick = true;
+ pn_selectable_t *sel;
+ int events;
+ while ((sel = pn_selector_next(messenger->selector, &events))) {
+ if (events & PN_READABLE) {
+ pn_selectable_readable(sel);
+ }
+ if (events & PN_WRITABLE) {
+ pn_selectable_writable(sel);
+ doMessengerTick = false;
+ }
+ if (events & PN_EXPIRED) {
+ pn_selectable_expired(sel);
+ }
+ if (events & PN_ERROR) {
+ pn_selectable_error(sel);
+ }
+ }
+ // ensure timer events are processed. Cannot call this inside the while loop
+ // as the timer events are not seen by the selector
+ if (doMessengerTick) {
+ pni_messenger_tick(messenger);
+ }
+ if (messenger->interrupted) {
+ messenger->interrupted = false;
+ return PN_INTR;
+ } else {
+ return 0;
+ }
+}
+
+pn_timestamp_t pn_messenger_deadline(pn_messenger_t *messenger)
+{
+ // If the scheduler detects credit imbalance on the links, wake up
+ // in time to service credit drain
+ return messenger->next_drain;
+}
+
+int pni_wait(pn_messenger_t *messenger, int timeout)
+{
+ bool wake = false;
+ pn_selectable_t *sel;
+ while ((sel = pn_messenger_selectable(messenger))) {
+ if (pn_selectable_is_terminal(sel)) {
+ if (pn_selectable_is_registered(sel)) {
+ pn_selector_remove(messenger->selector, sel);
+ }
+ pn_selectable_free(sel);
+ // we can't wait if we end up freeing anything because we could
+ // be waiting on the stopped predicate which might become true
+ // as a result of the free
+ wake = true;
+ } else if (pn_selectable_is_registered(sel)) {
+ pn_selector_update(messenger->selector, sel);
+ } else {
+ pn_selector_add(messenger->selector, sel);
+ pn_selectable_set_registered(sel, true);
+ }
+ }
+
+ if (wake) return 0;
+
+ return pn_selector_select(messenger->selector, timeout);
+}
+
+int pn_messenger_tsync(pn_messenger_t *messenger, bool (*predicate)(pn_messenger_t *), int timeout)
+{
+ if (messenger->passive) {
+ bool pred = predicate(messenger);
+ return pred ? 0 : PN_INPROGRESS;
+ }
+
+ pn_timestamp_t now = pn_i_now();
+ long int deadline = now + timeout;
+ bool pred;
+
+ while (true) {
+ int error = pn_messenger_process(messenger);
+ pred = predicate(messenger);
+ if (error == PN_INTR) {
+ return pred ? 0 : PN_INTR;
+ }
+ int remaining = deadline - now;
+ if (pred || (timeout >= 0 && remaining < 0)) break;
+
+ pn_timestamp_t mdeadline = pn_messenger_deadline(messenger);
+ if (mdeadline) {
+ if (now >= mdeadline)
+ remaining = 0;
+ else {
+ const int delay = mdeadline - now;
+ remaining = (remaining < 0) ? delay : pn_min( remaining, delay );
+ }
+ }
+ error = pni_wait(messenger, remaining);
+ if (error) return error;
+
+ if (timeout >= 0) {
+ now = pn_i_now();
+ }
+ }
+
+ return pred ? 0 : PN_TIMEOUT;
+}
+
+int pn_messenger_sync(pn_messenger_t *messenger, bool (*predicate)(pn_messenger_t *))
+{
+ if (messenger->blocking) {
+ return pn_messenger_tsync(messenger, predicate, messenger->timeout);
+ } else {
+ int err = pn_messenger_tsync(messenger, predicate, 0);
+ if (err == PN_TIMEOUT) {
+ return PN_INPROGRESS;
+ } else {
+ return err;
+ }
+ }
+}
+
+static void pni_parse(pn_address_t *address);
+pn_connection_t *pn_messenger_resolve(pn_messenger_t *messenger,
+ const char *address, char **name);
+int pn_messenger_work(pn_messenger_t *messenger, int timeout);
+
+int pn_messenger_start(pn_messenger_t *messenger)
+{
+ if (!messenger) return PN_ARG_ERR;
+
+ int error = 0;
+
+ // When checking of routes is required we attempt to resolve each route
+ // with a substitution that has a defined scheme, address and port. If
+ // any of theses routes is invalid an appropriate error code will be
+ // returned. Currently no attempt is made to check the name part of the
+ // address, as the intent here is to fail fast if the addressed host
+ // is invalid or unavailable.
+ if (messenger->flags & PN_FLAGS_CHECK_ROUTES) {
+ pn_list_t *substitutions = pn_list(PN_WEAKREF, 0);
+ pn_transform_get_substitutions(messenger->routes, substitutions);
+ for (size_t i = 0; i < pn_list_size(substitutions) && error == 0; i++) {
+ pn_string_t *substitution = (pn_string_t *)pn_list_get(substitutions, i);
+ if (substitution) {
+ pn_address_t addr;
+ addr.text = pn_string(NULL);
+ error = pn_string_copy(addr.text, substitution);
+ if (!error) {
+ pni_parse(&addr);
+ if (addr.scheme && strlen(addr.scheme) > 0 &&
+ !strstr(addr.scheme, "$") && addr.host && strlen(addr.host) > 0 &&
+ !strstr(addr.host, "$") && addr.port && strlen(addr.port) > 0 &&
+ !strstr(addr.port, "$")) {
+ pn_string_t *check_addr = pn_string(NULL);
+ // ipv6 hosts need to be wrapped in [] within a URI
+ if (strstr(addr.host, ":")) {
+ pn_string_format(check_addr, "%s://[%s]:%s/", addr.scheme,
+ addr.host, addr.port);
+ } else {
+ pn_string_format(check_addr, "%s://%s:%s/", addr.scheme,
+ addr.host, addr.port);
+ }
+ char *name = NULL;
+ pn_connection_t *connection = pn_messenger_resolve(
+ messenger, pn_string_get(check_addr), &name);
+ pn_free(check_addr);
+ if (!connection) {
+ if (pn_error_code(messenger->error) == 0)
+ pn_error_copy(messenger->error, pn_io_error(messenger->io));
+ pn_error_format(messenger->error, PN_ERR,
+ "CONNECTION ERROR (%s:%s): %s\n",
+ messenger->address.host, messenger->address.port,
+ pn_error_text(messenger->error));
+ error = pn_error_code(messenger->error);
+ } else {
+ // Send and receive outstanding messages until connection
+ // completes or an error occurs
+ int work = pn_messenger_work(messenger, -1);
+ pn_connection_ctx_t *cctx =
+ (pn_connection_ctx_t *)pn_connection_get_context(connection);
+ while ((work > 0 ||
+ (pn_connection_state(connection) & PN_REMOTE_UNINIT) ||
+ pni_connection_pending(cctx->selectable) != (ssize_t)0) &&
+ pn_error_code(messenger->error) == 0)
+ work = pn_messenger_work(messenger, 0);
+ if (work < 0 && work != PN_TIMEOUT) {
+ error = work;
+ } else {
+ error = pn_error_code(messenger->error);
+ }
+ }
+ }
+ pn_free(addr.text);
+ }
+ }
+ }
+ pn_free(substitutions);
+ }
+
+ return error;
+}
+
+bool pn_messenger_stopped(pn_messenger_t *messenger)
+{
+ return pn_list_size(messenger->connections) == 0 && pn_list_size(messenger->listeners) == 0;
+}
+
+int pn_messenger_stop(pn_messenger_t *messenger)
+{
+ if (!messenger) return PN_ARG_ERR;
+
+ for (size_t i = 0; i < pn_list_size(messenger->connections); i++) {
+ pn_connection_t *conn = (pn_connection_t *) pn_list_get(messenger->connections, i);
+ pn_link_t *link = pn_link_head(conn, PN_LOCAL_ACTIVE);
+ while (link) {
+ pn_link_close(link);
+ link = pn_link_next(link, PN_LOCAL_ACTIVE);
+ }
+ pn_connection_close(conn);
+ }
+
+ for (size_t i = 0; i < pn_list_size(messenger->listeners); i++) {
+ pn_listener_ctx_t *lnr = (pn_listener_ctx_t *) pn_list_get(messenger->listeners, i);
+ pn_selectable_terminate(lnr->selectable);
+ pni_lnr_modified(lnr);
+ }
+
+ return pn_messenger_sync(messenger, pn_messenger_stopped);
+}
+
+void pni_parse_url(char *url, char **scheme, char **user, char **pass, char **host, char **port, char **path);
+
+static void pni_parse(pn_address_t *address)
+{
+ address->passive = false;
+ address->scheme = NULL;
+ address->user = NULL;
+ address->pass = NULL;
+ address->host = NULL;
+ address->port = NULL;
+ address->name = NULL;
+ pni_parse_url(pn_string_buffer(address->text), &address->scheme, &address->user,
+ &address->pass, &address->host, &address->port, &address->name);
+ if (address->host[0] == '~') {
+ address->passive = true;
+ address->host++;
+ }
+}
+
+static int pni_route(pn_messenger_t *messenger, const char *address)
+{
+ pn_address_t *addr = &messenger->address;
+ int err = pn_transform_apply(messenger->routes, address, addr->text);
+ if (err) return pn_error_format(messenger->error, PN_ERR,
+ "transformation error");
+ pni_parse(addr);
+ return 0;
+}
+
+pn_connection_t *pn_messenger_resolve(pn_messenger_t *messenger, const char *address, char **name)
+{
+ assert(messenger);
+ messenger->connection_error = 0;
+ pn_string_t *domain = messenger->domain;
+
+ int err = pni_route(messenger, address);
+ if (err) return NULL;
+
+ bool passive = messenger->address.passive;
+ char *scheme = messenger->address.scheme;
+ char *user = messenger->address.user;
+ char *pass = messenger->address.pass;
+ char *host = messenger->address.host;
+ char *port = messenger->address.port;
+ *name = messenger->address.name;
+
+ if (passive) {
+ for (size_t i = 0; i < pn_list_size(messenger->listeners); i++) {
+ pn_listener_ctx_t *ctx = (pn_listener_ctx_t *) pn_list_get(messenger->listeners, i);
+ if (pn_streq(host, ctx->host) && pn_streq(port, ctx->port)) {
+ return NULL;
+ }
+ }
+
+ pn_listener_ctx(messenger, scheme, host, port);
+ return NULL;
+ }
+
+ pn_string_set(domain, "");
+
+ if (user) {
+ pn_string_addf(domain, "%s@", user);
+ }
+ pn_string_addf(domain, "%s", host);
+ if (port) {
+ pn_string_addf(domain, ":%s", port);
+ }
+
+ for (size_t i = 0; i < pn_list_size(messenger->connections); i++) {
+ pn_connection_t *connection = (pn_connection_t *) pn_list_get(messenger->connections, i);
+ pn_connection_ctx_t *ctx = (pn_connection_ctx_t *) pn_connection_get_context(connection);
+ if (pn_streq(scheme, ctx->scheme) && pn_streq(user, ctx->user) &&
+ pn_streq(pass, ctx->pass) && pn_streq(host, ctx->host) &&
+ pn_streq(port, ctx->port)) {
+ return connection;
+ }
+ const char *container = pn_connection_remote_container(connection);
+ if (pn_streq(container, pn_string_get(domain))) {
+ return connection;
+ }
+ }
+
+ pn_socket_t sock = pn_connect(messenger->io, host, port ? port : default_port(scheme));
+ if (sock == PN_INVALID_SOCKET) {
+ pn_error_copy(messenger->error, pn_io_error(messenger->io));
+ pn_error_format(messenger->error, PN_ERR, "CONNECTION ERROR (%s:%s): %s\n",
+ messenger->address.host, messenger->address.port,
+ pn_error_text(messenger->error));
+ return NULL;
+ }
+
+ pn_connection_t *connection =
+ pn_messenger_connection(messenger, sock, scheme, user, pass, host, port, NULL);
+ pn_transport_t *transport = pn_transport();
+ if (messenger->flags & PN_FLAGS_ALLOW_INSECURE_MECHS && user && pass) {
+ pn_sasl_t *s = pn_sasl(transport);
+ pn_sasl_set_allow_insecure_mechs(s, true);
+ }
+ pn_transport_bind(transport, connection);
+ pn_decref(transport);
+ pn_connection_ctx_t *ctx = (pn_connection_ctx_t *) pn_connection_get_context(connection);
+ pn_selectable_t *sel = ctx->selectable;
+ err = pn_transport_config(messenger, connection);
+ if (err) {
+ pn_selectable_free(sel);
+ messenger->connection_error = err;
+ return NULL;
+ }
+
+ pn_connection_open(connection);
+
+ return connection;
+}
+
+pn_link_t *pn_messenger_get_link(pn_messenger_t *messenger,
+ const char *address, bool sender)
+{
+ char *name = NULL;
+ pn_connection_t *connection = pn_messenger_resolve(messenger, address, &name);
+ if (!connection) return NULL;
+
+ pn_link_t *link = pn_link_head(connection, PN_LOCAL_ACTIVE);
+ while (link) {
+ if (pn_link_is_sender(link) == sender) {
+ const char *terminus = pn_link_is_sender(link) ?
+ pn_terminus_get_address(pn_link_target(link)) :
+ pn_terminus_get_address(pn_link_source(link));
+ if (pn_streq(name, terminus))
+ return link;
+ }
+ link = pn_link_next(link, PN_LOCAL_ACTIVE);
+ }
+ return NULL;
+}
+
+pn_link_t *pn_messenger_link(pn_messenger_t *messenger, const char *address,
+ bool sender, pn_seconds_t timeout)
+{
+ char *name = NULL;
+ pn_connection_t *connection = pn_messenger_resolve(messenger, address, &name);
+ if (!connection)
+ return NULL;
+ pn_connection_ctx_t *cctx =
+ (pn_connection_ctx_t *)pn_connection_get_context(connection);
+
+ pn_link_t *link = pn_messenger_get_link(messenger, address, sender);
+ if (link)
+ return link;
+
+ pn_session_t *ssn = pn_session(connection);
+ pn_session_open(ssn);
+ if (sender) {
+ link = pn_sender(ssn, "sender-xxx");
+ } else {
+ if (name) {
+ link = pn_receiver(ssn, name);
+ } else {
+ link = pn_receiver(ssn, "");
+ }
+ }
+
+ if ((sender && pn_messenger_get_outgoing_window(messenger)) ||
+ (!sender && pn_messenger_get_incoming_window(messenger))) {
+ if (messenger->snd_settle_mode == -1) { /* Choose default based on sender/receiver */
+ /* For a sender use MIXED so the application can decide whether each
+ message is settled or not. For a receiver request UNSETTLED, since the
+ user set an incoming_window which means they want to decide settlement.
+ */
+ pn_link_set_snd_settle_mode(link, sender ? PN_SND_MIXED : PN_SND_UNSETTLED);
+ } else { /* Respect user setting */
+ pn_link_set_snd_settle_mode(link, (pn_snd_settle_mode_t)messenger->snd_settle_mode);
+ }
+ pn_link_set_rcv_settle_mode(link, messenger->rcv_settle_mode);
+ }
+ if (pn_streq(name, "#")) {
+ if (pn_link_is_sender(link)) {
+ pn_terminus_set_dynamic(pn_link_target(link), true);
+ } else {
+ pn_terminus_set_dynamic(pn_link_source(link), true);
+ }
+ } else {
+ pn_terminus_set_address(pn_link_target(link), name);
+ pn_terminus_set_address(pn_link_source(link), name);
+ }
+ link_ctx_setup( messenger, connection, link );
+
+ if (timeout > 0) {
+ pn_terminus_set_expiry_policy(pn_link_target(link), PN_EXPIRE_WITH_LINK);
+ pn_terminus_set_expiry_policy(pn_link_source(link), PN_EXPIRE_WITH_LINK);
+ pn_terminus_set_timeout(pn_link_target(link), timeout);
+ pn_terminus_set_timeout(pn_link_source(link), timeout);
+ }
+
+ if (!sender) {
+ pn_link_ctx_t *ctx = (pn_link_ctx_t *)pn_link_get_context(link);
+ assert( ctx );
+ ctx->subscription = pn_subscription(messenger, cctx->scheme, cctx->host,
+ cctx->port);
+ }
+ pn_link_open(link);
+ return link;
+}
+
+pn_link_t *pn_messenger_source(pn_messenger_t *messenger, const char *source,
+ pn_seconds_t timeout)
+{
+ return pn_messenger_link(messenger, source, false, timeout);
+}
+
+pn_link_t *pn_messenger_target(pn_messenger_t *messenger, const char *target,
+ pn_seconds_t timeout)
+{
+ return pn_messenger_link(messenger, target, true, timeout);
+}
+
+pn_subscription_t *pn_messenger_subscribe(pn_messenger_t *messenger, const char *source)
+{
+ return pn_messenger_subscribe_ttl(messenger, source, 0);
+}
+
+pn_subscription_t *pn_messenger_subscribe_ttl(pn_messenger_t *messenger,
+ const char *source,
+ pn_seconds_t timeout)
+{
+ pni_route(messenger, source);
+ if (pn_error_code(messenger->error)) return NULL;
+
+ bool passive = messenger->address.passive;
+ char *scheme = messenger->address.scheme;
+ char *host = messenger->address.host;
+ char *port = messenger->address.port;
+
+ if (passive) {
+ pn_listener_ctx_t *ctx = pn_listener_ctx(messenger, scheme, host, port);
+ if (ctx) {
+ return ctx->subscription;
+ } else {
+ return NULL;
+ }
+ } else {
+ pn_link_t *src = pn_messenger_source(messenger, source, timeout);
+ if (!src) return NULL;
+ pn_link_ctx_t *ctx = (pn_link_ctx_t *) pn_link_get_context( src );
+ return ctx ? ctx->subscription : NULL;
+ }
+}
+
+int pn_messenger_get_outgoing_window(pn_messenger_t *messenger)
+{
+ return pni_store_get_window(messenger->outgoing);
+}
+
+int pn_messenger_set_outgoing_window(pn_messenger_t *messenger, int window)
+{
+ pni_store_set_window(messenger->outgoing, window);
+ return 0;
+}
+
+int pn_messenger_get_incoming_window(pn_messenger_t *messenger)
+{
+ return pni_store_get_window(messenger->incoming);
+}
+
+int pn_messenger_set_incoming_window(pn_messenger_t *messenger, int window)
+{
+ pni_store_set_window(messenger->incoming, window);
+ return 0;
+}
+
+static void outward_munge(pn_messenger_t *mng, pn_message_t *msg)
+{
+ char stackbuf[256];
+ char *heapbuf = NULL;
+ char *buf = stackbuf;
+ const char *address = pn_message_get_reply_to(msg);
+ int len = address ? strlen(address) : 0;
+ if (len > 1 && address[0] == '~' && address[1] == '/') {
+ unsigned needed = len + strlen(mng->name) + 9;
+ if (needed > sizeof(stackbuf)) {
+ heapbuf = (char *) malloc(needed);
+ buf = heapbuf;
+ }
+ sprintf(buf, "amqp://%s/%s", mng->name, address + 2);
+ pn_message_set_reply_to(msg, buf);
+ } else if (len == 1 && address[0] == '~') {
+ unsigned needed = strlen(mng->name) + 8;
+ if (needed > sizeof(stackbuf)) {
+ heapbuf = (char *) malloc(needed);
+ buf = heapbuf;
+ }
+ sprintf(buf, "amqp://%s", mng->name);
+ pn_message_set_reply_to(msg, buf);
+ }
+ if (heapbuf) free (heapbuf);
+}
+
+int pni_bump_out(pn_messenger_t *messenger, const char *address)
+{
+ pni_entry_t *entry = pni_store_get(messenger->outgoing, address);
+ if (!entry) return 0;
+
+ pni_entry_set_status(entry, PN_STATUS_ABORTED);
+ pni_entry_free(entry);
+ return 0;
+}
+
+int pni_pump_out(pn_messenger_t *messenger, const char *address, pn_link_t *sender)
+{
+ pni_entry_t *entry = pni_store_get(messenger->outgoing, address);
+ if (!entry) {
+ pn_link_drained(sender);
+ return 0;
+ }
+
+ pn_buffer_t *buf = pni_entry_bytes(entry);
+ pn_bytes_t bytes = pn_buffer_bytes(buf);
+ const char *encoded = bytes.start;
+ size_t size = bytes.size;
+
+ // XXX: proper tag
+ char tag[8];
+ void *ptr = &tag;
+ uint64_t next = messenger->next_tag++;
+ *((uint64_t *) ptr) = next;
+ pn_delivery_t *d = pn_delivery(sender, pn_dtag(tag, 8));
+ pni_entry_set_delivery(entry, d);
+ ssize_t n = pn_link_send(sender, encoded, size);
+ if (n < 0) {
+ pni_entry_free(entry);
+ return pn_error_format(messenger->error, n, "send error: %s",
+ pn_error_text(pn_link_error(sender)));
+ } else {
+ pn_link_advance(sender);
+ pni_entry_free(entry);
+ return 0;
+ }
+}
+
+static void pni_default_rewrite(pn_messenger_t *messenger, const char *address,
+ pn_string_t *dst)
+{
+ pn_address_t *addr = &messenger->address;
+ if (address && strstr(address, "@")) {
+ int err = pn_string_set(addr->text, address);
+ if (err) assert(false);
+ pni_parse(addr);
+ if (addr->user || addr->pass)
+ {
+ pn_string_format(messenger->rewritten, "%s%s%s%s%s%s%s",
+ addr->scheme ? addr->scheme : "",
+ addr->scheme ? "://" : "",
+ addr->host,
+ addr->port ? ":" : "",
+ addr->port ? addr->port : "",
+ addr->name ? "/" : "",
+ addr->name ? addr->name : "");
+ }
+ }
+}
+
+static void pni_rewrite(pn_messenger_t *messenger, pn_message_t *msg)
+{
+ const char *address = pn_message_get_address(msg);
+ pn_string_set(messenger->original, address);
+
+ int err = pn_transform_apply(messenger->rewrites, address,
+ messenger->rewritten);
+ if (err) assert(false);
+ if (!pn_transform_matched(messenger->rewrites)) {
+ pni_default_rewrite(messenger, pn_string_get(messenger->rewritten),
+ messenger->rewritten);
+ }
+ pn_message_set_address(msg, pn_string_get(messenger->rewritten));
+}
+
+static void pni_restore(pn_messenger_t *messenger, pn_message_t *msg)
+{
+ pn_message_set_address(msg, pn_string_get(messenger->original));
+}
+
+int pn_messenger_put(pn_messenger_t *messenger, pn_message_t *msg)
+{
+ if (!messenger) return PN_ARG_ERR;
+ if (!msg) return pn_error_set(messenger->error, PN_ARG_ERR, "null message");
+ outward_munge(messenger, msg);
+ const char *address = pn_message_get_address(msg);
+
+ pni_entry_t *entry = pni_store_put(messenger->outgoing, address);
+ if (!entry)
+ return pn_error_format(messenger->error, PN_ERR, "store error");
+
+ messenger->outgoing_tracker = pn_tracker(OUTGOING, pni_entry_track(entry));
+ pn_buffer_t *buf = pni_entry_bytes(entry);
+
+ pni_rewrite(messenger, msg);
+ while (true) {
+ char *encoded = pn_buffer_memory(buf).start;
+ size_t size = pn_buffer_capacity(buf);
+ int err = pn_message_encode(msg, encoded, &size);
+ if (err == PN_OVERFLOW) {
+ err = pn_buffer_ensure(buf, 2*pn_buffer_capacity(buf));
+ if (err) {
+ pni_entry_free(entry);
+ pni_restore(messenger, msg);
+ return pn_error_format(messenger->error, err, "put: error growing buffer");
+ }
+ } else if (err) {
+ pni_restore(messenger, msg);
+ return pn_error_format(messenger->error, err, "encode error: %s",
+ pn_message_error(msg));
+ } else {
+ pni_restore(messenger, msg);
+ pn_buffer_append(buf, encoded, size); // XXX
+ pn_link_t *sender = pn_messenger_target(messenger, address, 0);
+ if (!sender) {
+ int err = pn_error_code(messenger->error);
+ if (err) {
+ return err;
+ } else if (messenger->connection_error) {
+ return pni_bump_out(messenger, address);
+ } else {
+ return 0;
+ }
+ } else {
+ return pni_pump_out(messenger, address, sender);
+ }
+ }
+ }
+
+ return PN_ERR;
+}
+
+pn_tracker_t pn_messenger_outgoing_tracker(pn_messenger_t *messenger)
+{
+ assert(messenger);
+ return messenger->outgoing_tracker;
+}
+
+pni_store_t *pn_tracker_store(pn_messenger_t *messenger, pn_tracker_t tracker)
+{
+ if (pn_tracker_direction(tracker) == OUTGOING) {
+ return messenger->outgoing;
+ } else {
+ return messenger->incoming;
+ }
+}
+
+pn_status_t pn_messenger_status(pn_messenger_t *messenger, pn_tracker_t tracker)
+{
+ pni_store_t *store = pn_tracker_store(messenger, tracker);
+ pni_entry_t *e = pni_store_entry(store, pn_tracker_sequence(tracker));
+ if (e) {
+ return pni_entry_get_status(e);
+ } else {
+ return PN_STATUS_UNKNOWN;
+ }
+}
+
+pn_delivery_t *pn_messenger_delivery(pn_messenger_t *messenger,
+ pn_tracker_t tracker)
+{
+ pni_store_t *store = pn_tracker_store(messenger, tracker);
+ pni_entry_t *e = pni_store_entry(store, pn_tracker_sequence(tracker));
+ if (e) {
+ return pni_entry_get_delivery(e);
+ } else {
+ return NULL;
+ }
+}
+
+bool pn_messenger_buffered(pn_messenger_t *messenger, pn_tracker_t tracker)
+{
+ pni_store_t *store = pn_tracker_store(messenger, tracker);
+ pni_entry_t *e = pni_store_entry(store, pn_tracker_sequence(tracker));
+ if (e) {
+ pn_delivery_t *d = pni_entry_get_delivery(e);
+ if (d) {
+ bool b = pn_delivery_buffered(d);
+ return b;
+ } else {
+ return true;
+ }
+ } else {
+ return false;
+ }
+}
+
+int pn_messenger_settle(pn_messenger_t *messenger, pn_tracker_t tracker, int flags)
+{
+ pni_store_t *store = pn_tracker_store(messenger, tracker);
+ return pni_store_update(store, pn_tracker_sequence(tracker), PN_STATUS_UNKNOWN, flags, true, true);
+}
+
+// true if all pending output has been sent to peer
+bool pn_messenger_sent(pn_messenger_t *messenger)
+{
+ int total = pni_store_size(messenger->outgoing);
+
+ for (size_t i = 0; i < pn_list_size(messenger->connections); i++)
+ {
+ pn_connection_t *conn = (pn_connection_t *) pn_list_get(messenger->connections, i);
+ // check if transport is done generating output
+ pn_transport_t *transport = pn_connection_transport(conn);
+ if (transport) {
+ if (!pn_transport_quiesced(transport)) {
+ return false;
+ }
+ }
+
+ pn_link_t *link = pn_link_head(conn, PN_LOCAL_ACTIVE);
+ while (link) {
+ if (pn_link_is_sender(link)) {
+ total += pn_link_queued(link);
+
+ pn_delivery_t *d = pn_unsettled_head(link);
+ while (d) {
+ if (!pn_delivery_remote_state(d) && !pn_delivery_settled(d)) {
+ total++;
+ }
+ d = pn_unsettled_next(d);
+ }
+ }
+ link = pn_link_next(link, PN_LOCAL_ACTIVE);
+ }
+ }
+
+ return total <= messenger->send_threshold;
+}
+
+bool pn_messenger_rcvd(pn_messenger_t *messenger)
+{
+ if (pni_store_size(messenger->incoming) > 0) return true;
+
+ for (size_t i = 0; i < pn_list_size(messenger->connections); i++)
+ {
+ pn_connection_t *conn = (pn_connection_t *) pn_list_get(messenger->connections, i);
+
+ pn_delivery_t *d = pn_work_head(conn);
+ while (d) {
+ if (pn_delivery_readable(d) && !pn_delivery_partial(d)) {
+ return true;
+ }
+ d = pn_work_next(d);
+ }
+ }
+
+ if (!pn_list_size(messenger->connections) && !pn_list_size(messenger->listeners)) {
+ return true;
+ } else {
+ return false;
+ }
+}
+
+static bool work_pred(pn_messenger_t *messenger) {
+ return messenger->worked;
+}
+
+int pn_messenger_work(pn_messenger_t *messenger, int timeout)
+{
+ messenger->worked = false;
+ int err = pn_messenger_tsync(messenger, work_pred, timeout);
+ if (err) {
+ return err;
+ }
+ return (int) (messenger->worked ? 1 : 0);
+}
+
+int pni_messenger_work(pn_messenger_t *messenger)
+{
+ if (messenger->blocking) {
+ return pn_messenger_work(messenger, messenger->timeout);
+ } else {
+ int err = pn_messenger_work(messenger, 0);
+ if (err == PN_TIMEOUT) {
+ return PN_INPROGRESS;
+ } else {
+ return err;
+ }
+ }
+}
+
+int pn_messenger_interrupt(pn_messenger_t *messenger)
+{
+ assert(messenger);
+ ssize_t n = pn_write(messenger->io, messenger->ctrl[1], "x", 1);
+ if (n <= 0) {
+ return n;
+ } else {
+ return 0;
+ }
+}
+
+int pn_messenger_send(pn_messenger_t *messenger, int n)
+{
+ if (n == -1) {
+ messenger->send_threshold = 0;
+ } else {
+ messenger->send_threshold = pn_messenger_outgoing(messenger) - n;
+ if (messenger->send_threshold < 0)
+ messenger->send_threshold = 0;
+ }
+ return pn_messenger_sync(messenger, pn_messenger_sent);
+}
+
+int pn_messenger_recv(pn_messenger_t *messenger, int n)
+{
+ if (!messenger) return PN_ARG_ERR;
+ if (messenger->blocking && !pn_list_size(messenger->listeners)
+ && !pn_list_size(messenger->connections))
+ return pn_error_format(messenger->error, PN_STATE_ERR, "no valid sources");
+
+ // re-compute credit, and update credit scheduler
+ if (n == -2) {
+ messenger->credit_mode = LINK_CREDIT_MANUAL;
+ } else if (n == -1) {
+ messenger->credit_mode = LINK_CREDIT_AUTO;
+ } else {
+ messenger->credit_mode = LINK_CREDIT_EXPLICIT;
+ if (n > messenger->distributed)
+ messenger->credit = n - messenger->distributed;
+ else // cancel unallocated
+ messenger->credit = 0;
+ }
+ pn_messenger_flow(messenger);
+ int err = pn_messenger_sync(messenger, pn_messenger_rcvd);
+ if (err) return err;
+ if (!pn_messenger_incoming(messenger) &&
+ messenger->blocking &&
+ !pn_list_size(messenger->listeners) &&
+ !pn_list_size(messenger->connections)) {
+ return pn_error_format(messenger->error, PN_STATE_ERR, "no valid sources");
+ } else {
+ return 0;
+ }
+}
+
+int pn_messenger_receiving(pn_messenger_t *messenger)
+{
+ assert(messenger);
+ return messenger->credit + messenger->distributed;
+}
+
+int pn_messenger_get(pn_messenger_t *messenger, pn_message_t *msg)
+{
+ if (!messenger) return PN_ARG_ERR;
+
+ pni_entry_t *entry = pni_store_get(messenger->incoming, NULL);
+ // XXX: need to drain credit before returning EOS
+ if (!entry) return PN_EOS;
+
+ messenger->incoming_tracker = pn_tracker(INCOMING, pni_entry_track(entry));
+ pn_buffer_t *buf = pni_entry_bytes(entry);
+ pn_bytes_t bytes = pn_buffer_bytes(buf);
+ const char *encoded = bytes.start;
+ size_t size = bytes.size;
+
+ messenger->incoming_subscription = (pn_subscription_t *) pni_entry_get_context(entry);
+
+ if (msg) {
+ int err = pn_message_decode(msg, encoded, size);
+ pni_entry_free(entry);
+ if (err) {
+ return pn_error_format(messenger->error, err, "error decoding message: %s",
+ pn_message_error(msg));
+ } else {
+ return 0;
+ }
+ } else {
+ pni_entry_free(entry);
+ return 0;
+ }
+}
+
+pn_tracker_t pn_messenger_incoming_tracker(pn_messenger_t *messenger)
+{
+ assert(messenger);
+ return messenger->incoming_tracker;
+}
+
+pn_subscription_t *pn_messenger_incoming_subscription(pn_messenger_t *messenger)
+{
+ assert(messenger);
+ return messenger->incoming_subscription;
+}
+
+int pn_messenger_accept(pn_messenger_t *messenger, pn_tracker_t tracker, int flags)
+{
+ if (pn_tracker_direction(tracker) != INCOMING) {
+ return pn_error_format(messenger->error, PN_ARG_ERR,
+ "invalid tracker, incoming tracker required");
+ }
+
+ return pni_store_update(messenger->incoming, pn_tracker_sequence(tracker),
+ PN_STATUS_ACCEPTED, flags, false, false);
+}
+
+int pn_messenger_reject(pn_messenger_t *messenger, pn_tracker_t tracker, int flags)
+{
+ if (pn_tracker_direction(tracker) != INCOMING) {
+ return pn_error_format(messenger->error, PN_ARG_ERR,
+ "invalid tracker, incoming tracker required");
+ }
+
+ return pni_store_update(messenger->incoming, pn_tracker_sequence(tracker),
+ PN_STATUS_REJECTED, flags, false, false);
+}
+
+pn_link_t *pn_messenger_tracker_link(pn_messenger_t *messenger,
+ pn_tracker_t tracker)
+{
+ pni_store_t *store = pn_tracker_store(messenger, tracker);
+ pni_entry_t *e = pni_store_entry(store, pn_tracker_sequence(tracker));
+ if (e) {
+ pn_delivery_t *d = pni_entry_get_delivery(e);
+ if (d) {
+ return pn_delivery_link(d);
+ }
+ }
+ return NULL;
+}
+
+int pn_messenger_queued(pn_messenger_t *messenger, bool sender)
+{
+ if (!messenger) return 0;
+
+ int result = 0;
+
+ for (size_t i = 0; i < pn_list_size(messenger->connections); i++) {
+ pn_connection_t *conn = (pn_connection_t *) pn_list_get(messenger->connections, i);
+
+ pn_link_t *link = pn_link_head(conn, PN_LOCAL_ACTIVE);
+ while (link) {
+ if (pn_link_is_sender(link)) {
+ if (sender) {
+ result += pn_link_queued(link);
+ }
+ } else if (!sender) {
+ result += pn_link_queued(link);
+ }
+ link = pn_link_next(link, PN_LOCAL_ACTIVE);
+ }
+ }
+
+ return result;
+}
+
+int pn_messenger_outgoing(pn_messenger_t *messenger)
+{
+ return pni_store_size(messenger->outgoing) + pn_messenger_queued(messenger, true);
+}
+
+int pn_messenger_incoming(pn_messenger_t *messenger)
+{
+ return pni_store_size(messenger->incoming) + pn_messenger_queued(messenger, false);
+}
+
+int pn_messenger_route(pn_messenger_t *messenger, const char *pattern, const char *address)
+{
+ pn_transform_rule(messenger->routes, pattern, address);
+ return 0;
+}
+
+int pn_messenger_rewrite(pn_messenger_t *messenger, const char *pattern, const char *address)
+{
+ pn_transform_rule(messenger->rewrites, pattern, address);
+ return 0;
+}
+
+int pn_messenger_set_flags(pn_messenger_t *messenger, const int flags)
+{
+ if (!messenger)
+ return PN_ARG_ERR;
+ if (flags == 0) {
+ messenger->flags = 0;
+ } else if (flags & (PN_FLAGS_CHECK_ROUTES | PN_FLAGS_ALLOW_INSECURE_MECHS)) {
+ messenger->flags |= flags;
+ } else {
+ return PN_ARG_ERR;
+ }
+ return 0;
+}
+
+int pn_messenger_get_flags(pn_messenger_t *messenger)
+{
+ return messenger ? messenger->flags : 0;
+}
+
+int pn_messenger_set_snd_settle_mode(pn_messenger_t *messenger,
+ const pn_snd_settle_mode_t mode)
+{
+ if (!messenger)
+ return PN_ARG_ERR;
+ messenger->snd_settle_mode = mode;
+ return 0;
+}
+
+int pn_messenger_set_rcv_settle_mode(pn_messenger_t *messenger,
+ const pn_rcv_settle_mode_t mode)
+{
+ if (!messenger)
+ return PN_ARG_ERR;
+ messenger->rcv_settle_mode = mode;
+ return 0;
+}
+
+void pn_messenger_set_tracer(pn_messenger_t *messenger, pn_tracer_t tracer)
+{
+ assert(messenger);
+ assert(tracer);
+
+ messenger->tracer = tracer;
+}
+
+pn_millis_t pn_messenger_get_remote_idle_timeout(pn_messenger_t *messenger,
+ const char *address)
+{
+ if (!messenger)
+ return PN_ARG_ERR;
+
+ pn_address_t addr;
+ addr.text = pn_string(address);
+ pni_parse(&addr);
+
+ pn_millis_t timeout = -1;
+ for (size_t i = 0; i < pn_list_size(messenger->connections); i++) {
+ pn_connection_t *connection =
+ (pn_connection_t *)pn_list_get(messenger->connections, i);
+ pn_connection_ctx_t *ctx =
+ (pn_connection_ctx_t *)pn_connection_get_context(connection);
+ if (pn_streq(addr.scheme, ctx->scheme) && pn_streq(addr.host, ctx->host) &&
+ pn_streq(addr.port, ctx->port)) {
+ pn_transport_t *transport = pn_connection_transport(connection);
+ if (transport)
+ timeout = pn_transport_get_remote_idle_timeout(transport);
+ break;
+ }
+ }
+ return timeout;
+}
+
+int
+pn_messenger_set_ssl_peer_authentication_mode(pn_messenger_t *messenger,
+ const pn_ssl_verify_mode_t mode)
+{
+ if (!messenger)
+ return PN_ARG_ERR;
+ messenger->ssl_peer_authentication_mode = mode;
+ return 0;
+}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/37136940/c/src/messenger/messenger.h
----------------------------------------------------------------------
diff --git a/c/src/messenger/messenger.h b/c/src/messenger/messenger.h
new file mode 100644
index 0000000..01b1838
--- /dev/null
+++ b/c/src/messenger/messenger.h
@@ -0,0 +1,30 @@
+#ifndef _PROTON_MESSENGER_H
+#define _PROTON_MESSENGER_H 1
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include <proton/messenger.h>
+
+int pni_messenger_add_subscription(pn_messenger_t *messenger, pn_subscription_t *subscription);
+int pni_messenger_work(pn_messenger_t *messenger);
+
+#endif /* messenger.h */
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org