Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def send_recv_packets(packets, delay=0.0001) -> (list, float, float):
"""Starts a listener, send packets, receives packets
returns a tuple: ([(ts, export), ...], time_started_sending, time_stopped_sending)
"""
listener = ThreadedNetFlowListener(*CONNECTION)
tstart = time.time()
emit_packets(packets, delay=delay)
time.sleep(0.5) # Allow packets to be sent and recieved
tend = time.time()
listener.start()
pkts = []
while True:
try:
pkts.append(listener.get(timeout=0.5))
except queue.Empty:
break
listener.stop()
listener.join()
return pkts, tstart, tend
def send_recv_packets(packets, delay=0.0001, store_packets=-1) -> (list, float, float):
"""Starts a listener, send packets, receives packets
returns a tuple: ([(ts, export), ...], time_started_sending, time_stopped_sending)
"""
listener = ThreadedNetFlowListener(*CONNECTION)
tstart = time.time()
emit_packets(packets, delay=delay)
time.sleep(0.5) # Allow packets to be sent and recieved
tend = time.time()
listener.start()
pkts = []
to_pad = 0
while True:
try:
packet = listener.get(timeout=0.5)
if -1 == store_packets or store_packets > 0:
# Case where a programm yields from the queue and stores all packets.
pkts.append(packet)
if store_packets != -1 and len(pkts) > store_packets:
to_pad += len(pkts) # Hack for testing
def get_export_packets(host: str, port: int) -> ParsedPacket:
"""A threaded generator that will yield ExportPacket objects until it is killed
"""
listener = ThreadedNetFlowListener(host, port)
listener.start()
try:
while True:
yield listener.get()
finally:
listener.stop()
listener.join()