static void send_later_cb(gpointer data, gint source, const gchar * error)
{
	PurpleConnection *gc = data;
	struct fetion_account_data *sip;
	struct sip_connection *conn;

	if (!PURPLE_CONNECTION_IS_VALID(gc)) {
		if (source >= 0)
			close(source);
		return;
	}

	if (source < 0) {
		purple_connection_error_reason(gc,
					       PURPLE_CONNECTION_ERROR_NETWORK_ERROR,
					       _("Could not connect"));
		return;
	}

	sip = gc->proto_data;
	sip->fd = source;
	sip->connecting = FALSE;

	fetion_canwrite_cb(gc, sip->fd, PURPLE_INPUT_WRITE);

	/* If there is more to write now, we need to register a handler */
	if (sip->txbuf->bufused > 0)
		sip->tx_handler = purple_input_add(sip->fd, PURPLE_INPUT_WRITE,
						   fetion_canwrite_cb, gc);

	conn = connection_create(sip, source);
	conn->inputhandler =
	    purple_input_add(sip->fd, PURPLE_INPUT_READ, fetion_input_cb, gc);
}

static void
flap_connection_send_byte_stream(ByteStream *bs, FlapConnection *conn, size_t count)
{
	if (conn == NULL)
		return;

	/* Make sure we don't send past the end of the bs */
	if (count > byte_stream_bytes_left(bs))
		count = byte_stream_bytes_left(bs); /* truncate to remaining space */

	if (count == 0)
		return;

	/* Add everything to our outgoing buffer */
	purple_circ_buffer_append(conn->buffer_outgoing, bs->data, count);

	/* If we haven't already started writing stuff, then start the cycle */
	if (conn->watcher_outgoing == 0)
	{
		if (conn->gsc) {
			conn->watcher_outgoing = purple_input_add(conn->gsc->fd,
					PURPLE_INPUT_WRITE, send_cb, conn);
			send_cb(conn, -1, 0);
		} else if (conn->fd >= 0) {
			conn->watcher_outgoing = purple_input_add(conn->fd,
					PURPLE_INPUT_WRITE, send_cb, conn);
			send_cb(conn, -1, 0);
		}
	}
}

/* the callback function after socket is built
 * we setup the qq protocol related configuration here */
static void connect_cb(gpointer data, gint source, const gchar *error_message)
{
	PurpleConnection *gc;
	qq_data *qd;
	qq_connection *conn;

	gc = (PurpleConnection *) data;
	g_return_if_fail(gc != NULL && gc->proto_data != NULL);

	qd = (qq_data *) gc->proto_data;

	/* conn_data will be destoryed */
	qd->conn_data = NULL;

	if (!PURPLE_CONNECTION_IS_VALID(gc)) {
		purple_debug_info("QQ_CONN", "Invalid connection\n");
		close(source);
		return;
	}

	if (source < 0) {	/* socket returns -1 */
		purple_debug_info("QQ_CONN",
				"Could not establish a connection with the server:\n%s\n",
				error_message);
		if (qd->connect_watcher > 0)	purple_timeout_remove(qd->connect_watcher);
		qd->connect_watcher = purple_timeout_add_seconds(QQ_CONNECT_INTERVAL, qq_connect_later, gc);
		return;
	}

	/* _qq_show_socket("Got login socket", source); */
	/* ok, already connected to the server */
	qd->fd = source;
	conn = connection_create(qd, source);
	g_return_if_fail( conn != NULL );

	if (qd->use_tcp) {
		/* events which match "PURPLE_INPUT_READ" of
		 * "source" would trigger the callback function */
		conn->input_handler = purple_input_add(source, PURPLE_INPUT_READ, tcp_pending, gc);
	} else {
		conn->input_handler = purple_input_add(source, PURPLE_INPUT_READ, udp_pending, gc);
	}

	g_return_if_fail(qd->network_watcher == 0);
	qd->network_watcher = purple_timeout_add_seconds(qd->itv_config.resend, network_timeout, gc);

	set_all_keys( gc );

	if (qd->client_version >= 2007) {
		purple_connection_update_progress(gc, _("Getting server"), 2, QQ_CONNECT_STEPS);
		/* touch required */
		qq_request_get_server(gc);
		return;
	}

	purple_connection_update_progress(gc, _("Requesting token"), 2, QQ_CONNECT_STEPS);
	qq_request_token(gc);
}

static void
msn_dc_init(MsnDirectConn *dc)
{
	g_return_if_fail(dc != NULL);

	dc->in_size = DC_MAX_PACKET_SIZE + 4;
	dc->in_pos = 0;
	dc->in_buffer = g_malloc(dc->in_size);

	dc->recv_handle = purple_input_add(dc->fd, PURPLE_INPUT_READ, msn_dc_recv_cb, dc);
	dc->send_handle = purple_input_add(dc->fd, PURPLE_INPUT_WRITE, msn_dc_send_cb, dc);

	dc->timeout_handle = purple_timeout_add_seconds(DC_TIMEOUT, msn_dc_timeout, dc);
}

void
fb_mqtt_write(FbMqtt *mqtt, FbMqttMessage *msg)
{
	const GByteArray *bytes;
	FbMqttMessagePrivate *mriv;
	FbMqttPrivate *priv;

	g_return_if_fail(FB_IS_MQTT(mqtt));
	g_return_if_fail(FB_IS_MQTT_MESSAGE(msg));
	priv = mqtt->priv;
	mriv = msg->priv;

	bytes = fb_mqtt_message_bytes(msg);

	if (G_UNLIKELY(bytes == NULL)) {
		fb_mqtt_error(mqtt, FB_MQTT_ERROR_GENERAL,
		              _("Failed to format data"));
		return;
	}

	fb_util_debug_hexdump(FB_UTIL_DEBUG_INFO, mriv->bytes,
	                      "Writing %d (flags: 0x%0X)",
		              mriv->type, mriv->flags);

	g_byte_array_append(priv->wbuf, bytes->data, bytes->len);
	fb_mqtt_cb_write(mqtt, priv->gsc->fd, PURPLE_INPUT_WRITE);

	if (priv->wev > 0) {
		priv->wev = purple_input_add(priv->gsc->fd,
		                             PURPLE_INPUT_WRITE,
		                             fb_mqtt_cb_write, mqtt);
	}
}

void
purple_xfer_ui_ready(PurpleXfer *xfer)
{
	PurpleInputCondition cond;
	PurpleXferType type;
	PurpleXferPrivData *priv;

	g_return_if_fail(xfer != NULL);

	priv = g_hash_table_lookup(xfers_data, xfer);
	priv->ready |= PURPLE_XFER_READY_UI;

	if (0 == (priv->ready & PURPLE_XFER_READY_PRPL)) {
		purple_debug_misc("xfer", "UI is ready on ft %p, waiting for prpl\n", xfer);
		return;
	}

	purple_debug_misc("xfer", "UI (and prpl) ready on ft %p, so proceeding\n", xfer);

	type = purple_xfer_get_type(xfer);
	if (type == PURPLE_XFER_SEND)
		cond = PURPLE_INPUT_WRITE;
	else /* if (type == PURPLE_XFER_RECEIVE) */
		cond = PURPLE_INPUT_READ;

	if (xfer->watcher == 0 && xfer->fd != -1)
		xfer->watcher = purple_input_add(xfer->fd, cond, transfer_cb, xfer);

	priv->ready = PURPLE_XFER_READY_NONE;

	do_transfer(xfer);
}

static void
http_connection_send_request(PurpleHTTPConnection *conn, const GString *req)
{
	char *data;
	int ret;
	size_t len;

	/* Sending something to the server, restart the inactivity timer */
	jabber_stream_restart_inactivity_timer(conn->bosh->js);

	data = g_strdup_printf("POST %s HTTP/1.1\r\n"
	                       "Host: %s\r\n"
	                       "User-Agent: %s\r\n"
	                       "Content-Encoding: text/xml; charset=utf-8\r\n"
	                       "Content-Length: %" G_GSIZE_FORMAT "\r\n\r\n"
	                       "%s",
	                       conn->bosh->path, conn->bosh->host, bosh_useragent,
	                       req->len, req->str);

	len = strlen(data);

	++conn->requests;
	++conn->bosh->requests;

	if (purple_debug_is_unsafe() && purple_debug_is_verbose())
		/* Will contain passwords for SASL PLAIN and is verbose */
		purple_debug_misc("jabber", "BOSH (%p): Sending %s\n", conn, data);
	else if (purple_debug_is_verbose())
		purple_debug_misc("jabber", "BOSH (%p): Sending request of "
		                            "%" G_GSIZE_FORMAT " bytes.\n", conn, len);

	if (conn->writeh == 0)
		ret = http_connection_do_send(conn, data, len);
	else {
		ret = -1;
		errno = EAGAIN;
	}

	if (ret < 0 && errno != EAGAIN) {
		/*
		 * TODO: Handle this better. Probably requires a PurpleBOSHConnection
		 * buffer that stores what is "being sent" until the
		 * PurpleHTTPConnection reports it is fully sent.
		 */
		gchar *tmp = g_strdup_printf(_("Lost connection with server: %s"),
				g_strerror(errno));
		purple_connection_error_reason(conn->bosh->js->gc,
				PURPLE_CONNECTION_ERROR_NETWORK_ERROR,
				tmp);
		g_free(tmp);
		return;
	} else if (ret < len) {
		if (ret < 0)
			ret = 0;
		if (conn->writeh == 0)
			conn->writeh = purple_input_add(conn->psc ? conn->psc->fd : conn->fd,
					PURPLE_INPUT_WRITE, http_connection_send_cb, conn);
		purple_circ_buffer_append(conn->write_buf, data + ret, len - ret);
	}
}

static void ycht_got_connected(gpointer data, gint source, const gchar *error_message)
{
	YchtConn *ycht = data;
	PurpleConnection *gc = ycht->gc;
	YahooData *yd = purple_connection_get_protocol_data(gc);
	YchtPkt *pkt;
	char *buf;

	if (source < 0) {
		ycht_connection_error(ycht, _("Unable to connect"));
		return;
	}

	ycht->fd = source;

	pkt = ycht_packet_new(YCHT_VERSION, YCHT_SERVICE_LOGIN, 0);

	buf = g_strdup_printf("%s\001Y=%s; T=%s", purple_connection_get_display_name(gc), yd->cookie_y, yd->cookie_t);
	ycht_packet_append(pkt, buf);
	g_free(buf);

	ycht_packet_send(ycht, pkt);

	ycht_packet_free(pkt);

	ycht->inpa = purple_input_add(ycht->fd, PURPLE_INPUT_READ, ycht_pending, ycht);
}

void
peer_connection_finalize_connection(PeerConnection *conn)
{
	conn->watcher_incoming = purple_input_add(conn->fd,
			PURPLE_INPUT_READ, peer_connection_recv_cb, conn);

	if (conn->type == OSCAR_CAPABILITY_DIRECTIM)
	{
		/*
		 * If we are connecting to them then send our cookie so they
		 * can verify who we are.  Note: This doesn't seem to be
		 * necessary, but it also doesn't seem to hurt.
		 */
		if (!(conn->flags & PEER_CONNECTION_FLAG_IS_INCOMING))
			peer_odc_send_cookie(conn);
	}
	else if (conn->type == OSCAR_CAPABILITY_SENDFILE)
	{
		if (purple_xfer_get_type(conn->xfer) == PURPLE_XFER_SEND)
		{
			peer_oft_send_prompt(conn);
		}
	}

	/*
	 * Tell the remote user that we're connected (which may also imply
	 * that we've accepted their request).
	 */
	if (!(conn->flags & PEER_CONNECTION_FLAG_IS_INCOMING))
		aim_im_sendch2_connected(conn);
}

void mrim_send_xfer_connect_cb(gpointer data, gint source, const gchar *error_message) {
	purple_debug_info("mrim-prpl", "[%s]\n", __func__);
	MrimFT *ft = data;
	ft->proxy_conn = NULL;
	if (source >= 0) {
		purple_debug_info("mrim-prpl", "[%s] Connected!\n", __func__);
		ft->conn = source;
		ft->state = WAITING_FOR_HELLO_ACK;
		MrimData *mrim  = ft->mrim;
		MrimData *fake_mrim = g_new0(MrimData, 1);
		fake_mrim->fd = source;
		ft->fake_mrim = fake_mrim;
		MrimPackage *pack = mrim_package_new(0, MRIM_CS_PROXY_HELLO);
		pack->header->proto = 0x00010009;
		mrim_package_add_UL(pack, ft->proxy_id[0]);
		mrim_package_add_UL(pack, ft->proxy_id[1]);
		mrim_package_add_UL(pack, ft->proxy_id[2]);
		mrim_package_add_UL(pack, ft->proxy_id[3]);

		if (mrim_package_send(pack, fake_mrim)) {
			ft->inpa = purple_input_add(ft->conn, PURPLE_INPUT_READ, mrim_ft_send_input_cb, ft);
			purple_debug_info("mrim-prpl", "[%s] MRIM_CS_PROXY_HELLO sent!\n", __func__);
		} else {
			purple_debug_info("mrim-prpl", "[%s] Failed to send MRIM_CS_PROXY_HELLO!\n", __func__);
			purple_xfer_unref(ft->xfer);
		}
	} else {
		purple_debug_info("mrim-prpl", "[%s] Fail!\n", __func__);
		purple_xfer_unref(ft->xfer);
	}
}

static void
ggp_tcpsocket_connected(PurpleSocket *ps, const gchar *error, gpointer priv_gg)
{
	PurpleConnection *gc = purple_socket_get_connection(ps);
	GGPInfo *info = purple_connection_get_protocol_data(gc);
	int fd = -1;

	PURPLE_ASSERT_CONNECTION_IS_VALID(gc);

	if (error == NULL)
		fd = purple_socket_get_fd(ps);

	if (!gg_socket_manager_connected(ps, priv_gg, fd)) {
		purple_debug_error("gg", "socket not handled");
		purple_socket_destroy(ps);
	}

	if (info->inpa > 0)
		purple_input_remove(info->inpa);
	if (info->session->fd < 0)
		return;
	info->inpa = purple_input_add(info->session->fd,
		ggp_tcpsocket_inputcond_gg_to_purple(info->session->check),
		ggp_async_login_handler, gc);
}

static void yahoo_receivefile_connected(gpointer data, gint source, const gchar *error_message)
{
	PurpleXfer *xfer;
	struct yahoo_xfer_data *xd;

	purple_debug(PURPLE_DEBUG_INFO, "yahoo",
			   "AAA - in yahoo_receivefile_connected\n");
	if (!(xfer = data))
		return;
	if (!(xd = xfer->data))
		return;
	if ((source < 0) || (xd->path == NULL) || (xd->host == NULL)) {
		purple_xfer_error(PURPLE_XFER_RECEIVE, purple_xfer_get_account(xfer),
				xfer->who, _("Unable to connect."));
		purple_xfer_cancel_remote(xfer);
		return;
	}

	xfer->fd = source;

	/* The first time we get here, assemble the tx buffer */
	if (xd->txbuflen == 0) {
		xd->txbuf = g_strdup_printf("GET /%s HTTP/1.0\r\nHost: %s\r\n\r\n",
			      xd->path, xd->host);
		xd->txbuflen = strlen(xd->txbuf);
		xd->txbuf_written = 0;
	}

	if (!xd->tx_handler)
	{
		xd->tx_handler = purple_input_add(source, PURPLE_INPUT_WRITE,
			yahoo_receivefile_send_cb, xfer);
		yahoo_receivefile_send_cb(xfer, source, PURPLE_INPUT_WRITE);
	}
}

static void
connect_cb(gpointer data, gint source, const gchar *error_message)
{
	MsnServConn *servconn;

	servconn = data;
	servconn->connect_data = NULL;
	servconn->processing = FALSE;

	if (servconn->wasted)
	{
		if (source >= 0)
			close(source);
		msn_servconn_destroy(servconn);
		return;
	}

	servconn->fd = source;

	if (source >= 0)
	{
		servconn->connected = TRUE;

		/* Someone wants to know we connected. */
		servconn->connect_cb(servconn);
		servconn->inpa = purple_input_add(servconn->fd, PURPLE_INPUT_READ,
			read_cb, data);
	}
	else
	{
		purple_debug_error("msn", "Connection error: %s\n", error_message);
		msn_servconn_got_error(servconn, MSN_SERVCONN_ERROR_CONNECT);
	}
}

static void
nexus_connect_cb(gpointer data, PurpleSslConnection *gsc,
				 PurpleInputCondition cond)
{
	MsnNexus *nexus;
	MsnSession *session;

	nexus = data;
	g_return_if_fail(nexus != NULL);

	session = nexus->session;
	g_return_if_fail(session != NULL);

	msn_session_set_login_step(session, PECAN_LOGIN_STEP_AUTH);

	nexus->write_buf = g_strdup("GET /rdr/pprdr.asp\r\n\r\n");
	nexus->written_len = 0;

	nexus->read_len = 0;

	nexus->written_cb = nexus_connect_written_cb;

	nexus->input_handler = purple_input_add(gsc->fd, PURPLE_INPUT_WRITE,
		nexus_write_cb, nexus);

	nexus_write_cb(nexus, gsc->fd, PURPLE_INPUT_WRITE);
}

static void okc_post_or_get_connect_cb(gpointer data, gint source,
		const gchar *error_message)
{
	OkCupidConnection *okconn;
	ssize_t len;

	okconn = data;
	okconn->connect_data = NULL;

	if (error_message)
	{
		purple_debug_error("okcupid", "post_or_get_connect_cb %s\n",
				error_message);
		okc_fatal_connection_cb(okconn);
		return;
	}

	purple_debug_info("okcupid", "post_or_get_connect_cb\n");
	okconn->fd = source;

	/* TODO: Check the return value of write() */
	len = write(okconn->fd, okconn->request->str,
			okconn->request->len);
	okconn->input_watcher = purple_input_add(okconn->fd,
			PURPLE_INPUT_READ,
			okc_post_or_get_readdata_cb, okconn);
}

static void irc_dccsend_send_connected(gpointer data, int source, PurpleInputCondition cond) {
    PurpleXfer *xfer = (PurpleXfer *) data;
    struct irc_xfer_send_data *xd = purple_xfer_get_protocol_data(xfer);
    int conn;

    conn = accept(xd->fd, NULL, 0);
    if (conn == -1) {
        /* Accepting the connection failed. This could just be related
         * to the nonblocking nature of the listening socket, so we'll
         * just try again next time */
        /* Let's print an error message anyway */
        purple_debug_warning("irc", "accept: %s\n", g_strerror(errno));
        return;
    }

    purple_input_remove(purple_xfer_get_watcher(xfer));
    purple_xfer_set_watcher(xfer, 0);
    close(xd->fd);
    xd->fd = -1;

    _purple_network_set_common_socket_flags(conn);

    xd->inpa = purple_input_add(conn, PURPLE_INPUT_READ, irc_dccsend_send_read, xfer);
    /* Start the transfer */
    purple_xfer_start(xfer, conn, NULL, 0);
}

ConfigInterface::ConfigInterface(const std::string &sockfile, const LogSink &logInstance) : ConnectionTCPServer(this, logInstance, "localhost", -1) {
	// Create UNIX socket
	int m_socket;
	socklen_t length;
	struct sockaddr_un local;
	m_loaded = false;
	m_socketId = 0;
	
	m_admin = new AdhocAdmin();
	registerHandler(m_admin);
	
	if ((m_socket = getUnixSocket()) == -1)
		Log("ConfigInterface", "Could not create UNIX socket: " << strerror(errno));
	
	local.sun_family = AF_UNIX;
	
	strcpy(local.sun_path, sockfile.c_str());
	unlink(local.sun_path);
	
	length = offsetof(struct sockaddr_un, sun_path) + strlen(sockfile.c_str());

	if (bind(m_socket, (struct sockaddr *) &local, length) == -1) {
		Log("ConfigInterface", "Could not bind to UNIX socket: " << sockfile << " " << strerror(errno));
		return;
	}
	
	if (listen(m_socket, 5) == -1) {
		Log("ConfigInterface", "Could not listen to UNIX socket: " << sockfile << " " << strerror(errno));
		return;
	}

	setSocket(m_socket);
	m_socketId = purple_input_add(m_socket, PURPLE_INPUT_READ, gotData, this);
	m_loaded = true;
}

/*------------------------------------------------------------------------
 * Callback invoked once the connection has been established to the HTTP server,
 * or on connection failure.
 *
 *  @param user_data		The MXit session object
 *  @param source			The file-descriptor associated with the connection
 *  @param error_message	Message explaining why the connection failed
 */
static void mxit_cb_http_connect( gpointer user_data, gint source, const gchar* error_message )
{
	struct http_request*	req	= (struct http_request*) user_data;

	purple_debug_info( MXIT_PLUGIN_ID, "mxit_cb_http_connect\n" );

	/* source is the file descriptor of the new connection */
	if ( source < 0 ) {
		purple_debug_info( MXIT_PLUGIN_ID, "mxit_cb_http_connect failed: %s\n", error_message );
		purple_connection_error( req->session->con, _( "Unable to connect to the MXit HTTP server. Please check your server settings." ) );
		return;
	}

	/* we now have an open and active TCP connection to the mxit server */
	req->session->fd = source;

	/* reset the receive buffer */
	req->session->rx_state = RX_STATE_RLEN;
	req->session->rx_lbuf[0] = '\0';
	req->session->rx_i = 0;
	req->session->rx_res = 0;

	/* start listening on the open connection for messages from the server (reference: "libpurple/eventloop.h") */
	req->session->http_handler = purple_input_add( req->session->fd, PURPLE_INPUT_READ, mxit_cb_http_read, req->session );

	/* actually send the request to the HTTP server */
	mxit_http_raw_write( req->session->fd, req->data, req->datalen );

	/* free up resources */
	free_http_request( req );
	req = NULL;
}

/**
 * We tried to make an outgoing connection to a proxy server.  It
 * either connected or failed to connect.
 */
void
peer_proxy_connection_established_cb(gpointer data, gint source, const gchar *error_message)
{
	PeerConnection *conn;

	conn = data;

	conn->verified_connect_data = NULL;

	if (source < 0)
	{
		peer_connection_trynext(conn);
		return;
	}

	conn->fd = source;
	conn->watcher_incoming = purple_input_add(conn->fd,
			PURPLE_INPUT_READ, peer_proxy_connection_recv_cb, conn);

	if (conn->proxyip != NULL)
		/* Connect to the session created by the remote user */
		peer_proxy_send_join_existing_conn(conn, conn->port);
	else
		/* Create a new session */
		peer_proxy_send_create_new_conn(conn);
}

static gboolean
write_raw(MsnHttpConn *httpconn, const char *data, size_t data_len)
{
	gssize res; /* result of the write operation */

	if (httpconn->tx_handler == 0)
		res = write(httpconn->fd, data, data_len);
	else
	{
		res = -1;
		errno = EAGAIN;
	}

	if ((res <= 0) && ((errno != EAGAIN) && (errno != EWOULDBLOCK)))
	{
		msn_servconn_got_error(httpconn->servconn, MSN_SERVCONN_ERROR_WRITE);
		return FALSE;
	}

	if (res < 0 || res < data_len)
	{
		if (res < 0)
			res = 0;
		if (httpconn->tx_handler == 0 && httpconn->fd)
			httpconn->tx_handler = purple_input_add(httpconn->fd,
				PURPLE_INPUT_WRITE, httpconn_write_cb, httpconn);
		purple_circ_buffer_append(httpconn->tx_buf, data + res,
			data_len - res);
	}

	return TRUE;
}