/*
 * librdkafka - Apache Kafka C library
 *
 * Copyright (c) 2015, Magnus Edenhill
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met: 
 * 
 * 1. Redistributions of source code must retain the above copyright notice,
 *    this list of conditions and the following disclaimer. 
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution. 
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */
#ifdef _MSC_VER
#pragma comment(lib, "ws2_32.lib")
#endif

#define __need_IOV_MAX

#define _DARWIN_C_SOURCE  /* MSG_DONTWAIT */

#include "rdkafka_int.h"
#include "rdaddr.h"
#include "rdkafka_transport.h"
#include "rdkafka_transport_int.h"
#include "rdkafka_broker.h"
#include "rdkafka_interceptor.h"

#include <errno.h>

#if WITH_VALGRIND
/* OpenSSL relies on uninitialized memory, which Valgrind will whine about.
 * We use in-code Valgrind macros to suppress those warnings. */
#include <valgrind/memcheck.h>
#else
#define VALGRIND_MAKE_MEM_DEFINED(A,B)
#endif


#ifdef _MSC_VER
#define socket_errno WSAGetLastError()
#else
#include <sys/socket.h>
#define socket_errno errno
#define SOCKET_ERROR -1
#endif

/* AIX doesn't have MSG_DONTWAIT */
#ifndef MSG_DONTWAIT
#  define MSG_DONTWAIT MSG_NONBLOCK
#endif


#if WITH_SSL
static mtx_t *rd_kafka_ssl_locks;
static int    rd_kafka_ssl_locks_cnt;
#endif



/**
 * Low-level socket close
 */
static void rd_kafka_transport_close0 (rd_kafka_t *rk, int s) {
        if (rk->rk_conf.closesocket_cb)
                rk->rk_conf.closesocket_cb(s, rk->rk_conf.opaque);
        else {
#ifndef _MSC_VER
		close(s);
#else
		closesocket(s);
#endif
        }

}

/**
 * Close and destroy a transport handle
 */
void rd_kafka_transport_close (rd_kafka_transport_t *rktrans) {
#if WITH_SSL
	if (rktrans->rktrans_ssl) {
		SSL_shutdown(rktrans->rktrans_ssl);
		SSL_free(rktrans->rktrans_ssl);
	}
#endif

        rd_kafka_sasl_close(rktrans);

	if (rktrans->rktrans_recv_buf)
		rd_kafka_buf_destroy(rktrans->rktrans_recv_buf);

	if (rktrans->rktrans_s != -1)
                rd_kafka_transport_close0(rktrans->rktrans_rkb->rkb_rk,
                                          rktrans->rktrans_s);

	rd_free(rktrans);
}


static const char *socket_strerror(int err) {
#ifdef _MSC_VER
	static RD_TLS char buf[256];
        rd_strerror_w32(err, buf, sizeof(buf));
	return buf;
#else
	return rd_strerror(err);
#endif
}




#ifndef _MSC_VER
/**
 * @brief sendmsg() abstraction, converting a list of segments to iovecs.
 * @remark should only be called if the number of segments is > 1.
 */
ssize_t rd_kafka_transport_socket_sendmsg (rd_kafka_transport_t *rktrans,
                                           rd_slice_t *slice,
                                           char *errstr, size_t errstr_size) {
        struct iovec iov[IOV_MAX];
        struct msghdr msg = { .msg_iov = iov };
        size_t iovlen;
        ssize_t r;

        rd_slice_get_iov(slice, msg.msg_iov, &iovlen, IOV_MAX,
                         /* FIXME: Measure the effects of this */
                         rktrans->rktrans_sndbuf_size);
        msg.msg_iovlen = (int)iovlen;

#ifdef sun
        /* See recvmsg() comment. Setting it here to be safe. */
        socket_errno = EAGAIN;
#endif

        r = sendmsg(rktrans->rktrans_s, &msg, MSG_DONTWAIT
#ifdef MSG_NOSIGNAL
                    | MSG_NOSIGNAL
#endif
                );

        if (r == -1) {
                if (socket_errno == EAGAIN)
                        return 0;
                rd_snprintf(errstr, errstr_size, "%s", rd_strerror(errno));
        }

        /* Update buffer read position */
        rd_slice_read(slice, NULL, (size_t)r);

        return r;
}
#endif


/**
 * @brief Plain send() abstraction
 */
static ssize_t
rd_kafka_transport_socket_send0 (rd_kafka_transport_t *rktrans,
                                 rd_slice_t *slice,
                                 char *errstr, size_t errstr_size) {
        ssize_t sum = 0;
        const void *p;
        size_t rlen;

        while ((rlen = rd_slice_peeker(slice, &p))) {
                ssize_t r;

                r = send(rktrans->rktrans_s, p,
#ifdef _MSC_VER
                         (int)rlen, (int)0
#else
                         rlen, 0
#endif
                );

#ifdef _MSC_VER
                if (unlikely(r == SOCKET_ERROR)) {
                        if (sum > 0 || WSAGetLastError() == WSAEWOULDBLOCK)
                                return sum;
                        else {
                                rd_snprintf(errstr, errstr_size, "%s",
                                            socket_strerror(WSAGetLastError()));
                                return -1;
                        }
                }
#else
                if (unlikely(r <= 0)) {
                        if (r == 0 || errno == EAGAIN)
                                return 0;
                        rd_snprintf(errstr, errstr_size, "%s",
                                    socket_strerror(socket_errno));
                        return -1;
                }
#endif

                /* Update buffer read position */
                rd_slice_read(slice, NULL, (size_t)r);

                sum += r;

                /* FIXME: remove this and try again immediately and let
                 *        the next write() call fail instead? */
                if ((size_t)r < rlen)
                        break;
        }

        return sum;
}


static ssize_t
rd_kafka_transport_socket_send (rd_kafka_transport_t *rktrans,
                                rd_slice_t *slice,
                                char *errstr, size_t errstr_size) {
#ifndef _MSC_VER
        /* FIXME: Use sendmsg() with iovecs if there's more than one segment
         * remaining, otherwise (or if platform does not have sendmsg)
         * use plain send(). */
        return rd_kafka_transport_socket_sendmsg(rktrans, slice,
                                                 errstr, errstr_size);
#endif
        return rd_kafka_transport_socket_send0(rktrans, slice,
                                               errstr, errstr_size);
}



#ifndef _MSC_VER
/**
 * @brief recvmsg() abstraction, converting a list of segments to iovecs.
 * @remark should only be called if the number of segments is > 1.
 */
static ssize_t
rd_kafka_transport_socket_recvmsg (rd_kafka_transport_t *rktrans,
                                   rd_buf_t *rbuf,
                                   char *errstr, size_t errstr_size) {
        ssize_t r;
        struct iovec iov[IOV_MAX];
        struct msghdr msg = { .msg_iov = iov };
        size_t iovlen;

        rd_buf_get_write_iov(rbuf, msg.msg_iov, &iovlen, IOV_MAX,
                             /* FIXME: Measure the effects of this */
                             rktrans->rktrans_rcvbuf_size);
        msg.msg_iovlen = (int)iovlen;

#ifdef sun
        /* SunOS doesn't seem to set errno when recvmsg() fails
         * due to no data and MSG_DONTWAIT is set. */
        socket_errno = EAGAIN;
#endif
        r = recvmsg(rktrans->rktrans_s, &msg, MSG_DONTWAIT);
        if (unlikely(r <= 0)) {
                if (r == -1 && socket_errno == EAGAIN)
                        return 0;
                else if (r == 0) {
                        /* Receive 0 after POLLIN event means
                         * connection closed. */
                        rd_snprintf(errstr, errstr_size, "Disconnected");
                        return -1;
                } else if (r == -1) {
                        rd_snprintf(errstr, errstr_size, "%s",
                                    rd_strerror(errno));
                        return -1;
                }
        }

        /* Update buffer write position */
        rd_buf_write(rbuf, NULL, (size_t)r);

        return r;
}
#endif


/**
 * @brief Plain recv()
 */
static ssize_t
rd_kafka_transport_socket_recv0 (rd_kafka_transport_t *rktrans,
                                 rd_buf_t *rbuf,
                                 char *errstr, size_t errstr_size) {
        ssize_t sum = 0;
        void *p;
        size_t len;

        while ((len = rd_buf_get_writable(rbuf, &p))) {
                ssize_t r;

                r = recv(rktrans->rktrans_s, p,
#ifdef _MSC_VER
                         (int)
#endif
                         len,
                         0);

#ifdef _MSC_VER
                if (unlikely(r == SOCKET_ERROR)) {
                        if (WSAGetLastError() == WSAEWOULDBLOCK)
                                return sum;
                        rd_snprintf(errstr, errstr_size, "%s",
                                    socket_strerror(WSAGetLastError()));
                        return -1;
                }
#else
                if (unlikely(r <= 0)) {
                        if (r == -1 && socket_errno == EAGAIN)
                                return 0;
                        else if (r == 0) {
                                /* Receive 0 after POLLIN event means
                                 * connection closed. */
                                rd_snprintf(errstr, errstr_size,
                                            "Disconnected");
                                return -1;
                        } else if (r == -1) {
                                rd_snprintf(errstr, errstr_size, "%s",
                                            rd_strerror(errno));
                                return -1;
                        }
                }
#endif

                /* Update buffer write position */
                rd_buf_write(rbuf, NULL, (size_t)r);

                sum += r;

                /* FIXME: remove this and try again immediately and let
                 *        the next recv() call fail instead? */
                if ((size_t)r < len)
                        break;
        }
        return sum;
}


static ssize_t
rd_kafka_transport_socket_recv (rd_kafka_transport_t *rktrans,
                                rd_buf_t *buf,
                                char *errstr, size_t errstr_size) {
#ifndef _MSC_VER
        /* FIXME: Use recvmsg() with iovecs if there's more than one segment
         * remaining, otherwise (or if platform does not have sendmsg)
         * use plain send(). */
        return rd_kafka_transport_socket_recvmsg(rktrans, buf,
                                                 errstr, errstr_size);
#endif
        return rd_kafka_transport_socket_recv0(rktrans, buf,
                                               errstr, errstr_size);
}





/**
 * CONNECT state is failed (errstr!=NULL) or done (TCP is up, SSL is working..).
 * From this state we either hand control back to the broker code,
 * or if authentication is configured we ente the AUTH state.
 */
void rd_kafka_transport_connect_done (rd_kafka_transport_t *rktrans,
				      char *errstr) {
	rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;

	rd_kafka_broker_connect_done(rkb, errstr);
}



#if WITH_SSL


/**
 * Serves the entire OpenSSL error queue and logs each error.
 * The last error is not logged but returned in 'errstr'.
 *
 * If 'rkb' is non-NULL broker-specific logging will be used,
 * else it will fall back on global 'rk' debugging.
 */
static char *rd_kafka_ssl_error (rd_kafka_t *rk, rd_kafka_broker_t *rkb,
				 char *errstr, size_t errstr_size) {
    unsigned long l;
    const char *file, *data;
    int line, flags;
    int cnt = 0;

    while ((l = ERR_get_error_line_data(&file, &line, &data, &flags)) != 0) {
	char buf[256];

	if (cnt++ > 0) {
		/* Log last message */
		if (rkb)
			rd_rkb_log(rkb, LOG_ERR, "SSL", "%s", errstr);
		else
			rd_kafka_log(rk, LOG_ERR, "SSL", "%s", errstr);
	}
	
	ERR_error_string_n(l, buf, sizeof(buf));

	rd_snprintf(errstr, errstr_size, "%s:%d: %s: %s",
		    file, line, buf, (flags & ERR_TXT_STRING) ? data : "");

    }

    if (cnt == 0)
    	    rd_snprintf(errstr, errstr_size, "No error");
    
    return errstr;
}


static void rd_kafka_transport_ssl_lock_cb (int mode, int i,
					    const char *file, int line) {
	if (mode & CRYPTO_LOCK)
		mtx_lock(&rd_kafka_ssl_locks[i]);
	else
		mtx_unlock(&rd_kafka_ssl_locks[i]);
}

static unsigned long rd_kafka_transport_ssl_threadid_cb (void) {
#ifdef _MSC_VER
        /* Windows makes a distinction between thread handle
         * and thread id, which means we can't use the
         * thrd_current() API that returns the handle. */
        return (unsigned long)GetCurrentThreadId();
#else
        return (unsigned long)(intptr_t)thrd_current();
#endif
}


/**
 * Global OpenSSL cleanup.
 */
void rd_kafka_transport_ssl_term (void) {
	int i;

	CRYPTO_set_id_callback(NULL);
	CRYPTO_set_locking_callback(NULL);
        CRYPTO_cleanup_all_ex_data();

	for (i = 0 ; i < rd_kafka_ssl_locks_cnt ; i++)
		mtx_destroy(&rd_kafka_ssl_locks[i]);

	rd_free(rd_kafka_ssl_locks);

}


/**
 * Global OpenSSL init.
 */
void rd_kafka_transport_ssl_init (void) {
	int i;
	
	rd_kafka_ssl_locks_cnt = CRYPTO_num_locks();
	rd_kafka_ssl_locks = rd_malloc(rd_kafka_ssl_locks_cnt *
				       sizeof(*rd_kafka_ssl_locks));
	for (i = 0 ; i < rd_kafka_ssl_locks_cnt ; i++)
		mtx_init(&rd_kafka_ssl_locks[i], mtx_plain);

	CRYPTO_set_id_callback(rd_kafka_transport_ssl_threadid_cb);
	CRYPTO_set_locking_callback(rd_kafka_transport_ssl_lock_cb);
	
	SSL_load_error_strings();
	SSL_library_init();
	OpenSSL_add_all_algorithms();
}


/**
 * Set transport IO event polling based on SSL error.
 *
 * Returns -1 on permanent errors.
 *
 * Locality: broker thread
 */
static RD_INLINE int
rd_kafka_transport_ssl_io_update (rd_kafka_transport_t *rktrans, int ret,
				  char *errstr, size_t errstr_size) {
	int serr = SSL_get_error(rktrans->rktrans_ssl, ret);
	int serr2;

	switch (serr)
	{
	case SSL_ERROR_WANT_READ:
		rd_kafka_transport_poll_set(rktrans, POLLIN);
		break;

	case SSL_ERROR_WANT_WRITE:
	case SSL_ERROR_WANT_CONNECT:
		rd_kafka_transport_poll_set(rktrans, POLLOUT);
		break;

	case SSL_ERROR_SYSCALL:
		if (!(serr2 = SSL_get_error(rktrans->rktrans_ssl, ret))) {
			if (ret == 0)
				errno = ECONNRESET;
			rd_snprintf(errstr, errstr_size,
				    "SSL syscall error: %s", rd_strerror(errno));
		} else
			rd_snprintf(errstr, errstr_size,
				    "SSL syscall error number: %d: %s", serr2,
				    rd_strerror(errno));
		return -1;

        case SSL_ERROR_ZERO_RETURN:
                rd_snprintf(errstr, errstr_size, "Disconnected");
                return -1;

	default:
		rd_kafka_ssl_error(NULL, rktrans->rktrans_rkb,
				   errstr, errstr_size);
		return -1;
	}

	return 0;
}

static ssize_t
rd_kafka_transport_ssl_send (rd_kafka_transport_t *rktrans,
                             rd_slice_t *slice,
                             char *errstr, size_t errstr_size) {
	ssize_t sum = 0;
        const void *p;
        size_t rlen;

        while ((rlen = rd_slice_peeker(slice, &p))) {
                int r;

                r = SSL_write(rktrans->rktrans_ssl, p, (int)rlen);

		if (unlikely(r <= 0)) {
			if (rd_kafka_transport_ssl_io_update(rktrans, r,
							     errstr,
							     errstr_size) == -1)
				return -1;
			else
				return sum;
		}

                /* Update buffer read position */
                rd_slice_read(slice, NULL, (size_t)r);

		sum += r;
                 /* FIXME: remove this and try again immediately and let
                  *        the next SSL_write() call fail instead? */
                if ((size_t)r < rlen)
                        break;

	}
	return sum;
}

static ssize_t
rd_kafka_transport_ssl_recv (rd_kafka_transport_t *rktrans,
                             rd_buf_t *rbuf, char *errstr, size_t errstr_size) {
	ssize_t sum = 0;
        void *p;
        size_t len;

        while ((len = rd_buf_get_writable(rbuf, &p))) {
		int r;

                r = SSL_read(rktrans->rktrans_ssl, p, (int)len);

		if (unlikely(r <= 0)) {
			if (rd_kafka_transport_ssl_io_update(rktrans, r,
							     errstr,
							     errstr_size) == -1)
				return -1;
			else
				return sum;
		}

                VALGRIND_MAKE_MEM_DEFINED(p, r);

                /* Update buffer write position */
                rd_buf_write(rbuf, NULL, (size_t)r);

		sum += r;

                 /* FIXME: remove this and try again immediately and let
                  *        the next SSL_read() call fail instead? */
                if ((size_t)r < len)
                        break;

	}
	return sum;

}


/**
 * OpenSSL password query callback
 *
 * Locality: application thread
 */
static int rd_kafka_transport_ssl_passwd_cb (char *buf, int size, int rwflag,
					     void *userdata) {
	rd_kafka_t *rk = userdata;
	int pwlen;

	rd_kafka_dbg(rk, SECURITY, "SSLPASSWD",
		     "Private key file \"%s\" requires password",
		     rk->rk_conf.ssl.key_location);

	if (!rk->rk_conf.ssl.key_password) {
		rd_kafka_log(rk, LOG_WARNING, "SSLPASSWD",
			     "Private key file \"%s\" requires password but "
			     "no password configured (ssl.key.password)",
			     rk->rk_conf.ssl.key_location);
		return -1;
	}


	pwlen = (int) strlen(rk->rk_conf.ssl.key_password);
	memcpy(buf, rk->rk_conf.ssl.key_password, RD_MIN(pwlen, size));

	return pwlen;
}

/**
 * Set up SSL for a newly connected connection
 *
 * Returns -1 on failure, else 0.
 */
static int rd_kafka_transport_ssl_connect (rd_kafka_broker_t *rkb,
					   rd_kafka_transport_t *rktrans,
					   char *errstr, size_t errstr_size) {
	int r;
	char name[RD_KAFKA_NODENAME_SIZE];
	char *t;

	rktrans->rktrans_ssl = SSL_new(rkb->rkb_rk->rk_conf.ssl.ctx);
	if (!rktrans->rktrans_ssl)
		goto fail;

	if (!SSL_set_fd(rktrans->rktrans_ssl, rktrans->rktrans_s))
		goto fail;

#if (OPENSSL_VERSION_NUMBER >= 0x0090806fL) && !defined(OPENSSL_NO_TLSEXT)
	/* If non-numerical hostname, send it for SNI */
	rd_snprintf(name, sizeof(name), "%s", rkb->rkb_nodename);
	if ((t = strrchr(name, ':')))
		*t = '\0';
	if (!(/*ipv6*/(strchr(name, ':') &&
		       strspn(name, "0123456789abcdefABCDEF:.[]%") == strlen(name)) ||
	      /*ipv4*/strspn(name, "0123456789.") == strlen(name)) &&
	    !SSL_set_tlsext_host_name(rktrans->rktrans_ssl, name))
		goto fail;
#endif

	r = SSL_connect(rktrans->rktrans_ssl);
	if (r == 1) {
		/* Connected, highly unlikely since this is a
		 * non-blocking operation. */
		rd_kafka_transport_connect_done(rktrans, NULL);
		return 0;
	}

		
	if (rd_kafka_transport_ssl_io_update(rktrans, r,
					     errstr, errstr_size) == -1)
		return -1;
	
	return 0;

 fail:
	rd_kafka_ssl_error(NULL, rkb, errstr, errstr_size);
	return -1;
}


static RD_UNUSED int
rd_kafka_transport_ssl_io_event (rd_kafka_transport_t *rktrans, int events) {
	int r;
	char errstr[512];

	if (events & POLLOUT) {
		r = SSL_write(rktrans->rktrans_ssl, NULL, 0);
		if (rd_kafka_transport_ssl_io_update(rktrans, r,
						     errstr,
						     sizeof(errstr)) == -1)
			goto fail;
	}

	return 0;

 fail:
	/* Permanent error */
	rd_kafka_broker_fail(rktrans->rktrans_rkb, LOG_ERR,
                             RD_KAFKA_RESP_ERR__TRANSPORT,
			     "%s", errstr);
	return -1;
}


/**
 * Verify SSL handshake was valid.
 */
static int rd_kafka_transport_ssl_verify (rd_kafka_transport_t *rktrans) {
	long int rl;
	X509 *cert;

	cert = SSL_get_peer_certificate(rktrans->rktrans_ssl);
	X509_free(cert);
	if (!cert) {
		rd_kafka_broker_fail(rktrans->rktrans_rkb, LOG_ERR,
				     RD_KAFKA_RESP_ERR__SSL,
				     "Broker did not provide a certificate");
		return -1;
	}

	if ((rl = SSL_get_verify_result(rktrans->rktrans_ssl)) != X509_V_OK) {
		rd_kafka_broker_fail(rktrans->rktrans_rkb, LOG_ERR,
				     RD_KAFKA_RESP_ERR__SSL,
				     "Failed to verify broker certificate: %s",
				     X509_verify_cert_error_string(rl));
		return -1;
	}

	rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "SSLVERIFY",
		   "Broker SSL certificate verified");
	return 0;
}

/**
 * SSL handshake handling.
 * Call repeatedly (based on IO events) until handshake is done.
 *
 * Returns -1 on error, 0 if handshake is still in progress, or 1 on completion.
 */
static int rd_kafka_transport_ssl_handshake (rd_kafka_transport_t *rktrans) {
	rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;
	char errstr[512];
	int r;

	r = SSL_do_handshake(rktrans->rktrans_ssl);
	if (r == 1) {
		/* SSL handshake done. Verify. */
		if (rd_kafka_transport_ssl_verify(rktrans) == -1)
			return -1;

		rd_kafka_transport_connect_done(rktrans, NULL);
		return 1;

	} else if (rd_kafka_transport_ssl_io_update(rktrans, r,
						    errstr,
						    sizeof(errstr)) == -1) {
		rd_kafka_broker_fail(rkb, LOG_ERR, RD_KAFKA_RESP_ERR__SSL,
				     "SSL handshake failed: %s%s", errstr,
				     strstr(errstr, "unexpected message") ?
				     ": client authentication might be "
				     "required (see broker log)" : "");
		return -1;
	}

	return 0;
}


/**
 * Once per rd_kafka_t handle cleanup of OpenSSL
 *
 * Locality: any thread
 *
 * NOTE: rd_kafka_wrlock() MUST be held
 */
void rd_kafka_transport_ssl_ctx_term (rd_kafka_t *rk) {
	SSL_CTX_free(rk->rk_conf.ssl.ctx);
	rk->rk_conf.ssl.ctx = NULL;
}

/**
 * Once per rd_kafka_t handle initialization of OpenSSL
 *
 * Locality: application thread
 *
 * NOTE: rd_kafka_wrlock() MUST be held
 */
int rd_kafka_transport_ssl_ctx_init (rd_kafka_t *rk,
				     char *errstr, size_t errstr_size) {
	int r;
	SSL_CTX *ctx;

        if (errstr_size > 0)
                errstr[0] = '\0';

	ctx = SSL_CTX_new(SSLv23_client_method());
        if (!ctx) {
                rd_snprintf(errstr, errstr_size,
                            "SSLv23_client_method() failed: ");
                goto fail;
        }

#ifdef SSL_OP_NO_SSLv3
	/* Disable SSLv3 (unsafe) */
	SSL_CTX_set_options(ctx, SSL_OP_NO_SSLv3);
#endif

	/* Key file password callback */
	SSL_CTX_set_default_passwd_cb(ctx, rd_kafka_transport_ssl_passwd_cb);
	SSL_CTX_set_default_passwd_cb_userdata(ctx, rk);

	/* Ciphers */
	if (rk->rk_conf.ssl.cipher_suites) {
		rd_kafka_dbg(rk, SECURITY, "SSL",
			     "Setting cipher list: %s",
			     rk->rk_conf.ssl.cipher_suites);
		if (!SSL_CTX_set_cipher_list(ctx,
					     rk->rk_conf.ssl.cipher_suites)) {
                        /* Set a string that will prefix the
                         * the OpenSSL error message (which is lousy)
                         * to make it more meaningful. */
                        rd_snprintf(errstr, errstr_size,
                                    "ssl.cipher.suites failed: ");
                        goto fail;
		}
	}


	if (rk->rk_conf.ssl.ca_location) {
		/* CA certificate location, either file or directory. */
		int is_dir = rd_kafka_path_is_dir(rk->rk_conf.ssl.ca_location);

		rd_kafka_dbg(rk, SECURITY, "SSL",
			     "Loading CA certificate(s) from %s %s",
			     is_dir ? "directory":"file",
			     rk->rk_conf.ssl.ca_location);
		
		r = SSL_CTX_load_verify_locations(ctx,
						  !is_dir ?
						  rk->rk_conf.ssl.
						  ca_location : NULL,
						  is_dir ?
						  rk->rk_conf.ssl.
						  ca_location : NULL);

                if (r != 1) {
                        rd_snprintf(errstr, errstr_size,
                                    "ssl.ca.location failed: ");
                        goto fail;
                }
        } else {
                /* Use default CA certificate paths: ignore failures. */
                r = SSL_CTX_set_default_verify_paths(ctx);
                if (r != 1)
                        rd_kafka_dbg(rk, SECURITY, "SSL",
                                     "SSL_CTX_set_default_verify_paths() "
                                     "failed: ignoring");
        }

	if (rk->rk_conf.ssl.crl_location) {
		rd_kafka_dbg(rk, SECURITY, "SSL",
			     "Loading CRL from file %s",
			     rk->rk_conf.ssl.crl_location);

		r = SSL_CTX_load_verify_locations(ctx,
						  rk->rk_conf.ssl.crl_location,
						  NULL);

                if (r != 1) {
                        rd_snprintf(errstr, errstr_size,
                                    "ssl.crl.location failed: ");
                        goto fail;
                }


		rd_kafka_dbg(rk, SECURITY, "SSL",
			     "Enabling CRL checks");

		X509_STORE_set_flags(SSL_CTX_get_cert_store(ctx),
				     X509_V_FLAG_CRL_CHECK);
	}

	if (rk->rk_conf.ssl.cert_location) {
		rd_kafka_dbg(rk, SECURITY, "SSL",
			     "Loading certificate from file %s",
			     rk->rk_conf.ssl.cert_location);

		r = SSL_CTX_use_certificate_chain_file(ctx,
						       rk->rk_conf.ssl.cert_location);

                if (r != 1) {
                        rd_snprintf(errstr, errstr_size,
                                    "ssl.certificate.location failed: ");
                        goto fail;
                }
	}

	if (rk->rk_conf.ssl.key_location) {
		rd_kafka_dbg(rk, SECURITY, "SSL",
			     "Loading private key file from %s",
			     rk->rk_conf.ssl.key_location);

		r = SSL_CTX_use_PrivateKey_file(ctx,
						rk->rk_conf.ssl.key_location,
						SSL_FILETYPE_PEM);
                if (r != 1) {
                        rd_snprintf(errstr, errstr_size,
                                    "ssl.key.location failed: ");
                        goto fail;
                }
	}

	if (rk->rk_conf.ssl.keystore_location) {
		FILE *fp;
		EVP_PKEY *pkey;
		X509 *cert;
		STACK_OF(X509) *ca = NULL;
		PKCS12 *p12;

		rd_kafka_dbg(rk, SECURITY, "SSL",
			     "Loading client's keystore file from %s",
			     rk->rk_conf.ssl.keystore_location);

		if (!(fp = fopen(rk->rk_conf.ssl.keystore_location, "rb"))) {
			rd_snprintf(errstr, errstr_size,
				    "Failed to open ssl.keystore.location: %s: %s", 
				    rk->rk_conf.ssl.keystore_location, 
				    rd_strerror(errno));
			goto fail;
		}

		p12 = d2i_PKCS12_fp(fp, NULL);
		fclose(fp);
		if (!p12) {
			rd_snprintf(errstr, errstr_size,
				    "Error reading PKCS#12 file: ");
			goto fail;
		}

		pkey = EVP_PKEY_new();
		cert = X509_new();
		if (!PKCS12_parse(p12, rk->rk_conf.ssl.keystore_password, &pkey, &cert, &ca)) {
			EVP_PKEY_free(pkey);
			X509_free(cert);
			PKCS12_free(p12);
			if (ca != NULL)
				sk_X509_pop_free(ca, X509_free);
			rd_snprintf(errstr, errstr_size,
				    "Failed to parse PKCS#12 file: %s: ",
				    rk->rk_conf.ssl.keystore_location);
			goto fail;
		}

		if (ca != NULL)
			sk_X509_pop_free(ca, X509_free);

		PKCS12_free(p12);

		r = SSL_CTX_use_certificate(ctx, cert);
		X509_free(cert);
		if (r != 1) {
			EVP_PKEY_free(pkey);
			rd_snprintf(errstr, errstr_size,
				    "Failed to use ssl.keystore.location certificate: ");
			goto fail;
		}

		r = SSL_CTX_use_PrivateKey(ctx, pkey);
		EVP_PKEY_free(pkey);
		if (r != 1) {
			rd_snprintf(errstr, errstr_size,
				    "Failed to use ssl.keystore.location private key: ");
			goto fail;
		}
	}

	SSL_CTX_set_mode(ctx, SSL_MODE_ENABLE_PARTIAL_WRITE);

	rk->rk_conf.ssl.ctx = ctx;
	return 0;

 fail:
        r = (int)strlen(errstr);
        rd_kafka_ssl_error(rk, NULL, errstr+r,
                           (int)errstr_size > r ? (int)errstr_size - r : 0);
	SSL_CTX_free(ctx);

	return -1;
}


#endif /* WITH_SSL */


ssize_t
rd_kafka_transport_send (rd_kafka_transport_t *rktrans,
                         rd_slice_t *slice, char *errstr, size_t errstr_size) {

#if WITH_SSL
        if (rktrans->rktrans_ssl)
                return rd_kafka_transport_ssl_send(rktrans, slice,
                                                   errstr, errstr_size);
        else
#endif
                return rd_kafka_transport_socket_send(rktrans, slice,
                                                      errstr, errstr_size);
}


ssize_t
rd_kafka_transport_recv (rd_kafka_transport_t *rktrans, rd_buf_t *rbuf,
                         char *errstr, size_t errstr_size) {
#if WITH_SSL
	if (rktrans->rktrans_ssl)
                return rd_kafka_transport_ssl_recv(rktrans, rbuf,
                                                   errstr, errstr_size);
	else
#endif
                return rd_kafka_transport_socket_recv(rktrans, rbuf,
                                                      errstr, errstr_size);
}



/**
 * @brief Notify transport layer of full request sent.
 */
void rd_kafka_transport_request_sent (rd_kafka_broker_t *rkb,
                                      rd_kafka_buf_t *rkbuf) {
        rd_kafka_transport_t *rktrans = rkb->rkb_transport;

        /* Call on_request_sent interceptors */
        rd_kafka_interceptors_on_request_sent(
                rkb->rkb_rk,
                rktrans->rktrans_s,
                rkb->rkb_name, rkb->rkb_nodeid,
                rkbuf->rkbuf_reqhdr.ApiKey,
                rkbuf->rkbuf_reqhdr.ApiVersion,
                rkbuf->rkbuf_corrid,
                rd_slice_size(&rkbuf->rkbuf_reader));
}




/**
 * Length framed receive handling.
 * Currently only supports a the following framing:
 *     [int32_t:big_endian_length_of_payload][payload]
 *
 * To be used on POLLIN event, will return:
 *   -1: on fatal error (errstr will be updated, *rkbufp remains unset)
 *    0: still waiting for data (*rkbufp remains unset)
 *    1: data complete, (buffer returned in *rkbufp)
 */
int rd_kafka_transport_framed_recv (rd_kafka_transport_t *rktrans,
                                    rd_kafka_buf_t **rkbufp,
                                    char *errstr, size_t errstr_size) {
	rd_kafka_buf_t *rkbuf = rktrans->rktrans_recv_buf;
	ssize_t r;
	const int log_decode_errors = LOG_ERR;

	/* States:
	 *   !rktrans_recv_buf: initial state; set up buf to receive header.
	 *    rkbuf_totlen == 0:   awaiting header
	 *    rkbuf_totlen > 0:    awaiting payload
	 */

	if (!rkbuf) {
                rkbuf = rd_kafka_buf_new(1, 4/*length field's length*/);
                /* Set up buffer reader for the length field */
                rd_buf_write_ensure(&rkbuf->rkbuf_buf, 4, 4);
		rktrans->rktrans_recv_buf = rkbuf;
	}


        r = rd_kafka_transport_recv(rktrans, &rkbuf->rkbuf_buf,
                                    errstr, errstr_size);
	if (r == 0)
		return 0;
	else if (r == -1)
		return -1;

	if (rkbuf->rkbuf_totlen == 0) {
		/* Frame length not known yet. */
		int32_t frame_len;

		if (rd_buf_write_pos(&rkbuf->rkbuf_buf) < sizeof(frame_len)) {
			/* Wait for entire frame header. */
			return 0;
		}

                /* Initialize reader */
                rd_slice_init(&rkbuf->rkbuf_reader, &rkbuf->rkbuf_buf, 0, 4);

		/* Reader header: payload length */
		rd_kafka_buf_read_i32(rkbuf, &frame_len);

		if (frame_len < 0 ||
		    frame_len > rktrans->rktrans_rkb->
		    rkb_rk->rk_conf.recv_max_msg_size) {
			rd_snprintf(errstr, errstr_size,
				    "Invalid frame size %"PRId32, frame_len);
			return -1;
		}

		rkbuf->rkbuf_totlen = 4 + frame_len;
		if (frame_len == 0) {
			/* Payload is empty, we're done. */
			rktrans->rktrans_recv_buf = NULL;
			*rkbufp = rkbuf;
			return 1;
		}

		/* Allocate memory to hold entire frame payload in contigious
		 * memory. */
                rd_buf_write_ensure_contig(&rkbuf->rkbuf_buf, frame_len);

                /* Try reading directly, there is probably more data available*/
                return rd_kafka_transport_framed_recv(rktrans, rkbufp,
                                                      errstr, errstr_size);
	}

	if (rd_buf_write_pos(&rkbuf->rkbuf_buf) == rkbuf->rkbuf_totlen) {
		/* Payload is complete. */
		rktrans->rktrans_recv_buf = NULL;
		*rkbufp = rkbuf;
		return 1;
	}

	/* Wait for more data */
	return 0;

 err_parse:
	if (rkbuf)
		rd_kafka_buf_destroy(rkbuf);
        rd_snprintf(errstr, errstr_size, "Frame header parsing failed: %s",
                    rd_kafka_err2str(rkbuf->rkbuf_err));
	return -1;
}


/**
 * TCP connection established.
 * Set up socket options, SSL, etc.
 *
 * Locality: broker thread
 */
static void rd_kafka_transport_connected (rd_kafka_transport_t *rktrans) {
	rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;
        unsigned int slen;

        rd_rkb_dbg(rkb, BROKER, "CONNECT",
                   "Connected to %s",
                   rd_sockaddr2str(rkb->rkb_addr_last,
                                   RD_SOCKADDR2STR_F_PORT |
                                   RD_SOCKADDR2STR_F_FAMILY));

	/* Set socket send & receive buffer sizes if configuerd */
	if (rkb->rkb_rk->rk_conf.socket_sndbuf_size != 0) {
		if (setsockopt(rktrans->rktrans_s, SOL_SOCKET, SO_SNDBUF,
			       (void *)&rkb->rkb_rk->rk_conf.socket_sndbuf_size,
			       sizeof(rkb->rkb_rk->rk_conf.
				      socket_sndbuf_size)) == SOCKET_ERROR)
			rd_rkb_log(rkb, LOG_WARNING, "SNDBUF",
				   "Failed to set socket send "
				   "buffer size to %i: %s",
				   rkb->rkb_rk->rk_conf.socket_sndbuf_size,
				   socket_strerror(socket_errno));
	}

	if (rkb->rkb_rk->rk_conf.socket_rcvbuf_size != 0) {
		if (setsockopt(rktrans->rktrans_s, SOL_SOCKET, SO_RCVBUF,
			       (void *)&rkb->rkb_rk->rk_conf.socket_rcvbuf_size,
			       sizeof(rkb->rkb_rk->rk_conf.
				      socket_rcvbuf_size)) == SOCKET_ERROR)
			rd_rkb_log(rkb, LOG_WARNING, "RCVBUF",
				   "Failed to set socket receive "
				   "buffer size to %i: %s",
				   rkb->rkb_rk->rk_conf.socket_rcvbuf_size,
				   socket_strerror(socket_errno));
	}

        /* Get send and receive buffer sizes to allow limiting
         * the total number of bytes passed with iovecs to sendmsg()
         * and recvmsg(). */
        slen = sizeof(rktrans->rktrans_rcvbuf_size);
        if (getsockopt(rktrans->rktrans_s, SOL_SOCKET, SO_RCVBUF,
                       (void *)&rktrans->rktrans_rcvbuf_size,
                       &slen) == SOCKET_ERROR) {
                rd_rkb_log(rkb, LOG_WARNING, "RCVBUF",
                           "Failed to get socket receive "
                           "buffer size: %s: assuming 1MB",
                           socket_strerror(socket_errno));
                rktrans->rktrans_rcvbuf_size = 1024*1024;
        } else if (rktrans->rktrans_rcvbuf_size < 1024 * 64)
                rktrans->rktrans_rcvbuf_size = 1024*64; /* Use at least 64KB */

        slen = sizeof(rktrans->rktrans_sndbuf_size);
        if (getsockopt(rktrans->rktrans_s, SOL_SOCKET, SO_SNDBUF,
                       (void *)&rktrans->rktrans_sndbuf_size,
                       &slen) == SOCKET_ERROR) {
                rd_rkb_log(rkb, LOG_WARNING, "RCVBUF",
                           "Failed to get socket send "
                           "buffer size: %s: assuming 1MB",
                           socket_strerror(socket_errno));
                rktrans->rktrans_sndbuf_size = 1024*1024;
        } else if (rktrans->rktrans_sndbuf_size < 1024 * 64)
                rktrans->rktrans_sndbuf_size = 1024*64; /* Use at least 64KB */


#ifdef TCP_NODELAY
	if (rkb->rkb_rk->rk_conf.socket_nagle_disable) {
		int one = 1;
		if (setsockopt(rktrans->rktrans_s, IPPROTO_TCP, TCP_NODELAY,
			       (void *)&one, sizeof(one)) == SOCKET_ERROR)
			rd_rkb_log(rkb, LOG_WARNING, "NAGLE",
				   "Failed to disable Nagle (TCP_NODELAY) "
				   "on socket %d: %s",
				   socket_strerror(socket_errno));
	}
#endif


#if WITH_SSL
	if (rkb->rkb_proto == RD_KAFKA_PROTO_SSL ||
	    rkb->rkb_proto == RD_KAFKA_PROTO_SASL_SSL) {
		char errstr[512];

		/* Set up SSL connection.
		 * This is also an asynchronous operation so dont
		 * propagate to broker_connect_done() just yet. */
		if (rd_kafka_transport_ssl_connect(rkb, rktrans,
						   errstr,
						   sizeof(errstr)) == -1) {
			rd_kafka_transport_connect_done(rktrans, errstr);
			return;
		}
		return;
	}
#endif

	/* Propagate connect success */
	rd_kafka_transport_connect_done(rktrans, NULL);
}



/**
 * @brief the kernel SO_ERROR in \p errp for the given transport.
 * @returns 0 if getsockopt() was succesful (and \p and errp can be trusted),
 * else -1 in which case \p errp 's value is undefined.
 */
static int rd_kafka_transport_get_socket_error (rd_kafka_transport_t *rktrans,
						int *errp) {
	socklen_t intlen = sizeof(*errp);

	if (getsockopt(rktrans->rktrans_s, SOL_SOCKET,
		       SO_ERROR, (void *)errp, &intlen) == -1) {
		rd_rkb_dbg(rktrans->rktrans_rkb, BROKER, "SO_ERROR",
			   "Failed to get socket error: %s",
			   socket_strerror(socket_errno));
		return -1;
	}

	return 0;
}


/**
 * IO event handler.
 *
 * Locality: broker thread
 */
static void rd_kafka_transport_io_event (rd_kafka_transport_t *rktrans,
					 int events) {
	char errstr[512];
	int r;
	rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;

	switch (rkb->rkb_state)
	{
	case RD_KAFKA_BROKER_STATE_CONNECT:
#if WITH_SSL
		if (rktrans->rktrans_ssl) {
			/* Currently setting up SSL connection:
			 * perform handshake. */
			rd_kafka_transport_ssl_handshake(rktrans);
			return;
		}
#endif

		/* Asynchronous connect finished, read status. */
		if (!(events & (POLLOUT|POLLERR|POLLHUP)))
			return;

		if (rd_kafka_transport_get_socket_error(rktrans, &r) == -1) {
			rd_kafka_broker_fail(
                                rkb, LOG_ERR, RD_KAFKA_RESP_ERR__TRANSPORT,
                                "Connect to %s failed: "
                                "unable to get status from "
                                "socket %d: %s",
                                rd_sockaddr2str(rkb->rkb_addr_last,
                                                RD_SOCKADDR2STR_F_PORT |
                                                RD_SOCKADDR2STR_F_FAMILY),
                                rktrans->rktrans_s,
                                rd_strerror(socket_errno));
		} else if (r != 0) {
			/* Connect failed */
                        errno = r;
			rd_snprintf(errstr, sizeof(errstr),
				    "Connect to %s failed: %s",
                                    rd_sockaddr2str(rkb->rkb_addr_last,
                                                    RD_SOCKADDR2STR_F_PORT |
                                                    RD_SOCKADDR2STR_F_FAMILY),
                                    rd_strerror(r));

			rd_kafka_transport_connect_done(rktrans, errstr);
		} else {
			/* Connect succeeded */
			rd_kafka_transport_connected(rktrans);
		}
		break;

	case RD_KAFKA_BROKER_STATE_AUTH:
		/* SASL handshake */
		if (rd_kafka_sasl_io_event(rktrans, events,
					   errstr, sizeof(errstr)) == -1) {
			errno = EINVAL;
			rd_kafka_broker_fail(rkb, LOG_ERR,
					     RD_KAFKA_RESP_ERR__AUTHENTICATION,
					     "SASL authentication failure: %s",
					     errstr);
			return;
		}
		break;

	case RD_KAFKA_BROKER_STATE_APIVERSION_QUERY:
	case RD_KAFKA_BROKER_STATE_AUTH_HANDSHAKE:
	case RD_KAFKA_BROKER_STATE_UP:
	case RD_KAFKA_BROKER_STATE_UPDATE:

		if (events & POLLIN) {
			while (rkb->rkb_state >= RD_KAFKA_BROKER_STATE_UP &&
			       rd_kafka_recv(rkb) > 0)
				;
		}

		if (events & POLLHUP) {
			rd_kafka_broker_fail(rkb,
                                             rkb->rkb_rk->rk_conf.
                                             log_connection_close ?
                                             LOG_NOTICE : LOG_DEBUG,
                                             RD_KAFKA_RESP_ERR__TRANSPORT,
					     "Connection closed");
			return;
		}

		if (events & POLLOUT) {
			while (rd_kafka_send(rkb) > 0)
				;
		}
		break;

	case RD_KAFKA_BROKER_STATE_INIT:
	case RD_KAFKA_BROKER_STATE_DOWN:
		rd_kafka_assert(rkb->rkb_rk, !*"bad state");
	}
}


/**
 * Poll and serve IOs
 *
 * Locality: broker thread 
 */
void rd_kafka_transport_io_serve (rd_kafka_transport_t *rktrans,
                                  int timeout_ms) {
	rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;
	int events;

	if (rd_kafka_bufq_cnt(&rkb->rkb_waitresps) < rkb->rkb_max_inflight &&
	    rd_kafka_bufq_cnt(&rkb->rkb_outbufs) > 0)
		rd_kafka_transport_poll_set(rkb->rkb_transport, POLLOUT);

	if ((events = rd_kafka_transport_poll(rktrans, timeout_ms)) <= 0)
                return;

        rd_kafka_transport_poll_clear(rktrans, POLLOUT);

	rd_kafka_transport_io_event(rktrans, events);
}


/**
 * Initiate asynchronous connection attempt.
 *
 * Locality: broker thread
 */
rd_kafka_transport_t *rd_kafka_transport_connect (rd_kafka_broker_t *rkb,
						  const rd_sockaddr_inx_t *sinx,
						  char *errstr,
						  size_t errstr_size) {
	rd_kafka_transport_t *rktrans;
	int s = -1;
	int on = 1;
        int r;

        rkb->rkb_addr_last = sinx;

	s = rkb->rkb_rk->rk_conf.socket_cb(sinx->in.sin_family,
					   SOCK_STREAM, IPPROTO_TCP,
					   rkb->rkb_rk->rk_conf.opaque);
	if (s == -1) {
		rd_snprintf(errstr, errstr_size, "Failed to create socket: %s",
			    socket_strerror(socket_errno));
		return NULL;
	}


#ifdef SO_NOSIGPIPE
	/* Disable SIGPIPE signalling for this socket on OSX */
	if (setsockopt(s, SOL_SOCKET, SO_NOSIGPIPE, &on, sizeof(on)) == -1) 
		rd_rkb_dbg(rkb, BROKER, "SOCKET",
			   "Failed to set SO_NOSIGPIPE: %s",
			   socket_strerror(socket_errno));
#endif

	/* Enable TCP keep-alives, if configured. */
	if (rkb->rkb_rk->rk_conf.socket_keepalive) {
#ifdef SO_KEEPALIVE
		if (setsockopt(s, SOL_SOCKET, SO_KEEPALIVE,
			       (void *)&on, sizeof(on)) == SOCKET_ERROR)
			rd_rkb_dbg(rkb, BROKER, "SOCKET",
				   "Failed to set SO_KEEPALIVE: %s",
				   socket_strerror(socket_errno));
#else
		rd_rkb_dbg(rkb, BROKER, "SOCKET",
			   "System does not support "
			   "socket.keepalive.enable (SO_KEEPALIVE)");
#endif
	}

        /* Set the socket to non-blocking */
        if ((r = rd_fd_set_nonblocking(s))) {
                rd_snprintf(errstr, errstr_size,
                            "Failed to set socket non-blocking: %s",
                            socket_strerror(r));
                goto err;
        }

	rd_rkb_dbg(rkb, BROKER, "CONNECT", "Connecting to %s (%s) "
		   "with socket %i",
		   rd_sockaddr2str(sinx, RD_SOCKADDR2STR_F_FAMILY |
				   RD_SOCKADDR2STR_F_PORT),
		   rd_kafka_secproto_names[rkb->rkb_proto], s);

	/* Connect to broker */
        if (rkb->rkb_rk->rk_conf.connect_cb) {
                r = rkb->rkb_rk->rk_conf.connect_cb(
                        s, (struct sockaddr *)sinx, RD_SOCKADDR_INX_LEN(sinx),
                        rkb->rkb_name, rkb->rkb_rk->rk_conf.opaque);
        } else {
                if (connect(s, (struct sockaddr *)sinx,
                            RD_SOCKADDR_INX_LEN(sinx)) == SOCKET_ERROR &&
                    (socket_errno != EINPROGRESS
#ifdef _MSC_VER
                     && socket_errno != WSAEWOULDBLOCK
#endif
                            ))
                        r = socket_errno;
                else
                        r = 0;
        }

        if (r != 0) {
		rd_rkb_dbg(rkb, BROKER, "CONNECT",
			   "couldn't connect to %s: %s (%i)",
			   rd_sockaddr2str(sinx,
					   RD_SOCKADDR2STR_F_PORT |
					   RD_SOCKADDR2STR_F_FAMILY),
			   socket_strerror(r), r);
		rd_snprintf(errstr, errstr_size,
			    "Failed to connect to broker at %s: %s",
			    rd_sockaddr2str(sinx, RD_SOCKADDR2STR_F_NICE),
			    socket_strerror(r));
		goto err;
	}

	/* Create transport handle */
	rktrans = rd_calloc(1, sizeof(*rktrans));
	rktrans->rktrans_rkb = rkb;
	rktrans->rktrans_s = s;
	rktrans->rktrans_pfd[rktrans->rktrans_pfd_cnt++].fd = s;
        if (rkb->rkb_wakeup_fd[0] != -1) {
                rktrans->rktrans_pfd[rktrans->rktrans_pfd_cnt].events = POLLIN;
                rktrans->rktrans_pfd[rktrans->rktrans_pfd_cnt++].fd = rkb->rkb_wakeup_fd[0];
        }


	/* Poll writability to trigger on connection success/failure. */
	rd_kafka_transport_poll_set(rktrans, POLLOUT);

	return rktrans;

 err:
	if (s != -1)
                rd_kafka_transport_close0(rkb->rkb_rk, s);

	return NULL;
}



void rd_kafka_transport_poll_set(rd_kafka_transport_t *rktrans, int event) {
	rktrans->rktrans_pfd[0].events |= event;
}

void rd_kafka_transport_poll_clear(rd_kafka_transport_t *rktrans, int event) {
	rktrans->rktrans_pfd[0].events &= ~event;
}


int rd_kafka_transport_poll(rd_kafka_transport_t *rktrans, int tmout) {
        int r;
#ifndef _MSC_VER
	r = poll(rktrans->rktrans_pfd, rktrans->rktrans_pfd_cnt, tmout);
	if (r <= 0)
		return r;
#else
	r = WSAPoll(rktrans->rktrans_pfd, rktrans->rktrans_pfd_cnt, tmout);
	if (r == 0) {
		/* Workaround for broken WSAPoll() while connecting:
		 * failed connection attempts are not indicated at all by WSAPoll()
		 * so we need to check the socket error when Poll returns 0.
		 * Issue #525 */
		r = ECONNRESET;
		if (unlikely(rktrans->rktrans_rkb->rkb_state ==
			     RD_KAFKA_BROKER_STATE_CONNECT &&
			     (rd_kafka_transport_get_socket_error(rktrans,
								  &r) == -1 ||
			      r != 0))) {
			char errstr[512];
			errno = r;
			rd_snprintf(errstr, sizeof(errstr),
				    "Connect to %s failed: %s",
				    rd_sockaddr2str(rktrans->rktrans_rkb->
						    rkb_addr_last,
						    RD_SOCKADDR2STR_F_PORT |
                                                    RD_SOCKADDR2STR_F_FAMILY),
                                    socket_strerror(r));
			rd_kafka_transport_connect_done(rktrans, errstr);
			return -1;
		} else
			return 0;
	} else if (r == SOCKET_ERROR)
		return -1;
#endif
        rd_atomic64_add(&rktrans->rktrans_rkb->rkb_c.wakeups, 1);

        if (rktrans->rktrans_pfd[1].revents & POLLIN) {
                /* Read wake-up fd data and throw away, just used for wake-ups*/
                char buf[512];
                if (rd_read((int)rktrans->rktrans_pfd[1].fd,
                            buf, sizeof(buf)) == -1) {
                        /* Ignore warning */
                }
        }

        return rktrans->rktrans_pfd[0].revents;
}





#if 0
/**
 * Global cleanup.
 * This is dangerous and SHOULD NOT be called since it will rip
 * the rug from under the application if it uses any of this functionality
 * in its own code. This means we might leak some memory on exit.
 */
void rd_kafka_transport_term (void) {
#ifdef _MSC_VER
	(void)WSACleanup(); /* FIXME: dangerous */
#endif
}
#endif
 
void rd_kafka_transport_init(void) {
#ifdef _MSC_VER
	WSADATA d;
	(void)WSAStartup(MAKEWORD(2, 2), &d);
#endif
}
