Dans ce dernier article de la série « BPEL / Weblogic / Advanced Queuing », je me propose de vous montrer comment rendre la consommation synchrone de telle sorte que les messages soient déplacés en file d’exception si une erreur apparaît lors du traitement du message. Vous pourrez ainsi recycler facilement l’ensemble des messages dont le traitement a échoué.
En fin d’article, je proposerai quelques propriétés pour gérer plus finement la consommation des messages dans les queues de messages.
Consommation synchrone
On part d’un processus BPEL consommateur asynchrone (cf. troisième partie de la série) pour le rendre synchrone :
Modification du wsdl
Ci-desous, les parties à ajouter (en bleu) :
<?binding.jca Dequeue_jms.jca?>
<wsdl:definitions name="Dequeue"
targetNamespace="http://xmlns.oracle.com/pcbpel/adapter/jms/JMS/Dequeue/Dequeue%2F" xmlns:tns="http://xmlns.oracle.com/pcbpel/adapter/jms/JMS/Dequeue/Dequeue%2F"
xmlns:wsdl="http://schemas.xmlsoap.org/wsdl/"
xmlns:imp1="http://xmlns.oracle.com/JMS/Enqueue/EnqueueProcess"
xmlns:plt="http://schemas.xmlsoap.org/ws/2003/05/partner-link/"
xmlns:bpelx="http://schemas.oracle.com/bpel/extension">
<plt:partnerLinkType name="Consume_Message_plt">
<plt:role name="Consume_Message_role">
<plt:portType name="tns:Consume_Message_ptt"/>
</plt:role>
</plt:partnerLinkType>
<wsdl:import namespace="http://schemas.oracle.com/bpel/extension"
location="RuntimeFault.wsdl"/>
<wsdl:types>
<schema xmlns="http://www.w3.org/2001/XMLSchema"
targetNamespace="http://xmlns.oracle.com/pcbpel/adapter/jms/JMS/Dequeue/Dequeue%2F"
elementFormDefault="qualified">
<import namespace="http://xmlns.oracle.com/JMS/Dequeue/DequeueProcess"
schemaLocation="xsd/DequeueProcess.xsd"/>
<element name=”empty” />
</schema>
</wsdl:types>
<wsdl:message name="Consume_Message_msg">
<wsdl:part element="imp1:process"/>
</wsdl:message>
<wsdl:message>
<wsdl:part element="tns:empty"/>
</wsdl:message>
<wsdl:portType name="Consume_Message_ptt">
<wsdl:operation name="Consume_Message">
<wsdl:input message="tns:Consume_Message_msg"/>
<wsdl:output message="tns:ignoreMessage"/>
<wsdl:fault
message="bpelx:RuntimeFaultMessage"/>
</wsdl:operation>
</wsdl:portType>
</wsdl:definitions>
En modifiant ainsi le wsdl, on rend le service de dépilement des messages synchrone. Le message sera donc consommé dans la queue si et seulement si l’activité reply est exécutée.
Selon la configuration de la queue, en cas de non réponse de la part du BPEL, le message pourra être rejoué plusieurs fois avant d’être déplacé dans la queue d’exception.
Modification du bpel
Ajout d’une activité reply en fin de flux pour « signaler » le bon déroulement de la consommation (la variable retournée est de type « ignoreMessage »).
Gestion des exceptions
Comme on l’a mentionné plus haut, dans le cas où l’activité reply n’est pas exécutée, le message sera déplacé dans la queue d’exception associée à la queue d’origine du message (ou une autre queue d’exception si elle a été précisée dans l’en-tête du message).
Les messages en queue d’exception peuvent être consommés par un programme tiers et réinsérer dans la queue d’origine pour être à nouveau consommés par les flux BPEL (utilisation des API Java ou PL/SQL pour recycler simplement les messages dans les queues de messages d’origine).
Gérer le rythme de consommation des messages
Certaines propriétés de l’adaptateur de ressources JMS permettent de réguler la consommation des messages dans la queue de messages.
Les propriétés intéressantes sont les suivantes :
- « adapter.jms.receive.threads » : nombre de threads consommateurs en parallèle
- « minimumDelayBetweenMessages » : intervalle de consommation entre deux messages pour un thread donné
Pour ajouter ces propriétés, deux façons :
- Mode graphique
-
- Se positionner sur le service de consommation dans le fichier : composite.xml.
- Dans la vue « Property inspector », ajouter les propriétés évoquées ci-dessus dans la partie « binding properties ».
- Mode source
Insérer les lignes suivantes dans le composite.xml :
<service ui:wsdlLocation="Dequeue.wsdl">
<interface.wsdl interface="..."/>
<binding.jca config="Dequeue_jms.jca">
<property name="adapter.jms.receive.threads"
many="false" override="may">2</property>
<property type="xs:long" many="false"
override="may">10000</property>
</binding.jca>
</service>
Ces propriétés peuvent être mises à jour dynamiquement (pas de redémarrage du serveur) après le déploiement du processus via la console EM (clic droit sur le flux en question, « Services et Références », onglet « propriétés »). C’est très intéressant puisque cela vous permet de réguler très facilement la consommation, notamment par des personnes autres que les développeurs.
Si ces propriétés peuvent être modifiées sur le serveur, j’ai remarqué qu’elles ne pouvaient pas être ajoutées directement sur le serveur (erreur apparaissant dans les logs).
Enfin, j’ai également remarqué que ces propriétés ne sont efficaces que sur les consommateurs de queues dites « à consommateur unique » et non « multi-consommateurs ». En effet, l’ajout de ces propriétés sur un adaptateur destiné à consommer dans un topic (concept JMS pour une queue de messages « multi-consommateurs ») provoque des erreurs dans les logs et n’a aucun effet. Cette erreur stipule qu’il est impossible d’inscrire plusieurs threads consommateurs avec le même nom de consommateur …
Conclusion
Cette série d’articles vous aura permis d’explorer les possibilités offertes par BPEL en termes d’interaction avec des queues de messages basées sur Oracle Advanced Queuing.
