One-to-Many and Many-to-One Stream gateway

[nobodyinperson]

Hi Benoit,

In one of my projects I let my µCs provide a command-line interface (based on the Shell library). At some point it turned out to be very useful to provide this cli via different channels, i.e. the USB Serial, maybe a different Serial1 and even via MQTT. The best idea I could come up with to facilitate that was to implement a one-to-many and many-to-one Stream gateway or valve - I called it MultiStream for lack of a better term. This Stream subclass basically write()s and read()s everything to and from multiple other registered Streams. In my case it makes it possible to have a command-line interface via both MQTT and USB Serial simulataneously.

I think something like this would make a wonderful addition to your StreamUtils library.

Here is my (probably awful) implementation:

#pragma once

/**
  @file MultiStream.hpp
  @brief Multiple streams in one
*/

#include <Arduino.h>

/**
 *  @brief Interface multiple streams from a single one
 */
template<size_t nSlots>
class MultiStream : public Stream
{
public:
  MultiStream()
    : _slots{ nullptr } {};

  //! Add a stream
  bool add(Stream& stream)
  {
    uint8_t index;
    const bool anyFree = nextFreeSlot(index);
    if (anyFree)
      this->_slots[index] = &stream;
    return anyFree;
  }

  //! Return the first available byte of any stream in order
  int read()
  {
    int ret = -1;
    for (size_t i = 0; i < nSlots; i++) {
      Stream* stream = this->_slots[i];
      if (not stream)
        continue;
#if ESP8266
      // don't know why (yet) but only this works properly together with our
      // StreamGateway
      ret = stream->read();
      if (ret >= 0)
        return ret;
#else  // #if ESP8266
      char c;
      // don't use stream->read() as it doesn't honor timeout (needed for
      // EEPROM write from cli)
      const size_t n = stream->readBytes(&c, 1);
      if (n > 0)
        return c;
#endif // #if ESP8266
    }
    return ret;
  }

  //! fill given buffer from all streams in order until full or timeout
  size_t readBytes(char* buffer, size_t length)
  {
    size_t count = 0;
    for (size_t i = 0; i < nSlots; i++) {
      char* buf = buffer + count;
      const size_t len = length - count;
      if (not len)
        return count;
      Stream* stream = this->_slots[i];
      if (not stream)
        continue;
      count += stream->readBytes(buf, len);
    }
    return count;
  }

  //! Peek on first stream available in order
  int peek()
  {
    int ret = -1;
    for (size_t i = 0; i < nSlots; i++) {
      Stream* stream = this->_slots[i];
      if (not stream)
        continue;
      ret = stream->peek();
      if (ret >= 0)
        return ret;
    }
    return ret;
  }

  //! Total available bytes on all streams
  int available()
  {
    int total = 0;
    for (size_t i = 0; i < nSlots; i++) {
      Stream* stream = this->_slots[i];
      if (not stream)
        continue;
      const int av = stream->available();
      if (av > 0)
        total += av;
    }
    return total;
  }

  //! Write a byte to all streams
  size_t write(uint8_t b)
  {
    int total = 0;
    for (size_t i = 0; i < nSlots; i++) {
      Stream* stream = this->_slots[i];
      if (not stream)
        continue;
      const int written = stream->write(b);
      if (written > total)
        total = written;
    }
    return total;
  }

  template<typename TYPE>
  size_t write(TYPE b)
  {
    int total = 0;
    for (size_t i = 0; i < nSlots; i++) {
      Stream* stream = this->_slots[i];
      if (not stream)
        continue;
      const int written = stream->write(b);
      if (written > total)
        total = written;
    }
    return total;
  }

#if !defined(ESP8266)
  // For some reason, the ArduinoEPS8266 framework's Stream class doesn't
  // implement getTimeout()...
  //! Determine the maximum timeout of all available streams
  unsigned long getTimeout()
  {
    unsigned long timeout = 0;
    for (size_t i = 0; i < nSlots; i++) {
      Stream* stream = this->_slots[i];
      if (not stream)
        continue;
      const unsigned long streamTimeout = stream->getTimeout();
      if (streamTimeout > timeout)
        timeout = streamTimeout;
    }
    return timeout;
  }
#endif // #if ! defined(ESP8266)

  //! Set timeout of all available streams
  void setTimeout(const unsigned long timeout)
  {
    for (size_t i = 0; i < nSlots; i++) {
      Stream* stream = this->_slots[i];
      if (not stream)
        continue;
      stream->setTimeout(timeout);
    }
  }

  void flush()
  {
    for (size_t i = 0; i < nSlots; i++) {
      Stream* stream = this->_slots[i];
      if (not stream)
        continue;
      stream->flush();
    }
  }

protected:
  Stream* _slots[nSlots] = { nullptr };

  /**
   * @brief determine the next free slot index
   * @param index the next free index
   * @return whether a free slot was available
   */
  bool nextFreeSlot(uint8_t& index)
  {
    for (size_t i = 0; i < nSlots; i++) {
      if (not this->_slots[i]) {
        index = i;
        return true;
      }
    }
    return false;
  };
};

[bblanchon]

Hi @nobodyinperson,

Thank you very much for suggesting this class. However, I'm not sure I completely get the use case. From what I understood, this class allows exposing a shell to Serial and Serial1, but when a user interacts with Serial, the output shows in Serial1 as well. This seems really weird, doesn't it?

Best regards, Benoit

[nobodyinperson]

...but when a user interacts with Serial, the output shows in Serial1 as well

Exactly.

This seems really weird, doesn't it?

If you put it that way... probably yes :-)

A possibly cleaner implementation would be to have a command-line library that allows attaching multiple Streams independently (There aren't any. Every Arduino Shell library out there has flaws, the Shell library that I use also is very limited). But then each attached Stream would need to have its own internal input buffer (to store argv) which would need to be allocated dynamically or statically via a template argument for example. Assuming that only ever one CLI interface is used at a time, having Serial interaction also show up in Serial1 shouldn't be so much of a problem an saves memory.

Maybe what this MultiStream class does (namely gathering input from many other Streams and writing the same output to all Streams) can even be achived with a proper setup of ReadLoggingStream and WriteLoggingStream - though probably not in one object - but I couldn't really figure out how. The closes thing seems to be the LoggingStream, but this doesn't read from the other Stream, just writes, right?

[bblanchon]

If I were you, I would instantiate one shell per stream (which is exactly what happens when you open several SSH connections to the same server).

If one shell per stream is too consuming, I would use a Stream pointer that I would point to the active stream.

[nobodyinperson]

Yes, but unfortunately the Shell library really only allows one global shell (It's plain C, so no classes or at least they chose to implmement it this way). I also tried others but it turned out to be the most stable CLI library (no dynamic allocation all over the place, thus no heap fragmentation) among them. Also, no other library seems to directly use Streams as input, unfortunately. Maybe it's time to write an own library...

[bblanchon]

This library is written in pure C in order to allow it to work in other microcontrollers/compilers that do not support C++.

I'm curious to know what platform they are talking about.

Anyway, why don't you keep a global Stream* that you use in shell_reader() and shell_writer()?

[nobodyinperson]

I'm curious to know what platform they are talking about.

Maybe they don't like the Arduino framework that much, rather plain C AVR...

Anyway, why don't you keep a global Stream* that you use in shell_reader() and shell_writer()?

That's exactly what I'm doing. But this global Stream is a MultiStream as implemented above. And to this MultiStream I attached both Serial and a Stream class that uses MQTT in the background to recieve user input and transmit shell output via WIFI. Like this I have a command-line interface both via Serial and MQTT (together with ESP8266 OTA flashing one doesn't need a USB cable anymore then).

It works fine for me. But probably (as with all my suggestions/questions for your libraries :stuck_out_tongue_winking_eye:), this is too much of a corner case to be useful for others...

[bblanchon]

I'm afraid you're right: it's not a generic reusable solution.

[nobodyinperson]

Yes, a proper solution would be to fix the Shell libraries or write a proper one, not to write workaround code for existing shortcomings.

Thanks anyway!