Stream::send() (#6979)

Author: d-a-v

Diff for: cores/esp8266/Client.h

@@ -26,15 +26,15 @@
 class Client: public Stream {
 
     public:
-        virtual int connect(IPAddress ip, uint16_t port) =0;
-        virtual int connect(const char *host, uint16_t port) =0;
-        virtual size_t write(uint8_t) =0;
-        virtual size_t write(const uint8_t *buf, size_t size) =0;
-        virtual int available() = 0;
-        virtual int read() = 0;
-        virtual int read(uint8_t *buf, size_t size) = 0;
-        virtual int peek() = 0;
-        virtual void flush() = 0;
+        virtual int connect(IPAddress ip, uint16_t port) = 0;
+        virtual int connect(const char *host, uint16_t port) = 0;
+        virtual size_t write(uint8_t) override = 0;
+        virtual size_t write(const uint8_t *buf, size_t size) override = 0;
+        virtual int available() override = 0;
+        virtual int read() override = 0;
+        virtual int read(uint8_t *buf, size_t size) override = 0;
+        virtual int peek() override = 0;
+        virtual void flush() override = 0;
         virtual void stop() = 0;
         virtual uint8_t connected() = 0;
         virtual operator bool() = 0;

Diff for: cores/esp8266/FS.cpp

@@ -66,7 +66,7 @@ int File::read() {
     return result;
 }
 
-size_t File::read(uint8_t* buf, size_t size) {
+int File::read(uint8_t* buf, size_t size) {
     if (!_p)
         return 0;
 

Diff for: cores/esp8266/FS.h

@@ -67,13 +67,14 @@ class File : public Stream
     size_t readBytes(char *buffer, size_t length) override {
         return read((uint8_t*)buffer, length);
     }
-    size_t read(uint8_t* buf, size_t size);
+    int read(uint8_t* buf, size_t size) override;
     bool seek(uint32_t pos, SeekMode mode);
     bool seek(uint32_t pos) {
         return seek(pos, SeekSet);
     }
     size_t position() const;
     size_t size() const;
+    virtual ssize_t streamRemaining() override { return (ssize_t)size() - (ssize_t)position(); }
     void close();
     operator bool() const;
     const char* name() const;
@@ -84,6 +85,7 @@ class File : public Stream
     bool isDirectory() const;
 
     // Arduino "class SD" methods for compatibility
+    //TODO use stream::send / check read(buf,size) result
     template<typename T> size_t write(T &src){
       uint8_t obuf[256];
       size_t doneLen = 0;

Diff for: cores/esp8266/FSImpl.h

@@ -30,7 +30,7 @@ class FileImpl {
 public:
     virtual ~FileImpl() { }
     virtual size_t write(const uint8_t *buf, size_t size) = 0;
-    virtual size_t read(uint8_t* buf, size_t size) = 0;
+    virtual int read(uint8_t* buf, size_t size) = 0;
     virtual void flush() = 0;
     virtual bool seek(uint32_t pos, SeekMode mode) = 0;
     virtual size_t position() const = 0;

Diff for: cores/esp8266/HardwareSerial.h

@@ -135,16 +135,45 @@ class HardwareSerial: public Stream
         // return -1 when data is unvailable (arduino api)
         return uart_peek_char(_uart);
     }
+
+    virtual bool hasPeekBufferAPI () const override
+    {
+        return true;
+    }
+
+    // return a pointer to available data buffer (size = available())
+    // semantic forbids any kind of read() before calling peekConsume()
+    const char* peekBuffer () override
+    {
+        return uart_peek_buffer(_uart);
+    }
+
+    // return number of byte accessible by peekBuffer()
+    size_t peekAvailable () override
+    {
+        return uart_peek_available(_uart);
+    }
+
+    // consume bytes after use (see peekBuffer)
+    void peekConsume (size_t consume) override
+    {
+        return uart_peek_consume(_uart, consume);
+    }
+
     int read(void) override
     {
         // return -1 when data is unvailable (arduino api)
         return uart_read_char(_uart);
     }
     // ::read(buffer, size): same as readBytes without timeout
-    size_t read(char* buffer, size_t size)
+    int read(char* buffer, size_t size)
     {
         return uart_read(_uart, buffer, size);
     }
+    int read(uint8_t* buffer, size_t size) override
+    {
+        return uart_read(_uart, (char*)buffer, size);
+    }
     size_t readBytes(char* buffer, size_t size) override;
     size_t readBytes(uint8_t* buffer, size_t size) override
     {

Diff for: cores/esp8266/Print.cpp

@@ -33,15 +33,7 @@
 
 /* default implementation: may be overridden */
 size_t Print::write(const uint8_t *buffer, size_t size) {
-
-#ifdef DEBUG_ESP_CORE
-    static char not_the_best_way [] PROGMEM STORE_ATTR = "Print::write(data,len) should be overridden for better efficiency\r\n";
-    static bool once = false;
-    if (!once) {
-        once = true;
-        os_printf_plus(not_the_best_way);
-    }
-#endif
+    IAMSLOW();
 
     size_t n = 0;
     while (size--) {

Diff for: cores/esp8266/Print.h

@@ -111,6 +111,10 @@ class Print {
         size_t println(void);
 
         virtual void flush() { /* Empty implementation for backward compatibility */ }
+
+        // by default write timeout is possible (outgoing data from network,serial..)
+        // (children can override to false (like String))
+        virtual bool outputCanTimeout () { return true; }
 };
 
 template<> size_t Print::printNumber(double number, uint8_t digits);

Diff for: cores/esp8266/Stream.cpp

@@ -22,6 +22,7 @@
 
 #include <Arduino.h>
 #include <Stream.h>
+
 #define PARSE_TIMEOUT 1000  // default number of milli-seconds to wait
 #define NO_SKIP_CHAR  1  // a magic char not found in a valid ASCII numeric field
 
@@ -210,6 +211,8 @@ float Stream::parseFloat(char skipChar) {
 // the buffer is NOT null terminated.
 //
 size_t Stream::readBytes(char *buffer, size_t length) {
+    IAMSLOW();
+
     size_t count = 0;
     while(count < length) {
         int c = timedRead();
@@ -258,3 +261,20 @@ String Stream::readStringUntil(char terminator) {
     }
     return ret;
 }
+
+// read what can be read, immediate exit on unavailable data
+// prototype similar to Arduino's `int Client::read(buf, len)`
+int Stream::read (uint8_t* buffer, size_t maxLen)
+{
+    IAMSLOW();
+
+    size_t nbread = 0;
+    while (nbread < maxLen && available())
+    {
+        int c = read();
+        if (c == -1)
+            break;
+        buffer[nbread++] = read();
+    }
+    return nbread;
+}

Diff for: cores/esp8266/Stream.h

@@ -22,10 +22,13 @@
 #ifndef Stream_h
 #define Stream_h
 
+#include <debug.h>
 #include <inttypes.h>
-#include "Print.h"
+#include <Print.h>
+#include <PolledTimeout.h>
+#include <sys/types.h> // ssize_t
 
-// compatability macros for testing
+// compatibility macros for testing
 /*
  #define   getInt()            parseInt()
  #define   getInt(skipChar)    parseInt(skipchar)
@@ -35,6 +38,15 @@
  readBytesBetween( pre_string, terminator, buffer, length)
  */
 
+// Arduino `Client: public Stream` class defines `virtual int read(uint8_t *buf, size_t size) = 0;`
+// This function is now imported into `Stream::` for `Stream::send*()`.
+// Other classes inheriting from `Stream::` and implementing `read(uint8_t *buf, size_t size)`
+// must consequently use `int` as return type, namely Hardware/SoftwareSerial, FileSystems...
+#define STREAM_READ_RETURNS_INT 1
+
+// Stream::send API is present
+#define STREAMSEND_API 1
+
 class Stream: public Print {
     protected:
         unsigned long _timeout = 1000;  // number of milliseconds to wait for the next char before aborting timed read
@@ -53,6 +65,7 @@ class Stream: public Print {
 // parsing methods
 
         void setTimeout(unsigned long timeout);  // sets maximum milliseconds to wait for stream data, default is 1 second
+        unsigned long getTimeout () const { return _timeout; }
 
         bool find(const char *target);   // reads data from the stream until the target string is found
         bool find(uint8_t *target) {
@@ -102,12 +115,114 @@ class Stream: public Print {
         virtual String readString();
         String readStringUntil(char terminator);
 
+        virtual int read (uint8_t* buffer, size_t len);
+        int read (char* buffer, size_t len) { return read((uint8_t*)buffer, len); }
+
+        //////////////////// extension: direct access to input buffer
+        // to provide when possible a pointer to available data for read
+
+        // informs user and ::to*() on effective buffered peek API implementation
+        // by default: not available
+        virtual bool hasPeekBufferAPI () const { return false; }
+
+        // returns number of byte accessible by peekBuffer()
+        virtual size_t peekAvailable () { return 0; }
+
+        // returns a pointer to available data buffer (size = peekAvailable())
+        // semantic forbids any kind of ::read()
+        //     - after calling peekBuffer()
+        //     - and before calling peekConsume()
+        virtual const char* peekBuffer () { return nullptr; }
+
+        // consumes bytes after peekBuffer() use
+        // (then ::read() is allowed)
+        virtual void peekConsume (size_t consume) { (void)consume; }
+
+        // by default read timeout is possible (incoming data from network,serial..)
+        // children can override to false (like String::)
+        virtual bool inputCanTimeout () { return true; }
+
+        // (outputCanTimeout() is defined in Print::)
+
+    ////////////////////////
+        //////////////////// extensions: Streaming streams to streams
+        // Stream::send*()
+        //
+        // Stream::send*() uses 1-copy transfers when peekBuffer API is
+        // available, or makes a regular transfer through a temporary buffer.
+        //
+        // - for efficiency, Stream classes should implement peekAPI when
+        //   possible
+        // - for an efficient timeout management, Print/Stream classes
+        //   should implement {output,input}CanTimeout()
+
+        using oneShotMs = esp8266::polledTimeout::oneShotFastMs;
+        static constexpr int temporaryStackBufferSize = 64;
+
+        // ::send*() methods:
+        // - always stop before timeout when "no-more-input-possible-data"
+        //   or "no-more-output-possible-data" condition is met
+        // - always return number of transfered bytes
+        // When result is 0 or less than requested maxLen, Print::getLastSend()
+        // contains an error reason.
+
+        // transfers already buffered / immediately available data (no timeout)
+        // returns number of transfered bytes
+        size_t sendAvailable (Print* to) { return sendGeneric(to, -1, -1, oneShotMs::alwaysExpired); }
+        size_t sendAvailable (Print& to) { return sendAvailable(&to); }
+
+        // transfers data until timeout
+        // returns number of transfered bytes
+        size_t sendAll (Print* to, const oneShotMs::timeType timeoutMs = oneShotMs::neverExpires) { return sendGeneric(to, -1, -1, timeoutMs); }
+        size_t sendAll (Print& to, const oneShotMs::timeType timeoutMs = oneShotMs::neverExpires) { return sendAll(&to, timeoutMs); }
+
+        // transfers data until a char is encountered (the char is swallowed but not transfered) with timeout
+        // returns number of transfered bytes
+        size_t sendUntil (Print* to, const int readUntilChar, const oneShotMs::timeType timeoutMs = oneShotMs::neverExpires) { return sendGeneric(to, -1, readUntilChar, timeoutMs); }
+        size_t sendUntil (Print& to, const int readUntilChar, const oneShotMs::timeType timeoutMs = oneShotMs::neverExpires) { return sendUntil(&to, readUntilChar, timeoutMs); }
+
+        // transfers data until requested size or timeout
+        // returns number of transfered bytes
+        size_t sendSize (Print* to, const ssize_t maxLen, const oneShotMs::timeType timeoutMs = oneShotMs::neverExpires) { return sendGeneric(to, maxLen, -1, timeoutMs); }
+        size_t sendSize (Print& to, const ssize_t maxLen, const oneShotMs::timeType timeoutMs = oneShotMs::neverExpires) { return sendSize(&to, maxLen, timeoutMs); }
+
+        // remaining size (-1 by default = unknown)
+        virtual ssize_t streamRemaining () { return -1; }
+
+        enum class Report
+        {
+            Success = 0,
+            TimedOut,
+            ReadError,
+            WriteError,
+            ShortOperation,
+        };
+
+        Report getLastSendReport () const { return _sendReport; }
+
+    protected:
+        size_t sendGeneric (Print* to,
+                            const ssize_t len = -1,
+                            const int readUntilChar = -1,
+                            oneShotMs::timeType timeoutMs = oneShotMs::neverExpires /* neverExpires=>getTimeout() */);
+
+        size_t SendGenericPeekBuffer(Print* to, const ssize_t len, const int readUntilChar, const oneShotMs::timeType timeoutMs);
+        size_t SendGenericRegularUntil(Print* to, const ssize_t len, const int readUntilChar, const oneShotMs::timeType timeoutMs);
+        size_t SendGenericRegular(Print* to, const ssize_t len, const oneShotMs::timeType timeoutMs);
+
+        void setReport (Report report) { _sendReport = report; }
+
+    private:
+
+        Report _sendReport = Report::Success;
+
+    //////////////////// end of extensions
+
     protected:
-        long parseInt(char skipChar); // as above but the given skipChar is ignored
-        // as above but the given skipChar is ignored
+        long parseInt(char skipChar); // as parseInt() but the given skipChar is ignored
         // this allows format characters (typically commas) in values to be ignored
 
-        float parseFloat(char skipChar);  // as above but the given skipChar is ignored
+        float parseFloat(char skipChar);  // as parseFloat() but the given skipChar is ignored
 };
 
 #endif