blob: 73ffc70c84cc54bf37bde3adae987321a5cad8e7 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
import collections
import typing # noqa
import blinker
from mitmproxy import command
from mitmproxy.log import LogEntry
class EventStore:
def __init__(self, size=10000):
self.data = collections.deque(maxlen=size) # type: typing.Deque[LogEntry]
self.sig_add = blinker.Signal()
self.sig_refresh = blinker.Signal()
@property
def size(self) -> int:
return self.data.maxlen
def log(self, entry: LogEntry) -> None:
self.data.append(entry)
self.sig_add.send(self, entry=entry)
@command.command("eventstore.clear")
def clear(self) -> None:
"""
Clear the event log.
"""
self.data.clear()
self.sig_refresh.send(self)
|