swoole_websocket_server.c

/*
  +----------------------------------------------------------------------+
  | Swoole                                                               |
  +----------------------------------------------------------------------+
  | This source file is subject to version 2.0 of the Apache license,    |
  | that is bundled with this package in the file LICENSE, and is        |
  | available through the world-wide-web at the following url:           |
  | http://www.apache.org/licenses/LICENSE-2.0.html                      |
  | If you did not receive a copy of the Apache2.0 license and are unable|
  | to obtain it through the world-wide-web, please send a note to       |
  | license@swoole.com so we can mail you a copy immediately.            |
  +----------------------------------------------------------------------+
  | Author: Tianfeng Han  <mikan.tenny@gmail.com>                        |
  +----------------------------------------------------------------------+
*/

#include "php_swoole.h"
#include "swoole_http.h"
#ifdef SW_COROUTINE
#include "swoole_coroutine.h"
#endif

#include <ext/standard/url.h>
#include <ext/standard/sha1.h>
#include <ext/standard/php_var.h>
#include <ext/standard/php_string.h>
#include <ext/date/php_date.h>
#include <main/php_variables.h>

#include "websocket.h"
#include "connection.h"
#include "base64.h"
#include "thirdparty/php_http_parser.h"

static zend_class_entry swoole_websocket_server_ce;
static zend_class_entry *swoole_websocket_server_class_entry_ptr;

static zend_class_entry swoole_websocket_frame_ce;
static zend_class_entry *swoole_websocket_frame_class_entry_ptr;

static int websocket_handshake(swListenPort *, http_context *);

static PHP_METHOD(swoole_websocket_server, on);
static PHP_METHOD(swoole_websocket_server, push);
static PHP_METHOD(swoole_websocket_server, exist);
static PHP_METHOD(swoole_websocket_server, isEstablished);
static PHP_METHOD(swoole_websocket_server, pack);
static PHP_METHOD(swoole_websocket_server, unpack);
static PHP_METHOD(swoole_websocket_server, disconnect);

ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_websocket_server_on, 0, 0, 2)
    ZEND_ARG_INFO(0, event_name)
    ZEND_ARG_INFO(0, callback)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_websocket_server_push, 0, 0, 2)
    ZEND_ARG_INFO(0, fd)
    ZEND_ARG_INFO(0, data)
    ZEND_ARG_INFO(0, opcode)
    ZEND_ARG_INFO(0, finish)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_websocket_server_disconnect, 0, 0, 1)
    ZEND_ARG_INFO(0, fd)
    ZEND_ARG_INFO(0, code)
    ZEND_ARG_INFO(0, reason)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_websocket_server_pack, 0, 0, 1)
    ZEND_ARG_INFO(0, data)
    ZEND_ARG_INFO(0, opcode)
    ZEND_ARG_INFO(0, finish)
    ZEND_ARG_INFO(0, mask)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_websocket_server_unpack, 0, 0, 1)
    ZEND_ARG_INFO(0, data)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_websocket_server_exist, 0, 0, 1)
    ZEND_ARG_INFO(0, fd)
ZEND_END_ARG_INFO()

ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_websocket_server_isEstablished, 0, 0, 1)
    ZEND_ARG_INFO(0, fd)
ZEND_END_ARG_INFO()

const zend_function_entry swoole_websocket_server_methods[] =
{
    PHP_ME(swoole_websocket_server, on,         arginfo_swoole_websocket_server_on, ZEND_ACC_PUBLIC)
    PHP_ME(swoole_websocket_server, push,       arginfo_swoole_websocket_server_push, ZEND_ACC_PUBLIC)
    PHP_ME(swoole_websocket_server, disconnect,       arginfo_swoole_websocket_server_disconnect, ZEND_ACC_PUBLIC)
    PHP_ME(swoole_websocket_server, exist,      arginfo_swoole_websocket_server_exist, ZEND_ACC_PUBLIC)
    PHP_ME(swoole_websocket_server, isEstablished,      arginfo_swoole_websocket_server_isEstablished, ZEND_ACC_PUBLIC)
    PHP_ME(swoole_websocket_server, pack,       arginfo_swoole_websocket_server_pack, ZEND_ACC_PUBLIC | ZEND_ACC_STATIC)
    PHP_ME(swoole_websocket_server, unpack,     arginfo_swoole_websocket_server_unpack, ZEND_ACC_PUBLIC | ZEND_ACC_STATIC)
    PHP_FE_END
};

void swoole_websocket_onOpen(http_context *ctx)
{
    int fd = ctx->fd;

    swConnection *conn = swWorker_get_connection(SwooleG.serv, fd);
    if (!conn)
    {
        swoole_error_log(SW_LOG_NOTICE, SW_ERROR_SESSION_CLOSED, "session[%d] is closed.", fd);
        return;
    }
    conn->websocket_status = WEBSOCKET_STATUS_ACTIVE;

    zend_fcall_info_cache *cache = php_swoole_server_get_cache(SwooleG.serv, conn->from_fd, SW_SERVER_CB_onOpen);
    if (cache)
    {
        swServer *serv = SwooleG.serv;
        zval *zserv = (zval *) serv->ptr2;
        zval *zrequest_object = ctx->request.zobject;
        zval *retval = NULL;

        if (SwooleG.enable_coroutine)
        {
            zval *args[2];
            args[0] = zserv;
            args[1] = zrequest_object;

            int ret = coro_create(cache, args, 2, &retval, NULL, NULL);
            if (ret == CORO_LIMIT)
            {
                SwooleG.serv->factory.end(&SwooleG.serv->factory, fd);
                return;
            }
        }
        else
        {
            zval **args[2];
            args[0] = &zserv;
            args[1] = &zrequest_object;

            zval *zcallback = php_swoole_server_get_callback(SwooleG.serv, conn->from_fd, SW_SERVER_CB_onOpen);
            if (sw_call_user_function_fast(zcallback, cache, &retval, 2, args TSRMLS_CC) == FAILURE)
            {
                swoole_php_error(E_WARNING, "onOpen handler error");
            }
        }

        if (EG(exception))
        {
            zend_exception_error(EG(exception), E_ERROR TSRMLS_CC);
        }
        if (retval)
        {
            sw_zval_ptr_dtor(&retval);
        }
    }
}

/**
 * default onRequest callback
 */
void swoole_websocket_onRequest(http_context *ctx)
{
    char *bad_request =
            "HTTP/1.1 400 Bad Request\r\n"\
            "Connection: close\r\n"\
            "Content-Type: text/html; charset=UTF-8\r\n"\
            "Cache-Control: must-revalidate,no-cache,no-store\r\n"\
            "Content-Length: 83\r\n"\
            "Server: "SW_HTTP_SERVER_SOFTWARE"\r\n\r\n"\
            "<html><body><h2>HTTP 400 Bad Request</h2><hr><i>Powered by Swoole</i></body></html>";

    swServer_tcp_send(SwooleG.serv, ctx->fd, bad_request, strlen(bad_request));
    ctx->end = 1;
    swServer_tcp_close(SwooleG.serv, ctx->fd, 0);
    swoole_http_context_free(ctx TSRMLS_CC);
}

void php_swoole_sha1(const char *str, int _len, unsigned char *digest)
{
    PHP_SHA1_CTX context;
    PHP_SHA1Init(&context);
    PHP_SHA1Update(&context, (unsigned char *) str, _len);
    PHP_SHA1Final(digest, &context);
}

static int websocket_handshake(swListenPort *port, http_context *ctx)
{
    zval *header = ctx->request.zheader;
    HashTable *ht = Z_ARRVAL_P(header);
    zval *pData;

    if (sw_zend_hash_find(ht, ZEND_STRS("sec-websocket-key"), (void **) &pData) == FAILURE)
    {
        php_error_docref(NULL TSRMLS_CC, E_WARNING, "header no sec-websocket-key");
        return SW_ERR;
    }
    convert_to_string(pData);

    swString_clear(swoole_http_buffer);
    swString_append_ptr(swoole_http_buffer, ZEND_STRL("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"));

    int n;
    char sec_websocket_accept[128];
    memcpy(sec_websocket_accept, Z_STRVAL_P(pData), Z_STRLEN_P(pData));
    memcpy(sec_websocket_accept + Z_STRLEN_P(pData), SW_WEBSOCKET_GUID, sizeof(SW_WEBSOCKET_GUID) - 1);

    char sha1_str[20];
    bzero(sha1_str, sizeof(sha1_str));
    php_swoole_sha1(sec_websocket_accept, Z_STRLEN_P(pData) + sizeof(SW_WEBSOCKET_GUID) - 1, (unsigned char *) sha1_str);

    char encoded_str[50];
    bzero(encoded_str, sizeof(encoded_str));
    n = swBase64_encode((unsigned char *) sha1_str, sizeof(sha1_str), encoded_str);

    char _buf[128];
    n = snprintf(_buf, sizeof(_buf), "Sec-WebSocket-Accept: %*s\r\n", n, encoded_str);

    swString_append_ptr(swoole_http_buffer, _buf, n);
    swString_append_ptr(swoole_http_buffer, ZEND_STRL("Sec-WebSocket-Version: "SW_WEBSOCKET_VERSION"\r\n"));
    if (port->websocket_subprotocol)
    {
        swString_append_ptr(swoole_http_buffer, ZEND_STRL("Sec-WebSocket-Protocol: "));
        swString_append_ptr(swoole_http_buffer, port->websocket_subprotocol, port->websocket_subprotocol_length);
        swString_append_ptr(swoole_http_buffer, ZEND_STRL("\r\n"));
    }
    swString_append_ptr(swoole_http_buffer, ZEND_STRL("Server: "SW_WEBSOCKET_SERVER_SOFTWARE"\r\n\r\n"));

    swTrace("websocket header len:%ld\n%s \n", swoole_http_buffer->length, swoole_http_buffer->str);

    return swServer_tcp_send(SwooleG.serv, ctx->fd, swoole_http_buffer->str, swoole_http_buffer->length);
}

int swoole_websocket_onMessage(swEventData *req)
{
    int fd = req->info.fd;
    long finish = 0;
    long opcode = 0;
    long close_code = WEBSOCKET_CLOSE_STATUS_ERROR;
    long payload_length = 0;

    zval *zdata;
    SW_MAKE_STD_ZVAL(zdata);

    char frame_header[2];
    php_swoole_get_recv_data(zdata, req, frame_header, SW_WEBSOCKET_HEADER_LEN);

    finish = frame_header[0] & 0x80 ? 1 : 0;
    opcode = frame_header[0] & 0x0F;   // Opcode: low 4 bits of first byte

    if (opcode == WEBSOCKET_OPCODE_CLOSE)
    {
        if (!SwooleG.serv->listen_list->open_websocket_close_frame)
        {
            sw_zval_ptr_dtor(&zdata);
            return SW_OK;
        }
        // WebSocket Close code and reason message
        long length = Z_STRLEN_P(zdata);
        payload_length = frame_header[1] & 0x7F;
        if (payload_length >= SW_WEBSOCKET_CLOSE_CODE_LEN)
        {
            char *data = Z_STRVAL_P(zdata);
            close_code = (data[length - payload_length] << 8) ^ (data[length - payload_length + 1] & 0x00FF);
        }
    }

    zval *zframe;
    SW_MAKE_STD_ZVAL(zframe);
    object_init_ex(zframe, swoole_websocket_frame_class_entry_ptr);

    zend_update_property_long(swoole_websocket_frame_class_entry_ptr, zframe, ZEND_STRL("fd"), fd TSRMLS_CC);
    zend_update_property_bool(swoole_websocket_frame_class_entry_ptr, zframe, ZEND_STRL("finish"), finish TSRMLS_CC);
    zend_update_property_long(swoole_websocket_frame_class_entry_ptr, zframe, ZEND_STRL("opcode"), opcode TSRMLS_CC);
    zend_update_property(swoole_websocket_frame_class_entry_ptr, zframe, ZEND_STRL("data"), zdata TSRMLS_CC);
    
    if (opcode == WEBSOCKET_OPCODE_CLOSE)
    {
        // websocket close frame info
        zend_update_property_long(swoole_websocket_frame_class_entry_ptr, zframe, ZEND_STRL("code"), close_code TSRMLS_CC);
        if (payload_length > 2)
        {
            zend_update_property_stringl(swoole_websocket_frame_class_entry_ptr, zframe, ZEND_STRL("reason"), 
                Z_STRVAL_P(zdata) + Z_STRLEN_P(zdata) - payload_length + SW_WEBSOCKET_CLOSE_CODE_LEN, 
                payload_length - SW_WEBSOCKET_CLOSE_CODE_LEN TSRMLS_CC);
        }
    }

    swServer *serv = SwooleG.serv;
    zval *zserv = (zval *) serv->ptr2;
    zval *retval = NULL;

    if (SwooleG.enable_coroutine)
    {
        zval *args[2];
        args[0] = zserv;
        args[1] = zframe;

        zend_fcall_info_cache *cache = php_swoole_server_get_cache(serv, req->info.from_fd, SW_SERVER_CB_onMessage);
        int ret = coro_create(cache, args, 2, &retval, NULL, NULL);
        if (ret == CORO_LIMIT)
        {
            sw_zval_ptr_dtor(&zdata);
            sw_zval_ptr_dtor(&zframe);
            SwooleG.serv->factory.end(&SwooleG.serv->factory, fd);
            return SW_OK;
        }
    }
    else
    {
        zval **args[2];
        args[0] = &zserv;
        args[1] = &zframe;

        zend_fcall_info_cache *fci_cache = php_swoole_server_get_cache(serv, req->info.from_fd, SW_SERVER_CB_onMessage);
        zval *zcallback = php_swoole_server_get_callback(SwooleG.serv, req->info.from_fd, SW_SERVER_CB_onMessage);

        if (sw_call_user_function_fast(zcallback, fci_cache, &retval, 2, args TSRMLS_CC) == FAILURE)
        {
            swoole_php_error(E_WARNING, "onMessage handler error");
        }
    }

    if (EG(exception))
    {
        zend_exception_error(EG(exception), E_ERROR TSRMLS_CC);
    }
    if (retval)
    {
        sw_zval_ptr_dtor(&retval);
    }
    sw_zval_ptr_dtor(&zdata);
    sw_zval_ptr_dtor(&zframe);
    return SW_OK;
}

int swoole_websocket_onHandshake(swListenPort *port, http_context *ctx)
{
    int fd = ctx->fd;
    int ret = websocket_handshake(port, ctx);
    if (ret == SW_ERR)
    {
        swServer_tcp_close(SwooleG.serv, fd, 1);
    }
    else
    {
        swoole_websocket_onOpen(ctx);
    }

    //free client data
    if (!ctx->end)
    {
        swoole_http_context_free(ctx TSRMLS_CC);
    }

    return SW_OK;
}

void swoole_websocket_init(int module_number TSRMLS_DC)
{
    SWOOLE_INIT_CLASS_ENTRY(swoole_websocket_server_ce, "swoole_websocket_server", "Swoole\\WebSocket\\Server", swoole_websocket_server_methods);
    swoole_websocket_server_class_entry_ptr = sw_zend_register_internal_class_ex(&swoole_websocket_server_ce, swoole_http_server_class_entry_ptr, "swoole_http_server" TSRMLS_CC);
    SWOOLE_CLASS_ALIAS(swoole_websocket_server, "Swoole\\WebSocket\\Server");

    SWOOLE_INIT_CLASS_ENTRY(swoole_websocket_frame_ce, "swoole_websocket_frame", "Swoole\\WebSocket\\Frame", NULL);
    swoole_websocket_frame_class_entry_ptr = zend_register_internal_class(&swoole_websocket_frame_ce TSRMLS_CC);
    SWOOLE_CLASS_ALIAS(swoole_websocket_frame, "Swoole\\WebSocket\\Frame");
    zend_declare_property_long(swoole_websocket_frame_class_entry_ptr, ZEND_STRL("fd"), 0, ZEND_ACC_PUBLIC TSRMLS_CC);
    zend_declare_property_null(swoole_websocket_frame_class_entry_ptr, ZEND_STRL("data"), ZEND_ACC_PUBLIC TSRMLS_CC);
    zend_declare_property_long(swoole_websocket_frame_class_entry_ptr, ZEND_STRL("opcode"), 0, ZEND_ACC_PUBLIC TSRMLS_CC);
    zend_declare_property_bool(swoole_websocket_frame_class_entry_ptr, ZEND_STRL("finish"), 0, ZEND_ACC_PUBLIC TSRMLS_CC);

    if (SWOOLE_G(use_shortname))
    {
        sw_zend_register_class_alias("Co\\WebSocket\\Server", swoole_websocket_server_class_entry_ptr);
        sw_zend_register_class_alias("Co\\WebSocket\\Frame", swoole_websocket_frame_class_entry_ptr);
    }

    // status
    SWOOLE_RAW_DEFINE(WEBSOCKET_STATUS_CONNECTION);
    SWOOLE_RAW_DEFINE(WEBSOCKET_STATUS_HANDSHAKE);
    REGISTER_LONG_CONSTANT("WEBSOCKET_STATUS_FRAME", WEBSOCKET_STATUS_ACTIVE, CONST_CS | CONST_PERSISTENT);
    SWOOLE_RAW_DEFINE(WEBSOCKET_STATUS_ACTIVE);
    SWOOLE_RAW_DEFINE(WEBSOCKET_STATUS_CLOSING);
    // all opcodes
    SWOOLE_RAW_DEFINE(WEBSOCKET_OPCODE_CONTINUATION);
    SWOOLE_RAW_DEFINE(WEBSOCKET_OPCODE_TEXT);
    SWOOLE_RAW_DEFINE(WEBSOCKET_OPCODE_BINARY);
    SWOOLE_RAW_DEFINE(WEBSOCKET_OPCODE_CLOSE);
    SWOOLE_RAW_DEFINE(WEBSOCKET_OPCODE_PING);
    SWOOLE_RAW_DEFINE(WEBSOCKET_OPCODE_PONG);
    // close error
    SWOOLE_RAW_DEFINE(WEBSOCKET_CLOSE_NORMAL);
    SWOOLE_RAW_DEFINE(WEBSOCKET_CLOSE_GOING_AWAY);
    SWOOLE_RAW_DEFINE(WEBSOCKET_CLOSE_PROTOCOL_ERROR);
    SWOOLE_RAW_DEFINE(WEBSOCKET_CLOSE_DATA_ERROR);
    SWOOLE_RAW_DEFINE(WEBSOCKET_CLOSE_STATUS_ERROR);
    SWOOLE_RAW_DEFINE(WEBSOCKET_CLOSE_ABNORMAL);
    SWOOLE_RAW_DEFINE(WEBSOCKET_CLOSE_MESSAGE_ERROR);
    SWOOLE_RAW_DEFINE(WEBSOCKET_CLOSE_POLICY_ERROR);
    SWOOLE_RAW_DEFINE(WEBSOCKET_CLOSE_MESSAGE_TOO_BIG);
    SWOOLE_RAW_DEFINE(WEBSOCKET_CLOSE_EXTENSION_MISSING);
    SWOOLE_RAW_DEFINE(WEBSOCKET_CLOSE_SERVER_ERROR);
    SWOOLE_RAW_DEFINE(WEBSOCKET_CLOSE_TLS);
}

void php_swoole_websocket_unpack(swString *data, zval *zframe TSRMLS_DC)
{
    swWebSocket_frame frame;

    if (data->length < sizeof(frame.header))
    {
        ZVAL_BOOL(zframe, 0);
        return;
    }

    swWebSocket_decode(&frame, data);

    object_init_ex(zframe, swoole_websocket_frame_class_entry_ptr);

    zend_update_property_bool(swoole_websocket_frame_class_entry_ptr, zframe, ZEND_STRL("finish"), frame.header.FIN TSRMLS_CC);
    zend_update_property_long(swoole_websocket_frame_class_entry_ptr, zframe, ZEND_STRL("opcode"), frame.header.OPCODE TSRMLS_CC);
    zend_update_property_stringl(swoole_websocket_frame_class_entry_ptr, zframe, ZEND_STRL("data"), frame.payload,  frame.payload_length TSRMLS_CC);
}

static PHP_METHOD(swoole_websocket_server, on)
{
    zval *callback;
    zval *event_name;

    swServer *serv = swoole_get_object(getThis());
    if (serv->gs->start > 0)
    {
        php_error_docref(NULL TSRMLS_CC, E_WARNING, "can't register event callback function after server started.");
        RETURN_FALSE;
    }

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "zz", &event_name, &callback) == FAILURE)
    {
        return;
    }

    char *func_name = NULL;
    zend_fcall_info_cache *func_cache = emalloc(sizeof(zend_fcall_info_cache));
    if (!sw_zend_is_callable_ex(callback, NULL, 0, &func_name, NULL, func_cache, NULL TSRMLS_CC))
    {
        swoole_php_fatal_error(E_ERROR, "function '%s' is not callable", func_name);
        efree(func_name);
        RETURN_FALSE;
    }
    efree(func_name);

    serv->listen_list->open_websocket_protocol = 1;

    if (strncasecmp("open", Z_STRVAL_P(event_name), Z_STRLEN_P(event_name)) == 0)
    {
        zend_update_property(swoole_websocket_server_class_entry_ptr, getThis(), ZEND_STRL("onOpen"), callback TSRMLS_CC);
        php_sw_server_callbacks[SW_SERVER_CB_onOpen] = sw_zend_read_property(swoole_websocket_server_class_entry_ptr, getThis(), ZEND_STRL("onOpen"), 0 TSRMLS_CC);
        sw_copy_to_stack(php_sw_server_callbacks[SW_SERVER_CB_onOpen], _php_sw_server_callbacks[SW_SERVER_CB_onOpen]);
        php_sw_server_caches[SW_SERVER_CB_onOpen] = func_cache;
    }
    else if (strncasecmp("message", Z_STRVAL_P(event_name), Z_STRLEN_P(event_name)) == 0)
    {
        zend_update_property(swoole_websocket_server_class_entry_ptr, getThis(), ZEND_STRL("onMessage"), callback TSRMLS_CC);
        php_sw_server_callbacks[SW_SERVER_CB_onMessage] = sw_zend_read_property(swoole_websocket_server_class_entry_ptr, getThis(), ZEND_STRL("onMessage"), 0 TSRMLS_CC);
        sw_copy_to_stack(php_sw_server_callbacks[SW_SERVER_CB_onMessage], _php_sw_server_callbacks[SW_SERVER_CB_onMessage]);
        php_sw_server_caches[SW_SERVER_CB_onMessage] = func_cache;
    }
    else
    {
        efree(func_cache);
        zval *obj = getThis();
        sw_zend_call_method_with_2_params(&obj, swoole_http_server_class_entry_ptr, NULL, "on", &return_value, event_name, callback);
    }
}

static PHP_METHOD(swoole_websocket_server, disconnect)
{
    zval *zdata;
    zend_long fd = 0;
    zend_long code = WEBSOCKET_CLOSE_NORMAL;

    char *data;
    int length;

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "l|lz", &fd, &code, &zdata) == FAILURE)
    {
        return;
    }

    if (fd <= 0)
    {
        swoole_php_fatal_error(E_WARNING, "fd[%d] is invalid.", (int)fd);
        RETURN_FALSE;
    }

    if ((code < 3000 && code != 1000) || code >= 5000)
    {
        swoole_php_fatal_error(E_WARNING, "Status code [" ZEND_LONG_FMT "] is invalid.", code);
        RETURN_FALSE;
    }

    length = php_swoole_get_send_data(zdata, &data TSRMLS_CC);

    if (length < 0 || length > SW_WEBSOCKET_CLOSE_REASON_MAX_LEN)
    {
        RETURN_FALSE;
    }

    swConnection *conn = swWorker_get_connection(SwooleG.serv, fd);

    if (!conn || conn->websocket_status < WEBSOCKET_STATUS_HANDSHAKE)
    {
        SwooleG.error = SW_ERROR_WEBSOCKET_BAD_CLIENT;
        RETURN_FALSE;
    }
    swString_clear(swoole_http_buffer);

    char payload_length = length + SW_WEBSOCKET_CLOSE_CODE_LEN;

    swoole_http_buffer->str[0] = 0x88;
    swoole_http_buffer->str[1] = 0x7F & payload_length;     // Avoid bit 1 being set by accident

    memcpy(swoole_http_buffer->str + SW_WEBSOCKET_HEADER_LEN + SW_WEBSOCKET_CLOSE_CODE_LEN, data, length);

    // Encode close code
    swoole_http_buffer->str[2] = (char)((code >> 8 & 0x00FF));
    swoole_http_buffer->str[3] = (char)((code & 0x00FF));

    swoole_http_buffer->length = payload_length + SW_WEBSOCKET_HEADER_LEN;

    int ret = swServer_tcp_send(SwooleG.serv, fd, swoole_http_buffer->str, swoole_http_buffer->length);

    // Change status immediately to avoid double close
    conn->websocket_status = WEBSOCKET_STATUS_CLOSING;

    // Format swEventData
    swEventData req;
    req.info.fd         = fd;
    req.info.len        = swoole_http_buffer->length;
    req.info.from_id    = -1;
    req.info.from_fd    = conn->from_fd;
    memcpy(&req.data, swoole_http_buffer->str, swoole_http_buffer->length);

    // Call onClose
    swoole_websocket_onMessage(&req);

#ifdef SW_COROUTINE
    swServer *serv = SwooleG.serv;
    if (ret < 0 && SwooleG.error == SW_ERROR_OUTPUT_BUFFER_OVERFLOW && serv->send_yield)
    {
        zval _yield_data;
        ZVAL_STRINGL(&_yield_data, swoole_http_buffer->str, swoole_http_buffer->length);
        php_swoole_server_send_yield(serv, fd, &_yield_data, return_value);
    }
    else
#endif
    {
        // Server close connection immediately
        SW_CHECK_RETURN(SwooleG.serv->close(serv, (int)fd, (int)SW_FALSE));
    }
}

static PHP_METHOD(swoole_websocket_server, push)
{
    zval *zdata;
    long fd = 0;
    long opcode = WEBSOCKET_OPCODE_TEXT;
    zend_bool fin = 1;

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "lz|lb", &fd, &zdata, &opcode, &fin) == FAILURE)
    {
        return;
    }

    if (fd <= 0)
    {
        swoole_php_fatal_error(E_WARNING, "fd[%d] is invalid.", (int )fd);
        RETURN_FALSE;
    }

    if (opcode > WEBSOCKET_OPCODE_PONG)
    {
        swoole_php_fatal_error(E_WARNING, "the maximum value of opcode is 10.");
        RETURN_FALSE;
    }

    char *data;
    int length = php_swoole_get_send_data(zdata, &data TSRMLS_CC);

    if (length < 0)
    {
        RETURN_FALSE;
    }

    swConnection *conn = swWorker_get_connection(SwooleG.serv, fd);
    if (!conn || conn->websocket_status < WEBSOCKET_STATUS_HANDSHAKE)
    {
        SwooleG.error = SW_ERROR_WEBSOCKET_BAD_CLIENT;
        swoole_php_fatal_error(E_WARNING, "the connected client of connection[%d] is not a websocket client.", (int ) fd);
        RETURN_FALSE;
    }
    swString_clear(swoole_http_buffer);
    swWebSocket_encode(swoole_http_buffer, data, length, opcode, (int) fin, 0);

    int ret = swServer_tcp_send(SwooleG.serv, fd, swoole_http_buffer->str, swoole_http_buffer->length);
#ifdef SW_COROUTINE
    swServer *serv = SwooleG.serv;
    if (ret < 0 && SwooleG.error == SW_ERROR_OUTPUT_BUFFER_OVERFLOW && serv->send_yield)
    {
        zval _yield_data;
        ZVAL_STRINGL(&_yield_data, swoole_http_buffer->str, swoole_http_buffer->length);
        php_swoole_server_send_yield(serv, fd, &_yield_data, return_value);
    }
    else
#endif
    {
        SW_CHECK_RETURN(ret);
    }
}

static PHP_METHOD(swoole_websocket_server, pack)
{
    char *data;
    zend_size_t length;
    long opcode = WEBSOCKET_OPCODE_TEXT;
    zend_bool finish = 1;
    zend_bool mask = 0;

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "s|lbb", &data, &length, &opcode, &finish, &mask) == FAILURE)
    {
        return;
    }

    if (opcode > WEBSOCKET_OPCODE_PONG)
    {
        swoole_php_fatal_error(E_WARNING, "the maximum value of opcode is 10.");
        RETURN_FALSE;
    }

    swString *buffer = SwooleTG.buffer_stack;
    swString_clear(buffer);
    swWebSocket_encode(buffer, data, length, opcode, (int) finish, mask);
    RETVAL_STRINGL(buffer->str, buffer->length);
}

static PHP_METHOD(swoole_websocket_server, unpack)
{
    swString buffer;
    bzero(&buffer, sizeof(buffer));

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "s", &buffer.str, &buffer.length) == FAILURE)
    {
        return;
    }

    php_swoole_websocket_unpack(&buffer, return_value TSRMLS_CC);
}

static PHP_METHOD(swoole_websocket_server, exist)
{
    zend_long fd;

    swServer *serv = swoole_get_object(getThis());
    if (serv->gs->start == 0)
    {
        php_error_docref(NULL TSRMLS_CC, E_WARNING, "the server is not running.");
        RETURN_FALSE;
    }

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "l", &fd) == FAILURE)
    {
        return;
    }

    swConnection *conn = swWorker_get_connection(serv, fd);
    if (!conn)
    {
        RETURN_FALSE;
    }
    //connection is closed
    if (conn->active == 0 || conn->closed)
    {
        RETURN_FALSE;
    }
    swConnection *server_sock = swServer_connection_get(serv, conn->from_fd);
    if (server_sock)
    {
        swListenPort *port = server_sock->object;
        //not websocket port
        if (port && !port->open_websocket_protocol)
        {
            RETURN_TRUE;
        }
    }
    //have not handshake
    if (conn->websocket_status < WEBSOCKET_STATUS_ACTIVE)
    {
        RETURN_FALSE;
    }
    RETURN_TRUE;
}

static PHP_METHOD(swoole_websocket_server, isEstablished)
{
    zend_long fd;

    swServer *serv = swoole_get_object(getThis());
    if (serv->gs->start == 0)
    {
        php_error_docref(NULL TSRMLS_CC, E_WARNING, "the server is not running.");
        RETURN_FALSE;
    }

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "l", &fd) == FAILURE)
    {
        return;
    }

    swConnection *conn = swWorker_get_connection(serv, fd);
    //not isEstablished
    if (!conn || conn->active == 0 || conn->closed || conn->websocket_status < WEBSOCKET_STATUS_ACTIVE)
    {
        RETURN_FALSE;
    }
    else
    {
        RETURN_TRUE;
    }
}