# coding=utf-8
import datetime
import itertools
from typing import Optional, List
from aioxmpp import JID
from .message import Message
def _agent_in_msg(agent, msg):
to = msg.to.bare() if msg.to else None
sender = msg.sender.bare() if msg.sender else None
return to == agent or sender == agent
[docs]class TraceStore(object):
"""Stores and allows queries about events."""
def __init__(self, size: int):
self.size = size
self.store = []
[docs] def reset(self) -> None:
"""Resets the trace store"""
self.store = []
[docs] def append(self, event: Message, category: Optional[str] = None) -> None:
"""
Adds a new event to the trace store.
The event may hava a category
Args:
event (spade.message.Message): the event to be stored
category (str, optional): a category to classify the event (Default value = None)
"""
date = datetime.datetime.now()
self.store.insert(0, (date, event, category))
if len(self.store) > self.size:
del self.store[-1]
[docs] def len(self) -> int:
"""
Length of the store
Returns:
int: the size of the trace store
"""
return len(self.store)
[docs] def all(self, limit: Optional[int] = None) -> List[Message]:
"""
Returns all the events, until a limit if defined
Args:
limit (int, optional): the max length of the events to return (Default value = None)
Returns:
list: a list of events
"""
return self.store[:limit][::-1]
[docs] def received(self, limit: Optional[int] = None) -> List[Message]:
"""
Returns all the events that have been received (excluding sent events), until a limit if defined
Args:
limit (int, optional): the max length of the events to return (Default value = None)
Returns:
list: a list of received events
"""
return list(
itertools.islice(
(itertools.filterfalse(lambda x: x[1].sent, self.store)), limit
)
)[::-1]
[docs] def filter(
self,
limit: Optional[int] = None,
to: Optional[str] = None,
category: Optional[str] = None,
) -> List[Message]:
"""
Returns the events that match the filters
Args:
limit (int, optional): the max length of the events to return (Default value = None)
to (str, optional): only events that have been sent or received by 'to' (Default value = None)
category (str, optional): only events belonging to the category (Default value = None)
Returns:
list: a list of filtered events
"""
if category and not to:
msg_slice = itertools.islice(
(x for x in self.store if x[2] == category), limit
)
elif to and not category:
to = JID.fromstr(to)
msg_slice = itertools.islice(
(x for x in self.store if _agent_in_msg(to, x[1])), limit
)
elif to and category:
to = JID.fromstr(to)
msg_slice = itertools.islice(
(x for x in self.store if _agent_in_msg(to, x[1]) and x[2] == category),
limit,
)
else:
msg_slice = self.all(limit=limit)
return msg_slice
return list(msg_slice)[::-1]