Subversion Repositories ESP8266_P1_Meter

/**
 * @file  ESPAsyncTCPbuffer.cpp
 * @date  22.01.2016
 * @author Markus Sattler
 *
 * Copyright (c) 2015 Markus Sattler. All rights reserved.
 * This file is part of the Asynv TCP for ESP.
 *
 * 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 <debug.h>

#include "ESPAsyncTCPbuffer.h"


AsyncTCPbuffer::AsyncTCPbuffer(AsyncClient* client) {
    if(client == NULL) {
        DEBUG_ASYNC_TCP("[A-TCP] client is null!!!\n");
        panic();
    }

    _client = client;
    _TXbufferWrite = new (std::nothrow) cbuf(TCP_MSS);
    _TXbufferRead = _TXbufferWrite;
    _RXbuffer = new (std::nothrow) cbuf(100);
    _RXmode = ATB_RX_MODE_FREE;
    _rxSize = 0;
    _rxTerminator = 0x00;
    _rxReadBytesPtr = NULL;
    _rxReadStringPtr = NULL;
    _cbDisconnect = NULL;

    _cbRX = NULL;
    _cbDone = NULL;
    _attachCallbacks();
}

AsyncTCPbuffer::~AsyncTCPbuffer() {
    if(_client) {
        _client->close();
    }

    if(_RXbuffer) {
        delete _RXbuffer;
        _RXbuffer = NULL;
    }

    if(_TXbufferWrite) {
        // will be deleted in _TXbufferRead chain
        _TXbufferWrite = NULL;
    }

    if(_TXbufferRead) {
        cbuf * next = _TXbufferRead->next;
        delete _TXbufferRead;
        while(next != NULL) {
            _TXbufferRead = next;
            next = _TXbufferRead->next;
            delete _TXbufferRead;
        }
        _TXbufferRead = NULL;
    }
}

size_t AsyncTCPbuffer::write(String & data) {
    return write(data.c_str(), data.length());
}

size_t AsyncTCPbuffer::write(uint8_t data) {
    return write(&data, 1);
}

size_t AsyncTCPbuffer::write(const char* data) {
    return write((const uint8_t *) data, strlen(data));
}

size_t AsyncTCPbuffer::write(const char *data, size_t len) {
    return write((const uint8_t *) data, len);
}

/**
 * write data in to buffer and try to send the data
 * @param data
 * @param len
 * @return
 */
size_t AsyncTCPbuffer::write(const uint8_t *data, size_t len) {
    if(_TXbufferWrite == NULL || _client == NULL || !_client->connected() || data == NULL || len == 0) {
        return 0;
    }

    size_t bytesLeft = len;
    while(bytesLeft) {
        size_t w = _TXbufferWrite->write((const char*) data, bytesLeft);
        bytesLeft -= w;
        data += w;
        _sendBuffer();

        // add new buffer since we have more data
        if(_TXbufferWrite->full() && bytesLeft > 0) {

            // to less ram!!!
            if(ESP.getFreeHeap() < 4096) {
                DEBUG_ASYNC_TCP("[A-TCP] run out of Heap can not send all Data!\n");
                return (len - bytesLeft);
            }

            cbuf * next = new (std::nothrow) cbuf(TCP_MSS);
            if(next == NULL) {
                DEBUG_ASYNC_TCP("[A-TCP] run out of Heap!\n");
                panic();
            } else {
                DEBUG_ASYNC_TCP("[A-TCP] new cbuf\n");
            }

            // add new buffer to chain (current cbuf)
            _TXbufferWrite->next = next;

            // move ptr for next data
            _TXbufferWrite = next;
        }
    }

    return len;

}

/**
 * wait until all data has send out
 */
void AsyncTCPbuffer::flush() {
    while(!_TXbufferWrite->empty()) {
        while(connected() && !_client->canSend()) {
          delay(0);
        }
        if(!connected())
          return;
        _sendBuffer();
    }
}

void AsyncTCPbuffer::noCallback() {
    _RXmode = ATB_RX_MODE_NONE;
}

void AsyncTCPbuffer::readStringUntil(char terminator, String * str, AsyncTCPbufferDoneCb done) {
    if(_client == NULL) {
        return;
    }
    DEBUG_ASYNC_TCP("[A-TCP] readStringUntil terminator: %02X\n", terminator);
    _RXmode = ATB_RX_MODE_NONE;
    _cbDone = done;
    _rxReadStringPtr = str;
    _rxTerminator = terminator;
    _rxSize = 0;
    _RXmode = ATB_RX_MODE_TERMINATOR_STRING;
}

/*
 void AsyncTCPbuffer::readBytesUntil(char terminator, char *buffer, size_t length, AsyncTCPbufferDoneCb done) {
 _RXmode = ATB_RX_MODE_NONE;
 _cbDone = done;
 _rxReadBytesPtr = (uint8_t *) buffer;
 _rxTerminator = terminator;
 _rxSize = length;
 _RXmode = ATB_RX_MODE_TERMINATOR;
 _handleRxBuffer(NULL, 0);
 }

 void AsyncTCPbuffer::readBytesUntil(char terminator, uint8_t *buffer, size_t length, AsyncTCPbufferDoneCb done) {
 readBytesUntil(terminator, (char *) buffer, length, done);
 }
 */

void AsyncTCPbuffer::readBytes(char *buffer, size_t length, AsyncTCPbufferDoneCb done) {
    if(_client == NULL) {
        return;
    }
    DEBUG_ASYNC_TCP("[A-TCP] readBytes length: %d\n", length);
    _RXmode = ATB_RX_MODE_NONE;
    _cbDone = done;
    _rxReadBytesPtr = (uint8_t *) buffer;
    _rxSize = length;
    _RXmode = ATB_RX_MODE_READ_BYTES;
}

void AsyncTCPbuffer::readBytes(uint8_t *buffer, size_t length, AsyncTCPbufferDoneCb done) {
    readBytes((char *) buffer, length, done);
}

void AsyncTCPbuffer::onData(AsyncTCPbufferDataCb cb) {
    if(_client == NULL) {
        return;
    }
    DEBUG_ASYNC_TCP("[A-TCP] onData\n");
    _RXmode = ATB_RX_MODE_NONE;
    _cbDone = NULL;
    _cbRX = cb;
    _RXmode = ATB_RX_MODE_FREE;
}

void AsyncTCPbuffer::onDisconnect(AsyncTCPbufferDisconnectCb cb) {
    _cbDisconnect = cb;
}

IPAddress AsyncTCPbuffer::remoteIP() {
    if(!_client) {
        return IPAddress(0U);
    }
    return _client->remoteIP();
}

uint16_t AsyncTCPbuffer::remotePort() {
    if(!_client) {
        return 0;
    }
    return _client->remotePort();
}

bool AsyncTCPbuffer::connected() {
    if(!_client) {
        return false;
    }
    return _client->connected();
}

void AsyncTCPbuffer::stop() {

    if(!_client) {
        return;
    }
    _client->stop();
    _client = NULL;

    if(_cbDone) {
        switch(_RXmode) {
            case ATB_RX_MODE_READ_BYTES:
            case ATB_RX_MODE_TERMINATOR:
            case ATB_RX_MODE_TERMINATOR_STRING:
                _RXmode = ATB_RX_MODE_NONE;
                _cbDone(false, NULL);
                break;
            default:
              break;
        }
    }
    _RXmode = ATB_RX_MODE_NONE;
}

void AsyncTCPbuffer::close() {
    stop();
}


///--------------------------------

/**
 * attachCallbacks to AsyncClient class
 */
void AsyncTCPbuffer::_attachCallbacks() {
    if(!_client) {
        return;
    }
    DEBUG_ASYNC_TCP("[A-TCP] attachCallbacks\n");

    _client->onPoll([](void *obj, AsyncClient* c) {
        (void)c;
        AsyncTCPbuffer* b = ((AsyncTCPbuffer*)(obj));
        if((b->_TXbufferRead != NULL) && !b->_TXbufferRead->empty()) {
            b->_sendBuffer();
        }
        //    if(!b->_RXbuffer->empty()) {
        //       b->_handleRxBuffer(NULL, 0);
        //   }
    }, this);

    _client->onAck([](void *obj, AsyncClient* c, size_t len, uint32_t time) {
        (void)c;
        (void)len;
        (void)time;
        DEBUG_ASYNC_TCP("[A-TCP] onAck\n");
        ((AsyncTCPbuffer*)(obj))->_sendBuffer();
    }, this);

    _client->onDisconnect([](void *obj, AsyncClient* c) {
        DEBUG_ASYNC_TCP("[A-TCP] onDisconnect\n");
        AsyncTCPbuffer* b = ((AsyncTCPbuffer*)(obj));
        b->_client = NULL;
        bool del = true;
        if(b->_cbDisconnect) {
            del = b->_cbDisconnect(b);
        }
        delete c;
        if(del) {
            delete b;
        }
    }, this);

    _client->onData([](void *obj, AsyncClient* c, void *buf, size_t len) {
        (void)c;
        AsyncTCPbuffer* b = ((AsyncTCPbuffer*)(obj));
        b->_rxData((uint8_t *)buf, len);
    }, this);

    _client->onTimeout([](void *obj, AsyncClient* c, uint32_t time){
        (void)obj;
        (void)time;
        DEBUG_ASYNC_TCP("[A-TCP] onTimeout\n");
        c->close();
    }, this);

    DEBUG_ASYNC_TCP("[A-TCP] attachCallbacks Done.\n");
}

/**
 * send TX buffer if possible
 */
void AsyncTCPbuffer::_sendBuffer() {
    //DEBUG_ASYNC_TCP("[A-TCP] _sendBuffer...\n");
    size_t available = _TXbufferRead->available();
    if(available == 0 || _client == NULL || !_client->connected() || !_client->canSend()) {
        return;
    }

    while(connected() && (_client->space() > 0) && (_TXbufferRead->available() > 0) && _client->canSend()) {

        available = _TXbufferRead->available();

        if(available > _client->space()) {
            available = _client->space();
        }

        char *out = new (std::nothrow) char[available];
        if(out == NULL) {
            DEBUG_ASYNC_TCP("[A-TCP] to less heap, try later.\n");
            return;
        }

        // read data from buffer
        _TXbufferRead->peek(out, available);

        // send data
        size_t send = _client->write((const char*) out, available);
        if(send != available) {
            DEBUG_ASYNC_TCP("[A-TCP] write failed send: %d available: %d \n", send, available);
            if(!connected()) {
                DEBUG_ASYNC_TCP("[A-TCP] incomplete transfer, connection lost.\n");
            }
        }

        // remove really send data from buffer
        _TXbufferRead->remove(send);

        // if buffer is empty and there is a other buffer in chain delete the empty one
        if(_TXbufferRead->available() == 0 && _TXbufferRead->next != NULL) {
            cbuf * old = _TXbufferRead;
            _TXbufferRead = _TXbufferRead->next;
            delete old;
            DEBUG_ASYNC_TCP("[A-TCP] delete cbuf\n");
        }

        delete out;
    }

}

/**
 * called on incoming data
 * @param buf
 * @param len
 */
void AsyncTCPbuffer::_rxData(uint8_t *buf, size_t len) {
    if(!_client || !_client->connected()) {
        DEBUG_ASYNC_TCP("[A-TCP] not connected!\n");
        return;
    }
    if(!_RXbuffer) {
        DEBUG_ASYNC_TCP("[A-TCP] _rxData no _RXbuffer!\n");
        return;
    }
    DEBUG_ASYNC_TCP("[A-TCP] _rxData len: %d RXmode: %d\n", len, _RXmode);

    size_t handled = 0;

    if(_RXmode != ATB_RX_MODE_NONE) {
        handled = _handleRxBuffer((uint8_t *) buf, len);
        buf += handled;
        len -= handled;

        // handle as much as possible before using the buffer
        if(_RXbuffer->empty()) {
            while(_RXmode != ATB_RX_MODE_NONE && handled != 0 && len > 0) {
                handled = _handleRxBuffer(buf, len);
                buf += handled;
                len -= handled;
            }
        }
    }

    if(len > 0) {

        if(_RXbuffer->room() < len) {
            // to less space
            DEBUG_ASYNC_TCP("[A-TCP] _rxData buffer full try resize\n");
            _RXbuffer->resizeAdd((len + _RXbuffer->room()));

            if(_RXbuffer->room() < len) {
                DEBUG_ASYNC_TCP("[A-TCP] _rxData buffer to full can only handle %d!!!\n", _RXbuffer->room());
            }
        }

        _RXbuffer->write((const char *) (buf), len);
    }

    if(!_RXbuffer->empty() && _RXmode != ATB_RX_MODE_NONE) {
        // handle as much as possible data in buffer
        handled = _handleRxBuffer(NULL, 0);
        while(_RXmode != ATB_RX_MODE_NONE && handled != 0) {
            handled = _handleRxBuffer(NULL, 0);
        }
    }

    // clean up ram
    if(_RXbuffer->empty() && _RXbuffer->room() != 100) {
        _RXbuffer->resize(100);
    }

}

/**
 *
 */
size_t AsyncTCPbuffer::_handleRxBuffer(uint8_t *buf, size_t len) {
    if(!_client || !_client->connected() || _RXbuffer == NULL) {
        return 0;
    }

    DEBUG_ASYNC_TCP("[A-TCP] _handleRxBuffer len: %d RXmode: %d\n", len, _RXmode);

    size_t BufferAvailable = _RXbuffer->available();
    size_t r = 0;

    if(_RXmode == ATB_RX_MODE_NONE) {
        return 0;
    } else if(_RXmode == ATB_RX_MODE_FREE) {
        if(_cbRX == NULL) {
            return 0;
        }

        if(BufferAvailable > 0) {
            uint8_t * b = new (std::nothrow) uint8_t[BufferAvailable];
            if(b == NULL){
              panic(); //TODO: What action should this be ?
            }
            _RXbuffer->peek((char *) b, BufferAvailable);
            r = _cbRX(b, BufferAvailable);
            _RXbuffer->remove(r);
        }

        if(r == BufferAvailable && buf && (len > 0)) {
            return _cbRX(buf, len);
        } else {
            return 0;
        }

    } else if(_RXmode == ATB_RX_MODE_READ_BYTES) {
        if(_rxReadBytesPtr == NULL || _cbDone == NULL) {
            return 0;
        }

        size_t newReadCount = 0;

        if(BufferAvailable) {
            r = _RXbuffer->read((char *) _rxReadBytesPtr, _rxSize);
            _rxSize -= r;
            _rxReadBytesPtr += r;
        }

        if(_RXbuffer->empty() && (len > 0) && buf) {
            r = len;
            if(r > _rxSize) {
                r = _rxSize;
            }
            memcpy(_rxReadBytesPtr, buf, r);
            _rxReadBytesPtr += r;
            _rxSize -= r;
            newReadCount += r;
        }

        if(_rxSize == 0) {
            _RXmode = ATB_RX_MODE_NONE;
            _cbDone(true, NULL);
        }

        // add left over bytes to Buffer
        return newReadCount;

    } else if(_RXmode == ATB_RX_MODE_TERMINATOR) {
        // TODO implement read terminator non string

    } else if(_RXmode == ATB_RX_MODE_TERMINATOR_STRING) {
        if(_rxReadStringPtr == NULL || _cbDone == NULL) {
            return 0;
        }

        // handle Buffer
        if(BufferAvailable > 0) {
            while(!_RXbuffer->empty()) {
                char c = _RXbuffer->read();
                if(c == _rxTerminator || c == 0x00) {
                    _RXmode = ATB_RX_MODE_NONE;
                    _cbDone(true, _rxReadStringPtr);
                    return 0;
                } else {
                    (*_rxReadStringPtr) += c;
                }
            }
        }

        if(_RXbuffer->empty() && (len > 0) && buf) {
            size_t newReadCount = 0;
            while(newReadCount < len) {
                char c = (char) *buf;
                buf++;
                newReadCount++;
                if(c == _rxTerminator || c == 0x00) {
                    _RXmode = ATB_RX_MODE_NONE;
                    _cbDone(true, _rxReadStringPtr);
                    return newReadCount;
                } else {
                    (*_rxReadStringPtr) += c;
                }
            }
            return newReadCount;
        }
    }

    return 0;
}