Decorators

API package path: mcdreforged.api.decorator

@new_thread

@mcdreforged.api.decorator.new_thread.new_thread[source]

This is a one line solution to make your function executes in parallels. When decorated with this decorator, functions will be executed in a new daemon thread

This decorator only changes the return value of the function to the created Thread object. Beside the return value, it reserves all signatures of the decorated function, so you can safely use the decorated function as if there’s no decorating at all

It’s also a simple compatible upgrade method for old MCDR 0.x plugins

The return value of the decorated function is changed to the Thread object that executes this function

The decorated function has 1 extra field:

  • original field: The original undecorated function

Examples:

>>> import time

>>> @new_thread('My Plugin Thread')
... def do_something(text: str):
...     time.sleep(1)
...     print(threading.current_thread().name)
>>> callable(do_something.original)
True
>>> t = do_something('foo')
>>> isinstance(t, FunctionThread)
True
>>> t.join()
My Plugin Thread
Parameters:

arg – A str, the name of the thread. It’s recommend to specify the thread name, so when you log something by server.logger, a meaningful thread name will be displayed instead of a plain and meaningless Thread-3

class mcdreforged.api.decorator.new_thread.FunctionThread(target: Callable[[...], _R], name: str | None, args: Iterable = (), kwargs: dict | None = None)[source]

A Thread subclass which is used in decorator new_thread() to wrap a synchronized function call

get_return_value(block: bool = False, timeout: float | None = None)[source]

Get the return value of the original function

If an exception has occurred during the original function call, the exception will be risen again here

Examples:

>>> import time
>>> @new_thread
... def do_something(text: str):
...     time.sleep(1)
...     return text

>>> do_something('task').get_return_value(block=True)
'task'
Parameters:

  • block – If it should join the thread before getting the return value to make sure the function invocation finishes

  • timeout – The maximum timeout for the thread join

Raises:

RuntimeError – If the thread is still alive when getting return value. Might be caused by block=False while the thread is still running, or thread join operation times out

Returns:

The return value of the original function

@event_listener

@mcdreforged.api.decorator.event_listener.event_listener(event: PluginEvent | str, *, priority: int | None = None)[source]

This decorator is used to register a custom event listener without involving register_event_listener()

It accepts a single str or PluginEvent indicating the event you are listening to as parameter, and will register the function as the callback of the given listener

It’s highly suggested to use this decorator only in the entry point of your plugin, so it can work correctly and register the event listener in the correct time

Example:

@event_listener(MCDRPluginEvents.GENERAL_INFO)
def my_on_info(server, info):
        server.logger.info('on info in my own listener')

The above example is equivalent to:

def on_load(server, old):
        server.register_event_listener(MCDRPluginEvents.GENERAL_INFO, my_on_info)
Parameters:

event – The event to register a listener

Keyword Arguments:

priority – Optional, the priority of the event listener

Raises:

@spam_proof

@mcdreforged.api.decorator.spam_proof.spam_proof(arg=None, *, lock_class=<function RLock>, skip_callback: Callable | None = None)[source]

Use a lock to protect the decorated function from being invoked on multiple threads at the same time

If a multiple-invocation happens, only the first invocation can be executed normally, other invocations will be skipped

The return value of the decorated function is modified into a bool, indicating if this invocation is executed normally

The decorated function has 2 extra fields:

  • original field: stores the original undecorated function

  • lock field: stores the lock object used in the spam proof logic

Example:

@spam_proof
def some_work(value):
        # doing some important logics
        foo = value

The above example is equivalent to:

lock = threading.RLock()

def some_work(value) -> bool:
        acquired = lock.acquire(blocking=False)
        if acquired:
                try:
                        # doing some not thread-safe logics
                        foo = value
                finally:
                        lock.release()
        return acquired
Keyword Arguments:

  • lock_class – The type of the lock. It can be threading.Lock or threading.RLock (default)

  • skip_callback – (optional) The callback function that will be invoked with all parameters of the decorated function when the invocation is skipped

Keyword skip_callback example:

>>> def my_callback(value):
...     print('skip', value)

>>> @spam_proof(skip_callback=my_callback)
... def some_work(value):
...     event.wait()

>>> def threaded_invoke():
...     print(some_work(0.1))  # invocation normal

>>> from threading import Thread, Event
>>> t, event = Thread(target=threaded_invoke), Event()
>>> t.start()
>>> some_work(123)  # invocation skipped
skip 123
False
>>> _ = event.set(), t.join()
True

Added in version v2.5.0.

Added in version v2.7.0: Added skip_callback keyword argument