Exemple de processus scalable

unit TestSYFREWF;
interface
 
Type
  MyProcessus = Class(TitObject)
  public
    Procedure doDeQueue;
    Procedure doInQueue;
    Procedure doProcess(obj:TQueueObject);
    Procedure doTask;
    Procedure Execute;
  end;
 
Implementation
 
{MyProcessus}
 
Procedure MyProcessus.doDeQueue;
//Procedure doDeQueue;
var obj:MyQueue; indx:Integer; aTopic:string;
begin
  // In this step we dequeue and process the ready objects.
 
  // Request objects in initial state in case of the queue contains multiple states
  aTopic := 'wfclassea/0/%';
 
  // This is just an estimation of the count, could be false if several processes run in parallel
  ProgressMax(MyQueue.CountWhere('queueTopic like %1',true,[aTopic]));
 
  // Get an enumerator from the queue
  foreach obj in MyQueue.Topic(aTopic) index indx do
   try
     // Progress indicator
     ProgressCount(indx);
 
     // Process this queue object
     // Separate the code to support straight through processing
     doProcess(obj);
 
     // delete the queue object
     obj.DeleteThisObject;
   except
     // on error requeue the object on an error topic
     // do not requeue on the same topic if the processus must be used in a straight through mode
     obj.ReQueue(StringReplace(obj.queueTopic,'/0/','/99/'));
   end;
  //
  ProgressFull();
end;;
 
Procedure MyProcessus.doInQueue;
//Procedure doInQueue;
var sel:TSelector;
begin
  // In this step we InQueue all objects which match the process criteria.
 
  // In this example the process criteria is objects in initial state
  // Use a selector for objects in initial state
  sel := WFClasseA.CreateSelector('unEtat=%1','',true,[WFCAState_Initial]);
  // Insert the objects in the queue using the selector, this is a transactional operation.
  // Topic is //classname/state/code/oid
  sel.InQueue('MyQueue','wfclassea/@unEtat/@unCode/@oid');
end;;
 
Procedure MyProcessus.doProcess(obj:TQueueObject);
//Procedure doProcess(obj:TQueueObject);
var inst:WFClasseA;
begin
  // One transaction by object
  withP transaction do
   begin
     // Get the actual object
     inst := obj.queueRef as WFClasseA;
     if Assigned(inst) then
      begin
        // this is the process main code where business transformations are applied.
        // here we change the state of the object
        inst.unEtat.value := WFCAState_Etat2;
        ProgressMessage(obj.queueTopic);
      end;
   end;
end;;
 
Procedure MyProcessus.doTask;
//Procedure doTask;
begin
  // Step One : inqueue object to be processed.
  doInQueue;
  // Step two : Process objects from queue.
  doDeQueue;
end;;
 
Procedure MyProcessus.Execute;
//Procedure Execute;
var aMsg:TdbmQueueMessage; aTopic:string; idx:Integer;
begin
  // Main code of the processus
  if Assigned(UserContext.TaskContext) then
   begin
     // The processus has been started by the scheduler
     //
     UserContext.TaskContext.AddMessage('MyProcessus.Execute');
     if Assigned(UserContext.TaskContext.EventContext.receivedMsg) then
       begin
         // The scheduler task has consumed one object from the queue.
         //
         aMsg := UserContext.TaskContext.EventContext.receivedMsg;
         UserContext.TaskContext.AddMessage(Format('Message :%s',[aMsg.queueTopic]]);
 
         // a TdbmQueueMessage is also a TQueueObject
         // Original properties of the MyQueue object has been copied into aMsg
         doProcess(aMsg);
       end
       else
       begin
         // The scheduler task is not queue based. 
         //
         UserContext.TaskContext.AddMessage('No message, process as a regular task');
         doTask;
       end;
   end
   else doTask; // The processus has been started by the UI
end;

Procedure MyProcessus.Execute;
//Procedure Execute;
var aMsg:TdbmQueueMessage; aTopic:string; idx:Integer;
begin
  if Assigned(UserContext.TaskContext) then
   begin
     UserContext.TaskContext.AddMessage('MyProcessus.Execute');
     if Assigned(UserContext.TaskContext.EventContext.receivedMsg) then
       begin
         aMsg := UserContext.TaskContext.EventContext.receivedMsg;
         UserContext.TaskContext.AddMessage(Format('Message :%s',[aMsg.queueTopic]]);
 
         // a TdbmQueueMessage is also a TQueueObject
         // Original properties of the MyQueue object has been copied into aMsg
         doProcess(aMsg);
       end
       else
       begin
         UserContext.TaskContext.AddMessage('No message, process as a regular task');
         doTask;
       end;
   end
   else doTask; // started by the UI
end;

//Procedure doDeQueue;
begin
  ...
  foreach obj in MyQueue.Topic(aTopic) index indx do
   try
     ....
   except
     // on error requeue the object on an error topic
     // do not requeue on the same topic if the processus must be used in a straight through mode
     obj.ReQueue(StringReplace(obj.queueTopic,'/0/','/99/'));
   end;
end;

//Procedure doDeQueueWithGroupBy;
var obj:MyQueue; aList:TQueueObjectList; indx:Integer; aTopic,aGroupByTopic:string;
begin
  // In this step we dequeue and process the ready objects.
  // The objects are dequeued group by unCode
 
  // Request objects in initial state in case of the queue contains multiple states
  aTopic := 'wfclassea/0/%';
 
  // This is just an estimation of the count, could be false if several processes run in parallel
  ProgressMax(MyQueue.CountWhere('queueTopic like %1',true,[aTopic]));
 
  obj := MyQueue.GetFirst(aTopic);
  Repeat
    if Assigned(obj) then
     begin
       // Progress indicator
       ProgressCount(indx);
 
       // Get the topic on which to group by
       aGroupByTopic := 'wfclassea/0/'+StringPart(obj.queueTopic,'/',2)+'/%';
 
       // Get a list of queue objects with this topic
       aList := MyQueue.deQueueList(aGroupByTopic);
       // the objects may have been consumed by an other process, so we have to check it again.
       if Assigned(aList) and (aList.Count>0) then
         try
           // process the list of objects
           doProcessList(aList);
           aList.DeleteObjects;
         except
         // On error ReQueue the list
         MyQueue.ReQueueList(aList,'/0/','/99/');
         end;
     end;
    obj := MyQueue.GetNext(aTopic,obj);
  Until not Assigned(obj);
  //
  ProgressFull();
end;
 
//Procedure doProcessList(aList:TQueueObjectList);
var inst:WFClasseA; idx:Integer;
begin
  // One transaction by object
  withP transaction do
   begin
     for idx:=0 to aList.Count-1 do
      begin
        // Get the actual object
        inst := aList.Refs[idx].queueRef as WFClasseA;
        if Assigned(inst) then
         begin
           // this is the process core where business transformations are apply.
           // here we change the state of the object
           inst.unEtat.value := WFCAState_Etat2;
           ProgressMessage(aList.Refs[idx].queueTopic);
         end;
      end;
   end;
end;