spacepaste

Paste details

reply | raw

  1.  
  2. import asyncio
  3. from itertools import count
  5. async def position_stream():
  6. try:
  7. for position in count():
  8. await asyncio.sleep(1)
  9. yield position
  10. finally:
  11. print("position stream cancelled")
  13. async def altitude_stream():
  14. try:
  15. for altitude in count():
  16. await asyncio.sleep(1)
  17. yield altitude
  18. finally:
  19. print("altitude stream cancelled")
  21. async def print_position():
  22. async for position in position_stream():
  23. print(f"position: {position}")
  25. if position == 2:
  26. cleanup()
  28. async def print_altitude():
  29. async for altitude in altitude_stream():
  30. print(f"altitude: {altitude}")
  32. def cleanup():
  33. loop = asyncio.get_event_loop()
  34. asyncio.ensure_future(loop.shutdown_asyncgens())
  35. loop.close()
  37. if __name__ == '__main__':
  38. try:
  39. asyncio.ensure_future(print_position())
  40. asyncio.ensure_future(print_altitude())
  41. asyncio.get_event_loop().run_forever()
  42. except KeyboardInterrupt:
  43. pass
  44.  
Powered by spacepaste. Paste removal.