target 시간에 비동기 콜백 호출

target시간(시작 후 1분 뒤)이 되면 코루틴기반 비동기로 my_callback 콜백 함수를 호출 시킨다

 -- fut객체.add_done_callback 으로 등록하는 콜백함수에는 반드시 파라미터가 하나이상 이여야 한다.

 

import asyncio
import datetime
from typing import Union

class AA():
    def __init__(self):
        self.prevent_multi_task = True

    async def wait_target_time(self, target_time: Union[datetime.datetime, str]):
        if self.prevent_multi_task is True:
            self.prevent_multi_task = False
            if isinstance(target_time, str):
                target_time = datetime.datetime.fromisoformat(target_time)
            print(f"Waiting {target_time}")
            print(f"res : {target_time > datetime.datetime.now()}")
            i = 0
            while target_time > datetime.datetime.now():
                print(i)
                i = i + 1
                await asyncio.sleep(1)
            print(f"wait_time def is END")
            self.prevent_multi_task = True
        else:
            print("Task 누적 방지")
            return False


if __name__ == '__main__':
    def my_callback(arg):
        # future 함수의 콜백으로 등록될 시, 인자값으로 _asyncio.Task 객체가 넘어온다. <- future 객체를 상속받음
        print(f"\n\nmy_callback : {arg}")
        print(type(arg))
        print(arg.__dir__())
        print(arg.result())


    async def f_sleep():
        aa = AA()

        _one_minute = datetime.timedelta(minutes=1)
        target_time = datetime.datetime.now() + _one_minute
        # _one_minute = datetime.timedelta(seconds=7)
        # target_time = datetime.datetime.now() + _one_minute

        fut = asyncio.create_task(aa.wait_target_time(target_time))
        fut.add_done_callback(my_callback)

        while True:
            # 프로그램 종료 block
            print("  프로그램 Processing")
            await asyncio.sleep(1)

    ev = asyncio.get_event_loop()

    ev.run_until_complete(f_sleep())

 

 

출력