first version

master
Ugo Finnendahl 5 years ago
parent 5b127c8eef
commit f93180bbc7
  1. 3
      README.md
  2. 60
      ewents/__init__.py
  3. 25
      examples/basic.py
  4. 35
      setup.py

@ -1,2 +1,5 @@
# ewents
## Installation
`pip install git+https://git.finnendahl.de/Yugon/ewents`

@ -0,0 +1,60 @@
import asyncio
import functools
global_signals = {}
async def emit(signal, payload=None):
for q in global_signals.get(signal, []):
if payload is not None:
await q.put({"payload": payload})
else:
await q.put({})
async def shutdown():
for key in global_signals:
for q in global_signals[key]:
await q.put(None)
def shutdown(loop=None):
if loop is None:
loop = asyncio.get_event_loop()
loop.run_until_complete(shutdown())
async def consumer(queue, callback, signal):
while True:
item = await queue.get()
if item is None:
global_signals[signal].remove(queue)
break
if "payload" in item:
await callback(item["payload"])
else:
await callback()
def listen_on(signal, fun=None, *, loop=None):
if loop is None:
loop = asyncio.get_event_loop()
def decorator(func, deco=True):
@functools.wraps(func)
def fun(*args, **kwargs):
return func(*args, **kwargs)
q = asyncio.Queue()
global_signals.setdefault(signal, []).append(q)
handle = loop.call_soon(
lambda: asyncio.ensure_future(consumer(q, func, signal))
)
if deco:
return fun
else:
return handle
if fun is not None:
return decorator(fun, False)
return decorator

@ -0,0 +1,25 @@
import ewents
import asyncio
@ewents.listen_on("eventA")
async def callbackA():
print("Listener A")
await ewents.emit("eventB")
@ewents.listen_on("eventB")
async def CallbackB1():
print("Listener B1")
@ewents.listen_on("eventB")
async def CallbackB2():
print("Listener B2")
async def main():
while True:
await ewents.emit("eventA")
await asyncio.sleep(2)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

@ -0,0 +1,35 @@
from setuptools import setup, find_packages
import os
from setuptools import setup
# Utility function to read the README file.
# Used for the long_description. It's nice, because now 1) we have a top level
# README file and 2) it's easier to type in the README file than to put a raw
# string in below ...
def read(fname):
return open(os.path.join(os.path.dirname(__file__), fname)).read()
setup(
name="ewents",
version="0.1",
packages=find_packages(),
# scripts=['say_hello.py'],
# install_requires=['python_version>=3.5'],
# metadata to display on PyPI
author="Yugon",
author_email="ugogon@hotmail.de",
description="This is an async eventlistener library.",
keywords="async, ",
# url="http://example.com/HelloWorld/", # project home page, if any
# project_urls={
# "Bug Tracker": "https://bugs.example.com/HelloWorld/",
# "Documentation": "https://docs.example.com/HelloWorld/",
# "Source Code": "https://code.example.com/HelloWorld/",
# },
# classifiers=[
# 'License :: OSI Approved :: Python Software Foundation License'
# ],
python_requires=">=3.5",
long_description=read('README.md')
)
Loading…
Cancel
Save