(self,
pks, # type: SuperSocket
pkt, # type: _PacketIterable
timeout=None, # type: Optional[int]
inter=0, # type: int
verbose=None, # type: Optional[int]
chainCC=False, # type: bool
retry=0, # type: int
multi=False, # type: bool
first=False, # type: bool
rcv_pks=None, # type: Optional[SuperSocket]
prebuild=False, # type: bool
_flood=None, # type: Optional[_FloodGenerator]
threaded=True, # type: bool
session=None, # type: Optional[_GlobSessionType]
chainEX=False, # type: bool
stop_filter=None # type: Optional[Callable[[Packet], bool]]
)
| 118 | to avoid races that could result in negative latency. We aren't Stadia |
| 119 | """ |
| 120 | def __init__(self, |
| 121 | pks, # type: SuperSocket |
| 122 | pkt, # type: _PacketIterable |
| 123 | timeout=None, # type: Optional[int] |
| 124 | inter=0, # type: int |
| 125 | verbose=None, # type: Optional[int] |
| 126 | chainCC=False, # type: bool |
| 127 | retry=0, # type: int |
| 128 | multi=False, # type: bool |
| 129 | first=False, # type: bool |
| 130 | rcv_pks=None, # type: Optional[SuperSocket] |
| 131 | prebuild=False, # type: bool |
| 132 | _flood=None, # type: Optional[_FloodGenerator] |
| 133 | threaded=True, # type: bool |
| 134 | session=None, # type: Optional[_GlobSessionType] |
| 135 | chainEX=False, # type: bool |
| 136 | stop_filter=None # type: Optional[Callable[[Packet], bool]] |
| 137 | ): |
| 138 | # type: (...) -> None |
| 139 | # Instantiate all arguments |
| 140 | if verbose is None: |
| 141 | verbose = conf.verb |
| 142 | if conf.debug_match: |
| 143 | debug.recv = PacketList([], "Received") |
| 144 | debug.sent = PacketList([], "Sent") |
| 145 | debug.match = SndRcvList([], "Matched") |
| 146 | self.nbrecv = 0 |
| 147 | self.ans = [] # type: List[QueryAnswer] |
| 148 | self.pks = pks |
| 149 | self.rcv_pks = rcv_pks or pks |
| 150 | self.inter = inter |
| 151 | self.verbose = verbose |
| 152 | self.chainCC = chainCC |
| 153 | self.multi = multi |
| 154 | self.timeout = timeout |
| 155 | self.first = first |
| 156 | self.session = session |
| 157 | self.chainEX = chainEX |
| 158 | self.stop_filter = stop_filter |
| 159 | self._send_done = False |
| 160 | self.notans = 0 |
| 161 | self.noans = 0 |
| 162 | self._flood = _flood |
| 163 | self.threaded = threaded |
| 164 | self.breakout = Event() |
| 165 | # Instantiate packet holders |
| 166 | if prebuild and not self._flood: |
| 167 | self.tobesent = list(pkt) # type: _PacketIterable |
| 168 | else: |
| 169 | self.tobesent = pkt |
| 170 | |
| 171 | if retry < 0: |
| 172 | autostop = retry = -retry |
| 173 | else: |
| 174 | autostop = 0 |
| 175 | |
| 176 | if timeout is not None and timeout < 0: |
| 177 | self.timeout = None |
nothing calls this directly
no test coverage detected