devs/libstrophe-gh
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sock: Introduce xmpp_sock_t abstraction

Browse Source 
libstrophe uses non-blocking sockets and the connect() syscall may
return before a TCP connection is established. This doesn't allow
to catch all possible errors synchronously and some of the errors
are handled in the event handler.

In a scenario with multiple SRV records and/or multiple IP addresses
resolution, we need to repeat connection attempt on a failure.

xmpp_sock_t resolves the above problem. It keeps resolved records and
addresses to repeat connect attempt asynchronously.
This commit is contained in:
Dmitry Podgorny
2023-06-05 16:57:07 +03:00
parent 37555868f6
commit f3460460e9
7 changed files with 269 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/common.h
View File

@@ -198,6 +198,7 @@ struct _xmpp_conn_t {
int error;
xmpp_stream_error_t *stream_error;
xmpp_sock_t *xsock;
sock_t sock;
int ka_timeout; /* TCP keepalive timeout */
int ka_interval; /* TCP keepalive interval */

80
src/conn.c
View File

@@ -29,7 +29,6 @@
#include "common.h"
#include "util.h"
#include "parser.h"
#include "resolver.h"
#ifndef DEFAULT_SEND_QUEUE_MAX
/** @def DEFAULT_SEND_QUEUE_MAX
@@ -103,8 +102,6 @@ static unsigned short _conn_default_port(xmpp_conn_t *conn,
static void _conn_reset(xmpp_conn_t *conn);
static int _conn_connect(xmpp_conn_t *conn,
const char *domain,
const char *host,
unsigned short port,
xmpp_conn_type_t type,
xmpp_conn_handler callback,
void *userdata);
@@ -148,7 +145,8 @@ xmpp_conn_t *xmpp_conn_new(xmpp_ctx_t *ctx)
conn->type = XMPP_UNKNOWN;
conn->state = XMPP_STATE_DISCONNECTED;
conn->sock = -1;
conn->xsock = NULL;
conn->sock = INVALID_SOCKET;
conn->ka_timeout = KEEPALIVE_TIMEOUT;
conn->ka_interval = KEEPALIVE_INTERVAL;
conn->ka_count = KEEPALIVE_COUNT;
@@ -402,6 +400,7 @@ int xmpp_conn_release(xmpp_conn_t *conn)
if (conn->sm_state)
xmpp_free_sm_state(conn->sm_state);
tls_clear_password_cache(conn);
sock_free(conn->xsock);
strophe_free(ctx, conn);
released = 1;
}
@@ -705,12 +704,7 @@ int xmpp_connect_client(xmpp_conn_t *conn,
xmpp_conn_handler callback,
void *userdata)
{
resolver_srv_rr_t *srv_rr_list = NULL;
resolver_srv_rr_t *rr;
char *domain;
const char *host = NULL;
unsigned short port = 0;
int found = XMPP_DOMAIN_NOT_FOUND;
int rc;
if (!conn->jid && (conn->tls_client_cert || conn->tls_client_key)) {
@@ -739,48 +733,35 @@ int xmpp_connect_client(xmpp_conn_t *conn,
if (!conn->sm_state) {
conn->sm_state = strophe_alloc(conn->ctx, sizeof(*conn->sm_state));
if (!conn->sm_state)
return XMPP_EMEM;
goto err_mem;
memset(conn->sm_state, 0, sizeof(*conn->sm_state));
conn->sm_state->ctx = conn->ctx;
}
if (altdomain != NULL) {
strophe_debug(conn->ctx, "xmpp", "Connecting via altdomain.");
host = altdomain;
port = altport ? altport : _conn_default_port(conn, XMPP_CLIENT);
found = XMPP_DOMAIN_ALTDOMAIN;
if (altdomain != NULL)
strophe_debug(conn->ctx, "conn", "Connecting via altdomain.");
if (conn->tls_legacy_ssl && !altdomain) {
/* SSL tunneled connection on 5223 port is legacy and doesn't
* have an SRV record. */
} else if (!conn->tls_legacy_ssl) {
found = resolver_srv_lookup(conn->ctx, "xmpp-client", "tcp", domain,
&srv_rr_list);
altdomain = domain;
}
altport = altport ? altport : _conn_default_port(conn, XMPP_CLIENT);
if (XMPP_DOMAIN_NOT_FOUND == found) {
strophe_debug(conn->ctx, "xmpp",
"SRV lookup failed, "
"connecting via domain.");
host = domain;
port = altport ? altport : _conn_default_port(conn, XMPP_CLIENT);
found = XMPP_DOMAIN_ALTDOMAIN;
}
rr = srv_rr_list;
do {
if (XMPP_DOMAIN_FOUND == found && rr != NULL) {
host = rr->target;
port = rr->port;
rr = rr->next;
}
rc = _conn_connect(conn, domain, host, port, XMPP_CLIENT, callback,
userdata);
} while (rc != 0 && rr != NULL);
if (conn->xsock)
sock_free(conn->xsock);
conn->xsock = sock_new(conn, domain, altdomain, altport);
if (!conn->xsock)
goto err_mem;
rc = _conn_connect(conn, domain, XMPP_CLIENT, callback, userdata);
strophe_free(conn->ctx, domain);
resolver_srv_free(conn->ctx, srv_rr_list);
return rc;
err_mem:
strophe_free(conn->ctx, domain);
return XMPP_EMEM;
}
/** Initiate a component connection to server.
@@ -826,10 +807,15 @@ int xmpp_connect_component(xmpp_conn_t *conn,
}
port = port ? port : _conn_default_port(conn, XMPP_COMPONENT);
if (conn->xsock)
sock_free(conn->xsock);
conn->xsock = sock_new(conn, NULL, server, port);
if (!conn->xsock)
return XMPP_EMEM;
/* JID serves as an identifier here and will be used as "to" attribute
of the stream */
return _conn_connect(conn, conn->jid, server, port, XMPP_COMPONENT,
callback, userdata);
return _conn_connect(conn, conn->jid, XMPP_COMPONENT, callback, userdata);
}
/** Initiate a raw connection to the XMPP server.
@@ -977,7 +963,8 @@ void conn_disconnect(xmpp_conn_t *conn)
tls_free(conn->tls);
conn->tls = NULL;
}
sock_close(conn->sock);
if (conn->sock != INVALID_SOCKET)
sock_close(conn->sock);
_reset_sm_state_for_reconnect(conn);
/* fire off connection handler */
@@ -1845,8 +1832,6 @@ static void _conn_reset(xmpp_conn_t *conn)
static int _conn_connect(xmpp_conn_t *conn,
const char *domain,
const char *host,
unsigned short port,
xmpp_conn_type_t type,
xmpp_conn_handler callback,
void *userdata)
@@ -1857,8 +1842,6 @@ static int _conn_connect(xmpp_conn_t *conn,
return XMPP_EINVOP;
if (type != XMPP_CLIENT && type != XMPP_COMPONENT)
return XMPP_EINVOP;
if (host == NULL || port == 0)
return XMPP_EINT;
_conn_reset(conn);
@@ -1867,10 +1850,8 @@ static int _conn_connect(xmpp_conn_t *conn,
if (!conn->domain)
return XMPP_EMEM;
conn->sock = sock_connect(conn, host, port);
strophe_debug(conn->ctx, "xmpp", "sock_connect() to %s:%u returned %d",
host, port, conn->sock);
if (conn->sock == -1)
conn->sock = sock_connect(conn->xsock);
if (conn->sock == INVALID_SOCKET)
return XMPP_EINT;
/* setup handler */
@@ -1889,7 +1870,6 @@ static int _conn_connect(xmpp_conn_t *conn,
conn->state = XMPP_STATE_CONNECTING;
conn->timeout_stamp = time_stamp();
strophe_debug(conn->ctx, "xmpp", "Attempting to connect to %s", host);
return 0;
}

27
src/event.c
View File

@@ -60,6 +60,18 @@
/** Max buffer size for receiving messages. */
#define STROPE_MESSAGE_BUFFER_SIZE 4096
static int _connect_next(xmpp_conn_t *conn)
{
sock_close(conn->sock);
conn->sock = sock_connect(conn->xsock);
if (conn->sock == INVALID_SOCKET)
return -1;
conn->timeout_stamp = time_stamp();
return 0;
}
/** Run the event loop once.
* This function will run send any data that has been queued by
* xmpp_send and related functions and run through the Strophe even
@@ -209,9 +221,14 @@ next_item:
conn->connect_timeout)
FD_SET(conn->sock, &wfds);
else {
conn->error = ETIMEDOUT;
strophe_info(ctx, "xmpp", "Connection attempt timed out.");
conn_disconnect(conn);
ret = _connect_next(conn);
if (ret != 0) {
conn->error = ETIMEDOUT;
conn_disconnect(conn);
} else {
FD_SET(conn->sock, &wfds);
}
}
break;
case XMPP_STATE_CONNECTED:
@@ -272,7 +289,11 @@ next_item:
/* connection failed */
strophe_debug(ctx, "xmpp", "connection failed, error %d",
ret);
conn_disconnect(conn);
ret = _connect_next(conn);
if (ret != 0) {
conn->error = ret;
conn_disconnect(conn);
}
break;
}

43
src/resolver.c
View File

@@ -91,6 +91,25 @@ void resolver_shutdown(void)
#endif
}
resolver_srv_rr_t *resolver_srv_rr_new(xmpp_ctx_t *ctx,
const char *host,
unsigned short port,
unsigned short prio,
unsigned short weight)
{
resolver_srv_rr_t *rr = strophe_alloc(ctx, sizeof(*rr));
if (rr) {
memset(rr, 0, sizeof(*rr));
rr->port = port;
rr->priority = prio;
rr->weight = weight;
if (host) {
snprintf(rr->target, sizeof(rr->target), "%s", host);
}
}
return rr;
}
static void resolver_srv_list_sort(resolver_srv_rr_t **srv_rr_list)
{
resolver_srv_rr_t *rr_head;
@@ -438,17 +457,19 @@ static int resolver_raw_srv_lookup_buf(xmpp_ctx_t *ctx,
rdlength = xmpp_ntohs_ptr(&buf[j + 8]);
j += 10;
if (type == MESSAGE_T_SRV && class == MESSAGE_C_IN) {
rr = strophe_alloc(ctx, sizeof(*rr));
rr->next = *srv_rr_list;
rr->priority = xmpp_ntohs_ptr(&buf[j]);
rr->weight = xmpp_ntohs_ptr(&buf[j + 2]);
rr->port = xmpp_ntohs_ptr(&buf[j + 4]);
name_len = message_name_get(buf, len, j + 6, rr->target,
sizeof(rr->target));
if (name_len > 0)
*srv_rr_list = rr;
else
strophe_free(ctx, rr); /* skip broken record */
rr = resolver_srv_rr_new(ctx, NULL, 0, 0, 0);
if (rr) {
rr->next = *srv_rr_list;
rr->priority = xmpp_ntohs_ptr(&buf[j]);
rr->weight = xmpp_ntohs_ptr(&buf[j + 2]);
rr->port = xmpp_ntohs_ptr(&buf[j + 4]);
name_len = message_name_get(buf, len, j + 6, rr->target,
sizeof(rr->target));
if (name_len > 0)
*srv_rr_list = rr;
else
strophe_free(ctx, rr); /* skip broken record */
}
}
j += rdlength;
}

6
src/resolver.h
View File

@@ -36,6 +36,12 @@ typedef struct resolver_srv_rr_struc {
void resolver_initialize(void);
void resolver_shutdown(void);
resolver_srv_rr_t *resolver_srv_rr_new(xmpp_ctx_t *ctx,
const char *host,
unsigned short port,
unsigned short prio,
unsigned short weight);
/** Perform lookup for RFC1035 message format.
* This function allocates all elements.
*

207
src/sock.c
View File

@@ -24,6 +24,7 @@
#include <iphlpapi.h>
#include <mstcpip.h> /* tcp_keepalive */
#else
#include <arpa/inet.h>
#include <errno.h>
#include <unistd.h>
#include <sys/socket.h>
@@ -34,6 +35,18 @@
#endif
#include "common.h"
#include "resolver.h"
struct _xmpp_sock_t {
xmpp_ctx_t *ctx;
xmpp_conn_t *conn;
struct addrinfo *ainfo_list;
struct addrinfo *ainfo_cur;
resolver_srv_rr_t *srv_rr_list;
resolver_srv_rr_t *srv_rr_cur;
const char *host;
unsigned short port;
};
void sock_initialize(void)
{
@@ -68,50 +81,164 @@ static int _in_progress(int error)
#endif
}
sock_t sock_connect(xmpp_conn_t *conn, const char *host, unsigned short port)
static void sock_getaddrinfo(xmpp_sock_t *xsock)
{
sock_t sock;
char service[6];
struct addrinfo *res, *ainfo, hints;
int err;
struct addrinfo hints;
int rc;
strophe_snprintf(service, 6, "%u", port);
memset(&hints, 0, sizeof(struct addrinfo));
hints.ai_family = AF_UNSPEC;
#ifdef AI_ADDRCONFIG
hints.ai_flags = AI_ADDRCONFIG;
#endif /* AI_ADDRCONFIG */
hints.ai_protocol = IPPROTO_TCP;
hints.ai_socktype = SOCK_STREAM;
err = getaddrinfo(host, service, &hints, &res);
if (err != 0)
return -1;
for (ainfo = res; ainfo != NULL; ainfo = ainfo->ai_next) {
sock = socket(ainfo->ai_family, ainfo->ai_socktype, ainfo->ai_protocol);
if (sock < 0)
continue;
if (conn->sockopt_cb != NULL)
err = (conn->sockopt_cb)(conn, &sock);
if (err != 0) {
sock_close(sock);
continue;
}
err = sock_set_nonblocking(sock);
if (err == 0) {
err = connect(sock, ainfo->ai_addr, ainfo->ai_addrlen);
if (err == 0 || _in_progress(sock_error()))
break;
}
sock_close(sock);
if (xsock->ainfo_list) {
freeaddrinfo(xsock->ainfo_list);
xsock->ainfo_list = NULL;
}
freeaddrinfo(res);
sock = ainfo == NULL ? -1 : sock;
if (xsock->srv_rr_cur) {
/* Cache host and port for debug logs. */
xsock->host = xsock->srv_rr_cur->target;
xsock->port = xsock->srv_rr_cur->port;
strophe_snprintf(service, 6, "%u", xsock->srv_rr_cur->port);
memset(&hints, 0, sizeof(struct addrinfo));
hints.ai_family = AF_UNSPEC;
#ifdef AI_ADDRCONFIG
hints.ai_flags = AI_ADDRCONFIG;
#endif /* AI_ADDRCONFIG */
hints.ai_protocol = IPPROTO_TCP;
hints.ai_socktype = SOCK_STREAM;
rc = getaddrinfo(xsock->srv_rr_cur->target, service, &hints,
&xsock->ainfo_list);
if (rc != 0) {
strophe_debug(xsock->ctx, "sock", "getaddrinfo() failed with %d",
rc);
xsock->ainfo_list = NULL;
}
}
xsock->ainfo_cur = xsock->ainfo_list;
}
xmpp_sock_t *sock_new(xmpp_conn_t *conn,
const char *domain,
const char *host,
unsigned short port)
{
xmpp_ctx_t *ctx = conn->ctx;
xmpp_sock_t *xsock;
int found = XMPP_DOMAIN_NOT_FOUND;
xsock = strophe_alloc(ctx, sizeof(*xsock));
if (!xsock) {
return NULL;
}
xsock->ctx = ctx;
xsock->conn = conn;
xsock->host = NULL;
xsock->port = 0;
if (!host) {
found = resolver_srv_lookup(ctx, "xmpp-client", "tcp", domain,
&xsock->srv_rr_list);
if (XMPP_DOMAIN_NOT_FOUND == found)
strophe_debug(ctx, "sock",
"SRV lookup failed, connecting via domain.");
}
if (XMPP_DOMAIN_NOT_FOUND == found) {
/* Resolution failed or the host is provided explicitly. */
xsock->srv_rr_list =
resolver_srv_rr_new(ctx, host ? host : domain, port, 0, 0);
}
xsock->srv_rr_cur = xsock->srv_rr_list;
xsock->ainfo_list = NULL;
sock_getaddrinfo(xsock);
if (xsock->srv_rr_cur)
xsock->srv_rr_cur = xsock->srv_rr_cur->next;
return xsock;
}
void sock_free(xmpp_sock_t *xsock)
{
if (!xsock)
return;
if (xsock->ainfo_list)
freeaddrinfo(xsock->ainfo_list);
if (xsock->srv_rr_list)
resolver_srv_free(xsock->ctx, xsock->srv_rr_list);
strophe_free(xsock->ctx, xsock);
}
static const char *_sockaddr2str(struct sockaddr *sa, char *buf, size_t buflen)
{
buf[0] = '\0';
switch (sa->sa_family) {
case AF_INET:
inet_ntop(AF_INET, &((struct sockaddr_in *)sa)->sin_addr, buf, buflen);
break;
case AF_INET6:
inet_ntop(AF_INET6, &((struct sockaddr_in6 *)sa)->sin6_addr, buf,
buflen);
break;
default:
strophe_snprintf(buf, buflen, "<Unknown>");
}
return buf;
}
sock_t sock_connect(xmpp_sock_t *xsock)
{
struct addrinfo *ainfo;
sock_t sock;
int rc = 0;
char buf[64];
do {
if (!xsock->ainfo_cur) {
sock_getaddrinfo(xsock);
if (xsock->srv_rr_cur)
xsock->srv_rr_cur = xsock->srv_rr_cur->next;
}
if (!xsock->ainfo_cur) {
/* We tried all available addresses. */
return INVALID_SOCKET;
}
ainfo = xsock->ainfo_cur;
strophe_debug(xsock->ctx, "sock", "Connecting to %s:%u via %s",
xsock->host, xsock->port,
_sockaddr2str(ainfo->ai_addr, buf, sizeof(buf)));
sock = socket(ainfo->ai_family, ainfo->ai_socktype, ainfo->ai_protocol);
if (sock != INVALID_SOCKET) {
if (xsock->conn->sockopt_cb) {
/* Don't allow user to overwrite sockfd value. */
sock_t sock_copy = sock;
rc = xsock->conn->sockopt_cb(xsock->conn, &sock_copy);
if (rc != 0) {
strophe_debug(xsock->ctx, "sock",
"User's setsockopt callback"
"failed with %d (errno=%d)",
rc, errno);
}
}
if (rc == 0)
rc = sock_set_nonblocking(sock);
if (rc == 0)
rc = connect(sock, ainfo->ai_addr, ainfo->ai_addrlen);
/* Assume only connect() can cause "in progress" error. */
if (rc != 0 && !_in_progress(sock_error())) {
sock_close(sock);
sock = INVALID_SOCKET;
}
}
strophe_debug(xsock->ctx, "sock", "sock_connect() result %d", sock);
xsock->ainfo_cur = xsock->ainfo_cur->ai_next;
} while (sock == INVALID_SOCKET);
return sock;
}

10
src/sock.h
View File

@@ -20,6 +20,7 @@
#ifndef _WIN32
typedef int sock_t;
#define INVALID_SOCKET (-1)
#else
#include <winsock2.h>
#include <ws2tcpip.h>
@@ -28,12 +29,19 @@ typedef int sock_t;
typedef SOCKET sock_t;
#endif
typedef struct _xmpp_sock_t xmpp_sock_t;
void sock_initialize(void);
void sock_shutdown(void);
int sock_error(void);
sock_t sock_connect(xmpp_conn_t *conn, const char *host, unsigned short port);
xmpp_sock_t *sock_new(xmpp_conn_t *conn,
const char *domain,
const char *host,
unsigned short port);
void sock_free(xmpp_sock_t *xsock);
sock_t sock_connect(xmpp_sock_t *xsock);
int sock_close(sock_t sock);
int sock_set_blocking(sock_t sock);