Messaging e message-driven bean (MDB)

Di solito le comunicazioni fra i componenti di un’applicazione sono sincrone il che significa che l’oggetto che effettua la chiamata e l’oggetto che viene invocato devono essere entrambi presenti affinchè la comunicazione possa avvenire.
La comunicazione sincrona comporta anche il fatto che il client deve attendere che l’oggetto invocato completi l’elaborazione della richiesta prima di poter procedere oltre.
I Message-Oriented Middleware (MOM) permettono a due oggetti di comunicare in maniera asincrona facendo da ponte fra l’uno e l’altro: in questo caso non è più necessaria la presenza contemporanea dei due oggetti in quanto il messaggio verrà trattenuto dal MOM e consegnato solo quando il destinatario potrà riceverlo.

Message-oriented middleware (MOM)

Un message-oriented middleware è un software che consente a due componenti di comunicare in maniera asincrona.
Quando un componente invia un messaggio, il MOM lo memorizza nella locazione specificata dal mittente e ne comunica l’avvenuta ricezione.
Il componente che invia il messaggio è detto producer, la locazione nella quale il messaggio viene memorizzato è detta destinazione.
Successivamente il componente interessato al messaggio, detto consumer, può recuperarlo dalla locazione nella quale è stato memorizzato.
Un modello di messaging definisce il modo in cui il processo di messaging coinvolge mittenti e destinatari.
I due modelli più popolari, standardizzati in Java EE, sono il point-to-point e il pubish-subscribe.
Nello schema point to point, un messaggio viaggia da un produttore a un singolo consumatore e le destinazioni del messaggio sono dette code; questo modello non garantisce che i messaggi siano consegnati in secondo un particolare ordine.
Nello schema publish-subscribe un singolo produttore invia un messaggio ad una destinazione che prende il nome di topic e tutti i consumatori che sono registrati presso quel topic possono riceverlo.
Questo modello lavora si adatta particolarmente bene nel caso in sui si voglia effettuare il broadcasting di informazioni verso diversi sistemi.

Java MEssaging Service (JMS)

Java Messaging Service (JMS) fornisce uno standard di accesso ai MOM che costituisce un’alternativa all’utilizzo di API proprietarie; ad eccezione di Microsoft Message Queuing, la maggior parte dei prodotti MOM infatti supporta JMS.
Supponiamo di voler lanciare dei processi attraverso una piccola web application senza volerne attendere il completamento.
In questo caso sfruttando il modello point-to-point potremmo inviare un messaggio contenente il nome del comando che deve essere eseguito ad una coda, e fare in modo che un consumer prelevi i messaggi dalla coda e ne esegua i comandi.
Dal momento che abbiamo supposto l’invio di messaggi mediante una piccola web application, abbiamo bisogno di un session bean stateless che svolga le funzioni di producer dei messaggi e di una servlet che utilizzi il session bean per inviare il messaggio contenente il comando specificato dall’utente.
L’interfaccia del session bean deve essere locale (perché utilizzata da una servlet) e prevedere un metodo "accoda" che riceve in ingresso una stringa corrispondente al comando da inviare.

package miopackage;

import javax.ejb.Local;
   

public interface SessionBeanProducerLocal 
{
  public void accoda(String comando);
}

L’implementazione dell’interfaccia invece potrebbe avvenire nel seguente modo:

package miopackage;
   
import javax.annotation.Resource;
import javax.ejb.Stateless;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSession;
import javax.jms.Session;
   

public class SessionBeanProducer implements SessionBeanProducerLocal 
{
   
  (mappedName="ConnectionFactory")
  private QueueConnectionFactory connectionfactory;
        
  (mappedName="queue/MiaCodaMessaggi")
  Queue destinazione;
        
      
  public SessionBeanProducer() 
  {
  }
   
  
  public void accoda(String comando) 
  {
    try
    {
      QueueConnection connessione = connectionfactory.createQueueConnection();
      QueueSession sessione = connessione.createQueueSession(true,Session.AUTO_ACKNOWLEDGE);
      MessageProducer produttore = sessione.createProducer(destinazione);
      Message messaggio = sessione.createTextMessage(comando);
      produttore.send(messaggio);
      sessione.commit();
      sessione.close();
      connessione.close();
    }
    catch (Exception ecc)
    {
      System.out.println(ecc);
    }
  }
}

Nell’esempio, mediante dependency injection si ottengono i riferimenti alla Queue e al QueueConnectionFactory che verranno utilizzati all’interno del metodo accoda.

Questo metodo svolge nell’ordine le seguenti operazioni:

• crea una QueueConnection mediante il metodo createQueueConnection del QueueConnectionFactory
• crea una QueueSession mediante il metodo createQueueSession del QueueConnection (il primo parametro specifica che la sessione è transazionale e quindi il messaggio non verrà inviato finchè non verrà effettuato il commit, il secondo indica lamodalità di acknowledge e ha effetto soltanto per le sessioni non-transazionali).
• crea un MessageProducer mediante il metodo createProducer passando come parametro la Queue di destinazione
• crea un TextMessage inizializzato col valore del parametro “comando” passato come parametro del metodo
• invia il messaggio mediante il metodo send del MessageProducer passando come parametro il TextMessage
• effettua il commit della transazione
• chiude sessione e connessione

Il session bean definito verrà quindi utilizzato all’interno di una servlet per inviare il comando passato ad esempio mediante metodo get:

package miopackageweb;
   
import java.io.IOException;
import java.io.PrintWriter;
   
import javax.ejb.EJB;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
   
import miopackage.SessionBeanProducerLocal;
   
public class MiaServlet extends HttpServlet 
{
  private static final long serialVersionUID = 1L;
   
  @EJB
  private SessionBeanProducerLocal sessionbeanproducer;
      
  public MiaServlet() 
  {
  }
   
  protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException 
  {
    sessionbeanproducer.accoda(request.getParameter("comando"));
    PrintWriter out = response.getWriter();
    out.println( "Comando inviato" );
    out.close();
  }
   
  protected void doPost(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException 
  {
  }
}

L'interfaccia Message

L’interfaccia Message rappresenta il messaggio che viene scambiato mediante JMS e costituisce un meccanismo di incapsulamento molto robuto.
Un messaggio è costituito da tre parti: header, proprietà e body.
L’header è costituito da coppie nome-valore comuni a tutti i messaggi (alcuni header comuni sono JMSCorrelationID, JMSReplyTo, JMSMEssageID).
Le proprietà sono simili agli header ma sono create esplicitamente per un’applicazione.

Se ad esempio si vuole creare una proprietà booleana Originale si può scrivere:

message.setBooleanProperty("Originale", true);

Una proprietà può essere boolean, byte, double, int, long, short, String o Object.
Il body del messaggio è dato dal contenuto dello stesso, il cosiddetto payload; a seconda del tipo del contenuto è opportuno scegliere il tipo di messaggio da utilizzare.
Nel precedente esempio è stato utilizzato un TextMessage perché il contenuto del messaggio era una semplice stringa, alternativamente si può usare BytesMessage, MapMessate, StreamMessate o ObjectMessage.

Message-driven bean

I message-driven bean (MDB) sono componenti EJB progettati per fare da consumer di messaggi asincroni.
In generale un’applicazione potrebbe avere la necessità di disporre di consumer di messaggi multithreading in grado di elaborare i messaggi in parallelo.
I MDB sollevano anche dalla necessità di codificare gli aspetti meccanici dell’elaborazione del messaggio come il look up di connection factory o destinazioni, la creazione di connessioni, l’apertura di sessioni, la creazione di consumer e l’aggiuna di listener.

L’implementazione di un MDB deve seguire alcune semplici regole:

• la classe MDB deve implementare direttamente (mediante implements) o indirettamente (mediante annotazioni) l’interfaccia di un message listener
• la classe MDB deve essere concreta
• MDB deve essere un POJO e non una sottoclasse di un altro MDB
• la classe MDB deve essere dichiarata publica
• la classe MDB deve essere un costruttore senza argomenti
• non è possibile definire un metodo finalize
• occorre implementare tutti i metodi dell’interfaccia (questi devono essere publici e non possono essere static o final).

Implementazione del consumer mediante message-driven bean

Nel nostro esempio il consumer può essere implementato mediante un message-driven bean che preleva i messaggi dalla Queue e ne esegue i comandi indicati.

Una possibile implementazione del MDB potrebbe essere la seguente:

package miopackage;
   
import java.io.File;
import java.io.FileOutputStream;
import javax.ejb.ActivationConfigProperty;
import javax.ejb.MessageDriven;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
   
(activationConfig = { 
gProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
gProperty(propertyName = "destination", propertyValue = "queue/MiaCodaMessaggi"),
})
public class MioMDB implements MessageListener 
{
  public MioMDB() 
  {
  }
        
  public void onMessage(Message message) 
  {
    try
    {
      TextMessage messaggio = (TextMessage)message;
      
      if (messaggio.getText().equals(“comando1”)
      {
        ...
      }
      else
      {
        ...
      }
    }
    catch(Exception ecc)
    {
    }
  }
}

L’annotazione identifica l’oggetto come un MDB e ne specifica la configurazione (incluso il fatto che siamo in ascolto dalla queue MiaCodaMessaggi).
Il MDB implementa implementa l’interfaccia MessageListener e di conseguenza il metodo onMessage(Message message) che si occupa di elaborare i messaggi entranti: per ogni messaggio si estrae il comando mediante il metodo getText() e si svolge l’azione appropriata.

Uso dell'annotazione

I MDB sono uno dei tipi più semplici di EJB da sviluppare: le uniche annotazioni supportate sono e gProperty innestata al suo interno.

L’annotazione è definita da:

(TYPE)
(RUNTIME)
public  MessageDriven 
{
  String name() default "";
  Class messageListenerInterface default Object.class;
  ActivationConfigProperty[] activationConfig() default {};
  String mappedName();
  String description();
}

Tutti e tre gli argomenti dell’annotazione sono opzionali pertanto è possibile scrivere:

public class MioMDB

e definire i dettagli altrove (ad esempio in un Deployment Descriptor).
Il primo elemento name specifica il nome del MDB (se omesso il codice usa il nome della classe), il secondo parametro messageListenerInterface specifica quale message listener implementa l’MDB, l’ultimo parametro activationConfig è usato per specificare proprietà di configurazione.

Implementare il message listener

Un MDB implementa un’interfaccia message listener che il container usa per registrarlo presso il message provider e quindi passare i messaggi entranti invocando i metodi del message listener.
Per specificare il message listener si può usare il parametro messageListenerInterface dell’annotazione :

(name="NomeMDB",messageListenerInterface="javax.jms.MessageListener")
public class MioMDB {

In alternativa è possibile specificare l’interfaccia che si vuole implementare usando la parola chiave implements

public class MioMDB implements MessageListener {

Un’altra possibile opzione è quella di specificare l’interfaccia mediante il deployment descriptor.

ActivationConfigProperty

La proprietà activationConfig dell’annotazione consente di fornire informazioni di configurazione un array di istanze ActivationConfigProperty definite come segue:

public  ActivationConfigProperty 
{
  String propertyName();
  String propertyValue();
}

Ciascuna di queste è essenzialmente una coppia nome valore che il messaging provider sottostante usa per configurare il MDB.
Tre comuni proprietà di configurazione sono: destinationType, connectionFactoryJndiName e destinationName.

(name="NomeMDB",activationConfig = { 
gProperty( propertyName="destinationType", propertyValue="javax.jms.Queue"),
gProperty( propertyName="connectionFactoryJndiName", propertyValue="ConnectionFactory"),
gProperty( propertyName="destinationName", propertyValue="queue/MiaCodaMessaggi")
})

destinationType dice al container che l’MDB è in ascolto da una coda, connectionFactoryJndiName specifica il nome del connection factory che deve essere usato per creare la connessione per l’MDB e destinationName specifica la destinazione della quale devono arrivare i messaggi.

acknowledgeMode

I messaggi non vengono rimossi dalla coda finchè il consumer non ne invia l’ack.

Per impostare la modalità di acknowledge si usa:

gProperty(propertyName="acknowledgeMode",propertyValue="DUPS_OK_ACKNOWLEDGE")

dove propertyValue specifica la modalità da utilizzare.

subscriptionDurability

Se l’MDB è in ascolto su un topic è possibile specificare se la sottoscrizione è durable o nondurable.
In un dominio publish-subscribe, il messaggio viene distribuito a tutti i consumers sottoscritti: in generale chi non è connesso al topic ad un dato tempo non riceve la copia del messaggio.
L’eccezione a questa regola è conosciuta come durable subscription: una volta che il consumer ottiene una durable subscription al topic viene garantito che tutti i messaggi gli siano inviati.
Se il consumer non è connesso il MOM mantiene il messaggio finchè non si connette e quindi lo consegna.
Se si vuole che l’MDB sia un durable subscriber allora l’ActivationConfigProperty dovrebbe apparire nel seguente modo:

gProperty(propertyName="destinationType", propertyValue="javax.jms.Topic"),
gProperty(propertyName="subscriptionDurability", propertyValue="Durable")

Per sottoscrizioni non durable è sufficiente porre la proprietà subscriptionDurability a NonDurable (che è impostato di default).

messageSelector

La proprietà MessageSelector consente di applicare una selezione ai messaggi provenienti dalla destinazione secondo un certo criterio.
Per esempio se si vogliono ricevere solo i messaggi che hanno Originale pari a true si può scrivere:

gProperty(propertyName="messageSelector", propertyValue="Originale IS TRUE")