"""handler intf"""
from abc import ABC, abstractmethod
from typing import List
from yr.err_type import ErrorInfo
class HandlerIntf(ABC):
"""
Handler interface
"""
@abstractmethod
def execute_function(self, func_meta, args: List, invoke_type, return_num: int, is_actor_async: bool):
"""
Execute the function.
"""
@abstractmethod
def shutdown(self, grace_period_second: int) -> ErrorInfo:
"""
Shutdown the instance gracefully.
Args:
grace_period_second (int): The time to wait for the instance to shutdown gracefully.
Returns:
The ErrorInfo of the execution of shutdown function.
Raises:
RuntimeError: If the instance has not been initialized.
"""
@abstractmethod
def before_snapshot(self) -> ErrorInfo:
"""
Trigger snapshot preparation hook before taking snapshot.
This method is called by libruntime before creating a snapshot.
It invokes the user-defined __yr_before_snapshot__ method if present.
Returns:
ErrorInfo: Error information if hook execution failed.
"""
@abstractmethod
def after_snapstart(self) -> ErrorInfo:
"""
Trigger snapshot recovery hook after restoring from snapshot.
This method is called by libruntime after restoring from a snapshot.
It invokes the user-defined __yr_after_snapstart__ method if present.
Returns:
ErrorInfo: Error information if hook execution failed.
"""