22

✏️ Editing: persistent_message.py

import queue
import time

from defence360agent.model.instance import db
from defence360agent.model.messages_to_send import MessageToSend

class PersistentMessagesQueue:
    """
    The queue to store messages sent to the server if it is unavailable.
    - stores more recent data; if a limit is exceeded,
       older messages are deleted.
    - no duplicate messages are sent

NOTE: it is worth remembering that when writing a large number of messages,
          the amount of memory used may increase by the size of the sqlite
          cache (this may not be immediately obvious).
          https://www.sqlite.org/pragma.html#pragma_cache_size
    """

def __init__(self, buffer_limit=20, storage_limit=1000, model=None):
        self._buffer_limit = buffer_limit
        self._storage_limit = storage_limit
        self._buffer = []  # [(timestamp, message),...]
        self._model = model or MessageToSend

def push_buffer_to_storage(self) -> None:
        if self._buffer:
            with db.atomic():
                # buffer may contain older messages than db,
                # so remove oldest items after insert
                self._model.insert_many(self._buffer)
                need_to_remove = self.storage_size - self._storage_limit
                if need_to_remove > 0:
                    # keep only the most recent messages
                    self._model.delete_old(need_to_remove)
                self._buffer = []

def pop_all(self) -> list:
        items = []
        with db.atomic():
            items += list(
                self._model.select(
                    self._model.timestamp, self._model.message
                ).tuples()
            )
            self._model.delete().execute()
        items += self._buffer
        self._buffer = []
        return sorted(items)  # older first

def empty(self) -> bool:
        return self.qsize() == 0

def qsize(self) -> int:
        return self.storage_size + len(self._buffer)

@property
    def buffer_size(self) -> int:
        return len(self._buffer)

@property
    def storage_size(self) -> int:
        return self._model.select().count()

def put(self, message: bytes, timestamp=None):
        if timestamp is None:
            timestamp = time.time()
        self._buffer.append((timestamp, message))
        if self.buffer_size >= self._buffer_limit:
            self.push_buffer_to_storage()

def put_many(self, messages: list[tuple[float, bytes]]) -> None:
        self._buffer.extend(messages)
        if self.buffer_size >= self._buffer_limit:
            self.push_buffer_to_storage()

def get(self) -> bytes:
        if not self._buffer and self.storage_size != 0:
            # reduce the number of database calls
            items = self._model.get_oldest(limit=self._buffer_limit)
            # not save id value
            self._buffer += [item[1:] for item in items.tuples()]
            self._model.delete_in(items)

if self._buffer:
            _, message = self._buffer.pop(0)
            return message
        raise queue.Empty()

MR SKB