# 需要导入模块: from scapy import plist [as 别名]

# 或者: from scapy.plist import SndRcvList [as 别名]

def __sr_loop(srfunc, pkts, prn=lambda x: x[1].summary(),

prnfail=lambda x: x.summary(),

inter=1, timeout=None, count=None, verbose=None, store=1,

*args, **kargs):

n = 0

r = 0

ct = conf.color_theme

if verbose is None:

verbose = conf.verb

parity = 0

ans = []

unans = []

if timeout is None:

timeout = min(2 * inter, 5)

try:

while True:

parity ^= 1

col = [ct.even, ct.odd][parity]

if count is not None:

if count == 0:

break

count -= 1

start = time.time()

if verbose > 1:

print("\rsend...\r", end=' ')

res = srfunc(pkts, timeout=timeout, verbose=0, chainCC=True, *args, **kargs) # noqa: E501

n += len(res[0]) + len(res[1])

r += len(res[0])

if verbose > 1 and prn and len(res[0]) > 0:

msg = "RECV %i:" % len(res[0])

print("\r" + ct.success(msg), end=' ')

for p in res[0]:

print(col(prn(p)))

print(" " * len(msg), end=' ')

if verbose > 1 and prnfail and len(res[1]) > 0:

msg = "fail %i:" % len(res[1])

print("\r" + ct.fail(msg), end=' ')

for p in res[1]:

print(col(prnfail(p)))

print(" " * len(msg), end=' ')

if verbose > 1 and not (prn or prnfail):

print("recv:%i fail:%i" % tuple(map(len, res[:2])))

if store:

ans += res[0]

unans += res[1]

end = time.time()

if end - start < inter:

time.sleep(inter + start - end)

except KeyboardInterrupt:

pass

if verbose and n > 0:

print(ct.normal("\nSent %i packets, received %i packets. %3.1f%% hits." % (n, r, 100.0 * r / n))) # noqa: E501

return SndRcvList(ans), PacketList(unans)