Apply a pattern to a routing key.
Source code in faststream/rabbit/testing.py
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428 | def apply_pattern(pattern: str, current: str) -> bool:
"""Apply a pattern to a routing key."""
pattern_queue = iter(pattern.split("."))
current_queue = iter(current.split("."))
pattern_symb = next(pattern_queue, None)
while pattern_symb:
if (next_symb := next(current_queue, None)) is None:
return False
if pattern_symb == "#":
next_pattern = next(pattern_queue, None)
if next_pattern is None:
return True
if (next_symb := next(current_queue, None)) is None:
return False
while next_pattern == "*":
next_pattern = next(pattern_queue, None)
if (next_symb := next(current_queue, None)) is None:
return False
while next_symb != next_pattern:
if (next_symb := next(current_queue, None)) is None:
return False
pattern_symb = next(pattern_queue, None)
elif pattern_symb in {"*", next_symb}:
pattern_symb = next(pattern_queue, None)
else:
return False
return next(current_queue, None) is None
|