# Module `lib.server.stream`

These classes are used to manage asynchronous stream.

Expand source code

    # Distributed under Pycameresp License
    # Copyright (c) 2023 Remi BERTHOLET
    """ These classes are used to manage asynchronous stream. """
    import io
    import tools.filesystem

    class Stream:
            """ Class stream """
            # trace = open("stream.txt","wb")
            trace = None
            def __init__(self, reader, writer):
                    self.reader = reader
                    self.writer = writer
                    self.buffer = b""
                    if tools.filesystem.ismicropython():
                            self.close      = self.close_mcp
                            self.awrite     = self.awrite_mcp
                            self.is_closing = self.is_closing_mcp
                    else:
                            self.close      = self.close_pc
                            self.awrite     = self.awrite_pc
                            self.is_closing = self.is_closing_pc

            async def readline(self):
                    """ The firt time, this method completely reads the stream.
                    It then returns only the requested row. This significantly increases read performance. """
                    pos = self.buffer.find(b"\r\n")
                    if pos == -1:
                            self.buffer += await self.reader.read(1440)
                            pos = self.buffer.find(b"\r\n")
                    if pos == -1:
                            result = self.buffer
                            self.buffer = b""
                    else:
                            result = self.buffer[:pos+2]
                            self.buffer = self.buffer[pos+2:]
                    return result

            async def read(self, length):
                    """ Read data from the stream """
                    if len(self.buffer) < length:
                            data = await self.reader.read(length - len(self.buffer))
                            self.buffer += data

                    data = self.buffer[:length]
                    self.buffer = self.buffer[length:]
                    if Stream.trace:
                            Stream.trace.write(b"\n# read\n")
                            Stream.trace.write(data)
                            Stream.trace.flush()
                    return data

            async def write(self, data):
                    """ Write data in the stream """
                    result = await self.awrite(data)
                    if self.is_closing():
                            raise OSError(104,"Closed connection")
                    if Stream.trace:
                            Stream.trace.write(b"\n# write\n")
                            Stream.trace.write(data)
                            Stream.trace.flush()
                    if result is None:
                            result = len(data)
                    else:
                            result = -1
                    return result

            async def awrite_mcp(self, data):
                    """ Awrite micropython """
                    return await self.writer.awrite(data)

            async def awrite_pc(self, data):
                    """ Awrite micropython """
                    return self.writer.write(data)

            async def close_mcp(self):
                    """ Close the stream """
                    await self.writer.aclose()

            async def close_pc(self):
                    """ Close the stream """
                    self.writer.close()

            def is_closing_pc(self):
                    """ Check if it closed """
                    return self.writer.is_closing()

            def is_closing_mcp(self):
                    """ Check if it closed """
                    return False

    class Socket:
            """ Class stream which wrap socket """
            def __init__(self, socket):
                    """ Constructor """
                    self.socket = socket

            def read(self):
                    """ Read data from the stream """
                    data = self.socket.readlines()
                    return data

            def write(self, data):
                    """ Write data in the stream """
                    length = self.socket.sendall(data)
                    if length is None:
                            length = len(data)
                    return length

            def close(self):
                    """ Close the stream """
                    self.socket.close()

    class Bytesio:
            """ Class stream which wrap BytesIO """
            def __init__(self):
                    """ Constructor """
                    self.streamio = io.BytesIO()

            async def read(self):
                    """ Read data from the stream """
                    return self.streamio.read()

            async def write(self, data):
                    """ Write data in the stream """
                    return self.streamio.write(data)

            async def close(self):
                    """ Close the stream """
                    self.streamio.close()

    class Bufferedio:
            """ Bufferized bytes io stream """
            memorysize = [None]
            """ Class used to buffered stream write """
            def __init__(self, streamio, part=1440*20):
                    """ Constructor """
                    self.buffered = io.BytesIO()

                    if Bufferedio.is_enough_memory():
                            self.part = part
                    else:
                            self.part = None

                    self.streamio = streamio

            @staticmethod
            def is_enough_memory():
                    """ Indicate if it has enough memory """
                    if Bufferedio.memorysize[0] is None:
                            import gc
                            try:
                                    # pylint: disable=no-member
                                    Bufferedio.memorysize[0]  = gc.mem_free()
                            except:
                                    Bufferedio.memorysize[0]  = 256*1024

                    if Bufferedio.memorysize[0] < 200*1024:
                            return False
                    return True

            async def read(self):
                    """ Read data from the stream """
                    return self.streamio.read()

            async def write(self, data):
                    """ Write data in the stream """
                    if self.part is None:
                            result = len(data)
                            await self.streamio.write(data)
                    else:
                            result = self.buffered.write(data)
                            if self.buffered.tell() > self.part:
                                    await self.streamio.write(self.buffered.getvalue())
                                    self.buffered = io.BytesIO()
                    return result

            async def close(self):
                    """ Close the stream """
                    try:
                            if self.buffered.tell() > 0:
                                    await self.streamio.write(self.buffered.getvalue())
                    finally:
                            self.buffered.close()

## Classes

`class Bufferedio (streamio, part=28800)`

  Bufferized bytes io stream

  Constructor

  Expand source code

      class Bufferedio:
              """ Bufferized bytes io stream """
              memorysize = [None]
              """ Class used to buffered stream write """
              def __init__(self, streamio, part=1440*20):
                      """ Constructor """
                      self.buffered = io.BytesIO()

                      if Bufferedio.is_enough_memory():
                              self.part = part
                      else:
                              self.part = None

                      self.streamio = streamio

              @staticmethod
              def is_enough_memory():
                      """ Indicate if it has enough memory """
                      if Bufferedio.memorysize[0] is None:
                              import gc
                              try:
                                      # pylint: disable=no-member
                                      Bufferedio.memorysize[0]  = gc.mem_free()
                              except:
                                      Bufferedio.memorysize[0]  = 256*1024

                      if Bufferedio.memorysize[0] < 200*1024:
                              return False
                      return True

              async def read(self):
                      """ Read data from the stream """
                      return self.streamio.read()

              async def write(self, data):
                      """ Write data in the stream """
                      if self.part is None:
                              result = len(data)
                              await self.streamio.write(data)
                      else:
                              result = self.buffered.write(data)
                              if self.buffered.tell() > self.part:
                                      await self.streamio.write(self.buffered.getvalue())
                                      self.buffered = io.BytesIO()
                      return result

              async def close(self):
                      """ Close the stream """
                      try:
                              if self.buffered.tell() > 0:
                                      await self.streamio.write(self.buffered.getvalue())
                      finally:
                              self.buffered.close()

  ### Class variables

  `var memorysize`

    Class used to buffered stream write

  ### Static methods

  `def is_enough_memory()`

    Indicate if it has enough memory

    Expand source code

        @staticmethod
        def is_enough_memory():
                """ Indicate if it has enough memory """
                if Bufferedio.memorysize[0] is None:
                        import gc
                        try:
                                # pylint: disable=no-member
                                Bufferedio.memorysize[0]  = gc.mem_free()
                        except:
                                Bufferedio.memorysize[0]  = 256*1024

                if Bufferedio.memorysize[0] < 200*1024:
                        return False
                return True

  ### Methods

  `async def close(self)`

    Close the stream

    Expand source code

        async def close(self):
                """ Close the stream """
                try:
                        if self.buffered.tell() > 0:
                                await self.streamio.write(self.buffered.getvalue())
                finally:
                        self.buffered.close()

  `async def read(self)`

    Read data from the stream

    Expand source code

        async def read(self):
                """ Read data from the stream """
                return self.streamio.read()

  `async def write(self, data)`

    Write data in the stream

    Expand source code

        async def write(self, data):
                """ Write data in the stream """
                if self.part is None:
                        result = len(data)
                        await self.streamio.write(data)
                else:
                        result = self.buffered.write(data)
                        if self.buffered.tell() > self.part:
                                await self.streamio.write(self.buffered.getvalue())
                                self.buffered = io.BytesIO()
                return result

`class Bytesio`

  Class stream which wrap BytesIO

  Constructor

  Expand source code

      class Bytesio:
              """ Class stream which wrap BytesIO """
              def __init__(self):
                      """ Constructor """
                      self.streamio = io.BytesIO()

              async def read(self):
                      """ Read data from the stream """
                      return self.streamio.read()

              async def write(self, data):
                      """ Write data in the stream """
                      return self.streamio.write(data)

              async def close(self):
                      """ Close the stream """
                      self.streamio.close()

  ### Methods

  `async def close(self)`

    Close the stream

    Expand source code

        async def close(self):
                """ Close the stream """
                self.streamio.close()

  `async def read(self)`

    Read data from the stream

    Expand source code

        async def read(self):
                """ Read data from the stream """
                return self.streamio.read()

  `async def write(self, data)`

    Write data in the stream

    Expand source code

        async def write(self, data):
                """ Write data in the stream """
                return self.streamio.write(data)

`class Socket (socket)`

  Class stream which wrap socket

  Constructor

  Expand source code

      class Socket:
              """ Class stream which wrap socket """
              def __init__(self, socket):
                      """ Constructor """
                      self.socket = socket

              def read(self):
                      """ Read data from the stream """
                      data = self.socket.readlines()
                      return data

              def write(self, data):
                      """ Write data in the stream """
                      length = self.socket.sendall(data)
                      if length is None:
                              length = len(data)
                      return length

              def close(self):
                      """ Close the stream """
                      self.socket.close()

  ### Methods

  `def close(self)`

    Close the stream

    Expand source code

        def close(self):
                """ Close the stream """
                self.socket.close()

  `def read(self)`

    Read data from the stream

    Expand source code

        def read(self):
                """ Read data from the stream """
                data = self.socket.readlines()
                return data

  `def write(self, data)`

    Write data in the stream

    Expand source code

        def write(self, data):
                """ Write data in the stream """
                length = self.socket.sendall(data)
                if length is None:
                        length = len(data)
                return length

`class Stream (reader, writer)`

  Class stream

  Expand source code

      class Stream:
              """ Class stream """
              # trace = open("stream.txt","wb")
              trace = None
              def __init__(self, reader, writer):
                      self.reader = reader
                      self.writer = writer
                      self.buffer = b""
                      if tools.filesystem.ismicropython():
                              self.close      = self.close_mcp
                              self.awrite     = self.awrite_mcp
                              self.is_closing = self.is_closing_mcp
                      else:
                              self.close      = self.close_pc
                              self.awrite     = self.awrite_pc
                              self.is_closing = self.is_closing_pc

              async def readline(self):
                      """ The firt time, this method completely reads the stream.
                      It then returns only the requested row. This significantly increases read performance. """
                      pos = self.buffer.find(b"\r\n")
                      if pos == -1:
                              self.buffer += await self.reader.read(1440)
                              pos = self.buffer.find(b"\r\n")
                      if pos == -1:
                              result = self.buffer
                              self.buffer = b""
                      else:
                              result = self.buffer[:pos+2]
                              self.buffer = self.buffer[pos+2:]
                      return result

              async def read(self, length):
                      """ Read data from the stream """
                      if len(self.buffer) < length:
                              data = await self.reader.read(length - len(self.buffer))
                              self.buffer += data

                      data = self.buffer[:length]
                      self.buffer = self.buffer[length:]
                      if Stream.trace:
                              Stream.trace.write(b"\n# read\n")
                              Stream.trace.write(data)
                              Stream.trace.flush()
                      return data

              async def write(self, data):
                      """ Write data in the stream """
                      result = await self.awrite(data)
                      if self.is_closing():
                              raise OSError(104,"Closed connection")
                      if Stream.trace:
                              Stream.trace.write(b"\n# write\n")
                              Stream.trace.write(data)
                              Stream.trace.flush()
                      if result is None:
                              result = len(data)
                      else:
                              result = -1
                      return result

              async def awrite_mcp(self, data):
                      """ Awrite micropython """
                      return await self.writer.awrite(data)

              async def awrite_pc(self, data):
                      """ Awrite micropython """
                      return self.writer.write(data)

              async def close_mcp(self):
                      """ Close the stream """
                      await self.writer.aclose()

              async def close_pc(self):
                      """ Close the stream """
                      self.writer.close()

              def is_closing_pc(self):
                      """ Check if it closed """
                      return self.writer.is_closing()

              def is_closing_mcp(self):
                      """ Check if it closed """
                      return False

  ### Class variables

  `var trace`

  ### Methods

  `async def awrite_mcp(self, data)`

    Awrite micropython

    Expand source code

        async def awrite_mcp(self, data):
                """ Awrite micropython """
                return await self.writer.awrite(data)

  `async def awrite_pc(self, data)`

    Awrite micropython

    Expand source code

        async def awrite_pc(self, data):
                """ Awrite micropython """
                return self.writer.write(data)

  `async def close_mcp(self)`

    Close the stream

    Expand source code

        async def close_mcp(self):
                """ Close the stream """
                await self.writer.aclose()

  `async def close_pc(self)`

    Close the stream

    Expand source code

        async def close_pc(self):
                """ Close the stream """
                self.writer.close()

  `def is_closing_mcp(self)`

    Check if it closed

    Expand source code

        def is_closing_mcp(self):
                """ Check if it closed """
                return False

  `def is_closing_pc(self)`

    Check if it closed

    Expand source code

        def is_closing_pc(self):
                """ Check if it closed """
                return self.writer.is_closing()

  `async def read(self, length)`

    Read data from the stream

    Expand source code

        async def read(self, length):
                """ Read data from the stream """
                if len(self.buffer) < length:
                        data = await self.reader.read(length - len(self.buffer))
                        self.buffer += data

                data = self.buffer[:length]
                self.buffer = self.buffer[length:]
                if Stream.trace:
                        Stream.trace.write(b"\n# read\n")
                        Stream.trace.write(data)
                        Stream.trace.flush()
                return data

  `async def readline(self)`

    The firt time, this method completely reads the stream. It then returns only the requested row. This significantly increases read performance.

    Expand source code

        async def readline(self):
                """ The firt time, this method completely reads the stream.
                It then returns only the requested row. This significantly increases read performance. """
                pos = self.buffer.find(b"\r\n")
                if pos == -1:
                        self.buffer += await self.reader.read(1440)
                        pos = self.buffer.find(b"\r\n")
                if pos == -1:
                        result = self.buffer
                        self.buffer = b""
                else:
                        result = self.buffer[:pos+2]
                        self.buffer = self.buffer[pos+2:]
                return result

  `async def write(self, data)`

    Write data in the stream

    Expand source code

        async def write(self, data):
                """ Write data in the stream """
                result = await self.awrite(data)
                if self.is_closing():
                        raise OSError(104,"Closed connection")
                if Stream.trace:
                        Stream.trace.write(b"\n# write\n")
                        Stream.trace.write(data)
                        Stream.trace.flush()
                if result is None:
                        result = len(data)
                else:
                        result = -1
                return result