/* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */

/*

 * soup-websocket-connection.c: This file was originally part of Cockpit.

 *

 * Copyright 2013, 2014 Red Hat, Inc.

 *

 * Cockpit is free software; you can redistribute it and/or modify it

 * under the terms of the GNU Lesser General Public License as published by

 * the Free Software Foundation; either version 2.1 of the License, or

 * (at your option) any later version.

 *

 * Cockpit is distributed in the hope that it will be useful, but

 * WITHOUT ANY WARRANTY; without even the implied warranty of

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU

 * Lesser General Public License for more details.

 *

 * You should have received a copy of the GNU Lesser General Public License

 * along with this library; If not, see <http://www.gnu.org/licenses/>.

 */

#include "config.h"

#include <string.h>

#include "soup-websocket-connection.h"

#include "soup-enum-types.h"

#include "soup-uri.h"

/*

 * SECTION:websocketconnection

 * @title: SoupWebsocketConnection

 * @short_description: A WebSocket connection

 *

 * A #SoupWebsocketConnection is a WebSocket connection to a peer.

 * This API is modeled after the W3C API for interacting with

 * WebSockets.

 *

 * The #SoupWebsocketConnection:state property will indicate the

 * state of the connection.

 *

 * Use soup_websocket_connection_send() to send a message to the peer.

 * When a message is received the #SoupWebsocketConnection::message

 * signal will fire.

 *

 * The soup_websocket_connection_close() function will perform an

 * orderly close of the connection. The

 * #SoupWebsocketConnection::closed signal will fire once the

 * connection closes, whether it was initiated by this side or the

 * peer.

 *

 * Connect to the #SoupWebsocketConnection::closing signal to detect

 * when either peer begins closing the connection.

 */

/**

 * SoupWebsocketConnection:

 *

 * A class representing a WebSocket connection.

 *

 * Since: 2.50

 */

/**

 * SoupWebsocketConnectionClass:

 * @message: default handler for the #SoupWebsocketConnection::message signal

 * @error: default handler for the #SoupWebsocketConnection::error signal

 * @closing: the default handler for the #SoupWebsocketConnection:closing signal

 * @closed: default handler for the #SoupWebsocketConnection::closed signal

 * @pong: default handler for the #SoupWebsocketConnection::pong signal

 *

 * The abstract base class for #SoupWebsocketConnection

 *

 * Since: 2.50

 */

enum {

	PROP_0,

	PROP_IO_STREAM,

	PROP_CONNECTION_TYPE,

	PROP_URI,

	PROP_ORIGIN,

	PROP_PROTOCOL,

	PROP_STATE,

	PROP_MAX_INCOMING_PAYLOAD_SIZE,

	PROP_KEEPALIVE_INTERVAL,

};

enum {

	MESSAGE,

	ERROR,

	CLOSING,

	CLOSED,

	PONG,

	NUM_SIGNALS

};

static guint signals[NUM_SIGNALS] = { 0, };

typedef enum {

	SOUP_WEBSOCKET_QUEUE_NORMAL = 0,

	SOUP_WEBSOCKET_QUEUE_URGENT = 1 << 0,

	SOUP_WEBSOCKET_QUEUE_LAST = 1 << 1,

} SoupWebsocketQueueFlags;

typedef struct {

	GBytes *data;

	gsize sent;

	gsize amount;

	SoupWebsocketQueueFlags flags;

	gboolean pending;

} Frame;

struct _SoupWebsocketConnectionPrivate {

	GIOStream *io_stream;

	SoupWebsocketConnectionType connection_type;

	SoupURI *uri;

	char *origin;

	char *protocol;

	guint64 max_incoming_payload_size;

	guint keepalive_interval;

	gushort peer_close_code;

	char *peer_close_data;

	gboolean close_sent;

	gboolean close_received;

	gboolean dirty_close;

	GSource *close_timeout;

	GMainContext *main_context;

	gboolean io_closing;

	gboolean io_closed;

	GPollableInputStream *input;

	GSource *input_source;

	GByteArray *incoming;

	GPollableOutputStream *output;

	GSource *output_source;

	GQueue outgoing;

	/* Current message being assembled */

	guint8 message_opcode;

	GByteArray *message_data;

	GSource *keepalive_timeout;

};

#define MAX_INCOMING_PAYLOAD_SIZE_DEFAULT   128 * 1024

#define READ_BUFFER_SIZE 1024

G_DEFINE_TYPE_WITH_PRIVATE (SoupWebsocketConnection, soup_websocket_connection, G_TYPE_OBJECT)

static void queue_frame (SoupWebsocketConnection *self, SoupWebsocketQueueFlags flags,

			 gpointer data, gsize len, gsize amount);

static void protocol_error_and_close (SoupWebsocketConnection *self);

static gboolean on_web_socket_input (GObject *pollable_stream,

				     gpointer user_data);

static gboolean on_web_socket_output (GObject *pollable_stream,

				      gpointer user_data);

/* Code below is based on g_utf8_validate() implementation,

 * but handling NULL characters as valid, as expected by

 * WebSockets and compliant with RFC 3629.

 */

#define VALIDATE_BYTE(mask, expect)                             \

        G_STMT_START {                                          \

          if (G_UNLIKELY((*(guchar *)p & (mask)) != (expect)))  \

                  return FALSE;                                 \

        } G_STMT_END

/* see IETF RFC 3629 Section 4 */

static gboolean

utf8_validate (const char *str,

               gsize max_len)

{

        const gchar *p;

        for (p = str; ((p - str) < max_len); p++) {

                if (*(guchar *)p < 128)

                        /* done */;

                else {

                        if (*(guchar *)p < 0xe0) { /* 110xxxxx */

                                if (G_UNLIKELY (max_len - (p - str) < 2))

                                        return FALSE;

                                if (G_UNLIKELY (*(guchar *)p < 0xc2))

                                        return FALSE;

                        } else {

                                if (*(guchar *)p < 0xf0) { /* 1110xxxx */

                                        if (G_UNLIKELY (max_len - (p - str) < 3))

                                                return FALSE;

                                        switch (*(guchar *)p++ & 0x0f) {

                                        case 0:

                                                VALIDATE_BYTE(0xe0, 0xa0); /* 0xa0 ... 0xbf */

                                                break;

                                        case 0x0d:

                                                VALIDATE_BYTE(0xe0, 0x80); /* 0x80 ... 0x9f */

                                                break;

                                        default:

                                                VALIDATE_BYTE(0xc0, 0x80); /* 10xxxxxx */

                                        }

                                } else if (*(guchar *)p < 0xf5) { /* 11110xxx excluding out-of-range */

                                        if (G_UNLIKELY (max_len - (p - str) < 4))

                                                return FALSE;

                                        switch (*(guchar *)p++ & 0x07) {

                                        case 0:

                                                VALIDATE_BYTE(0xc0, 0x80); /* 10xxxxxx */

                                                if (G_UNLIKELY((*(guchar *)p & 0x30) == 0))

                                                        return FALSE;

                                                break;

                                        case 4:

                                                VALIDATE_BYTE(0xf0, 0x80); /* 0x80 ... 0x8f */

                                                break;

                                        default:

                                                VALIDATE_BYTE(0xc0, 0x80); /* 10xxxxxx */

                                        }

                                        p++;

                                        VALIDATE_BYTE(0xc0, 0x80); /* 10xxxxxx */

                                } else {

                                        return FALSE;

                                }

                        }

                        p++;

                        VALIDATE_BYTE(0xc0, 0x80); /* 10xxxxxx */

                }

        }

        return TRUE;

}

#undef VALIDATE_BYTE

static void

frame_free (gpointer data)

{

	Frame *frame = data;

	if (frame) {

		g_bytes_unref (frame->data);

		g_slice_free (Frame, frame);

	}

}

static void

soup_websocket_connection_init (SoupWebsocketConnection *self)

{

	SoupWebsocketConnectionPrivate *pv;

	pv = self->pv = soup_websocket_connection_get_instance_private (self);

	pv->incoming = g_byte_array_sized_new (1024);

	g_queue_init (&pv->outgoing);

	pv->main_context = g_main_context_ref_thread_default ();

}

static void

on_iostream_closed (GObject *source,

                    GAsyncResult *result,

                    gpointer user_data)

{

	SoupWebsocketConnection *self = user_data;

	SoupWebsocketConnectionPrivate *pv = self->pv;

	GError *error = NULL;

	/* We treat connection as closed even if close fails */

	pv->io_closed = TRUE;

	g_io_stream_close_finish (pv->io_stream, result, &error);

	if (error) {

		g_debug ("error closing web socket stream: %s", error->message);

		if (!pv->dirty_close)

			g_signal_emit (self, signals[ERROR], 0, error);

		pv->dirty_close = TRUE;

		g_error_free (error);

	}

	g_assert (soup_websocket_connection_get_state (self) == SOUP_WEBSOCKET_STATE_CLOSED);

	g_debug ("closed: completed io stream close");

	g_signal_emit (self, signals[CLOSED], 0);

	g_object_unref (self);

}

static void

soup_websocket_connection_start_input_source (SoupWebsocketConnection *self)

{

	SoupWebsocketConnectionPrivate *pv = self->pv;

	if (pv->input_source)

		return;

	pv->input_source = g_pollable_input_stream_create_source (pv->input, NULL);

	g_source_set_callback (pv->input_source, (GSourceFunc)on_web_socket_input, self, NULL);

	g_source_attach (pv->input_source, pv->main_context);

}

static void

soup_websocket_connection_stop_input_source (SoupWebsocketConnection *self)

{

	SoupWebsocketConnectionPrivate *pv = self->pv;

	if (pv->input_source) {

		g_debug ("stopping input source");

		g_source_destroy (pv->input_source);

		g_source_unref (pv->input_source);

		pv->input_source = NULL;

	}

}

static void

soup_websocket_connection_start_output_source (SoupWebsocketConnection *self)

{

	SoupWebsocketConnectionPrivate *pv = self->pv;

	if (pv->output_source)

		return;

	pv->output_source = g_pollable_output_stream_create_source (pv->output, NULL);

	g_source_set_callback (pv->output_source, (GSourceFunc)on_web_socket_output, self, NULL);

	g_source_attach (pv->output_source, pv->main_context);

}

static void

soup_websocket_connection_stop_output_source (SoupWebsocketConnection *self)

{

	SoupWebsocketConnectionPrivate *pv = self->pv;

	if (pv->output_source) {

		g_debug ("stopping output source");

		g_source_destroy (pv->output_source);

		g_source_unref (pv->output_source);

		pv->output_source = NULL;

	}

}

static void

keepalive_stop_timeout (SoupWebsocketConnection *self)

{

	SoupWebsocketConnectionPrivate *pv = self->pv;

	if (pv->keepalive_timeout) {

		g_source_destroy (pv->keepalive_timeout);

		g_source_unref (pv->keepalive_timeout);

		pv->keepalive_timeout = NULL;

	}

}

static void

close_io_stop_timeout (SoupWebsocketConnection *self)

{

	SoupWebsocketConnectionPrivate *pv = self->pv;

	if (pv->close_timeout) {

		g_source_destroy (pv->close_timeout);

		g_source_unref (pv->close_timeout);

		pv->close_timeout = NULL;

	}

}

static void

close_io_stream (SoupWebsocketConnection *self)

{

	SoupWebsocketConnectionPrivate *pv = self->pv;

	keepalive_stop_timeout (self);

	close_io_stop_timeout (self);

	if (!pv->io_closing) {

		soup_websocket_connection_stop_input_source (self);

		soup_websocket_connection_stop_output_source (self);

		pv->io_closing = TRUE;

		g_debug ("closing io stream");

		g_io_stream_close_async (pv->io_stream, G_PRIORITY_DEFAULT,

					 NULL, on_iostream_closed, g_object_ref (self));

	}

	g_object_notify (G_OBJECT (self), "state");

}

static void

shutdown_wr_io_stream (SoupWebsocketConnection *self)

{

	SoupWebsocketConnectionPrivate *pv = self->pv;

	GSocket *socket;

	GError *error = NULL;

	soup_websocket_connection_stop_output_source (self);

	if (G_IS_SOCKET_CONNECTION (pv->io_stream)) {

		socket = g_socket_connection_get_socket (G_SOCKET_CONNECTION (pv->io_stream));

		g_socket_shutdown (socket, FALSE, TRUE, &error);

		if (error != NULL) {

			g_debug ("error shutting down io stream: %s", error->message);

			g_error_free (error);

		}

	}

	g_object_notify (G_OBJECT (self), "state");

}

static gboolean

on_timeout_close_io (gpointer user_data)

{

	SoupWebsocketConnection *self = SOUP_WEBSOCKET_CONNECTION (user_data);

	SoupWebsocketConnectionPrivate *pv = self->pv;

	pv->close_timeout = 0;

	g_debug ("peer did not close io when expected");

	close_io_stream (self);

	return FALSE;

}

static void

close_io_after_timeout (SoupWebsocketConnection *self)

{

	SoupWebsocketConnectionPrivate *pv = self->pv;

	const int timeout = 5;

	if (pv->close_timeout)

		return;

	g_debug ("waiting %d seconds for peer to close io", timeout);

	pv->close_timeout = g_timeout_source_new_seconds (timeout);

	g_source_set_callback (pv->close_timeout, on_timeout_close_io, self, NULL);

	g_source_attach (pv->close_timeout, pv->main_context);

}

static void

xor_with_mask (const guint8 *mask,

	       guint8 *data,

	       gsize len)

{

	gsize n;

	/* Do the masking */

	for (n = 0; n < len; n++)

		data[n] ^= mask[n & 3];

}

static void

send_message (SoupWebsocketConnection *self,

	      SoupWebsocketQueueFlags flags,

	      guint8 opcode,

	      const guint8 *data,

	      gsize length)

{

	gsize buffered_amount = length;

	GByteArray *bytes;

	gsize frame_len;

	guint8 *outer;

	guint8 *mask = 0;

	guint8 *at;

	if (!(soup_websocket_connection_get_state (self) == SOUP_WEBSOCKET_STATE_OPEN)) {

		g_debug ("Ignoring message since the connection is closed or is closing");

		return;

	}

	bytes = g_byte_array_sized_new (14 + length);

	outer = bytes->data;

	outer[0] = 0x80 | opcode;

	/* If control message, check payload size */

	if (opcode & 0x08) {

		if (length > 125) {

			g_warning ("WebSocket control message payload exceeds size limit");

			protocol_error_and_close (self);

			return;

		}

		buffered_amount = 0;

	}

	if (length < 126) {

		outer[1] = (0xFF & length); /* mask | 7-bit-len */

		bytes->len = 2;

	} else if (length < 65536) {

		outer[1] = 126; /* mask | 16-bit-len */

		outer[2] = (length >> 8) & 0xFF;

		outer[3] = (length >> 0) & 0xFF;

		bytes->len = 4;

	} else {

		outer[1] = 127; /* mask | 64-bit-len */

#if GLIB_SIZEOF_SIZE_T > 4

		outer[2] = (length >> 56) & 0xFF;

		outer[3] = (length >> 48) & 0xFF;

		outer[4] = (length >> 40) & 0xFF;

		outer[5] = (length >> 32) & 0xFF;

#else

		outer[2] = outer[3] = outer[4] = outer[5] = 0;

#endif

		outer[6] = (length >> 24) & 0xFF;

		outer[7] = (length >> 16) & 0xFF;

		outer[8] = (length >> 8) & 0xFF;

		outer[9] = (length >> 0) & 0xFF;

		bytes->len = 10;

	}

	/* The server side doesn't need to mask, so we don't. There's

	 * probably a client somewhere that's not expecting it.

	 */

	if (self->pv->connection_type == SOUP_WEBSOCKET_CONNECTION_CLIENT) {

		guint32 rnd = g_random_int ();

		outer[1] |= 0x80;

		mask = outer + bytes->len;

		memcpy (mask, &rnd, sizeof (rnd));

		bytes->len += 4;

	}

	at = bytes->data + bytes->len;

	g_byte_array_append (bytes, data, length);

	if (self->pv->connection_type == SOUP_WEBSOCKET_CONNECTION_CLIENT)

		xor_with_mask (mask, at, length);

	frame_len = bytes->len;

	queue_frame (self, flags, g_byte_array_free (bytes, FALSE),

		     frame_len, buffered_amount);

	g_debug ("queued %d frame of len %u", (int)opcode, (guint)frame_len);

}

static void

send_close (SoupWebsocketConnection *self,

	    SoupWebsocketQueueFlags flags,

	    gushort code,

	    const char *reason)

{

	/* Note that send_message truncates as expected */

	char buffer[128];

	gsize len = 0;

	if (code != 0) {

		buffer[len++] = code >> 8;

		buffer[len++] = code & 0xFF;

		if (reason)

			len += g_strlcpy (buffer + len, reason, sizeof (buffer) - len);

	}

	send_message (self, flags, 0x08, (guint8 *)buffer, len);

	self->pv->close_sent = TRUE;

	keepalive_stop_timeout (self);

}

static void

emit_error_and_close (SoupWebsocketConnection *self,

		      GError *error,

		      gboolean prejudice)

{

	gboolean ignore = FALSE;

	gushort code;

	if (soup_websocket_connection_get_state (self) == SOUP_WEBSOCKET_STATE_CLOSED) {

		g_error_free (error);

		return;

	}

	if (error && error->domain == SOUP_WEBSOCKET_ERROR)

		code = error->code;

	else

		code = SOUP_WEBSOCKET_CLOSE_GOING_AWAY;

	self->pv->dirty_close = TRUE;

	g_signal_emit (self, signals[ERROR], 0, error);

	g_error_free (error);

	/* If already closing, just ignore this stuff */

	switch (soup_websocket_connection_get_state (self)) {

	case SOUP_WEBSOCKET_STATE_CLOSED:

		ignore = TRUE;

		break;

	case SOUP_WEBSOCKET_STATE_CLOSING:

		ignore = !prejudice;

		break;

	default:

		break;

	}

	if (ignore) {

		g_debug ("already closing/closed, ignoring error");

	} else if (prejudice) {

		g_debug ("forcing close due to error");

		close_io_stream (self);

	} else {

		g_debug ("requesting close due to error");

		send_close (self, SOUP_WEBSOCKET_QUEUE_URGENT | SOUP_WEBSOCKET_QUEUE_LAST, code, NULL);

	}

}

static void

protocol_error_and_close_full (SoupWebsocketConnection *self,

                               gboolean prejudice)

{

	GError *error;

	error = g_error_new_literal (SOUP_WEBSOCKET_ERROR,

				     SOUP_WEBSOCKET_CLOSE_PROTOCOL_ERROR,

				     self->pv->connection_type == SOUP_WEBSOCKET_CONNECTION_SERVER ?

				     "Received invalid WebSocket response from the client" :

				     "Received invalid WebSocket response from the server");

	emit_error_and_close (self, error, prejudice);

}

static void

protocol_error_and_close (SoupWebsocketConnection *self)

{

	protocol_error_and_close_full (self, FALSE);

}

static void

bad_data_error_and_close (SoupWebsocketConnection *self)

{

	GError *error;

	error = g_error_new_literal (SOUP_WEBSOCKET_ERROR,

				     SOUP_WEBSOCKET_CLOSE_BAD_DATA,

				     self->pv->connection_type == SOUP_WEBSOCKET_CONNECTION_SERVER ?

				     "Received invalid WebSocket data from the client" :

				     "Received invalid WebSocket data from the server");

	emit_error_and_close (self, error, FALSE);

}

static void

too_big_error_and_close (SoupWebsocketConnection *self,

                         guint64 payload_len)

{

	GError *error;

	error = g_error_new_literal (SOUP_WEBSOCKET_ERROR,

				     SOUP_WEBSOCKET_CLOSE_TOO_BIG,

				     self->pv->connection_type == SOUP_WEBSOCKET_CONNECTION_SERVER ?

				     "Received extremely large WebSocket data from the client" :

				     "Received extremely large WebSocket data from the server");

	g_debug ("%s is trying to frame of size %" G_GUINT64_FORMAT " or greater, but max supported size is %" G_GUINT64_FORMAT,

		 self->pv->connection_type == SOUP_WEBSOCKET_CONNECTION_SERVER ? "server" : "client",

	         payload_len, self->pv->max_incoming_payload_size);

	emit_error_and_close (self, error, TRUE);

}

static void

close_connection (SoupWebsocketConnection *self,

                  gushort                  code,

                  const char              *data)

{

	SoupWebsocketQueueFlags flags;

	SoupWebsocketConnectionPrivate *pv;

	pv = self->pv;

	if (pv->close_sent) {

		g_debug ("close code already sent");

		return;

	}

	/* Validate the closing code received by the peer */

	switch (code) {

	case SOUP_WEBSOCKET_CLOSE_NORMAL:

	case SOUP_WEBSOCKET_CLOSE_GOING_AWAY:

	case SOUP_WEBSOCKET_CLOSE_PROTOCOL_ERROR:

	case SOUP_WEBSOCKET_CLOSE_UNSUPPORTED_DATA:

	case SOUP_WEBSOCKET_CLOSE_BAD_DATA:

	case SOUP_WEBSOCKET_CLOSE_POLICY_VIOLATION:

	case SOUP_WEBSOCKET_CLOSE_TOO_BIG:

		break;

	case SOUP_WEBSOCKET_CLOSE_NO_EXTENSION:

		if (pv->connection_type == SOUP_WEBSOCKET_CONNECTION_SERVER) {

			g_debug ("Wrong closing code %d received for a server connection",

			         code);

		}

		break;

	case SOUP_WEBSOCKET_CLOSE_SERVER_ERROR:

		if (pv->connection_type != SOUP_WEBSOCKET_CONNECTION_SERVER) {

			g_debug ("Wrong closing code %d received for a non server connection",

			         code);

		}

		break;

	default:

		if (code < 3000) {

			g_debug ("Wrong closing code %d received", code);

			protocol_error_and_close (self);

			return;

		}

	}

	g_signal_emit (self, signals[CLOSING], 0);

	if (pv->close_received)

		g_debug ("responding to close request");

	flags = 0;

	if (pv->connection_type == SOUP_WEBSOCKET_CONNECTION_SERVER && pv->close_received)

		flags |= SOUP_WEBSOCKET_QUEUE_LAST;

	send_close (self, flags, code, data);

	close_io_after_timeout (self);

}

static void

receive_close (SoupWebsocketConnection *self,

	       const guint8 *data,

	       gsize len)

{

	SoupWebsocketConnectionPrivate *pv = self->pv;

	pv->peer_close_code = 0;

	g_free (pv->peer_close_data);

	pv->peer_close_data = NULL;

	pv->close_received = TRUE;

	switch (len) {

	case 0:

		/* Send a clean close when having an empty payload */

		close_connection (self, 1000, NULL);

		return;

	case 1:

		/* Send a protocol error since the close code is incomplete */

		protocol_error_and_close (self);

		return;

	default:

		/* Store the code/data payload */

		pv->peer_close_code = (guint16)data[0] << 8 | data[1];

		break;

	}

	if (len > 2) {

		data += 2;

		len -= 2;

		if (!utf8_validate ((const char *)data, len)) {

			g_debug ("received non-UTF8 close data: %d '%.*s' %d", (int)len, (int)len, (char *)data, (int)data[0]);

			protocol_error_and_close (self);

			return;

		}

		pv->peer_close_data = g_strndup ((char *)data, len);

	}

	/* Once we receive close response on server, close immediately */

	if (pv->close_sent) {

		shutdown_wr_io_stream (self);

		if (pv->connection_type == SOUP_WEBSOCKET_CONNECTION_SERVER)

			close_io_stream (self);

	} else {

		close_connection (self, pv->peer_close_code, pv->peer_close_data);

	}

}

static void

receive_ping (SoupWebsocketConnection *self,

                      const guint8 *data,

                      gsize len)

{

	/* Send back a pong with same data */

	g_debug ("received ping, responding");

	send_message (self, SOUP_WEBSOCKET_QUEUE_URGENT, 0x0A, data, len);

}

static void

receive_pong (SoupWebsocketConnection *self,

                      const guint8 *data,

                      gsize len)

{

	GByteArray *bytes;

	g_debug ("received pong message");

	bytes = g_byte_array_sized_new (len + 1);

	g_byte_array_append (bytes, data, len);

	/* Always null terminate, as a convenience */

	g_byte_array_append (bytes, (guchar *)"\0", 1);

	/* But don't include the null terminator in the byte count */

	bytes->len--;

	g_signal_emit (self, signals[PONG], 0, bytes);

	g_byte_array_unref (bytes);

}

static void

process_contents (SoupWebsocketConnection *self,

		  gboolean control,

		  gboolean fin,

		  guint8 opcode,

		  gconstpointer payload,

		  gsize payload_len)

{

	SoupWebsocketConnectionPrivate *pv = self->pv;

	GBytes *message;

	if (pv->close_sent && pv->close_received)

		return;

	if (control) {

		/* Control frames must never be fragmented */

		if (!fin) {

			g_debug ("received fragmented control frame");

			protocol_error_and_close (self);

			return;

		}

		g_debug ("received control frame %d with %d payload", (int)opcode, (int)payload_len);

		switch (opcode) {

		case 0x08:

			receive_close (self, payload, payload_len);

			break;

		case 0x09:

			receive_ping (self, payload, payload_len);

			break;

		case 0x0A:

			receive_pong (self, payload, payload_len);

			break;

		default:

			g_debug ("received unsupported control frame: %d", (int)opcode);

			protocol_error_and_close (self);

			return;

		}

	} else if (pv->close_received) {

		g_debug ("received message after close was received");

	} else {

		/* A message frame */

		if (!fin && opcode) {

			/* Initial fragment of a message */

			if (pv->message_data) {

				g_debug ("received out of order initial message fragment");

				protocol_error_and_close (self);

				return;

			}

			g_debug ("received initial fragment frame %d with %d payload", (int)opcode, (int)payload_len);

		} else if (!fin && !opcode) {

			/* Middle fragment of a message */

			if (!pv->message_data) {

				g_debug ("received out of order middle message fragment");

				protocol_error_and_close (self);

				return;

			}

			g_debug ("received middle fragment frame with %d payload", (int)payload_len);

		} else if (fin && !opcode) {

			/* Last fragment of a message */

			if (!pv->message_data) {

				g_debug ("received out of order ending message fragment");

				protocol_error_and_close (self);

				return;

			}

			g_debug ("received last fragment frame with %d payload", (int)payload_len);

		} else {

			/* An unfragmented message */

			g_assert (opcode != 0);

			if (pv->message_data) {

				g_debug ("received unfragmented message when fragment was expected");

				protocol_error_and_close (self);

				return;

			}

			g_debug ("received frame %d with %d payload", (int)opcode, (int)payload_len);

		}

		if (opcode) {

			pv->message_opcode = opcode;

			pv->message_data = g_byte_array_sized_new (payload_len + 1);

		}

		switch (pv->message_opcode) {

		case 0x01:

		case 0x02:

			g_byte_array_append (pv->message_data, payload, payload_len);

			break;

		default:

			g_debug ("received unknown data frame: %d", (int)opcode);

			protocol_error_and_close (self);

			return;

		}