def test_listening()

in asfpy/pubsub.py [0:0]

• 24 lines of code
• 1 McCabe index (conditional complexity)

def test_listening():
    logging.basicConfig(level=logging.DEBUG)

    start = time.time()
    count = 0

    # We'll say "now" is when we believe the connection to be alive.
    last_traffic = time.time()

    async def report_stats():
        while True:
            # NOTE: do not set this lower than 60, or at startup,
            # DURATION will be zero, creating a div-by-zero error.
            await asyncio.sleep(70)

            duration = int((time.time() - start) / 60)
            alive_since = int(time.time() - last_traffic)
            print(f'[{duration}m] {count} events.  {count/duration:.1f}/min'
                  f'  traffic: {alive_since}s ago')

    async def print_events():
        async for payload in listen('https://pubsub.apache.org:2070/'):
            nonlocal last_traffic
            last_traffic = time.time()

            if 'stillalive' not in payload:
                print(f'PAYLOAD: [{payload.get("pubsub_path", "none")}]'
                      f'  KEYS: {sorted(payload.keys())}')
                nonlocal count
                count += 1

    async def run_test():
        await asyncio.gather(report_stats(), print_events())

    asyncio.run(run_test())