text: /*
   * 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 _WIN32
  #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>
  
  /* AIX doesn't have MSG_DONTWAIT */
  #ifndef MSG_DONTWAIT
  #  define MSG_DONTWAIT MSG_NONBLOCK
  #endif
  
  #if WITH_SSL
  #include "rdkafka_ssl.h"
  #endif
  
  /**< Current thread's rd_kafka_transport_t instance.
   *   This pointer is set up when calling any OpenSSL APIs that might
   *   trigger SSL callbacks, and is used to retrieve the SSL object's
   *   corresponding rd_kafka_transport_t instance.
   *   There is an set/get_ex_data() API in OpenSSL, but it requires storing
   *   a unique index somewhere, which we can't do without having a singleton
   *   object, so instead we cut out the middle man and store the
   *   rd_kafka_transport_t pointer directly in the thread-local memory. */
  RD_TLS rd_kafka_transport_t *rd_kafka_curr_transport;
  
  
  
  /**
   * Low-level socket close
   */
  static void rd_kafka_transport_close0 (rd_kafka_t *rk, rd_socket_t s) {
          if (rk->rk_conf.closesocket_cb)
                  rk->rk_conf.closesocket_cb((int)s, rk->rk_conf.opaque);
          else
                  rd_close(s);
  }
  
  /**
   * Close and destroy a transport handle
   */
  void rd_kafka_transport_close (rd_kafka_transport_t *rktrans) {
  #if WITH_SSL
          rd_kafka_curr_transport = rktrans;
          if (rktrans->rktrans_ssl)
                  rd_kafka_transport_ssl_close(rktrans);
  #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);
  }
  
  
  #ifndef _WIN32
  /**
   * @brief sendmsg() 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_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;
          size_t r2;
  
          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. */
          rd_socket_errno = EAGAIN;
  #endif
  
          r = sendmsg(rktrans->rktrans_s, &msg, MSG_DONTWAIT
  #ifdef MSG_NOSIGNAL
                      | MSG_NOSIGNAL
  #endif
                  );
  
          if (r == -1) {
                  if (rd_socket_errno == EAGAIN)
                          return 0;
                  rd_snprintf(errstr, errstr_size, "%s", rd_strerror(errno));
                  return -1;
          }
  
          /* Update buffer read position */
          r2 = rd_slice_read(slice, NULL, (size_t)r);
          rd_assert((size_t)r == r2 &&
                    *"BUG: wrote more bytes than available in slice");
  
          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;
                  size_t r2;
  
                  r = send(rktrans->rktrans_s, p,
  #ifdef _WIN32
                           (int)rlen, (int)0
  #else
                           rlen, 0
  #endif
                  );
  
  #ifdef _WIN32
                  if (unlikely(r == RD_SOCKET_ERROR)) {
                          if (sum > 0 || rd_socket_errno == WSAEWOULDBLOCK)
                                  return sum;
                          else {
                                  rd_snprintf(errstr, errstr_size, "%s",
                                              rd_socket_strerror(rd_socket_errno));
                                  return -1;
                          }
                  }
  #else
                  if (unlikely(r <= 0)) {
                          if (r == 0 || rd_socket_errno == EAGAIN)
                                  return 0;
                          rd_snprintf(errstr, errstr_size, "%s",
                                      rd_socket_strerror(rd_socket_errno));
                          return -1;
                  }
  #endif
  
                  /* Update buffer read position */
                  r2 = rd_slice_read(slice, NULL, (size_t)r);
                  rd_assert((size_t)r == r2 &&
                            *"BUG: wrote more bytes than available in slice");
  
  
                  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 _WIN32
          /* 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 _WIN32
  /**
   * @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. */
          rd_socket_errno = EAGAIN;
  #endif
          r = recvmsg(rktrans->rktrans_s, &msg, MSG_DONTWAIT);
          if (unlikely(r <= 0)) {
                  if (r == -1 && rd_socket_errno == EAGAIN)
                          return 0;
                  else if (r == 0 ||
                           (r == -1 && rd_socket_errno == ECONNRESET)) {
                          /* 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 _WIN32
                           (int)
  #endif
                           len,
                           0);
  
                  if (unlikely(r == RD_SOCKET_ERROR)) {
                          if (rd_socket_errno == EAGAIN
  #ifdef _WIN32
                              || rd_socket_errno == WSAEWOULDBLOCK
  #endif
                                  )
                                  return sum;
                          else {
                                  rd_snprintf(errstr, errstr_size, "%s",
                                              rd_socket_strerror(rd_socket_errno));
                                  return -1;
                          }
                  } else if (unlikely(r == 0)) {
                          /* Receive 0 after POLLIN event means
                           * connection closed. */
                          rd_snprintf(errstr, errstr_size,
                                      "Disconnected");
                          return -1;
                  }
  
                  /* 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 _WIN32
          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_curr_transport = rktrans;
  
          rd_kafka_broker_connect_done(rkb, errstr);
  }
  
  
  
  
  
  
  ssize_t
  rd_kafka_transport_send (rd_kafka_transport_t *rktrans,
                           rd_slice_t *slice, char *errstr, size_t errstr_size) {
          ssize_t r;
  #if WITH_SSL
          if (rktrans->rktrans_ssl) {
                  rd_kafka_curr_transport = rktrans;
                  r = rd_kafka_transport_ssl_send(rktrans, slice,
                                                  errstr, errstr_size);
          } else
  #endif
                  r = rd_kafka_transport_socket_send(rktrans, slice,
                                                     errstr, errstr_size);
  
          return r;
  }
  
  
  ssize_t
  rd_kafka_transport_recv (rd_kafka_transport_t *rktrans, rd_buf_t *rbuf,
                           char *errstr, size_t errstr_size) {
          ssize_t r;
  
  #if WITH_SSL
          if (rktrans->rktrans_ssl) {
                  rd_kafka_curr_transport = rktrans;
                  r = rd_kafka_transport_ssl_recv(rktrans, rbuf,
                                                  errstr, errstr_size);
          } else
  #endif
                  r = rd_kafka_transport_socket_recv(rktrans, rbuf,
                                                     errstr, errstr_size);
  
          return r;
  }
  
  
  
  /**
   * @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,
                  (int)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:
          rd_snprintf(errstr, errstr_size, "Frame header parsing failed: %s",
                      rd_kafka_err2str(rkbuf->rkbuf_err));
  	return -1;
  }
  
  
  /**
   * @brief Final socket setup after a connection has been established
   */
  void rd_kafka_transport_post_connect_setup (rd_kafka_transport_t *rktrans) {
          rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;
          unsigned int slen;
  
          /* 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)) == RD_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,
                                     rd_socket_strerror(rd_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)) == RD_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,
                                     rd_socket_strerror(rd_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) == RD_SOCKET_ERROR) {
                  rd_rkb_log(rkb, LOG_WARNING, "RCVBUF",
                             "Failed to get socket receive "
                             "buffer size: %s: assuming 1MB",
                             rd_socket_strerror(rd_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) == RD_SOCKET_ERROR) {
                  rd_rkb_log(rkb, LOG_WARNING, "RCVBUF",
                             "Failed to get socket send "
                             "buffer size: %s: assuming 1MB",
                             rd_socket_strerror(rd_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)) == RD_SOCKET_ERROR)
                          rd_rkb_log(rkb, LOG_WARNING, "NAGLE",
                                     "Failed to disable Nagle (TCP_NODELAY) "
                                     "on socket: %s",
                                     rd_socket_strerror(rd_socket_errno));
          }
  #endif
  }
  
  
  /**
   * 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;
  
          rd_rkb_dbg(rkb, BROKER, "CONNECT",
                     "Connected to %s",
                     rd_sockaddr2str(rkb->rkb_addr_last,
                                     RD_SOCKADDR2STR_F_PORT |
                                     RD_SOCKADDR2STR_F_FAMILY));
  
          rd_kafka_transport_post_connect_setup(rktrans);
  
  #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",
  			   rd_socket_strerror(rd_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(rd_socket_errno));
  		} else if (r != 0) {
  			/* Connect failed */
  			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_LEGACY:
                  /* SASL authentication.
                   * Prior to broker version v1.0.0 this is performed
                   * directly on the socket without Kafka framing. */
                  if (rd_kafka_sasl_io_event(rktrans, events,
                                             errstr,
                                             sizeof(errstr)) == -1) {
                          rd_kafka_broker_fail(
                                  rkb, LOG_ERR,
                                  RD_KAFKA_RESP_ERR__AUTHENTICATION,
                                  "SASL authentication failure: %s",
                                  errstr);
                          return;
                  }
  
                  if (events & POLLHUP) {
                          rd_kafka_broker_fail(
                                  rkb, LOG_ERR,
                                  RD_KAFKA_RESP_ERR__AUTHENTICATION,
                                  "Disconnected");
  
                          return;
                  }
  
                  break;
  
  	case RD_KAFKA_BROKER_STATE_APIVERSION_QUERY:
  	case RD_KAFKA_BROKER_STATE_AUTH_HANDSHAKE:
                  case RD_KAFKA_BROKER_STATE_AUTH_REQ:
  	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 connection went down: bail out early */
                          if (rkb->rkb_state == RD_KAFKA_BROKER_STATE_DOWN)
                                  return;
  		}
  
                  if (events & POLLHUP) {
                          rd_kafka_broker_conn_closed(
                                  rkb, RD_KAFKA_RESP_ERR__TRANSPORT,
                                  "Disconnected");
                          return;
                  }
  
  		if (events & POLLOUT) {
  			while (rd_kafka_send(rkb) > 0)
  				;
  		}
  		break;
  
  	case RD_KAFKA_BROKER_STATE_INIT:
  	case RD_KAFKA_BROKER_STATE_DOWN:
          case RD_KAFKA_BROKER_STATE_TRY_CONNECT:
  		rd_kafka_assert(rkb->rkb_rk, !*"bad state");
  	}
  }
  
  
  /**
   * @brief Poll and serve IOs
   *
   * @returns 1 if at least one IO event was triggered, else 0, or -1 on error.
   *
   * @locality broker thread
   */
  int rd_kafka_transport_io_serve (rd_kafka_transport_t *rktrans,
                                    int timeout_ms) {
  	rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;
          int events;
          int r;
  
          rd_kafka_curr_transport = rktrans;
  
          if (rkb->rkb_state == RD_KAFKA_BROKER_STATE_CONNECT ||
              (rkb->rkb_state > RD_KAFKA_BROKER_STATE_CONNECT &&
               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 ((r = rd_kafka_transport_poll(rktrans, timeout_ms)) <= 0)
                  return r;
  
          /* Only handle events on the broker socket, the wakeup
           * socket is just for waking up the blocking boll. */
          events = rktrans->rktrans_pfd[0].revents;
          if (events) {
                  rd_kafka_transport_poll_clear(rktrans, POLLOUT);
  
                  rd_kafka_transport_io_event(rktrans, events);
          }
  
          return 1;
  }
  
  
  /**
   * @brief Create a new transport object using existing socket \p s.
   */
  rd_kafka_transport_t *rd_kafka_transport_new (rd_kafka_broker_t *rkb,
                                                rd_socket_t s,
                                                char *errstr,
                                                size_t errstr_size) {
          rd_kafka_transport_t *rktrans;
          int on = 1;
          int r;
  
  #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",
                             rd_socket_strerror(rd_socket_errno));
  #endif
  
  #ifdef SO_KEEPALIVE
          /* Enable TCP keep-alives, if configured. */
          if (rkb->rkb_rk->rk_conf.socket_keepalive) {
                  if (setsockopt(s, SOL_SOCKET, SO_KEEPALIVE,
                                 (void *)&on, sizeof(on)) == RD_SOCKET_ERROR)
                          rd_rkb_dbg(rkb, BROKER, "SOCKET",
                                     "Failed to set SO_KEEPALIVE: %s",
                                     rd_socket_strerror(rd_socket_errno));
          }
  #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",
                              rd_socket_strerror(r));
                  return NULL;
          }
  
  
          rktrans = rd_calloc(1, sizeof(*rktrans));
          rktrans->rktrans_rkb = rkb;
          rktrans->rktrans_s = s;
  
          return rktrans;
  }
  
  
  /**
   * 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 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",
  			    rd_socket_strerror(rd_socket_errno));
  		return NULL;
  	}
  
          rktrans = rd_kafka_transport_new(rkb, s, errstr, errstr_size);
          if (!rktrans) {
                  rd_kafka_transport_close0(rkb->rkb_rk, s);
                  return NULL;
          }
  
  	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) {
                  rd_kafka_broker_lock(rkb); /* for rkb_nodename */
                  r = rkb->rkb_rk->rk_conf.connect_cb(
                          s, (struct sockaddr *)sinx, RD_SOCKADDR_INX_LEN(sinx),
                          rkb->rkb_nodename, rkb->rkb_rk->rk_conf.opaque);
                  rd_kafka_broker_unlock(rkb);
          } else {
                  if (connect(s, (struct sockaddr *)sinx,
                              RD_SOCKADDR_INX_LEN(sinx)) == RD_SOCKET_ERROR &&
                      (rd_socket_errno != EINPROGRESS
  #ifdef _WIN32
                       && rd_socket_errno != WSAEWOULDBLOCK
  #endif
                              ))
                          r = rd_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),
  			   rd_socket_strerror(r), r);
  		rd_snprintf(errstr, errstr_size,
  			    "Failed to connect to broker at %s: %s",
  			    rd_sockaddr2str(sinx, RD_SOCKADDR2STR_F_NICE),
  			    rd_socket_strerror(r));
  
                  rd_kafka_transport_close(rktrans);
                  return NULL;
  	}
  
          /* Set up transport handle */
          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;
  }
  
  
  
  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;
  }
  
  /**
   * @brief Poll transport fds.
   *
   * @returns 1 if an event was raised, else 0, or -1 on error.
   */
  int rd_kafka_transport_poll(rd_kafka_transport_t *rktrans, int tmout) {
          int r;
  
  #ifndef _WIN32
  	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];
  			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),
                                      rd_socket_strerror(r));
  			rd_kafka_transport_connect_done(rktrans, errstr);
  			return -1;
  		} else
  			return 0;
  	} else if (r == RD_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[1024];
                  while (rd_read((int)rktrans->rktrans_pfd[1].fd,
                                 buf, sizeof(buf)) > 0)
                          ; /* Read all buffered signalling bytes */
          }
  
          return 1;
  }
  
  
  
  
  
  #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 _WIN32
  	(void)WSACleanup(); /* FIXME: dangerous */
  #endif
  }
  #endif
  
  void rd_kafka_transport_init (void) {
  #ifdef _WIN32
  	WSADATA d;
  	(void)WSAStartup(MAKEWORD(2, 2), &d);
  #endif
  }
  