Afonso Muralha / Pixels Camp LED matrix badge - Files

AsyncTCP.cpp
/*
  Asynchronous TCP library for Espressif MCUs

  Copyright (c) 2016 Hristo Gochkov. All rights reserved.
  This file is part of the esp8266 core for Arduino environment.

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  Lesser General Public License for more details.

  You should have received a copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 */

#include "Arduino.h"

#include "AsyncTCP.h"
extern "C"{
#include "lwip/opt.h"
#include "lwip/tcp.h"
#include "lwip/inet.h"
#include "lwip/dns.h"
}

#if CONFIG_FREERTOS_UNICORE
#define ASYNCTCP_RUNNING_CORE 0
#else
#define ASYNCTCP_RUNNING_CORE 1
#endif

/*
 * TCP/IP Event Task
 * */

typedef enum {
    LWIP_TCP_SENT, LWIP_TCP_RECV, LWIP_TCP_ERROR, LWIP_TCP_POLL, LWIP_TCP_CLEAR
} lwip_event_t;

typedef struct {
        lwip_event_t event;
        void *arg;
        union {
                struct {
                        void * pcb;
                        int8_t err;
                } connected;
                struct {
                        int8_t err;
                } error;
                struct {
                        tcp_pcb * pcb;
                        uint16_t len;
                } sent;
                struct {
                        tcp_pcb * pcb;
                        pbuf * pb;
                        int8_t err;
                } recv;
                struct {
                        tcp_pcb * pcb;
                } poll;
                struct {
                        tcp_pcb * pcb;
                        int8_t err;
                } accept;
                struct {
                        const char * name;
                        ip_addr_t addr;
                } dns;
        };
} lwip_event_packet_t;

static xQueueHandle _async_queue;
static TaskHandle_t _async_service_task_handle = NULL;

static inline bool _init_async_event_queue(){
    if(!_async_queue){
        _async_queue = xQueueCreate(32, sizeof(lwip_event_packet_t *));
        if(!_async_queue){
            return false;
        }
    }
    return true;
}

static inline bool _send_async_event(lwip_event_packet_t ** e){
    return _async_queue && xQueueSend(_async_queue, e, portMAX_DELAY) == pdPASS;
}

static inline bool _prepend_async_event(lwip_event_packet_t ** e){
    return _async_queue && xQueueSendToFront(_async_queue, e, portMAX_DELAY) == pdPASS;
}

static inline bool _get_async_event(lwip_event_packet_t ** e){
    return _async_queue && xQueueReceive(_async_queue, e, portMAX_DELAY) == pdPASS;
}

static bool _remove_events_with_arg(void * arg){
    lwip_event_packet_t * first_packet = NULL;
    lwip_event_packet_t * packet = NULL;

    if(!_async_queue){
        return false;
    }
    //figure out which is the first packet so we can keep the order
    while(!first_packet){
        if(xQueueReceive(_async_queue, &first_packet, 0) != pdPASS){
            return false;
        }
        //discard packet if matching
        if((int)first_packet->arg == (int)arg){
            //ets_printf("X: 0x%08x\n", (uint32_t)first_packet->arg);
            free(first_packet);
            first_packet = NULL;
        //return first packet to the back of the queue
        } else if(xQueueSend(_async_queue, &first_packet, portMAX_DELAY) != pdPASS){
            return false;
        }
    }

    while(xQueuePeek(_async_queue, &packet, 0) == pdPASS && packet != first_packet){
        if(xQueueReceive(_async_queue, &packet, 0) != pdPASS){
            return false;
        }
        if((int)packet->arg == (int)arg){
            //ets_printf("X: 0x%08x\n", (uint32_t)packet->arg);
            free(packet);
            packet = NULL;
        } else if(xQueueSend(_async_queue, &packet, portMAX_DELAY) != pdPASS){
            return false;
        }
    }
    return true;
}

static void _handle_async_event(lwip_event_packet_t * e){
    if(e->event == LWIP_TCP_CLEAR){
        _remove_events_with_arg(e->arg);
    } else if(e->event == LWIP_TCP_RECV){
        //ets_printf("%c: 0x%08x 0x%08x\n", e->recv.pb?'R':'D', e->arg, e->recv.pcb);
        AsyncClient::_s_recv(e->arg, e->recv.pcb, e->recv.pb, e->recv.err);
    } else if(e->event == LWIP_TCP_SENT){
        //ets_printf("S: 0x%08x 0x%08x\n", e->arg, e->sent.pcb);
        AsyncClient::_s_sent(e->arg, e->sent.pcb, e->sent.len);
    } else if(e->event == LWIP_TCP_POLL){
        //ets_printf("P: 0x%08x 0x%08x\n", e->arg, e->poll.pcb);
        AsyncClient::_s_poll(e->arg, e->poll.pcb);
    } else if(e->event == LWIP_TCP_ERROR){
        AsyncClient::_s_error(e->arg, e->error.err);
    }
    free((void*)(e));
}

static void _async_service_task(void *pvParameters){
    lwip_event_packet_t * packet = NULL;
    for (;;) {
        if(_get_async_event(&packet)){
            _handle_async_event(packet);
        }
    }
    vTaskDelete(NULL);
    _async_service_task_handle = NULL;
}
/*
static void _stop_async_task(){
    if(_async_service_task_handle){
        vTaskDelete(_async_service_task_handle);
        _async_service_task_handle = NULL;
    }
}
*/
static bool _start_async_task(){
    if(!_init_async_event_queue()){
        return false;
    }
    if(!_async_service_task_handle){
        xTaskCreatePinnedToCore(_async_service_task, "async_tcp", 8192, NULL, 3, &_async_service_task_handle, ASYNCTCP_RUNNING_CORE);
        if(!_async_service_task_handle){
            return false;
        }
    }
    return true;
}

/*
 * LwIP Callbacks
 * */

static int8_t _tcp_clear_events(void * arg) {
    lwip_event_packet_t * e = (lwip_event_packet_t *)malloc(sizeof(lwip_event_packet_t));
    e->event = LWIP_TCP_CLEAR;
    e->arg = arg;
    if (!_prepend_async_event(&e)) {
        free((void*)(e));
    }
    return ERR_OK;
}

static int8_t _tcp_poll(void * arg, struct tcp_pcb * pcb) {
    lwip_event_packet_t * e = (lwip_event_packet_t *)malloc(sizeof(lwip_event_packet_t));
    e->event = LWIP_TCP_POLL;
    e->arg = arg;
    e->poll.pcb = pcb;
    if (!_send_async_event(&e)) {
        free((void*)(e));
    }
    return ERR_OK;
}

static int8_t _tcp_recv(void * arg, struct tcp_pcb * pcb, struct pbuf *pb, int8_t err) {
    lwip_event_packet_t * e = (lwip_event_packet_t *)malloc(sizeof(lwip_event_packet_t));
    e->event = LWIP_TCP_RECV;
    e->arg = arg;
    e->recv.pcb = pcb;
    e->recv.pb = pb;
    e->recv.err = err;
    if (!_send_async_event(&e)) {
        free((void*)(e));
    }
    return ERR_OK;
}

static int8_t _tcp_sent(void * arg, struct tcp_pcb * pcb, uint16_t len) {
    lwip_event_packet_t * e = (lwip_event_packet_t *)malloc(sizeof(lwip_event_packet_t));
    e->event = LWIP_TCP_SENT;
    e->arg = arg;
    e->sent.pcb = pcb;
    e->sent.len = len;
    if (!_send_async_event(&e)) {
        free((void*)(e));
    }
    return ERR_OK;
}

static void _tcp_error(void * arg, int8_t err) {
    lwip_event_packet_t * e = (lwip_event_packet_t *)malloc(sizeof(lwip_event_packet_t));
    e->event = LWIP_TCP_ERROR;
    e->arg = arg;
    e->error.err = err;
    if (!_send_async_event(&e)) {
        free((void*)(e));
    }
}

/*
 * TCP/IP API Calls
 * */

#include "lwip/priv/tcpip_priv.h"

typedef struct {
    struct tcpip_api_call_data call;
    tcp_pcb * pcb;
    int8_t err;
    union {
            struct {
                    const char* data;
                    size_t size;
                    uint8_t apiflags;
            } write;
            size_t received;
            struct {
                    ip_addr_t * addr;
                    uint16_t port;
                    tcp_connected_fn cb;
            } connect;
            struct {
                    ip_addr_t * addr;
                    uint16_t port;
            } bind;
            uint8_t backlog;
    };
} tcp_api_call_t;

static err_t _tcp_output_api(struct tcpip_api_call_data *api_call_msg){
    tcp_api_call_t * msg = (tcp_api_call_t *)api_call_msg;
    if(msg->pcb){
      msg->err = tcp_output(msg->pcb);
    } else {
      msg->err = 0;
    }
    return msg->err;
}

static esp_err_t _tcp_output(tcp_pcb * pcb) {
    tcp_api_call_t msg;
    msg.pcb = pcb;
    tcpip_api_call(_tcp_output_api, (struct tcpip_api_call_data*)&msg);
    return msg.err;
}

static err_t _tcp_write_api(struct tcpip_api_call_data *api_call_msg){
    tcp_api_call_t * msg = (tcp_api_call_t *)api_call_msg;
    msg->err = tcp_write(msg->pcb, msg->write.data, msg->write.size, msg->write.apiflags);
    return msg->err;
}

static esp_err_t _tcp_write(tcp_pcb * pcb, const char* data, size_t size, uint8_t apiflags) {
    tcp_api_call_t msg;
    msg.pcb = pcb;
    msg.write.data = data;
    msg.write.size = size;
    msg.write.apiflags = apiflags;
    tcpip_api_call(_tcp_write_api, (struct tcpip_api_call_data*)&msg);
    return msg.err;
}

static err_t _tcp_recved_api(struct tcpip_api_call_data *api_call_msg){
    tcp_api_call_t * msg = (tcp_api_call_t *)api_call_msg;
    msg->err = 0;
    tcp_recved(msg->pcb, msg->received);
    return msg->err;
}

static esp_err_t _tcp_recved(tcp_pcb * pcb, size_t len) {
    tcp_api_call_t msg;
    msg.pcb = pcb;
    msg.received = len;
    tcpip_api_call(_tcp_recved_api, (struct tcpip_api_call_data*)&msg);
    return msg.err;
}

static err_t _tcp_connect_api(struct tcpip_api_call_data *api_call_msg){
    tcp_api_call_t * msg = (tcp_api_call_t *)api_call_msg;
    msg->err = tcp_connect(msg->pcb, msg->connect.addr, msg->connect.port, msg->connect.cb);
    return msg->err;
}

static esp_err_t _tcp_connect(tcp_pcb * pcb, ip_addr_t * addr, uint16_t port, tcp_connected_fn cb) {
    tcp_api_call_t msg;
    msg.pcb = pcb;
    msg.connect.addr = addr;
    msg.connect.port = port;
    msg.connect.cb = cb;
    tcpip_api_call(_tcp_connect_api, (struct tcpip_api_call_data*)&msg);
    return msg.err;
}

static err_t _tcp_close_api(struct tcpip_api_call_data *api_call_msg){
    tcp_api_call_t * msg = (tcp_api_call_t *)api_call_msg;
    msg->err = tcp_close(msg->pcb);
    return msg->err;
}

static esp_err_t _tcp_close(tcp_pcb * pcb) {
    tcp_api_call_t msg;
    msg.pcb = pcb;
    //ets_printf("close 0x%08x\n", (uint32_t)pcb);
    tcpip_api_call(_tcp_close_api, (struct tcpip_api_call_data*)&msg);
    return msg.err;
}

static err_t _tcp_abort_api(struct tcpip_api_call_data *api_call_msg){
    tcp_api_call_t * msg = (tcp_api_call_t *)api_call_msg;
    msg->err = 0;
    tcp_abort(msg->pcb);
    return msg->err;
}

static esp_err_t _tcp_abort(tcp_pcb * pcb) {
    tcp_api_call_t msg;
    msg.pcb = pcb;
    //ets_printf("abort 0x%08x\n", (uint32_t)pcb);
    tcpip_api_call(_tcp_abort_api, (struct tcpip_api_call_data*)&msg);
    return msg.err;
}

static err_t _tcp_bind_api(struct tcpip_api_call_data *api_call_msg){
    tcp_api_call_t * msg = (tcp_api_call_t *)api_call_msg;
    msg->err = tcp_bind(msg->pcb, msg->bind.addr, msg->bind.port);
    return msg->err;
}

static esp_err_t _tcp_bind(tcp_pcb * pcb, ip_addr_t * addr, uint16_t port) {
    tcp_api_call_t msg;
    msg.pcb = pcb;
    msg.bind.addr = addr;
    msg.bind.port = port;
    tcpip_api_call(_tcp_bind_api, (struct tcpip_api_call_data*)&msg);
    return msg.err;
}

static err_t _tcp_listen_api(struct tcpip_api_call_data *api_call_msg){
    tcp_api_call_t * msg = (tcp_api_call_t *)api_call_msg;
    msg->err = 0;
    msg->pcb = tcp_listen_with_backlog(msg->pcb, msg->backlog);
    return msg->err;
}

static tcp_pcb * _tcp_listen_with_backlog(tcp_pcb * pcb, uint8_t backlog) {
    tcp_api_call_t msg;
    msg.pcb = pcb;
    msg.backlog = backlog?backlog:0xFF;
    tcpip_api_call(_tcp_listen_api, (struct tcpip_api_call_data*)&msg);
    return msg.pcb;
}
#define _tcp_listen(p) _tcp_listen_with_backlog(p, 0xFF);



/*
  Async TCP Client
 */

AsyncClient::AsyncClient(tcp_pcb* pcb)
: _connect_cb(0)
, _connect_cb_arg(0)
, _discard_cb(0)
, _discard_cb_arg(0)
, _sent_cb(0)
, _sent_cb_arg(0)
, _error_cb(0)
, _error_cb_arg(0)
, _recv_cb(0)
, _recv_cb_arg(0)
, _pb_cb(0)
, _pb_cb_arg(0)
, _timeout_cb(0)
, _timeout_cb_arg(0)
, _pcb_busy(false)
, _pcb_sent_at(0)
, _close_pcb(false)
, _ack_pcb(true)
, _rx_last_packet(0)
, _rx_since_timeout(0)
, _ack_timeout(ASYNC_MAX_ACK_TIME)
, _connect_port(0)
, prev(NULL)
, next(NULL)
, _in_lwip_thread(false)
{
    //ets_printf("+: 0x%08x\n", (uint32_t)this);

    _pcb = pcb;
    if(_pcb){
        _rx_last_packet = millis();
        tcp_arg(_pcb, this);
        tcp_recv(_pcb, &_tcp_recv);
        tcp_sent(_pcb, &_tcp_sent);
        tcp_err(_pcb, &_tcp_error);
        tcp_poll(_pcb, &_tcp_poll, 1);
        //ets_printf("accept 0x%08x\n", (uint32_t)_pcb);
    }
}

AsyncClient::~AsyncClient(){
    if(_pcb)
        _close();

    //ets_printf("-: 0x%08x\n", (uint32_t)this);
}

bool AsyncClient::connect(IPAddress ip, uint16_t port){
    if (_pcb){
        log_w("already connected, state %d", _pcb->state);
        return false;
    }
    if(!_start_async_task()){
        log_e("failed to start task");
        return false;
    }

    ip_addr_t addr;
    addr.type = IPADDR_TYPE_V4;
    addr.u_addr.ip4.addr = ip;

    tcp_pcb* pcb = tcp_new_ip_type(IPADDR_TYPE_V4);
    if (!pcb){
        log_e("pcb == NULL");
        return false;
    }

    tcp_arg(pcb, this);
    tcp_err(pcb, &_tcp_error);
    if(_in_lwip_thread){
        tcp_connect(pcb, &addr, port,(tcp_connected_fn)&_s_connected);
    } else {
        _tcp_connect(pcb, &addr, port,(tcp_connected_fn)&_s_connected);
    }
    return true;
}

AsyncClient& AsyncClient::operator=(const AsyncClient& other){
    if (_pcb)
        _close();

    _pcb = other._pcb;
    if (_pcb) {
        _rx_last_packet = millis();
        tcp_arg(_pcb, this);
        tcp_recv(_pcb, &_tcp_recv);
        tcp_sent(_pcb, &_tcp_sent);
        tcp_err(_pcb, &_tcp_error);
        tcp_poll(_pcb, &_tcp_poll, 1);
    }
    return *this;
}

int8_t AsyncClient::_connected(void* pcb, int8_t err){
    _pcb = reinterpret_cast<tcp_pcb*>(pcb);
    if(_pcb){
        _rx_last_packet = millis();
        _pcb_busy = false;
        tcp_recv(_pcb, &_tcp_recv);
        tcp_sent(_pcb, &_tcp_sent);
        tcp_poll(_pcb, &_tcp_poll, 1);
    }
    _in_lwip_thread = true;
    if(_connect_cb)
        _connect_cb(_connect_cb_arg, this);
    _in_lwip_thread = false;
    return ERR_OK;
}

int8_t AsyncClient::_close(){
    //ets_printf("X: 0x%08x\n", (uint32_t)this);
    int8_t err = ERR_OK;
    if(_pcb) {
        //log_i("");
        tcp_arg(_pcb, NULL);
        tcp_sent(_pcb, NULL);
        tcp_recv(_pcb, NULL);
        tcp_err(_pcb, NULL);
        tcp_poll(_pcb, NULL, 0);
        _tcp_clear_events(this);
        if(_in_lwip_thread){
            err = tcp_close(_pcb);
        } else {
            err = _tcp_close(_pcb);
        }
        if(err != ERR_OK) {
            err = abort();
        }
        _pcb = NULL;
        if(_discard_cb)
            _discard_cb(_discard_cb_arg, this);
    }
    return err;
}

void AsyncClient::_error(int8_t err) {
    if(_pcb){
        tcp_arg(_pcb, NULL);
        tcp_sent(_pcb, NULL);
        tcp_recv(_pcb, NULL);
        tcp_err(_pcb, NULL);
        tcp_poll(_pcb, NULL, 0);
        _pcb = NULL;
    }
    if(_error_cb)
        _error_cb(_error_cb_arg, this, err);
    if(_discard_cb)
        _discard_cb(_discard_cb_arg, this);
}

int8_t AsyncClient::_sent(tcp_pcb* pcb, uint16_t len) {
    _in_lwip_thread = false;
    _rx_last_packet = millis();
    //log_i("%u", len);
    _pcb_busy = false;
    if(_sent_cb)
        _sent_cb(_sent_cb_arg, this, len, (millis() - _pcb_sent_at));
    return ERR_OK;
}

int8_t AsyncClient::_recv(tcp_pcb* pcb, pbuf* pb, int8_t err) {
    if(!_pcb || pcb != _pcb){
        log_e("0x%08x != 0x%08x", (uint32_t)pcb, (uint32_t)_pcb);
        if(pb){
            pbuf_free(pb);
        }
        return ERR_OK;
    }
    _in_lwip_thread = false;
    if(pb == NULL){
        return _close();
    }

    while(pb != NULL){
        _rx_last_packet = millis();
        //we should not ack before we assimilate the data
        //log_i("%u", pb->len);
        //Serial.write((const uint8_t *)pb->payload, pb->len);
        _ack_pcb = true;
        pbuf *b = pb;
        pb = b->next;
        b->next = NULL;
        if(_pb_cb){
            _pb_cb(_pb_cb_arg, this, b);
        } else {
            if(_recv_cb)
                _recv_cb(_recv_cb_arg, this, b->payload, b->len);
            if(!_ack_pcb)
                _rx_ack_len += b->len;
            else
                _tcp_recved(pcb, b->len);
            pbuf_free(b);
        }
    }
    return ERR_OK;
}

int8_t AsyncClient::_poll(tcp_pcb* pcb){
    _in_lwip_thread = false;
    // Close requested
    if(_close_pcb){
        _close_pcb = false;
        _close();
        return ERR_OK;
    }
    uint32_t now = millis();

    // ACK Timeout
    if(_pcb_busy && _ack_timeout && (now - _pcb_sent_at) >= _ack_timeout){
        _pcb_busy = false;
        log_w("ack timeout %d", pcb->state);
        if(_timeout_cb)
            _timeout_cb(_timeout_cb_arg, this, (now - _pcb_sent_at));
        return ERR_OK;
    }
    // RX Timeout
    if(_rx_since_timeout && (now - _rx_last_packet) >= (_rx_since_timeout * 1000)){
        log_w("rx timeout %d", pcb->state);
        _close();
        return ERR_OK;
    }
    // Everything is fine
    if(_poll_cb)
        _poll_cb(_poll_cb_arg, this);
    return ERR_OK;
}

void AsyncClient::_dns_found(struct ip_addr *ipaddr){
    _in_lwip_thread = true;
    if(ipaddr){
        connect(IPAddress(ipaddr->u_addr.ip4.addr), _connect_port);
    } else {
        log_e("dns fail");
        if(_error_cb)
            _error_cb(_error_cb_arg, this, -55);
        if(_discard_cb)
            _discard_cb(_discard_cb_arg, this);
    }
    _in_lwip_thread = false;
}

bool AsyncClient::operator==(const AsyncClient &other) {
    return _pcb == other._pcb;
}

bool AsyncClient::connect(const char* host, uint16_t port){
    ip_addr_t addr;
    err_t err = dns_gethostbyname(host, &addr, (dns_found_callback)&_s_dns_found, this);
    if(err == ERR_OK) {
        return connect(IPAddress(addr.u_addr.ip4.addr), port);
    } else if(err == ERR_INPROGRESS) {
        _connect_port = port;
        return true;
    }
    log_e("error: %d", err);
    return false;
}

int8_t AsyncClient::abort(){
    if(_pcb) {
        log_w("state %d", _pcb->state);
        if(_in_lwip_thread){
            tcp_abort(_pcb);
        } else {
            _tcp_abort(_pcb);
        }
        _pcb = NULL;
    }
    return ERR_ABRT;
}

void AsyncClient::close(bool now){
    if(_pcb){
        if(_in_lwip_thread){
            tcp_recved(_pcb, _rx_ack_len);
        } else {
            _tcp_recved(_pcb, _rx_ack_len);
        }
    }
    if(now)
        _close();
    else
        _close_pcb = true;
}

void AsyncClient::stop() {
    close(false);
}

bool AsyncClient::free(){
    if(!_pcb)
        return true;
    if(_pcb->state == 0 || _pcb->state > 4)
        return true;
    return false;
}

size_t AsyncClient::space(){
    if((_pcb != NULL) && (_pcb->state == 4)){
        return tcp_sndbuf(_pcb);
    }
    return 0;
}

size_t AsyncClient::write(const char* data) {
    if(data == NULL)
        return 0;
    return write(data, strlen(data));
}

size_t AsyncClient::write(const char* data, size_t size, uint8_t apiflags) {
    size_t will_send = add(data, size, apiflags);
    if(!will_send || !send())
        return 0;
    return will_send;
}


size_t AsyncClient::add(const char* data, size_t size, uint8_t apiflags) {
    if(!_pcb || size == 0 || data == NULL)
        return 0;
    size_t room = space();
    if(!room)
        return 0;
    size_t will_send = (room < size) ? room : size;
    int8_t err = ERR_OK;
    if(_in_lwip_thread){
        err = tcp_write(_pcb, data, will_send, apiflags);
    } else {
        err = _tcp_write(_pcb, data, will_send, apiflags);
    }
    if(err != ERR_OK)
        return 0;
    return will_send;
}

bool AsyncClient::send(){
    int8_t err = ERR_OK;
    if(_in_lwip_thread){
        err = tcp_output(_pcb);
    } else {
        err = _tcp_output(_pcb);
    }
    if(err == ERR_OK){
        _pcb_busy = true;
        _pcb_sent_at = millis();
        return true;
    }
    return false;
}

size_t AsyncClient::ack(size_t len){
    if(len > _rx_ack_len)
        len = _rx_ack_len;
    if(len){
        if(_in_lwip_thread){
            tcp_recved(_pcb, len);
        } else {
            _tcp_recved(_pcb, len);
        }
    }
    _rx_ack_len -= len;
    return len;
}

// Operators

AsyncClient & AsyncClient::operator+=(const AsyncClient &other) {
    if(next == NULL){
        next = (AsyncClient*)(&other);
        next->prev = this;
    } else {
        AsyncClient *c = next;
        while(c->next != NULL) c = c->next;
        c->next =(AsyncClient*)(&other);
        c->next->prev = c;
    }
    return *this;
}

void AsyncClient::setRxTimeout(uint32_t timeout){
    _rx_since_timeout = timeout;
}

uint32_t AsyncClient::getRxTimeout(){
    return _rx_since_timeout;
}

uint32_t AsyncClient::getAckTimeout(){
    return _ack_timeout;
}

void AsyncClient::setAckTimeout(uint32_t timeout){
    _ack_timeout = timeout;
}

void AsyncClient::setNoDelay(bool nodelay){
    if(!_pcb)
        return;
    if(nodelay)
        tcp_nagle_disable(_pcb);
    else
        tcp_nagle_enable(_pcb);
}

bool AsyncClient::getNoDelay(){
    if(!_pcb)
        return false;
    return tcp_nagle_disabled(_pcb);
}

uint16_t AsyncClient::getMss(){
    if(_pcb)
        return tcp_mss(_pcb);
    return 0;
}

uint32_t AsyncClient::getRemoteAddress() {
    if(!_pcb)
        return 0;
    return _pcb->remote_ip.u_addr.ip4.addr;
}

uint16_t AsyncClient::getRemotePort() {
    if(!_pcb)
        return 0;
    return _pcb->remote_port;
}

uint32_t AsyncClient::getLocalAddress() {
    if(!_pcb)
        return 0;
    return _pcb->local_ip.u_addr.ip4.addr;
}

uint16_t AsyncClient::getLocalPort() {
    if(!_pcb)
        return 0;
    return _pcb->local_port;
}

IPAddress AsyncClient::remoteIP() {
    return IPAddress(getRemoteAddress());
}

uint16_t AsyncClient::remotePort() {
    return getRemotePort();
}

IPAddress AsyncClient::localIP() {
    return IPAddress(getLocalAddress());
}

uint16_t AsyncClient::localPort() {
    return getLocalPort();
}

uint8_t AsyncClient::state() {
    if(!_pcb)
        return 0;
    return _pcb->state;
}

bool AsyncClient::connected(){
    if (!_pcb)
        return false;
    return _pcb->state == 4;
}

bool AsyncClient::connecting(){
    if (!_pcb)
        return false;
    return _pcb->state > 0 && _pcb->state < 4;
}

bool AsyncClient::disconnecting(){
    if (!_pcb)
        return false;
    return _pcb->state > 4 && _pcb->state < 10;
}

bool AsyncClient::disconnected(){
    if (!_pcb)
        return true;
    return _pcb->state == 0 || _pcb->state == 10;
}

bool AsyncClient::freeable(){
    if (!_pcb)
        return true;
    return _pcb->state == 0 || _pcb->state > 4;
}

bool AsyncClient::canSend(){
    return space() > 0;
}

void AsyncClient::ackPacket(struct pbuf * pb){
  if(!pb){
    return;
  }
  _tcp_recved(_pcb, pb->len);
  pbuf_free(pb);
}


// Callback Setters

void AsyncClient::onConnect(AcConnectHandler cb, void* arg){
    _connect_cb = cb;
    _connect_cb_arg = arg;
}

void AsyncClient::onDisconnect(AcConnectHandler cb, void* arg){
    _discard_cb = cb;
    _discard_cb_arg = arg;
}

void AsyncClient::onAck(AcAckHandler cb, void* arg){
    _sent_cb = cb;
    _sent_cb_arg = arg;
}

void AsyncClient::onError(AcErrorHandler cb, void* arg){
    _error_cb = cb;
    _error_cb_arg = arg;
}

void AsyncClient::onData(AcDataHandler cb, void* arg){
    _recv_cb = cb;
    _recv_cb_arg = arg;
}

void AsyncClient::onPacket(AcPacketHandler cb, void* arg){
  _pb_cb = cb;
  _pb_cb_arg = arg;
}

void AsyncClient::onTimeout(AcTimeoutHandler cb, void* arg){
    _timeout_cb = cb;
    _timeout_cb_arg = arg;
}

void AsyncClient::onPoll(AcConnectHandler cb, void* arg){
    _poll_cb = cb;
    _poll_cb_arg = arg;
}


void AsyncClient::_s_dns_found(const char * name, struct ip_addr * ipaddr, void * arg){
    if(arg){
        reinterpret_cast<AsyncClient*>(arg)->_dns_found(ipaddr);
    } else {
        log_e("Bad Arg: 0x%08x", arg);
    }
}

int8_t AsyncClient::_s_poll(void * arg, struct tcp_pcb * pcb) {
    if(arg && pcb){
        reinterpret_cast<AsyncClient*>(arg)->_poll(pcb);
    } else {
        log_e("Bad Args: 0x%08x 0x%08x", arg, pcb);
    }
    return ERR_OK;
}

int8_t AsyncClient::_s_recv(void * arg, struct tcp_pcb * pcb, struct pbuf *pb, int8_t err) {
    if(arg && pcb){
        reinterpret_cast<AsyncClient*>(arg)->_recv(pcb, pb, err);
    } else {
        if(pb){
           pbuf_free(pb);
        }
        log_e("Bad Args: 0x%08x 0x%08x", arg, pcb);
    }
    return ERR_OK;
}

int8_t AsyncClient::_s_sent(void * arg, struct tcp_pcb * pcb, uint16_t len) {
    if(arg && pcb){
        reinterpret_cast<AsyncClient*>(arg)->_sent(pcb, len);
    } else {
        log_e("Bad Args: 0x%08x 0x%08x", arg, pcb);
    }
    return ERR_OK;
}

void AsyncClient::_s_error(void * arg, int8_t err) {
    if(arg){
        reinterpret_cast<AsyncClient*>(arg)->_error(err);
    } else {
        log_e("Bad Arg: 0x%08x", arg);
    }
}

int8_t AsyncClient::_s_connected(void * arg, void * pcb, int8_t err){
    if(arg && pcb){
        reinterpret_cast<AsyncClient*>(arg)->_connected(pcb, err);
    } else {
        log_e("Bad Args: 0x%08x 0x%08x", arg, pcb);
    }
    return ERR_OK;
}

const char * AsyncClient::errorToString(int8_t error){
    switch(error){
        case 0: return "OK";
        case -1: return "Out of memory error";
        case -2: return "Buffer error";
        case -3: return "Timeout";
        case -4: return "Routing problem";
        case -5: return "Operation in progress";
        case -6: return "Illegal value";
        case -7: return "Operation would block";
        case -8: return "Connection aborted";
        case -9: return "Connection reset";
        case -10: return "Connection closed";
        case -11: return "Not connected";
        case -12: return "Illegal argument";
        case -13: return "Address in use";
        case -14: return "Low-level netif error";
        case -15: return "Already connected";
        case -55: return "DNS failed";
        default: return "UNKNOWN";
    }
}

const char * AsyncClient::stateToString(){
    switch(state()){
        case 0: return "Closed";
        case 1: return "Listen";
        case 2: return "SYN Sent";
        case 3: return "SYN Received";
        case 4: return "Established";
        case 5: return "FIN Wait 1";
        case 6: return "FIN Wait 2";
        case 7: return "Close Wait";
        case 8: return "Closing";
        case 9: return "Last ACK";
        case 10: return "Time Wait";
        default: return "UNKNOWN";
    }
}

/*
  Async TCP Server
 */
struct pending_pcb {
        tcp_pcb* pcb;
        pbuf *pb;
        struct pending_pcb * next;
};

AsyncServer::AsyncServer(IPAddress addr, uint16_t port)
: _port(port)
, _addr(addr)
, _noDelay(false)
, _in_lwip_thread(false)
, _pcb(0)
, _connect_cb(0)
, _connect_cb_arg(0)
{}

AsyncServer::AsyncServer(uint16_t port)
: _port(port)
, _addr((uint32_t) IPADDR_ANY)
, _noDelay(false)
, _in_lwip_thread(false)
, _pcb(0)
, _connect_cb(0)
, _connect_cb_arg(0)
{}

AsyncServer::~AsyncServer(){
    end();
}

void AsyncServer::onClient(AcConnectHandler cb, void* arg){
    _connect_cb = cb;
    _connect_cb_arg = arg;
}

int8_t AsyncServer::_s_accept(void * arg, tcp_pcb * pcb, int8_t err){
    reinterpret_cast<AsyncServer*>(arg)->_accept(pcb, err);
    return ERR_OK;
}

int8_t AsyncServer::_accept(tcp_pcb* pcb, int8_t err){
    tcp_accepted(_pcb);
    if(_connect_cb){

        if (_noDelay)
            tcp_nagle_disable(pcb);
        else
            tcp_nagle_enable(pcb);

        AsyncClient *c = new AsyncClient(pcb);
        if(c){
            _in_lwip_thread = true;
            c->_in_lwip_thread = true;
            _connect_cb(_connect_cb_arg, c);
            c->_in_lwip_thread = false;
            _in_lwip_thread = false;
            return ERR_OK;
        }
    }
    if(tcp_close(pcb) != ERR_OK){
        tcp_abort(pcb);
    }
    log_e("FAIL");
    return ERR_OK;
}

void AsyncServer::begin(){
    if(_pcb)
        return;

    if(!_start_async_task()){
        log_e("failed to start task");
        return;
    }
    int8_t err;
    _pcb = tcp_new_ip_type(IPADDR_TYPE_V4);
    if (!_pcb){
        log_e("_pcb == NULL");
        return;
    }

    ip_addr_t local_addr;
    local_addr.type = IPADDR_TYPE_V4;
    local_addr.u_addr.ip4.addr = (uint32_t) _addr;
    err = _tcp_bind(_pcb, &local_addr, _port);

    if (err != ERR_OK) {
        _tcp_close(_pcb);
        log_e("bind error: %d", err);
        return;
    }

    static uint8_t backlog = 5;
    _pcb = _tcp_listen_with_backlog(_pcb, backlog);
    //_pcb = _tcp_listen(_pcb);
    if (!_pcb) {
        log_e("listen_pcb == NULL");
        return;
    }
    tcp_arg(_pcb, (void*) this);
    tcp_accept(_pcb, &_s_accept);
}

void AsyncServer::end(){
    if(_pcb){
        tcp_arg(_pcb, NULL);
        tcp_accept(_pcb, NULL);
        if(_in_lwip_thread){
            tcp_close(_pcb);
        } else {
            _tcp_close(_pcb);
        }
        _pcb = NULL;
    }
}

void AsyncServer::setNoDelay(bool nodelay){
    _noDelay = nodelay;
}

bool AsyncServer::getNoDelay(){
    return _noDelay;
}

uint8_t AsyncServer::status(){
    if (!_pcb)
        return 0;
    return _pcb->state;
}