FUSE - MokaByte n. 134 - Ottobre 2008

A fronte della necessità di integrare applicazioni rapidamente, molti sviluppatori preferiscono un approccio "code-first", senza doversi preoccupare dei dettagli specifici legati a piattaforma o dominio. Con FUSE Mediation Router uno sviluppatore Java può implementare rapidamente pattern di integrazione, tramite semplici API direttamente all'interno di un programma Java. Questi pattern di integrazione possono poi essere distribuiti in diverse piattaforme, fra cui FUSE ESB, fornendo un percorso di migrazione verso un'infrastruttura SOA.

L'approccio alternativo è quello di invocare il processor recipientList(), che ha come argomento una lista di destinatari (recipent list dinamica). Il vantaggio derivante dall'utilizzo del processor recipientList() è che la lista di destinatari può essere valorizzata a runtime: ad esempio, la regola seguente genera una lista di destinatari leggendo il valore di recipientListHeader dal messaggio ricevuto:

from("SourceURL").recipientList(header("recipientListHeader").tokenize(","));

Aggregator Il processor aggregator() è utilizzato per aggregare messaggi correlati in un singolo messaggio, come illustrato nella Figura 1.

Allo scopo di distinguere i messaggi che devono essere aggregati insieme, deve essere definita una correlation key per l'aggregator(): i messaggi che hanno la stessa correlation key devono essere aggregati. Solitamente, la correlation key deriva da uno dei campi del messaggio, ad esempio un campo dell'header. È comunque possibile definire un algoritmo di aggregazione per il processor aggregator(): l'algoritmo di default è di scegliere l'ultimo messaggio con un dato valore per la correlation key e di scartare i messaggi più vecchi con quel valore di correlation key. Ad esempio, nel caso si stia monitorando uno stream di dati che fornisce i prezzi delle azioni in tempo reale, potremmo essere interessati solo all'ultimo prezzo per ogni azione. In questo caso è possibile configurare un aggregator() che trasmetta solo l'ultimo prezzo di un'azione e scarti quelli più vecchi. Questa funzionalità è implementata dalla seguente regola, dove la correlation key è presa dall'header stockSymbol e viene utilizzato l'algoritmo di default:

from("SourceURL").aggregator(header("stockSymbol")).to("TargetURL");

Allo scopo di combinare tutti i messaggi di exchange, relativi ad una specifica correlation key, in un singolo messaggio exchange, si utilizza il processor AggregationStrategy().

Multicast Il processor multicast() consente di eseguire il routing dello stesso messaggio verso diversi endpoint e di processarli in maniera differente. Il seguente esempio mostra come prendere una richiesta dall'endpoint direct:a e di eseguire il multicast verso direct:x, direct:y e direct:z.

from("direct:a").multicast().to("direct:x", "direct:y", "direct:z");

Implementazione del service provider dell'agente di prestito Come anticipato, l'esempio descritto in questo articolo prevede l'esposizione di interfacce SOAP/HTTP da parte dei service provider dell'agente di prestito e dell'agenzia di credito e di interfacce SOAP/JMS da parte dei service provider delle banche .Il sequence diagram UML corrispondente è illustrato nella seguente figura.

Dal punto di vista progettuale, si tratta di implementare seguente workflow illustrato nella seguente figura, realizzato con le componenti descritte da Gregor Hohpe e Bobby Woolf.

Per prima cosa definiamo l'interfaccia che il service provider dell'agente di prestito esporrà. L'operazione da esporre è getLoanQuote e restituisce una String contenente le informazioni sul valore del rateo minore e sulla banca che lo applica in base ai seguenti parametri: Social Security Number dell'utente (String ssn), quantità del prestito richiesto (double loanAmount), durata del prestito (int loanDuration). Si tratta degli stessi parametri richiesti dall'operazione getQuote esposta dalle banche, a meno dei parametri (int creditHistory e int creditScore) forniti dall'agenzia di credito e la cui valorizzazione è quindi inclusa nel workflow dell'agente di prestito e derivante dall'invocazione dell'agenzia di credito. L'interfaccia del service provider dell'agente di prestito pertanto sarà:

import javax.jws.WebService;

@WebService
public interface LoanBrokerWS {
String getLoanQuote(String ssn, Double loanAmount, Integer loanDuriation);
}

Prima di vedere come implementare il codice con FUSE Mediation Router, è importante sapere che org.apache.camel.builder.RouteBuilder è la classe base per la definizione della regola di routing. Questa classe definisce il metodo astratto configure() che deve essere specializzato nella classe derivata. Inoltre, RouteBuilder definisce anche metodi utilizzati per iniziare le regole di routing, come from(), intercept(), and exception(). È possibile definire quante classi RouteBuilder sono necessarie, l'importante è instanziare ognuna di queste con l'oggetto CamelContext. La classe LoanBroker che implementa l'agente di prestito sarà quindi derivata da RouteBuilder:

public class LoanBroker extends RouteBuilder {
}

all'interno del main sarà instanziato l'oggetto CamelContext

public static void main(String... args) throws Exception {
System.setProperty("java.util.logging.config.file","logging.properties");
System.out.println("Starting...");
CamelContext context = new DefaultCamelContext();
context.addRoutes(new LoanBroker());
context.start();
System.out.println("...Ready!");
Thread.sleep(Long.MAX_VALUE);
context.stop();
}

e la regola di routing si troverà all'interno del metodo configure()

public void configure() {
/**
- BANK = JMS
- CREDIT AGENCY = WS
- MULTICAST = PARALLEL
*/
from(Constants.PARALLEL_LOANBROKER_WS_BANK_JMS_CREDITAGENCY_WS_URI)
.process(new CreditScoreProcessor(Constants.CREDITAGENCY_WS_ADDRESS))
.intercept(new Transform())
.multicast(new BankResponseAggregationStrategy(),true)
.to(Constants.BANK1_JMS_URI,Constants.BANK2_JMS_URI,Constants.BANK3_JMS_URI);
}

Analizziamo i singoli processor che compongono la regola di routing. Il processo from(), dopo aver inizializzato l'endpoint, intercetta il messaggio inviato dal client all'endpoint FUSE Services Framework definito dall'URI

cxf://http://localhost:9008/parallelLoanBrokerWithBankJms?serviceClass
=com.iona.fuse.mokabyte.articolo2.broker.LoanBrokerWS

il cui WSDL esposto può essere visualizzato lanciando dal proprio browser il seguente URL:

http://localhost:9008/parallelLoanBrokerWithBankJms?wsdl

Il secondo componente della regola di routing è process() che crea un'istanza del processor CreditScoreProcessor() prendendo come argomento l'URI

http://localhost:9006/creditAgency

che identifica il service provider dell'agenzia di credito che fornisce i dati aggiuntivi necessari all'invocazione delle 3 banche.

Il processor CreditScoreProcessor() si occupa dell'enrichment dei dati forniti dall'utente estraendo il valore ssn dall'exchange in input e utilizzandolo per invocare le operazioni getCreditScore() e getCreditHistoryLength() esposte dall'agenzia di credito, come descritto di seguito:

public void process(Exchange exchange) throws Exception {
System.out.println("\n[CreditScoreProcessor] process has been invoked!!!");
Message requestMessage = exchange.getIn();
List<Object> request = (List<Object>) requestMessage.getBody();

String ssn = (String)request.get(0);

int historyLength = proxy.getCreditHistoryLength(ssn);
System.out.println("   Credit Agency getCreditHistoryLength Response for ssn:"
+ ssn + " -> historyLength:" + historyLength);
int score = proxy.getCreditScore(ssn);
System.out.println("   Credit Agency getCreditScore Response for ssn:"
+ ssn + " -> score:" + score);

exchange.getOut().setBody(exchange.getIn().getBody());

exchange.setProperty(Transform.HISTORY_LENGTH, historyLength);
exchange.setProperty(Transform.SCORE, score);
}

I valori ottenuti vengono poi inseriti come property direttamente all'interno dell'exchange.

Il successivo componente della regole di routing è il processor intercept() che crea un'istanza del processor Transform() il quale interviene due volte durante il dispatching di una singola richiesta. Nel primo caso è responsabile di adattare richiesta contenuta nell'exchange proveniente da CreditScoreProcessor() per inviarlo ai service provider delle banche, mentre nel secondo caso di adattare la risposta dei service provider delle banche in quella da restituire all'utente.

Il codice chiave di questo processor è riportato di seguito:

int historyLength = (int)exchange.getProperty(HISTORY_LENGTH, Integer.class);
int score = (int)exchange.getProperty(SCORE, Integer.class);

Message requestMessage = exchange.getIn();
List<Object> request = (List<Object>) requestMessage.getBody();

String ssn = (String)request.get(0);
Double amount = (Double) request.get(1);
Integer loanDuriation = (Integer)request.get(2);

List<Object> bankRequest = new ArrayList<Object>();
bankRequest.add(ssn);
bankRequest.add(amount);
bankRequest.add(loanDuriation);
bankRequest.add(historyLength);
bankRequest.add(score);

exchange.getIn().setBody(bankRequest);
exchange.getIn().setHeader(CxfConstants.OPERATION_NAME, "getQuote");

processNext(exchange);

Object[] bankResponse = (Object[]) exchange.getOut().getBody();
BankQuote newQuote = (BankQuote) bankResponse[0];

List<Object> brokerRequest = new ArrayList<Object>();
brokerRequest
.add("[ssn:" + newQuote.getSsn() + ", Bank:"
+ newQuote.getBankName() + ", Rate:"
+ newQuote.getRate() + "]");
exchange.getOut().setBody(brokerRequest);

Da notare come i due interventi di trasformazione del messaggio verso e da i service provider delle banche siano separati dall'istruzione processNext(exchange).

Il successivo componente della regole di routing è il processor multicast() necessario al processamento dei risultati derivanti dall'invocazione dei service provider delle banche. il processor multicast() crea un'istanza della classe BankResponseAggregationStrategy(), definita come estensione del processor AggregationStrategy() e che, come visto, serve a combinare tutti i messaggi di exchange relativi ad una specifica correlation key, in un singolo messaggio exchange.

La specializzazione del metodo aggregate() viene richiamato da FUSE Mediation Router a partire dalla seconda risposta di una delle banche ricevuta dallo stadio multicast. Gli argomenti che vengono passati al metodo sono due exchange (oldExchange, newExchange) contenenti le risposte dei service provider delle prime due banche che rispondono. Ricordiamo a questo proposito che, nel caso in cui il secondo argomento del processor() multi cast sia true, la comunicazione con i service provider delle banche è asincrona, pertanto l'ordine di ricezione delle risposte è indipendente dall'ordine di invocazione.

Come si nota dal codice mostrato di seguito, viene fatta una prima comparazione tra i valori di rateo forniti dalle banche invocate e viene restituito l'exchange contenente il rateo più basso. A questo punto, quando il processor multicast() riceve la risposta dell'ultima banca, viene richiamato nuovamente il metodo aggregate passando l'exchange calcolato precedentemente (oldExchange) e l'exchange con la risposta appena ricevuta (newExchange).

Object[] oldResult = (Object[])oldExchange.getOut().getBody();
BankQuote oldQuote = (BankQuote) oldResult[0];

Object[] newResult = (Object[]) newExchange.getOut().getBody();
BankQuote newQuote = (BankQuote) newResult[0];

Exchange result = null;

if (newQuote.getRate() >= oldQuote.getRate()) {
result = oldExchange;
} else {
result = newExchange;
}
return result;

L'ultimo processor della regola di routing è to() che specifica i destinatari del messaggio, in questo caso gli URI:

cxf://jms://?serviceClass=com.iona.fuse.mokabyte.articolo2.bank.BankWS&portName= {http://bank.articolo2.mokabyte.fuse.iona.com/}Bank1&dataFormat=POJO

cxf://jms://?serviceClass=com.iona.fuse.mokabyte.articolo2.bank.BankWS&portName= {http://bank.articolo2.mokabyte.fuse.iona.com/}Bank2&dataFormat=POJO

cxf://jms://?serviceClass=com.iona.fuse.mokabyte.articolo2.bank.BankWS&portName= {http://bank.articolo2.mokabyte.fuse.iona.com/}Bank3&dataFormat=POJO

Esecuzione del processo completo Di seguito viene descritta l'esecuzione dell'intero processo utilizzando il codice disponibile nel file articolo2.zip e distribuito all'URL HYPERLINK "http://www.iona.com/FUSEEIP/articolo2.zip" www.iona.com/FUSEEIP/articolo_2.zip.

Dopo aver eseguito l'installazione di Maven, è utile preparare uno script per inizializzare le variabili d'ambiente. Per gli utenti Windows le seguenti istruzioni del file mavenenv.bat contenuto nel file articolo1.zip dovranno essere modificate in base alle directory di installazione del JDK e di Maven:

set JAVAHOME=<directory di installazione del JDK> set M2HOME=<directory di installazione di maven> set PATH=%M2HOME%\bin;%JAVAHOME%\bin;%PATH%

Per gli utenti Linux le seguenti istruzioni del file mavenenv.sh contenuto nel file articolo1.zip dovranno essere modificate in base alle directory di installazione del JDK e di Maven:

export JAVAHOME=<directory di installazione del JDK> export M2HOME=<directory di installazione di maven> export PATH=$JAVAHOME/bin:$M2HOME/bin:$PATH

Per modificare i permessi di esecuzione del file dovrà essere eseguita la seguente istruzione:

chmod u+x maven_env.sh

e per inizializzare le variabili d'ambiente dovrà essere eseguita la seguente istruzione:

. ./maven_env.sh

Prima di lanciare i servizi, è comunque sempre consigliabile ricompilare tutto con il comando:

mvn compile

Per lanciare il broker JMS è sufficiente posizionarsi sulla directory articolo2, aprire la shell con il file batch maven_env.bat ed eseguire il comando:

mvn exec:java -Dexec.mainClass=com.iona.fuse.mokabyte.articolo2.EmbeddedBroker