sctp.c

/*! \file    sctp.c
 * \author   Lorenzo Miniero <lorenzo@meetecho.com>
 * \copyright GNU General Public License v3
 * \brief    SCTP processing for data channels
 * \details  Implementation (based on libusrsctp) of the SCTP Data Channels.
 * The code takes care of the SCTP association between peers and the gateway,
 * and allows for sending and receiving text messages (binary stuff yet to
 * be implemented) after that.
 * 
 * \note Right now, the code is heavily based on the rtcweb.c sample code
 * provided in the \c usrsctp library code, and as such the copyright notice
 * that appears at the beginning of that code is ideally present here as
 * well: http://code.google.com/p/sctp-refimpl/source/browse/trunk/KERN/usrsctp/programs/rtcweb.c 
 * 
 * \note If you want/need to debug SCTP messages for any reason, you can
 * do so by uncommenting the definition of \c DEBUG_SCTP in sctp.h. This
 * will force this code to save all the SCTP messages being exchanged to
 * a separate file for each session. You can choose what folder to save
 * these files in by modifying the \c debug_folder variable. Once a file
 * has been saved, you need to process it using the \c text2pcap tool
 * that is usually shipped with Wireshark, e.g.:
 * 
\verbatim
cd /tmp/sctp
/usr/sbin/text2pcap -n -l 248 -D -t '%H:%M:%S.' sctp-debug-XYZ.txt sctp-debug-XYZ.pcapng
/usr/sbin/wireshark sctp-debug-XYZ.pcapng
\endverbatim
 * 
 * \ingroup protocols
 * \ref protocols
 */

#ifdef HAVE_SCTP
 
#include "sctp.h"
#include "dtls.h"
#include "janus.h"
#include "ice.h"
#include "debug.h"

#ifdef DEBUG_SCTP
/* If we're debugging the SCTP messaging, save the files here */
const char *debug_folder = "/tmp/sctp";
#endif


#define SCTP_MAX_PACKET_SIZE (1<<16)

static uint16_t event_types[] = {
	SCTP_ASSOC_CHANGE,
	SCTP_PEER_ADDR_CHANGE,
	SCTP_REMOTE_ERROR,
	SCTP_SHUTDOWN_EVENT,
	SCTP_ADAPTATION_INDICATION,
	SCTP_SEND_FAILED_EVENT,
	SCTP_STREAM_RESET_EVENT,
	SCTP_STREAM_CHANGE_EVENT
};

int janus_sctp_data_to_dtls(void *instance, void *buffer, size_t length, uint8_t tos, uint8_t set_df);
static int janus_sctp_incoming_data(struct socket *sock, union sctp_sockstore addr, void *data, size_t datalen, struct sctp_rcvinfo rcv, int flags, void *ulp_info);
janus_sctp_channel *janus_sctp_find_channel_by_stream(janus_sctp_association *sctp, uint16_t stream);
janus_sctp_channel *janus_sctp_find_free_channel(janus_sctp_association *sctp);
uint16_t janus_sctp_find_free_stream(janus_sctp_association *sctp);
void janus_sctp_request_more_streams(janus_sctp_association *sctp);
int janus_sctp_send_open_request_message(struct socket *sock, uint16_t stream, uint8_t unordered, uint16_t pr_policy, uint32_t pr_value);
int janus_sctp_send_open_response_message(struct socket *sock, uint16_t stream);
int janus_sctp_send_open_ack_message(struct socket *sock, uint16_t stream);
void janus_sctp_send_deferred_messages(janus_sctp_association *sctp);
int janus_sctp_open_channel(janus_sctp_association *sctp, uint8_t unordered, uint16_t pr_policy, uint32_t pr_value);
int janus_sctp_send_text(janus_sctp_association *sctp, uint16_t id, char *text, size_t length);
void janus_sctp_reset_outgoing_stream(janus_sctp_association *sctp, uint16_t stream);
void janus_sctp_send_outgoing_stream_reset(janus_sctp_association *sctp);
int janus_sctp_close_channel(janus_sctp_association *sctp, uint16_t id);
void janus_sctp_handle_open_request_message(janus_sctp_association *sctp, janus_datachannel_open_request *req, size_t length, uint16_t stream);
void janus_sctp_handle_open_response_message(janus_sctp_association *sctp, janus_datachannel_open_response *rsp, size_t length, uint16_t stream);
void janus_sctp_handle_open_ack_message(janus_sctp_association *sctp, janus_datachannel_ack *ack, size_t length, uint16_t stream);
void janus_sctp_handle_unknown_message(char *msg, size_t length, uint16_t stream);
void janus_sctp_handle_data_message(janus_sctp_association *sctp, char *buffer, size_t length, uint16_t stream);
void janus_sctp_handle_message(janus_sctp_association *sctp, char *buffer, size_t length, uint32_t ppid, uint16_t stream);
void janus_sctp_handle_association_change_event(struct sctp_assoc_change *sac);
void janus_sctp_handle_peer_address_change_event(struct sctp_paddr_change *spc);
void janus_sctp_handle_adaptation_indication(struct sctp_adaptation_event *sai);
void janus_sctp_handle_shutdown_event(struct sctp_shutdown_event *sse);
void janus_sctp_handle_stream_reset_event(janus_sctp_association *sctp, struct sctp_stream_reset_event *strrst);
void janus_sctp_handle_remote_error_event(struct sctp_remote_error *sre);
void janus_sctp_handle_send_failed_event(struct sctp_send_failed_event *ssfe);
void janus_sctp_handle_notification(janus_sctp_association *sctp, union sctp_notification *notif, size_t n);

void *janus_sctp_thread(void *data);
janus_sctp_message *janus_sctp_message_create(char *buffer, size_t length);
void janus_sctp_message_destroy(janus_sctp_message *message);

static gboolean sctp_running;
int janus_sctp_init(void) {
	/* Initialize the SCTP stack */
	usrsctp_init(0, janus_sctp_data_to_dtls, NULL);
	sctp_running = TRUE;

#ifdef DEBUG_SCTP
	JANUS_LOG(LOG_WARN, "SCTP debugging to files enabled: going to save them in %s\n", debug_folder);
	if(janus_mkdir(debug_folder, 0755) < 0) {
		JANUS_LOG(LOG_ERR, "Error creating folder %s, expect problems...\n", debug_folder);
	}
#endif
	return 0;
}

void janus_sctp_deinit(void) {
	usrsctp_finish();
	sctp_running = FALSE;
}

janus_sctp_association *janus_sctp_association_create(void *dtls, uint64_t handle_id, uint16_t udp_port) {
	if(dtls == NULL || udp_port == 0)
		return NULL;
	
	/* usrsctp provides UDP encapsulation of SCTP, but we need these messages to
	 * be encapsulated in DTLS and actually sent/received by libnice, and not by
	 * usrsctp itself... as such, we make use of the AF_CONN approach */

	janus_sctp_association *sctp = (janus_sctp_association *)calloc(1, sizeof(janus_sctp_association));
	if(sctp == NULL) {
		JANUS_LOG(LOG_FATAL, "Memory error!\n");
		return NULL;
	}
	sctp->dtls = dtls;
	sctp->handle_id = handle_id;
	sctp->local_port = 5000;	/* FIXME We always use this one */
	sctp->remote_port = udp_port;
	sctp->sock = NULL;

	struct socket *sock = NULL;
	unsigned int i = 0;
	struct sockaddr_conn sconn;

	/* Now go on with SCTP */
	janus_sctp_channel *channel = NULL;

	for(i = 0; i < NUMBER_OF_CHANNELS; i++) {
		channel = &(sctp->channels[i]);
		channel->id = i;
		channel->state = DATA_CHANNEL_CLOSED;
		channel->pr_policy = SCTP_PR_SCTP_NONE;
		channel->pr_value = 0;
		channel->stream = 0;
		channel->unordered = 0;
		channel->flags = 0;
	}
	for(i = 0; i < NUMBER_OF_STREAMS; i++) {
		sctp->stream_channel[i] = NULL;
		sctp->stream_buffer[i] = 0;
	}
	sctp->stream_buffer_counter = 0;
	sctp->sock = NULL;
	janus_mutex_init(&sctp->mutex);

	usrsctp_register_address((void *)sctp);
	usrsctp_sysctl_set_sctp_ecn_enable(0);
	if((sock = usrsctp_socket(AF_CONN, SOCK_STREAM, IPPROTO_SCTP, janus_sctp_incoming_data, NULL, 0, (void *)sctp)) == NULL) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] Error creating usrsctp socket...\n", handle_id);
		free(sctp);
		sctp = NULL;
		return NULL;
	}
	/* Set SO_LINGER */
	struct linger linger_opt;
	linger_opt.l_onoff = 1;
	linger_opt.l_linger = 0;
	if(usrsctp_setsockopt(sock, SOL_SOCKET, SO_LINGER, &linger_opt, sizeof(linger_opt))) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] setsockopt error: SO_LINGER\n", handle_id);
		free(sctp);
		sctp = NULL;
		return NULL;
	}
	/* Allow resetting streams */
	struct sctp_assoc_value av;
	av.assoc_id = SCTP_ALL_ASSOC;
	av.assoc_value = 1;
	if(usrsctp_setsockopt(sock, IPPROTO_SCTP, SCTP_ENABLE_STREAM_RESET, &av, sizeof(struct sctp_assoc_value)) < 0) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] setsockopt error: SCTP_ENABLE_STREAM_RESET\n", handle_id);
		free(sctp);
		sctp = NULL;
		return NULL;
	}
	/* Disable Nagle */
	uint32_t nodelay = 1;
	if(usrsctp_setsockopt(sock, IPPROTO_SCTP, SCTP_NODELAY, &nodelay, sizeof(nodelay))) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] setsockopt error: SCTP_NODELAY\n", handle_id);
		free(sctp);
		sctp = NULL;
		return NULL;
	}	
	/* Enable the events of interest */
	struct sctp_event event;
	memset(&event, 0, sizeof(event));
	event.se_assoc_id = SCTP_ALL_ASSOC;
	event.se_on = 1;
	for(i = 0; i < sizeof(event_types)/sizeof(uint16_t); i++) {
		event.se_type = event_types[i];
		if(usrsctp_setsockopt(sock, IPPROTO_SCTP, SCTP_EVENT, &event, sizeof(event)) < 0) {
			JANUS_LOG(LOG_ERR, "[%"SCNu64"] setsockopt error: SCTP_EVENT\n", handle_id);
			free(sctp);
			sctp = NULL;
			return NULL;
		}
	}
	/* Configure our INIT message */
	struct sctp_initmsg initmsg;
	memset(&initmsg, 0, sizeof(struct sctp_initmsg));
	initmsg.sinit_num_ostreams = 16;	/* What Firefox says in the INIT (Chrome says 1023) */
	initmsg.sinit_max_instreams = 2048;	/* What both Chrome and Firefox say in the INIT */
	if(usrsctp_setsockopt(sock, IPPROTO_SCTP, SCTP_INITMSG, &initmsg, sizeof(struct sctp_initmsg)) < 0) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] setsockopt error: SCTP_INITMSG\n", handle_id);
		free(sctp);
		sctp = NULL;
		return NULL;
	}
	/* Bind our side of the communication, using AF_CONN as we're doing the actual delivery ourselves */
	memset(&sconn, 0, sizeof(struct sockaddr_conn));
	sconn.sconn_family = AF_CONN;
	sconn.sconn_port = htons(sctp->local_port);
	sconn.sconn_addr = (void *)sctp;
	if(usrsctp_bind(sock, (struct sockaddr *)&sconn, sizeof(struct sockaddr_conn)) < 0) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] Error binding client on port %"SCNu16"\n", handle_id, sctp->local_port);
		free(sctp);
		sctp = NULL;
		return NULL;
	}

#ifdef DEBUG_SCTP
	char debug_file[1024];
	g_snprintf(debug_file, 1024, "%s/sctp-debug-%"SCNu64".txt", debug_folder, handle_id);
	sctp->debug_dump = fopen(debug_file, "wt");
#endif

	/* We're done for now, the setup is done elsewhere */
	janus_mutex_lock(&sctp->mutex);
	sctp->sock = sock;
	sctp->in_messages = g_queue_new();
	sctp->out_messages = g_queue_new();
	GError *error = NULL;
	sctp->thread = g_thread_try_new("JanusSCTP", &janus_sctp_thread, sctp, &error);
	if(error != NULL) {
		/* Something went wrong... */
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] Got error %d (%s) trying to launch the SCTP thread...\n", handle_id, error->code, error->message ? error->message : "??");
		g_queue_free(sctp->in_messages);
		sctp->in_messages = NULL;
		g_queue_free(sctp->out_messages);
		sctp->out_messages = NULL;
		g_free(sctp);
		sctp = NULL;
		return NULL;
	}
	janus_mutex_unlock(&sctp->mutex);
	return sctp;
}

int janus_sctp_association_setup(janus_sctp_association *sctp) {
	if(sctp == NULL)
		return -1;

	struct socket *sock = sctp->sock;
	struct sockaddr_conn sconn;

	/* Operating as client */
	JANUS_LOG(LOG_VERB, "[%"SCNu64"] Connecting the SCTP association\n", sctp->handle_id);
	memset(&sconn, 0, sizeof(struct sockaddr_conn));
	sconn.sconn_family = AF_CONN;
	sconn.sconn_port = htons(sctp->remote_port);
	sconn.sconn_addr = (void *)sctp;
#ifdef HAVE_SCONN_LEN
	sconn.sconn_len = sizeof(struct sockaddr_conn);
#endif
	if(usrsctp_connect(sock, (struct sockaddr *)&sconn, sizeof(struct sockaddr_conn)) < 0) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] Error connecting to SCTP server at port %"SCNu16"\n", sctp->handle_id, sctp->remote_port);
		return -1;
	}
	JANUS_LOG(LOG_INFO, "[%"SCNu64"] Connected to the DataChannel peer\n", sctp->handle_id);

	janus_mutex_lock(&sctp->mutex);
	sctp->sock = sock;
	janus_mutex_unlock(&sctp->mutex);
	return 0;
}

void janus_sctp_association_destroy(janus_sctp_association *sctp) {
	if(sctp == NULL)
		return;
	JANUS_LOG(LOG_INFO, "[%"SCNu64"] Destroying SCTP association\n", sctp->handle_id);
	usrsctp_deregister_address(sctp);
	usrsctp_shutdown(sctp->sock, SHUT_RDWR);
	usrsctp_close(sctp->sock);
	janus_mutex_lock(&sctp->mutex);
	sctp->dtls = NULL;	/* This will get rid of the thread */
	janus_mutex_unlock(&sctp->mutex);
}

void janus_sctp_data_from_dtls(janus_sctp_association *sctp, char *buf, int len) {
	if(sctp == NULL || sctp->dtls == NULL || buf == NULL || len <= 0)
		return;
	JANUS_LOG(LOG_VERB, "[%"SCNu64"] Data from DTLS to SCTP stack: %d bytes\n", sctp->handle_id, len);
	janus_mutex_lock(&sctp->mutex);
	if(sctp->in_messages != NULL)
		g_queue_push_tail(sctp->in_messages, janus_sctp_message_create(buf, len));
	janus_mutex_unlock(&sctp->mutex);
}

int janus_sctp_data_to_dtls(void *instance, void *buffer, size_t length, uint8_t tos, uint8_t set_df) {
	janus_sctp_association *sctp = (janus_sctp_association *)instance;
	if(sctp == NULL || sctp->dtls == NULL)
		return -1;
	JANUS_LOG(LOG_VERB, "[%"SCNu64"] Data from SCTP to DTLS stack: %zu bytes\n", sctp->handle_id, length);
	janus_mutex_lock(&sctp->mutex);
	if(sctp->out_messages != NULL)
		g_queue_push_tail(sctp->out_messages, janus_sctp_message_create(buffer, length));
	janus_mutex_unlock(&sctp->mutex);
	return 0;
}

static int janus_sctp_incoming_data(struct socket *sock, union sctp_sockstore addr, void *data, size_t datalen, struct sctp_rcvinfo rcv, int flags, void *ulp_info) {
	janus_sctp_association *sctp = (janus_sctp_association *)ulp_info;
	if(sctp == NULL || sctp->dtls == NULL)
		return 0;
	if(data) {
		if(flags & MSG_NOTIFICATION) {
			janus_sctp_handle_notification(sctp, (union sctp_notification *)data, datalen);
		} else {
			janus_sctp_handle_message(sctp, data, datalen, ntohl(rcv.rcv_ppid), rcv.rcv_sid);
		}
	}
	return 1;
}

void janus_sctp_send_data(janus_sctp_association *sctp, char *buf, int len) {
	if(sctp == NULL || buf == NULL || len <= 0)
		return;
	JANUS_LOG(LOG_VERB, "[%"SCNu64"] SCTP data to send (%d bytes) coming from a plugin: %s\n", sctp->handle_id, len, buf);
	/* FIXME Is there any open channel we can use? */
	int i = 0, found = 0;
	for(i = 0; i < NUMBER_OF_CHANNELS; i++) {
		if(sctp->channels[i].state != DATA_CHANNEL_CLOSED) {
			found = 1;
			JANUS_LOG(LOG_VERB, "[%"SCNu64"]   -- Using open channel %i\n", sctp->handle_id, i);
			break;
		}
	}
	if(!found) {
		JANUS_LOG(LOG_WARN, "[%"SCNu64"] Couldn't send data, channel %i is not open yet\n", sctp->handle_id, i);
		return;
		//~ /* FIXME There's no open channel (shouldn't happen, we always create it in janus.js), try opening one now */
		//~ if(janus_sctp_open_channel(sctp, 0, 0, 0) < 0) {
			//~ JANUS_LOG(LOG_ERR, "[%"SCNu64"] Couldn't open channel...\n", sctp->handle_id);
			//~ return;
		//~ }
		//~ for(i = 0; i < NUMBER_OF_CHANNELS; i++) {
			//~ if(sctp->channels[i].state != DATA_CHANNEL_CLOSED) {
				//~ found = 1;
				//~ JANUS_LOG(LOG_VERB, "[%"SCNu64"]   -- Using open channel %i\n", sctp->handle_id, i);
				//~ break;
			//~ }
		//~ }
		//~ if(!found) {
			//~ JANUS_LOG(LOG_ERR, "[%"SCNu64"] Channel opened but not found?? giving up...\n", sctp->handle_id);
			//~ return;
		//~ }
	}
	/* FIXME We're assuming this is a string (we don't support binary data yet) */
	janus_sctp_send_text(sctp, i, buf, len);
}


/* From now on, it's SCTP stuff */
janus_sctp_channel *janus_sctp_find_channel_by_stream(janus_sctp_association *sctp, uint16_t stream) {
	if(sctp == NULL)
		return NULL;
	if(stream < NUMBER_OF_STREAMS) {
		return (sctp->stream_channel[stream]);
	} else {
		return NULL;
	}
}

janus_sctp_channel *janus_sctp_find_free_channel(janus_sctp_association *sctp) {
	uint32_t i;

	for(i = 0; i < NUMBER_OF_CHANNELS; i++) {
		if(sctp->channels[i].state == DATA_CHANNEL_CLOSED) {
			break;
		}
	}
	if(i == NUMBER_OF_CHANNELS) {
		return NULL;
	} else {
		return (&(sctp->channels[i]));
	}
}

uint16_t janus_sctp_find_free_stream(janus_sctp_association *sctp) {
	struct sctp_status status;
	uint32_t i, limit;
	socklen_t len;

	len = (socklen_t)sizeof(struct sctp_status);
	if(usrsctp_getsockopt(sctp->sock, IPPROTO_SCTP, SCTP_STATUS, &status, &len) < 0) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] getsockopt error: SCTP_STATUS\n", sctp->handle_id);
		return 0;
	}
	if(status.sstat_outstrms < NUMBER_OF_STREAMS) {
		limit = status.sstat_outstrms;
	} else {
		limit = NUMBER_OF_STREAMS;
	}
	/* stream id 0 is reserved */
	for(i = 1; i < limit; i++) {
		if(sctp->stream_channel[i] == NULL) {
			break;
		}
	}
	if(i == limit) {
		return 0;
	} else {
		return ((uint16_t)i);
	}
}

void janus_sctp_request_more_streams(janus_sctp_association *sctp) {
	struct sctp_status status;
	struct sctp_add_streams sas;
	uint32_t i, streams_needed;
	socklen_t len;

	streams_needed = 0;
	for(i = 0; i < NUMBER_OF_CHANNELS; i++) {
		if((sctp->channels[i].state == DATA_CHANNEL_CONNECTING) &&
		    (sctp->channels[i].stream == 0)) {
			streams_needed++;
		}
	}
	len = (socklen_t)sizeof(struct sctp_status);
	if(usrsctp_getsockopt(sctp->sock, IPPROTO_SCTP, SCTP_STATUS, &status, &len) < 0) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] getsockopt error: SCTP_STATUS\n", sctp->handle_id);
		return;
	}
	if(status.sstat_outstrms + streams_needed > NUMBER_OF_STREAMS) {
		streams_needed = NUMBER_OF_STREAMS - status.sstat_outstrms;
	}
	if(streams_needed == 0) {
		return;
	}
	memset(&sas, 0, sizeof(struct sctp_add_streams));
	sas.sas_instrms = 0;
	sas.sas_outstrms = (uint16_t)streams_needed; /* XXX eror handling */
	if(usrsctp_setsockopt(sctp->sock, IPPROTO_SCTP, SCTP_ADD_STREAMS, &sas, (socklen_t)sizeof(struct sctp_add_streams)) < 0) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] setsockopt error: SCTP_ADD_STREAMS\n", sctp->handle_id);
	}
	return;
}

int janus_sctp_send_open_request_message(struct socket *sock, uint16_t stream, uint8_t unordered, uint16_t pr_policy, uint32_t pr_value) {
	/* XXX: This should be encoded in a better way */
	janus_datachannel_open_request *req = NULL;
	struct sctp_sndinfo sndinfo;

	/* FIXME For open requests we send, we always use this label */
	const char *label = "JanusDataChannel";
	guint label_size = (strlen(label)+3) & ~3;
	JANUS_LOG(LOG_VERB, "Using label '%s' (%zu, %u with padding)\n", label, strlen(label), label_size);

	req = calloc(sizeof(janus_datachannel_open_request) + label_size, sizeof(char));
	memset(req, 0, sizeof(janus_datachannel_open_request) + label_size);
	req->msg_type = DATA_CHANNEL_OPEN_REQUEST;
	switch (pr_policy) {
		case SCTP_PR_SCTP_NONE:
			/* XXX: What about DATA_CHANNEL_RELIABLE_STREAM */
			req->channel_type = DATA_CHANNEL_RELIABLE;
			break;
		case SCTP_PR_SCTP_TTL:
			/* XXX: What about DATA_CHANNEL_UNRELIABLE */
			req->channel_type = DATA_CHANNEL_PARTIAL_RELIABLE_TIMED;
			break;
		case SCTP_PR_SCTP_RTX:
			req->channel_type = DATA_CHANNEL_PARTIAL_RELIABLE_REXMIT;
			break;
		default:
			return 0;
	}
	req->priority = htons(0); /* XXX: add support */
	req->reliability_params = htonl((uint32_t)pr_value);
	req->label_length = htons(label_size);
	memcpy(&req->label, label, strlen(label));
	
	memset(&sndinfo, 0, sizeof(struct sctp_sndinfo));
	sndinfo.snd_sid = stream;
	sndinfo.snd_flags = SCTP_EOR;
	sndinfo.snd_ppid = htonl(DATA_CHANNEL_PPID_CONTROL);
	
	if(usrsctp_sendv(sock,
	                  req, sizeof(janus_datachannel_open_request) + label_size,
	                  NULL, 0,
	                  &sndinfo, (socklen_t)sizeof(struct sctp_sndinfo),
	                  SCTP_SENDV_SNDINFO, 0) < 0) {
		JANUS_LOG(LOG_ERR, "usrsctp_sendv error\n");
		g_free(req);
		req = NULL;
		return 0;
	} else {
		g_free(req);
		req = NULL;
		return 1;
	}
}

int janus_sctp_send_open_response_message(struct socket *sock, uint16_t stream) {
	/* XXX: This should be encoded in a better way */
	janus_datachannel_open_response rsp;
	struct sctp_sndinfo sndinfo;

	memset(&rsp, 0, sizeof(janus_datachannel_open_response));
	rsp.msg_type = DATA_CHANNEL_OPEN_RESPONSE;
	rsp.error = 0;
	rsp.flags = htons(0);
	rsp.reverse_stream = htons(stream);
	memset(&sndinfo, 0, sizeof(struct sctp_sndinfo));
	sndinfo.snd_sid = stream;
	sndinfo.snd_flags = SCTP_EOR;
	sndinfo.snd_ppid = htonl(DATA_CHANNEL_PPID_CONTROL);
	if(usrsctp_sendv(sock,
	                  &rsp, sizeof(janus_datachannel_open_response),
	                  NULL, 0,
	                  &sndinfo, (socklen_t)sizeof(struct sctp_sndinfo),
	                  SCTP_SENDV_SNDINFO, 0) < 0) {
		JANUS_LOG(LOG_ERR, "usrsctp_sendv error\n");
		return 0;
	} else {
		return 1;
	}
}

int janus_sctp_send_open_ack_message(struct socket *sock, uint16_t stream) {
	/* XXX: This should be encoded in a better way */
	janus_datachannel_ack ack;
	struct sctp_sndinfo sndinfo;

	memset(&ack, 0, sizeof(janus_datachannel_ack));
	ack.msg_type = DATA_CHANNEL_ACK;
	memset(&sndinfo, 0, sizeof(struct sctp_sndinfo));
	sndinfo.snd_sid = stream;
	sndinfo.snd_flags = SCTP_EOR;
	sndinfo.snd_ppid = htonl(DATA_CHANNEL_PPID_CONTROL);
	if(usrsctp_sendv(sock,
	                  &ack, sizeof(janus_datachannel_ack),
	                  NULL, 0,
	                  &sndinfo, (socklen_t)sizeof(struct sctp_sndinfo),
	                  SCTP_SENDV_SNDINFO, 0) < 0) {
		JANUS_LOG(LOG_ERR, "usrsctp_sendv error\n");
		return 0;
	} else {
		return 1;
	}
}

void janus_sctp_send_deferred_messages(janus_sctp_association *sctp) {
	uint32_t i;
	janus_sctp_channel *channel;

	for(i = 0; i < NUMBER_OF_CHANNELS; i++) {
		channel = &(sctp->channels[i]);
		if(channel->flags & DATA_CHANNEL_FLAGS_SEND_REQ) {
			if(janus_sctp_send_open_request_message(sctp->sock, channel->stream, channel->unordered, channel->pr_policy, channel->pr_value)) {
				channel->flags &= ~DATA_CHANNEL_FLAGS_SEND_REQ;
			} else {
				if(errno != EAGAIN) {
					/* XXX: error handling */
				}
			}
		}
		if(channel->flags & DATA_CHANNEL_FLAGS_SEND_RSP) {
			if(janus_sctp_send_open_response_message(sctp->sock, channel->stream)) {
				channel->flags &= ~DATA_CHANNEL_FLAGS_SEND_RSP;
			} else {
				if(errno != EAGAIN) {
					/* XXX: error handling */
				}
			}
		}
		if(channel->flags & DATA_CHANNEL_FLAGS_SEND_ACK) {
			if(janus_sctp_send_open_ack_message(sctp->sock, channel->stream)) {
				channel->flags &= ~DATA_CHANNEL_FLAGS_SEND_ACK;
			} else {
				if(errno != EAGAIN) {
					/* XXX: error handling */
				}
			}
		}
	}
	return;
}

int janus_sctp_open_channel(janus_sctp_association *sctp, uint8_t unordered, uint16_t pr_policy, uint32_t pr_value) {
	if(sctp == NULL)
		return -1;
	janus_sctp_channel *channel;
	uint16_t stream;

	if((pr_policy != SCTP_PR_SCTP_NONE) &&
			(pr_policy != SCTP_PR_SCTP_TTL) &&
			(pr_policy != SCTP_PR_SCTP_RTX)) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] Invalid pr_policy %"SCNu32"\n", sctp->handle_id, pr_policy);
		return -1;
	}
	if((pr_policy == SCTP_PR_SCTP_NONE) && (pr_value != 0)) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] Invalid pr_value %"SCNu32" for SCTP_PR_SCTP_NONE\n", sctp->handle_id, pr_value);
		return -1;
	}
	if((channel = janus_sctp_find_free_channel(sctp)) == NULL) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] No more free channels available\n", sctp->handle_id);
		return -1;
	}
	stream = janus_sctp_find_free_stream(sctp);
	channel->state = DATA_CHANNEL_CONNECTING;
	channel->unordered = unordered ? 1 : 0;
	channel->pr_policy = pr_policy;
	channel->pr_value = pr_value;
	channel->stream = stream;
	channel->flags = 0;
	if(stream == 0) {
		janus_sctp_request_more_streams(sctp);
	} else {
		if(janus_sctp_send_open_request_message(sctp->sock, stream, unordered, pr_policy, pr_value)) {
			sctp->stream_channel[stream] = channel;
		} else {
			if(errno == EAGAIN) {
				sctp->stream_channel[stream] = channel;
				channel->flags |= DATA_CHANNEL_FLAGS_SEND_REQ;
			} else {
				channel->state = DATA_CHANNEL_CLOSED;
				channel->unordered = 0;
				channel->pr_policy = 0;
				channel->pr_value = 0;
				channel->stream = 0;
				channel->flags = 0;
				channel = NULL;
			}
		}
	}
	return 0;
}

int janus_sctp_send_text(janus_sctp_association *sctp, uint16_t id, char *text, size_t length) {
	if(id >= NUMBER_OF_CHANNELS || text == NULL)
		return -1;
	struct sctp_sendv_spa spa;
	janus_sctp_channel *channel = &sctp->channels[id];
	if(channel == NULL) {
		/* No such channel */
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] No such channel %"SCNu16"...\n", sctp->handle_id, id); 
		return -1;
	}
	if((channel->state != DATA_CHANNEL_OPEN) && (channel->state != DATA_CHANNEL_CONNECTING)) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] Channel %"SCNu16" is neither open nor connecting (state=%d)...\n", sctp->handle_id, id, channel->state); 
		return -1;
	}

	memset(&spa, 0, sizeof(struct sctp_sendv_spa));
	spa.sendv_sndinfo.snd_sid = channel->stream;
	if((channel->state == DATA_CHANNEL_OPEN) && (channel->unordered)) {
		spa.sendv_sndinfo.snd_flags = SCTP_EOR | SCTP_UNORDERED;
	} else {
		spa.sendv_sndinfo.snd_flags = SCTP_EOR;
	}
	spa.sendv_sndinfo.snd_ppid = htonl(DATA_CHANNEL_PPID_DOMSTRING);
	spa.sendv_flags = SCTP_SEND_SNDINFO_VALID;
	if((channel->pr_policy == SCTP_PR_SCTP_TTL) || (channel->pr_policy == SCTP_PR_SCTP_RTX)) {
		spa.sendv_prinfo.pr_policy = channel->pr_policy;
		spa.sendv_prinfo.pr_value = channel->pr_value;
		spa.sendv_flags |= SCTP_SEND_PRINFO_VALID;
	}
	if(usrsctp_sendv(sctp->sock, text, length, NULL, 0,
			&spa, (socklen_t)sizeof(struct sctp_sendv_spa),
			SCTP_SENDV_SPA, 0) < 0) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] sctp_sendv error\n", sctp->handle_id);
		return -1;
	}
	JANUS_LOG(LOG_VERB, "[%"SCNu64"] Message sent on channel %"SCNu16"\n", sctp->handle_id, id); 
	return 0;
}

void janus_sctp_reset_outgoing_stream(janus_sctp_association *sctp, uint16_t stream) {
	uint32_t i;

	for(i = 0; i < sctp->stream_buffer_counter; i++) {
		if(sctp->stream_buffer[i] == stream) {
			return;
		}
	}
	sctp->stream_buffer[sctp->stream_buffer_counter++] = stream;
	return;
}

void janus_sctp_send_outgoing_stream_reset(janus_sctp_association *sctp) {
	struct sctp_reset_streams *srs;
	uint32_t i;
	size_t len;

	if(sctp->stream_buffer_counter == 0) {
		return;
	}
	len = sizeof(sctp_assoc_t) + (2 + sctp->stream_buffer_counter) * sizeof(uint16_t);
	srs = (struct sctp_reset_streams *)malloc(len);
	if(srs == NULL) {
		return;
	}
	memset(srs, 0, len);
	srs->srs_flags = SCTP_STREAM_RESET_OUTGOING;
	srs->srs_number_streams = sctp->stream_buffer_counter;
	for(i = 0; i < sctp->stream_buffer_counter; i++) {
		srs->srs_stream_list[i] = sctp->stream_buffer[i];
	}
	if(usrsctp_setsockopt(sctp->sock, IPPROTO_SCTP, SCTP_RESET_STREAMS, srs, (socklen_t)len) < 0) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] setsockopt error: SCTP_RESET_STREAMS\n", sctp->handle_id);
	} else {
		for(i = 0; i < sctp->stream_buffer_counter; i++) {
			srs->srs_stream_list[i] = 0;
		}
		sctp->stream_buffer_counter = 0;
	}
	free(srs);
	return;
}

int janus_sctp_close_channel(janus_sctp_association *sctp, uint16_t id) {
	if(id >= NUMBER_OF_CHANNELS)
		return -1;
	janus_sctp_channel *channel = &sctp->channels[id];
	if(channel == NULL) {
		return -1;
	}
	if(channel->state != DATA_CHANNEL_OPEN) {
		return -1;
	}
	janus_sctp_reset_outgoing_stream(sctp, channel->stream);
	janus_sctp_send_outgoing_stream_reset(sctp);
	channel->state = DATA_CHANNEL_CLOSING;
	return 0;
}

void janus_sctp_handle_open_request_message(janus_sctp_association *sctp, janus_datachannel_open_request *req, size_t length, uint16_t stream) {
	janus_sctp_channel *channel;
	uint32_t pr_value;
	uint16_t pr_policy;
	uint8_t unordered;

	if((channel = janus_sctp_find_channel_by_stream(sctp, stream))) {
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] channel %d is in state %d instead of CLOSED.\n", sctp->handle_id, channel->id, channel->state);
		/* XXX: some error handling */
		return;
	}
	if((channel = janus_sctp_find_free_channel(sctp)) == NULL) {
		/* XXX: some error handling */
		return;
	}
	switch (req->channel_type) {
		case DATA_CHANNEL_RELIABLE:
			pr_policy = SCTP_PR_SCTP_NONE;
			unordered = 0;
			break;
		case DATA_CHANNEL_RELIABLE_UNORDERED:
			pr_policy = SCTP_PR_SCTP_NONE;
			unordered = 1;
			break;
		case DATA_CHANNEL_PARTIAL_RELIABLE_TIMED:
			pr_policy = SCTP_PR_SCTP_TTL;
			unordered = 0;
			break;
		case DATA_CHANNEL_PARTIAL_RELIABLE_TIMED_UNORDERED:
			pr_policy = SCTP_PR_SCTP_TTL;
			unordered = 1;
			break;
		case DATA_CHANNEL_PARTIAL_RELIABLE_REXMIT:
			pr_policy = SCTP_PR_SCTP_RTX;
			unordered = 1;
			break;
		case DATA_CHANNEL_PARTIAL_RELIABLE_REXMIT_UNORDERED:
			pr_policy = SCTP_PR_SCTP_RTX;
			unordered = 1;
			break;
		default:
			pr_policy = SCTP_PR_SCTP_NONE;
			unordered = 0;
			/* FIXME Should we handle some error, here? */
			break;
	}
	pr_value = ntohs(req->reliability_params);
	channel->state = DATA_CHANNEL_CONNECTING;
	channel->unordered = unordered;
	channel->pr_policy = pr_policy;
	channel->pr_value = pr_value;
	channel->stream = stream;
	channel->flags = 0;
	sctp->stream_channel[stream] = channel;
	if(stream == 0) {
		janus_sctp_request_more_streams(sctp);
	} else {
		if(janus_sctp_send_open_ack_message(sctp->sock, stream)) {
			sctp->stream_channel[stream] = channel;
		} else {
			if(errno == EAGAIN) {
				channel->flags |= DATA_CHANNEL_FLAGS_SEND_ACK;
				sctp->stream_channel[stream] = channel;
			} else {
				/* XXX: Signal error to the other end */
				sctp->stream_channel[stream] = NULL;
				channel->state = DATA_CHANNEL_CLOSED;
				channel->unordered = 0;
				channel->pr_policy = 0;
				channel->pr_value = 0;
				channel->stream = 0;
				channel->flags = 0;
			}
		}
	}
	/* Read label, if available */
	char *label = NULL;
	guint len = ntohs(req->label_length);
	if(len > 0 && len < length) {
		label = calloc(len+1, sizeof(char));
		memcpy(label, req->label, len);
		label[len] = '\0'; 
	}
	JANUS_LOG(LOG_VERB, "[%"SCNu64"] Opened channel '%s' (id=%"SCNu16") (%d/%d/%d)\n", sctp->handle_id,
		label ? label : "??",
		channel->stream, channel->unordered, channel->pr_policy, channel->pr_value); 
}

void janus_sctp_handle_open_response_message(janus_sctp_association *sctp, janus_datachannel_open_response *rsp, size_t length, uint16_t stream) {
	janus_sctp_channel *channel;

	channel = janus_sctp_find_channel_by_stream(sctp, stream);
	if(channel == NULL) {
		/* XXX: improve error handling */
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] Can't find channel for outgoing steam %d.\n", sctp->handle_id, stream);
		return;
	}
	if(channel->state != DATA_CHANNEL_CONNECTING) {
		/* XXX: improve error handling */
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] Channel with id %d for outgoing steam %d is in state %d.\n", sctp->handle_id, channel->id, stream, channel->state);
		return;
	}
	if(janus_sctp_find_channel_by_stream(sctp, stream)) {
		/* XXX: improve error handling */
		JANUS_LOG(LOG_ERR, "[%"SCNu64"] Channel collision for channel with id %d and streams (in/out) = (%d/%d).\n", sctp->handle_id, channel->id, stream, stream);
		return;
	}
	channel->stream = stream;
	channel->state = DATA_CHANNEL_OPEN;
	sctp->stream_channel[stream] = channel;
	if(janus_sctp_send_open_ack_message(sctp->sock, stream)) {
		channel->flags = 0;
	} else {
		channel->flags |= DATA_CHANNEL_FLAGS_SEND_ACK;
	}
	return;
}

void janus_sctp_handle_open_ack_message(janus_sctp_association *sctp, janus_datachannel_ack *ack, size_t length, uint16_t stream) {
	janus_sctp_channel *channel;

	channel = janus_sctp_find_channel_by_stream(sctp, stream);
	if(channel == NULL) {
		/* XXX: some error handling */
		JANUS_LOG(LOG_ERR, "Ops, no channel with stream %"SCNu16"?\n", stream);
		return;
	}
	if(channel->state == DATA_CHANNEL_OPEN) {
		return;
	}
	if(channel->state != DATA_CHANNEL_CONNECTING) {
		/* XXX: error handling */
		return;
	}
	channel->state = DATA_CHANNEL_OPEN;
	return;
}

void janus_sctp_handle_unknown_message(char *msg, size_t length, uint16_t stream) {
	/* XXX: Send an error message */
	return;
}

void janus_sctp_handle_data_message(janus_sctp_association *sctp, char *buffer, size_t length, uint16_t stream) {
	janus_sctp_channel *channel;

	channel = janus_sctp_find_channel_by_stream(sctp, stream);
	if(channel == NULL) {
		/* XXX: Some error handling */
		JANUS_LOG(LOG_WARN, "[%"SCNu64"] Got data from this SCTP association but there is no channel with stream %"SCNu16"...\n", sctp->handle_id, stream);
		return;
	}
	if(channel->state == DATA_CHANNEL_CONNECTING) {
		/* Implicit ACK */
		channel->state = DATA_CHANNEL_OPEN;
	}
	if(channel->state != DATA_CHANNEL_OPEN) {
		/* XXX: What about other states? */
		/* XXX: Some error handling */
		JANUS_LOG(LOG_WARN, "[%"SCNu64"] Got data from this SCTP association but channel isn't open yet...\n", sctp->handle_id);
		return;
	} else {
		/* Assuming DATA_CHANNEL_PPID_DOMSTRING */
		/* XXX: Protect for non 0 terminated buffer */
		JANUS_LOG(LOG_VERB, "[%"SCNu64"] Message received of length %zu on channel with id %d: %.*s\n",
		       sctp->handle_id, length, channel->id, (int)length, buffer);
		/* FIXME: notify this to the core */
		janus_dtls_notify_data((janus_dtls_srtp *)sctp->dtls, buffer, (int)length);
	}
	return;
}

void janus_sctp_handle_message(janus_sctp_association *sctp, char *buffer, size_t length, uint32_t ppid, uint16_t stream) {
	janus_datachannel_open_request *req;
	janus_datachannel_open_response *rsp;
	janus_datachannel_ack *ack, *msg;

	switch (ppid) {
		case DATA_CHANNEL_PPID_CONTROL:
			if(length < sizeof(janus_datachannel_ack)) {
				return;
			}
			msg = (janus_datachannel_ack *)buffer;
			switch (msg->msg_type) {
				case DATA_CHANNEL_OPEN_REQUEST:
					if(length < sizeof(janus_datachannel_open_request)) {
						/* XXX: error handling? */
						return;
					}
					req = (janus_datachannel_open_request *)buffer;
					janus_sctp_handle_open_request_message(sctp, req, length, stream);
					break;
				case DATA_CHANNEL_OPEN_RESPONSE:
					if(length < sizeof(janus_datachannel_open_response)) {
						/* XXX: error handling? */
						return;
					}
					rsp = (janus_datachannel_open_response *)buffer;
					janus_sctp_handle_open_response_message(sctp, rsp, length, stream);
					break;
				case DATA_CHANNEL_ACK:
					if(length < sizeof(janus_datachannel_ack)) {
						/* XXX: error handling? */
						return;
					}
					ack = (janus_datachannel_ack *)buffer;
					janus_sctp_handle_open_ack_message(sctp, ack, length, stream);
					break;
				default:
					janus_sctp_handle_unknown_message(buffer, length, stream);
					break;
			}
			break;
		case DATA_CHANNEL_PPID_DOMSTRING:
		case DATA_CHANNEL_PPID_BINARY:
		case DATA_CHANNEL_PPID_DOMSTRING_PARTIAL:
		case DATA_CHANNEL_PPID_BINARY_PARTIAL:
			janus_sctp_handle_data_message(sctp, buffer, length, stream);
			break;
		default:
			JANUS_LOG(LOG_VERB, "[%"SCNu64"] Message of length %zu, PPID %u on stream %u received.\n",
				   sctp->handle_id, length, ppid, stream);
			break;
	}
}

void janus_sctp_handle_association_change_event(struct sctp_assoc_change *sac) {
	unsigned int i, n;

	JANUS_LOG(LOG_VERB, "Association change ");
	switch (sac->sac_state) {
		case SCTP_COMM_UP:
			JANUS_LOG(LOG_VERB, "SCTP_COMM_UP");
			break;
		case SCTP_COMM_LOST:
			JANUS_LOG(LOG_VERB, "SCTP_COMM_LOST");
			break;
		case SCTP_RESTART:
			JANUS_LOG(LOG_VERB, "SCTP_RESTART");
			break;
		case SCTP_SHUTDOWN_COMP:
			JANUS_LOG(LOG_VERB, "SCTP_SHUTDOWN_COMP");
			break;
		case SCTP_CANT_STR_ASSOC:
			JANUS_LOG(LOG_VERB, "SCTP_CANT_STR_ASSOC");
			break;
		default:
			JANUS_LOG(LOG_VERB, "UNKNOWN");
			break;
	}
	JANUS_LOG(LOG_VERB, ", streams (in/out) = (%u/%u)",
	       sac->sac_inbound_streams, sac->sac_outbound_streams);
	n = sac->sac_length - sizeof(struct sctp_assoc_change);
	if(((sac->sac_state == SCTP_COMM_UP) ||
	     (sac->sac_state == SCTP_RESTART)) && (n > 0)) {
		JANUS_LOG(LOG_VERB, ", supports");
		for(i = 0; i < n; i++) {
			switch (sac->sac_info[i]) {
				case SCTP_ASSOC_SUPPORTS_PR:
					JANUS_LOG(LOG_VERB, " PR");
					break;
				case SCTP_ASSOC_SUPPORTS_AUTH:
					JANUS_LOG(LOG_VERB, " AUTH");
					break;
				case SCTP_ASSOC_SUPPORTS_ASCONF:
					JANUS_LOG(LOG_VERB, " ASCONF");
					break;
				case SCTP_ASSOC_SUPPORTS_MULTIBUF:
					JANUS_LOG(LOG_VERB, " MULTIBUF");
					break;
				case SCTP_ASSOC_SUPPORTS_RE_CONFIG:
					JANUS_LOG(LOG_VERB, " RE-CONFIG");
					break;
				default:
					JANUS_LOG(LOG_VERB, " UNKNOWN(0x%02x)", sac->sac_info[i]);
					break;
			}
		}
	} else if(((sac->sac_state == SCTP_COMM_LOST) ||
	            (sac->sac_state == SCTP_CANT_STR_ASSOC)) && (n > 0)) {
		JANUS_LOG(LOG_VERB, ", ABORT =");
		for(i = 0; i < n; i++) {
			JANUS_LOG(LOG_VERB, " 0x%02x", sac->sac_info[i]);
		}
	}
	JANUS_LOG(LOG_VERB, ".\n");
	if((sac->sac_state == SCTP_CANT_STR_ASSOC) ||
			(sac->sac_state == SCTP_SHUTDOWN_COMP) ||
			(sac->sac_state == SCTP_COMM_LOST)) {
		/* FIXME Should we notify the application that data channels were lost? */
	}
	return;
}

void janus_sctp_handle_peer_address_change_event(struct sctp_paddr_change *spc) {
	char addr_buf[INET6_ADDRSTRLEN];
	const char *addr;
	struct sockaddr_in *sin;
	struct sockaddr_in6 *sin6;

	switch (spc->spc_aaddr.ss_family) {
		case AF_INET:
			sin = (struct sockaddr_in *)&spc->spc_aaddr;
			addr = inet_ntop(AF_INET, &sin->sin_addr, addr_buf, INET_ADDRSTRLEN);
			break;
		case AF_INET6:
			sin6 = (struct sockaddr_in6 *)&spc->spc_aaddr;
			addr = inet_ntop(AF_INET6, &sin6->sin6_addr, addr_buf, INET6_ADDRSTRLEN);
			break;
		default:
			snprintf(addr_buf, INET6_ADDRSTRLEN, "Unknown family %d", spc->spc_aaddr.ss_family);
			addr = addr_buf;
			break;
	}
	JANUS_LOG(LOG_VERB, "Peer address %s is now ", addr);
	switch (spc->spc_state) {
		case SCTP_ADDR_AVAILABLE:
			JANUS_LOG(LOG_VERB, "SCTP_ADDR_AVAILABLE");
			break;
		case SCTP_ADDR_UNREACHABLE:
			JANUS_LOG(LOG_VERB, "SCTP_ADDR_UNREACHABLE");
			break;
		case SCTP_ADDR_REMOVED:
			JANUS_LOG(LOG_VERB, "SCTP_ADDR_REMOVED");
			break;
		case SCTP_ADDR_ADDED:
			JANUS_LOG(LOG_VERB, "SCTP_ADDR_ADDED");
			break;
		case SCTP_ADDR_MADE_PRIM:
			JANUS_LOG(LOG_VERB, "SCTP_ADDR_MADE_PRIM");
			break;
		case SCTP_ADDR_CONFIRMED:
			JANUS_LOG(LOG_VERB, "SCTP_ADDR_CONFIRMED");
			break;
		default:
			JANUS_LOG(LOG_VERB, "UNKNOWN");
			break;
	}
	JANUS_LOG(LOG_VERB, " (error = 0x%08x).\n", spc->spc_error);
	return;
}

void janus_sctp_handle_adaptation_indication(struct sctp_adaptation_event *sai) {
	JANUS_LOG(LOG_VERB, "Adaptation indication: %x.\n", sai-> sai_adaptation_ind);
	return;
}

void janus_sctp_handle_shutdown_event(struct sctp_shutdown_event *sse) {
	JANUS_LOG(LOG_VERB, "Shutdown event.\n");
	/* XXX: notify all channels */
	return;
}

void janus_sctp_handle_stream_reset_event(janus_sctp_association *sctp, struct sctp_stream_reset_event *strrst) {
	uint32_t n, i;
	janus_sctp_channel *channel;

	n = (strrst->strreset_length - sizeof(struct sctp_stream_reset_event)) / sizeof(uint16_t);
	JANUS_LOG(LOG_VERB, "[%"SCNu64"] Stream reset event: flags = %x, ", sctp->handle_id, strrst->strreset_flags);
	if(strrst->strreset_flags & SCTP_STREAM_RESET_INCOMING_SSN) {
		if(strrst->strreset_flags & SCTP_STREAM_RESET_OUTGOING_SSN) {
			JANUS_LOG(LOG_VERB, "incoming/");
		}
		JANUS_LOG(LOG_VERB, "incoming ");
	}
	if(strrst->strreset_flags & SCTP_STREAM_RESET_OUTGOING_SSN) {
		JANUS_LOG(LOG_VERB, "outgoing ");
	}
	JANUS_LOG(LOG_VERB, "stream ids = ");
	for(i = 0; i < n; i++) {
		if(i > 0) {
			JANUS_LOG(LOG_VERB, ", ");
		}
		JANUS_LOG(LOG_VERB, "%d", strrst->strreset_stream_list[i]);
	}
	JANUS_LOG(LOG_VERB, ".\n");	
	if(!(strrst->strreset_flags & SCTP_STREAM_RESET_DENIED) &&
	    !(strrst->strreset_flags & SCTP_STREAM_RESET_FAILED)) {
		for(i = 0; i < n; i++) {
			if(strrst->strreset_flags & SCTP_STREAM_RESET_INCOMING_SSN ||
					strrst->strreset_flags & SCTP_STREAM_RESET_OUTGOING_SSN) {
				channel = janus_sctp_find_channel_by_stream(sctp, strrst->strreset_stream_list[i]);
				if(channel != NULL) {
					sctp->stream_channel[channel->stream] = NULL;
					if(channel->stream == 0) {
						channel->pr_policy = SCTP_PR_SCTP_NONE;
						channel->pr_value = 0;
						channel->unordered = 0;
						channel->flags = 0;
						channel->state = DATA_CHANNEL_CLOSED;
					} else {
						if(channel->state == DATA_CHANNEL_OPEN) {
							janus_sctp_reset_outgoing_stream(sctp, channel->stream);
							channel->state = DATA_CHANNEL_CLOSING;
						} else {
							/* XXX: What to do? */
						}
					}
				}
			}
		}
	}
	return;
}

void janus_sctp_handle_remote_error_event(struct sctp_remote_error *sre) {
	size_t i, n;

	n = sre->sre_length - sizeof(struct sctp_remote_error);
	JANUS_LOG(LOG_VERB, "Remote Error (error = 0x%04x): ", sre->sre_error);
	for(i = 0; i < n; i++) {
		JANUS_LOG(LOG_VERB, " 0x%02x", sre-> sre_data[i]);
	}
	JANUS_LOG(LOG_VERB, ".\n");
	return;
}

void janus_sctp_handle_send_failed_event(struct sctp_send_failed_event *ssfe) {
	size_t i, n;

	if(ssfe->ssfe_flags & SCTP_DATA_UNSENT) {
		JANUS_LOG(LOG_VERB, "Unsent ");
	}
	if(ssfe->ssfe_flags & SCTP_DATA_SENT) {
		JANUS_LOG(LOG_VERB, "Sent ");
	}
	if(ssfe->ssfe_flags & ~(SCTP_DATA_SENT | SCTP_DATA_UNSENT)) {
		JANUS_LOG(LOG_VERB, "(flags = %x) ", ssfe->ssfe_flags);
	}
	JANUS_LOG(LOG_VERB, "message with PPID = %d, SID = %d, flags: 0x%04x due to error = 0x%08x",
	       ntohl(ssfe->ssfe_info.snd_ppid), ssfe->ssfe_info.snd_sid,
	       ssfe->ssfe_info.snd_flags, ssfe->ssfe_error);
	n = ssfe->ssfe_length - sizeof(struct sctp_send_failed_event);
	for(i = 0; i < n; i++) {
		JANUS_LOG(LOG_VERB, " 0x%02x", ssfe->ssfe_data[i]);
	}
	JANUS_LOG(LOG_VERB, ".\n");
	return;
}

void janus_sctp_handle_notification(janus_sctp_association *sctp, union sctp_notification *notif, size_t n) {
	if(notif->sn_header.sn_length != (uint32_t)n) {
		return;
	}
	switch (notif->sn_header.sn_type) {
		case SCTP_ASSOC_CHANGE:
			janus_sctp_handle_association_change_event(&(notif->sn_assoc_change));
			break;
		case SCTP_PEER_ADDR_CHANGE:
			janus_sctp_handle_peer_address_change_event(&(notif->sn_paddr_change));
			break;
		case SCTP_REMOTE_ERROR:
			janus_sctp_handle_remote_error_event(&(notif->sn_remote_error));
			break;
		case SCTP_SHUTDOWN_EVENT:
			janus_sctp_handle_shutdown_event(&(notif->sn_shutdown_event));
			break;
		case SCTP_ADAPTATION_INDICATION:
			janus_sctp_handle_adaptation_indication(&(notif->sn_adaptation_event));
			break;
		case SCTP_PARTIAL_DELIVERY_EVENT:
			break;
		case SCTP_AUTHENTICATION_EVENT:
			break;
		case SCTP_SENDER_DRY_EVENT:
			break;
		case SCTP_NOTIFICATIONS_STOPPED_EVENT:
			break;
		case SCTP_SEND_FAILED_EVENT:
			janus_sctp_handle_send_failed_event(&(notif->sn_send_failed_event));
			break;
		case SCTP_STREAM_RESET_EVENT:
			janus_sctp_handle_stream_reset_event(sctp, &(notif->sn_strreset_event));
			janus_sctp_send_deferred_messages(sctp);
			janus_sctp_send_outgoing_stream_reset(sctp);
			janus_sctp_request_more_streams(sctp);
			break;
		case SCTP_ASSOC_RESET_EVENT:
			break;
		case SCTP_STREAM_CHANGE_EVENT:
			break;
		default:
			break;
	}
}


void *janus_sctp_thread(void *data) {
	janus_sctp_association *sctp = (janus_sctp_association *)data;
	if(sctp == NULL) {
		JANUS_LOG(LOG_ERR, "Invalid SCTP association, closing thread\n");
		g_thread_unref(g_thread_self());
		return NULL;
	}
	JANUS_LOG(LOG_INFO, "[%"SCNu64"] Starting thread for SCTP association\n", sctp->handle_id);
	janus_sctp_message *message = NULL;
	gboolean sent_data = FALSE;
	while(sctp->dtls != NULL && sctp_running) {
		/* Anything to do at all? */
		janus_mutex_lock(&sctp->mutex);
		if(g_queue_is_empty(sctp->in_messages) && g_queue_is_empty(sctp->out_messages)) {
			janus_mutex_unlock(&sctp->mutex);
			g_usleep (10000);
			continue;
		}
		/* Check incoming messages */
		if(!g_queue_is_empty(sctp->in_messages) && sent_data) {
			while(!g_queue_is_empty(sctp->in_messages)) {
				message = g_queue_pop_head(sctp->in_messages);
				if(message == NULL)
					continue;
#ifdef DEBUG_SCTP
				if(sctp->debug_dump != NULL) {
					/* Dump incoming message */
					char *dump = usrsctp_dumppacket(message->buffer, message->length, SCTP_DUMP_INBOUND);
					if(dump != NULL) {
						fwrite(dump, sizeof(char), strlen(dump), sctp->debug_dump);
						fflush(sctp->debug_dump);
						usrsctp_freedumpbuffer(dump);
					}
				}
#endif
				/* Pass this data to the SCTP association */
				janus_mutex_unlock(&sctp->mutex);
				usrsctp_conninput((void *)sctp, message->buffer, message->length, 0);
				janus_mutex_lock(&sctp->mutex);
				janus_sctp_message_destroy(message);
				message = NULL;
			}
		}
		/* Check outgoing messages */
		if(sctp->dtls == NULL) {
			/* No DTLS stack anymore, we're done */
			janus_mutex_unlock(&sctp->mutex);
			break;
		}
		if(!g_queue_is_empty(sctp->out_messages)) {
			while(!g_queue_is_empty(sctp->out_messages)) {
				message = g_queue_pop_head(sctp->out_messages);
				if(message == NULL)
					continue;
#ifdef DEBUG_SCTP
				if(sctp->debug_dump != NULL) {
					/* Dump incoming message */
					char *dump = usrsctp_dumppacket(message->buffer, message->length, SCTP_DUMP_OUTBOUND);
					if(dump != NULL) {
						fwrite(dump, sizeof(char), strlen(dump), sctp->debug_dump);
						fflush(sctp->debug_dump);
						usrsctp_freedumpbuffer(dump);
					}
				}
#endif
				/* Encapsulate this data in DTLS and send it */
				janus_dtls_send_sctp_data((janus_dtls_srtp *)sctp->dtls, message->buffer, message->length);
				janus_sctp_message_destroy(message);
				message = NULL;
				if(!sent_data)
					sent_data = TRUE;
			}
		}
		janus_mutex_unlock(&sctp->mutex);
	}
	JANUS_LOG(LOG_INFO, "[%"SCNu64"] Leaving SCTP association thread\n", sctp->handle_id);
	/* This association has been destroyed, wait a bit and then free all the resources */
	g_usleep (1*G_USEC_PER_SEC);
	g_queue_free(sctp->in_messages);
	sctp->in_messages = NULL;
	g_queue_free(sctp->out_messages);
	sctp->out_messages = NULL;
	sctp->thread = NULL;
#ifdef DEBUG_SCTP
	if(sctp->debug_dump != NULL)
		fclose(sctp->debug_dump);
	sctp->debug_dump = NULL;
#endif
	g_free(sctp);
	sctp = NULL;
	g_thread_unref(g_thread_self());
	return NULL;
}

janus_sctp_message *janus_sctp_message_create(char *buffer, size_t length) {
	if(buffer == NULL || length == 0)
		return NULL;
	janus_sctp_message *message = calloc(1, sizeof(janus_sctp_message));
	if(message == NULL)
		return NULL;
	message->buffer = calloc(length, sizeof(char));
	if(message->buffer == NULL) {
		free(message);
		message = NULL;
		return NULL;
	}
	memcpy(message->buffer, buffer, length);
	message->length = length;
	return message;
}

void janus_sctp_message_destroy(janus_sctp_message *message) {
	if(message == NULL)
		return;
	if(message->buffer != NULL)
		free(message->buffer);
	message->buffer = NULL;
	free(message);
	message = NULL;
}

#endif