refactor async_ng code to introduce random delay in MyAsyncNGBO

Author: grongierisc

src/iop/_business_host.py

@@ -580,43 +580,41 @@ def __init__(self, target, request, timeout=-1, description=None, host=None):
         self._iris_handle = host.iris_handle
         asyncio.create_task(self.send())
 
-    async def send_async(self):
-        # Call the synchronous function
-        self._iris_handle.dispatchSendRequestAsyncNG(self.target, self.host._dispatch_serializer(self.request), self.timeout, self.description)
-        
-        # Periodically check if the request is done
-        while not self._iris_handle.dispatchIsRequestDone():
-            await asyncio.sleep(0.1)  # Adjust the sleep duration as needed
-        
-        # Set the result of the Future
-        self.set_result("Request sent and completed")
-
     async def send(self):
+        # init parameters
         message_header_id = iris.ref()
         queue_name = iris.ref()
         end_time = iris.ref()
         request = self.host._dispatch_serializer(self.request)
-        self._iris_handle.dispatchSendRequestAsyncNG(self.target, request, self.timeout, self.description, message_header_id, queue_name, end_time)
+
+        # send request
+        self._iris_handle.dispatchSendRequestAsyncNG(
+            self.target, request, self.timeout, self.description,
+            message_header_id, queue_name, end_time)
+        
+        # get byref values
         self._message_header_id = message_header_id.value
         self._queue_name = queue_name.value
         self._end_time = end_time.value
 
         while not self._done:
             await asyncio.sleep(0.1)
-            self.done()
+            self.is_done()
 
         self.set_result(self._response)
 
-    def done(self):
+    def is_done(self):
         response = iris.ref()
-        status = self._iris_handle.dispatchIsRequestDone(self.timeout, self._end_time, self._queue_name, self._message_header_id, response)
+        status = self._iris_handle.dispatchIsRequestDone(self.timeout, self._end_time,
+                                                         self._queue_name, self._message_header_id,
+                                                         response)
+        
         self._response = self.host._dispatch_deserializer(response.value)
-        if status == 1 and self._response is not None:
-            self._done = True
 
-    # def __await__(self):
-    #     if not self._done:
-    #         self.done()
-    #         yield self
-    #     else:
-    #         return self._response
+        if status == 2: # message found
+            self._done = True
+        elif status == 1: # message not found
+            pass
+        else:
+            self._done = True
+            self.set_exception(RuntimeError(iris.system.Status.GetOneStatusText(status)))

src/iop/cls/IOP/Common.cls

@@ -223,24 +223,6 @@ ClassMethod OnGetConnections(
     quit
 }
 
-Method dispatchSendRequestAsyncNGTest(
-	pTarget,
-	pRequest,
-	pTimeout,
-	pDescription,
-	ByRef pMessageHeaderId,
-	ByRef pQueueName,
-	ByRef pEndTime) As %String
-{
-    Set tTargetConfigName = $get($$$DispatchNameToConfigName(pTarget)) Quit:""=tTargetConfigName $$$EnsError($$$EnsErrBusinessDispatchNameNotRegistered,pTarget)
-    Set tTargetBusinessClass=$$$ConfigClassName(tTargetConfigName)
-    Set tPriority=$$$eMessagePriorityAsync
-
-    Set tSC=$classmethod(tTargetBusinessClass,"acceptRequestAsync",..%ConfigName,tTargetConfigName,pRequest,tPriority,$$$queueSyncCallQueueName,..%SessionId,"",.tRequestHeader,pDescription,..%SuperSession)
-
-	Quit tSC
-}
-
 Method dispatchSendRequestAsyncNG(
 	pTarget,
 	pRequest,
@@ -305,37 +287,50 @@ Method dispatchIsRequestDone(
 	pMessageHeaderId,
 	ByRef pResponse) As %Status
 {
+
     set tSC=$$$OK
     try {
-        set tTimeout=$s(pTimeout=-1:-1,1:pEndTime-$zh) if (pTimeout'=-1)&&(tTimeout<0) quit
+        set tTimeout=$s(pTimeout=-1:-1,1:pEndTime-$zh)
+
         set tSC = ##class(Ens.Queue).DeQueue($$$queueSyncCallQueueName,.tResponseHeader,tTimeout,.tIsTimedOut,0) Quit:$$$ISERR(tSC)
-        quit:tIsTimedOut 
+
+        quit:$IsObject(tResponseHeader)=0
 
         set tFound = $select(tResponseHeader.CorrespondingMessageId: pMessageHeaderId=tResponseHeader.CorrespondingMessageId, 1: 0)
         if tFound=0 {
-            $$$sysTRACE("Out-of-band message '"_tResponseHeader.%Id()_"' discarded")
-            do tResponseHeader.SetStatus($$$eMessageStatusDiscarded)
-            quit
+
+            set tSC = ##class(Ens.Queue).EnQueue(tResponseHeader)
+            Kill $$$EnsActiveMessage($$$SystemName_":"_$Job)
         }
-        if tResponseHeader.IsError {
+        else {
+
+            if tIsTimedOut || ((pTimeout'=-1)&&(tTimeout<0)) {
+
+                do tResponseHeader.SetStatus($$$eMessageStatusDiscarded)
+                return $$$ERROR($$$EnsErrFailureTimeout, tTimeout, $$$StatusDisplayString(tSC), $$$CurrentClass)
+            }
+            if tResponseHeader.IsError {
+
+                do tResponseHeader.SetStatus($$$eMessageStatusCompleted)
+                return $$$EnsError($$$EnsErrGeneral,"Error message received: "_tResponseHeader.ErrorText)
+                
+            }
+            if tResponseHeader.MessageBodyClassName'="" {
+
+                set tResponse = $classmethod(tResponseHeader.MessageBodyClassName,"%OpenId",tResponseHeader.MessageBodyId,,.tSC)
+                if '$IsObject(tResponse) return $$$EnsError($$$EnsErrGeneral,"Could not open MessageBody "_tResponseHeader.MessageBodyId_" for MessageHeader #"_tResponseHeader.%Id()_" with body class "_tResponseHeader.MessageBodyClassName_":"_$$$StatusDisplayString(tSC)) 
+            } else {
+
+                set tResponse=$$$NULLOREF
+            }
+            set pResponse=tResponse
             do tResponseHeader.SetStatus($$$eMessageStatusCompleted)
-            set tSC = $$$EnsError($$$EnsErrGeneral,"Error message received: "_tResponseHeader.ErrorText)
-            quit
-        }
-        if tResponseHeader.MessageBodyClassName'="" {
-            set tResponse = $classmethod(tResponseHeader.MessageBodyClassName,"%OpenId",tResponseHeader.MessageBodyId,,.tSC)
-            if '$IsObject(tResponse) Set tSC=$$$EnsError($$$EnsErrGeneral,"Could not open MessageBody "_tResponseHeader.MessageBodyId_" for MessageHeader #"_tResponseHeader.%Id()_" with body class "_tResponseHeader.MessageBodyClassName_":"_$$$StatusDisplayString(tSC))  Quit
-        } else {
-            set tResponse=$$$NULLOREF
+            set tSC = 2
+
         }
-        set pResponse=tResponse
-        do tResponseHeader.SetStatus($$$eMessageStatusCompleted)
-        
-    
-    Set tSC2 = ##class(Ens.Queue).Delete($$$queueSyncCallQueueName,"*") quit:$$$ISERR(tSC2)
     }
-	catch {
-		set tSC = $$$EnsSystemError
+	catch ex {
+		set tSC = ex.AsStatus()
 	}
 	quit tSC
 }