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
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
|
#
# This file is part of pysnmp software.
#
# Copyright (c) 2005-2020, Ilya Etingof <etingof@gmail.com>
# License: https://www.pysnmp.com/pysnmp/license.html
#
from typing import TYPE_CHECKING
from pysnmp.proto import error
from pysnmp.proto.mpmod import cache
if TYPE_CHECKING:
from pysnmp.entity.engine import SnmpEngine
class AbstractMessageProcessingModel:
"""Create a message processing model object."""
SNMP_MSG_SPEC = NotImplementedError
def __init__(self):
"""Create a message processing model object."""
self._snmpMsgSpec = self.SNMP_MSG_SPEC() # local copy
self._cache = cache.Cache()
def prepare_outgoing_message(
self,
snmpEngine: "SnmpEngine",
transportDomain,
transportAddress,
messageProcessingModel,
securityModel,
securityName,
securityLevel,
contextEngineId,
contextName,
pduVersion,
pdu,
expectResponse,
sendPduHandle,
):
"""Prepare SNMP message for dispatch."""
raise error.ProtocolError("method not implemented")
def prepare_response_message(
self,
snmpEngine: "SnmpEngine",
messageProcessingModel,
securityModel,
securityName,
securityLevel,
contextEngineId,
contextName,
pduVersion,
pdu,
maxSizeResponseScopedPDU,
stateReference,
statusInformation,
):
"""Prepare SNMP message for response."""
raise error.ProtocolError("method not implemented")
def prepare_data_elements(
self, snmpEngine: "SnmpEngine", transportDomain, transportAddress, wholeMsg
):
"""Prepare SNMP message data elements."""
raise error.ProtocolError("method not implemented")
def release_state_information(self, sendPduHandle):
"""Release state information."""
try:
self._cache.pop_by_send_pdu_handle(sendPduHandle)
except error.ProtocolError:
pass # XXX maybe these should all follow some scheme?
def receive_timer_tick(self, snmpEngine: "SnmpEngine", timeNow):
"""Process a timer tick."""
self._cache.expire_caches()
|
