File size: 2,960 Bytes
d1e2d14 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | from threading import Condition
from typing import Any
from enum import Enum
"""
" Enum RequtestFutureState
"""
class FutureState(Enum):
DEFER = 0
READY = 1
FAILED = 2
"""
" class FutureException
"""
class FutureResult:
FUTURE_SUCC = 0
FUTUTE_ERR_TIMEOUT = 1
FUTURE_ERR_FAILED = 2
FUTURE_ERR_UNKNOWN = 3
def __init__(self, code: int, msg: str, value: Any = None):
self.code = code
self.msg = msg
self.value = value
def __str__(self):
return f"FutureResult(code={str(self.code)}, msg='{self.msg}', value={self.value})"
class Future:
def __init__(self):
self.__state = FutureState.DEFER
self.__msg = None
self.__condition = Condition()
def GetResult(self, timeout: float = None):
with self.__condition:
return self.__WaitResult(timeout)
def Wait(self, timeout: float = None):
with self.__condition:
return self.__Wait(timeout)
def Ready(self, value):
with self.__condition:
ready = self.__Ready(value)
self.__condition.notify()
return ready
def Fail(self, reason: str):
with self.__condition:
fail = self.__Fail(reason)
self.__condition.notify()
return fail
def __Wait(self, timeout: float = None):
if not self.__IsDeferred():
return True
try:
if timeout is None:
return self.__condition.wait()
else:
return self.__condition.wait(timeout)
except:
print("[Future] future wait error")
return False
def __WaitResult(self, timeout: float = None):
if not self.__Wait(timeout):
return FutureResult(FutureResult.FUTUTE_ERR_TIMEOUT, "future wait timeout")
if self.__IsReady():
return FutureResult(FutureResult.FUTURE_SUCC, "success", self.__value)
elif self.__IsFailed():
return FutureResult(FutureResult.FUTURE_ERR_FAILED, self.__msg)
else:
return FutureResult(FutureResult.FUTURE_ERR_UNKNOWN, "future state error:" + str(self.__state))
def __Ready(self, value):
if not self.__IsDeferred():
print("[Future] futrue state is not defer")
return False
else:
self.__value = value
self.__state = FutureState.READY
return True
def __Fail(self, message: str):
if not self.__IsDeferred():
print("[Future] futrue state is not DEFER")
return False
else:
self.__msg = message
self.__state = FutureState.FAILED
return True
def __IsDeferred(self):
return self.__state == FutureState.DEFER
def __IsReady(self):
return self.__state == FutureState.READY
def __IsFailed(self):
return self.__state == FutureState.FAILED |