py-event-bus's Introduction

py-event-bus

Super simple event bus in Python 3, built with asyncio. No decorators required. Built around a subset of the NodeJS EventEmitter API.

A PyPI package implementing this as a module can be found here.

A full write up and explanation can be found here.

Usage

from py_event_bus.EventBus import EventBus
event_bus = EventBus()

async def func(event):
  for pet in event['pets']:
    print(pet)
  
event_data = {
  'pets': ['cats', 'dogs']  
}
event_bus.add_listener('some-event', func)
event_bus.emit('some-event', event_data)
event_bus.remove_listener('some-event', func)

This will give:

> cats
> dogs

Development Setup

git clone [email protected]:joeltok/py-event-bus.git
cd ./py-event-bus
python3 -m venv ./venv
source venv/bin/activate
pip3 install pytest
pip3 install pytest-asyncio

Testing

python3 -m pytest event_bus/ event_bus_advanced_examples/

py-event-bus's People

Contributors

Stargazers

Watchers

py-event-bus's Issues

Trying to use this to emit between modules

I'm basing an EventEmitter on your approach . I created a package I'm calling EventEmitter with a singleton class called Events. I'm trying to emit and listen to events between two different modules that use this package. But what I've found is each instance of the class, even though the class is a singleton, is running on its own thread. Because of that, I'm not able to emit from one module and get the event on the listener in the other.

Do you have any thoughts about how I can get this to work?

(Note: I'm relatively new to python, so I might not be aware of some other python way to do what I'm attempting)

EventEmitter:

import asyncio


class Events:
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(Events, cls).__new__(cls)
        return cls._instance

    def __init__(self):
        self.listeners = {}

    def on(self, event_name, listener):
        if not self.listeners.get(event_name, None):
            self.listeners[event_name] = {listener}
        else:
            self.listeners[event_name].add(listener)

    def off(self, event_name, listener):
        self.listeners[event_name].remove(listener)
        if len(self.listeners[event_name]) == 0:
            del self.listeners[event_name]

    def emit(self, event_name, event):
        listeners = self.listeners.get(event_name, [])
        for listener in listeners:
            asyncio.create_task(listener(event))

Emitter

import time

from Drivers.Events import EventEmitter

em = EventEmitter.Events()
print(em)


while True:
    em.emit("hi", "hello")
    time.sleep(5)

Listener

from Drivers.Events import EventEmitter
import asyncio

em = EventEmitter.Events()
print(em)


async def received(message, *args, **kwargs):
    print(message)
    return message

while True:
    em.on("hi", received)

Can I use it between multiprocessing?

Hi, I have two questions:

Can I use it between multi processes? like one process emits event and another process listen it.
Can I emit event which is user defined class object nor string, int, and so on ?

Thank you.

Recommend Projects

joeltok / py-event-bus Goto Github PK

py-event-bus's Introduction

py-event-bus

Usage

Development Setup

Testing

py-event-bus's People

Contributors

Stargazers

Watchers

Forkers

py-event-bus's Issues

Trying to use this to emit between modules

EventEmitter:

Emitter

Listener

Can I use it between multiprocessing?

Recommend Projects

React

Vue.js

Typescript

TensorFlow

Django

Laravel

D3

Recommend Topics

javascript

web

server

Machine learning

Visualization

Game

Recommend Org

Facebook

Microsoft

Google

Alibaba

D3

Tencent