Skip to content

A powerful Python publishing subscriber framework that supports synchronous and asynchronous scheduling, timed tasks, topic management, publishing subscriber callbacks, and other functions.

License

Notifications You must be signed in to change notification settings

Undertone0809/broadcast-service

Repository files navigation

broadcast-service

A powerful python broadcast library. You can easily construct a Broadcast pattern/Publish subscriber pattern through this library.

github stars

Features

  • A publishing subscriber pattern can be built with a very simple syntax
  • Support different application scenarios, such as asynchronous and synchronous
  • Provide different syntax writing modes for lambda, callback functions, decorators, etc
  • A callback function listens on multiple subscriptions
  • Provide publisher dispatch callback manage

Quick Start

Setup

pip install broadcast-service

Usage

There is an easy demo to show how to use broadcast-service.

from broadcast_service import broadcast_service

# callback of decorator
@broadcast_service.on_listen('my_topic')
def handle_decorator_msg(params):
    print(f"handle_decorator_msg receive params: {params}")

if __name__ == '__main__':
    info = 'This is very important msg'

    # publish broadcast
    broadcast_service.publish('my_topic', info)
  • You can use publish, emit, broadcast to send your topic msg and use listen, on, subscribe to listen your topic msg.

  • You can also add more arguments or no argument when you publish thr broadcast.

from broadcast_service import broadcast_service

# subscribe topic
@broadcast_service.on_listen(['my_topic'])
def handle_msg(info, info2):
    print(info)
    print(info2)

if __name__ == '__main__':
    info = 'This is very important msg'
    info2 = 'This is also a very important msg.'

    # publish broadcast
    broadcast_service.publish('my_topic', info, info2)
from broadcast_service import broadcast_service

# subscribe topic
@broadcast_service.on_listen(['my_topic'])
def handle_msg():
    print('handle_msg callback')

if __name__ == '__main__':
    # publish broadcast
    broadcast_service.publish('Test')

You can use config to make publisher callback If you want.

from broadcast_service import broadcast_service


@broadcast_service.on_listen("topic")
def handle_subscriber_callback():
    print("handle_subscriber_callback")


def handle_publisher_callback():
    print("handle_publisher_callback")


if __name__ == '__main__':
    broadcast_service.config(
        num_of_executions=5,
        callback=handle_publisher_callback,
        enable_final_return=True,
        interval=0.1
    ).publish("topic")

Moreover, you can see more example in document.

TODO

  • optimize documents and show more examples.
  • optimize the syntax expression of broadcast-service
  • provide more test cases
  • provide the ability to subscribe the topic and callback once
  • support for fuzzy subscriptions
  • the publisher of the topic can provide a return value
  • optimize usage in class ('self' params problem)
  • build observer mode
  • provide publisher callback when all subscriber have completed callback

Contribution

If you want to contribute to this project, you can submit pr or issue. I am glad to see more people involved and optimize it.

About

A powerful Python publishing subscriber framework that supports synchronous and asynchronous scheduling, timed tasks, topic management, publishing subscriber callbacks, and other functions.

Resources

License

Stars

Watchers

Forks

Packages

No packages published