Partydeck/Websocket-Extension
3
0
Fork 0
mirror of https://github.com/defold/extension-websocket.git synced 2025-06-27 17:57:43 +02:00
Code Issues Projects Releases Wiki Activity

Compare commits

base: Partydeck:master
Branches Tags
Partydeck:master
Partydeck:initial
Partydeck:4.1.0
Partydeck:4.0.0
Partydeck:3.2.0
Partydeck:3.1.0
Partydeck:3.0.0
Partydeck:2.2.3
Partydeck:2.2.2
Partydeck:2.2.1
Partydeck:2.2.0
Partydeck:2.1.0
Partydeck:2.0.0
Partydeck:1.6.0
Partydeck:1.5.2
Partydeck:1.5.1
Partydeck:1.5.0
Partydeck:1.4.0
Partydeck:1.3.0
Partydeck:1.2.0
Partydeck:1.1.0
Partydeck:1.0.0
..
compare: Partydeck:1.5.2
Branches Tags
Partydeck:master
Partydeck:initial
Partydeck:4.1.0
Partydeck:4.0.0
Partydeck:3.2.0
Partydeck:3.1.0
Partydeck:3.0.0
Partydeck:2.2.3
Partydeck:2.2.2
Partydeck:2.2.1
Partydeck:2.2.0
Partydeck:2.1.0
Partydeck:2.0.0
Partydeck:1.6.0
Partydeck:1.5.2
Partydeck:1.5.1
Partydeck:1.5.0
Partydeck:1.4.0
Partydeck:1.3.0
Partydeck:1.2.0
Partydeck:1.1.0
Partydeck:1.0.0

No commits in common. "master" and "1.5.2" have entirely different histories.
master ... 1.5.2

15 changed files with 5816 additions and 547 deletions
Show Stats Expand all files Collapse all files

76
.github/workflows/bob.yml vendored
View File

@ -1,76 +0,0 @@
name: Build with bob
on:
push:
pull_request_target:
schedule:
# nightly at 05:00 on the 1st and 15th
- cron: 0 5 1,15 * *
env:
VERSION_FILENAME: 'info.json'
BUILD_SERVER: 'https://build.defold.com'
jobs:
build_with_bob:
strategy:
matrix:
platform: [armv7-android, x86_64-linux, x86_64-win32, x86-win32, js-web]
runs-on: ubuntu-latest
name: Build
steps:
- uses: actions/checkout@v2
- uses: actions/setup-java@v1
with:
java-version: '11.0.2'
- name: Get Defold version
run: |
TMPVAR=`curl -s http://d.defold.com/stable/${{env.VERSION_FILENAME}} | jq -r '.sha1'`
echo "DEFOLD_VERSION=${TMPVAR}" >> $GITHUB_ENV
echo "Found version ${TMPVAR}"
- name: Download bob.jar
run: |
wget -q http://d.defold.com/archive/stable/${{env.DEFOLD_VERSION}}/bob/bob.jar
java -jar bob.jar --version
- name: Resolve libraries
run: java -jar bob.jar resolve --email a@b.com --auth 123456
- name: Build
run: java -jar bob.jar --platform=${{ matrix.platform }} build --archive --build-server=${{env.BUILD_SERVER}}
- name: Bundle
run: java -jar bob.jar --platform=${{ matrix.platform }} bundle
# macOS is not technically needed for building, but we want to test bundling as well, since we're also testing the manifest merging
build_with_bob_macos:
strategy:
matrix:
platform: [armv7-darwin, x86_64-darwin]
runs-on: macOS-latest
name: Build
steps:
- uses: actions/checkout@v2
- uses: actions/setup-java@v1
with:
java-version: '11.0.2'
- name: Get Defold version
run: |
TMPVAR=`curl -s http://d.defold.com/stable/${{env.VERSION_FILENAME}} | jq -r '.sha1'`
echo "DEFOLD_VERSION=${TMPVAR}" >> $GITHUB_ENV
echo "Found version ${TMPVAR}"
- name: Download bob.jar
run: |
wget -q http://d.defold.com/archive/stable/${{env.DEFOLD_VERSION}}/bob/bob.jar
java -jar bob.jar --version
- name: Resolve libraries
run: java -jar bob.jar resolve --email a@b.com --auth 123456
- name: Build
run: java -jar bob.jar --platform=${{ matrix.platform }} build --archive --build-server=${{env.BUILD_SERVER}}
- name: Bundle
run: java -jar bob.jar --platform=${{ matrix.platform }} bundle

5
.github/workflows/trigger-site-rebuild.yml vendored
View File

@ -1,9 +1,6 @@
name: Trigger site rebuild
on:
push:
branches:
- 'master'
on: [push]
jobs:
site-rebuild:

2
.gitignore vendored
View File

@ -12,5 +12,3 @@ builtins
lws_source
lws_build
*.profraw
*.der
/.editor_settings

17
README.md
View File

@ -1,7 +1,5 @@
# Defold websocket extension
[![Build Status](https://github.com/defold/extension-websocket/workflows/Build%20with%20bob/badge.svg)](https://github.com/defold/extension-websocket/actions)
## Installation
To use this library in your Defold project, add the following URL to your `game.project` dependencies:
@ -11,18 +9,11 @@ We recommend using a link to a zip file of a [specific release](https://github.c
## API reference
https://defold.com/extension-websocket/
https://defold.com/extension-websocket/websocket_api/
https://defold.com/extension-websocket/api/
## Debugging
In order to make it easier to debug this extension, we provide a `game.project` setting `websocket.debug` (edit `game.project` as text and add):
```
[websocket]
debug = level
```
In order to make it easier to debug this extension, we provide a `game.project` setting `websocket.debug`.
Set it to:
@ -55,7 +46,3 @@ For command line debugging, there's
This extension makes use of the C library WSlay by @tatsuhiro-t:
* https://github.com/tatsuhiro-t/wslay
The test server used by the example:
* https://www.lob.com/blog/websocket-org-is-down-here-is-an-alternative

5523
docs/Defold-Websocket/DefoldWebsocket.symbols Normal file
View File

File diff suppressed because it is too large Load Diff

32
docs/index.md
View File

@ -14,23 +14,16 @@ Here is how you connect to a websocket and listen to events:
```lua
local function websocket_callback(self, conn, data)
if data.event == websocket.EVENT_DISCONNECTED then
print("Disconnected: " .. tostring(conn))
log("Disconnected: " .. tostring(conn))
self.connection = nil
update_gui(self)
elseif data.event == websocket.EVENT_CONNECTED then
update_gui(self)
print("Connected: " .. tostring(conn))
log("Connected: " .. tostring(conn))
elseif data.event == websocket.EVENT_ERROR then
print("Error: '" .. tostring(data.message) .. "'")
if data.handshake_response then
print("Handshake response status: '" .. tostring(data.handshake_response.status) .. "'")
for key, value in pairs(data.handshake_response.headers) do
log("Handshake response header: '" .. key .. ": " .. value .. "'")
end
print("Handshake response body: '" .. tostring(data.handshake_response.response) .. "'")
end
log("Error: '" .. data.error .. "'")
elseif data.event == websocket.EVENT_MESSAGE then
print("Receiving: '" .. tostring(data.message) .. "'")
log("Receiving: '" .. tostring(data.message) .. "'")
end
end
@ -56,19 +49,10 @@ https://github.com/defold/extension-websocket/archive/master.zip
We recommend using a link to a zip file of a [specific release](https://github.com/defold/extension-websocket/releases).
## Configuration
The following configuration options can be set in the game.project file:
```
[websocket]
debug = 1
socket_timeout = 10000000
```
* `debug` - Log level where 0 means no logs and higher values shows more logs (currently 1 or 2).
* `socket_timeout` - Timeout for the underlying socket connection. In microseconds.
## Source code
The source code is available on [GitHub](https://github.com/defold/extension-websocket)
## API reference
https://defold.com/extension-websocket/api/

24
examples/websocket.gui_script
View File

@ -1,4 +1,4 @@
local URL = "echo.websocket.events"
local URL="://echo.websocket.org"
local function click_button(node, action)
return gui.is_enabled(node) and action.pressed and gui.pick_node(node, action.x, action.y)
@ -33,6 +33,12 @@ local function log(...)
gui.set_text(node, gui.get_text(node) .. "\n" .. text)
end
local function http_result(self, _, response)
print(response.status)
--print(response.response)
pprint(response.headers)
end
function init(self)
msg.post(".", "acquire_input_focus")
msg.post("@render:", "clear_color", { color = vmath.vector4(0.2, 0.4, 0.8, 1.0) })
@ -43,6 +49,9 @@ function init(self)
self.connection_text = gui.get_node("connection_text")
self.connection = nil
update_gui(self)
--http.request("https://defold.com", "GET", http_result)
end
function final(self)
@ -57,21 +66,14 @@ end
local function websocket_callback(self, conn, data)
if data.event == websocket.EVENT_DISCONNECTED then
log("Disconnected: " .. tostring(conn) .. " Code: " .. data.code .. " Message: " .. tostring(data.message))
log("Disconnected: " .. tostring(conn))
self.connection = nil
update_gui(self)
elseif data.event == websocket.EVENT_CONNECTED then
update_gui(self)
log("Connected: " .. tostring(conn))
elseif data.event == websocket.EVENT_ERROR then
log("Error: '" .. tostring(data.message) .. "'")
if data.handshake_response then
log("Handshake response status: '" .. tostring(data.handshake_response.status) .. "'")
for key, value in pairs(data.handshake_response.headers) do
log("Handshake response header: '" .. key .. ": " .. value .. "'")
end
log("Handshake response body: '" .. tostring(data.handshake_response.response) .. "'")
end
log("Error: '" .. data.error .. "'")
elseif data.event == websocket.EVENT_MESSAGE then
log("Receiving: '" .. tostring(data.message) .. "'")
end
@ -80,7 +82,7 @@ end
local function connect(self, scheme)
local params = {}
self.url = scheme .. "://" .. URL
self.url = scheme .. URL
log("Connecting to " .. self.url)
self.connection = websocket.connect(self.url, params, websocket_callback)
end

1
game.project
View File

@ -14,3 +14,4 @@ _dependencies = https://github.com/GameAnalytics/defold-openssl/archive/1.0.0.zi
[library]
include_dirs = websocket

10
websocket/api/api.script_api
View File

@ -61,14 +61,6 @@
type: string
desc: The received data if event is `websocket.EVENT_MESSAGE`. Error message otherwise
- name: handshake_response
type: table
desc: Handshake response information (status, headers etc)
- name: code
type: number
desc: Status code received from the server if the server closed the connection. Only present if event is `EVENT_DISCONNECTED`.
returns:
- name: connection
@ -94,7 +86,7 @@
end
function init(self)
self.url = "ws://echo.websocket.events"
self.url = "ws://echo.websocket.org"
local params = {
timeout = 3000,
headers = "Sec-WebSocket-Protocol: chat\r\nOrigin: mydomain.com\r\n"

4
websocket/ext.manifest
View File

@ -6,7 +6,3 @@ platforms:
context:
includes: ["upload/websocket/include/wslay"]
defines: ["HAVE_CONFIG_H"]
web:
context:
linkFlags: ["-lwebsocket.js"]

42
websocket/src/emscripten_callbacks.cpp
View File

@ -1,42 +0,0 @@
#include "websocket.h"
#if defined(__EMSCRIPTEN__)
namespace dmWebsocket
{
EM_BOOL Emscripten_WebSocketOnOpen(int eventType, const EmscriptenWebSocketOpenEvent *websocketEvent, void *userData) {
DebugLog(1, "WebSocket OnOpen");
WebsocketConnection* conn = (WebsocketConnection*)userData;
SetState(conn, STATE_CONNECTED);
HandleCallback(conn, EVENT_CONNECTED, 0, 0);
return EM_TRUE;
}
EM_BOOL Emscripten_WebSocketOnError(int eventType, const EmscriptenWebSocketErrorEvent *websocketEvent, void *userData) {
DebugLog(1, "WebSocket OnError");
WebsocketConnection* conn = (WebsocketConnection*)userData;
conn->m_Status = RESULT_ERROR;
SetState(conn, STATE_DISCONNECTED);
return EM_TRUE;
}
EM_BOOL Emscripten_WebSocketOnClose(int eventType, const EmscriptenWebSocketCloseEvent *websocketEvent, void *userData) {
DebugLog(1, "WebSocket OnClose");
WebsocketConnection* conn = (WebsocketConnection*)userData;
int length = strlen(websocketEvent->reason);
PushMessage(conn, MESSAGE_TYPE_CLOSE, length, (uint8_t*)websocketEvent->reason, websocketEvent->code);
return EM_TRUE;
}
EM_BOOL Emscripten_WebSocketOnMessage(int eventType, const EmscriptenWebSocketMessageEvent *websocketEvent, void *userData) {
DebugLog(1, "WebSocket OnMessage");
WebsocketConnection* conn = (WebsocketConnection*)userData;
int length = websocketEvent->numBytes;
if (websocketEvent->isText)
{
length--;
}
PushMessage(conn, MESSAGE_TYPE_NORMAL, length, websocketEvent->data, 0);
return EM_TRUE;
}
} // namespace
#endif // __EMSCRIPTEN__

138
websocket/src/handshake.cpp
View File

@ -1,6 +1,5 @@
#include "websocket.h"
#include <dmsdk/dlib/socket.h>
#include <dmsdk/dlib/http_client.h>
#include <ctype.h> // tolower
namespace dmWebsocket
@ -171,37 +170,66 @@ Result ReceiveHeaders(WebsocketConnection* conn)
}
#endif
static void HandleVersion(void* user_data, int major, int minor, int status, const char* status_str)
static int dmStriCmp(const char* s1, const char* s2)
{
HandshakeResponse* response = (HandshakeResponse*)user_data;
response->m_HttpMajor = major;
response->m_HttpMinor = minor;
response->m_ResponseStatusCode = status;
}
static void HandleHeader(void* user_data, const char* key, const char* value)
{
HandshakeResponse* response = (HandshakeResponse*)user_data;
if (response->m_Headers.Remaining() == 0)
for (;;)
{
response->m_Headers.OffsetCapacity(4);
if (!*s1 || !*s2 || tolower((unsigned char) *s1) != tolower((unsigned char) *s2))
{
return (unsigned char) *s1 - (unsigned char) *s2;
}
s1++;
s2++;
}
HttpHeader* new_header = new HttpHeader(key, value);
response->m_Headers.Push(new_header);
}
static void HandleContent(void* user_data, int offset)
#if defined(__EMSCRIPTEN__)
Result VerifyHeaders(WebsocketConnection* conn)
{
HandshakeResponse* response = (HandshakeResponse*)user_data;
response->m_BodyOffset = offset;
return RESULT_OK;
}
bool ValidateSecretKey(WebsocketConnection* conn, const char* server_key)
#else
Result VerifyHeaders(WebsocketConnection* conn)
{
char* r = conn->m_Buffer;
// According to protocol, the response should start with "HTTP/1.1 <statuscode> <message>"
const char* http_version_and_status_protocol = "HTTP/1.1 101";
if (strstr(r, http_version_and_status_protocol) != r) {
return SetStatus(conn, RESULT_HANDSHAKE_FAILED, "Missing: '%s' in header", http_version_and_status_protocol);
}
const char* endtag = strstr(conn->m_Buffer, "\r\n\r\n");
r = strstr(r, "\r\n") + 2;
bool connection = false;
bool upgrade = false;
bool valid_key = false;
// parse the headers in place
while (r < endtag)
{
// Tokenize the each header line: "Key: Value\r\n"
const char* key = r;
r = strchr(r, ':');
*r = 0;
++r;
const char* value = r;
while(*value == ' ')
++value;
r = strstr(r, "\r\n");
*r = 0;
r += 2;
// Page 18 in https://tools.ietf.org/html/rfc6455#section-11.3.3
if (dmStriCmp(key, "Connection") == 0 && dmStriCmp(value, "Upgrade") == 0)
connection = true;
else if (dmStriCmp(key, "Upgrade") == 0 && dmStriCmp(value, "websocket") == 0)
upgrade = true;
else if (dmStriCmp(key, "Sec-WebSocket-Accept") == 0)
{
uint8_t client_key[32 + 40];
uint32_t client_key_len = sizeof(client_key);
dmCrypt::Base64Encode(conn->m_Key, sizeof(conn->m_Key), client_key, &client_key_len);
@ -223,48 +251,24 @@ bool ValidateSecretKey(WebsocketConnection* conn, const char* server_key)
client_key_len = sizeof(client_key);
dmCrypt::Base64Encode(client_key_sha1, sizeof(client_key_sha1), client_key, &client_key_len);
client_key[client_key_len] = 0;
DebugLog(2, "Client key (base64): %s", client_key);
DebugLog(2, "Server key (base64): %s", server_key);
return strcmp(server_key, (const char*)client_key) == 0;
}
DebugLog(2, "Server key (base64): %s", value);
#if defined(__EMSCRIPTEN__)
Result VerifyHeaders(WebsocketConnection* conn)
{
return RESULT_OK;
}
#else
Result VerifyHeaders(WebsocketConnection* conn)
{
char* r = conn->m_Buffer;
// Find start of payload now because dmHttpClient::ParseHeader is destructive
const char* start_of_payload = strstr(conn->m_Buffer, "\r\n\r\n");
start_of_payload += 4;
HandshakeResponse* response = new HandshakeResponse();
conn->m_HandshakeResponse = response;
dmHttpClient::ParseResult parse_result = dmHttpClient::ParseHeader(r, response, true, &HandleVersion, &HandleHeader, &HandleContent);
if (parse_result != dmHttpClient::ParseResult::PARSE_RESULT_OK)
{
return SetStatus(conn, RESULT_HANDSHAKE_FAILED, "Failed to parse handshake response. 'dmHttpClient::ParseResult=%i'", parse_result);
if (strcmp(value, (const char*)client_key) == 0)
valid_key = true;
}
}
if (response->m_ResponseStatusCode != 101) {
return SetStatus(conn, RESULT_HANDSHAKE_FAILED, "Wrong response status: %i", response->m_ResponseStatusCode);
}
// The response might contain both the headers, but also (if successful) the first batch of data
endtag += 4;
uint32_t size = conn->m_BufferSize - (endtag - conn->m_Buffer);
conn->m_BufferSize = size;
memmove(conn->m_Buffer, endtag, size);
conn->m_Buffer[size] = 0;
conn->m_HasHandshakeData = conn->m_BufferSize != 0 ? 1 : 0;
HttpHeader *connection_header, *upgrade_header, *websocket_secret_header;
connection_header = response->GetHeader("Connection");
upgrade_header = response->GetHeader("Upgrade");
websocket_secret_header = response->GetHeader("Sec-WebSocket-Accept");
bool connection = connection_header && dmStrCaseCmp(connection_header->m_Value, "Upgrade") == 0;
bool upgrade = upgrade_header && dmStrCaseCmp(upgrade_header->m_Value, "websocket") == 0;
bool valid_key = websocket_secret_header && ValidateSecretKey(conn, websocket_secret_header->m_Value);
// Send error to lua?
if (!connection)
dmLogError("Failed to find the Connection keyword in the response headers");
if (!upgrade)
@ -273,21 +277,11 @@ Result VerifyHeaders(WebsocketConnection* conn)
dmLogError("Failed to find valid key in the response headers");
bool ok = connection && upgrade && valid_key;
if(!ok)
{
return RESULT_HANDSHAKE_FAILED;
if (!ok) {
dmLogError("Response:\n\"%s\"\n", conn->m_Buffer);
}
delete conn->m_HandshakeResponse;
conn->m_HandshakeResponse = 0;
// The response might contain both the headers, but also (if successful) the first batch of data
uint32_t size = conn->m_BufferSize - (start_of_payload - conn->m_Buffer);
conn->m_BufferSize = size;
memmove(conn->m_Buffer, start_of_payload, size);
conn->m_Buffer[size] = 0;
conn->m_HasHandshakeData = conn->m_BufferSize != 0 ? 1 : 0;
return RESULT_OK;
return ok ? RESULT_OK : RESULT_HANDSHAKE_FAILED;
}
#endif

357
websocket/src/websocket.cpp
View File

@ -7,13 +7,12 @@
#include "websocket.h"
#include "script_util.h"
#include <dmsdk/dlib/connection_pool.h>
#include <dmsdk/dlib/thread.h>
#include <dmsdk/dlib/dns.h>
#include <dmsdk/dlib/sslsocket.h>
#include <ctype.h> // isprint et al
#if defined(__EMSCRIPTEN__)
#include <emscripten/emscripten.h> // for EM_ASM
#include <emscripten/websocket.h>
#endif
#if defined(WIN32)
@ -31,10 +30,14 @@ struct WebsocketContext
int m_Timeout;
dmArray<WebsocketConnection*> m_Connections;
dmConnectionPool::HPool m_Pool;
dmDNS::HChannel m_Channel;
uint32_t m_Initialized:1;
} g_Websocket;
static void HandleCallback(WebsocketConnection* conn, int event);
#define STRING_CASE(_X) case _X: return #_X;
const char* ResultToString(Result err)
@ -53,12 +56,10 @@ const char* ResultToString(Result err)
const char* StateToString(State err)
{
switch(err) {
STRING_CASE(STATE_CREATE);
STRING_CASE(STATE_CONNECTING);
STRING_CASE(STATE_HANDSHAKE_WRITE);
STRING_CASE(STATE_HANDSHAKE_READ);
STRING_CASE(STATE_CONNECTED);
STRING_CASE(STATE_DISCONNECTING);
STRING_CASE(STATE_DISCONNECTED);
default: return "Unknown error";
};
@ -138,7 +139,7 @@ void DebugPrint(int level, const char* msg, const void* _bytes, uint32_t num_byt
CloseConnection(conn);
void SetState(WebsocketConnection* conn, State state)
static void SetState(WebsocketConnection* conn, State state)
{
State prev_state = conn->m_State;
if (prev_state != state)
@ -168,6 +169,8 @@ Result SetStatus(WebsocketConnection* conn, Result status, const char* format, .
// ***************************************************************************************************
// LUA functions
static WebsocketConnection* CreateConnection(const char* url)
{
WebsocketConnection* conn = new WebsocketConnection;
@ -177,13 +180,14 @@ static WebsocketConnection* CreateConnection(const char* url)
conn->m_BufferSize = 0;
conn->m_ConnectTimeout = 0;
dmURI::Parts uri;
dmURI::Parse(url, &conn->m_Url);
if (strcmp(conn->m_Url.m_Scheme, "https") == 0)
strcpy(conn->m_Url.m_Scheme, "wss");
conn->m_SSL = strcmp(conn->m_Url.m_Scheme, "wss") == 0 ? 1 : 0;
conn->m_State = STATE_CREATE;
conn->m_State = STATE_CONNECTING;
conn->m_Callback = 0;
conn->m_Connection = 0;
@ -191,15 +195,11 @@ static WebsocketConnection* CreateConnection(const char* url)
conn->m_SSLSocket = 0;
conn->m_Status = RESULT_OK;
conn->m_HasHandshakeData = 0;
conn->m_HandshakeResponse = 0;
conn->m_ConnectionThread = 0;
conn->m_WasConnected = 0;
#if defined(HAVE_WSLAY)
conn->m_Ctx = 0;
#endif
#if defined(__EMSCRIPTEN__)
conn->m_WS = 0;
#endif
return conn;
}
@ -218,27 +218,18 @@ static void DestroyConnection(WebsocketConnection* conn)
dmScript::DestroyCallback(conn->m_Callback);
#if defined(__EMSCRIPTEN__)
if (conn->m_WS)
{
emscripten_websocket_delete(conn->m_WS);
if (conn->m_Socket != dmSocket::INVALID_SOCKET_HANDLE) {
// We would normally do a shutdown() first, but Emscripten returns ENOSYS
//dmSocket::Shutdown(conn->m_Socket, dmSocket::SHUTDOWNTYPE_READWRITE);
dmSocket::Delete(conn->m_Socket);
}
#else
if (conn->m_Connection)
dmConnectionPool::Close(g_Websocket.m_Pool, conn->m_Connection);
dmConnectionPool::Return(g_Websocket.m_Pool, conn->m_Connection);
#endif
if (conn->m_HandshakeResponse)
delete conn->m_HandshakeResponse;
free((void*)conn->m_Buffer);
if (conn->m_ConnectionThread)
{
dmThread::Join(conn->m_ConnectionThread);
conn->m_ConnectionThread = 0;
}
delete conn;
DebugLog(2, "DestroyConnection: %p", conn);
}
@ -246,37 +237,26 @@ static void DestroyConnection(WebsocketConnection* conn)
static void CloseConnection(WebsocketConnection* conn)
{
State prev_state = conn->m_State;
// we want it to send this message in the polling
if (conn->m_State == STATE_CONNECTED) {
#if defined(HAVE_WSLAY)
WSL_Close(conn->m_Ctx);
#else
// start disconnecting by closing the WebSocket through the JS API
// we transition to the DISCONNECTED state when we receive the
// Emscripten callback that the connection has closed
emscripten_websocket_close(conn->m_WS, 1000, "CloseConnection");
SetState(conn, STATE_DISCONNECTING);
#endif
}
#if defined(HAVE_WSLAY)
// close the connection and immediately transition to the DISCONNECTED
// state
SetState(conn, STATE_DISCONNECTED);
#endif
}
static bool IsConnectionValid(WebsocketConnection* conn)
static int FindConnection(WebsocketConnection* conn)
{
if (conn)
{
for (int i = 0; i < g_Websocket.m_Connections.Size(); ++i )
{
if (g_Websocket.m_Connections[i] == conn)
return true;
return i;
}
}
return false;
return -1;
}
/*#
@ -329,7 +309,8 @@ static int LuaDisconnect(lua_State* L)
WebsocketConnection* conn = (WebsocketConnection*)lua_touserdata(L, 1);
if (IsConnectionValid(conn))
int i = FindConnection(conn);
if (i != -1)
{
CloseConnection(conn);
}
@ -348,7 +329,8 @@ static int LuaSend(lua_State* L)
WebsocketConnection* conn = (WebsocketConnection*)lua_touserdata(L, 1);
if (!IsConnectionValid(conn))
int i = FindConnection(conn);
if (i == -1)
return DM_LUA_ERROR("Invalid connection");
if (conn->m_State != STATE_CONNECTED)
@ -367,17 +349,9 @@ static int LuaSend(lua_State* L)
wslay_event_queue_msg(conn->m_Ctx, &msg); // it makes a copy of the data
#else
EMSCRIPTEN_RESULT result;
int write_mode = dmScript::CheckTableNumber(L, 3, "type", DATA_TYPE_BINARY);
if (write_mode == DATA_TYPE_BINARY)
{
result = emscripten_websocket_send_binary(conn->m_WS, (void*)string, string_length);
}
else
{
result = emscripten_websocket_send_utf8_text(conn->m_WS, string);
}
if (result)
dmSocket::Result sr = Send(conn, string, string_length, 0);
if (dmSocket::RESULT_OK != sr)
{
CLOSE_CONN("Failed to send on websocket");
}
@ -386,7 +360,7 @@ static int LuaSend(lua_State* L)
return 0;
}
void HandleCallback(WebsocketConnection* conn, int event, int msg_offset, int msg_length)
static void HandleCallback(WebsocketConnection* conn, int event, int msg_offset, int msg_length)
{
if (!dmScript::IsCallbackValid(conn->m_Callback))
return;
@ -410,80 +384,12 @@ void HandleCallback(WebsocketConnection* conn, int event, int msg_offset, int ms
lua_pushlstring(L, conn->m_Buffer + msg_offset, msg_length);
lua_setfield(L, -2, "message");
if(conn->m_HandshakeResponse)
{
HandshakeResponse* response = conn->m_HandshakeResponse;
lua_newtable(L);
lua_pushnumber(L, response->m_ResponseStatusCode);
lua_setfield(L, -2, "status");
lua_pushstring(L, &conn->m_Buffer[response->m_BodyOffset]);
lua_setfield(L, -2, "response");
lua_newtable(L);
for (uint32_t i = 0; i < response->m_Headers.Size(); ++i)
{
lua_pushstring(L, response->m_Headers[i]->m_Value);
lua_setfield(L, -2, response->m_Headers[i]->m_Key);
}
lua_setfield(L, -2, "headers");
lua_setfield(L, -2, "handshake_response");
delete conn->m_HandshakeResponse;
conn->m_HandshakeResponse = 0;
}
if (event == EVENT_DISCONNECTED)
{
lua_pushinteger(L, conn->m_CloseCode);
lua_setfield(L, -2, "code");
}
dmScript::PCall(L, 3, 0);
dmScript::TeardownCallback(conn->m_Callback);
}
HttpHeader::HttpHeader(const char* key, const char* value)
{
m_Key = strdup(key);
m_Value = strdup(value);
}
HttpHeader::~HttpHeader()
{
free((void*)m_Key);
free((void*)m_Value);
m_Key = 0;
m_Value = 0;
}
HttpHeader* HandshakeResponse::GetHeader(const char* header_key)
{
for(uint32_t i = 0; i < m_Headers.Size(); ++i)
{
if (dmStrCaseCmp(m_Headers[i]->m_Key, header_key) == 0)
{
return m_Headers[i];
}
}
return 0;
}
HandshakeResponse::~HandshakeResponse()
{
for(uint32_t i = 0; i < m_Headers.Size(); ++i)
{
delete m_Headers[i];
}
}
// ***************************************************************************************************
// Life cycle functions
@ -533,8 +439,8 @@ static dmExtension::Result AppInitialize(dmExtension::AppParams* params)
g_Websocket.m_BufferSize = dmConfigFile::GetInt(params->m_ConfigFile, "websocket.buffer_size", 64 * 1024);
g_Websocket.m_Timeout = dmConfigFile::GetInt(params->m_ConfigFile, "websocket.socket_timeout", 500 * 1000);
g_Websocket.m_Connections.SetCapacity(4);
g_Websocket.m_Channel = 0;
g_Websocket.m_Pool = 0;
g_Websocket.m_Initialized = 0;
dmConnectionPool::Params pool_params;
pool_params.m_MaxConnections = dmConfigFile::GetInt(params->m_ConfigFile, "websocket.max_connections", 2);
@ -547,10 +453,39 @@ static dmExtension::Result AppInitialize(dmExtension::AppParams* params)
if (dmConnectionPool::RESULT_OK != result)
{
dmLogError("Failed to create connection pool: %d", result);
return dmExtension::RESULT_INIT_ERROR;
}
// We can do without the channel, it will then fallback to the dmSocket::GetHostname (as opposed to dmDNS::GetHostname)
#if defined(HAVE_WSLAY)
dmDNS::Result dns_result = dmDNS::NewChannel(&g_Websocket.m_Channel);
if (dmDNS::RESULT_OK != dns_result)
{
dmLogError("Failed to create connection pool: %d", dns_result);
}
#endif
#if defined(__EMSCRIPTEN__)
// avoid mixed content warning if trying to access wss resource from http page
// If not using this, we get EHOSTUNREACH
EM_ASM({
Module["websocket"].url = window["location"]["protocol"].replace("http", "ws") + "//";
});
#endif
g_Websocket.m_Initialized = 1;
if (!g_Websocket.m_Pool)
{
if (!g_Websocket.m_Pool)
{
dmLogInfo("pool is null!");
dmConnectionPool::Delete(g_Websocket.m_Pool);
}
dmLogInfo("%s extension not initialized", MODULE_NAME);
g_Websocket.m_Initialized = 0;
}
return dmExtension::RESULT_OK;
}
@ -567,6 +502,7 @@ static dmExtension::Result Initialize(dmExtension::Params* params)
static dmExtension::Result AppFinalize(dmExtension::AppParams* params)
{
dmConnectionPool::Shutdown(g_Websocket.m_Pool, dmSocket::SHUTDOWNTYPE_READWRITE);
return dmExtension::RESULT_OK;
}
@ -582,7 +518,7 @@ static dmExtension::Result Finalize(dmExtension::Params* params)
return dmExtension::RESULT_OK;
}
Result PushMessage(WebsocketConnection* conn, MessageType type, int length, const uint8_t* buffer, uint16_t code)
Result PushMessage(WebsocketConnection* conn, MessageType type, int length, const uint8_t* buffer)
{
if (conn->m_Messages.Full())
conn->m_Messages.OffsetCapacity(4);
@ -590,9 +526,11 @@ Result PushMessage(WebsocketConnection* conn, MessageType type, int length, cons
Message msg;
msg.m_Type = (uint32_t)type;
msg.m_Length = length;
msg.m_Code = code;
conn->m_Messages.Push(msg);
// No need to copy itself (html5)
if (buffer != (const uint8_t*)conn->m_Buffer)
{
if ((conn->m_BufferSize + length) >= conn->m_BufferCapacity)
{
conn->m_BufferCapacity = conn->m_BufferSize + length + 1;
@ -600,6 +538,7 @@ Result PushMessage(WebsocketConnection* conn, MessageType type, int length, cons
}
// append to the end of the buffer
memcpy(conn->m_Buffer + conn->m_BufferSize, buffer, length);
}
conn->m_BufferSize += length;
conn->m_Buffer[conn->m_BufferCapacity-1] = 0;
@ -617,19 +556,6 @@ static bool CheckConnectTimeout(WebsocketConnection* conn)
return t >= conn->m_ConnectTimeout;
}
static void ConnectionWorker(void* _conn)
{
WebsocketConnection* conn = (WebsocketConnection*)_conn;
dmSocket::Result sr;
dmConnectionPool::Result pool_result = dmConnectionPool::Dial(g_Websocket.m_Pool, conn->m_Url.m_Hostname, conn->m_Url.m_Port, conn->m_SSL, g_Websocket.m_Timeout, &conn->m_Connection, &sr);
if (dmConnectionPool::RESULT_OK != pool_result)
{
CLOSE_CONN("Failed to open connection: %s", dmSocket::ResultToString(sr));
return;
}
SetState(conn, STATE_HANDSHAKE_WRITE);
}
static dmExtension::Result OnUpdate(dmExtension::Params* params)
{
uint32_t size = g_Websocket.m_Connections.Size();
@ -646,14 +572,17 @@ static dmExtension::Result OnUpdate(dmExtension::Params* params)
conn->m_BufferSize = 0;
}
if (conn->m_WasConnected)
{
HandleCallback(conn, EVENT_DISCONNECTED, 0, conn->m_BufferSize);
}
g_Websocket.m_Connections.EraseSwap(i);
--i;
--size;
DestroyConnection(conn);
}
else if ((STATE_CONNECTED == conn->m_State) || (STATE_DISCONNECTING == conn->m_State))
else if (STATE_CONNECTED == conn->m_State)
{
#if defined(HAVE_WSLAY)
int r = WSL_Poll(conn->m_Ctx);
@ -662,9 +591,27 @@ static dmExtension::Result OnUpdate(dmExtension::Params* params)
CLOSE_CONN("Websocket closing for %s (%s)", conn->m_Url.m_Hostname, WSL_ResultToString(r));
continue;
}
#else
int recv_bytes = 0;
dmSocket::Result sr = Receive(conn, conn->m_Buffer, conn->m_BufferCapacity-1, &recv_bytes);
if( sr == dmSocket::RESULT_WOULDBLOCK )
{
continue;
}
if (dmSocket::RESULT_OK == sr)
{
PushMessage(conn, MESSAGE_TYPE_NORMAL, recv_bytes, (const uint8_t*)conn->m_Buffer);
}
else
{
CLOSE_CONN("Websocket failed to receive data %s", dmSocket::ResultToString(sr));
continue;
}
#endif
uint32_t offset = 0;
bool close_received = false;
for (uint32_t i = 0; i < conn->m_Messages.Size(); ++i)
{
const Message& msg = conn->m_Messages[i];
@ -672,19 +619,25 @@ static dmExtension::Result OnUpdate(dmExtension::Params* params)
if (EVENT_DISCONNECTED == msg.m_Type)
{
conn->m_Status = RESULT_OK;
// close the connection and immediately transition to the DISCONNECTED
// state
SetState(conn, STATE_DISCONNECTED);
conn->m_CloseCode = msg.m_Code;
CloseConnection(conn);
// Put the message at the front of the buffer
conn->m_Messages.SetSize(0);
conn->m_BufferSize = 0;
PushMessage(conn, MESSAGE_TYPE_CLOSE, msg.m_Length, (const uint8_t*)conn->m_Buffer+offset);
close_received = true;
break;
}
HandleCallback(conn, EVENT_MESSAGE, offset, msg.m_Length);
offset += msg.m_Length;
}
if (!close_received) // saving the close message for next step
{
conn->m_Messages.SetSize(0);
conn->m_BufferSize = 0;
}
}
else if (STATE_HANDSHAKE_READ == conn->m_State)
{
if (CheckConnectTimeout(conn))
@ -726,9 +679,10 @@ static dmExtension::Result OnUpdate(dmExtension::Params* params)
dmSocket::SetReceiveTimeout(conn->m_Socket, 1000);
if (conn->m_SSLSocket)
dmSSLSocket::SetReceiveTimeout(conn->m_SSLSocket, 1000);
dmSocket::SetBlocking(conn->m_Socket, false);
#endif
dmSocket::SetBlocking(conn->m_Socket, false);
conn->m_WasConnected = 1;
SetState(conn, STATE_CONNECTED);
HandleCallback(conn, EVENT_CONNECTED, 0, 0);
}
@ -740,13 +694,6 @@ static dmExtension::Result OnUpdate(dmExtension::Params* params)
continue;
}
if (conn->m_ConnectionThread)
{
dmThread::Join(conn->m_ConnectionThread);
conn->m_ConnectionThread = 0;
}
conn->m_Socket = dmConnectionPool::GetSocket(g_Websocket.m_Pool, conn->m_Connection);
conn->m_SSLSocket = dmConnectionPool::GetSSLSocket(g_Websocket.m_Pool, conn->m_Connection);
Result result = SendClientHandshake(conn);
if (RESULT_WOULDBLOCK == result)
{
@ -760,67 +707,63 @@ static dmExtension::Result OnUpdate(dmExtension::Params* params)
SetState(conn, STATE_HANDSHAKE_READ);
}
else if (STATE_CREATE == conn->m_State)
{
if (CheckConnectTimeout(conn))
{
CLOSE_CONN("Connect sequence timed out");
continue;
}
#if defined(__EMSCRIPTEN__)
char uri_buffer[dmURI::MAX_URI_LEN];
const char* uri;
bool no_path = conn->m_Url.m_Path[0] == '\0';
bool no_port = conn->m_Url.m_Port == -1;
if (no_path && no_port)
{
dmSnPrintf(uri_buffer, sizeof(uri_buffer), "%s://%s", conn->m_Url.m_Scheme, conn->m_Url.m_Hostname);
}
else if (no_port)
{
dmSnPrintf(uri_buffer, sizeof(uri_buffer), "%s://%s%s", conn->m_Url.m_Scheme, conn->m_Url.m_Hostname, conn->m_Url.m_Path);
}
else if (no_path)
{
dmSnPrintf(uri_buffer, sizeof(uri_buffer), "%s://%s:%d", conn->m_Url.m_Scheme, conn->m_Url.m_Hostname, conn->m_Url.m_Port);
}
else {
dmSnPrintf(uri_buffer, sizeof(uri_buffer), "%s://%s:%d%s", conn->m_Url.m_Scheme, conn->m_Url.m_Hostname, conn->m_Url.m_Port, conn->m_Url.m_Path);
}
uri = uri_buffer;
EmscriptenWebSocketCreateAttributes ws_attrs = {
uri,
conn->m_Protocol,
EM_TRUE
};
EMSCRIPTEN_WEBSOCKET_T ws = emscripten_websocket_new(&ws_attrs);
if (ws < 0)
{
CLOSE_CONN("Failed to connect to '%s:%d': %d", conn->m_Url.m_Hostname, (int)conn->m_Url.m_Port, ws);
continue;
}
conn->m_WS = ws;
emscripten_websocket_set_onopen_callback(ws, conn, Emscripten_WebSocketOnOpen);
emscripten_websocket_set_onerror_callback(ws, conn, Emscripten_WebSocketOnError);
emscripten_websocket_set_onclose_callback(ws, conn, Emscripten_WebSocketOnClose);
emscripten_websocket_set_onmessage_callback(ws, conn, Emscripten_WebSocketOnMessage);
#else
conn->m_ConnectionThread = dmThread::New((dmThread::ThreadStart)ConnectionWorker, 0x80000, conn, "WSConnect");
#endif
SetState(conn, STATE_CONNECTING);
}
else if (STATE_CONNECTING == conn->m_State)
{
#if defined(__EMSCRIPTEN__)
if (CheckConnectTimeout(conn))
{
CLOSE_CONN("Connect sequence timed out");
continue;
}
#if defined(__EMSCRIPTEN__)
conn->m_SSLSocket = dmSSLSocket::INVALID_SOCKET_HANDLE;
EM_ASM({
// https://emscripten.org/docs/porting/networking.html#emulated-posix-tcp-sockets-over-websockets
Module["websocket"]["subprotocol"] = $0 ? UTF8ToString($0) : null;
}, conn->m_Protocol);
char uri_buffer[dmURI::MAX_URI_LEN];
const char* uri;
if (conn->m_Url.m_Path[0] != '\0') {
dmSnPrintf(uri_buffer, sizeof(uri_buffer), "%s%s", conn->m_Url.m_Hostname, conn->m_Url.m_Path);
uri = uri_buffer;
} else {
uri = conn->m_Url.m_Hostname;
}
dmSocket::Address address;
dmSocket::Result sr = dmSocket::GetHostByName(uri, &address, true, false);
if (dmSocket::RESULT_OK != sr) {
CLOSE_CONN("Failed to get address from host name '%s': %s", uri, dmSocket::ResultToString(sr));
continue;
}
sr = dmSocket::New(address.m_family, dmSocket::TYPE_STREAM, dmSocket::PROTOCOL_TCP, &conn->m_Socket);
if (dmSocket::RESULT_OK != sr) {
CLOSE_CONN("Failed to create socket for '%s': %s", conn->m_Url.m_Hostname, dmSocket::ResultToString(sr));
continue;
}
sr = dmSocket::Connect(conn->m_Socket, address, conn->m_Url.m_Port);
if (dmSocket::RESULT_OK != sr) {
CLOSE_CONN("Failed to connect to '%s:%d': %s", conn->m_Url.m_Hostname, (int)conn->m_Url.m_Port, dmSocket::ResultToString(sr));
continue;
}
#else
dmSocket::Result sr;
int timeout = g_Websocket.m_Timeout;
dmConnectionPool::Result pool_result = dmConnectionPool::Dial(g_Websocket.m_Pool, conn->m_Url.m_Hostname, conn->m_Url.m_Port, g_Websocket.m_Channel, conn->m_SSL, timeout, &conn->m_Connection, &sr);
if (dmConnectionPool::RESULT_OK != pool_result)
{
CLOSE_CONN("Failed to open connection: %s", dmSocket::ResultToString(sr));
continue;
}
conn->m_Socket = dmConnectionPool::GetSocket(g_Websocket.m_Pool, conn->m_Connection);
conn->m_SSLSocket = dmConnectionPool::GetSSLSocket(g_Websocket.m_Pool, conn->m_Connection);
#endif
SetState(conn, STATE_HANDSHAKE_WRITE);
}
}

65
websocket/src/websocket.h
View File

@ -13,17 +13,13 @@
#if defined(HAVE_WSLAY)
#include <wslay/wslay.h>
#endif
#if defined(__EMSCRIPTEN__)
#include <emscripten/websocket.h>
#endif
#include <dmsdk/dlib/connection_pool.h>
#include <dmsdk/dlib/socket.h>
#include <dmsdk/dlib/dns.h>
#include <dmsdk/dlib/uri.h>
#include <dmsdk/dlib/array.h>
#include <dmsdk/dlib/thread.h>
namespace dmCrypt
{
@ -39,12 +35,10 @@ namespace dmWebsocket
enum State
{
STATE_CREATE,
STATE_CONNECTING,
STATE_HANDSHAKE_WRITE,
STATE_HANDSHAKE_READ,
STATE_CONNECTED,
STATE_DISCONNECTING,
STATE_DISCONNECTED,
};
@ -80,46 +74,20 @@ namespace dmWebsocket
struct Message
{
uint16_t m_Code;
uint32_t m_Length:30;
uint32_t m_Type:2;
};
struct HttpHeader
{
const char* m_Key;
const char* m_Value;
HttpHeader(const char* key, const char* value);
~HttpHeader();
};
struct HandshakeResponse
{
int m_HttpMajor;
int m_HttpMinor;
int m_ResponseStatusCode;
int m_BodyOffset;
dmArray<HttpHeader*> m_Headers;
~HandshakeResponse();
HttpHeader* GetHeader(const char* header);
};
struct WebsocketConnection
{
dmScript::LuaCallbackInfo* m_Callback;
#if defined(HAVE_WSLAY)
wslay_event_context_ptr m_Ctx;
#endif
#if defined(__EMSCRIPTEN__)
EMSCRIPTEN_WEBSOCKET_T m_WS;
#endif
dmURI::Parts m_Url;
dmConnectionPool::HConnection m_Connection;
dmSocket::Socket m_Socket;
dmSSLSocket::Socket m_SSLSocket;
dmThread::Thread m_ConnectionThread;
dmArray<Message> m_Messages; // lengths of the messages in the data buffer
uint64_t m_ConnectTimeout;
uint8_t m_Key[16];
@ -130,11 +98,10 @@ namespace dmWebsocket
int m_BufferSize;
uint32_t m_BufferCapacity;
Result m_Status;
uint16_t m_CloseCode;
uint8_t m_SSL:1;
uint8_t m_HasHandshakeData:1;
uint8_t :7;
HandshakeResponse* m_HandshakeResponse;
uint8_t m_WasConnected:1;
uint8_t :6;
};
// Set error message
@ -144,9 +111,6 @@ namespace dmWebsocket
Result SetStatus(WebsocketConnection* conn, Result status, const char* fmt, ...);
#endif
// Set socket state
void SetState(WebsocketConnection* conn, State state);
// Communication
dmSocket::Result Send(WebsocketConnection* conn, const char* buffer, int length, int* out_sent_bytes);
dmSocket::Result Receive(WebsocketConnection* conn, void* buffer, int length, int* received_bytes);
@ -157,11 +121,8 @@ namespace dmWebsocket
Result ReceiveHeaders(WebsocketConnection* conn);
Result VerifyHeaders(WebsocketConnection* conn);
// Callback to Lua
void HandleCallback(WebsocketConnection* conn, int event, int msg_offset, int msg_length);
// Messages
Result PushMessage(WebsocketConnection* conn, MessageType type, int length, const uint8_t* msg, uint16_t code);
Result PushMessage(WebsocketConnection* conn, MessageType type, int length, const uint8_t* msg);
#if defined(HAVE_WSLAY)
// Wslay callbacks
@ -175,12 +136,6 @@ namespace dmWebsocket
int WSL_GenmaskCallback(wslay_event_context_ptr ctx, uint8_t *buf, size_t len, void *user_data);
const char* WSL_ResultToString(int err);
#endif
#if defined(__EMSCRIPTEN__)
EM_BOOL Emscripten_WebSocketOnOpen(int eventType, const EmscriptenWebSocketOpenEvent *websocketEvent, void *userData);
EM_BOOL Emscripten_WebSocketOnError(int eventType, const EmscriptenWebSocketErrorEvent *websocketEvent, void *userData);
EM_BOOL Emscripten_WebSocketOnClose(int eventType, const EmscriptenWebSocketCloseEvent *websocketEvent, void *userData);
EM_BOOL Emscripten_WebSocketOnMessage(int eventType, const EmscriptenWebSocketMessageEvent *websocketEvent, void *userData);
#endif
// Random numbers (PCG)
typedef struct { uint64_t state; uint64_t inc; } pcg32_random_t;
@ -193,5 +148,17 @@ namespace dmWebsocket
#else
void DebugLog(int level, const char* fmt, ...);
#endif
void DebugPrint(int level, const char* msg, const void* _bytes, uint32_t num_bytes);
}

13
websocket/src/wslay_callbacks.cpp
View File

@ -39,7 +39,11 @@ const char* WSL_ResultToString(int err)
int WSL_Init(wslay_event_context_ptr* ctx, ssize_t buffer_size, void* userctx)
{
// Currently only supports client implementation
return wslay_event_context_client_init(ctx, &g_WslCallbacks, userctx);
int ret = -1;
ret = wslay_event_context_client_init(ctx, &g_WslCallbacks, userctx);
if (ret == 0)
wslay_event_config_set_max_recv_msg_length(*ctx, buffer_size);
return ret;
}
@ -120,7 +124,7 @@ void WSL_OnMsgRecvCallback(wslay_event_context_ptr ctx, const struct wslay_event
WebsocketConnection* conn = (WebsocketConnection*)user_data;
if (arg->opcode == WSLAY_TEXT_FRAME || arg->opcode == WSLAY_BINARY_FRAME)
{
PushMessage(conn, MESSAGE_TYPE_NORMAL, arg->msg_length, arg->msg, 0);
PushMessage(conn, MESSAGE_TYPE_NORMAL, arg->msg_length, arg->msg);
} else if (arg->opcode == WSLAY_CONNECTION_CLOSE)
{
// The first two bytes is the close code
@ -133,9 +137,8 @@ void WSL_OnMsgRecvCallback(wslay_event_context_ptr ctx, const struct wslay_event
}
char buffer[1024];
uint16_t status_code = wslay_event_get_status_code_received(ctx);
len = dmSnPrintf(buffer, sizeof(buffer), "Server closing (%u). Reason: '%s'", status_code, reason);
PushMessage(conn, MESSAGE_TYPE_CLOSE, len, (const uint8_t*)buffer, status_code);
len = dmSnPrintf(buffer, sizeof(buffer), "Server closing (%u). Reason: '%s'", wslay_event_get_status_code_received(ctx), reason);
PushMessage(conn, MESSAGE_TYPE_CLOSE, len, (const uint8_t*)buffer);
if (!wslay_event_get_close_sent(ctx))
{