Merging upstream version 1.44.3.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

14 files changed, 4997 insertions, 0 deletions

diff --git a/mqtt_websockets/src/common_public.c b/mqtt_websockets/src/common_public.c
new file mode 100644
index 000000000..7f74fa511
--- /dev/null
+++ b/mqtt_websockets/src/common_public.c
@@ -0,0 +1,7 @@
+#include "common_public.h"
+
+// this dummy exists to have a special pointer with special meaning
+// other than NULL
+void _caller_responsibility(void *ptr) {
+    (void)(ptr);
+}
diff --git a/mqtt_websockets/src/include/common_internal.h b/mqtt_websockets/src/include/common_internal.h
new file mode 100644
index 000000000..2e6563552
--- /dev/null
+++ b/mqtt_websockets/src/include/common_internal.h
@@ -0,0 +1,37 @@
+// SPDX-License-Identifier: GPL-3.0-only
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free Software Foundation, version 3.
+//
+// This program 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 General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along with this program.
+// If not, see <https://www.gnu.org/licenses/>.
+
+#ifndef COMMON_INTERNAL_H
+#define COMMON_INTERNAL_H
+
+#include "endian_compat.h"
+
+#ifdef MQTT_WSS_CUSTOM_ALLOC
+#include "mqtt_wss_pal.h"
+#else
+#define mw_malloc(...) malloc(__VA_ARGS__)
+#define mw_calloc(...) calloc(__VA_ARGS__)
+#define mw_free(...) free(__VA_ARGS__)
+#define mw_strdup(...) strdup(__VA_ARGS__)
+#define mw_realloc(...) realloc(__VA_ARGS__)
+#endif
+
+#ifndef MQTT_WSS_FRAG_MEMALIGN
+#define MQTT_WSS_FRAG_MEMALIGN (8)
+#endif
+
+#define OPENSSL_VERSION_095 0x00905100L
+#define OPENSSL_VERSION_097 0x00907000L
+#define OPENSSL_VERSION_110 0x10100000L
+#define OPENSSL_VERSION_111 0x10101000L
+
+#endif /* COMMON_INTERNAL_H */
diff --git a/mqtt_websockets/src/include/common_public.h b/mqtt_websockets/src/include/common_public.h
new file mode 100644
index 000000000..a855737f9
--- /dev/null
+++ b/mqtt_websockets/src/include/common_public.h
@@ -0,0 +1,33 @@
+#ifndef MQTT_WEBSOCKETS_COMMON_PUBLIC_H
+#define MQTT_WEBSOCKETS_COMMON_PUBLIC_H
+
+#include <stddef.h>
+
+/* free_fnc_t in general (in whatever function or struct it is used)
+ * decides how the related data will be handled.
+ * - If NULL the data are copied internally (causing malloc and later free)
+ * - If pointer provided the free function pointed will be called when data are no longer needed
+ *   to free associated memory. This is effectively transfering ownership of that pointer to the library.
+ *   This also allows caller to provide custom free function other than system one.
+ * - If == CALLER_RESPONSIBILITY the library will not copy the data pointed to and will not call free
+ *   at the end. This is usefull to avoid copying memory (and associated malloc/free) when data are for
+ *   example static. In this case caller has to guarantee the memory pointed to will be valid for entire duration
+ *   it is needed. For example by freeing the data after PUBACK is received or by data being static.
+ */
+typedef void (*free_fnc_t)(void *ptr);
+void _caller_responsibility(void *ptr);
+#define CALLER_RESPONSIBILITY ((free_fnc_t)&_caller_responsibility)
+
+struct mqtt_ng_stats {
+    size_t tx_bytes_queued;
+    int tx_messages_queued;
+    int tx_messages_sent;
+    int rx_messages_rcvd;
+    size_t tx_buffer_used;
+    size_t tx_buffer_free;
+    size_t tx_buffer_size;
+    // part of transaction buffer that containes mesages we can free alredy during the garbage colleciton step
+    size_t tx_buffer_reclaimable;
+};
+
+#endif /* MQTT_WEBSOCKETS_COMMON_PUBLIC_H */
diff --git a/mqtt_websockets/src/include/endian_compat.h b/mqtt_websockets/src/include/endian_compat.h
new file mode 100644
index 000000000..076ccbe84
--- /dev/null
+++ b/mqtt_websockets/src/include/endian_compat.h
@@ -0,0 +1,41 @@
+// SPDX-License-Identifier: GPL-3.0-only
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free Software Foundation, version 3.
+//
+// This program 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 General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along with this program.
+// If not, see <https://www.gnu.org/licenses/>.
+
+#ifndef MQTT_WSS_ENDIAN_COMPAT_H
+#define MQTT_WSS_ENDIAN_COMPAT_H
+
+#ifdef __APPLE__
+    #include <libkern/OSByteOrder.h>
+
+    #define htobe16(x) OSSwapHostToBigInt16(x)
+    #define htole16(x) OSSwapHostToLittleInt16(x)
+    #define be16toh(x) OSSwapBigToHostInt16(x)
+    #define le16toh(x) OSSwapLittleToHostInt16(x)
+
+    #define htobe32(x) OSSwapHostToBigInt32(x)
+    #define htole32(x) OSSwapHostToLittleInt32(x)
+    #define be32toh(x) OSSwapBigToHostInt32(x)
+    #define le32toh(x) OSSwapLittleToHostInt32(x)
+
+    #define htobe64(x) OSSwapHostToBigInt64(x)
+    #define htole64(x) OSSwapHostToLittleInt64(x)
+    #define be64toh(x) OSSwapBigToHostInt64(x)
+    #define le64toh(x) OSSwapLittleToHostInt64(x)
+#else
+#ifdef __FreeBSD__
+    #include <sys/endian.h>
+#else
+    #include <endian.h>
+#endif
+#endif
+
+#endif /* MQTT_WSS_ENDIAN_COMPAT_H */
diff --git a/mqtt_websockets/src/include/mqtt_constants.h b/mqtt_websockets/src/include/mqtt_constants.h
new file mode 100644
index 000000000..1db498976
--- /dev/null
+++ b/mqtt_websockets/src/include/mqtt_constants.h
@@ -0,0 +1,101 @@
+#ifndef MQTT_CONSTANTS_H
+#define MQTT_CONSTANTS_H
+
+#define MQTT_MAX_QOS 0x02
+
+#define MQTT_VERSION_5_0     0x5
+
+/* [MQTT-1.5.5] most significant bit
+   of MQTT Variable Byte Integer signifies
+   there are more bytes following */
+#define MQTT_VBI_CONTINUATION_FLAG 0x80
+#define MQTT_VBI_DATA_MASK         0x7F
+#define MQTT_VBI_MAXBYTES          4
+
+/* MQTT control packet types as defined in
+   2.1.2 MQTT Control Packet type */
+#define MQTT_CPT_CONNECT     0x1
+#define MQTT_CPT_CONNACK     0x2
+#define MQTT_CPT_PUBLISH     0x3
+#define MQTT_CPT_PUBACK      0x4
+#define MQTT_CPT_PUBREC      0x5
+#define MQTT_CPT_PUBREL      0x6
+#define MQTT_CPT_PUBCOMP     0x7
+#define MQTT_CPT_SUBSCRIBE   0x8
+#define MQTT_CPT_SUBACK      0x9
+#define MQTT_CPT_UNSUBSCRIBE 0xA
+#define MQTT_CPT_UNSUBACK    0xB
+#define MQTT_CPT_PINGREQ     0xC
+#define MQTT_CPT_PINGRESP    0xD
+#define MQTT_CPT_DISCONNECT  0xE
+#define MQTT_CPT_AUTH        0xF
+
+// MQTT CONNECT FLAGS (spec:3.1.2.3)
+#define MQTT_CONNECT_FLAG_USERNAME    0x80
+#define MQTT_CONNECT_FLAG_PASSWORD    0x40
+#define MQTT_CONNECT_FLAG_LWT_RETAIN  0x20
+#define MQTT_CONNECT_FLAG_LWT         0x04
+#define MQTT_CONNECT_FLAG_CLEAN_START 0x02
+
+#define MQTT_CONNECT_FLAG_QOS_MASK    0x18
+#define MQTT_CONNECT_FLAG_QOS_BITSHIFT 3
+
+#define MQTT_MAX_CLIENT_ID 23 /* [MQTT-3.1.3-5] */
+
+// MQTT Property identifiers [MQTT-2.2.2.2]
+#define MQTT_PROP_PAYLOAD_FMT_INDICATOR        0x01
+#define MQTT_PROP_PAYLOAD_FMT_INDICATOR_NAME   "Payload Format Indicator"
+#define MQTT_PROP_MSG_EXPIRY_INTERVAL          0x02
+#define MQTT_PROP_MSG_EXPIRY_INTERVAL_NAME     "Message Expiry Interval"
+#define MQTT_PROP_CONTENT_TYPE                 0x03
+#define MQTT_PROP_CONTENT_TYPE_NAME            "Content Type"
+#define MQTT_PROP_RESPONSE_TOPIC               0x08
+#define MQTT_PROP_RESPONSE_TOPIC_NAME          "Response Topic"
+#define MQTT_PROP_CORRELATION_DATA             0x09
+#define MQTT_PROP_CORRELATION_DATA_NAME        "Correlation Data"
+#define MQTT_PROP_SUB_IDENTIFIER               0x0B
+#define MQTT_PROP_SUB_IDENTIFIER_NAME          "Subscription Identifier"
+#define MQTT_PROP_SESSION_EXPIRY_INTERVAL      0x11
+#define MQTT_PROP_SESSION_EXPIRY_INTERVAL_NAME "Session Expiry Interval"
+#define MQTT_PROP_ASSIGNED_CLIENT_ID           0x12
+#define MQTT_PROP_ASSIGNED_CLIENT_ID_NAME      "Assigned Client Identifier"
+#define MQTT_PROP_SERVER_KEEP_ALIVE            0x13
+#define MQTT_PROP_SERVER_KEEP_ALIVE_NAME       "Server Keep Alive"
+#define MQTT_PROP_AUTH_METHOD                  0x15
+#define MQTT_PROP_AUTH_METHOD_NAME             "Authentication Method"
+#define MQTT_PROP_AUTH_DATA                    0x16
+#define MQTT_PROP_AUTH_DATA_NAME               "Authentication Data"
+#define MQTT_PROP_REQ_PROBLEM_INFO             0x17
+#define MQTT_PROP_REQ_PROBLEM_INFO_NAME        "Request Problem Information"
+#define MQTT_PROP_WILL_DELAY_INTERVAL          0x18
+#define MQTT_PROP_WIIL_DELAY_INTERVAL_NAME     "Will Delay Interval"
+#define MQTT_PROP_REQ_RESP_INFORMATION         0x19
+#define MQTT_PROP_REQ_RESP_INFORMATION_NAME    "Request Response Information"
+#define MQTT_PROP_RESP_INFORMATION             0x1A
+#define MQTT_PROP_RESP_INFORMATION_NAME        "Response Information"
+#define MQTT_PROP_SERVER_REF                   0x1C
+#define MQTT_PROP_SERVER_REF_NAME              "Server Reference"
+#define MQTT_PROP_REASON_STR                   0x1F
+#define MQTT_PROP_REASON_STR_NAME              "Reason String"
+#define MQTT_PROP_RECEIVE_MAX                  0x21
+#define MQTT_PROP_RECEIVE_MAX_NAME             "Receive Maximum"
+#define MQTT_PROP_TOPIC_ALIAS_MAX              0x22
+#define MQTT_PROP_TOPIC_ALIAS_MAX_NAME         "Topic Alias Maximum"
+#define MQTT_PROP_TOPIC_ALIAS                  0x23
+#define MQTT_PROP_TOPIC_ALIAS_NAME             "Topic Alias"
+#define MQTT_PROP_MAX_QOS                      0x24
+#define MQTT_PROP_MAX_QOS_NAME                 "Maximum QoS"
+#define MQTT_PROP_RETAIN_AVAIL                 0x25
+#define MQTT_PROP_RETAIN_AVAIL_NAME            "Retain Available"
+#define MQTT_PROP_USR                          0x26
+#define MQTT_PROP_USR_NAME                     "User Property"
+#define MQTT_PROP_MAX_PKT_SIZE                 0x27
+#define MQTT_PROP_MAX_PKT_SIZE_NAME            "Maximum Packet Size"
+#define MQTT_PROP_WILDCARD_SUB_AVAIL           0x28
+#define MQTT_PROP_WILDCARD_SUB_AVAIL_NAME      "Wildcard Subscription Available"
+#define MQTT_PROP_SUB_ID_AVAIL                 0x29
+#define MQTT_PROP_SUB_ID_AVAIL_NAME            "Subscription Identifier Available"
+#define MQTT_PROP_SHARED_SUB_AVAIL             0x2A
+#define MQTT_PROP_SHARED_SUB_AVAIL_NAME        "Shared Subscription Available"
+
+#endif /* MQTT_CONSTANTS_H */
diff --git a/mqtt_websockets/src/include/mqtt_ng.h b/mqtt_websockets/src/include/mqtt_ng.h
new file mode 100644
index 000000000..09668d09b
--- /dev/null
+++ b/mqtt_websockets/src/include/mqtt_ng.h
@@ -0,0 +1,97 @@
+#include <stdint.h>
+#include <sys/types.h>
+#include <time.h>
+
+#include "ringbuffer.h"
+#include "common_public.h"
+
+#define MQTT_NG_MSGGEN_OK         0
+// MQTT_NG_MSGGEN_USER_ERROR means parameters given to this function
+// do not make sense or are out of MQTT specs
+#define MQTT_NG_MSGGEN_USER_ERROR  1
+#define MQTT_NG_MSGGEN_BUFFER_OOM  2
+#define MQTT_NG_MSGGEN_MSG_TOO_BIG 3
+
+struct mqtt_ng_client;
+
+/* Converts integer to MQTT Variable Byte Integer as per 1.5.5 of MQTT 5 specs
+ * @param input value to be converted
+ * @param output pointer to memory where output will be written to. Must allow up to 4 bytes to be written.
+ * @return number of bytes written to output or <= 0 if error in which case contents of output are undefined
+ */
+int uint32_to_mqtt_vbi(uint32_t input, char *output);
+
+struct mqtt_lwt_properties {
+    char *will_topic;
+    free_fnc_t will_topic_free;
+
+    void *will_message;
+    free_fnc_t will_message_free;
+    size_t will_message_size;
+
+    int will_qos;
+    int will_retain;
+};
+
+struct mqtt_auth_properties {
+    char *client_id;
+    free_fnc_t client_id_free;
+    char *username;
+    free_fnc_t username_free;
+    char *password;
+    free_fnc_t password_free;
+};
+
+int mqtt_ng_connect(struct mqtt_ng_client *client,
+                    struct mqtt_auth_properties *auth,
+                    struct mqtt_lwt_properties *lwt,
+                    uint8_t clean_start,
+                    uint16_t keep_alive);
+
+int mqtt_ng_publish(struct mqtt_ng_client *client,
+                    char *topic,
+                    free_fnc_t topic_free,
+                    void *msg,
+                    free_fnc_t msg_free,
+                    size_t msg_len,
+                    uint8_t publish_flags,
+                    uint16_t *packet_id);
+
+struct mqtt_sub {
+    char *topic;
+    free_fnc_t topic_free;
+    uint8_t options;
+};
+
+int mqtt_ng_subscribe(struct mqtt_ng_client *client, struct mqtt_sub *subscriptions, size_t subscription_count);
+
+int mqtt_ng_ping(struct mqtt_ng_client *client);
+
+typedef ssize_t (*mqtt_ng_send_fnc_t)(void *user_ctx, const void* buf, size_t len);
+
+struct mqtt_ng_init {
+    mqtt_wss_log_ctx_t log;
+    rbuf_t data_in;
+    mqtt_ng_send_fnc_t data_out_fnc;
+    void *user_ctx;
+
+    void (*puback_callback)(uint16_t packet_id);
+    void (*connack_callback)(void* user_ctx, int connack_reply);
+    void (*msg_callback)(const char *topic, const void *msg, size_t msglen, int qos);
+};
+
+struct mqtt_ng_client *mqtt_ng_init(struct mqtt_ng_init *settings);
+
+void mqtt_ng_destroy(struct mqtt_ng_client *client);
+
+int mqtt_ng_disconnect(struct mqtt_ng_client *client, uint8_t reason_code);
+
+int mqtt_ng_sync(struct mqtt_ng_client *client);
+
+time_t mqtt_ng_last_send_time(struct mqtt_ng_client *client);
+
+void mqtt_ng_set_max_mem(struct mqtt_ng_client *client, size_t bytes);
+
+void mqtt_ng_get_stats(struct mqtt_ng_client *client, struct mqtt_ng_stats *stats);
+
+int mqtt_ng_set_topic_alias(struct mqtt_ng_client *client, const char *topic);
diff --git a/mqtt_websockets/src/include/mqtt_wss_client.h b/mqtt_websockets/src/include/mqtt_wss_client.h
new file mode 100644
index 000000000..e325961b7
--- /dev/null
+++ b/mqtt_websockets/src/include/mqtt_wss_client.h
@@ -0,0 +1,172 @@
+// Copyright (C) 2020 Timotej Šiškovič
+// SPDX-License-Identifier: GPL-3.0-only
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free Software Foundation, version 3.
+//
+// This program 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 General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along with this program.
+// If not, see <https://www.gnu.org/licenses/>.
+
+#ifndef MQTT_WSS_CLIENT_H
+#define MQTT_WSS_CLIENT_H
+
+#include <stdint.h>
+#include <stddef.h> //size_t
+
+#include "mqtt_wss_log.h"
+#include "common_public.h"
+
+// All OK call me at your earliest convinience
+#define MQTT_WSS_OK              0
+/* All OK, poll timeout you requested when calling mqtt_wss_service expired - you might want to know if timeout
+ * happened or we got some data or handle same as MQTT_WSS_OK
+ */
+#define MQTT_WSS_OK_TO           1
+// Connection was closed by remote
+#define MQTT_WSS_ERR_CONN_DROP  -1
+// Error in MQTT protocol (e.g. malformed packet)
+#define MQTT_WSS_ERR_PROTO_MQTT -2
+// Error in WebSocket protocol (e.g. malformed packet)
+#define MQTT_WSS_ERR_PROTO_WS   -3
+
+#define MQTT_WSS_ERR_TX_BUF_TOO_SMALL  -4
+#define MQTT_WSS_ERR_RX_BUF_TOO_SMALL  -5
+
+#define MQTT_WSS_ERR_TOO_BIG_FOR_SERVER -6
+// if client was initialized with MQTT 3 but MQTT 5 feature
+// was requested by user of library
+#define MQTT_WSS_ERR_CANT_DO       -8
+
+typedef struct mqtt_wss_client_struct *mqtt_wss_client;
+
+typedef void (*msg_callback_fnc_t)(const char *topic, const void *msg, size_t msglen, int qos);
+/* Creates new instance of MQTT over WSS. Doesn't start connection.
+ * @param log_prefix this is prefix to be used when logging to discern between multiple
+ *        mqtt_wss instances. Can be NULL.
+ * @param log_callback is function pointer to fnc to be called when mqtt_wss wants
+ *        to log. This allows plugging this library into your own logging system/solution.
+ *        If NULL STDOUT/STDERR will be used.
+ * @param msg_callback is function pointer to function which will be called
+ *        when application level message arrives from broker (for subscribed topics).
+ *        Can be NULL if you are not interested about incoming messages.
+ * @param puback_callback is function pointer to function to be called when QOS1 Publish
+ *        is acknowledged by server
+ */
+mqtt_wss_client mqtt_wss_new(const char *log_prefix,
+                             mqtt_wss_log_callback_t log_callback,
+                             msg_callback_fnc_t msg_callback,
+                             void (*puback_callback)(uint16_t packet_id));
+
+void mqtt_wss_set_max_buf_size(mqtt_wss_client client, size_t size);
+
+void mqtt_wss_destroy(mqtt_wss_client client);
+
+struct mqtt_connect_params;
+struct mqtt_wss_proxy;
+
+#define MQTT_WSS_SSL_CERT_CHECK_FULL   0x00
+#define MQTT_WSS_SSL_ALLOW_SELF_SIGNED 0x01
+#define MQTT_WSS_SSL_DONT_CHECK_CERTS  0x08
+
+/* Will block until the MQTT over WSS connection is established or return error
+ * @param client mqtt_wss_client which should connect
+ * @param host to connect to (where MQTT over WSS server is listening)
+ * @param port to connect to (where MQTT over WSS server is listening)
+ * @param mqtt_params pointer to mqtt_connect_params structure which contains MQTT credentials and settings
+ * @param ssl_flags parameters for OpenSSL, 0=MQTT_WSS_SSL_CERT_CHECK_FULL
+ */
+int mqtt_wss_connect(mqtt_wss_client client, char *host, int port, struct mqtt_connect_params *mqtt_params, int ssl_flags, struct mqtt_wss_proxy *proxy);
+int mqtt_wss_service(mqtt_wss_client client, int timeout_ms);
+void mqtt_wss_disconnect(mqtt_wss_client client, int timeout_ms);
+
+// we redefine this instead of using MQTT-C flags as in future
+// we want to support different MQTT implementations if needed
+enum mqtt_wss_publish_flags {
+    MQTT_WSS_PUB_QOS0    = 0x0,
+    MQTT_WSS_PUB_QOS1    = 0x1,
+    MQTT_WSS_PUB_QOS2    = 0x2,
+    MQTT_WSS_PUB_QOSMASK = 0x3,
+    MQTT_WSS_PUB_RETAIN  = 0x4
+};
+
+struct mqtt_connect_params {
+    const char *clientid;
+    const char *username;
+    const char *password;
+    const char *will_topic;
+    const void *will_msg;
+    enum mqtt_wss_publish_flags will_flags;
+    size_t will_msg_len;
+    int keep_alive;
+    int drop_on_publish_fail;
+};
+
+enum mqtt_wss_proxy_type {
+    MQTT_WSS_DIRECT = 0,
+    MQTT_WSS_PROXY_HTTP
+};
+
+struct mqtt_wss_proxy {
+    enum mqtt_wss_proxy_type type;
+    const char *host;
+    int port;
+    const char *username;
+    const char *password;
+};
+
+/* TODO!!! update the description
+ * Publishes MQTT message and gets message id
+ * @param client mqtt_wss_client which should transfer the message
+ * @param topic MQTT topic to publish message to (0 terminated C string)
+ * @param msg Message to be published (no need for 0 termination)
+ * @param msg_len Length of the message to be published
+ * @param publish_flags see enum mqtt_wss_publish_flags e.g. (MQTT_WSS_PUB_QOS1 | MQTT_WSS_PUB_RETAIN)
+ * @param packet_id is 16 bit unsigned int representing ID that can be used to pair with PUBACK callback
+ *        for usages where application layer wants to know which messages are delivered when
+ * @return Returns 0 on success
+ */
+int mqtt_wss_publish5(mqtt_wss_client client,
+                      char *topic,
+                      free_fnc_t topic_free,
+                      void *msg,
+                      free_fnc_t msg_free,
+                      size_t msg_len,
+                      uint8_t publish_flags,
+                      uint16_t *packet_id);
+
+int mqtt_wss_set_topic_alias(mqtt_wss_client client, const char *topic);
+
+/* Subscribes to MQTT topic
+ * @param client mqtt_wss_client which should do the subscription
+ * @param topic MQTT topic to subscribe to
+ * @param max_qos_level maximum QOS level that broker can send to us on this subscription
+ * @return Returns 0 on success
+ */
+int mqtt_wss_subscribe(mqtt_wss_client client, char *topic, int max_qos_level);
+
+
+struct mqtt_wss_stats {
+    uint64_t bytes_tx;
+    uint64_t bytes_rx;
+#ifdef MQTT_WSS_CPUSTATS
+    uint64_t time_keepalive;
+    uint64_t time_read_socket;
+    uint64_t time_write_socket;
+    uint64_t time_process_websocket;
+    uint64_t time_process_mqtt;
+#endif
+    struct mqtt_ng_stats mqtt;
+};
+
+struct mqtt_wss_stats mqtt_wss_get_stats(mqtt_wss_client client);
+
+#ifdef MQTT_WSS_DEBUG
+#include <openssl/ssl.h>
+void mqtt_wss_set_SSL_CTX_keylog_cb(mqtt_wss_client client, void (*ssl_ctx_keylog_cb)(const SSL *ssl, const char *line));
+#endif
+
+#endif /* MQTT_WSS_CLIENT_H */
diff --git a/mqtt_websockets/src/include/mqtt_wss_log.h b/mqtt_websockets/src/include/mqtt_wss_log.h
new file mode 100644
index 000000000..a33c460c9
--- /dev/null
+++ b/mqtt_websockets/src/include/mqtt_wss_log.h
@@ -0,0 +1,37 @@
+#ifndef MQTT_WSS_LOG_H
+#define MQTT_WSS_LOG_H
+
+typedef enum mqtt_wss_log_type {
+    MQTT_WSS_LOG_DEBUG = 0x01,
+    MQTT_WSS_LOG_INFO  = 0x02,
+    MQTT_WSS_LOG_WARN  = 0x03,
+    MQTT_WSS_LOG_ERROR = 0x81,
+    MQTT_WSS_LOG_FATAL = 0x88
+} mqtt_wss_log_type_t;
+
+typedef void (*mqtt_wss_log_callback_t)(mqtt_wss_log_type_t, const char*);
+
+typedef struct mqtt_wss_log_ctx *mqtt_wss_log_ctx_t;
+
+/** Creates logging context with optional prefix and optional callback
+ *   @param ctx_prefix String to be prefixed to every log message.
+ *   This is useful if multiple clients are instantiated to be able to
+ *   know which one this message belongs to. Can be `NULL` for no prefix.
+ *   @param log_callback Callback to be called instead of logging to
+ *   `STDOUT` or `STDERR` (if debug enabled otherwise silent). Callback has to be
+ *   pointer to function of `void function(mqtt_wss_log_type_t, const char*)` type.
+ *   If `NULL` default will be used (silent or STDERR/STDOUT).
+ *   @return mqtt_wss_log_ctx_t or `NULL` on error */
+mqtt_wss_log_ctx_t mqtt_wss_log_ctx_create(const char *ctx_prefix, mqtt_wss_log_callback_t log_callback);
+
+/** Destroys logging context and cleans up the memory
+ *  @param ctx Context to destroy */
+void mqtt_wss_log_ctx_destroy(mqtt_wss_log_ctx_t ctx);
+
+void mws_fatal(mqtt_wss_log_ctx_t ctx, const char *fmt, ...);
+void mws_error(mqtt_wss_log_ctx_t ctx, const char *fmt, ...);
+void mws_warn (mqtt_wss_log_ctx_t ctx, const char *fmt, ...);
+void mws_info (mqtt_wss_log_ctx_t ctx, const char *fmt, ...);
+void mws_debug(mqtt_wss_log_ctx_t ctx, const char *fmt, ...);
+
+#endif /* MQTT_WSS_LOG_H */
diff --git a/mqtt_websockets/src/include/ws_client.h b/mqtt_websockets/src/include/ws_client.h
new file mode 100644
index 000000000..de4fac40b
--- /dev/null
+++ b/mqtt_websockets/src/include/ws_client.h
@@ -0,0 +1,130 @@
+// Copyright (C) 2020 Timotej Šiškovič
+// SPDX-License-Identifier: GPL-3.0-only
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free Software Foundation, version 3.
+//
+// This program 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 General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along with this program.
+// If not, see <https://www.gnu.org/licenses/>.
+
+#ifndef WS_CLIENT_H
+#define WS_CLIENT_H
+
+#include "ringbuffer.h"
+#include "mqtt_wss_log.h"
+
+#include <stdint.h>
+
+#define WS_CLIENT_NEED_MORE_BYTES     0x10
+#define WS_CLIENT_PARSING_DONE        0x11
+#define WS_CLIENT_CONNECTION_CLOSED   0x12
+#define WS_CLIENT_PROTOCOL_ERROR     -0x10
+#define WS_CLIENT_BUFFER_FULL        -0x11
+#define WS_CLIENT_INTERNAL_ERROR     -0x12
+
+enum websocket_client_conn_state {
+    WS_RAW = 0,
+    WS_HANDSHAKE,
+    WS_ESTABLISHED,
+    WS_ERROR,        // connection has to be restarted if this is reached
+    WS_CONN_CLOSED_GRACEFUL
+};
+
+enum websocket_client_hdr_parse_state {
+    WS_HDR_HTTP = 0,        // need to check HTTP/1.1
+    WS_HDR_RC,              // need to read HTTP code
+    WS_HDR_ENDLINE,         // need to read rest of the first line
+    WS_HDR_PARSE_HEADERS,   // rest of the header until CRLF CRLF
+    WS_HDR_PARSE_DONE,
+    WS_HDR_ALL_DONE
+};
+
+enum websocket_client_rx_ws_parse_state {
+    WS_FIRST_2BYTES = 0,
+    WS_PAYLOAD_EXTENDED_16,
+    WS_PAYLOAD_EXTENDED_64,
+    WS_PAYLOAD_DATA, // BINARY payload to be passed to MQTT
+    WS_PAYLOAD_CONNECTION_CLOSE,
+    WS_PAYLOAD_CONNECTION_CLOSE_EC,
+    WS_PAYLOAD_CONNECTION_CLOSE_MSG,
+    WS_PAYLOAD_SKIP_UNKNOWN_PAYLOAD,
+    WS_PAYLOAD_PING_REQ_PAYLOAD, // PING payload to be sent back as PONG
+    WS_PACKET_DONE
+};
+
+enum websocket_opcode {
+    WS_OP_CONTINUATION_FRAME = 0x0,
+    WS_OP_TEXT_FRAME         = 0x1,
+    WS_OP_BINARY_FRAME       = 0x2,
+    WS_OP_CONNECTION_CLOSE   = 0x8,
+    WS_OP_PING               = 0x9,
+    WS_OP_PONG               = 0xA
+};
+
+struct ws_op_close_payload {
+    uint16_t ec;
+    char *reason;
+};
+
+struct http_header {
+    char *key;
+    char *value;
+    struct http_header *next;
+};
+
+typedef struct websocket_client {
+    enum websocket_client_conn_state state;
+
+    struct ws_handshake {
+        enum websocket_client_hdr_parse_state hdr_state;
+        char *nonce_reply;
+        int nonce_matched;
+        int http_code;
+        char *http_reply_msg;
+        struct http_header *headers;
+        struct http_header *headers_tail;
+        int hdr_count;
+    } hs;
+
+    struct ws_rx {
+        enum websocket_client_rx_ws_parse_state parse_state;
+        enum websocket_opcode opcode;
+        uint64_t payload_length;
+        uint64_t payload_processed;
+        union {
+            struct ws_op_close_payload op_close;
+            char *ping_msg;
+        } specific_data;
+    } rx;
+
+    rbuf_t buf_read;    // from SSL
+    rbuf_t buf_write;   // to SSL and then to socket
+    // TODO if ringbuffer gets multiple tail support
+    // we can work without buf_to_mqtt and thus reduce
+    // memory usage and remove one more memcpy buf_read->buf_to_mqtt
+    rbuf_t buf_to_mqtt; // RAW data for MQTT lib
+
+    int entropy_fd;
+
+    // careful host is borrowed, don't free
+    char **host;
+    mqtt_wss_log_ctx_t log;
+} ws_client;
+
+ws_client *ws_client_new(size_t buf_size, char **host, mqtt_wss_log_ctx_t log);
+void ws_client_destroy(ws_client *client);
+void ws_client_reset(ws_client *client);
+
+int ws_client_start_handshake(ws_client *client);
+
+int ws_client_want_write(ws_client *client);
+
+int ws_client_process(ws_client *client);
+
+int ws_client_send(ws_client *client, enum websocket_opcode frame_type, const char *data, size_t size);
+
+#endif /* WS_CLIENT_H */
diff --git a/mqtt_websockets/src/mqtt_ng.c b/mqtt_websockets/src/mqtt_ng.c
new file mode 100644
index 000000000..81cffccf0
--- /dev/null
+++ b/mqtt_websockets/src/mqtt_ng.c
@@ -0,0 +1,2233 @@
+#define _GNU_SOURCE
+
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+#include <pthread.h>
+#include <inttypes.h>
+
+#include "c_rhash.h"
+
+#include "common_internal.h"
+#include "mqtt_constants.h"
+#include "mqtt_wss_log.h"
+#include "mqtt_ng.h"
+
+#define UNIT_LOG_PREFIX "mqtt_client: "
+#define FATAL(fmt, ...) mws_fatal(client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+#define ERROR(fmt, ...) mws_error(client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+#define WARN(fmt, ...)  mws_warn (client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+#define INFO(fmt, ...)  mws_info (client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+#define DEBUG(fmt, ...) mws_debug(client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+
+#define SMALL_STRING_DONT_FRAGMENT_LIMIT 128
+
+#define MIN(a,b) (((a)<(b))?(a):(b))
+
+#define LOCK_HDR_BUFFER(buffer) pthread_mutex_lock(&((buffer)->mutex))
+#define UNLOCK_HDR_BUFFER(buffer) pthread_mutex_unlock(&((buffer)->mutex))
+
+#define BUFFER_FRAG_GARBAGE_COLLECT         0x01
+// some packets can be marked for garbage collection
+// immediately when they are sent (e.g. sent PUBACK on QoS1)
+#define BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND 0x02
+// as buffer fragment can point to both
+// external data and data in the same buffer
+// we mark the former case with BUFFER_FRAG_DATA_EXTERNAL
+#define BUFFER_FRAG_DATA_EXTERNAL           0x04
+// as single MQTT Packet can be stored into multiple
+// buffer fragments (depending on copy requirements)
+// this marks this fragment to be the first/last
+#define BUFFER_FRAG_MQTT_PACKET_HEAD        0x10
+#define BUFFER_FRAG_MQTT_PACKET_TAIL        0x20
+
+typedef uint16_t buffer_frag_flag_t;
+struct buffer_fragment {
+    size_t len;
+    size_t sent;
+    buffer_frag_flag_t flags;
+    void (*free_fnc)(void *ptr);
+    char *data;
+
+    uint16_t packet_id;
+
+    struct buffer_fragment *next;
+};
+
+typedef struct buffer_fragment *mqtt_msg_data;
+
+// buffer used for MQTT headers only
+// not for actual data sent
+struct header_buffer {
+    size_t size;
+    char *data;
+    char *tail;
+    struct buffer_fragment *tail_frag;
+};
+
+struct transaction_buffer {
+    struct header_buffer hdr_buffer;
+    // used while building new message
+    // to be able to revert state easily
+    // in case of error mid processing
+    struct header_buffer state_backup;
+    pthread_mutex_t mutex;
+    struct buffer_fragment *sending_frag;
+};
+
+enum mqtt_client_state {
+    RAW = 0,
+    CONNECT_PENDING,
+    CONNECTING,
+    CONNECTED, 
+    ERROR,
+    DISCONNECTED
+};
+
+enum parser_state {
+    MQTT_PARSE_FIXED_HEADER_PACKET_TYPE = 0,
+    MQTT_PARSE_FIXED_HEADER_LEN,
+    MQTT_PARSE_VARIABLE_HEADER,
+    MQTT_PARSE_MQTT_PACKET_DONE
+};
+
+enum varhdr_parser_state {
+    MQTT_PARSE_VARHDR_INITIAL = 0,
+    MQTT_PARSE_VARHDR_OPTIONAL_REASON_CODE,
+    MQTT_PARSE_VARHDR_PROPS,
+    MQTT_PARSE_VARHDR_TOPICNAME,
+    MQTT_PARSE_VARHDR_POST_TOPICNAME,
+    MQTT_PARSE_VARHDR_PACKET_ID,
+    MQTT_PARSE_REASONCODES,
+    MQTT_PARSE_PAYLOAD
+};
+
+struct mqtt_vbi_parser_ctx {
+    char data[MQTT_VBI_MAXBYTES];
+    uint8_t bytes;
+    uint32_t result;
+};
+
+enum mqtt_datatype {
+    MQTT_TYPE_UNKNOWN = 0,
+    MQTT_TYPE_UINT_8,
+    MQTT_TYPE_UINT_16,
+    MQTT_TYPE_UINT_32,
+    MQTT_TYPE_VBI,
+    MQTT_TYPE_STR,
+    MQTT_TYPE_STR_PAIR,
+    MQTT_TYPE_BIN
+};
+
+struct mqtt_property {
+    uint8_t id;
+    enum mqtt_datatype type;
+    union {
+        char *strings[2];
+        void *bindata;
+        uint8_t uint8;
+        uint16_t uint16;
+        uint32_t uint32;
+    } data;
+    size_t bindata_len;
+    struct mqtt_property *next;
+};
+
+enum mqtt_properties_parser_state {
+    PROPERTIES_LENGTH = 0,
+    PROPERTY_CREATE,
+    PROPERTY_ID,
+    PROPERTY_TYPE_UINT8,
+    PROPERTY_TYPE_UINT16,
+    PROPERTY_TYPE_UINT32,
+    PROPERTY_TYPE_STR_BIN_LEN,
+    PROPERTY_TYPE_STR,
+    PROPERTY_TYPE_BIN,
+    PROPERTY_TYPE_VBI,
+    PROPERTY_NEXT
+};
+
+struct mqtt_properties_parser_ctx {
+    enum mqtt_properties_parser_state state;
+    struct mqtt_property *head;
+    struct mqtt_property *tail;
+    uint32_t properties_length;
+    uint32_t vbi_length;
+    struct mqtt_vbi_parser_ctx vbi_parser_ctx;
+    size_t bytes_consumed;
+    int str_idx;
+};
+
+struct mqtt_connack {
+    uint8_t flags;
+    uint8_t reason_code;
+};
+struct mqtt_puback {
+    uint16_t packet_id;
+    uint8_t reason_code;
+};
+
+struct mqtt_suback {
+    uint16_t packet_id;
+    uint8_t *reason_codes;
+    uint8_t reason_code_count;
+    uint8_t reason_codes_pending;
+};
+
+struct mqtt_publish {
+    uint16_t topic_len;
+    char *topic;
+    uint16_t packet_id;
+    size_t data_len;
+    char *data;
+    uint8_t qos;
+};
+
+struct mqtt_disconnect {
+    uint8_t reason_code;
+};
+
+struct mqtt_ng_parser {
+    rbuf_t received_data;
+
+    uint8_t mqtt_control_packet_type;
+    uint32_t mqtt_fixed_hdr_remaining_length;
+    size_t mqtt_parsed_len;
+
+    struct mqtt_vbi_parser_ctx vbi_parser;
+    struct mqtt_properties_parser_ctx properties_parser;
+
+    enum parser_state state;
+    enum varhdr_parser_state varhdr_state;
+
+    struct mqtt_property *varhdr_properties;
+
+    union {
+        struct mqtt_connack connack;
+        struct mqtt_puback puback;
+        struct mqtt_suback suback;
+        struct mqtt_publish publish;
+        struct mqtt_disconnect disconnect;
+    } mqtt_packet;
+};
+
+struct topic_alias_data {
+    uint16_t idx;
+    uint32_t usage_count;
+};
+
+struct topic_aliases_data {
+    c_rhash stoi_dict;
+    uint32_t idx_max;
+    uint32_t idx_assigned;
+    pthread_rwlock_t rwlock;
+};
+
+struct mqtt_ng_client {
+    struct transaction_buffer main_buffer;
+
+    enum mqtt_client_state client_state;
+
+    mqtt_msg_data connect_msg;
+
+    mqtt_wss_log_ctx_t log;
+
+    mqtt_ng_send_fnc_t send_fnc_ptr;
+    void *user_ctx;
+
+    // time when last fragment of MQTT message was sent
+    time_t time_of_last_send;
+
+    struct mqtt_ng_parser parser;
+
+    size_t max_mem_bytes;
+
+    void (*puback_callback)(uint16_t packet_id);
+    void (*connack_callback)(void* user_ctx, int connack_reply);
+    void (*msg_callback)(const char *topic, const void *msg, size_t msglen, int qos);
+
+    unsigned int ping_pending:1;
+
+    struct mqtt_ng_stats stats;
+    pthread_mutex_t stats_mutex;
+
+    struct topic_aliases_data tx_topic_aliases;
+    c_rhash rx_aliases;
+
+    size_t max_msg_size;
+};
+
+char pingreq[] = { MQTT_CPT_PINGREQ << 4, 0x00 };
+
+struct buffer_fragment ping_frag = {
+    .data = pingreq,
+    .flags =  BUFFER_FRAG_MQTT_PACKET_HEAD | BUFFER_FRAG_MQTT_PACKET_TAIL,
+    .free_fnc = NULL,
+    .len = sizeof(pingreq),
+    .next = NULL,
+    .sent = 0,
+    .packet_id = 0
+};
+
+int uint32_to_mqtt_vbi(uint32_t input, char *output) {
+    int i = 1;
+    *output = 0;
+
+    /* MQTT 5 specs allows max 4 bytes of output
+       making it 0xFF, 0xFF, 0xFF, 0x7F
+       representing number 268435455 decimal
+       see 1.5.5. Variable Byte Integer */
+    if(input >= 256 * 1024 * 1024)
+        return 0;
+
+    if(!input) {
+        *output = 0;
+        return 1;
+    }
+
+    while(input) {
+        output[i-1] = input & MQTT_VBI_DATA_MASK;
+        input >>= 7;
+        if (input)
+            output[i-1] |= MQTT_VBI_CONTINUATION_FLAG;
+        i++;
+    }
+    return i - 1;
+}
+
+int mqtt_vbi_to_uint32(char *input, uint32_t *output) {
+    // dont want to operate directly on output
+    // as I want it to be possible for input and output
+    // pointer to be the same
+    uint32_t result = 0;
+    uint32_t multiplier = 1;
+
+    do {
+        result += (uint32_t)(*input & MQTT_VBI_DATA_MASK) * multiplier;
+        if (multiplier > 128*128*128)
+            return 1;
+        multiplier <<= 7;
+    } while (*input++ & MQTT_VBI_CONTINUATION_FLAG);
+    *output = result;
+    return 0;
+}
+
+#ifdef TESTS
+#include <stdio.h>
+#define MQTT_VBI_MAXLEN 4
+// we add extra byte to check we dont write out of bounds
+// in case where 4 bytes are supposed to be written
+static const char _mqtt_vbi_0[MQTT_VBI_MAXLEN + 1] = { 0x00, 0x00, 0x00, 0x00, 0x00 };
+static const char _mqtt_vbi_127[MQTT_VBI_MAXLEN + 1] = { 0x7F, 0x00, 0x00, 0x00, 0x00 };
+static const char _mqtt_vbi_128[MQTT_VBI_MAXLEN + 1] = { 0x80, 0x01, 0x00, 0x00, 0x00 };
+static const char _mqtt_vbi_16383[MQTT_VBI_MAXLEN + 1] = { 0xFF, 0x7F, 0x00, 0x00, 0x00 };
+static const char _mqtt_vbi_16384[MQTT_VBI_MAXLEN + 1] = { 0x80, 0x80, 0x01, 0x00, 0x00 };
+static const char _mqtt_vbi_2097151[MQTT_VBI_MAXLEN + 1] = { 0xFF, 0xFF, 0x7F, 0x00, 0x00 };
+static const char _mqtt_vbi_2097152[MQTT_VBI_MAXLEN + 1] = { 0x80, 0x80, 0x80, 0x01, 0x00 };
+static const char _mqtt_vbi_268435455[MQTT_VBI_MAXLEN + 1] = { 0xFF, 0xFF, 0xFF, 0x7F, 0x00 };
+static const char _mqtt_vbi_999999999[MQTT_VBI_MAXLEN + 1] = { 0x80, 0x80, 0x80, 0x80, 0x01 };
+
+#define MQTT_VBI_TESTCASE(case, expected_len) \
+    { \
+    memset(buf, 0, MQTT_VBI_MAXLEN + 1); \
+    int len; \
+    if ((len=uint32_to_mqtt_vbi(case, buf)) != expected_len) { \
+        fprintf(stderr, "uint32_to_mqtt_vbi(case:%d, line:%d): Incorrect length returned. Expected %d, Got %d\n", case, __LINE__, expected_len, len); \
+        return 1; \
+    } \
+    if (memcmp(buf, _mqtt_vbi_ ## case, MQTT_VBI_MAXLEN + 1 )) { \
+        fprintf(stderr, "uint32_to_mqtt_vbi(case:%d, line:%d): Wrong output\n", case, __LINE__); \
+        return 1; \
+    } }
+
+
+int test_uint32_mqtt_vbi() {
+    char buf[MQTT_VBI_MAXLEN + 1];
+
+    MQTT_VBI_TESTCASE(0,         1)
+    MQTT_VBI_TESTCASE(127,       1)
+    MQTT_VBI_TESTCASE(128,       2)
+    MQTT_VBI_TESTCASE(16383,     2)
+    MQTT_VBI_TESTCASE(16384,     3)
+    MQTT_VBI_TESTCASE(2097151,   3)
+    MQTT_VBI_TESTCASE(2097152,   4)
+    MQTT_VBI_TESTCASE(268435455, 4)
+
+    memset(buf, 0, MQTT_VBI_MAXLEN + 1);
+    int len;
+    if ((len=uint32_to_mqtt_vbi(268435456, buf)) != 0) {
+        fprintf(stderr, "uint32_to_mqtt_vbi(case:268435456, line:%d): Incorrect length returned. Expected 0, Got %d\n", __LINE__, len);
+        return 1;
+    }
+
+    return 0;
+}
+
+#define MQTT_VBI2UINT_TESTCASE(case, expected_error) \
+    { \
+    uint32_t result; \
+    int ret = mqtt_vbi_to_uint32(_mqtt_vbi_ ## case, &result); \
+    if (ret && !(expected_error)) { \
+        fprintf(stderr, "mqtt_vbi_to_uint(case:%d, line:%d): Unexpectedly Errored\n", (case), __LINE__); \
+        return 1; \
+    } \
+    if (!ret && (expected_error)) { \
+        fprintf(stderr, "mqtt_vbi_to_uint(case:%d, line:%d): Should return error but didnt\n", (case), __LINE__); \
+        return 1; \
+    } \
+    if (!ret && result != (case)) { \
+        fprintf(stderr, "mqtt_vbi_to_uint(case:%d, line:%d): Returned wrong result %d\n", (case), __LINE__, result); \
+        return 1; \
+    }}
+
+
+int test_mqtt_vbi_to_uint32() {
+    MQTT_VBI2UINT_TESTCASE(0,         0)
+    MQTT_VBI2UINT_TESTCASE(127,       0)
+    MQTT_VBI2UINT_TESTCASE(128,       0)
+    MQTT_VBI2UINT_TESTCASE(16383,     0)
+    MQTT_VBI2UINT_TESTCASE(16384,     0)
+    MQTT_VBI2UINT_TESTCASE(2097151,   0)
+    MQTT_VBI2UINT_TESTCASE(2097152,   0)
+    MQTT_VBI2UINT_TESTCASE(268435455, 0)
+    MQTT_VBI2UINT_TESTCASE(999999999, 1)
+    return 0;
+}
+#endif /* TESTS */
+
+// this helps with switch statements
+// as they have to use integer type (not pointer)
+enum memory_mode {
+    MEMCPY,
+    EXTERNAL_FREE_AFTER_USE,
+    CALLER_RESPONSIBLE
+};
+
+static inline enum memory_mode ptr2memory_mode(void * ptr) {
+    if (ptr == NULL)
+        return MEMCPY;
+    if (ptr == CALLER_RESPONSIBILITY)
+        return CALLER_RESPONSIBLE;
+    return EXTERNAL_FREE_AFTER_USE;
+}
+
+#define frag_is_marked_for_gc(frag) ((frag->flags & BUFFER_FRAG_GARBAGE_COLLECT) || ((frag->flags & BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND) && frag->sent == frag->len))
+#define FRAG_SIZE_IN_BUFFER(frag) (sizeof(struct buffer_fragment) + ((frag->flags & BUFFER_FRAG_DATA_EXTERNAL) ? 0 : frag->len))
+
+static void buffer_frag_free_data(struct buffer_fragment *frag)
+{
+    if ( frag->flags & BUFFER_FRAG_DATA_EXTERNAL && frag->data != NULL) {
+        switch (ptr2memory_mode(frag->free_fnc)) {
+            case MEMCPY:
+                mw_free(frag->data);
+                break;
+            case EXTERNAL_FREE_AFTER_USE:
+                frag->free_fnc(frag->data);
+                break;
+            case CALLER_RESPONSIBLE:
+                break;
+        }
+        frag->data = NULL;
+    }
+}
+
+#define HEADER_BUFFER_SIZE 1024*1024
+#define GROWTH_FACTOR 1.25
+
+#define BUFFER_BYTES_USED(buf) ((size_t)((buf)->tail - (buf)->data))
+#define BUFFER_BYTES_AVAILABLE(buf) ((buf)->size - BUFFER_BYTES_USED(buf))
+#define BUFFER_FIRST_FRAG(buf) ((struct buffer_fragment *)((buf)->tail_frag ? (buf)->data : NULL))
+static void buffer_purge(struct header_buffer *buf) {
+    struct buffer_fragment *frag = BUFFER_FIRST_FRAG(buf);
+    while (frag) {
+        buffer_frag_free_data(frag);
+        frag = frag->next;
+    }
+    buf->tail = buf->data;
+    buf->tail_frag = NULL;
+}
+
+#define FRAG_PADDING(addr) ((MQTT_WSS_FRAG_MEMALIGN - ((uintptr_t)addr % MQTT_WSS_FRAG_MEMALIGN)) % MQTT_WSS_FRAG_MEMALIGN)
+static struct buffer_fragment *buffer_new_frag(struct header_buffer *buf, buffer_frag_flag_t flags)
+{
+    uint8_t padding = FRAG_PADDING(buf->tail);
+
+    if (BUFFER_BYTES_AVAILABLE(buf) < sizeof(struct buffer_fragment) + padding)
+        return NULL;
+
+    struct buffer_fragment *frag = (struct buffer_fragment *)(buf->tail + padding);
+
+    memset(frag, 0, sizeof(*frag));
+    buf->tail += sizeof(*frag) + padding;
+
+    if (/*!((frag)->flags & BUFFER_FRAG_MQTT_PACKET_HEAD) &&*/ buf->tail_frag)
+        buf->tail_frag->next = frag;
+
+    buf->tail_frag = frag;
+
+    frag->data = buf->tail;
+
+    frag->flags = flags;
+
+    return frag;
+}
+
+static void buffer_rebuild(struct header_buffer *buf)
+{
+    struct buffer_fragment *frag = (struct buffer_fragment*)buf->data;
+    do {
+        buf->tail = (char*)frag + sizeof(struct buffer_fragment);
+        buf->tail_frag = frag;
+        if (!(frag->flags & BUFFER_FRAG_DATA_EXTERNAL)) {
+            buf->tail_frag->data = buf->tail;
+            buf->tail += frag->len;
+        }
+        if (frag->next != NULL)
+            frag->next = (struct buffer_fragment*)(buf->tail + FRAG_PADDING(buf->tail));
+        frag = frag->next;
+    } while(frag);
+}
+
+static void buffer_garbage_collect(struct header_buffer *buf, mqtt_wss_log_ctx_t log_ctx)
+{
+#if !defined(MQTT_DEBUG_VERBOSE) && !defined(ADDITIONAL_CHECKS)
+    (void) log_ctx;
+#endif
+#ifdef MQTT_DEBUG_VERBOSE
+        mws_debug(log_ctx, "Buffer Garbage Collection!");
+#endif
+
+    struct buffer_fragment *frag = BUFFER_FIRST_FRAG(buf);
+    while (frag) {
+        if (!frag_is_marked_for_gc(frag))
+            break;
+
+        buffer_frag_free_data(frag);
+
+        frag = frag->next;
+    }
+
+    if (frag == BUFFER_FIRST_FRAG(buf)) {
+#ifdef MQTT_DEBUG_VERBOSE
+        mws_debug(log_ctx, "Buffer Garbage Collection! No Space Reclaimed!");
+#endif
+        return;
+    }
+
+    if (!frag) {
+        buf->tail_frag = NULL;
+        buf->tail = buf->data;
+        return;
+    }
+
+#ifdef ADDITIONAL_CHECKS
+    if (!(frag->flags & BUFFER_FRAG_MQTT_PACKET_HEAD)) {
+        mws_error(log_ctx, "Expected to find end of buffer (NULL) or next packet head!");
+        return;
+    }
+#endif
+
+    memmove(buf->data, frag, buf->tail - (char*)frag);
+    buffer_rebuild(buf);
+}
+
+static void transaction_buffer_garbage_collect(struct transaction_buffer *buf, mqtt_wss_log_ctx_t log_ctx)
+{
+#ifdef MQTT_DEBUG_VERBOSE
+    mws_debug(log_ctx, "Transaction Buffer Garbage Collection! %s", buf->sending_frag == NULL ? "NULL" : "in flight message");
+#endif
+
+    // Invalidate the cached sending fragment
+    // as we will move data around
+    if (buf->sending_frag != &ping_frag)
+        buf->sending_frag = NULL;
+
+    buffer_garbage_collect(&buf->hdr_buffer, log_ctx);
+}
+
+static int transaction_buffer_grow(struct transaction_buffer *buf, mqtt_wss_log_ctx_t log_ctx, float rate, size_t max)
+{
+    if (buf->hdr_buffer.size >= max)
+        return 0;
+
+    // Invalidate the cached sending fragment
+    // as we will move data around
+    if (buf->sending_frag != &ping_frag)
+        buf->sending_frag = NULL;
+
+    buf->hdr_buffer.size *= rate;
+    if (buf->hdr_buffer.size > max)
+        buf->hdr_buffer.size = max;
+
+    void *ret = mw_realloc(buf->hdr_buffer.data, buf->hdr_buffer.size);
+    if (ret == NULL) {
+        mws_warn(log_ctx, "Buffer growth failed (realloc)");
+        return 1;
+    }
+
+    mws_debug(log_ctx, "Message metadata buffer was grown");
+
+    buf->hdr_buffer.data = ret;
+    buffer_rebuild(&buf->hdr_buffer);
+    return 0;
+}
+
+inline static int transaction_buffer_init(struct transaction_buffer *to_init, size_t size)
+{
+    pthread_mutex_init(&to_init->mutex, NULL);
+
+    to_init->hdr_buffer.size = size;
+    to_init->hdr_buffer.data = mw_malloc(size);
+    if (to_init->hdr_buffer.data == NULL)
+        return 1;
+
+    to_init->hdr_buffer.tail = to_init->hdr_buffer.data;
+    to_init->hdr_buffer.tail_frag = NULL;
+    return 0;
+}
+
+static void transaction_buffer_destroy(struct transaction_buffer *to_init)
+{
+    buffer_purge(&to_init->hdr_buffer);
+    pthread_mutex_destroy(&to_init->mutex);
+    mw_free(to_init->hdr_buffer.data);
+}
+
+// Creates transaction
+// saves state of buffer before any operation was done
+// allowing for rollback if things go wrong
+#define transaction_buffer_transaction_start(buf) \
+  { LOCK_HDR_BUFFER(buf); \
+    memcpy(&(buf)->state_backup, &(buf)->hdr_buffer, sizeof((buf)->hdr_buffer)); }
+
+#define transaction_buffer_transaction_commit(buf) UNLOCK_HDR_BUFFER(buf);
+
+void transaction_buffer_transaction_rollback(struct transaction_buffer *buf, struct buffer_fragment *frag)
+{
+    memcpy(&buf->hdr_buffer, &buf->state_backup, sizeof(buf->hdr_buffer));
+    if (buf->hdr_buffer.tail_frag != NULL)
+        buf->hdr_buffer.tail_frag->next = NULL;
+
+    while(frag) {
+        buffer_frag_free_data(frag);
+        // we are not actually freeing the structure itself
+        // just the data it manages
+        // structure itself is in permanent buffer
+        // which is locked by HDR_BUFFER lock
+        frag = frag->next;
+    }
+
+    UNLOCK_HDR_BUFFER(buf);
+}
+
+#define TX_ALIASES_INITIALIZE() c_rhash_new(0)
+#define RX_ALIASES_INITIALIZE() c_rhash_new(UINT16_MAX >> 8)
+struct mqtt_ng_client *mqtt_ng_init(struct mqtt_ng_init *settings)
+{
+    struct mqtt_ng_client *client = mw_calloc(1, sizeof(struct mqtt_ng_client));
+    if (client == NULL)
+        return NULL;
+
+    if (transaction_buffer_init(&client->main_buffer, HEADER_BUFFER_SIZE))
+        goto err_free_client;
+
+    client->rx_aliases = RX_ALIASES_INITIALIZE();
+    if (client->rx_aliases == NULL)
+        goto err_free_trx_buf;
+
+    if (pthread_mutex_init(&client->stats_mutex, NULL))
+        goto err_free_rx_alias;
+
+    client->tx_topic_aliases.stoi_dict = TX_ALIASES_INITIALIZE();
+    if (client->tx_topic_aliases.stoi_dict == NULL)
+        goto err_free_stats_mutex;
+    client->tx_topic_aliases.idx_max = UINT16_MAX;
+
+    if (pthread_rwlock_init(&client->tx_topic_aliases.rwlock, NULL))
+        goto err_free_tx_alias;
+
+    // TODO just embed the struct into mqtt_ng_client
+    client->parser.received_data = settings->data_in;
+    client->send_fnc_ptr = settings->data_out_fnc;
+    client->user_ctx = settings->user_ctx;
+
+    client->log = settings->log;
+
+    client->puback_callback = settings->puback_callback;
+    client->connack_callback = settings->connack_callback;
+    client->msg_callback = settings->msg_callback;
+
+    return client;
+
+err_free_tx_alias:
+    c_rhash_destroy(client->tx_topic_aliases.stoi_dict);
+err_free_stats_mutex:
+    pthread_mutex_destroy(&client->stats_mutex);
+err_free_rx_alias:
+    c_rhash_destroy(client->rx_aliases);
+err_free_trx_buf:
+    transaction_buffer_destroy(&client->main_buffer);
+err_free_client:
+    mw_free(client);
+    return NULL;
+}
+
+static inline uint8_t get_control_packet_type(uint8_t first_hdr_byte)
+{
+    return first_hdr_byte >> 4;
+}
+
+static void mqtt_ng_destroy_rx_alias_hash(c_rhash hash)
+{
+    c_rhash_iter_t i = C_RHASH_ITER_T_INITIALIZER;
+    uint64_t stored_key;
+    void *to_free;
+    while(!c_rhash_iter_uint64_keys(hash, &i, &stored_key)) {
+        c_rhash_get_ptr_by_uint64(hash, stored_key, &to_free);
+        mw_free(to_free);
+    }
+    c_rhash_destroy(hash);
+}
+
+static void mqtt_ng_destroy_tx_alias_hash(c_rhash hash)
+{
+    c_rhash_iter_t i = C_RHASH_ITER_T_INITIALIZER;
+    const char *stored_key;
+    void *to_free;
+    while(!c_rhash_iter_str_keys(hash, &i, &stored_key)) {
+        c_rhash_get_ptr_by_str(hash, stored_key, &to_free);
+        mw_free(to_free);
+    }
+    c_rhash_destroy(hash);
+}
+
+void mqtt_ng_destroy(struct mqtt_ng_client *client)
+{
+    transaction_buffer_destroy(&client->main_buffer);
+    pthread_mutex_destroy(&client->stats_mutex);
+
+    mqtt_ng_destroy_tx_alias_hash(client->tx_topic_aliases.stoi_dict);
+    pthread_rwlock_destroy(&client->tx_topic_aliases.rwlock);
+    mqtt_ng_destroy_rx_alias_hash(client->rx_aliases);
+
+    mw_free(client);
+}
+
+int frag_set_external_data(mqtt_wss_log_ctx_t log, struct buffer_fragment *frag, void *data, size_t data_len, free_fnc_t data_free_fnc)
+{
+    if (frag->len) {
+        // TODO?: This could potentially be done in future if we set rule
+        // external data always follows in buffer data
+        // could help reduce fragmentation in some messages but
+        // currently not worth it considering time is tight
+        mws_fatal(log, UNIT_LOG_PREFIX "INTERNAL ERROR: Cannot set external data to fragment already containing in buffer data!");
+        return 1;
+    }
+
+    switch (ptr2memory_mode(data_free_fnc)) {
+        case MEMCPY:
+            frag->data = mw_malloc(data_len);
+            if (frag->data == NULL) {
+                mws_error(log, UNIT_LOG_PREFIX "OOM while malloc @_optimized_add");
+                return 1;
+            }
+            memcpy(frag->data, data, data_len);
+            break;
+        case EXTERNAL_FREE_AFTER_USE:
+        case CALLER_RESPONSIBLE:
+            frag->data = data;
+            break;
+    }
+    frag->free_fnc = data_free_fnc;
+    frag->len = data_len;
+
+    frag->flags |= BUFFER_FRAG_DATA_EXTERNAL;
+    return 0;
+ }
+
+// this is fixed part of variable header for connect packet
+// mqtt-v5.0-cs1, 3.1.2.1, 2.1.2.2
+static const char mqtt_protocol_name_frag[] =
+    { 0x00, 0x04, 'M', 'Q', 'T', 'T', MQTT_VERSION_5_0 };
+
+#define MQTT_UTF8_STRING_SIZE(string) (2 + strlen(string))
+
+// see 1.5.5
+#define MQTT_VARSIZE_INT_BYTES(value) ( value > 2097152 ? 4 : ( value > 16384 ? 3 : ( value > 128 ? 2 : 1 ) ) )
+
+static size_t mqtt_ng_connect_size(struct mqtt_auth_properties *auth,
+                    struct mqtt_lwt_properties *lwt)
+{
+    // First get the size of payload + variable header
+    size_t size =
+        + sizeof(mqtt_protocol_name_frag) /* Proto Name and Version */
+        + 1 /* Connect Flags */
+        + 2 /* Keep Alive */
+        + 4 /* 3.1.2.11.1 Property Length - for now fixed to only Topic Alias Maximum, TODO TODO*/;
+
+    // CONNECT payload. 3.1.3
+    if (auth->client_id)
+        size += MQTT_UTF8_STRING_SIZE(auth->client_id);
+
+    if (lwt) {
+        // 3.1.3.2 will properties TODO TODO
+        size += 1;
+
+        // 3.1.3.3
+        if (lwt->will_topic)
+            size += MQTT_UTF8_STRING_SIZE(lwt->will_topic);
+
+        // 3.1.3.4 will payload
+        if (lwt->will_message) {
+            size += 2 + lwt->will_message_size;
+        }
+    }
+
+    // 3.1.3.5
+    if (auth->username)
+        size += MQTT_UTF8_STRING_SIZE(auth->username);
+
+    // 3.1.3.6
+    if (auth->password)
+        size += MQTT_UTF8_STRING_SIZE(auth->password);
+
+    return size;
+}
+
+#define BUFFER_TRANSACTION_NEW_FRAG(buf, flags, frag, on_fail) \
+    { if(frag==NULL) { \
+        frag = buffer_new_frag(buf, (flags)); } \
+      if(frag==NULL) { on_fail; }}
+
+#define CHECK_BYTES_AVAILABLE(buf, needed, fail) \
+    { if (BUFFER_BYTES_AVAILABLE(buf) < (size_t)needed) { \
+        fail; } }
+
+#define DATA_ADVANCE(buf, bytes, frag) { size_t b = (bytes); (buf)->tail += b; (frag)->len += b; }
+
+// TODO maybe just user client->buf.tail?
+#define WRITE_POS(frag) (&(frag->data[frag->len]))
+
+// [MQTT-1.5.2] Two Byte Integer
+#define PACK_2B_INT(buffer, integer, frag) { *(uint16_t *)WRITE_POS(frag) = htobe16((integer)); \
+            DATA_ADVANCE(buffer, sizeof(uint16_t), frag); }
+
+static int _optimized_add(struct header_buffer *buf, mqtt_wss_log_ctx_t log_ctx, void *data, size_t data_len, free_fnc_t data_free_fnc, struct buffer_fragment **frag)
+{
+    if (data_len > SMALL_STRING_DONT_FRAGMENT_LIMIT) {
+        buffer_frag_flag_t flags = BUFFER_FRAG_DATA_EXTERNAL;
+        if ((*frag)->flags & BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND)
+            flags |= BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND;
+        if( (*frag = buffer_new_frag(buf, flags)) == NULL ) {
+            mws_error(log_ctx, "Out of buffer space while generating the message");
+            return 1;
+        }
+        if (frag_set_external_data(log_ctx, *frag, data, data_len, data_free_fnc)) {
+            mws_error(log_ctx, "Error adding external data to newly created fragment");
+            return 1;
+        }
+        // we dont want to write to this fragment anymore
+        *frag = NULL;
+    } else if (data_len) {
+        // if the data are small dont bother creating new fragments
+        // store in buffer directly
+        CHECK_BYTES_AVAILABLE(buf, data_len, return 1);
+        memcpy(buf->tail, data, data_len);
+        DATA_ADVANCE(buf, data_len, *frag);
+    }
+    return 0;
+}
+
+#define TRY_GENERATE_MESSAGE(generator_function, client, ...) \
+    int rc = generator_function(&client->main_buffer, client->log, ##__VA_ARGS__); \
+    if (rc == MQTT_NG_MSGGEN_BUFFER_OOM) { \
+        LOCK_HDR_BUFFER(&client->main_buffer); \
+        transaction_buffer_garbage_collect((&client->main_buffer), client->log); \
+        UNLOCK_HDR_BUFFER(&client->main_buffer); \
+        rc = generator_function(&client->main_buffer, client->log, ##__VA_ARGS__); \
+        if (rc == MQTT_NG_MSGGEN_BUFFER_OOM && client->max_mem_bytes) { \
+            LOCK_HDR_BUFFER(&client->main_buffer); \
+            transaction_buffer_grow((&client->main_buffer), client->log, GROWTH_FACTOR, client->max_mem_bytes); \
+            UNLOCK_HDR_BUFFER(&client->main_buffer); \
+            rc = generator_function(&client->main_buffer, client->log, ##__VA_ARGS__); \
+        } \
+        if (rc == MQTT_NG_MSGGEN_BUFFER_OOM) \
+            mws_error(client->log, "%s failed to generate message due to insufficient buffer space (line %d)", __FUNCTION__, __LINE__); \
+    } \
+    if (rc == MQTT_NG_MSGGEN_OK) { \
+        pthread_mutex_lock(&client->stats_mutex); \
+        client->stats.tx_messages_queued++; \
+        pthread_mutex_unlock(&client->stats_mutex); \
+    } \
+    return rc;
+
+mqtt_msg_data mqtt_ng_generate_connect(struct transaction_buffer *trx_buf,
+                                       mqtt_wss_log_ctx_t log_ctx,
+                                       struct mqtt_auth_properties *auth,
+                                       struct mqtt_lwt_properties *lwt,
+                                       uint8_t clean_start,
+                                       uint16_t keep_alive)
+{
+    // Sanity Checks First (are given parameters correct and up to MQTT spec)
+    if (!auth->client_id) {
+        mws_error(log_ctx, "ClientID must be set. [MQTT-3.1.3-3]");
+        return NULL;
+    }
+
+    size_t len = strlen(auth->client_id);
+    if (!len) {
+        // [MQTT-3.1.3-6] server MAY allow empty client_id and treat it
+        // as specific client_id (not same as client_id not given)
+        // however server MUST allow ClientIDs between 1-23 bytes [MQTT-3.1.3-5]
+        // so we will warn client server might not like this and he is using it
+        // at his own risk!
+        mws_warn(log_ctx, "client_id provided is empty string. This might not be allowed by server [MQTT-3.1.3-6]");
+    }
+    if(len > MQTT_MAX_CLIENT_ID) {
+        // [MQTT-3.1.3-5] server MUST allow client_id length 1-32
+        // server MAY allow longer client_id, if user provides longer client_id
+        // warn them he is doing so at his own risk!
+        mws_warn(log_ctx, "client_id provided is longer than 23 bytes, server might not allow that [MQTT-3.1.3-5]");
+    }
+
+    if (lwt) {
+        if (lwt->will_message && lwt->will_message_size > 65535) {
+            mws_error(log_ctx, "Will message cannot be longer than 65535 bytes due to MQTT protocol limitations [MQTT-3.1.3-4] and [MQTT-1.5.6]");
+            return NULL;
+        }
+
+        if (!lwt->will_topic) { //TODO topic given with strlen==0 ? check specs
+            mws_error(log_ctx, "If will message is given will topic must also be given [MQTT-3.1.3.3]");
+            return NULL;
+        }
+
+        if (lwt->will_qos > MQTT_MAX_QOS) {
+            // refer to [MQTT-3-1.2-12]
+            mws_error(log_ctx, "QOS for LWT message is bigger than max");
+            return NULL;
+        }
+    }
+
+    // >> START THE RODEO <<
+    transaction_buffer_transaction_start(trx_buf);
+
+    // Calculate the resulting message size sans fixed MQTT header
+    size_t size = mqtt_ng_connect_size(auth, lwt);
+
+    // Start generating the message
+    struct buffer_fragment *frag = NULL;
+    mqtt_msg_data ret = NULL;
+
+    BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, BUFFER_FRAG_MQTT_PACKET_HEAD, frag, goto fail_rollback );
+    ret = frag;
+
+    // MQTT Fixed Header
+    size_t needed_bytes = 1 /* Packet type */ + MQTT_VARSIZE_INT_BYTES(size) + sizeof(mqtt_protocol_name_frag) + 1 /* CONNECT FLAGS */ + 2 /* keepalive */ + 1 /* Properties TODO now fixed 0*/;
+    CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, needed_bytes, goto fail_rollback);
+
+    *WRITE_POS(frag) = MQTT_CPT_CONNECT << 4;
+    DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+    DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(size, WRITE_POS(frag)), frag);
+
+    memcpy(WRITE_POS(frag), mqtt_protocol_name_frag, sizeof(mqtt_protocol_name_frag));
+    DATA_ADVANCE(&trx_buf->hdr_buffer, sizeof(mqtt_protocol_name_frag), frag);
+
+    // [MQTT-3.1.2.3] Connect flags
+    char *connect_flags = WRITE_POS(frag);
+    *connect_flags = 0;
+    if (auth->username)
+        *connect_flags |= MQTT_CONNECT_FLAG_USERNAME;
+    if (auth->password)
+        *connect_flags |= MQTT_CONNECT_FLAG_PASSWORD;
+    if (lwt) {
+        *connect_flags |= MQTT_CONNECT_FLAG_LWT;
+        *connect_flags |= lwt->will_qos << MQTT_CONNECT_FLAG_QOS_BITSHIFT;
+        if (lwt->will_retain)
+            *connect_flags |= MQTT_CONNECT_FLAG_LWT_RETAIN;
+    }
+    if (clean_start)
+        *connect_flags |= MQTT_CONNECT_FLAG_CLEAN_START;
+
+    DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+
+    PACK_2B_INT(&trx_buf->hdr_buffer, keep_alive, frag);
+
+    // TODO Property Length [MQTT-3.1.3.2.1] temporary fixed to 3 (one property topic alias max)
+    DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(3, WRITE_POS(frag)), frag);
+    *WRITE_POS(frag) = MQTT_PROP_TOPIC_ALIAS_MAX;
+    DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+
+    PACK_2B_INT(&trx_buf->hdr_buffer, 65535, frag);
+
+    // [MQTT-3.1.3.1] Client identifier
+    CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 2, goto fail_rollback);
+    PACK_2B_INT(&trx_buf->hdr_buffer, strlen(auth->client_id), frag);
+    if (_optimized_add(&trx_buf->hdr_buffer, log_ctx, auth->client_id, strlen(auth->client_id), auth->client_id_free, &frag))
+        goto fail_rollback;
+
+    if (lwt != NULL) {
+        // Will Properties [MQTT-3.1.3.2]
+        // TODO for now fixed 0
+        BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback);
+        CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 1, goto fail_rollback);
+        *WRITE_POS(frag) = 0;
+        DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+
+        // Will Topic [MQTT-3.1.3.3]
+        CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 2, goto fail_rollback);
+        PACK_2B_INT(&trx_buf->hdr_buffer, strlen(lwt->will_topic), frag);
+        if (_optimized_add(&trx_buf->hdr_buffer, log_ctx, lwt->will_topic, strlen(lwt->will_topic), lwt->will_topic_free, &frag))
+            goto fail_rollback;
+
+        // Will Payload [MQTT-3.1.3.4]
+        if (lwt->will_message_size) {
+            BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback);
+            CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 2, goto fail_rollback);
+            PACK_2B_INT(&trx_buf->hdr_buffer, lwt->will_message_size, frag);
+            if (_optimized_add(&trx_buf->hdr_buffer, log_ctx, lwt->will_message, lwt->will_message_size, lwt->will_topic_free, &frag))
+                goto fail_rollback;
+        }
+    }
+
+    // [MQTT-3.1.3.5]
+    if (auth->username) {
+        BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback);
+        CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 2, goto fail_rollback);
+        PACK_2B_INT(&trx_buf->hdr_buffer, strlen(auth->username), frag);
+        if (_optimized_add(&trx_buf->hdr_buffer, log_ctx, auth->username, strlen(auth->username), auth->username_free, &frag))
+            goto fail_rollback;
+    }
+
+    // [MQTT-3.1.3.6]
+    if (auth->password) {
+        BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback);
+        CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, 2, goto fail_rollback);
+        PACK_2B_INT(&trx_buf->hdr_buffer, strlen(auth->password), frag);
+        if (_optimized_add(&trx_buf->hdr_buffer, log_ctx, auth->password, strlen(auth->password), auth->password_free, &frag))
+            goto fail_rollback;
+    }
+    trx_buf->hdr_buffer.tail_frag->flags |= BUFFER_FRAG_MQTT_PACKET_TAIL;
+    transaction_buffer_transaction_commit(trx_buf);
+    return ret;
+fail_rollback:
+    transaction_buffer_transaction_rollback(trx_buf, ret);
+    return NULL;
+}
+
+int mqtt_ng_connect(struct mqtt_ng_client *client,
+                    struct mqtt_auth_properties *auth,
+                    struct mqtt_lwt_properties *lwt,
+                    uint8_t clean_start,
+                    uint16_t keep_alive)
+{
+    client->client_state = RAW;
+    client->parser.state = MQTT_PARSE_FIXED_HEADER_PACKET_TYPE;
+
+    LOCK_HDR_BUFFER(&client->main_buffer);
+    client->main_buffer.sending_frag = NULL;
+    if (clean_start)
+        buffer_purge(&client->main_buffer.hdr_buffer);
+    UNLOCK_HDR_BUFFER(&client->main_buffer);
+
+    pthread_rwlock_wrlock(&client->tx_topic_aliases.rwlock);
+    // according to MQTT spec topic aliases should not be persisted
+    // even if clean session is true
+    mqtt_ng_destroy_tx_alias_hash(client->tx_topic_aliases.stoi_dict);
+    client->tx_topic_aliases.stoi_dict = TX_ALIASES_INITIALIZE();
+    if (client->tx_topic_aliases.stoi_dict == NULL) {
+        pthread_rwlock_unlock(&client->tx_topic_aliases.rwlock);
+        return 1;
+    }
+    client->tx_topic_aliases.idx_assigned = 0;
+    pthread_rwlock_unlock(&client->tx_topic_aliases.rwlock);
+
+    mqtt_ng_destroy_rx_alias_hash(client->rx_aliases);
+    client->rx_aliases = RX_ALIASES_INITIALIZE();
+    if (client->rx_aliases == NULL)
+        return 1;
+
+    client->connect_msg = mqtt_ng_generate_connect(&client->main_buffer, client->log, auth, lwt, clean_start, keep_alive);
+    if (client->connect_msg == NULL)
+        return 1;
+
+    pthread_mutex_lock(&client->stats_mutex);
+    if (clean_start)
+        client->stats.tx_messages_queued = 1;
+    else
+        client->stats.tx_messages_queued++;
+
+    client->stats.tx_messages_sent = 0;
+    client->stats.rx_messages_rcvd = 0;
+    pthread_mutex_unlock(&client->stats_mutex);
+
+    client->client_state = CONNECT_PENDING;
+    return 0;
+}
+
+uint16_t get_unused_packet_id() {
+    static uint16_t packet_id = 0;
+    packet_id++;
+    return packet_id ? packet_id : ++packet_id;
+}
+
+static inline size_t mqtt_ng_publish_size(const char *topic,
+                            size_t msg_len,
+                            uint16_t topic_id)
+{
+    size_t retval = 2 /* Topic Name Length */
+        + (topic == NULL ? 0 : strlen(topic))
+        + 2 /* Packet identifier */
+        + 1 /* Properties Length TODO for now fixed to 1 property */
+        + msg_len;
+
+    if (topic_id)
+        retval += 3;
+
+    return retval;
+}
+
+int mqtt_ng_generate_publish(struct transaction_buffer *trx_buf,
+                             mqtt_wss_log_ctx_t log_ctx,
+                             char *topic,
+                             free_fnc_t topic_free,
+                             void *msg,
+                             free_fnc_t msg_free,
+                             size_t msg_len,
+                             uint8_t publish_flags,
+                             uint16_t *packet_id,
+                             uint16_t topic_alias)
+{
+    // >> START THE RODEO <<
+    transaction_buffer_transaction_start(trx_buf);
+
+    // Calculate the resulting message size sans fixed MQTT header
+    size_t size = mqtt_ng_publish_size(topic, msg_len, topic_alias);
+
+    // Start generating the message
+    struct buffer_fragment *frag = NULL;
+    mqtt_msg_data mqtt_msg = NULL;
+
+    BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, BUFFER_FRAG_MQTT_PACKET_HEAD, frag, goto fail_rollback );
+    // in case of QOS 0 we can garbage collect immediatelly after sending
+    uint8_t qos = (publish_flags >> 1) & 0x03;
+    if (!qos)
+        frag->flags |= BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND;
+    mqtt_msg = frag;
+
+    // MQTT Fixed Header
+    size_t needed_bytes = 1 /* Packet type */ + MQTT_VARSIZE_INT_BYTES(size) + size - msg_len;
+    CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, needed_bytes, goto fail_rollback);
+
+    *WRITE_POS(frag) = (MQTT_CPT_PUBLISH << 4) | (publish_flags & 0xF);
+    DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+    DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(size, WRITE_POS(frag)), frag);
+
+    // MQTT Variable Header
+    // [MQTT-3.3.2.1]
+    PACK_2B_INT(&trx_buf->hdr_buffer, topic == NULL ? 0 : strlen(topic), frag);
+    if (topic != NULL) {
+        if (_optimized_add(&trx_buf->hdr_buffer, log_ctx, topic, strlen(topic), topic_free, &frag))
+            goto fail_rollback;
+        BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback);
+    }
+
+    // [MQTT-3.3.2.2]
+    mqtt_msg->packet_id = get_unused_packet_id();
+    *packet_id = mqtt_msg->packet_id;
+    PACK_2B_INT(&trx_buf->hdr_buffer, mqtt_msg->packet_id, frag);
+
+    // [MQTT-3.3.2.3.1] TODO Property Length for now fixed 0
+    *WRITE_POS(frag) = topic_alias ? 3 : 0;
+    DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+
+    if(topic_alias) {
+        *WRITE_POS(frag) = MQTT_PROP_TOPIC_ALIAS;
+        DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+
+        PACK_2B_INT(&trx_buf->hdr_buffer, topic_alias, frag);
+    }
+
+    if( (frag = buffer_new_frag(&trx_buf->hdr_buffer, BUFFER_FRAG_DATA_EXTERNAL)) == NULL )
+        goto fail_rollback;
+
+    if (frag_set_external_data(log_ctx, frag, msg, msg_len, msg_free))
+        goto fail_rollback;
+
+    trx_buf->hdr_buffer.tail_frag->flags |= BUFFER_FRAG_MQTT_PACKET_TAIL;
+    if (!qos)
+        trx_buf->hdr_buffer.tail_frag->flags |= BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND;
+    transaction_buffer_transaction_commit(trx_buf);
+    return MQTT_NG_MSGGEN_OK;
+fail_rollback:
+    transaction_buffer_transaction_rollback(trx_buf, mqtt_msg);
+    return MQTT_NG_MSGGEN_BUFFER_OOM;
+}
+
+#define PUBLISH_SP_SIZE 64
+int mqtt_ng_publish(struct mqtt_ng_client *client,
+                    char *topic,
+                    free_fnc_t topic_free,
+                    void *msg,
+                    free_fnc_t msg_free,
+                    size_t msg_len,
+                    uint8_t publish_flags,
+                    uint16_t *packet_id)
+{
+    struct topic_alias_data *alias = NULL;
+    pthread_rwlock_rdlock(&client->tx_topic_aliases.rwlock);
+    c_rhash_get_ptr_by_str(client->tx_topic_aliases.stoi_dict, topic, (void**)&alias);
+    pthread_rwlock_unlock(&client->tx_topic_aliases.rwlock);
+
+    uint16_t topic_id = 0;
+
+    if (alias != NULL) {
+        topic_id = alias->idx;
+        uint32_t cnt = __atomic_fetch_add(&alias->usage_count, 1, __ATOMIC_SEQ_CST);
+        if (cnt) {
+            topic = NULL;
+            topic_free = NULL;
+        }
+    }
+
+    if (client->max_msg_size && PUBLISH_SP_SIZE + mqtt_ng_publish_size(topic, msg_len, topic_id) > client->max_msg_size) {
+        mws_error(client->log, "Message too big for server: %zu", msg_len);
+        return MQTT_NG_MSGGEN_MSG_TOO_BIG;
+    }
+
+    TRY_GENERATE_MESSAGE(mqtt_ng_generate_publish, client, topic, topic_free, msg, msg_free, msg_len, publish_flags, packet_id, topic_id);
+}
+
+static inline size_t mqtt_ng_subscribe_size(struct mqtt_sub *subs, size_t sub_count)
+{
+    size_t len = 2 /* Packet Identifier */ + 1 /* Properties Length TODO for now fixed 0 */;
+    len += sub_count * (2 /* topic filter string length */ + 1 /* [MQTT-3.8.3.1] Subscription Options Byte */);
+
+    for (size_t i = 0; i < sub_count; i++) {
+        len += strlen(subs[i].topic);
+    }
+    return len;
+}
+
+int mqtt_ng_generate_subscribe(struct transaction_buffer *trx_buf, mqtt_wss_log_ctx_t log_ctx, struct mqtt_sub *subs, size_t sub_count)
+{
+    // >> START THE RODEO <<
+    transaction_buffer_transaction_start(trx_buf);
+
+    // Calculate the resulting message size sans fixed MQTT header
+    size_t size = mqtt_ng_subscribe_size(subs, sub_count);
+
+    // Start generating the message
+    struct buffer_fragment *frag = NULL;
+    mqtt_msg_data ret = NULL;
+
+    BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, BUFFER_FRAG_MQTT_PACKET_HEAD, frag, goto fail_rollback);
+    ret = frag;
+
+    // MQTT Fixed Header
+    size_t needed_bytes = 1 /* Packet type */ + MQTT_VARSIZE_INT_BYTES(size) + 3 /*Packet ID + Property Length*/;
+    CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, needed_bytes, goto fail_rollback);
+
+    *WRITE_POS(frag) = (MQTT_CPT_SUBSCRIBE << 4) | 0x2 /* [MQTT-3.8.1-1] */;
+    DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+    DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(size, WRITE_POS(frag)), frag);
+
+    // MQTT Variable Header
+    // [MQTT-3.8.2] PacketID
+    ret->packet_id = get_unused_packet_id();
+    PACK_2B_INT(&trx_buf->hdr_buffer, ret->packet_id, frag);
+
+    // [MQTT-3.8.2.1.1] Property Length // TODO for now fixed 0
+    *WRITE_POS(frag) = 0;
+    DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+
+    for (size_t i = 0; i < sub_count; i++) {
+        BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback);
+        PACK_2B_INT(&trx_buf->hdr_buffer, strlen(subs[i].topic), frag);
+        if (_optimized_add(&trx_buf->hdr_buffer, log_ctx, subs[i].topic, strlen(subs[i].topic), subs[i].topic_free, &frag))
+            goto fail_rollback;
+        BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, 0, frag, goto fail_rollback);
+        *WRITE_POS(frag) = subs[i].options;
+        DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+    }
+
+    trx_buf->hdr_buffer.tail_frag->flags |= BUFFER_FRAG_MQTT_PACKET_TAIL;
+    transaction_buffer_transaction_commit(trx_buf);
+    return MQTT_NG_MSGGEN_OK;
+fail_rollback:
+    transaction_buffer_transaction_rollback(trx_buf, ret);
+    return MQTT_NG_MSGGEN_BUFFER_OOM;
+}
+
+int mqtt_ng_subscribe(struct mqtt_ng_client *client, struct mqtt_sub *subs, size_t sub_count)
+{
+    TRY_GENERATE_MESSAGE(mqtt_ng_generate_subscribe, client, subs, sub_count);
+}
+
+int mqtt_ng_generate_disconnect(struct transaction_buffer *trx_buf, mqtt_wss_log_ctx_t log_ctx, uint8_t reason_code)
+{
+    (void) log_ctx;
+    // >> START THE RODEO <<
+    transaction_buffer_transaction_start(trx_buf);
+
+    // Calculate the resulting message size sans fixed MQTT header
+    size_t size = reason_code ? 1 : 0;
+
+    // Start generating the message
+    struct buffer_fragment *frag = NULL;
+    mqtt_msg_data ret = NULL;
+
+    BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, BUFFER_FRAG_MQTT_PACKET_HEAD, frag, goto fail_rollback);
+    ret = frag;
+
+    // MQTT Fixed Header
+    size_t needed_bytes = 1 /* Packet type */ + MQTT_VARSIZE_INT_BYTES(size) + (reason_code ? 1 : 0);
+    CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, needed_bytes, goto fail_rollback);
+
+    *WRITE_POS(frag) = MQTT_CPT_DISCONNECT << 4;
+    DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+    DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(size, WRITE_POS(frag)), frag);
+
+    if (reason_code) {
+        // MQTT Variable Header
+        // [MQTT-3.14.2.1] PacketID
+        *WRITE_POS(frag) = reason_code;
+        DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+    }
+
+    trx_buf->hdr_buffer.tail_frag->flags |= BUFFER_FRAG_MQTT_PACKET_TAIL;
+    transaction_buffer_transaction_commit(trx_buf);
+    return MQTT_NG_MSGGEN_OK;
+fail_rollback:
+    transaction_buffer_transaction_rollback(trx_buf, ret);
+    return MQTT_NG_MSGGEN_BUFFER_OOM;
+}
+
+int mqtt_ng_disconnect(struct mqtt_ng_client *client, uint8_t reason_code)
+{
+    TRY_GENERATE_MESSAGE(mqtt_ng_generate_disconnect, client, reason_code);
+}
+
+static int mqtt_generate_puback(struct transaction_buffer *trx_buf, mqtt_wss_log_ctx_t log_ctx, uint16_t packet_id, uint8_t reason_code)
+{
+    (void) log_ctx;
+    // >> START THE RODEO <<
+    transaction_buffer_transaction_start(trx_buf);
+
+    // Calculate the resulting message size sans fixed MQTT header
+    size_t size = 2 /* Packet ID */ + (reason_code ? 1 : 0) /* reason code */;
+
+    // Start generating the message
+    struct buffer_fragment *frag = NULL;
+
+    BUFFER_TRANSACTION_NEW_FRAG(&trx_buf->hdr_buffer, BUFFER_FRAG_MQTT_PACKET_HEAD | BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND, frag, goto fail_rollback);
+
+    // MQTT Fixed Header
+    size_t needed_bytes = 1 /* Packet type */ + MQTT_VARSIZE_INT_BYTES(size) + size;
+    CHECK_BYTES_AVAILABLE(&trx_buf->hdr_buffer, needed_bytes, goto fail_rollback);
+
+    *WRITE_POS(frag) = MQTT_CPT_PUBACK << 4;
+    DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+    DATA_ADVANCE(&trx_buf->hdr_buffer, uint32_to_mqtt_vbi(size, WRITE_POS(frag)), frag);
+
+    // MQTT Variable Header
+    PACK_2B_INT(&trx_buf->hdr_buffer, packet_id, frag);
+
+    if (reason_code) {
+        // MQTT Variable Header
+        // [MQTT-3.14.2.1] PacketID
+        *WRITE_POS(frag) = reason_code;
+        DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag);
+    }
+
+    trx_buf->hdr_buffer.tail_frag->flags |= BUFFER_FRAG_MQTT_PACKET_TAIL;
+    transaction_buffer_transaction_commit(trx_buf);
+    return MQTT_NG_MSGGEN_OK;
+fail_rollback:
+    transaction_buffer_transaction_rollback(trx_buf, frag);
+    return MQTT_NG_MSGGEN_BUFFER_OOM;
+}
+
+static int mqtt_ng_puback(struct mqtt_ng_client *client, uint16_t packet_id, uint8_t reason_code)
+{
+    TRY_GENERATE_MESSAGE(mqtt_generate_puback, client, packet_id, reason_code);
+}
+
+int mqtt_ng_ping(struct mqtt_ng_client *client)
+{
+    client->ping_pending = 1;
+    return MQTT_NG_MSGGEN_OK;
+}
+
+#define MQTT_NG_CLIENT_NEED_MORE_BYTES         0x10
+#define MQTT_NG_CLIENT_MQTT_PACKET_DONE        0x11
+#define MQTT_NG_CLIENT_PARSE_DONE              0x12
+#define MQTT_NG_CLIENT_WANT_WRITE              0x13
+#define MQTT_NG_CLIENT_OK_CALL_AGAIN           0
+#define MQTT_NG_CLIENT_PROTOCOL_ERROR         -1
+#define MQTT_NG_CLIENT_SERVER_RETURNED_ERROR  -2
+#define MQTT_NG_CLIENT_NOT_IMPL_YET           -3
+#define MQTT_NG_CLIENT_OOM                    -4
+#define MQTT_NG_CLIENT_INTERNAL_ERROR         -5
+
+#define BUF_READ_CHECK_AT_LEAST(buf, x)                 \
+    if (rbuf_bytes_available(buf) < (x)) \
+        return MQTT_NG_CLIENT_NEED_MORE_BYTES;
+
+#define vbi_parser_reset_ctx(ctx) memset(ctx, 0, sizeof(struct mqtt_vbi_parser_ctx))
+
+static int vbi_parser_parse(struct mqtt_vbi_parser_ctx *ctx, rbuf_t data, mqtt_wss_log_ctx_t log)
+{
+    if (ctx->bytes > MQTT_VBI_MAXBYTES - 1) {
+        mws_error(log, "MQTT Variable Byte Integer can't be longer than %d bytes", MQTT_VBI_MAXBYTES);
+        return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+    }
+    if (!ctx->bytes || ctx->data[ctx->bytes-1] & MQTT_VBI_CONTINUATION_FLAG) {
+        BUF_READ_CHECK_AT_LEAST(data, 1);
+        ctx->bytes++;
+        rbuf_pop(data, &ctx->data[ctx->bytes-1], 1);
+        if ( ctx->data[ctx->bytes-1] & MQTT_VBI_CONTINUATION_FLAG )
+            return MQTT_NG_CLIENT_OK_CALL_AGAIN;
+    }
+
+    if (mqtt_vbi_to_uint32(ctx->data, &ctx->result)) {
+            mws_error(log, "MQTT Variable Byte Integer failed to be parsed.");
+            return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+    }
+
+    return MQTT_NG_CLIENT_PARSE_DONE;
+}
+
+static void mqtt_properties_parser_ctx_reset(struct mqtt_properties_parser_ctx *ctx)
+{
+    ctx->state = PROPERTIES_LENGTH;
+    while (ctx->head) {
+        struct mqtt_property *f = ctx->head;
+        ctx->head = ctx->head->next;
+        if (f->type == MQTT_TYPE_STR || f->type == MQTT_TYPE_STR_PAIR)
+            mw_free(f->data.strings[0]);
+        if (f->type == MQTT_TYPE_STR_PAIR)
+            mw_free(f->data.strings[1]);
+        if (f->type == MQTT_TYPE_BIN)
+            mw_free(f->data.bindata);
+        mw_free(f);
+    }
+    ctx->tail = NULL;
+    ctx->properties_length = 0;
+    ctx->bytes_consumed = 0;
+    vbi_parser_reset_ctx(&ctx->vbi_parser_ctx);
+}
+
+struct mqtt_property_type {
+    uint8_t id;
+    enum mqtt_datatype datatype;
+    const char* name;
+};
+
+const struct mqtt_property_type mqtt_property_types[] = {
+    { .id = MQTT_PROP_TOPIC_ALIAS,             .name = MQTT_PROP_TOPIC_ALIAS_NAME,             .datatype = MQTT_TYPE_UINT_16  },
+
+    { .id = MQTT_PROP_PAYLOAD_FMT_INDICATOR,   .name = MQTT_PROP_PAYLOAD_FMT_INDICATOR_NAME,   .datatype = MQTT_TYPE_UINT_8   },
+    { .id = MQTT_PROP_MSG_EXPIRY_INTERVAL,     .name = MQTT_PROP_MSG_EXPIRY_INTERVAL_NAME,     .datatype = MQTT_TYPE_UINT_32  },
+    { .id = MQTT_PROP_CONTENT_TYPE,            .name = MQTT_PROP_CONTENT_TYPE_NAME,            .datatype = MQTT_TYPE_STR      },
+    { .id = MQTT_PROP_RESPONSE_TOPIC,          .name = MQTT_PROP_RESPONSE_TOPIC_NAME,          .datatype = MQTT_TYPE_STR      },
+    { .id = MQTT_PROP_CORRELATION_DATA,        .name = MQTT_PROP_CORRELATION_DATA_NAME,        .datatype = MQTT_TYPE_BIN      },
+    { .id = MQTT_PROP_SUB_IDENTIFIER,          .name = MQTT_PROP_SUB_IDENTIFIER_NAME,          .datatype = MQTT_TYPE_VBI      },
+    { .id = MQTT_PROP_SESSION_EXPIRY_INTERVAL, .name = MQTT_PROP_SESSION_EXPIRY_INTERVAL_NAME, .datatype = MQTT_TYPE_UINT_32  },
+    { .id = MQTT_PROP_ASSIGNED_CLIENT_ID,      .name = MQTT_PROP_ASSIGNED_CLIENT_ID_NAME,      .datatype = MQTT_TYPE_STR      },
+    { .id = MQTT_PROP_SERVER_KEEP_ALIVE,       .name = MQTT_PROP_SERVER_KEEP_ALIVE_NAME,       .datatype = MQTT_TYPE_UINT_16  },
+    { .id = MQTT_PROP_AUTH_METHOD,             .name = MQTT_PROP_AUTH_METHOD_NAME,             .datatype = MQTT_TYPE_STR      },
+    { .id = MQTT_PROP_AUTH_DATA,               .name = MQTT_PROP_AUTH_DATA_NAME,               .datatype = MQTT_TYPE_BIN      },
+    { .id = MQTT_PROP_REQ_PROBLEM_INFO,        .name = MQTT_PROP_REQ_PROBLEM_INFO_NAME,        .datatype = MQTT_TYPE_UINT_8   },
+    { .id = MQTT_PROP_WILL_DELAY_INTERVAL,     .name = MQTT_PROP_WIIL_DELAY_INTERVAL_NAME,     .datatype = MQTT_TYPE_UINT_32  },
+    { .id = MQTT_PROP_REQ_RESP_INFORMATION,    .name = MQTT_PROP_REQ_RESP_INFORMATION_NAME,    .datatype = MQTT_TYPE_UINT_8   },
+    { .id = MQTT_PROP_RESP_INFORMATION,        .name = MQTT_PROP_RESP_INFORMATION_NAME,        .datatype = MQTT_TYPE_STR      },
+    { .id = MQTT_PROP_SERVER_REF,              .name = MQTT_PROP_SERVER_REF_NAME,              .datatype = MQTT_TYPE_STR      },
+    { .id = MQTT_PROP_REASON_STR,              .name = MQTT_PROP_REASON_STR_NAME,              .datatype = MQTT_TYPE_STR      },
+    { .id = MQTT_PROP_RECEIVE_MAX,             .name = MQTT_PROP_RECEIVE_MAX_NAME,             .datatype = MQTT_TYPE_UINT_16  },
+    { .id = MQTT_PROP_TOPIC_ALIAS_MAX,         .name = MQTT_PROP_TOPIC_ALIAS_MAX_NAME,         .datatype = MQTT_TYPE_UINT_16  },
+    // MQTT_PROP_TOPIC_ALIAS is first as it is most often used
+    { .id = MQTT_PROP_MAX_QOS,                 .name = MQTT_PROP_MAX_QOS_NAME,                 .datatype = MQTT_TYPE_UINT_8   },
+    { .id = MQTT_PROP_RETAIN_AVAIL,            .name = MQTT_PROP_RETAIN_AVAIL_NAME,            .datatype = MQTT_TYPE_UINT_8   },
+    { .id = MQTT_PROP_USR,                     .name = MQTT_PROP_USR_NAME,                     .datatype = MQTT_TYPE_STR_PAIR },
+    { .id = MQTT_PROP_MAX_PKT_SIZE,            .name = MQTT_PROP_MAX_PKT_SIZE_NAME,            .datatype = MQTT_TYPE_UINT_32  },
+    { .id = MQTT_PROP_WILDCARD_SUB_AVAIL,      .name = MQTT_PROP_WILDCARD_SUB_AVAIL_NAME,      .datatype = MQTT_TYPE_UINT_8   },
+    { .id = MQTT_PROP_SUB_ID_AVAIL,            .name = MQTT_PROP_SUB_ID_AVAIL_NAME,            .datatype = MQTT_TYPE_UINT_8   },
+    { .id = MQTT_PROP_SHARED_SUB_AVAIL,        .name = MQTT_PROP_SHARED_SUB_AVAIL_NAME,        .datatype = MQTT_TYPE_UINT_8   },
+    { .id = 0,                                 .name = NULL,                                   .datatype = MQTT_TYPE_UNKNOWN  }
+};
+
+static int get_property_type_by_id(uint8_t property_id) {
+    for (int i = 0; mqtt_property_types[i].datatype != MQTT_TYPE_UNKNOWN; i++) {
+        if (mqtt_property_types[i].id == property_id)
+            return mqtt_property_types[i].datatype;
+    }
+    return MQTT_TYPE_UNKNOWN;
+}
+
+struct mqtt_property *get_property_by_id(struct mqtt_property *props, uint8_t property_id)
+{
+    while (props) {
+        if (props->id == property_id) {
+            return props;
+        }
+        props = props->next;
+    }
+    return NULL;
+}
+
+// Parses [MQTT-2.2.2]
+static int parse_properties_array(struct mqtt_properties_parser_ctx *ctx, rbuf_t data, mqtt_wss_log_ctx_t log)
+{
+    int rc;
+    switch (ctx->state) {
+        case PROPERTIES_LENGTH:
+            rc = vbi_parser_parse(&ctx->vbi_parser_ctx, data, log);
+            if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
+                ctx->properties_length = ctx->vbi_parser_ctx.result;
+                ctx->bytes_consumed += ctx->vbi_parser_ctx.bytes;
+                ctx->vbi_length = ctx->vbi_parser_ctx.bytes;
+                if (!ctx->properties_length)
+                    return MQTT_NG_CLIENT_PARSE_DONE;
+                ctx->state = PROPERTY_CREATE;
+                break;
+            }
+            return rc;
+        case PROPERTY_CREATE:
+            BUF_READ_CHECK_AT_LEAST(data, 1);
+            struct mqtt_property *prop = mw_calloc(1, sizeof(struct mqtt_property));
+            if (ctx->head == NULL) {
+                ctx->head = prop;
+                ctx->tail = prop;
+            } else {
+                ctx->tail->next = prop;
+                ctx->tail = ctx->tail->next;
+            }
+            ctx->state = PROPERTY_ID;
+            /* FALLTHROUGH */
+        case PROPERTY_ID:
+            rbuf_pop(data, (char*)&ctx->tail->id, 1);
+            ctx->bytes_consumed += 1;
+            ctx->tail->type = get_property_type_by_id(ctx->tail->id);
+            switch (ctx->tail->type) {
+                case MQTT_TYPE_UINT_16:
+                    ctx->state = PROPERTY_TYPE_UINT16;
+                    break;
+                case MQTT_TYPE_UINT_32:
+                    ctx->state = PROPERTY_TYPE_UINT32;
+                    break;
+                case MQTT_TYPE_UINT_8:
+                    ctx->state = PROPERTY_TYPE_UINT8;
+                    break;
+                case MQTT_TYPE_VBI:
+                    ctx->state = PROPERTY_TYPE_VBI;
+                    vbi_parser_reset_ctx(&ctx->vbi_parser_ctx);
+                    break;
+                case MQTT_TYPE_STR:
+                case MQTT_TYPE_STR_PAIR:
+                    ctx->str_idx = 0;
+                    /* FALLTHROUGH */
+                case MQTT_TYPE_BIN:
+                    ctx->state = PROPERTY_TYPE_STR_BIN_LEN;
+                    break;
+                default:
+                    mws_error(log, "Unsupported property type %d for property id %d.", (int)ctx->tail->type, (int)ctx->tail->id);
+                    return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+            }
+            break;
+        case PROPERTY_TYPE_STR_BIN_LEN:
+            BUF_READ_CHECK_AT_LEAST(data, sizeof(uint16_t));
+            rbuf_pop(data, (char*)&ctx->tail->bindata_len, sizeof(uint16_t));
+            ctx->tail->bindata_len = be16toh(ctx->tail->bindata_len);
+            ctx->bytes_consumed += 2;
+            switch (ctx->tail->type) {
+                case MQTT_TYPE_BIN:
+                    ctx->state = PROPERTY_TYPE_BIN;
+                    break;
+                case MQTT_TYPE_STR:
+                case MQTT_TYPE_STR_PAIR:
+                    ctx->state = PROPERTY_TYPE_STR;
+                    break;
+                default:
+                    mws_error(log, "Unexpected datatype in PROPERTY_TYPE_STR_BIN_LEN %d", (int)ctx->tail->type);
+                    return MQTT_NG_CLIENT_INTERNAL_ERROR;
+            }
+            break;
+        case PROPERTY_TYPE_STR:
+            BUF_READ_CHECK_AT_LEAST(data, ctx->tail->bindata_len);
+            ctx->tail->data.strings[ctx->str_idx] = mw_malloc(ctx->tail->bindata_len + 1);
+            rbuf_pop(data, ctx->tail->data.strings[ctx->str_idx], ctx->tail->bindata_len);
+            ctx->tail->data.strings[ctx->str_idx][ctx->tail->bindata_len] = 0;
+            ctx->str_idx++;
+            ctx->bytes_consumed += ctx->tail->bindata_len;
+            if (ctx->tail->type == MQTT_TYPE_STR_PAIR && ctx->str_idx < 2) {
+                ctx->state = PROPERTY_TYPE_STR_BIN_LEN;
+                break;
+            }
+            ctx->state = PROPERTY_NEXT;
+            break;
+        case PROPERTY_TYPE_BIN:
+            BUF_READ_CHECK_AT_LEAST(data, ctx->tail->bindata_len);
+            ctx->tail->data.bindata = mw_malloc(ctx->tail->bindata_len);
+            rbuf_pop(data, ctx->tail->data.bindata, ctx->tail->bindata_len);
+            ctx->bytes_consumed += ctx->tail->bindata_len;
+            ctx->state = PROPERTY_NEXT;
+            break;
+        case PROPERTY_TYPE_VBI:
+            rc = vbi_parser_parse(&ctx->vbi_parser_ctx, data, log);
+            if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
+                ctx->tail->data.uint32 = ctx->vbi_parser_ctx.result;
+                ctx->bytes_consumed += ctx->vbi_parser_ctx.bytes;
+                ctx->state = PROPERTY_NEXT;
+                break;
+            }
+            return rc;
+        case PROPERTY_TYPE_UINT8:
+            BUF_READ_CHECK_AT_LEAST(data, sizeof(uint8_t));
+            rbuf_pop(data, (char*)&ctx->tail->data.uint8, sizeof(uint8_t));
+            ctx->bytes_consumed += sizeof(uint8_t);
+            ctx->state = PROPERTY_NEXT;
+            break;
+        case PROPERTY_TYPE_UINT32:
+            BUF_READ_CHECK_AT_LEAST(data, sizeof(uint32_t));
+            rbuf_pop(data, (char*)&ctx->tail->data.uint32, sizeof(uint32_t));
+            ctx->tail->data.uint32 = be32toh(ctx->tail->data.uint32);
+            ctx->bytes_consumed += sizeof(uint32_t);
+            ctx->state = PROPERTY_NEXT;
+            break;
+        case PROPERTY_TYPE_UINT16:
+            BUF_READ_CHECK_AT_LEAST(data, sizeof(uint16_t));
+            rbuf_pop(data, (char*)&ctx->tail->data.uint16, sizeof(uint16_t));
+            ctx->tail->data.uint16 = be16toh(ctx->tail->data.uint16);
+            ctx->bytes_consumed += sizeof(uint16_t);
+            ctx->state = PROPERTY_NEXT;
+            /* FALLTHROUGH */
+        case PROPERTY_NEXT:
+            if (ctx->properties_length > ctx->bytes_consumed - ctx->vbi_length) {
+                ctx->state = PROPERTY_CREATE;
+                break;
+            } else
+            return MQTT_NG_CLIENT_PARSE_DONE;
+    }
+    return MQTT_NG_CLIENT_OK_CALL_AGAIN;
+}
+
+static int parse_connack_varhdr(struct mqtt_ng_client *client)
+{
+    struct mqtt_ng_parser *parser = &client->parser;
+    switch (parser->varhdr_state) {
+        case MQTT_PARSE_VARHDR_INITIAL:
+            BUF_READ_CHECK_AT_LEAST(parser->received_data, 2);
+            rbuf_pop(parser->received_data, (char*)&parser->mqtt_packet.connack.flags, 1);
+            rbuf_pop(parser->received_data, (char*)&parser->mqtt_packet.connack.reason_code, 1);
+            parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
+            mqtt_properties_parser_ctx_reset(&parser->properties_parser);
+            break;
+        case MQTT_PARSE_VARHDR_PROPS:
+            return parse_properties_array(&parser->properties_parser, parser->received_data, client->log);
+        default:
+            ERROR("invalid state for connack varhdr parser");
+            return MQTT_NG_CLIENT_INTERNAL_ERROR;
+    }
+    return MQTT_NG_CLIENT_OK_CALL_AGAIN;
+}
+
+static int parse_disconnect_varhdr(struct mqtt_ng_client *client)
+{
+    struct mqtt_ng_parser *parser = &client->parser;
+    switch (parser->varhdr_state) {
+        case MQTT_PARSE_VARHDR_INITIAL:
+            if (!parser->mqtt_fixed_hdr_remaining_length) {
+                // [MQTT-3.14.2.1] if reason code omitted act same as == 0
+                parser->mqtt_packet.disconnect.reason_code = 0;
+                return MQTT_NG_CLIENT_PARSE_DONE;
+            }
+            BUF_READ_CHECK_AT_LEAST(parser->received_data, 1);
+            rbuf_pop(parser->received_data, (char*)&parser->mqtt_packet.connack.reason_code, 1);
+            if (parser->mqtt_fixed_hdr_remaining_length == 1)
+                return MQTT_NG_CLIENT_PARSE_DONE;
+            parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
+            mqtt_properties_parser_ctx_reset(&parser->properties_parser);
+            break;
+        case MQTT_PARSE_VARHDR_PROPS:
+            return parse_properties_array(&parser->properties_parser, parser->received_data, client->log);
+        default:
+            ERROR("invalid state for connack varhdr parser");
+            return MQTT_NG_CLIENT_INTERNAL_ERROR;
+    }
+    return MQTT_NG_CLIENT_OK_CALL_AGAIN;
+}
+
+static int parse_puback_varhdr(struct mqtt_ng_client *client)
+{
+    struct mqtt_ng_parser *parser = &client->parser;
+    switch (parser->varhdr_state) {
+        case MQTT_PARSE_VARHDR_INITIAL:
+            BUF_READ_CHECK_AT_LEAST(parser->received_data, 2);
+            rbuf_pop(parser->received_data, (char*)&parser->mqtt_packet.puback.packet_id, 2);
+            parser->mqtt_packet.puback.packet_id = be16toh(parser->mqtt_packet.puback.packet_id);
+            if (parser->mqtt_fixed_hdr_remaining_length < 3) {
+                // [MQTT-3.4.2.1] if length is not big enough for reason code
+                // it is omitted and handled same as if it was present and == 0
+                // initially missed this detail and was wondering WTF is going on (sigh)
+                parser->mqtt_packet.puback.reason_code = 0;
+                return MQTT_NG_CLIENT_PARSE_DONE;
+            }
+            parser->varhdr_state = MQTT_PARSE_VARHDR_OPTIONAL_REASON_CODE;
+            /* FALLTHROUGH */
+        case MQTT_PARSE_VARHDR_OPTIONAL_REASON_CODE:
+            BUF_READ_CHECK_AT_LEAST(parser->received_data, 1);
+            rbuf_pop(parser->received_data, (char*)&parser->mqtt_packet.puback.reason_code, 1);
+            // LOL so in CONNACK you have to have 0 byte to
+            // signify empty properties list
+            // but in PUBACK it can be omitted if remaining length doesn't allow it (sigh)
+            if (parser->mqtt_fixed_hdr_remaining_length < 4)
+                return MQTT_NG_CLIENT_PARSE_DONE;
+
+            parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
+            mqtt_properties_parser_ctx_reset(&parser->properties_parser);
+            /* FALLTHROUGH */
+        case MQTT_PARSE_VARHDR_PROPS:
+            return parse_properties_array(&parser->properties_parser, parser->received_data, client->log);
+        default:
+            ERROR("invalid state for puback varhdr parser");
+            return MQTT_NG_CLIENT_INTERNAL_ERROR;
+    }
+    return MQTT_NG_CLIENT_OK_CALL_AGAIN;
+}
+
+static int parse_suback_varhdr(struct mqtt_ng_client *client)
+{
+    int rc;
+    size_t avail;
+    struct mqtt_ng_parser *parser = &client->parser;
+    struct mqtt_suback *suback = &client->parser.mqtt_packet.suback;
+    switch (parser->varhdr_state) {
+        case MQTT_PARSE_VARHDR_INITIAL:
+            suback->reason_codes = NULL;
+            BUF_READ_CHECK_AT_LEAST(parser->received_data, 2);
+            rbuf_pop(parser->received_data, (char*)&suback->packet_id, 2);
+            suback->packet_id = be16toh(suback->packet_id);
+            parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
+            parser->mqtt_parsed_len = 2;
+            mqtt_properties_parser_ctx_reset(&parser->properties_parser);
+            /* FALLTHROUGH */
+        case MQTT_PARSE_VARHDR_PROPS:
+           rc = parse_properties_array(&parser->properties_parser, parser->received_data, client->log);
+            if (rc != MQTT_NG_CLIENT_PARSE_DONE) 
+                return rc;
+            parser->mqtt_parsed_len += parser->properties_parser.bytes_consumed;
+            suback->reason_code_count = parser->mqtt_fixed_hdr_remaining_length - parser->mqtt_parsed_len;
+            suback->reason_codes = mw_calloc(suback->reason_code_count, sizeof(*suback->reason_codes));
+            suback->reason_codes_pending = suback->reason_code_count;
+            parser->varhdr_state = MQTT_PARSE_REASONCODES;
+            /* FALLTHROUGH */
+        case MQTT_PARSE_REASONCODES:
+            avail = rbuf_bytes_available(parser->received_data);
+            if (avail < 1)
+                return MQTT_NG_CLIENT_NEED_MORE_BYTES;
+
+            suback->reason_codes_pending -= rbuf_pop(parser->received_data, (char*)suback->reason_codes, MIN(suback->reason_codes_pending, avail));
+
+            if (!suback->reason_codes_pending)
+                return MQTT_NG_CLIENT_PARSE_DONE;
+
+            return MQTT_NG_CLIENT_NEED_MORE_BYTES;
+        default:
+            ERROR("invalid state for suback varhdr parser");
+            return MQTT_NG_CLIENT_INTERNAL_ERROR;
+    }
+    return MQTT_NG_CLIENT_OK_CALL_AGAIN;
+}
+
+static int parse_publish_varhdr(struct mqtt_ng_client *client)
+{
+    int rc;
+    struct mqtt_ng_parser *parser = &client->parser;
+    struct mqtt_publish *publish = &client->parser.mqtt_packet.publish;
+    switch (parser->varhdr_state) {
+        case MQTT_PARSE_VARHDR_INITIAL:
+            BUF_READ_CHECK_AT_LEAST(parser->received_data, 2);
+            publish->topic = NULL;
+            publish->qos = ((parser->mqtt_control_packet_type >> 1) & 0x03);
+            rbuf_pop(parser->received_data, (char*)&publish->topic_len, 2);
+            publish->topic_len = be16toh(publish->topic_len);
+            parser->mqtt_parsed_len = 2;
+            if (!publish->topic_len) {
+                parser->varhdr_state = MQTT_PARSE_VARHDR_POST_TOPICNAME;
+                break;
+            }
+            publish->topic = mw_calloc(1, publish->topic_len + 1 /* add 0x00 */);
+            if (publish->topic == NULL)
+                return MQTT_NG_CLIENT_OOM;
+            parser->varhdr_state = MQTT_PARSE_VARHDR_TOPICNAME;
+            /* FALLTHROUGH */
+        case MQTT_PARSE_VARHDR_TOPICNAME:
+            // TODO check empty topic can be valid? In which case we have to skip this step
+            BUF_READ_CHECK_AT_LEAST(parser->received_data, publish->topic_len);
+            rbuf_pop(parser->received_data, publish->topic, publish->topic_len);
+            parser->mqtt_parsed_len += publish->topic_len;
+            parser->varhdr_state = MQTT_PARSE_VARHDR_POST_TOPICNAME;
+            /* FALLTHROUGH */
+        case MQTT_PARSE_VARHDR_POST_TOPICNAME:
+            mqtt_properties_parser_ctx_reset(&parser->properties_parser);
+            if (!publish->qos) { // PacketID present only for QOS > 0 [MQTT-3.3.2.2]
+                parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
+                break;
+            }
+            parser->varhdr_state = MQTT_PARSE_VARHDR_PACKET_ID;
+            /* FALLTHROUGH */
+        case MQTT_PARSE_VARHDR_PACKET_ID:
+            BUF_READ_CHECK_AT_LEAST(parser->received_data, 2);
+            rbuf_pop(parser->received_data, (char*)&publish->packet_id, 2);
+            publish->packet_id = be16toh(publish->packet_id);
+            parser->varhdr_state = MQTT_PARSE_VARHDR_PROPS;
+            parser->mqtt_parsed_len += 2;
+            /* FALLTHROUGH */
+        case MQTT_PARSE_VARHDR_PROPS:
+            rc = parse_properties_array(&parser->properties_parser, parser->received_data, client->log);
+            if (rc != MQTT_NG_CLIENT_PARSE_DONE) 
+                return rc;
+            parser->mqtt_parsed_len += parser->properties_parser.bytes_consumed;
+            parser->varhdr_state = MQTT_PARSE_PAYLOAD;
+            /* FALLTHROUGH */
+        case MQTT_PARSE_PAYLOAD:
+            if (parser->mqtt_fixed_hdr_remaining_length < parser->mqtt_parsed_len) {
+                mw_free(publish->topic);
+                publish->topic = NULL;
+                ERROR("Error parsing PUBLISH message");
+                return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+            }
+            publish->data_len = parser->mqtt_fixed_hdr_remaining_length - parser->mqtt_parsed_len;
+            if (!publish->data_len) {
+                publish->data = NULL;
+                return MQTT_NG_CLIENT_PARSE_DONE; // 0 length payload is OK [MQTT-3.3.3]
+            }
+            BUF_READ_CHECK_AT_LEAST(parser->received_data, publish->data_len);
+
+            publish->data = mw_malloc(publish->data_len);
+            if (publish->data == NULL) {
+                mw_free(publish->topic);
+                publish->topic = NULL;
+                return MQTT_NG_CLIENT_OOM;
+            }
+
+            rbuf_pop(parser->received_data, publish->data, publish->data_len);
+            parser->mqtt_parsed_len += publish->data_len;
+
+            return MQTT_NG_CLIENT_PARSE_DONE;
+        default:
+            ERROR("invalid state for publish varhdr parser");
+            return MQTT_NG_CLIENT_INTERNAL_ERROR;
+    }
+    return MQTT_NG_CLIENT_OK_CALL_AGAIN;
+}
+
+// TODO move to separate file, dont send whole client pointer just to be able
+// to access LOG context send parser only which should include log
+static int parse_data(struct mqtt_ng_client *client)
+{
+    int rc;
+    struct mqtt_ng_parser *parser = &client->parser;
+    switch(parser->state) {
+        case MQTT_PARSE_FIXED_HEADER_PACKET_TYPE:
+            BUF_READ_CHECK_AT_LEAST(parser->received_data, 1);
+            rbuf_pop(parser->received_data, (char*)&parser->mqtt_control_packet_type, 1);
+            vbi_parser_reset_ctx(&parser->vbi_parser);
+            parser->state = MQTT_PARSE_FIXED_HEADER_LEN;
+            break;
+        case MQTT_PARSE_FIXED_HEADER_LEN:
+            rc = vbi_parser_parse(&parser->vbi_parser, parser->received_data, client->log);
+            if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
+                parser->mqtt_fixed_hdr_remaining_length = parser->vbi_parser.result;
+                parser->state = MQTT_PARSE_VARIABLE_HEADER;
+                parser->varhdr_state = MQTT_PARSE_VARHDR_INITIAL;
+                break;
+            }
+            return rc;
+        case MQTT_PARSE_VARIABLE_HEADER:
+            switch (get_control_packet_type(parser->mqtt_control_packet_type)) {
+                case MQTT_CPT_CONNACK:
+                    rc = parse_connack_varhdr(client);
+                    if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
+                        parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
+                        break;
+                    }
+                    return rc;
+                case MQTT_CPT_PUBACK:
+                    rc = parse_puback_varhdr(client);
+                    if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
+                        parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
+                        break;
+                    }
+                    return rc;
+                case MQTT_CPT_SUBACK:
+                    rc = parse_suback_varhdr(client);
+                    if (rc != MQTT_NG_CLIENT_NEED_MORE_BYTES && rc != MQTT_NG_CLIENT_OK_CALL_AGAIN) {
+                        mw_free(parser->mqtt_packet.suback.reason_codes);
+                    }
+                    if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
+                        parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
+                        break;
+                    }
+                    return rc;
+                case MQTT_CPT_PUBLISH:
+                    rc = parse_publish_varhdr(client);
+                    if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
+                        parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
+                        break;
+                    }
+                    return rc;
+                case MQTT_CPT_PINGRESP:
+                    if (parser->mqtt_fixed_hdr_remaining_length) {
+                        ERROR ("PINGRESP has to be 0 Remaining Length."); // [MQTT-3.13.1]
+                        return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+                    }
+                    parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
+                    break;
+                case MQTT_CPT_DISCONNECT:
+                    rc = parse_disconnect_varhdr(client);
+                    if (rc == MQTT_NG_CLIENT_PARSE_DONE) {
+                        parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
+                        break;
+                    }
+                    return rc;
+                default:
+                    ERROR("Parsing Control Packet Type %" PRIu8 " not implemented yet.", get_control_packet_type(parser->mqtt_control_packet_type));
+                    rbuf_bump_tail(parser->received_data, parser->mqtt_fixed_hdr_remaining_length);
+                    parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
+                    return MQTT_NG_CLIENT_NOT_IMPL_YET;
+            }
+            // we could also return MQTT_NG_CLIENT_OK_CALL_AGAIN
+            // and be called again later
+            /* FALLTHROUGH */
+        case MQTT_PARSE_MQTT_PACKET_DONE:
+            parser->state = MQTT_PARSE_FIXED_HEADER_PACKET_TYPE;
+            return MQTT_NG_CLIENT_MQTT_PACKET_DONE;
+    }
+    return MQTT_NG_CLIENT_OK_CALL_AGAIN;
+}
+
+// set next MQTT fragment to send
+// return 1 if nothing to send
+// return -1 on error
+// return 0 if there is fragment set
+static int mqtt_ng_next_to_send(struct mqtt_ng_client *client) {
+    if (client->client_state == CONNECT_PENDING) {
+        client->main_buffer.sending_frag = client->connect_msg;
+        client->client_state = CONNECTING;
+        return 0;
+    }
+    if (client->client_state != CONNECTED)
+        return -1;
+
+    struct buffer_fragment *frag = BUFFER_FIRST_FRAG(&client->main_buffer.hdr_buffer);
+    while (frag) {
+        if ( frag->sent != frag->len )
+            break;
+        frag = frag->next;
+    }
+
+    if ( client->ping_pending && (!frag || (frag->flags & BUFFER_FRAG_MQTT_PACKET_HEAD && frag->sent == 0)) ) {
+        client->ping_pending = 0;
+        ping_frag.sent = 0;
+        client->main_buffer.sending_frag = &ping_frag;
+        return 0;
+    }
+
+    client->main_buffer.sending_frag = frag;
+    return frag == NULL ? 1 : 0;
+}
+
+// send current fragment
+// return 0 if whole remaining length could be sent as a whole
+// return -1 if send buffer was filled and
+// nothing could be written anymore
+// return 1 if last fragment of a message was fully sent
+static int send_fragment(struct mqtt_ng_client *client) {
+    struct buffer_fragment *frag = client->main_buffer.sending_frag;
+
+    // for readability
+    char *ptr = frag->data + frag->sent;
+    size_t bytes = frag->len - frag->sent;
+
+    size_t processed = 0;
+
+    if (bytes)
+        processed = client->send_fnc_ptr(client->user_ctx, ptr, bytes);
+    else
+        WARN("This fragment was fully sent already. This should not happen!");
+
+    frag->sent += processed;
+    if (frag->sent != frag->len)
+        return -1;
+
+    if (frag->flags & BUFFER_FRAG_MQTT_PACKET_TAIL) {
+        client->time_of_last_send = time(NULL);
+        pthread_mutex_lock(&client->stats_mutex);
+        if (client->main_buffer.sending_frag != &ping_frag)
+            client->stats.tx_messages_queued--;
+        client->stats.tx_messages_sent++;
+        pthread_mutex_unlock(&client->stats_mutex);
+        client->main_buffer.sending_frag = NULL;
+        return 1;
+    }
+
+    client->main_buffer.sending_frag = frag->next;
+    
+    return 0;
+}
+
+// attempt sending all fragments of current single MQTT packet
+static int send_all_message_fragments(struct mqtt_ng_client *client) {
+    int rc;
+    while ( !(rc = send_fragment(client)) );
+    return rc;
+}
+
+static void try_send_all(struct mqtt_ng_client *client) {
+    do {
+        if (client->main_buffer.sending_frag == NULL && mqtt_ng_next_to_send(client))
+            return;
+    } while(send_all_message_fragments(client) >= 0);
+}
+
+static inline void mark_message_for_gc(struct buffer_fragment *frag)
+{
+    while (frag) {
+        frag->flags |= BUFFER_FRAG_GARBAGE_COLLECT;
+        buffer_frag_free_data(frag);
+        if (frag->flags & BUFFER_FRAG_MQTT_PACKET_TAIL)
+            return;
+        frag = frag->next;
+    }
+}
+
+static int mark_packet_acked(struct mqtt_ng_client *client, uint16_t packet_id)
+{
+    LOCK_HDR_BUFFER(&client->main_buffer);
+    struct buffer_fragment *frag = BUFFER_FIRST_FRAG(&client->main_buffer.hdr_buffer);
+    while (frag) {
+        if ( (frag->flags & BUFFER_FRAG_MQTT_PACKET_HEAD) && frag->packet_id == packet_id) {
+            if (!frag->sent) {
+                ERROR("Received packet_id (%" PRIu16 ") belongs to MQTT packet which was not yet sent!", packet_id);
+                UNLOCK_HDR_BUFFER(&client->main_buffer);
+                return 1;
+            }
+            mark_message_for_gc(frag);
+            UNLOCK_HDR_BUFFER(&client->main_buffer);
+            return 0;
+        }
+        frag = frag->next;
+    }
+    ERROR("Received packet_id (%" PRIu16 ") is unknown!", packet_id);
+    UNLOCK_HDR_BUFFER(&client->main_buffer);
+    return 1;
+}
+
+int handle_incoming_traffic(struct mqtt_ng_client *client)
+{
+    int rc;
+    struct mqtt_publish *pub;
+    while( (rc = parse_data(client)) == MQTT_NG_CLIENT_OK_CALL_AGAIN );
+    if ( rc == MQTT_NG_CLIENT_MQTT_PACKET_DONE ) {
+        struct mqtt_property *prop;
+#ifdef MQTT_DEBUG_VERBOSE
+        DEBUG("MQTT Packet Parsed Successfully!");
+#endif
+        pthread_mutex_lock(&client->stats_mutex);
+        client->stats.rx_messages_rcvd++;
+        pthread_mutex_unlock(&client->stats_mutex);
+
+        switch (get_control_packet_type(client->parser.mqtt_control_packet_type)) {
+            case MQTT_CPT_CONNACK:
+#ifdef MQTT_DEBUG_VERBOSE
+                DEBUG("Received CONNACK");
+#endif
+                LOCK_HDR_BUFFER(&client->main_buffer);
+                mark_message_for_gc(client->connect_msg);
+                UNLOCK_HDR_BUFFER(&client->main_buffer);
+                client->connect_msg = NULL;
+                if (client->client_state != CONNECTING) {
+                    ERROR("Received unexpected CONNACK");
+                    client->client_state = ERROR;
+                    return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+                }
+                if ((prop = get_property_by_id(client->parser.properties_parser.head, MQTT_PROP_MAX_PKT_SIZE)) != NULL) {
+                    INFO("MQTT server limits message size to %" PRIu32, prop->data.uint32);
+                    client->max_msg_size = prop->data.uint32;
+                }
+                if (client->connack_callback)
+                    client->connack_callback(client->user_ctx, client->parser.mqtt_packet.connack.reason_code);
+                if (!client->parser.mqtt_packet.connack.reason_code) {
+                    INFO("MQTT Connection Accepted By Server");
+                    client->client_state = CONNECTED;
+                    break;
+                }
+                client->client_state = ERROR;
+                return MQTT_NG_CLIENT_SERVER_RETURNED_ERROR;
+            case MQTT_CPT_PUBACK:
+#ifdef MQTT_DEBUG_VERBOSE
+                DEBUG("Received PUBACK %" PRIu16, client->parser.mqtt_packet.puback.packet_id);
+#endif
+                if (mark_packet_acked(client, client->parser.mqtt_packet.puback.packet_id))
+                    return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+                if (client->puback_callback)
+                    client->puback_callback(client->parser.mqtt_packet.puback.packet_id);
+                break;
+            case MQTT_CPT_PINGRESP:
+#ifdef MQTT_DEBUG_VERBOSE
+                DEBUG("Received PINGRESP");
+#endif
+                break;
+            case MQTT_CPT_SUBACK:
+#ifdef MQTT_DEBUG_VERBOSE
+                DEBUG("Received SUBACK %" PRIu16, client->parser.mqtt_packet.suback.packet_id);
+#endif
+                if (mark_packet_acked(client, client->parser.mqtt_packet.suback.packet_id))
+                    return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+                break;
+            case MQTT_CPT_PUBLISH:
+#ifdef MQTT_DEBUG_VERBOSE
+                DEBUG("Recevied PUBLISH");
+#endif
+                pub = &client->parser.mqtt_packet.publish;
+                if (pub->qos > 1) {
+                    mw_free(pub->topic);
+                    mw_free(pub->data);
+                    return MQTT_NG_CLIENT_NOT_IMPL_YET;
+                }
+                if ( pub->qos == 1 && (rc = mqtt_ng_puback(client, pub->packet_id, 0)) ) {
+                    client->client_state = ERROR;
+                    ERROR("Error generating PUBACK reply for PUBLISH");
+                    return rc;
+                }
+                if ( (prop = get_property_by_id(client->parser.properties_parser.head, MQTT_PROP_TOPIC_ALIAS)) != NULL ) {
+                    // Topic Alias property was sent from server
+                    void *topic_ptr;
+                    if (!c_rhash_get_ptr_by_uint64(client->rx_aliases, prop->data.uint8, &topic_ptr)) {
+                        if (pub->topic != NULL) {
+                            ERROR("We do not yet support topic alias reassignment");
+                            return MQTT_NG_CLIENT_NOT_IMPL_YET;
+                        }
+                        pub->topic = topic_ptr;
+                    } else {
+                        if (pub->topic == NULL) {
+                            ERROR("Topic alias with id %d unknown and topic not set by server!", prop->data.uint8);
+                            return MQTT_NG_CLIENT_PROTOCOL_ERROR;
+                        }
+                        c_rhash_insert_uint64_ptr(client->rx_aliases, prop->data.uint8, pub->topic);
+                    }
+                }
+                if (client->msg_callback)
+                    client->msg_callback(pub->topic, pub->data, pub->data_len, pub->qos);
+                // in case we have property topic alias and we have topic we take over the string
+                // and add pointer to it into topic alias list
+                if (prop == NULL)
+                    mw_free(pub->topic);
+                mw_free(pub->data);
+                return MQTT_NG_CLIENT_WANT_WRITE;
+            case MQTT_CPT_DISCONNECT:
+                INFO ("Got MQTT DISCONNECT control packet from server. Reason code: %d", (int)client->parser.mqtt_packet.disconnect.reason_code);
+                client->client_state = DISCONNECTED;
+                break;
+        }
+    }
+
+    return rc;
+}
+
+int mqtt_ng_sync(struct mqtt_ng_client *client)
+{
+    if (client->client_state == RAW || client->client_state == DISCONNECTED)
+        return 0;
+    
+    if (client->client_state == ERROR)
+        return 1;
+
+    LOCK_HDR_BUFFER(&client->main_buffer);
+    try_send_all(client);
+    UNLOCK_HDR_BUFFER(&client->main_buffer);
+
+    int rc;
+
+    while ((rc = handle_incoming_traffic(client)) != MQTT_NG_CLIENT_NEED_MORE_BYTES) {
+        if (rc < 0)
+            break;
+        if (rc == MQTT_NG_CLIENT_WANT_WRITE) {
+            LOCK_HDR_BUFFER(&client->main_buffer);
+            try_send_all(client);
+            UNLOCK_HDR_BUFFER(&client->main_buffer);
+        }
+    }
+
+    if (rc < 0)
+        return rc;
+
+    return 0;
+}
+
+time_t mqtt_ng_last_send_time(struct mqtt_ng_client *client)
+{
+    return client->time_of_last_send;
+}
+
+void mqtt_ng_set_max_mem(struct mqtt_ng_client *client, size_t bytes)
+{
+    client->max_mem_bytes = bytes;
+}
+
+void mqtt_ng_get_stats(struct mqtt_ng_client *client, struct mqtt_ng_stats *stats)
+{
+    pthread_mutex_lock(&client->stats_mutex);
+    memcpy(stats, &client->stats, sizeof(struct mqtt_ng_stats));
+    pthread_mutex_unlock(&client->stats_mutex);
+
+    stats->tx_bytes_queued = 0;
+    stats->tx_buffer_reclaimable = 0;
+
+    LOCK_HDR_BUFFER(&client->main_buffer);
+    stats->tx_buffer_used = BUFFER_BYTES_USED(&client->main_buffer.hdr_buffer);
+    stats->tx_buffer_free = BUFFER_BYTES_AVAILABLE(&client->main_buffer.hdr_buffer);
+    stats->tx_buffer_size = client->main_buffer.hdr_buffer.size;
+    struct buffer_fragment *frag = BUFFER_FIRST_FRAG(&client->main_buffer.hdr_buffer);
+    while (frag) {
+        stats->tx_bytes_queued += frag->len - frag->sent;
+        if (frag_is_marked_for_gc(frag))
+            stats->tx_buffer_reclaimable += FRAG_SIZE_IN_BUFFER(frag);
+
+        frag = frag->next;
+    }
+    UNLOCK_HDR_BUFFER(&client->main_buffer);
+}
+
+int mqtt_ng_set_topic_alias(struct mqtt_ng_client *client, const char *topic)
+{
+    uint16_t idx;
+    pthread_rwlock_wrlock(&client->tx_topic_aliases.rwlock);
+
+    if (client->tx_topic_aliases.idx_assigned >= client->tx_topic_aliases.idx_max) {
+        pthread_rwlock_unlock(&client->tx_topic_aliases.rwlock);
+        mws_error(client->log, "Tx topic alias indexes were exhausted (current version of the library doesn't support reassigning yet. Feel free to contribute.");
+        return 0; //0 is not a valid topic alias
+    }
+
+    struct topic_alias_data *alias;
+    if (!c_rhash_get_ptr_by_str(client->tx_topic_aliases.stoi_dict, topic, (void**)&alias)) {
+        // this is not a problem for library but might be helpful to warn user
+        // as it might indicate bug in their program (but also might be expected)
+        idx = alias->idx;
+        pthread_rwlock_unlock(&client->tx_topic_aliases.rwlock);
+        mws_debug(client->log, "%s topic \"%s\" already has alias set. Ignoring.", __FUNCTION__, topic);
+        return idx;
+    }
+
+    alias = mw_malloc(sizeof(struct topic_alias_data));
+    idx = ++client->tx_topic_aliases.idx_assigned;
+    alias->idx = idx;
+    __atomic_store_n(&alias->usage_count, 0, __ATOMIC_SEQ_CST);
+
+    c_rhash_insert_str_ptr(client->tx_topic_aliases.stoi_dict, topic, (void*)alias);
+
+    pthread_rwlock_unlock(&client->tx_topic_aliases.rwlock);
+    return idx;
+}
diff --git a/mqtt_websockets/src/mqtt_wss_client.c b/mqtt_websockets/src/mqtt_wss_client.c
new file mode 100644
index 000000000..01e2ffce7
--- /dev/null
+++ b/mqtt_websockets/src/mqtt_wss_client.c
@@ -0,0 +1,1138 @@
+// Copyright (C) 2020 Timotej Šiškovič
+// SPDX-License-Identifier: GPL-3.0-only
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free Software Foundation, version 3.
+//
+// This program 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 General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along with this program.
+// If not, see <https://www.gnu.org/licenses/>.
+#define _GNU_SOURCE
+
+#include "mqtt_wss_client.h"
+#include "mqtt_ng.h"
+#include "ws_client.h"
+#include "common_internal.h"
+
+#include <stdlib.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <poll.h>
+#include <string.h>
+#include <time.h>
+
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <netinet/tcp.h> //TCP_NODELAY
+#include <netdb.h>
+
+#include <openssl/err.h>
+#include <openssl/ssl.h>
+
+#define PIPE_READ_END  0
+#define PIPE_WRITE_END 1
+#define POLLFD_SOCKET  0
+#define POLLFD_PIPE    1
+
+#if (OPENSSL_VERSION_NUMBER < OPENSSL_VERSION_110) && (SSLEAY_VERSION_NUMBER >= OPENSSL_VERSION_097)
+#include <openssl/conf.h>
+#endif
+
+//TODO MQTT_PUBLISH_RETAIN should not be needed anymore
+#define MQTT_PUBLISH_RETAIN 0x01
+#define MQTT_CONNECT_CLEAN_SESSION 0x02
+#define MQTT_CONNECT_WILL_RETAIN 0x20
+
+char *util_openssl_ret_err(int err)
+{
+    switch(err){
+        case SSL_ERROR_WANT_READ:
+            return "SSL_ERROR_WANT_READ";
+        case SSL_ERROR_WANT_WRITE:
+            return "SSL_ERROR_WANT_WRITE";
+        case SSL_ERROR_NONE:
+            return "SSL_ERROR_NONE";
+        case SSL_ERROR_ZERO_RETURN:
+            return "SSL_ERROR_ZERO_RETURN";
+        case SSL_ERROR_WANT_CONNECT:
+            return "SSL_ERROR_WANT_CONNECT";
+        case SSL_ERROR_WANT_ACCEPT:
+            return "SSL_ERROR_WANT_ACCEPT";
+        case SSL_ERROR_WANT_X509_LOOKUP:
+            return "SSL_ERROR_WANT_X509_LOOKUP";
+#ifdef SSL_ERROR_WANT_ASYNC
+        case SSL_ERROR_WANT_ASYNC:
+            return "SSL_ERROR_WANT_ASYNC";
+#endif
+#ifdef SSL_ERROR_WANT_ASYNC_JOB
+        case SSL_ERROR_WANT_ASYNC_JOB:
+            return "SSL_ERROR_WANT_ASYNC_JOB";
+#endif
+#ifdef SSL_ERROR_WANT_CLIENT_HELLO_CB
+        case SSL_ERROR_WANT_CLIENT_HELLO_CB:
+            return "SSL_ERROR_WANT_CLIENT_HELLO_CB";
+#endif
+        case SSL_ERROR_SYSCALL:
+            return "SSL_ERROR_SYSCALL";
+        case SSL_ERROR_SSL:
+            return "SSL_ERROR_SSL";
+    }
+    return "UNKNOWN";
+}
+
+struct mqtt_wss_client_struct {
+    ws_client *ws_client;
+
+    mqtt_wss_log_ctx_t log;
+
+// immediate connection (e.g. proxy server)
+    char *host; 
+    int port;
+
+// target of connection (e.g. where we want to connect to)
+    char *target_host;
+    int target_port;
+
+    enum mqtt_wss_proxy_type proxy_type;
+    char *proxy_uname;
+    char *proxy_passwd;
+
+// nonblock IO related
+    int sockfd;
+    int write_notif_pipe[2];
+    struct pollfd poll_fds[2];
+
+    SSL_CTX *ssl_ctx;
+    SSL *ssl;
+    int ssl_flags;
+
+    struct mqtt_ng_client *mqtt;
+
+    int mqtt_keepalive;
+
+    pthread_mutex_t pub_lock;
+
+// signifies that we didn't write all MQTT wanted
+// us to write during last cycle (e.g. due to buffer
+// size) and thus we should arm POLLOUT
+    unsigned int mqtt_didnt_finish_write:1;
+
+    unsigned int mqtt_connected:1;
+    unsigned int mqtt_disconnecting:1;
+
+// Application layer callback pointers
+    void (*msg_callback)(const char *, const void *, size_t, int);
+    void (*puback_callback)(uint16_t packet_id);
+
+    pthread_mutex_t stat_lock;
+    struct mqtt_wss_stats stats;
+
+#ifdef MQTT_WSS_DEBUG
+    void (*ssl_ctx_keylog_cb)(const SSL *ssl, const char *line);
+#endif
+};
+
+static void mws_connack_callback_ng(void *user_ctx, int code)
+{
+    mqtt_wss_client client = user_ctx;
+    switch(code) {
+        case 0:
+            client->mqtt_connected = 1;
+            return;
+//TODO manual labor: all the CONNACK error codes with some nice error message
+        default:
+            mws_error(client->log, "MQTT CONNACK returned error %d", code);
+            return;
+    }
+}
+
+static ssize_t mqtt_send_cb(void *user_ctx, const void* buf, size_t len)
+{
+    mqtt_wss_client mqtt_wss_client = user_ctx;
+#ifdef DEBUG_ULTRA_VERBOSE
+    mws_debug(mqtt_wss_client->log, "mqtt_pal_sendall(len=%d)", len);
+#endif
+    int ret = ws_client_send(mqtt_wss_client->ws_client, WS_OP_BINARY_FRAME, buf, len);
+    if (ret >= 0 && (size_t)ret != len) {
+#ifdef DEBUG_ULTRA_VERBOSE
+        mws_debug(mqtt_wss_client->log, "Not complete message sent (Msg=%d,Sent=%d). Need to arm POLLOUT!", len, ret);
+#endif
+        mqtt_wss_client->mqtt_didnt_finish_write = 1;
+    }
+    return ret;
+}
+
+mqtt_wss_client mqtt_wss_new(const char *log_prefix,
+                             mqtt_wss_log_callback_t log_callback,
+                             msg_callback_fnc_t msg_callback,
+                             void (*puback_callback)(uint16_t packet_id))
+{
+    mqtt_wss_log_ctx_t log;
+
+    log = mqtt_wss_log_ctx_create(log_prefix, log_callback);
+    if(!log)
+        return NULL;
+
+    SSL_library_init();
+    SSL_load_error_strings();
+
+    mqtt_wss_client client = mw_calloc(1, sizeof(struct mqtt_wss_client_struct));
+    if (!client) {
+        mws_error(log, "OOM alocating mqtt_wss_client");
+        goto fail;
+    }
+
+    pthread_mutex_init(&client->pub_lock, NULL);
+    pthread_mutex_init(&client->stat_lock, NULL);
+
+    client->msg_callback = msg_callback;
+    client->puback_callback = puback_callback;
+
+    client->ws_client = ws_client_new(0, &client->target_host, log);
+    if (!client->ws_client) {
+        mws_error(log, "Error creating ws_client");
+        goto fail_1;
+    }
+
+    client->log = log;
+
+#ifdef __APPLE__
+    if (pipe(client->write_notif_pipe)) {
+#else
+    if (pipe2(client->write_notif_pipe, O_CLOEXEC /*| O_DIRECT*/)) {
+#endif
+        mws_error(log, "Couldn't create pipe");
+        goto fail_2;
+    }
+
+    client->poll_fds[POLLFD_PIPE].fd = client->write_notif_pipe[PIPE_READ_END];
+    client->poll_fds[POLLFD_PIPE].events = POLLIN;
+
+    client->poll_fds[POLLFD_SOCKET].events = POLLIN;
+
+    struct mqtt_ng_init settings = {
+        .log = log,
+        .data_in = client->ws_client->buf_to_mqtt,
+        .data_out_fnc = &mqtt_send_cb,
+        .user_ctx = client,
+        .connack_callback = &mws_connack_callback_ng,
+        .puback_callback = puback_callback,
+        .msg_callback = msg_callback
+    };
+    if ( (client->mqtt = mqtt_ng_init(&settings)) == NULL ) {
+        mws_error(log, "Error initializing internal MQTT client");
+        goto fail_3;
+    }
+
+    return client;
+
+fail_3:
+    close(client->write_notif_pipe[PIPE_WRITE_END]);
+    close(client->write_notif_pipe[PIPE_READ_END]);
+fail_2:
+    ws_client_destroy(client->ws_client);
+fail_1:
+    mw_free(client);
+fail:
+    mqtt_wss_log_ctx_destroy(log);
+    return NULL;
+}
+
+void mqtt_wss_set_max_buf_size(mqtt_wss_client client, size_t size)
+{
+    mqtt_ng_set_max_mem(client->mqtt, size);
+}
+
+void mqtt_wss_destroy(mqtt_wss_client client)
+{
+    mqtt_ng_destroy(client->mqtt);
+
+    close(client->write_notif_pipe[PIPE_WRITE_END]);
+    close(client->write_notif_pipe[PIPE_READ_END]);
+
+    ws_client_destroy(client->ws_client);
+
+    // deleted after client->ws_client
+    // as it "borrows" this pointer and might use it
+    if (client->target_host == client->host)
+        client->target_host = NULL;
+    if (client->target_host)
+        mw_free(client->target_host);
+    if (client->host)
+        mw_free(client->host);
+    mw_free(client->proxy_passwd);
+    mw_free(client->proxy_uname);
+
+    if (client->ssl)
+        SSL_free(client->ssl);
+    
+    if (client->ssl_ctx)
+        SSL_CTX_free(client->ssl_ctx);
+
+    if (client->sockfd > 0)
+        close(client->sockfd);
+
+    pthread_mutex_destroy(&client->pub_lock);
+    pthread_mutex_destroy(&client->stat_lock);
+
+    mqtt_wss_log_ctx_destroy(client->log);
+    mw_free(client);
+}
+
+static int cert_verify_callback(int preverify_ok, X509_STORE_CTX *ctx)
+{
+    SSL *ssl;
+    X509 *err_cert;
+    mqtt_wss_client client;
+    int err = 0, depth;
+    char *err_str;
+
+    ssl = X509_STORE_CTX_get_ex_data(ctx, SSL_get_ex_data_X509_STORE_CTX_idx());
+    client = SSL_get_ex_data(ssl, 0);
+
+    // TODO handle depth as per https://www.openssl.org/docs/man1.0.2/man3/SSL_CTX_set_verify.html
+
+    if (!preverify_ok) {
+        err = X509_STORE_CTX_get_error(ctx);
+        depth = X509_STORE_CTX_get_error_depth(ctx);
+        err_cert = X509_STORE_CTX_get_current_cert(ctx);
+        err_str = X509_NAME_oneline(X509_get_subject_name(err_cert), NULL, 0);
+
+        mws_error(client->log, "verify error:num=%d:%s:depth=%d:%s", err,
+                 X509_verify_cert_error_string(err), depth, err_str);
+
+        mw_free(err_str);
+    }
+
+    if (!preverify_ok && err == X509_V_ERR_DEPTH_ZERO_SELF_SIGNED_CERT &&
+        client->ssl_flags & MQTT_WSS_SSL_ALLOW_SELF_SIGNED)
+    {
+        preverify_ok = 1;
+        mws_error(client->log, "Self Signed Certificate Accepted as the connection was "
+                               "requested with MQTT_WSS_SSL_ALLOW_SELF_SIGNED");
+    }
+
+    return preverify_ok;
+}
+
+#define PROXY_CONNECT "CONNECT"
+#define PROXY_HTTP "HTTP/1.1"
+#define HTTP_ENDLINE "\x0D\x0A"
+#define HTTP_HDR_TERMINATOR "\x0D\x0A\x0D\x0A"
+#define HTTP_CODE_LEN 4
+#define HTTP_REASON_MAX_LEN 512
+static int http_parse_reply(mqtt_wss_client client, rbuf_t buf)
+{
+    char *ptr;
+    char http_code_s[4];
+    int http_code;
+    int idx;
+
+    if (rbuf_memcmp_n(buf, PROXY_HTTP, strlen(PROXY_HTTP))) {
+        mws_error(client->log, "http_proxy expected reply with \"" PROXY_HTTP "\"");
+        return 1;
+    }
+
+    rbuf_bump_tail(buf, strlen(PROXY_HTTP));
+
+    if (!rbuf_pop(buf, http_code_s, 1) || http_code_s[0] != 0x20) {
+        mws_error(client->log, "http_proxy missing space after \"" PROXY_HTTP "\"");
+        return 2;
+    }
+
+    if (!rbuf_pop(buf, http_code_s, HTTP_CODE_LEN)) {
+        mws_error(client->log, "http_proxy missing HTTP code");
+        return 3;
+    }
+
+    for (int i = 0; i < HTTP_CODE_LEN - 1; i++)
+        if (http_code_s[i] > 0x39 || http_code_s[i] < 0x30) {
+            mws_error(client->log, "http_proxy HTTP code non numeric");
+            return 4;
+        }
+
+    http_code_s[HTTP_CODE_LEN - 1] = 0;
+    http_code = atoi(http_code_s);
+
+    // TODO check if we ever have more headers here
+    rbuf_find_bytes(buf, HTTP_ENDLINE, strlen(HTTP_ENDLINE), &idx);
+    if (idx >= HTTP_REASON_MAX_LEN) {
+        mws_error(client->log, "http_proxy returned reason that is too long");
+        return 5;
+    }
+
+    if (http_code != 200) {
+        ptr = mw_malloc(idx + 1);
+        if (!ptr)
+            return 6;
+        rbuf_pop(buf, ptr, idx);
+        ptr[idx] = 0;
+
+        mws_error(client->log, "http_proxy returned error code %d \"%s\"", http_code, ptr);
+        mw_free(ptr);
+        return 7;
+    }/* else
+        rbuf_bump_tail(buf, idx);*/
+
+    rbuf_find_bytes(buf, HTTP_HDR_TERMINATOR, strlen(HTTP_HDR_TERMINATOR), &idx);
+    if (idx)
+        rbuf_bump_tail(buf, idx);
+
+    rbuf_bump_tail(buf, strlen(HTTP_HDR_TERMINATOR));
+
+    if (rbuf_bytes_available(buf)) {
+        mws_error(client->log, "http_proxy unexpected trailing bytes after end of HTTP hdr");
+        return 8;
+    }
+
+    mws_debug(client->log, "http_proxy CONNECT succeeded");
+    return 0;
+}
+
+#if defined(OPENSSL_VERSION_NUMBER) && OPENSSL_VERSION_NUMBER < OPENSSL_VERSION_110
+static EVP_ENCODE_CTX *EVP_ENCODE_CTX_new(void)
+{
+	EVP_ENCODE_CTX *ctx = OPENSSL_malloc(sizeof(*ctx));
+
+	if (ctx != NULL) {
+		memset(ctx, 0, sizeof(*ctx));
+	}
+	return ctx;
+}
+static void EVP_ENCODE_CTX_free(EVP_ENCODE_CTX *ctx)
+{
+	OPENSSL_free(ctx);
+	return;
+}
+#endif
+
+inline static int base64_encode_helper(unsigned char *out, int *outl, const unsigned char *in, int in_len)
+{
+    int len;
+    unsigned char *str = out;
+    EVP_ENCODE_CTX *ctx = EVP_ENCODE_CTX_new();
+    EVP_EncodeInit(ctx);
+    EVP_EncodeUpdate(ctx, str, outl, in, in_len);
+    str += *outl;
+    EVP_EncodeFinal(ctx, str, &len);
+    *outl += len;
+
+    str = out;
+    while(*str) {
+        if (*str != 0x0D && *str != 0x0A)
+            *out++ = *str++;
+        else
+            str++;
+    }
+    *out = 0;
+
+    EVP_ENCODE_CTX_free(ctx);
+    return 0;
+}
+
+static int http_proxy_connect(mqtt_wss_client client)
+{
+    int rc;
+    struct pollfd poll_fd;
+    rbuf_t r_buf = rbuf_create(4096);
+    if (!r_buf)
+        return 1;
+    char *r_buf_ptr;
+    size_t r_buf_linear_insert_capacity;
+
+    poll_fd.fd = client->sockfd;
+    poll_fd.events = POLLIN;
+
+    r_buf_ptr = rbuf_get_linear_insert_range(r_buf, &r_buf_linear_insert_capacity);
+    snprintf(r_buf_ptr, r_buf_linear_insert_capacity,"%s %s:%d %s" HTTP_ENDLINE, PROXY_CONNECT, client->target_host, client->target_port, PROXY_HTTP);
+    write(client->sockfd, r_buf_ptr, strlen(r_buf_ptr));
+
+    if (client->proxy_uname) {
+        size_t creds_plain_len = strlen(client->proxy_uname) + strlen(client->proxy_passwd) + 2;
+        char *creds_plain = mw_malloc(creds_plain_len);
+        if (!creds_plain) {
+            mws_error(client->log, "OOM creds_plain");
+            rc = 6;
+            goto cleanup;
+        }
+        int creds_base64_len = (((4 * creds_plain_len / 3) + 3) & ~3);
+        // OpenSSL encoder puts newline every 64 output bytes
+        // we remove those but during encoding we need that space in the buffer
+        creds_base64_len += (1+(creds_base64_len/64)) * strlen("\n");
+        char *creds_base64 = mw_malloc(creds_base64_len + 1);
+        if (!creds_base64) {
+            mw_free(creds_plain);
+            mws_error(client->log, "OOM creds_base64");
+            rc = 6;
+            goto cleanup;
+        }
+        char *ptr = creds_plain;
+        strcpy(ptr, client->proxy_uname);
+        ptr += strlen(client->proxy_uname);
+        *ptr++ = ':';
+        strcpy(ptr, client->proxy_passwd);
+
+        int b64_len;
+        base64_encode_helper((unsigned char*)creds_base64, &b64_len, (unsigned char*)creds_plain, strlen(creds_plain));
+        mw_free(creds_plain);
+
+        r_buf_ptr = rbuf_get_linear_insert_range(r_buf, &r_buf_linear_insert_capacity);
+        snprintf(r_buf_ptr, r_buf_linear_insert_capacity,"Proxy-Authorization: Basic %s" HTTP_ENDLINE, creds_base64);
+        write(client->sockfd, r_buf_ptr, strlen(r_buf_ptr));
+        mw_free(creds_base64);
+    }
+    write(client->sockfd, HTTP_ENDLINE, strlen(HTTP_ENDLINE));
+
+    // read until you find CRLF, CRLF (HTTP HDR end)
+    // or ring buffer is full
+    // or timeout
+    while ((rc = poll(&poll_fd, 1, 1000)) >= 0) {
+        if (!rc) {
+            mws_error(client->log, "http_proxy timeout waiting reply from proxy server");
+            rc = 2;
+            goto cleanup;
+        }
+        r_buf_ptr = rbuf_get_linear_insert_range(r_buf, &r_buf_linear_insert_capacity);
+        if (!r_buf_ptr) {
+            mws_error(client->log, "http_proxy read ring buffer full");
+            rc = 3;
+            goto cleanup;
+        }
+        if ((rc = read(client->sockfd, r_buf_ptr, r_buf_linear_insert_capacity)) < 0) {
+            if (errno == EWOULDBLOCK || errno == EAGAIN) {
+                continue;
+            }
+            mws_error(client->log, "http_proxy error reading from socket \"%s\"", strerror(errno));
+            rc = 4;
+            goto cleanup;
+        }
+        rbuf_bump_head(r_buf, rc);
+        if (rbuf_find_bytes(r_buf, HTTP_HDR_TERMINATOR, strlen(HTTP_HDR_TERMINATOR), &rc)) {
+            rc = 0;
+            if (http_parse_reply(client, r_buf))
+                rc = 5;
+
+            goto cleanup;
+        }
+    }
+    mws_error(client->log, "proxy negotiation poll error \"%s\"", strerror(errno));
+    rc = 5;
+cleanup:
+    rbuf_free(r_buf);
+    return rc;
+}
+
+int mqtt_wss_connect(mqtt_wss_client client, char *host, int port, struct mqtt_connect_params *mqtt_params, int ssl_flags, struct mqtt_wss_proxy *proxy)
+{
+    struct sockaddr_in addr;
+    memset(&addr, 0, sizeof(addr));
+    addr.sin_family = AF_INET;
+
+    struct hostent *he;
+    struct in_addr **addr_list;
+
+    if (!mqtt_params) {
+        mws_error(client->log, "mqtt_params can't be null!");
+        return -1;
+    }
+
+    // reset state in case this is reconnect
+    client->mqtt_didnt_finish_write = 0;
+    client->mqtt_connected = 0;
+    client->mqtt_disconnecting = 0;
+    ws_client_reset(client->ws_client);
+
+    if (client->target_host == client->host)
+        client->target_host = NULL;
+    if (client->target_host)
+        mw_free(client->target_host);
+    if (client->host)
+        mw_free(client->host);
+
+    if (proxy && proxy->type != MQTT_WSS_DIRECT) {
+        client->host = mw_strdup(proxy->host);
+        client->port = proxy->port;
+        client->target_host = mw_strdup(host);
+        client->target_port = port;
+        client->proxy_type = proxy->type;
+        if (proxy->username)
+            client->proxy_uname = mw_strdup(proxy->username);
+        if (proxy->password)
+            client->proxy_passwd = mw_strdup(proxy->password);
+    } else {
+        client->host = mw_strdup(host);
+        client->port = port;
+        client->target_host = client->host;
+        client->target_port = port;
+    }
+
+    client->ssl_flags = ssl_flags;
+
+    //TODO gethostbyname -> getaddinfo
+    //     hstrerror -> gai_strerror
+    if ((he = gethostbyname(client->host)) == NULL) {
+        mws_error(client->log, "gethostbyname() error \"%s\"", hstrerror(h_errno));
+        return -1;
+    }
+
+    addr_list = (struct in_addr **)he->h_addr_list;
+    if(!addr_list[0]) {
+        mws_error(client->log, "No IP addr resolved");
+        return -1;
+    }
+    mws_debug(client->log, "Resolved IP: %s", inet_ntoa(*addr_list[0]));
+    addr.sin_addr = *addr_list[0];
+    addr.sin_port = htons(client->port);
+
+    if (client->sockfd > 0)
+        close(client->sockfd);
+    client->sockfd = socket(AF_INET, SOCK_STREAM, 0);
+    if (client->sockfd < 0) {
+        mws_error(client->log, "Couldn't create socket()");
+        return -1;
+    }
+
+    int flag = 1;
+    int result = setsockopt(client->sockfd,
+                            IPPROTO_TCP,
+                            TCP_NODELAY,
+                            &flag,
+                            sizeof(int));
+    if (result < 0)
+       mws_error(client->log, "Could not dissable NAGLE");
+
+    if (connect(client->sockfd, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
+        mws_error(client->log, "Could not connect to remote endpoint \"%s\", port %d.\n", client->host, client->port);
+        return -3;
+    }
+
+    client->poll_fds[POLLFD_SOCKET].fd = client->sockfd;
+
+    if (fcntl(client->sockfd, F_SETFL, fcntl(client->sockfd, F_GETFL, 0) | O_NONBLOCK) == -1) {
+        mws_error(client->log, "Error setting O_NONBLOCK to TCP socket. \"%s\"", strerror(errno));
+        return -8;
+    }
+
+    if (client->proxy_type != MQTT_WSS_DIRECT)
+        if (http_proxy_connect(client))
+            return -4;
+
+#if OPENSSL_VERSION_NUMBER < OPENSSL_VERSION_110
+#if (SSLEAY_VERSION_NUMBER >= OPENSSL_VERSION_097)
+    OPENSSL_config(NULL);
+#endif
+    SSL_load_error_strings();
+    SSL_library_init();
+#else
+    if (OPENSSL_init_ssl(OPENSSL_INIT_LOAD_CONFIG, NULL) != 1) {
+        mws_error(client->log, "Failed to initialize SSL");
+        return -1;
+    };
+#endif
+
+    // free SSL structs from possible previous connections
+    if (client->ssl)
+        SSL_free(client->ssl);
+    if (client->ssl_ctx)
+        SSL_CTX_free(client->ssl_ctx);
+
+    client->ssl_ctx = SSL_CTX_new(SSLv23_client_method());
+    if (!(client->ssl_flags & MQTT_WSS_SSL_DONT_CHECK_CERTS)) {
+        SSL_CTX_set_default_verify_paths(client->ssl_ctx);
+        SSL_CTX_set_verify(client->ssl_ctx, SSL_VERIFY_PEER | SSL_VERIFY_CLIENT_ONCE, cert_verify_callback);
+    } else
+        mws_error(client->log, "SSL Certificate checking completely disabled!!!");
+
+#ifdef MQTT_WSS_DEBUG
+    if(client->ssl_ctx_keylog_cb)
+        SSL_CTX_set_keylog_callback(client->ssl_ctx, client->ssl_ctx_keylog_cb);
+#endif
+
+    client->ssl = SSL_new(client->ssl_ctx);
+    if (!(client->ssl_flags & MQTT_WSS_SSL_DONT_CHECK_CERTS)) {
+        if (!SSL_set_ex_data(client->ssl, 0, client)) {
+            mws_error(client->log, "Could not SSL_set_ex_data");
+            return -4;
+        }
+    }
+    SSL_set_fd(client->ssl, client->sockfd);
+    SSL_set_connect_state(client->ssl);
+
+    if (!SSL_set_tlsext_host_name(client->ssl, client->target_host)) {
+        mws_error(client->log, "Error setting TLS SNI host");
+        return -7;
+    }
+
+    result = SSL_connect(client->ssl);
+    if (result != -1 && result != 1) {
+        mws_error(client->log, "SSL could not connect");
+        return -5;
+    }
+    if (result == -1) {
+        int ec = SSL_get_error(client->ssl, result);
+        if (ec != SSL_ERROR_WANT_READ && ec != SSL_ERROR_WANT_WRITE) {
+            mws_error(client->log, "Failed to start SSL connection");
+            return -6;
+        }
+    }
+
+    uint8_t mqtt_flags = (mqtt_params->will_flags & MQTT_WSS_PUB_QOSMASK) << 3;
+    if (mqtt_params->will_flags & MQTT_WSS_PUB_RETAIN)
+        mqtt_flags |= MQTT_CONNECT_WILL_RETAIN;
+    mqtt_flags |= MQTT_CONNECT_CLEAN_SESSION;
+
+    client->mqtt_keepalive = (mqtt_params->keep_alive ? mqtt_params->keep_alive : 400);
+
+    mws_info(client->log, "Going to connect using internal MQTT 5 implementation");
+    struct mqtt_auth_properties auth;
+    auth.client_id = (char*)mqtt_params->clientid;
+    auth.client_id_free = NULL;
+    auth.username = (char*)mqtt_params->username;
+    auth.username_free = NULL;
+    auth.password = (char*)mqtt_params->password;
+    auth.password_free = NULL;
+    struct mqtt_lwt_properties lwt;
+    lwt.will_topic = (char*)mqtt_params->will_topic;
+    lwt.will_topic_free = NULL;
+    lwt.will_message = (void*)mqtt_params->will_msg;
+    lwt.will_message_free = NULL; // TODO expose no copy version to API
+    lwt.will_message_size = mqtt_params->will_msg_len;
+    lwt.will_qos = (mqtt_params->will_flags & MQTT_WSS_PUB_QOSMASK);
+    lwt.will_retain = mqtt_params->will_flags & MQTT_WSS_PUB_RETAIN;
+    int ret = mqtt_ng_connect(client->mqtt, &auth, mqtt_params->will_msg ? &lwt : NULL, 1, client->mqtt_keepalive);
+    if (ret) {
+        mws_error(client->log, "Error generating MQTT connect");
+        return 1;
+    }
+
+    client->poll_fds[POLLFD_PIPE].events = POLLIN;
+    client->poll_fds[POLLFD_SOCKET].events = POLLIN;
+    // wait till MQTT connection is established
+    while (!client->mqtt_connected) {
+        if(mqtt_wss_service(client, -1)) {
+            mws_error(client->log, "Error connecting to MQTT WSS server \"%s\", port %d.", host, port);
+            return 2;
+        }
+    }
+
+    return 0;
+}
+
+#define NSEC_PER_USEC   1000ULL
+#define USEC_PER_SEC    1000000ULL
+#define NSEC_PER_MSEC   1000000ULL
+#define NSEC_PER_SEC    1000000000ULL
+
+static inline uint64_t boottime_usec(mqtt_wss_client client) {
+    struct timespec ts;
+#if defined(__APPLE__) || defined(__FreeBSD__)
+    if (clock_gettime(CLOCK_MONOTONIC, &ts) == -1) {
+#else
+    if (clock_gettime(CLOCK_BOOTTIME, &ts) == -1) {
+#endif
+        mws_error(client->log, "clock_gettimte failed");
+        return 0;
+    }
+    return (uint64_t)ts.tv_sec * USEC_PER_SEC + (ts.tv_nsec % NSEC_PER_SEC) / NSEC_PER_USEC;
+}
+
+#define MWS_TIMED_OUT 1
+#define MWS_ERROR 2
+#define MWS_OK 0
+static inline const char *mqtt_wss_error_tos(int ec)
+{
+    switch(ec) {
+        case MWS_TIMED_OUT:
+            return "Error: Operation was not able to finish in time";
+        case MWS_ERROR:
+            return "Unspecified Error";
+        default:
+            return "Unknown Error Code!";
+    }
+
+}
+
+static inline int mqtt_wss_service_all(mqtt_wss_client client, int timeout_ms)
+{
+    uint64_t exit_by = boottime_usec(client) + (timeout_ms * NSEC_PER_MSEC);
+    uint64_t now;
+    client->poll_fds[POLLFD_SOCKET].events |= POLLOUT; // TODO when entering mwtt_wss_service use out buffer size to arm POLLOUT
+    while (rbuf_bytes_available(client->ws_client->buf_write)) {
+        now = boottime_usec(client);
+        if (now >= exit_by)
+            return MWS_TIMED_OUT;
+        if (mqtt_wss_service(client, exit_by - now))
+            return MWS_ERROR;
+    }
+    return MWS_OK;
+}
+
+void mqtt_wss_disconnect(mqtt_wss_client client, int timeout_ms)
+{
+    int ret;
+
+    // block application from sending more MQTT messages
+    client->mqtt_disconnecting = 1;
+
+    // send whatever was left at the time of calling this function
+    ret = mqtt_wss_service_all(client, timeout_ms / 4);
+    if(ret)
+        mws_error(client->log,
+                  "Error while trying to send all remaining data in an attempt "
+                  "to gracefully disconnect! EC=%d Desc:\"%s\"",
+                  ret,
+                  mqtt_wss_error_tos(ret));
+
+    // schedule and send MQTT disconnect
+    mqtt_ng_disconnect(client->mqtt, 0);
+    mqtt_ng_sync(client->mqtt);
+
+    ret = mqtt_wss_service_all(client, timeout_ms / 4);
+    if(ret)
+        mws_error(client->log,
+                  "Error while trying to send MQTT disconnect message in an attempt "
+                  "to gracefully disconnect! EC=%d Desc:\"%s\"",
+                  ret,
+                  mqtt_wss_error_tos(ret));
+
+    // send WebSockets close message
+    uint16_t ws_rc = htobe16(1000);
+    ws_client_send(client->ws_client, WS_OP_CONNECTION_CLOSE, (const char*)&ws_rc, sizeof(ws_rc));
+    ret = mqtt_wss_service_all(client, timeout_ms / 4);
+    if(ret) {
+        // Some MQTT/WSS servers will close socket on receipt of MQTT disconnect and
+        // do not wait for WebSocket to be closed properly
+        mws_warn(client->log,
+                 "Error while trying to send WebSocket disconnect message in an attempt "
+                 "to gracefully disconnect! EC=%d Desc:\"%s\".",
+                 ret,
+                 mqtt_wss_error_tos(ret));
+    }
+
+    // Service WSS connection until remote closes connection (usual)
+    // or timeout happens (unusual) in which case we close
+    mqtt_wss_service_all(client, timeout_ms / 4);
+
+    close(client->sockfd);
+    client->sockfd = -1;
+}
+
+static inline void mqtt_wss_wakeup(mqtt_wss_client client)
+{
+#ifdef DEBUG_ULTRA_VERBOSE
+    mws_debug(client->log, "mqtt_wss_wakup - forcing wake up of main loop");
+#endif
+    write(client->write_notif_pipe[PIPE_WRITE_END], " ", 1);
+}
+
+#define THROWAWAY_BUF_SIZE 32
+char throwaway[THROWAWAY_BUF_SIZE];
+static inline void util_clear_pipe(int fd)
+{
+    (void)read(fd, throwaway, THROWAWAY_BUF_SIZE);
+}
+
+static inline void set_socket_pollfds(mqtt_wss_client client, int ssl_ret) {
+    if (ssl_ret == SSL_ERROR_WANT_WRITE)
+        client->poll_fds[POLLFD_SOCKET].events |= POLLOUT;
+    if (ssl_ret == SSL_ERROR_WANT_READ)
+        client->poll_fds[POLLFD_SOCKET].events |= POLLIN;
+}
+
+static int handle_mqtt_internal(mqtt_wss_client client)
+{
+    int rc = mqtt_ng_sync(client->mqtt);
+    if (rc) {
+        mws_error(client->log, "mqtt_ng_sync returned %d != 0", rc);
+        client->mqtt_connected = 0;
+        return 1;
+    }
+    return 0;
+}
+
+#define SEC_TO_MSEC 1000
+static inline long long int t_till_next_keepalive_ms(mqtt_wss_client client)
+{
+    time_t last_send = mqtt_ng_last_send_time(client->mqtt);
+    long long int next_mqtt_keep_alive = (last_send * SEC_TO_MSEC)
+        + (client->mqtt_keepalive * (SEC_TO_MSEC * 0.75 /* SEND IN ADVANCE */));
+    return(next_mqtt_keep_alive - (time(NULL) * SEC_TO_MSEC));
+}
+
+#ifdef MQTT_WSS_CPUSTATS
+static inline uint64_t mqtt_wss_now_usec(mqtt_wss_client client) {
+    struct timespec ts;
+    if(clock_gettime(CLOCK_MONOTONIC, &ts) == -1) {
+        mws_error(client->log, "clock_gettime(CLOCK_MONOTONIC, &timespec) failed.");
+        return 0;
+    }
+    return (uint64_t)ts.tv_sec * USEC_PER_SEC + (ts.tv_nsec % NSEC_PER_SEC) / NSEC_PER_USEC;
+}
+#endif
+
+int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)
+{
+    char *ptr;
+    size_t size;
+    int ret;
+    int send_keepalive = 0;
+
+#ifdef MQTT_WSS_CPUSTATS
+    uint64_t t1,t2;
+    t1 = mqtt_wss_now_usec(client);
+#endif
+
+#ifdef DEBUG_ULTRA_VERBOSE
+    mws_debug(client->log, ">>>>> mqtt_wss_service <<<<<");
+    mws_debug(client->log, "Waiting for events: %s%s%s",
+        (client->poll_fds[POLLFD_SOCKET].events & POLLIN)  ? "SOCKET_POLLIN " : "",
+        (client->poll_fds[POLLFD_SOCKET].events & POLLOUT) ? "SOCKET_POLLOUT " : "",
+        (client->poll_fds[POLLFD_PIPE].events & POLLIN) ? "PIPE_POLLIN" : "" );
+#endif
+
+    // Check user requested TO doesn't interfere with MQTT keep alives
+    long long int till_next_keep_alive = t_till_next_keepalive_ms(client);
+    if (client->mqtt_connected && (timeout_ms < 0 || timeout_ms >= till_next_keep_alive)) {
+        #ifdef DEBUG_ULTRA_VERBOSE
+            mws_debug(client->log, "Shortening Timeout requested %d to %lld to ensure keep-alive can be sent", timeout_ms, till_next_keep_alive);
+        #endif
+        timeout_ms = till_next_keep_alive;
+        send_keepalive = 1;
+    }
+
+#ifdef MQTT_WSS_CPUSTATS
+    t2 = mqtt_wss_now_usec(client);
+    client->stats.time_keepalive += t2 - t1;
+#endif
+
+    if ((ret = poll(client->poll_fds, 2, timeout_ms >= 0 ? timeout_ms : -1)) < 0) {
+        if (errno == EINTR) {
+            mws_warn(client->log, "poll interrupted by EINTR");
+            return 0;
+        }
+        mws_error(client->log, "poll error \"%s\"", strerror(errno));
+        return -2;
+    }
+
+#ifdef DEBUG_ULTRA_VERBOSE
+    mws_debug(client->log, "Poll events happened: %s%s%s%s",
+        (client->poll_fds[POLLFD_SOCKET].revents & POLLIN)  ? "SOCKET_POLLIN " : "",
+        (client->poll_fds[POLLFD_SOCKET].revents & POLLOUT) ? "SOCKET_POLLOUT " : "",
+        (client->poll_fds[POLLFD_PIPE].revents & POLLIN) ? "PIPE_POLLIN " : "",
+        (!ret) ? "POLL_TIMEOUT" : "");
+#endif
+
+#ifdef MQTT_WSS_CPUSTATS
+    t1 = mqtt_wss_now_usec(client);
+#endif
+
+    if (ret == 0) {
+        if (send_keepalive) {
+            // otherwise we shortened the timeout ourselves to take care of
+            // MQTT keep alives
+#ifdef DEBUG_ULTRA_VERBOSE
+            mws_debug(client->log, "Forcing MQTT Ping/keep-alive");
+#endif
+            mqtt_ng_ping(client->mqtt);
+        } else {
+            // if poll timed out and user requested timeout was being used
+            // return here let user do his work and he will call us back soon
+            return 0;
+        }
+    }
+
+#ifdef MQTT_WSS_CPUSTATS
+    t2 = mqtt_wss_now_usec(client);
+    client->stats.time_keepalive += t2 - t1;
+#endif
+
+    client->poll_fds[POLLFD_SOCKET].events = 0;
+
+    if ((ptr = rbuf_get_linear_insert_range(client->ws_client->buf_read, &size))) {
+        if((ret = SSL_read(client->ssl, ptr, size)) > 0) {
+#ifdef DEBUG_ULTRA_VERBOSE
+            mws_debug(client->log, "SSL_Read: Read %d.", ret);
+#endif
+            pthread_mutex_lock(&client->stat_lock);
+            client->stats.bytes_rx += ret;
+            pthread_mutex_unlock(&client->stat_lock);
+            rbuf_bump_head(client->ws_client->buf_read, ret);
+        } else {
+            int errnobkp = errno;
+            ret = SSL_get_error(client->ssl, ret);
+#ifdef DEBUG_ULTRA_VERBOSE
+            mws_debug(client->log, "Read Err: %s", util_openssl_ret_err(ret));
+#endif
+            set_socket_pollfds(client, ret);
+            if (ret != SSL_ERROR_WANT_READ &&
+                ret != SSL_ERROR_WANT_WRITE) {
+                mws_error(client->log, "SSL_read error: %d %s", ret, util_openssl_ret_err(ret));
+                if (ret == SSL_ERROR_SYSCALL)
+                    mws_error(client->log, "SSL_read SYSCALL errno: %d %s", errnobkp, strerror(errnobkp));
+                return MQTT_WSS_ERR_CONN_DROP;
+            }
+        }
+    }
+
+#ifdef MQTT_WSS_CPUSTATS
+    t1 = mqtt_wss_now_usec(client);
+    client->stats.time_read_socket += t1 - t2;
+#endif
+
+    ret = ws_client_process(client->ws_client);
+    switch(ret) {
+        case WS_CLIENT_PROTOCOL_ERROR:
+            return MQTT_WSS_ERR_PROTO_WS;
+        case WS_CLIENT_NEED_MORE_BYTES:
+#ifdef DEBUG_ULTRA_VERBOSE
+            mws_debug(client->log, "WSCLIENT WANT READ");
+#endif
+            client->poll_fds[POLLFD_SOCKET].events |= POLLIN;
+            break;
+        case WS_CLIENT_CONNECTION_CLOSED:
+            return MQTT_WSS_ERR_CONN_DROP;
+    }
+
+#ifdef MQTT_WSS_CPUSTATS
+    t2 = mqtt_wss_now_usec(client);
+    client->stats.time_process_websocket += t2 - t1;
+#endif
+
+    // process MQTT stuff
+    if(client->ws_client->state == WS_ESTABLISHED)
+        if (handle_mqtt_internal(client))
+            return MQTT_WSS_ERR_PROTO_MQTT;
+
+    if (client->mqtt_didnt_finish_write) {
+        client->mqtt_didnt_finish_write = 0;
+        client->poll_fds[POLLFD_SOCKET].events |= POLLOUT;
+    }
+
+#ifdef MQTT_WSS_CPUSTATS
+    t1 = mqtt_wss_now_usec(client);
+    client->stats.time_process_mqtt += t1 - t2;
+#endif
+
+    if ((ptr = rbuf_get_linear_read_range(client->ws_client->buf_write, &size))) {
+#ifdef DEBUG_ULTRA_VERBOSE
+        mws_debug(client->log, "Have data to write to SSL");
+#endif
+        if ((ret = SSL_write(client->ssl, ptr, size)) > 0) {
+#ifdef DEBUG_ULTRA_VERBOSE
+            mws_debug(client->log, "SSL_Write: Written %d of avail %d.", ret, size);
+#endif
+            pthread_mutex_lock(&client->stat_lock);
+            client->stats.bytes_tx += ret;
+            pthread_mutex_unlock(&client->stat_lock);
+            rbuf_bump_tail(client->ws_client->buf_write, ret);
+        } else {
+            int errnobkp = errno;
+            ret = SSL_get_error(client->ssl, ret);
+#ifdef DEBUG_ULTRA_VERBOSE
+            mws_debug(client->log, "Write Err: %s", util_openssl_ret_err(ret));
+#endif
+            set_socket_pollfds(client, ret);
+            if (ret != SSL_ERROR_WANT_READ &&
+                ret != SSL_ERROR_WANT_WRITE) {
+                mws_error(client->log, "SSL_write error: %d %s", ret, util_openssl_ret_err(ret));
+                if (ret == SSL_ERROR_SYSCALL)
+                    mws_error(client->log, "SSL_write SYSCALL errno: %d %s", errnobkp, strerror(errnobkp));
+                return MQTT_WSS_ERR_CONN_DROP;
+            }
+        }
+    }
+
+    if(client->poll_fds[POLLFD_PIPE].revents & POLLIN)
+        util_clear_pipe(client->write_notif_pipe[PIPE_READ_END]);
+
+#ifdef MQTT_WSS_CPUSTATS
+    t2 = mqtt_wss_now_usec(client);
+    client->stats.time_write_socket += t2 - t1;
+#endif
+
+    return MQTT_WSS_OK;
+}
+
+int mqtt_wss_publish5(mqtt_wss_client client,
+                      char *topic,
+                      free_fnc_t topic_free,
+                      void *msg,
+                      free_fnc_t msg_free,
+                      size_t msg_len,
+                      uint8_t publish_flags,
+                      uint16_t *packet_id)
+{
+    if (client->mqtt_disconnecting) {
+        mws_error(client->log, "mqtt_wss is disconnecting can't publish");
+        return 1;
+    }
+
+    if (!client->mqtt_connected) {
+        mws_error(client->log, "MQTT is offline. Can't send message.");
+        return 1;
+    }
+    uint8_t mqtt_flags = 0;
+
+    mqtt_flags = (publish_flags & MQTT_WSS_PUB_QOSMASK) << 1;
+    if (publish_flags & MQTT_WSS_PUB_RETAIN)
+        mqtt_flags |= MQTT_PUBLISH_RETAIN;
+
+    int rc = mqtt_ng_publish(client->mqtt, topic, topic_free, msg, msg_free, msg_len, mqtt_flags, packet_id);
+    if (rc == MQTT_NG_MSGGEN_MSG_TOO_BIG)
+        return MQTT_WSS_ERR_TOO_BIG_FOR_SERVER;
+
+    mqtt_wss_wakeup(client);
+
+    return rc;
+}
+
+int mqtt_wss_subscribe(mqtt_wss_client client, char *topic, int max_qos_level)
+{
+    (void)max_qos_level; //TODO now hardcoded
+    if (!client->mqtt_connected) {
+        mws_error(client->log, "MQTT is offline. Can't subscribe.");
+        return 1;
+    }
+
+    if (client->mqtt_disconnecting) {
+        mws_error(client->log, "mqtt_wss is disconnecting can't subscribe");
+        return 1;
+    }
+
+    struct mqtt_sub sub = {
+        .topic = topic,
+        .topic_free = NULL,
+        .options = /* max_qos_level & 0x3 TODO when QOS > 1 implemented */ 0x01 | (0x01 << 3)
+    };
+    mqtt_ng_subscribe(client->mqtt, &sub, 1);
+
+    mqtt_wss_wakeup(client);
+    return 0;
+}
+
+struct mqtt_wss_stats mqtt_wss_get_stats(mqtt_wss_client client)
+{
+    struct mqtt_wss_stats current;
+    pthread_mutex_lock(&client->stat_lock);
+    current = client->stats;
+    memset(&client->stats, 0, sizeof(client->stats));
+    pthread_mutex_unlock(&client->stat_lock);
+    mqtt_ng_get_stats(client->mqtt, &current.mqtt);
+    return current;
+}
+
+int mqtt_wss_set_topic_alias(mqtt_wss_client client, const char *topic)
+{
+    return mqtt_ng_set_topic_alias(client->mqtt, topic);
+}
+
+#ifdef MQTT_WSS_DEBUG
+void mqtt_wss_set_SSL_CTX_keylog_cb(mqtt_wss_client client, void (*ssl_ctx_keylog_cb)(const SSL *ssl, const char *line))
+{
+    client->ssl_ctx_keylog_cb = ssl_ctx_keylog_cb;
+}
+#endif
diff --git a/mqtt_websockets/src/mqtt_wss_log.c b/mqtt_websockets/src/mqtt_wss_log.c
new file mode 100644
index 000000000..2c8cf32e5
--- /dev/null
+++ b/mqtt_websockets/src/mqtt_wss_log.c
@@ -0,0 +1,127 @@
+#include "mqtt_wss_log.h"
+#include <stdlib.h>
+#include <stdarg.h>
+#include <string.h>
+#include <stdio.h>
+#include "common_internal.h"
+
+struct mqtt_wss_log_ctx {
+    mqtt_wss_log_callback_t extern_log_fnc;
+    char *ctx_prefix;
+    char *buffer;
+    char *buffer_w_ptr;
+    size_t buffer_bytes_avail;
+};
+
+#define LOG_BUFFER_SIZE 1024 * 4
+#define LOG_CTX_PREFIX_SEV_STR "  : "
+#define LOG_CTX_PREFIX_LIMIT 15
+#define LOG_CTX_PREFIX_LIMIT_STR (LOG_CTX_PREFIX_LIMIT - (2 + strlen(LOG_CTX_PREFIX_SEV_STR)))  // with [] characters and affixed ' ' it is total 15 chars
+#if (LOG_CTX_PREFIX_LIMIT * 10) > LOG_BUFFER_SIZE
+#error "LOG_BUFFER_SIZE too small"
+#endif
+mqtt_wss_log_ctx_t mqtt_wss_log_ctx_create(const char *ctx_prefix, mqtt_wss_log_callback_t log_callback)
+{
+    mqtt_wss_log_ctx_t ctx = mw_calloc(1, sizeof(struct mqtt_wss_log_ctx));
+    if(!ctx)
+        return NULL;
+
+    if(log_callback) {
+        ctx->extern_log_fnc = log_callback;
+        ctx->buffer = mw_calloc(1, LOG_BUFFER_SIZE);
+        if(!ctx->buffer)
+            goto cleanup;
+
+        ctx->buffer_w_ptr = ctx->buffer;
+        if(ctx_prefix) {
+            *(ctx->buffer_w_ptr++) = '[';
+            strncpy(ctx->buffer_w_ptr, ctx_prefix, LOG_CTX_PREFIX_LIMIT_STR);
+            ctx->buffer_w_ptr += strnlen(ctx_prefix, LOG_CTX_PREFIX_LIMIT_STR);
+            *(ctx->buffer_w_ptr++) = ']';
+        }
+        strcpy(ctx->buffer_w_ptr, LOG_CTX_PREFIX_SEV_STR);
+        ctx->buffer_w_ptr += strlen(LOG_CTX_PREFIX_SEV_STR);
+        // no term '\0' -> calloc is used
+
+        ctx->buffer_bytes_avail = LOG_BUFFER_SIZE - strlen(ctx->buffer);
+
+        return ctx;
+    }
+
+    if(ctx_prefix) {
+        ctx->ctx_prefix = strndup(ctx_prefix, LOG_CTX_PREFIX_LIMIT_STR);
+        if(!ctx->ctx_prefix)
+            goto cleanup;
+    }
+
+    return ctx;
+
+cleanup:
+    mw_free(ctx);
+    return NULL;
+}
+
+void mqtt_wss_log_ctx_destroy(mqtt_wss_log_ctx_t ctx)
+{
+    mw_free(ctx->ctx_prefix);
+    mw_free(ctx->buffer);
+    mw_free(ctx);
+}
+
+static inline char severity_to_c(int severity)
+{
+    switch (severity) {
+        case MQTT_WSS_LOG_FATAL:
+            return 'F';
+        case MQTT_WSS_LOG_ERROR:
+            return 'E';
+        case MQTT_WSS_LOG_WARN:
+            return 'W';
+        case MQTT_WSS_LOG_INFO:
+            return 'I';
+        case MQTT_WSS_LOG_DEBUG:
+            return 'D';
+        default:
+            return '?';
+    }
+}
+
+void mws_log(int severity, mqtt_wss_log_ctx_t ctx, const char *fmt, va_list args)
+{
+    size_t size;
+
+    if(ctx->extern_log_fnc) {
+        size = vsnprintf(ctx->buffer_w_ptr, ctx->buffer_bytes_avail, fmt, args);
+        *(ctx->buffer_w_ptr - 3) = severity_to_c(severity);
+
+        ctx->extern_log_fnc(severity, ctx->buffer);
+
+        if(size >= ctx->buffer_bytes_avail)
+            mws_error(ctx, "Last message of this type was truncated! Consider what you log or increase LOG_BUFFER_SIZE if really needed.");
+
+        return;
+    }
+
+    if(ctx->ctx_prefix)
+        printf("[%s] ", ctx->ctx_prefix);
+
+    printf("%c: ", severity_to_c(severity));
+
+    vprintf(fmt, args);
+    putchar('\n');
+}
+
+#define DEFINE_MWS_SEV_FNC(severity_fncname, severity) \
+void mws_ ## severity_fncname(mqtt_wss_log_ctx_t ctx, const char *fmt, ...) \
+{ \
+    va_list args; \
+    va_start(args, fmt); \
+    mws_log(severity, ctx, fmt, args); \
+    va_end(args); \
+}
+
+DEFINE_MWS_SEV_FNC(fatal, MQTT_WSS_LOG_FATAL)
+DEFINE_MWS_SEV_FNC(error, MQTT_WSS_LOG_ERROR)
+DEFINE_MWS_SEV_FNC(warn,  MQTT_WSS_LOG_WARN )
+DEFINE_MWS_SEV_FNC(info,  MQTT_WSS_LOG_INFO )
+DEFINE_MWS_SEV_FNC(debug, MQTT_WSS_LOG_DEBUG)
diff --git a/mqtt_websockets/src/test.c b/mqtt_websockets/src/test.c
new file mode 100644
index 000000000..aee23545a
--- /dev/null
+++ b/mqtt_websockets/src/test.c
@@ -0,0 +1,100 @@
+// Copyright (C) 2020 Timotej Šiškovič
+// SPDX-License-Identifier: GPL-3.0-only
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free Software Foundation, version 3.
+//
+// This program 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 General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along with this program.
+// If not, see <https://www.gnu.org/licenses/>.
+
+#include <unistd.h>
+#include <stdio.h>
+#include <string.h>
+#include <stdlib.h>
+
+#include "mqtt_wss_client.h"
+
+int test_exit = 0;
+int port = 0;
+
+void mqtt_wss_log_cb(mqtt_wss_log_type_t log_type, const char* str)
+{
+    (void)log_type;
+    puts(str);
+}
+
+#define TEST_MSGLEN_MAX 512
+void msg_callback(const char *topic, const void *msg, size_t msglen, int qos)
+{
+    char cmsg[TEST_MSGLEN_MAX];
+    size_t len = (msglen < TEST_MSGLEN_MAX - 1) ? msglen : (TEST_MSGLEN_MAX - 1);
+    memcpy(cmsg,
+           msg,
+           len);
+    cmsg[len] = 0;
+
+    if (!strcmp(cmsg, "shutdown"))
+        test_exit = 1;
+
+    printf("Got Message From Broker Topic \"%s\" QOS %d MSG: \"%s\"\n", topic, qos, cmsg);
+}
+
+#define TESTMSG "Hello World!"
+int client_handle(mqtt_wss_client client)
+{
+    struct mqtt_connect_params params = {
+        .clientid = "test",
+        .username = "anon",
+        .password = "anon",
+        .keep_alive = 10
+    };
+
+/*    struct mqtt_wss_proxy proxy = {
+        .host = "127.0.0.1",
+        .port = 3128,
+        .type = MQTT_WSS_PROXY_HTTP
+    };*/
+
+    while (mqtt_wss_connect(client, "127.0.0.1", port, &params, MQTT_WSS_SSL_ALLOW_SELF_SIGNED, NULL /*&proxy*/)) {
+        printf("Connect failed\n");
+        sleep(1);
+        printf("Attempting Reconnect\n");
+    }
+    printf("Connection succeeded\n");
+
+    mqtt_wss_subscribe(client, "test", 1);
+
+    while (!test_exit) {
+        if(mqtt_wss_service(client, -1) < 0)
+            return 1;
+    }
+    return 0;
+}
+
+int main(int argc, char **argv)
+{
+    if (argc >= 2)
+        port = atoi(argv[1]);
+    if (!port)
+        port = 9002;
+    printf("Using port %d\n", port);
+
+    mqtt_wss_client client = mqtt_wss_new("main", mqtt_wss_log_cb, msg_callback, NULL);
+    if (!client) {
+        printf("Couldn't initialize mqtt_wss\n");
+        return 1;
+    }
+    while (!test_exit) {
+        printf("client_handle = %d\n", client_handle(client));
+    }
+    if (test_exit) {
+        mqtt_wss_disconnect(client, 2000);
+    }
+
+    mqtt_wss_destroy(client);
+    return 0;
+}
diff --git a/mqtt_websockets/src/ws_client.c b/mqtt_websockets/src/ws_client.c
new file mode 100644
index 000000000..47f633e74
--- /dev/null
+++ b/mqtt_websockets/src/ws_client.c
@@ -0,0 +1,744 @@
+// Copyright (C) 2020 Timotej Šiškovič
+// SPDX-License-Identifier: GPL-3.0-only
+//
+// This program is free software: you can redistribute it and/or modify it
+// under the terms of the GNU General Public License as published by the Free Software Foundation, version 3.
+//
+// This program 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 General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along with this program.
+// If not, see <https://www.gnu.org/licenses/>.
+
+#include <fcntl.h>
+#include <unistd.h>
+#include <string.h>
+#include <errno.h>
+#include <ctype.h>
+
+#include <openssl/evp.h>
+
+#include "ws_client.h"
+#include "common_internal.h"
+
+#ifdef MQTT_WEBSOCKETS_DEBUG
+#include "../c-rbuf/src/ringbuffer_internal.h"
+#endif
+
+#define UNIT_LOG_PREFIX "ws_client: "
+#define FATAL(fmt, ...) mws_fatal(client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+#define ERROR(fmt, ...) mws_error(client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+#define WARN(fmt, ...)  mws_warn (client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+#define INFO(fmt, ...)  mws_info (client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+#define DEBUG(fmt, ...) mws_debug(client->log, UNIT_LOG_PREFIX fmt, ##__VA_ARGS__)
+
+const char *websocket_upgrage_hdr = "GET /mqtt HTTP/1.1\x0D\x0A"
+                              "Host: %s\x0D\x0A"
+                              "Upgrade: websocket\x0D\x0A"
+                              "Connection: Upgrade\x0D\x0A"
+                              "Sec-WebSocket-Key: %s\x0D\x0A"
+                              "Origin: http://example.com\x0D\x0A"
+                              "Sec-WebSocket-Protocol: mqtt\x0D\x0A"
+                              "Sec-WebSocket-Version: 13\x0D\x0A\x0D\x0A";
+
+const char *mqtt_protoid = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
+
+#define DEFAULT_RINGBUFFER_SIZE (1024*128)
+#define ENTROPY_SOURCE "/dev/urandom"
+ws_client *ws_client_new(size_t buf_size, char **host, mqtt_wss_log_ctx_t log)
+{
+    ws_client *client;
+
+    if(!host)
+        return NULL;
+
+    client = mw_calloc(1, sizeof(ws_client));
+    if (!client)
+        return NULL;
+
+    client->host = host;
+    client->log = log;
+
+    client->buf_read = rbuf_create(buf_size ? buf_size : DEFAULT_RINGBUFFER_SIZE);
+    if (!client->buf_read)
+        goto cleanup;
+
+    client->buf_write = rbuf_create(buf_size ? buf_size : DEFAULT_RINGBUFFER_SIZE);
+    if (!client->buf_write)
+        goto cleanup_1;
+
+    client->buf_to_mqtt = rbuf_create(buf_size ? buf_size : DEFAULT_RINGBUFFER_SIZE);
+    if (!client->buf_to_mqtt)
+        goto cleanup_2;
+
+    client->entropy_fd = open(ENTROPY_SOURCE, O_RDONLY);
+    if (client->entropy_fd < 1) {
+        ERROR("Error opening entropy source \"" ENTROPY_SOURCE "\". Reason: \"%s\"", strerror(errno));
+        goto cleanup_3;
+    }
+
+    return client;
+
+cleanup_3:
+    rbuf_free(client->buf_to_mqtt);
+cleanup_2:
+    rbuf_free(client->buf_write);
+cleanup_1:
+    rbuf_free(client->buf_read);
+cleanup:
+    mw_free(client);
+    return NULL;
+}
+
+void ws_client_free_headers(ws_client *client)
+{
+    struct http_header *ptr = client->hs.headers;
+    struct http_header *tmp;
+
+    while (ptr) {
+        tmp = ptr;
+        ptr = ptr->next;
+        mw_free(tmp);
+    }
+
+    client->hs.headers = NULL;
+    client->hs.headers_tail = NULL;
+    client->hs.hdr_count = 0;
+}
+
+void ws_client_destroy(ws_client *client)
+{
+    ws_client_free_headers(client);
+    mw_free(client->hs.nonce_reply);
+    mw_free(client->hs.http_reply_msg);
+    close(client->entropy_fd);
+    rbuf_free(client->buf_read);
+    rbuf_free(client->buf_write);
+    rbuf_free(client->buf_to_mqtt);
+    mw_free(client);
+}
+
+void ws_client_reset(ws_client *client)
+{
+    ws_client_free_headers(client);
+    mw_free(client->hs.nonce_reply);
+    client->hs.nonce_reply = NULL;
+    mw_free(client->hs.http_reply_msg);
+    client->hs.http_reply_msg = NULL;
+    rbuf_flush(client->buf_read);
+    rbuf_flush(client->buf_write);
+    rbuf_flush(client->buf_to_mqtt);
+    client->state = WS_RAW;
+    client->hs.hdr_state = WS_HDR_HTTP;
+    client->rx.parse_state = WS_FIRST_2BYTES;
+}
+
+#define MAX_HTTP_HDR_COUNT 128
+int ws_client_add_http_header(ws_client *client, struct http_header *hdr)
+{
+    if (client->hs.hdr_count > MAX_HTTP_HDR_COUNT) {
+        ERROR("Too many HTTP response header fields");
+        return -1;
+    }
+
+    if (client->hs.headers)
+        client->hs.headers_tail->next = hdr;
+    else
+        client->hs.headers = hdr;
+
+    client->hs.headers_tail = hdr;
+    client->hs.hdr_count++;
+
+    return 0;
+}
+
+int ws_client_want_write(ws_client *client)
+{
+    return rbuf_bytes_available(client->buf_write);
+}
+
+#define RAND_SRC "/dev/urandom"
+static int ws_client_get_nonce(ws_client *client, char *dest, unsigned int size)
+{
+    // we do not need crypto secure random here
+    // it's just used for protocol negotiation
+    int rd;
+    int f = open(RAND_SRC, O_RDONLY);
+    if (f < 0) {
+        ERROR("Error opening \"%s\". Err: \"%s\"", RAND_SRC, strerror(errno));
+        return -2;
+    }
+
+    if ((rd = read(f, dest, size)) > 0) {
+        close(f);
+        return rd;
+    }
+    close(f);
+    return -1;
+}
+
+#define WEBSOCKET_NONCE_SIZE 16
+#define TEMP_BUF_SIZE 4096
+int ws_client_start_handshake(ws_client *client)
+{
+    char nonce[WEBSOCKET_NONCE_SIZE];
+    char nonce_b64[256];
+    char second[TEMP_BUF_SIZE];
+    unsigned int md_len;
+    unsigned char *digest;
+    EVP_MD_CTX *md_ctx;
+    const EVP_MD *md;
+
+    if(!*client->host) {
+        ERROR("Hostname has not been set. We should not be able to come here!");
+        return 1;
+    }
+
+    ws_client_get_nonce(client, nonce, WEBSOCKET_NONCE_SIZE);
+    EVP_EncodeBlock((unsigned char *)nonce_b64, (const unsigned char *)nonce, WEBSOCKET_NONCE_SIZE);
+    snprintf(second, TEMP_BUF_SIZE, websocket_upgrage_hdr,
+        *client->host,
+        nonce_b64);
+    if(rbuf_bytes_free(client->buf_write) < strlen(second)) {
+        ERROR("Write buffer capacity too low.");
+        return 1;
+    }
+
+    rbuf_push(client->buf_write, second, strlen(second));
+    client->state = WS_HANDSHAKE;
+
+    //Calculating expected Sec-WebSocket-Accept reply
+    snprintf(second, TEMP_BUF_SIZE, "%s%s", nonce_b64, mqtt_protoid);
+
+#if (OPENSSL_VERSION_NUMBER < OPENSSL_VERSION_110)
+    md_ctx = EVP_MD_CTX_create();
+#else
+    md_ctx = EVP_MD_CTX_new();
+#endif
+    if (md_ctx == NULL) {
+        ERROR("Cant create EVP_MD Context");
+        return 1;
+    }
+
+    md = EVP_get_digestbyname("sha1");
+    if (!md) {
+        ERROR("Unknown message digest");
+        return 1;
+    }
+
+    if ((digest = (unsigned char *)OPENSSL_malloc(EVP_MD_size(EVP_sha256()))) == NULL) {
+        ERROR("Cant alloc digest");
+        return 1;
+    }
+
+    EVP_DigestInit_ex(md_ctx, md, NULL);
+    EVP_DigestUpdate(md_ctx, second, strlen(second));
+    EVP_DigestFinal_ex(md_ctx, digest, &md_len);
+
+    EVP_EncodeBlock((unsigned char *)nonce_b64, digest, md_len);
+
+    mw_free(client->hs.nonce_reply);
+    client->hs.nonce_reply = mw_strdup(nonce_b64);
+
+    OPENSSL_free(digest);
+
+#if (OPENSSL_VERSION_NUMBER < OPENSSL_VERSION_110)
+    EVP_MD_CTX_destroy(md_ctx);
+#else
+    EVP_MD_CTX_free(md_ctx);
+#endif
+
+    return 0;
+}
+
+#define BUF_READ_MEMCMP_CONST(const, err)                                                                              \
+    if (rbuf_memcmp_n(client->buf_read, const, strlen(const))) {                                                       \
+        ERROR(err);                                                                                                    \
+        rbuf_flush(client->buf_read);                                                                                  \
+        return WS_CLIENT_PROTOCOL_ERROR;                                                                               \
+    }
+
+#define BUF_READ_CHECK_AT_LEAST(x)                                                                                     \
+    if (rbuf_bytes_available(client->buf_read) < x)                                                                    \
+        return WS_CLIENT_NEED_MORE_BYTES;
+
+#define MAX_HTTP_LINE_LENGTH 1024*4
+#define HTTP_SC_LENGTH 4 // "XXX " http status code as C string
+#define WS_CLIENT_HTTP_HDR "HTTP/1.1 "
+#define WS_CONN_ACCEPT "sec-websocket-accept"
+#define HTTP_HDR_SEPARATOR ": "
+#define WS_NONCE_STRLEN_B64 28
+#define WS_HTTP_NEWLINE "\r\n"
+#define HTTP_HEADER_NAME_MAX_LEN 256
+#if HTTP_HEADER_NAME_MAX_LEN > MAX_HTTP_LINE_LENGTH
+#error "Buffer too small"
+#endif
+#if WS_NONCE_STRLEN_B64 > MAX_HTTP_LINE_LENGTH
+#error "Buffer too small"
+#endif
+
+#define HTTP_HDR_LINE_CHECK_LIMIT(x) if ((x) >= MAX_HTTP_LINE_LENGTH) \
+{ \
+    ERROR("HTTP line received is too long. Maximum is %d", MAX_HTTP_LINE_LENGTH); \
+    return WS_CLIENT_PROTOCOL_ERROR; \
+}
+
+int ws_client_parse_handshake_resp(ws_client *client)
+{
+    char buf[HTTP_SC_LENGTH];
+    int idx_crlf, idx_sep;
+    char *ptr;
+    size_t bytes;
+    switch (client->hs.hdr_state) {
+        case WS_HDR_HTTP:
+            BUF_READ_CHECK_AT_LEAST(strlen(WS_CLIENT_HTTP_HDR))
+            BUF_READ_MEMCMP_CONST(WS_CLIENT_HTTP_HDR, "Expected \"HTTP1.1\" header");
+            rbuf_bump_tail(client->buf_read, strlen(WS_CLIENT_HTTP_HDR));
+            client->hs.hdr_state = WS_HDR_RC;
+            break;
+        case WS_HDR_RC:
+            BUF_READ_CHECK_AT_LEAST(HTTP_SC_LENGTH); // "XXX " http return code
+            rbuf_pop(client->buf_read, buf, HTTP_SC_LENGTH);
+            if (buf[HTTP_SC_LENGTH - 1] != 0x20) {
+                ERROR("HTTP status code received is not terminated by space (0x20)");
+                return WS_CLIENT_PROTOCOL_ERROR;
+            }
+            buf[HTTP_SC_LENGTH - 1] = 0;
+            client->hs.http_code = atoi(buf);
+            if (client->hs.http_code < 100 || client->hs.http_code >= 600) {
+                ERROR("HTTP status code received not in valid range 100-600");
+                return WS_CLIENT_PROTOCOL_ERROR;
+            }
+            client->hs.hdr_state = WS_HDR_ENDLINE;
+            break;
+        case WS_HDR_ENDLINE:
+            ptr = rbuf_find_bytes(client->buf_read, WS_HTTP_NEWLINE, strlen(WS_HTTP_NEWLINE), &idx_crlf);
+            if (!ptr) {
+                bytes = rbuf_bytes_available(client->buf_read);
+                HTTP_HDR_LINE_CHECK_LIMIT(bytes);
+                return WS_CLIENT_NEED_MORE_BYTES;
+            }
+            HTTP_HDR_LINE_CHECK_LIMIT(idx_crlf);
+
+            client->hs.http_reply_msg = mw_malloc(idx_crlf+1);
+            rbuf_pop(client->buf_read, client->hs.http_reply_msg, idx_crlf);
+            client->hs.http_reply_msg[idx_crlf] = 0;
+            rbuf_bump_tail(client->buf_read, strlen(WS_HTTP_NEWLINE));
+            client->hs.hdr_state = WS_HDR_PARSE_HEADERS;
+            break;
+        case WS_HDR_PARSE_HEADERS:
+            ptr = rbuf_find_bytes(client->buf_read, WS_HTTP_NEWLINE, strlen(WS_HTTP_NEWLINE), &idx_crlf);
+            if (!ptr) {
+                bytes = rbuf_bytes_available(client->buf_read);
+                HTTP_HDR_LINE_CHECK_LIMIT(bytes);
+                return WS_CLIENT_NEED_MORE_BYTES;
+            }
+            HTTP_HDR_LINE_CHECK_LIMIT(idx_crlf);
+
+            if (!idx_crlf) { // empty line, header end
+                rbuf_bump_tail(client->buf_read, strlen(WS_HTTP_NEWLINE));
+                client->hs.hdr_state = WS_HDR_PARSE_DONE;
+                return 0;
+            }
+
+            ptr = rbuf_find_bytes(client->buf_read, HTTP_HDR_SEPARATOR, strlen(HTTP_HDR_SEPARATOR), &idx_sep);
+            if (!ptr || idx_sep > idx_crlf) {
+                ERROR("Expected HTTP hdr field key/value separator \": \" before endline in non empty HTTP header line");
+                return WS_CLIENT_PROTOCOL_ERROR;
+            }
+            if (idx_crlf == idx_sep + (int)strlen(HTTP_HDR_SEPARATOR)) {
+                ERROR("HTTP Header value cannot be empty");
+                return WS_CLIENT_PROTOCOL_ERROR;
+            }
+
+            if (idx_sep > HTTP_HEADER_NAME_MAX_LEN) {
+                ERROR("HTTP header too long (%d)", idx_sep);
+                return WS_CLIENT_PROTOCOL_ERROR;
+            }
+
+            struct http_header *hdr = mw_calloc(1, sizeof(struct http_header) + idx_crlf); //idx_crlf includes ": " that will be used as 2 \0 bytes
+            hdr->key = ((char*)hdr) + sizeof(struct http_header);
+            hdr->value = hdr->key + idx_sep + 1;
+
+            bytes = rbuf_pop(client->buf_read, hdr->key, idx_sep);
+            rbuf_bump_tail(client->buf_read, strlen(HTTP_HDR_SEPARATOR));
+
+            bytes = rbuf_pop(client->buf_read, hdr->value, idx_crlf - idx_sep - strlen(HTTP_HDR_SEPARATOR));
+            rbuf_bump_tail(client->buf_read, strlen(WS_HTTP_NEWLINE));
+
+            for (int i = 0; hdr->key[i]; i++)
+                hdr->key[i] = tolower(hdr->key[i]);
+
+//            DEBUG("HTTP header \"%s\" received. Value \"%s\"", hdr->key, hdr->value);
+
+            if (ws_client_add_http_header(client, hdr))
+                return WS_CLIENT_PROTOCOL_ERROR;
+
+            if (!strcmp(hdr->key, WS_CONN_ACCEPT)) {
+                if (strcmp(client->hs.nonce_reply, hdr->value)) {
+                    ERROR("Received NONCE \"%s\" does not match expected nonce of \"%s\"", hdr->value, client->hs.nonce_reply);
+                    return WS_CLIENT_PROTOCOL_ERROR;
+                }
+                client->hs.nonce_matched = 1;
+            }
+
+            break;
+        case WS_HDR_PARSE_DONE:
+            if (!client->hs.nonce_matched) {
+                ERROR("Missing " WS_CONN_ACCEPT " header");
+                return WS_CLIENT_PROTOCOL_ERROR;
+            }
+            if (client->hs.http_code != 101) {
+                ERROR("HTTP return code not 101. Received %d with msg \"%s\".", client->hs.http_code, client->hs.http_reply_msg);
+                return WS_CLIENT_PROTOCOL_ERROR;
+            }
+
+            client->state = WS_ESTABLISHED;
+            client->hs.hdr_state = WS_HDR_ALL_DONE;
+            INFO("Websocket Connection Accepted By Server");
+            return WS_CLIENT_PARSING_DONE;
+        case WS_HDR_ALL_DONE:
+            FATAL("This is error we should never come here!");
+            return WS_CLIENT_PROTOCOL_ERROR;
+    }
+    return 0;
+}
+
+#define BYTE_MSB          0x80
+#define WS_FINAL_FRAG     BYTE_MSB
+#define WS_PAYLOAD_MASKED BYTE_MSB
+
+static inline size_t get_ws_hdr_size(size_t payload_size)
+{
+    size_t hdr_len = 2 + 4 /*mask*/;
+    if(payload_size > 125)
+        hdr_len += 2;
+    if(payload_size > 65535)
+        hdr_len += 6;
+    return hdr_len;
+}
+
+#define MAX_POSSIBLE_HDR_LEN 14
+int ws_client_send(ws_client *client, enum websocket_opcode frame_type, const char *data, size_t size)
+{
+    // TODO maybe? implement fragmenting, it is not necessary though
+    // as both tested MQTT brokers have no reuirement of one MQTT envelope
+    // be equal to one WebSockets envelope. Therefore there is no need to send
+    // one big MQTT message as single fragmented WebSocket envelope
+    char hdr[MAX_POSSIBLE_HDR_LEN];
+    char *ptr = hdr;
+    char *mask;
+    int size_written = 0;
+    size_t j = 0;
+
+    size_t w_buff_free = rbuf_bytes_free(client->buf_write);
+    size_t hdr_len = get_ws_hdr_size(size);
+
+    if (w_buff_free < hdr_len * 2) {
+#ifdef DEBUG_ULTRA_VERBOSE
+        DEBUG("Write buffer full. Can't write requested %d size.", size);
+#endif
+        return 0;
+    }
+
+    if (w_buff_free < (hdr_len + size)) {
+#ifdef DEBUG_ULTRA_VERBOSE
+        DEBUG("Can't write whole MQTT packet of %d bytes into the buffer. Will do partial send of %d.", size, w_buff_free - hdr_len);
+#endif
+        size = w_buff_free - hdr_len;
+        hdr_len = get_ws_hdr_size(size);
+        // the actual needed header size might decrease if we cut number of bytes
+        // if decrease of size crosses 65535 or 125 boundary
+        // but I can live with that at least for now
+        // worst case is we have 6 more bytes we could have written
+        // no bigus dealus
+    }
+
+    *ptr++ = frame_type | WS_FINAL_FRAG;
+
+    //generate length
+    *ptr = WS_PAYLOAD_MASKED;
+    if (size > 65535) {
+        *ptr++ |= 0x7f;
+        uint64_t be = htobe64(size);
+        memcpy(ptr, (void *)&be, sizeof(be));
+        ptr += sizeof(be);
+    } else if (size > 125) {
+        *ptr++ |= 0x7e;
+        uint16_t be = htobe16(size);
+        memcpy(ptr, (void *)&be, sizeof(be));
+        ptr += sizeof(be);
+    } else
+        *ptr++ |= size;
+    
+    mask = ptr;
+    if (read(client->entropy_fd, mask, sizeof(uint32_t)) < (ssize_t)sizeof(uint32_t)) {
+        ERROR("Unable to get mask from \"" ENTROPY_SOURCE "\"");
+        return -2;
+    }
+
+    rbuf_push(client->buf_write, hdr, hdr_len);
+
+    if (!size)
+        return 0;
+
+    // copy and mask data in the write ringbuffer
+    while (size - size_written) {
+        size_t writable_bytes;
+        char *w_ptr = rbuf_get_linear_insert_range(client->buf_write, &writable_bytes);
+        if(!writable_bytes)
+            break;
+
+        writable_bytes = (writable_bytes > size) ? (size - size_written) : writable_bytes;
+
+        memcpy(w_ptr, &data[size_written], writable_bytes);
+        rbuf_bump_head(client->buf_write, writable_bytes);
+
+        for (size_t i = 0; i < writable_bytes; i++, j++)
+            w_ptr[i] ^= mask[j % 4];
+        size_written += writable_bytes;
+    }
+    return size_written;
+}
+
+static int check_opcode(ws_client *client,enum websocket_opcode oc)
+{
+    switch(oc) {
+        case WS_OP_BINARY_FRAME:
+        case WS_OP_CONNECTION_CLOSE:
+        case WS_OP_PING:
+            return 0;
+        case WS_OP_CONTINUATION_FRAME:
+            FATAL("WS_OP_CONTINUATION_FRAME NOT IMPLEMENTED YET!!!!");
+            return 0;
+        case WS_OP_TEXT_FRAME:
+            FATAL("WS_OP_TEXT_FRAME NOT IMPLEMENTED YET!!!!");
+            return 0;
+        case WS_OP_PONG:
+            FATAL("WS_OP_PONG NOT IMPLEMENTED YET!!!!");
+            return 0;
+        default:
+            return WS_CLIENT_PROTOCOL_ERROR;
+    }
+}
+
+static inline void ws_client_rx_post_hdr_state(ws_client *client)
+{
+    switch(client->rx.opcode) {
+        case WS_OP_BINARY_FRAME:
+            client->rx.parse_state = WS_PAYLOAD_DATA;
+            return;
+        case WS_OP_CONNECTION_CLOSE:
+            client->rx.parse_state = WS_PAYLOAD_CONNECTION_CLOSE;
+            return;
+        case WS_OP_PING:
+            client->rx.parse_state = WS_PAYLOAD_PING_REQ_PAYLOAD;
+            return;
+        default:
+            client->rx.parse_state = WS_PAYLOAD_SKIP_UNKNOWN_PAYLOAD;
+            return;
+    }
+}
+
+#define LONGEST_POSSIBLE_HDR_PART 8
+int ws_client_process_rx_ws(ws_client *client)
+{
+    char buf[LONGEST_POSSIBLE_HDR_PART];
+    size_t size;
+    switch (client->rx.parse_state) {
+        case WS_FIRST_2BYTES:
+            BUF_READ_CHECK_AT_LEAST(2);
+            rbuf_pop(client->buf_read, buf, 2);
+            client->rx.opcode = buf[0] & (char)~BYTE_MSB;
+
+            if (!(buf[0] & (char)~WS_FINAL_FRAG)) {
+                ERROR("Not supporting fragmented messages yet!");
+                return WS_CLIENT_PROTOCOL_ERROR;
+            }
+
+            if (check_opcode(client, client->rx.opcode) == WS_CLIENT_PROTOCOL_ERROR)
+                return WS_CLIENT_PROTOCOL_ERROR;
+
+            if (buf[1] & (char)WS_PAYLOAD_MASKED) {
+                ERROR("Mask is not allowed in Server->Client Websocket direction.");
+                return WS_CLIENT_PROTOCOL_ERROR;
+            }
+
+            switch (buf[1]) {
+                case 127:
+                    client->rx.parse_state = WS_PAYLOAD_EXTENDED_64;
+                    break;
+                case 126:
+                    client->rx.parse_state = WS_PAYLOAD_EXTENDED_16;
+                    break;
+                default:
+                    client->rx.payload_length = buf[1];
+                    ws_client_rx_post_hdr_state(client);
+            }
+            break;
+        case WS_PAYLOAD_EXTENDED_16:
+            BUF_READ_CHECK_AT_LEAST(2);
+            rbuf_pop(client->buf_read, buf, 2);
+            client->rx.payload_length = be16toh(*((uint16_t *)buf));
+            ws_client_rx_post_hdr_state(client);
+            break;
+        case WS_PAYLOAD_EXTENDED_64:
+            BUF_READ_CHECK_AT_LEAST(LONGEST_POSSIBLE_HDR_PART);
+            rbuf_pop(client->buf_read, buf, LONGEST_POSSIBLE_HDR_PART);
+            client->rx.payload_length = be64toh(*((uint64_t *)buf));
+            ws_client_rx_post_hdr_state(client);
+            break;
+        case WS_PAYLOAD_DATA:
+            // TODO not pretty?
+            while (client->rx.payload_processed < client->rx.payload_length) {
+                size_t remaining = client->rx.payload_length - client->rx.payload_processed;
+                if (!rbuf_bytes_available(client->buf_read))
+                    return WS_CLIENT_NEED_MORE_BYTES;
+                char *insert = rbuf_get_linear_insert_range(client->buf_to_mqtt, &size);
+                if (!insert) {
+#ifdef DEBUG_ULTRA_VERBOSE
+                    DEBUG("BUFFER TOO FULL. Avail %d req %d", (int)size, (int)remaining);
+#endif
+                    return WS_CLIENT_BUFFER_FULL;
+                }
+                size = (size > remaining) ? remaining : size;
+                size = rbuf_pop(client->buf_read, insert, size);
+                rbuf_bump_head(client->buf_to_mqtt, size);
+                client->rx.payload_processed += size;
+            }
+            client->rx.parse_state = WS_PACKET_DONE;
+            break;
+        case WS_PAYLOAD_CONNECTION_CLOSE:
+            // for WS_OP_CONNECTION_CLOSE allowed is
+            // a) empty payload
+            // b) 2byte reason code
+            // c) 2byte reason code followed by message
+            if (client->rx.payload_length == 1) {
+                ERROR("WebScoket CONNECTION_CLOSE can't have payload of size 1");
+                return WS_CLIENT_PROTOCOL_ERROR;
+            }
+            if (!client->rx.payload_length) {
+                INFO("WebSocket server closed the connection without giving reason.");
+                client->rx.parse_state = WS_PACKET_DONE;
+                break;
+            }
+            client->rx.parse_state = WS_PAYLOAD_CONNECTION_CLOSE_EC;
+            break;
+        case WS_PAYLOAD_CONNECTION_CLOSE_EC:
+            BUF_READ_CHECK_AT_LEAST(sizeof(uint16_t));
+
+            rbuf_pop(client->buf_read, buf, sizeof(uint16_t));
+            client->rx.specific_data.op_close.ec = be16toh(*((uint16_t *)buf));
+            client->rx.payload_processed += sizeof(uint16_t);
+
+            if(client->rx.payload_processed == client->rx.payload_length) {
+                INFO("WebSocket server closed the connection with EC=%d. Without message.",
+                    client->rx.specific_data.op_close.ec);
+                client->rx.parse_state = WS_PACKET_DONE;
+                break;
+            }
+            client->rx.parse_state = WS_PAYLOAD_CONNECTION_CLOSE_MSG;
+            break;
+        case WS_PAYLOAD_CONNECTION_CLOSE_MSG:
+            if (!client->rx.specific_data.op_close.reason)
+                client->rx.specific_data.op_close.reason = mw_malloc(client->rx.payload_length + 1);
+
+            while (client->rx.payload_processed < client->rx.payload_length) {
+                if (!rbuf_bytes_available(client->buf_read))
+                    return WS_CLIENT_NEED_MORE_BYTES;
+                client->rx.payload_processed += rbuf_pop(client->buf_read,
+                                                         &client->rx.specific_data.op_close.reason[client->rx.payload_processed - sizeof(uint16_t)],
+                                                         client->rx.payload_length - client->rx.payload_processed);
+            }
+            client->rx.specific_data.op_close.reason[client->rx.payload_length] = 0;
+            INFO("WebSocket server closed the connection with EC=%d and reason \"%s\"",
+                client->rx.specific_data.op_close.ec,
+                client->rx.specific_data.op_close.reason);
+            mw_free(client->rx.specific_data.op_close.reason);
+            client->rx.specific_data.op_close.reason = NULL;
+            client->rx.parse_state = WS_PACKET_DONE;
+            break;
+        case WS_PAYLOAD_SKIP_UNKNOWN_PAYLOAD:
+            BUF_READ_CHECK_AT_LEAST(client->rx.payload_length);
+            WARN("Skipping Websocket Packet of unsupported/unknown type");
+            if (client->rx.payload_length)
+                rbuf_bump_tail(client->buf_read, client->rx.payload_length);
+            client->rx.parse_state = WS_PACKET_DONE;
+            return WS_CLIENT_PARSING_DONE;
+        case WS_PAYLOAD_PING_REQ_PAYLOAD:
+            if (client->rx.payload_length > rbuf_get_capacity(client->buf_read) / 2) {
+                ERROR("Ping arrived with payload which is too big!");
+                return WS_CLIENT_INTERNAL_ERROR;
+            }
+            BUF_READ_CHECK_AT_LEAST(client->rx.payload_length);
+            client->rx.specific_data.ping_msg = mw_malloc(client->rx.payload_length);
+            rbuf_pop(client->buf_read, client->rx.specific_data.ping_msg, client->rx.payload_length);
+            // TODO schedule this instead of sending right away
+            // then attempt to send as soon as buffer space clears up
+            size = ws_client_send(client, WS_OP_PONG, client->rx.specific_data.ping_msg, client->rx.payload_length);
+            if (size != client->rx.payload_length) {
+                ERROR("Unable to send the PONG as one packet back. Closing connection.");
+                return WS_CLIENT_PROTOCOL_ERROR;
+            }
+            client->rx.parse_state = WS_PACKET_DONE;
+            return WS_CLIENT_PARSING_DONE;
+        case WS_PACKET_DONE:
+            client->rx.parse_state = WS_FIRST_2BYTES;
+            client->rx.payload_processed = 0;
+            if (client->rx.opcode == WS_OP_CONNECTION_CLOSE)
+                return WS_CLIENT_CONNECTION_CLOSED;
+            return WS_CLIENT_PARSING_DONE;
+        default:
+            FATAL("Unknown parse state");
+            return WS_CLIENT_INTERNAL_ERROR;
+    }
+    return 0;
+}
+
+int ws_client_process(ws_client *client)
+{
+    int ret;
+    switch(client->state) {
+        case WS_RAW:
+            if (ws_client_start_handshake(client))
+                return WS_CLIENT_INTERNAL_ERROR;
+            return WS_CLIENT_NEED_MORE_BYTES;
+        case WS_HANDSHAKE:
+            do {
+                ret = ws_client_parse_handshake_resp(client);
+                if (ret == WS_CLIENT_PROTOCOL_ERROR)
+                    client->state = WS_ERROR;
+                if (ret == WS_CLIENT_PARSING_DONE && client->state == WS_ESTABLISHED)
+                    ret = WS_CLIENT_NEED_MORE_BYTES;
+            } while (!ret);
+            break;
+        case WS_ESTABLISHED:
+            do {
+                ret = ws_client_process_rx_ws(client);
+                switch(ret) {
+                    case WS_CLIENT_PROTOCOL_ERROR:
+                        client->state = WS_ERROR;
+                        break;
+                    case WS_CLIENT_CONNECTION_CLOSED:
+                        client->state = WS_CONN_CLOSED_GRACEFUL;
+                        break;
+                }
+                // if ret == 0 we can continue parsing
+                // if ret == WS_CLIENT_PARSING_DONE we processed
+                //    one websocket packet and attempt processing
+                //    next one if data available in the buffer
+            } while (!ret || ret == WS_CLIENT_PARSING_DONE);
+            break;
+        case WS_ERROR:
+            ERROR("ws_client is in error state. Restart the connection!");
+            return WS_CLIENT_PROTOCOL_ERROR;
+        case WS_CONN_CLOSED_GRACEFUL:
+            ERROR("Connection has been gracefully closed. Calling this is useless (and probably bug) until you reconnect again.");
+            return WS_CLIENT_CONNECTION_CLOSED;
+        default:
+            FATAL("Unknown connection state! Probably memory corruption.");
+            return WS_CLIENT_INTERNAL_ERROR;
+    }
+    return ret;
+}