JPlatform 10 - L'API ReplicaMessage

Dans un cluster JSync, seules les données du store (et éventuellement les fichiers associés) sont recopiées sur les différents réplicas du cluster. JcmsDB étant commune aux différents réplicas, les données sont accessibles par tous les réplicas.

JPlatform 10 introduit une nouvelle API, les ReplicaMessage, qui permet aux réplicas de communiquer en dehors des échanges JSync. Le protocole utilisé garanti la distribution des messages dans l'ordre d'émission, même si un réplica n'est pas joignable lors de l'envoi.

JPlatform 10 utilise cette API pour propager les changements sur la structure des Workflows entre les réplicas.

Cette API pourrait aussi être utilisée plus tard pour propager les changements sur certaines propriétés de paramétrage.

1. Principes de distribution des messages

L'algorithme de distribution des messages reprend les principes de celui des DBEventLog.

Lorsqu'un réplica veut envoyer un message aux autres réplicas :

• si il s'agit d'un simple réplica, il enregistre dans JcmsDB un message pour son leader
• si il s'agit du leader ou d'un sous-leader, il enregistre dans JcmsDB autant de message qu'il a de réplica explicitement rattachés
• le leader et chaque réplica sont ensuite notifiés par appel de la JSP processReplicaMessage.jsp

Chaque message comporte :

• l'URID du réplica émetteur (source)
• l'URID du réplica destinataire (target)
• une date d'émission (cdate)
• une date d'expiration (expiryDate)
• le type de message (type)
• la portée applicative du message (scope)
• des informations propre au message (payload).

Chaque réplica interroge régulièrement JcmsDB pour relever les messages qui lui sont destinés (i.e. ceux dont target correspond à l'URID du réplica)

Pour chaque message :

• Si la date d'expiration est dépassée alors le message est supprimé
• Sinon
  • Si le réplica est un leader ou un sous-leader alors le message est réémis, au nom du leader à l'ensemble des réplicas à l'exception du réplica émetteur (source)
  • le message est délivré à tous les composants qui écoutent les messages
  • le message est supprimé

2. La classe ReplicaMessage

Cette classe représente un message envoyé d'un réplica à un autre réplica.

class ReplicaMessage implements DBData {
  Long rowId;
  Date cdate;
  String source;
  String target;
  Date expiryDate;
  String type;
  String scope;
  String payload;
}

Champs :

• rowId : Identifiant du message
• cdate : Date de création du message
• source : URID de l'émetteur du message
• target : URID du destinataire du message
• type : type de message
• scope : portée ou donnée cible du message
• expiryDate : date d'expiration du message
• payload : contenu du message (limité dans le HBM à 512 KB)

Méthode :

• send() déclenche l'envoi du message en invoquant ReplicaMessageManager.getInstance().sendMessage(this).

3. La classe ReplicaMessageManager

Il s'agit du singleton qui traite l'envoi et la relève des messages.

3.1 La méthode sendMessage()

Cette méthode envoie un message au cluster (au leader si il s'agit d'un simple réplica, et à l'ensemble des réplicas rattachés si il s'agit d'un leader).

3.2 La méthode sendNotification()

Cette méthode déclenche l'appel à la JSP admin/processReplicaMessage.jsp sur le leader et les réplicas actuellement connectés.

3.3 Les méthodes addListener() / removeListener()

Ces deux méthodes permettent d'ajouter et de supprimer des listener de messages reçus par ce réplica.

3.4 La relève des messages

Une alarme se déclenche à fréquence régulière (propriété replica-message.scan-freq défini epar défaut à 1 minute).

A chaque déclenchement la méthode processNewMessages() est appelée pour relever les nouveaux messages, les envoyer aux listeners et les supprimer.

3.5. La JSP processReplicaMessage.jsp

Cette JSP ne fait qu'appeler ReplicaMessage.getInstance().processNewMessages().

4. La classe ReplicaMessageListener

Cette classe doit être implementée pour recevoir des messages.

La classe dispose d'une seule méthode : processMessage(ReplicaMessage)

Le listener doit commencer par vérifier si le message le concerne.

Le listener n'a pas à vérifier la date d'expiration car cela a été fait au préalable par le manager.

5. Exemple d'utilisation

Cet exemple illustre, d'une façon simplifiée, l'utilisation de l'API des ReplicaMessage pour la synchronisation des changements sur les workflows.

5.1 Envoi des messages

class WorkflowManager {
  // ...
  private static final String REPLICA_MESSAGE_WORKFLOW         = "core/workflow";
  private static final String REPLICA_MESSAGE_WORKFLOW_DELETE  = "core/workflow/delete";

  private void sendWFChangeMessage(Workflow wf) {
        
    ReplicaMessage msg = new ReplicaMessage();
    msg.setType(REPLICA_MESSAGE_WORKFLOW);
    msg.setScope(wf.getId());
    msg.setPayload(wf.toXml());
    msg.send();
  }
  
  private void sendWFDeleteMessage(String wfId) {
    ReplicaMessage msg = new ReplicaMessage();
    msg.setType(REPLICA_MESSAGE_WORKFLOW_DELETE);
    msg.setScope(wfId);
    msg.send();
  }
}

5.2 Traitement des messages

class WorkflowManager implements ReplicaMessageListener {
  // ...

  public void initReplicaMessageListener() {
    ReplicaMessageManager.getInstance().addListener(this);
  }

  @Override
  public void processMessage(ReplicaMessage msg) {
    
    if (!channel.isAvailable()) {
      return;
    }
    
    String type = msg.getType();
    
    if (type == null || !type.startsWith(REPLICA_MESSAGE_WORKFLOW)) {
      return; 
    }
    
    String wfId = msg.getScope();
    if (Util.isEmpty(wfId)) {
      logger.warn("Received workflow ID is empty");
      return;
    }
    
    if (REPLICA_MESSAGE_WORKFLOW_DELETE.equals(type)) {
      processWFDeleteMessage(wfId);
    }     
    else {
      processWFChangeMessage(wfId, msg.getPayload());
    }
  } 
}