Upd: transfer protect

Author: GoodManWEN

README.md

@@ -126,6 +126,8 @@ async def async_root(*args):
 
 def sync_prime_number(redis, mysql, x):
     # Example synchronous function to determine if the input x is a prime number.
+    # redis and mysql clients are entered by default, starting from the third parameter 
+    # is your custom parameters, only keyword parameters are supported. 
     import math, time
     if x == 1:
         return True
@@ -137,6 +139,12 @@ def sync_prime_number(redis, mysql, x):
 
 async def async_fibonacci(redis, mysql, n):
     # Example asynchronous function to calculate the nth position of the Fibonacci series.
+    # redis and mysql clients are entered by default, starting from the third parameter 
+    # is your custom parameters, only keyword parameters are supported.
+
+    # Be sure to note that all data to upload and download must be serializable by msgpack.
+    # This means that if you transfer some custom object, or in this case a very large integer, 
+    # the request will be responsed with an internal server error (http 500).
     a, b = 0, 1
     for _ in range(n):
         a, b = b, a + b 

fastapi_queue/gateway_manager.py

@@ -52,6 +52,8 @@ async def redis_channel_listening_result(self, *args, **kwargs):
                         if message['data'] == self._confirm_token:
                             task_status_activate = True
                             break
+                        else:
+                            raise RuntimeError('This is special class to account for confirmation failures')
         except asyncio.TimeoutError:
             asyncio.create_task(self._withdraw_query(*args, **kwargs))
         except Exception as e:
@@ -66,15 +68,17 @@ async def redis_channel_listening_result(self, *args, **kwargs):
                 while True:
                     message = await channel.get_message(ignore_subscribe_messages=True, timeout = self._process_timeout + 5)
                     if message is not None:
-                        result = json.loads(message['data'])
+                        run_success, result = json.loads(message['data'])
                         break
         except asyncio.TimeoutError:
             return False, -2
         except Exception as e:
             if debug:
                 raise e
+            return False, e
         else:
-            return True, result 
+            if run_success: return True, result 
+            else: return False, result
 
     async def _publish_query(self, form_data: dict = {}, task_level: int = 0):
         '''

fastapi_queue/worker.py

@@ -167,11 +167,16 @@ async def _worker_thread(self, thread_idx: int):
                     await self.redis.publish(result_channel_name, self._confirm_token)
                     self._logger.debug(f"Pid: {self.pid}, `{result_channel_name}` confirmed")
                     method = self._method_map[task_route]
-                    if method.__name__[0] == 'a':
-                        res = await method(self.redis, self.mysql, **form_data)
+                    try:
+                        if method.__name__[0] == 'a':
+                            res = await method(self.redis, self.mysql, **form_data)
+                        else:
+                            res = await loop.run_in_executor(self._executor, partial(method, self.redis, self.mysql, **form_data))
+                    except Exception as e:
+                        res = json.dumps([False, f'{type(e)}:{e}'])
                     else:
-                        res = await loop.run_in_executor(self._executor, partial(method, self.redis, self.mysql, **form_data))
-                    res = json.dumps(res)
+                        # format: [run_success: bool, result: Any]
+                        res = json.dumps([True, res])
                     self._logger.info(f"Pid: {self.pid}, return: {res[:20]}")
                     await self.redis.publish(result_channel_name, res)