proton-c/src/reactor/connection.c - qpid-proton - Git at Google

 /*
  *
  * 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/connection.h>
 #include <proton/object.h>
 #include <proton/sasl.h>
 #include <proton/ssl.h>
 #include <proton/transport.h>
 #include <proton/url.h>
 #include <assert.h>
 #include <stdio.h>
 #include <string.h>
 #include "selectable.h"
 #include "reactor.h"

 // XXX: overloaded for both directions
 PN_HANDLE(PN_TRANCTX)
 PN_HANDLE(PNI_CONN_PEER_ADDRESS)

 void pni_reactor_set_connection_peer_address(pn_connection_t *connection,
                                              const char *host,
                                              const char *port)
 {
     pn_url_t *url = pn_url();
     pn_url_set_host(url, host);
     pn_url_set_port(url, port);
     pn_record_t *record = pn_connection_attachments(connection);
     if (!pn_record_has(record, PNI_CONN_PEER_ADDRESS)) {
       pn_record_def(record, PNI_CONN_PEER_ADDRESS, PN_OBJECT);
     }
     pn_record_set(record, PNI_CONN_PEER_ADDRESS, url);
     pn_decref(url);
 }

 static pn_transport_t *pni_transport(pn_selectable_t *sel) {
   pn_record_t *record = pn_selectable_attachments(sel);
   return (pn_transport_t *) pn_record_get(record, PN_TRANCTX);
 }

 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;
 }

 static ssize_t pni_connection_pending(pn_selectable_t *sel)
 {
   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_reactor_t *reactor = (pn_reactor_t *) pni_selectable_get_context(sel);
   pn_transport_t *transport = pni_transport(sel);
   pn_timestamp_t deadline = pn_transport_tick(transport, pn_reactor_now(reactor));
   return deadline;
 }

 static void pni_connection_update(pn_selectable_t *sel) {
   ssize_t c = pni_connection_capacity(sel);
   ssize_t p = pni_connection_pending(sel);
   pn_selectable_set_reading(sel, c > 0);
   pn_selectable_set_writing(sel, p > 0);
   pn_selectable_set_deadline(sel, pni_connection_deadline(sel));
 }

 void pni_handle_transport(pn_reactor_t *reactor, pn_event_t *event) {
   assert(reactor);
   pn_transport_t *transport = pn_event_transport(event);
   pn_record_t *record = pn_transport_attachments(transport);
   pn_selectable_t *sel = (pn_selectable_t *) pn_record_get(record, PN_TRANCTX);
   if (sel && !pn_selectable_is_terminal(sel)) {
     pni_connection_update(sel);
     pn_reactor_update(reactor, sel);
   }
 }

 pn_selectable_t *pn_reactor_selectable_transport(pn_reactor_t *reactor, pn_socket_t sock, pn_transport_t *transport);

 void pni_handle_open(pn_reactor_t *reactor, pn_event_t *event) {
   assert(reactor);
   assert(event);

   pn_connection_t *conn = pn_event_connection(event);
   if (!(pn_connection_state(conn) & PN_REMOTE_UNINIT)) {
     return;
   }

   pn_transport_t *transport = pn_transport();
   pn_transport_bind(transport, conn);
   pn_decref(transport);
 }

 void pni_handle_bound(pn_reactor_t *reactor, pn_event_t *event) {
   assert(reactor);
   assert(event);

   pn_connection_t *conn = pn_event_connection(event);
   pn_transport_t *transport = pn_event_transport(event);
   pn_record_t *record = pn_connection_attachments(conn);
   pn_url_t *url = (pn_url_t *)pn_record_get(record, PNI_CONN_PEER_ADDRESS);
   const char *host = NULL;
   const char *port = "5672";
   pn_string_t *str = NULL;

   // link the new transport to its reactor:
   pni_record_init_reactor(pn_transport_attachments(transport), reactor);

   if (pn_connection_acceptor(conn) != NULL) {
       // this connection was created by the acceptor.  There is already a
       // socket assigned to this connection.  Nothing needs to be done.
       return;
   }

   if (url) {
       host = pn_url_get_host(url);
       const char *uport = pn_url_get_port(url);
       if (uport) {
           port = uport;
       } else {
           const char *scheme = pn_url_get_scheme(url);
           if (scheme && strcmp(scheme, "amqps") == 0) {
               port = "5671";
           }
       }
       if (!pn_connection_get_user(conn)) {
           // user did not manually set auth info
           const char *user = pn_url_get_username(url);
           if (user) pn_connection_set_user(conn, user);
           const char *passwd = pn_url_get_password(url);
           if (passwd) pn_connection_set_password(conn, passwd);
       }
   } else {
       // for backward compatibility, see if the connection's hostname can be
       // used for the remote address.  See JIRA PROTON-1133
       const char *hostname = pn_connection_get_hostname(conn);
       if (hostname) {
           str = pn_string(hostname);
           char *h = pn_string_buffer(str);
           // see if a port has been included in the hostname.  This is not
           // allowed by the spec, but the old reactor interface allowed it.
           char *colon = strrchr(h, ':');
           if (colon) {
               *colon = '\0';
               port = colon + 1;
           }
           host = h;
       }
   }

   if (!host) {
       // error: no address configured
       pn_condition_t *cond = pn_transport_condition(transport);
       pn_condition_set_name(cond, "proton:io");
       pn_condition_set_description(cond, "Connection failed: no address configured");
       pn_transport_close_tail(transport);
       pn_transport_close_head(transport);
   } else {
       pn_socket_t sock = pn_connect(pn_reactor_io(reactor), host, port);
       // invalid sockets are ignored by poll, so we need to do this manualy
       if (sock == PN_INVALID_SOCKET) {
           pn_condition_t *cond = pn_transport_condition(transport);
           pn_condition_set_name(cond, "proton:io");
           pn_condition_set_description(cond, pn_error_text(pn_io_error(pn_reactor_io(reactor))));
           pn_transport_close_tail(transport);
           pn_transport_close_head(transport);
       } else {
           pn_reactor_selectable_transport(reactor, sock, transport);
       }
   }
   pn_free(str);
 }

 void pni_handle_final(pn_reactor_t *reactor, pn_event_t *event) {
   assert(reactor);
   assert(event);
   pn_connection_t *conn = pn_event_connection(event);
   pn_list_remove(pn_reactor_children(reactor), conn);
 }

 static void pni_connection_readable(pn_selectable_t *sel)
 {
   pn_reactor_t *reactor = (pn_reactor_t *) pni_selectable_get_context(sel);
   pn_transport_t *transport = pni_transport(sel);
   ssize_t capacity = pn_transport_capacity(transport);
   if (capacity > 0) {
     ssize_t n = pn_recv(pn_reactor_io(reactor), pn_selectable_get_fd(sel),
                         pn_transport_tail(transport), capacity);
     if (n <= 0) {
       if (n == 0 || !pn_wouldblock(pn_reactor_io(reactor))) {
         if (n < 0) {
           pn_condition_t *cond = pn_transport_condition(transport);
           pn_condition_set_name(cond, "proton:io");
           pn_condition_set_description(cond, pn_error_text(pn_io_error(pn_reactor_io(reactor))));
         }
         pn_transport_close_tail(transport);
       }
     } else {
       pn_transport_process(transport, (size_t)n);
     }
   }

   ssize_t newcap = pn_transport_capacity(transport);
   //occasionally transport events aren't generated when expected, so
   //the following hack ensures we always update the selector
   if (1 || newcap != capacity) {
     pni_connection_update(sel);
     pn_reactor_update(reactor, sel);
   }
 }

 static void pni_connection_writable(pn_selectable_t *sel)
 {
   pn_reactor_t *reactor = (pn_reactor_t *) pni_selectable_get_context(sel);
   pn_transport_t *transport = pni_transport(sel);
   ssize_t pending = pn_transport_pending(transport);
   if (pending > 0) {
     ssize_t n = pn_send(pn_reactor_io(reactor), pn_selectable_get_fd(sel),
                         pn_transport_head(transport), pending);
     if (n < 0) {
       if (!pn_wouldblock(pn_reactor_io(reactor))) {
         pn_condition_t *cond = pn_transport_condition(transport);
         if (!pn_condition_is_set(cond)) {
           pn_condition_set_name(cond, "proton:io");
           pn_condition_set_description(cond, pn_error_text(pn_io_error(pn_reactor_io(reactor))));
         }
         pn_transport_close_head(transport);
       }
     } else {
       pn_transport_pop(transport, n);
     }
   }

   ssize_t newpending = pn_transport_pending(transport);
   if (newpending != pending) {
     pni_connection_update(sel);
     pn_reactor_update(reactor, sel);
   }
 }

 static void pni_connection_error(pn_selectable_t *sel) {
   pn_reactor_t *reactor = (pn_reactor_t *) pni_selectable_get_context(sel);
   pn_transport_t *transport = pni_transport(sel);
   pn_transport_close_head(transport);
   pn_transport_close_tail(transport);
   pn_selectable_terminate(sel);
   pn_reactor_update(reactor, sel);
 }

 static void pni_connection_expired(pn_selectable_t *sel) {
   pn_reactor_t *reactor = (pn_reactor_t *) pni_selectable_get_context(sel);
   pn_transport_t *transport = pni_transport(sel);
   pn_timestamp_t deadline = pn_transport_tick(transport, pn_reactor_now(reactor));
   pn_selectable_set_deadline(sel, deadline);
   ssize_t c = pni_connection_capacity(sel);
   ssize_t p = pni_connection_pending(sel);
   pn_selectable_set_reading(sel, c > 0);
   pn_selectable_set_writing(sel, p > 0);
   pn_reactor_update(reactor, sel);
 }

 static void pni_connection_finalize(pn_selectable_t *sel) {
   pn_reactor_t *reactor = (pn_reactor_t *) pni_selectable_get_context(sel);
   pn_transport_t *transport = pni_transport(sel);
   pn_record_t *record = pn_transport_attachments(transport);
   pn_record_set(record, PN_TRANCTX, NULL);
   pn_socket_t fd = pn_selectable_get_fd(sel);
   pn_close(pn_reactor_io(reactor), fd);
 }

 pn_selectable_t *pn_reactor_selectable_transport(pn_reactor_t *reactor, pn_socket_t sock, pn_transport_t *transport) {
   pn_selectable_t *sel = pn_reactor_selectable(reactor);
   pn_selectable_set_fd(sel, sock);
   pn_selectable_on_readable(sel, pni_connection_readable);
   pn_selectable_on_writable(sel, pni_connection_writable);
   pn_selectable_on_error(sel, pni_connection_error);
   pn_selectable_on_expired(sel, pni_connection_expired);
   pn_selectable_on_finalize(sel, pni_connection_finalize);
   pn_record_t *record = pn_selectable_attachments(sel);
   pn_record_def(record, PN_TRANCTX, PN_OBJECT);
   pn_record_set(record, PN_TRANCTX, transport);
   pn_record_t *tr = pn_transport_attachments(transport);
   pn_record_def(tr, PN_TRANCTX, PN_WEAKREF);
   pn_record_set(tr, PN_TRANCTX, sel);
   pni_connection_update(sel);
   pn_reactor_update(reactor, sel);
   return sel;
 }

 pn_connection_t *pn_reactor_connection(pn_reactor_t *reactor, pn_handler_t *handler) {
   assert(reactor);
   pn_connection_t *connection = pn_connection();
   pn_record_t *record = pn_connection_attachments(connection);
   pn_record_set_handler(record, handler);
   pn_connection_collect(connection, pn_reactor_collector(reactor));
   pn_list_add(pn_reactor_children(reactor), connection);
   pni_record_init_reactor(record, reactor);
   pn_decref(connection);
   return connection;
 }

 pn_connection_t *pn_reactor_connection_to_host(pn_reactor_t *reactor,
                                                const char *host,
                                                const char *port,
                                                pn_handler_t *handler) {
     pn_connection_t *connection = pn_reactor_connection(reactor, handler);
     pn_reactor_set_connection_host(reactor, connection, host, port);
     return connection;
 }


 void pn_reactor_set_connection_host(pn_reactor_t *reactor,
                                     pn_connection_t *connection,
                                     const char *host,
                                     const char *port)
 {
     (void)reactor;  // ignored
     if (pn_connection_acceptor(connection) != NULL) {
         // this is an inbound connection created by the acceptor. The peer
         // address cannot be modified.
         return;
     }
     pni_reactor_set_connection_peer_address(connection, host, port);
 }


 const char *pn_reactor_get_connection_address(pn_reactor_t *reactor,
                                               pn_connection_t *connection)
 {
     (void)reactor;  // ignored
     if (!connection) return NULL;
     pn_record_t *record = pn_connection_attachments(connection);
     pn_url_t *url = (pn_url_t *)pn_record_get(record, PNI_CONN_PEER_ADDRESS);
     if (url) {
         return pn_url_str(url);
     }
     return NULL;
 }
	/*
	*
	* 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/connection.h>
	#include <proton/object.h>
	#include <proton/sasl.h>
	#include <proton/ssl.h>
	#include <proton/transport.h>
	#include <proton/url.h>
	#include <assert.h>
	#include <stdio.h>
	#include <string.h>
	#include "selectable.h"
	#include "reactor.h"

	// XXX: overloaded for both directions
	PN_HANDLE(PN_TRANCTX)
	PN_HANDLE(PNI_CONN_PEER_ADDRESS)

	void pni_reactor_set_connection_peer_address(pn_connection_t *connection,
	const char *host,
	const char *port)
	{
	pn_url_t *url = pn_url();
	pn_url_set_host(url, host);
	pn_url_set_port(url, port);
	pn_record_t *record = pn_connection_attachments(connection);
	if (!pn_record_has(record, PNI_CONN_PEER_ADDRESS)) {
	pn_record_def(record, PNI_CONN_PEER_ADDRESS, PN_OBJECT);
	}
	pn_record_set(record, PNI_CONN_PEER_ADDRESS, url);
	pn_decref(url);
	}

	static pn_transport_t pni_transport(pn_selectable_t sel) {
	pn_record_t *record = pn_selectable_attachments(sel);
	return (pn_transport_t *) pn_record_get(record, PN_TRANCTX);
	}

	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;
	}

	static ssize_t pni_connection_pending(pn_selectable_t *sel)
	{
	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_reactor_t reactor = (pn_reactor_t ) pni_selectable_get_context(sel);
	pn_transport_t *transport = pni_transport(sel);
	pn_timestamp_t deadline = pn_transport_tick(transport, pn_reactor_now(reactor));
	return deadline;
	}

	static void pni_connection_update(pn_selectable_t *sel) {
	ssize_t c = pni_connection_capacity(sel);
	ssize_t p = pni_connection_pending(sel);
	pn_selectable_set_reading(sel, c > 0);
	pn_selectable_set_writing(sel, p > 0);
	pn_selectable_set_deadline(sel, pni_connection_deadline(sel));
	}

	void pni_handle_transport(pn_reactor_t reactor, pn_event_t event) {
	assert(reactor);
	pn_transport_t *transport = pn_event_transport(event);
	pn_record_t *record = pn_transport_attachments(transport);
	pn_selectable_t sel = (pn_selectable_t ) pn_record_get(record, PN_TRANCTX);
	if (sel && !pn_selectable_is_terminal(sel)) {
	pni_connection_update(sel);
	pn_reactor_update(reactor, sel);
	}
	}

	pn_selectable_t pn_reactor_selectable_transport(pn_reactor_t reactor, pn_socket_t sock, pn_transport_t *transport);

	void pni_handle_open(pn_reactor_t reactor, pn_event_t event) {
	assert(reactor);
	assert(event);

	pn_connection_t *conn = pn_event_connection(event);
	if (!(pn_connection_state(conn) & PN_REMOTE_UNINIT)) {
	return;
	}

	pn_transport_t *transport = pn_transport();
	pn_transport_bind(transport, conn);
	pn_decref(transport);
	}

	void pni_handle_bound(pn_reactor_t reactor, pn_event_t event) {
	assert(reactor);
	assert(event);

	pn_connection_t *conn = pn_event_connection(event);
	pn_transport_t *transport = pn_event_transport(event);
	pn_record_t *record = pn_connection_attachments(conn);
	pn_url_t url = (pn_url_t )pn_record_get(record, PNI_CONN_PEER_ADDRESS);
	const char *host = NULL;
	const char *port = "5672";
	pn_string_t *str = NULL;

	// link the new transport to its reactor:
	pni_record_init_reactor(pn_transport_attachments(transport), reactor);

	if (pn_connection_acceptor(conn) != NULL) {
	// this connection was created by the acceptor. There is already a
	// socket assigned to this connection. Nothing needs to be done.
	return;
	}

	if (url) {
	host = pn_url_get_host(url);
	const char *uport = pn_url_get_port(url);
	if (uport) {
	port = uport;
	} else {
	const char *scheme = pn_url_get_scheme(url);
	if (scheme && strcmp(scheme, "amqps") == 0) {
	port = "5671";
	}
	}
	if (!pn_connection_get_user(conn)) {
	// user did not manually set auth info
	const char *user = pn_url_get_username(url);
	if (user) pn_connection_set_user(conn, user);
	const char *passwd = pn_url_get_password(url);
	if (passwd) pn_connection_set_password(conn, passwd);
	}
	} else {
	// for backward compatibility, see if the connection's hostname can be
	// used for the remote address. See JIRA PROTON-1133
	const char *hostname = pn_connection_get_hostname(conn);
	if (hostname) {
	str = pn_string(hostname);
	char *h = pn_string_buffer(str);
	// see if a port has been included in the hostname. This is not
	// allowed by the spec, but the old reactor interface allowed it.
	char *colon = strrchr(h, ':');
	if (colon) {
	*colon = '\0';
	port = colon + 1;
	}
	host = h;
	}
	}

	if (!host) {
	// error: no address configured
	pn_condition_t *cond = pn_transport_condition(transport);
	pn_condition_set_name(cond, "proton:io");
	pn_condition_set_description(cond, "Connection failed: no address configured");
	pn_transport_close_tail(transport);
	pn_transport_close_head(transport);
	} else {
	pn_socket_t sock = pn_connect(pn_reactor_io(reactor), host, port);
	// invalid sockets are ignored by poll, so we need to do this manualy
	if (sock == PN_INVALID_SOCKET) {
	pn_condition_t *cond = pn_transport_condition(transport);
	pn_condition_set_name(cond, "proton:io");
	pn_condition_set_description(cond, pn_error_text(pn_io_error(pn_reactor_io(reactor))));
	pn_transport_close_tail(transport);
	pn_transport_close_head(transport);
	} else {
	pn_reactor_selectable_transport(reactor, sock, transport);
	}
	}
	pn_free(str);
	}

	void pni_handle_final(pn_reactor_t reactor, pn_event_t event) {
	assert(reactor);
	assert(event);
	pn_connection_t *conn = pn_event_connection(event);
	pn_list_remove(pn_reactor_children(reactor), conn);
	}

	static void pni_connection_readable(pn_selectable_t *sel)
	{
	pn_reactor_t reactor = (pn_reactor_t ) pni_selectable_get_context(sel);
	pn_transport_t *transport = pni_transport(sel);
	ssize_t capacity = pn_transport_capacity(transport);
	if (capacity > 0) {
	ssize_t n = pn_recv(pn_reactor_io(reactor), pn_selectable_get_fd(sel),
	pn_transport_tail(transport), capacity);
	if (n <= 0) {
	if (n == 0 \|\| !pn_wouldblock(pn_reactor_io(reactor))) {
	if (n < 0) {
	pn_condition_t *cond = pn_transport_condition(transport);
	pn_condition_set_name(cond, "proton:io");
	pn_condition_set_description(cond, pn_error_text(pn_io_error(pn_reactor_io(reactor))));
	}
	pn_transport_close_tail(transport);
	}
	} else {
	pn_transport_process(transport, (size_t)n);
	}
	}

	ssize_t newcap = pn_transport_capacity(transport);
	//occasionally transport events aren't generated when expected, so
	//the following hack ensures we always update the selector
	if (1 \|\| newcap != capacity) {
	pni_connection_update(sel);
	pn_reactor_update(reactor, sel);
	}
	}

	static void pni_connection_writable(pn_selectable_t *sel)
	{
	pn_reactor_t reactor = (pn_reactor_t ) pni_selectable_get_context(sel);
	pn_transport_t *transport = pni_transport(sel);
	ssize_t pending = pn_transport_pending(transport);
	if (pending > 0) {
	ssize_t n = pn_send(pn_reactor_io(reactor), pn_selectable_get_fd(sel),
	pn_transport_head(transport), pending);
	if (n < 0) {
	if (!pn_wouldblock(pn_reactor_io(reactor))) {
	pn_condition_t *cond = pn_transport_condition(transport);
	if (!pn_condition_is_set(cond)) {
	pn_condition_set_name(cond, "proton:io");
	pn_condition_set_description(cond, pn_error_text(pn_io_error(pn_reactor_io(reactor))));
	}
	pn_transport_close_head(transport);
	}
	} else {
	pn_transport_pop(transport, n);
	}
	}

	ssize_t newpending = pn_transport_pending(transport);
	if (newpending != pending) {
	pni_connection_update(sel);
	pn_reactor_update(reactor, sel);
	}
	}

	static void pni_connection_error(pn_selectable_t *sel) {
	pn_reactor_t reactor = (pn_reactor_t ) pni_selectable_get_context(sel);
	pn_transport_t *transport = pni_transport(sel);
	pn_transport_close_head(transport);
	pn_transport_close_tail(transport);
	pn_selectable_terminate(sel);
	pn_reactor_update(reactor, sel);
	}

	static void pni_connection_expired(pn_selectable_t *sel) {
	pn_reactor_t reactor = (pn_reactor_t ) pni_selectable_get_context(sel);
	pn_transport_t *transport = pni_transport(sel);
	pn_timestamp_t deadline = pn_transport_tick(transport, pn_reactor_now(reactor));
	pn_selectable_set_deadline(sel, deadline);
	ssize_t c = pni_connection_capacity(sel);
	ssize_t p = pni_connection_pending(sel);
	pn_selectable_set_reading(sel, c > 0);
	pn_selectable_set_writing(sel, p > 0);
	pn_reactor_update(reactor, sel);
	}

	static void pni_connection_finalize(pn_selectable_t *sel) {
	pn_reactor_t reactor = (pn_reactor_t ) pni_selectable_get_context(sel);
	pn_transport_t *transport = pni_transport(sel);
	pn_record_t *record = pn_transport_attachments(transport);
	pn_record_set(record, PN_TRANCTX, NULL);
	pn_socket_t fd = pn_selectable_get_fd(sel);
	pn_close(pn_reactor_io(reactor), fd);
	}

	pn_selectable_t pn_reactor_selectable_transport(pn_reactor_t reactor, pn_socket_t sock, pn_transport_t *transport) {
	pn_selectable_t *sel = pn_reactor_selectable(reactor);
	pn_selectable_set_fd(sel, sock);
	pn_selectable_on_readable(sel, pni_connection_readable);
	pn_selectable_on_writable(sel, pni_connection_writable);
	pn_selectable_on_error(sel, pni_connection_error);
	pn_selectable_on_expired(sel, pni_connection_expired);
	pn_selectable_on_finalize(sel, pni_connection_finalize);
	pn_record_t *record = pn_selectable_attachments(sel);
	pn_record_def(record, PN_TRANCTX, PN_OBJECT);
	pn_record_set(record, PN_TRANCTX, transport);
	pn_record_t *tr = pn_transport_attachments(transport);
	pn_record_def(tr, PN_TRANCTX, PN_WEAKREF);
	pn_record_set(tr, PN_TRANCTX, sel);
	pni_connection_update(sel);
	pn_reactor_update(reactor, sel);
	return sel;
	}

	pn_connection_t pn_reactor_connection(pn_reactor_t reactor, pn_handler_t *handler) {
	assert(reactor);
	pn_connection_t *connection = pn_connection();
	pn_record_t *record = pn_connection_attachments(connection);
	pn_record_set_handler(record, handler);
	pn_connection_collect(connection, pn_reactor_collector(reactor));
	pn_list_add(pn_reactor_children(reactor), connection);
	pni_record_init_reactor(record, reactor);
	pn_decref(connection);
	return connection;
	}

	pn_connection_t pn_reactor_connection_to_host(pn_reactor_t reactor,
	const char *host,
	const char *port,
	pn_handler_t *handler) {
	pn_connection_t *connection = pn_reactor_connection(reactor, handler);
	pn_reactor_set_connection_host(reactor, connection, host, port);
	return connection;
	}


	void pn_reactor_set_connection_host(pn_reactor_t *reactor,
	pn_connection_t *connection,
	const char *host,
	const char *port)
	{
	(void)reactor; // ignored
	if (pn_connection_acceptor(connection) != NULL) {
	// this is an inbound connection created by the acceptor. The peer
	// address cannot be modified.
	return;
	}
	pni_reactor_set_connection_peer_address(connection, host, port);
	}


	const char pn_reactor_get_connection_address(pn_reactor_t reactor,
	pn_connection_t *connection)
	{
	(void)reactor; // ignored
	if (!connection) return NULL;
	pn_record_t *record = pn_connection_attachments(connection);
	pn_url_t url = (pn_url_t )pn_record_get(record, PNI_CONN_PEER_ADDRESS);
	if (url) {
	return pn_url_str(url);
	}
	return NULL;
	}