Integrazione con Apache Camel

Apache Camel 2 è un Integration Framework, ovvero uno strumento che permette a a sistemi software complessi di scambiarsi messaggi o utilizzare servizi anche se essi sono implementati su ambienti, piattaforme o linguaggi diversi tra loro.

Perché questi questi servizi collaborino c'è bisogno di integrare un sistema di comunicazione che consenta lo scambio di messaggi che presentino:

• diverse forme (con determinati protocolli);
• diverse modalità (ad esempio sincrona o asincrona).

A risolvere questo problema (fornendo un contesto dove avviene l'interazione e una serie di servizi di 'appoggio') ci sono i cosiddetti ESB sia gratuiti (come Mule o Apache serviceMix), sia a pagamento (Bea Aqualogic Bus).

Pur non rientrando nella categoria di ESB, gli Integration Framework (come Apache Camel o Spring Integration Framework) rappresentano comunque strumenti efficaci nel risolvere il problema in questione.

Cosa differenzia Camel da un ESB?

Apache Camel non è un ESB poichè non ha un proprio Container dove viene eseguito ne offre tutta una serie di servizi che tipicamente un ESB offre eppure è altrettanto efficace per quanto riguarda il lavoro per il quale è stato realizzato che riguarda l'orchestrazione tra servizi, l'interazione attraverso lo scambio di messaggi , monitoring, creazione di componenti per la trasformazione dei messaggi ecc.

Fatto sta che Camel viene usato come strumento di routing da Apache serviceMix stesso (come motore interno per il routing).

Cosa realizzeremo

In questo articolo ci addentreremo tra le caratteristiche di Apache Camel con il pretesto di realizzare un esempio pratico, vediamo di che si tratta.

Scriveremo un'applicazione con Camel che prevede:

• un server FTP
• un server dove sarà installata una JMS queue (una coda di messaggi, secondo lo standard JMS) realizzata sfruttando Apache ActiveMQ

Con Apache Camel interfacceremo i due ambienti: trasporteremo il contenuto di un file
dal server FTP e ne scriveremo il contenuto nella coda JMS. Tutto in pochissime righe di codice.

Concetti fondamentali

Prima di addentrarci nell'implementazione è importante aver presente alcuni concetti. Anzitutto è importante sottolineare che Apache Camel ha un motore di routing e mediation engine. Routing significa 'instradamento' (dei messaggi) e Camel permette di definire routes in due forme:

• EIP (Enterprise Integration Patterns)
• DSL (Domain Specific Language)

EIP

L'EIP rappresenta una serie di pattern o linee guida da seguire per risolvere i più comuni problemi riguardo l'integrazione tra sistemi, in sostanza si tratta di soluzioni pronte all'uso e raccolte anche in un libro scritto da Gregor Hohpe e Bobby Woolf.

Camel utilizza la maggior parte di questi pattern per implementare parte della logica di integrazione attraverso il routing e in aggiunta utilizza anche un DSL.

DSL

Rappresenta un linguaggio specifico di un dominio. Java e C#, ad esempio, rappresentano linguaggi generici (non a dominio 'ristretto') perchè sono progettati per risolvere problemi di qualsiasi natura mentre un DSL è un linguaggio realizzato per risolvere un problema specifico.

Camel utilizza un suo DSL, un linguaggio per implementare la logica di routing, ovvero una serie di regole scritte con questo linguaggio che determina come i messaggi viaggiano, tra chi e in quale modalità.

Acnche se il DSL è un linguaggio 'neutro' col quale poter descrivere il tutto, esso non è indispensabile per usare Camel, poiché esso rappresenta un meta-Linguaggio. In sostanza possiamo utilizzare il nostro linguaggio preferito per programmare Camel.

Questo concetto non è del tutto nuovo. Ad esempio il vecchio CORBA prevedeva il DSL in maniera del tutto analoga a Camel (anche se CORBA era più un ESB che un framework di Integrazione).

I linguaggi che possiamo usare per descrivere un DSL in Camel sono diversi e vanno da Java a Scala a Spring DSL.

Vediamo degli esempi.

Java DSL:

from("file:data/inbox").to("jms:queue:order");

Spring DSL:

<route>
<from uri="file:data/inbox"/>
<to uri="jms:queue:order"/>
</route>

Scala DSL:

from "file:data/inbox" -> "jms:queue:order"

È piuttosto chiaro che le istruzioni indicano che qualcosa deve 'viaggiare' a partire da un file nella cartella data/inbox sullo stesso server ed essere scritto su una coda JMS order. Camel, indipendentemente dal linguaggio usato interpreta le tre istruzioni nei tre linguaggi diversi come una sola regola DSL: quella scritta con il suo linguaggio (quello neutro).

Usare il linguaggio che meglio conosciamo per scrivere il DSL significa avere supporto del proprio IDE preferito con tanto di vantaggi (vedi autocompletamento).

Le caratteristiche di Apache Camel

Riassumendo ancora le caratteristiche di Apache Camel:

• Motore di routing e mediation
• DSL
• Router Agnostico cioè Camel non si pone il problema sulla natura o tipologia del messaggio per lui il messaggio è un Object punto, è poi compito di chi riceve il messaggio sapere come trattarlo (vedi casting)
• Utilizzo di EIP
• Libreria di componenti estesa
• Meccanismo interno di casting implicito (automatico)
• Architettura modulare ed estendibile a plugin
• Configurazione molto semplice
• Non necessita di container e può essere installato in Container (anche web) di terze parti.

Vediamo ora velocemente quali entità esistono in Camel riguardo lo scambio dati.

Entità in gioco nello scambio dei messaggi in Camel

Messaggio

È un'entità che viaggia tra un Sender ed un Receiver (chi trasmette e chi riceve) ed è composto da tre parti:

Parte del messaggio	Descrizione
Header	Contiene metadati e informazioni sul messaggio, ad esempio chi l'ha inviato
Attachments	Dati aggiuntivi come file o immagini
Payload	Il corpo vero e proprio del messaggio

2. Exchange

Camel non utilizza direttamente un messaggio ma un contenitore che racchiude altri elementi ad esso strettamente correlati:

Campo	Descrizione
Exchange ID	Identificativo del messaggio
MEP	acronimo di Message Exchange Pattern che specifica se lo scambio è di tipo InOnly (cioè a "senso unico"): il messaggio viene inviato senza che ci si aspetti una risposta (tipico di una coda JMS) InOut (messaggio request-response): il sender invia un messaggio ed attende la risposta
InMessage	messaggio di ingresso
OutMessage	messaggio di uscita (solo per il MEP inOut)

Contesto di Camel

Finalmente possiamo riepilogare e integrare tutti i concetti che portano a definire ciò che con Camel chiamiamo Contesto (Context), in cui vivono e vengono offerti una serie di servizi ed entità:

Entità	Descrizione
DSL	Già menzionato
Messaggio	L'entità che viene scambiata
Exchange	Contenitore di tutto ciò che riguarda i dati scambiati
Processor	Nodo capace di modificare o creare un messaggio
Channel (o Canale)	Il 'condotto' o 'tubo' sul quale viaggia il messaggio
Endpoint	La fine del canale dove in sostanza il messaggio è diretto
Consumer	Il Receiver che utilizza il messaggio
Producer	Il Sender e cioè chi invia il messaggio
Component	La factory per creare un Endpoint ed un Producer per inviargli il messaggio

Un consumer infine può essere di tipo:

• Event-Driven : riceve il messaggio in modalità asincrona
• Polling-Consumer: riceve il messaggio in modalità sincrona (aspetta passivo che il messaggio arrivi)

Ecco un tipico schema di esempio, in cui un web service che funge da sender invia informazioni ad un processore (translator) che traduce il messaggio e lo invia tramite email, dopo averlo elaborato, sia ai destinatari, sia a un server per il backup:

Dopo questa panoramica e l'introduzine di alcuni elementi fondamentali non resta che iniziare a realizzare il nostro esempio.

Installare il server FTP

Il primo passo che andremo ora a fare è l'installazione del server FTP, prima di fare questo ho innanzitutto associato ad uno dei miei server un IP Statico LOCALE :

Quindi selezionando 'risorse di rete' e successivamente col tasto destro del mouse 'proprietà' si imposta in 'Tcp/Ip' la risoluzione MANUALE dell'IP, senza quindi l'utilizzo del DHCP (il computer/software che assegna gli indirizzi in automatico).

Per chi fosse interessato a questi concetti (NAT, Router, DNS; DHCP) può far riferimento alle sezioni di competenza o sul web.

Sia chiaro che non è un passo essenziale ma questo permette di fare riferimento poi al server ftp sempre con lo stesso IP locale.

Per testare il funzionamento dovrete disporre di un altro pc sul quale c'è la STESSA connessione web (ad esempio due pc collegati in wifi o cavi tramite lo stesso router) , lanciare sul pc dove NON è installato il server FTP e lanciare il comando 'cmd' per aprire su windows la dos shell dove digitando:

> ping 192.168.1.102

dovresete riceve conferma che i due pc comunicano tra di loro:

Ovvio che gli smanettoni possono utilizzare un IP Statico pubblico e collegarsi da un'altra rete per verificare il tutto in maniera 'completa' ma lasciamo questa attività a loro..

Fatto questo scarichiamo FTP Filezilla Server, lanciamo l'eseguibile nell'archivio ed installiamo il server in una cartella a scelta, vi verrà richiesto di definire la porta di ascolto del server per l'amministrazione , lasciamo quella standard che è la 14147.

A fine installazione apriamo il pannello di amministrazione:

In Edit->Settings potete personalizzare tutti i parametri del server, quali ad esempio la porta su cui porre il demone in attesa (di default è porta la 21), se ci sono , ip da bloccare, opzioni di logging ed eventuali limiti di velocità.

Dobbiamo quindi creare almeno un utente con password, io ho creato un utente con nome 'brutocaterano' (un personaggio di Magic the gathering) e password 'asterix111'.

Passiamo quindi a creare una cartella condivisa con i relativi permessi: I permessi ovviamente sono strettamente legati al ruolo dell'utente , se amministratore potrà avere accesso a tutte le funzionalità ed operazioni mentre per un utente 'guest' (visitatore) non dovrebbe essere permesso di cancellare file o directories (o rinominarle).

Per semplicità abbiamo creato una sola direcotry , quella Home e che rappresenta quella che di default viene 'aperta' all'atto del collegamento.

Filezilla server permette di definire più cartelle con relativi Alias associati (un alias è un nome fittizio) ma non complichiamo le cose.

Testiamo se il tutto funziona installando sull'altro pc filezilla client. Apriamolo e inseriamo i parametri di connessione all'altro pc:

Nel caso in cui non dovesse funzionare impostate la modalità di trasferimento 'passiva':

Colleghiamoci quindi al server:

Sul server ftp stesso, creiamo una cartella 'incoming', che ci servirà successivamente.

Installare ActiveMQ

È il momento di configurare activeMq, ma non senza ricordare brevemente di cosa si tratti. Ecco la definizione di Wikipedia:

Apache ActiveMQ is an open source message broker which fully implements the Java Message Service 1.1 (JMS).
It provides "Enterprise Features"[1] like clustering, multiple message stores[clarify], and ability to use any database as a JMS persistence provider besides virtual memory, cache, and journal persistency.
Apart from Java, ActiveMQ can also be used from .NET,[2] C/C++[3] or Delphi[4] or from scripting languages like Perl, Python, PHP and Ruby via various "Cross Language Clients"[4] together with connecting to many protocols and platforms.[5] These include several standard wire-level protocols , plus their own protocol called OpenWire.
ActiveMQ is used in enterprise service bus implementations such as Apache ServiceMix, Apache Camel, and Mule.
ActiveMQ is often used with Apache ServiceMix, Apache Camel and Apache CXF in SOA infrastructure projects.[6]

ActiveMQ quindi è più che una semplice 'implementazione' delle specifiche JMS, infatti prevede funzionalità per l'interoperabilità tra ambienti eterogenei che lo accomunano ad Apache Camel. Nel nostro esempio però consideriamola come coda JMS da utilizzare per interfacciare Camel.

Scarichiamo ActiveMQ, apriamo l'archivio ed installiamo il contenuto in una directory specifica:

Ci posizioniamo sotto la cartella 'bin' di installazione e digitiamo il comando 'activemq' che farà partire activeMQ:

Infatti il comando lancerà il servizio sulla porta 8161, prima di entrarci richiederà user e password, inserite per entrambe 'admin' e verrà caricata la seguente schermata:

Installiamo ActiveMQ Senza entrare troppo nel merito della spiegazione di Jms diciamo che esistono due entità da poter utilizzare per inviare e/o ricevere dati (due modelli) Point-to-Point e Publish/Subscribe in sostanza nel primo caso un messaggio verrà scritto sulla coda e un listener una volta scritto il messaggio lo leggerà 'distruggendolo' (non sarà più contenuto nella queue) mentre il modello Publish/Subscribe funziona per il broadcasting, il messaggio viene inviato a tutti i listener che si registrano come ascoltatori sulla coda (in questo caso si chiamerà Topic).

I concetti legati a jms sono tanti e riguardano l'affidabilità (message reliability) l'asincronicità ecc.

Ma ne discuteremo in separata sede, per ora quello che ci interessa è semplicemente avere una coda jms sulla quale scrivere un messaggio tramite Camel.

La pagina del browser infine visualizza lo stato della coda tra cui i messaggi inviati, da elaborare, numero di consumers (chi è in ascolto sulla coda per prelevare il/i messaggi) ecc:

Ritorniamo ad Apache Camel

Apriamo MyEclipse, la prima cosa da fare è scaricare le librerie necessarie, possiamo farlo attraverso Maven ecco il file pom.xml nel sorgente:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.camel</groupId>
<artifactId>examples</artifactId>
<version>2.11.0</version>
</parent>
<artifactId>camel-example-jms-file</artifactId>
<name>Camel :: Example :: JMS-File</name>
<description>An example that persists messages from FTP site to JMS</description>
<dependencies>
<!-- Camel dependencies -->
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-core</artifactId>
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-jms</artifactId>
</dependency>
<!-- ActiveMQ dependencies -->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-broker</artifactId>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-client</artifactId>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-camel</artifactId>
</dependency>
<dependency>
<groupId>org.apache.xbean</groupId>
<artifactId>xbean-spring</artifactId>
<exclusions>
<exclusion>
<groupId>org.springframework</groupId>
<artifactId>spring</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
<profiles>
<profile>
<id>Example</id>
<properties>
<target.main.class>com.ftpToJms.FtpToJMSExample</target.main.class>
</properties>
</profile>
</profiles>
</project>

Poiché Maven copia le librerie da scaricare in una cartella temporanea sull'hard disk possiamo
tranquillamente 'prelevare' le librerie da quella cartella:

Grazie ad un plugin, Maven permette di scaricare le librerie direttamente all'interno del progetto ma per ora ci serve individuare per benele librerie necessarie quindi operiamo nel modo descritto.

Nota: se Maven è stato lanciato in precedenza troverete nel vostro Local
Repository delle librerie aggiuntive quindi per il vostro esempio utilizzate Maven.

Ora create un Java Project 'FtpToJms' e create la classe seguente nel relativo package:

package com.ftpToJms;
import javax.jms.ConnectionFactory;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.jms.JmsComponent;
import org.apache.camel.impl.DefaultCamelContext;
public class FtpToJMSExample
{
private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;
private static String ftpLocation = "ftp://192.168.1.102/incoming?username=brutocaterano&password=asterix111";
public void start() throws Exception
{
CamelContext context = new DefaultCamelContext();
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
context.addComponent("jms", JmsComponent.jmsComponentAutoAcknowledge(connectionFactory));
context.addRoutes(
new RouteBuilder() {
public void configure()
{
from(ftpLocation).to("jms:MYTESTQUEUE");
}
});
System.out.println("start context!");
context.start();
System.out.println("wait");
System.out.println(loaded);
while (loaded!=true) { System.out.println("in attesan"); }
context.stop();
System.out.println("stop context!");
}
public static void main(String args[]) throws Exception
{
FtpToJMSExample j = new FtpToJMSExample();
j.start();
}
}

Analizziamo il codice:

Innanzitutto la classe è un POJO, non estende nessuna classe ne implementa interfacce, mentre:

1. Url punta al broker (letteralmente 'intermediario') , il valore della variabile
DEFAULT_BROKER_URL è 'tcp://127.0.0.1:61616'

private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;

Notiamo che non si tratta della porta di ascolto del servizio di amministrazione web di activeMQ (in quel caso l'url è http://localhost:8161/admin/queues.jsp) Quindi il valore di url permette di interfacciarsi al broker di activeMQ.

2. FtpLocation rappresenta l'url (protocollo ftp) che punta al contenuto della directory 'incoming' (creata in precedenza ricordate?) Fornendo nell'URI user e password.

private static String ftpLocation = "ftp://192.168.1.102/incoming?username=brutocaterano&password=asterix111";

3. Istanziamo il Context di Camel e la Connectionfactory per creare la connectione ad ActiveMQ
utilizzando la variabile url.

CamelContext context = new DefaultCamelContext();
ConnectionFactory connectionFactory = new
ActiveMQConnectionFactory(url);

4. Questo è un punto critico: aggiungiamo un COMPONENTE di Camel che si occupa di creare il produce e l'endPoint verso una coda JMS.

context.addComponent("jms", JmsComponent.jmsComponentAutoAcknowledge(connectionFactory));

Questo componente è presente già in Apache Camel, oltretutto riceve una factory più precisamente un qualunque oggetto che implementa l'interfaccia 'javax.jms.ConnectionFactory;', i più smaliziati avranno capito che questo significa potergli passare qualsiasi factory di qualsiasi provider Jms, a patto che questi implementi le specifiche di jms.

Infine con 'AutoAcknowledge' il componente in sostanza riconosce in automatico che il messaggio è stato ricevuto e activeMQ lo rimuove non appena il receiver appunto lo riceve.

5. Il codice:

context.addRoutes(new RouteBuilder() {
public void configure() {
from(ftpLocation).to("jms:queue:MYTESTQUEUE");
}
});

rappresenta un'altra sezione critica, qui inseriamo nel context di Camel un RouteBuilder come classe anonima, questa classe ha un metodo 'configure()' che in sostanza permette di definire il DSL, infatti quello che gli abbiamo detto è di creare un routes (instradamento) da (from) ftpLocation alla (to) coda 'MYTESTQUEUE'.

Attenzione che se non viene trovata una queue con questo nome dal broker semplicemente la creerà. La parte restante del codice:

System.out.println("start context!");
context.start();
System.out.println("wait");
System.out.println(loaded);
Thread.sleep(3000);
context.stop();
System.out.println("stop context!");

Avvia il context, attende 3 secondi necessari per prelevare il file e scrivere il suo contenuto sulla coda jms.

Creiamo ora un file 'info.txt' inserendo la stringa 'Camel Message':

Proviamo ad eseguire la classe (ha il metodo main) senza dimenticare di avere in funzione il server ftp e la coda ActiveMQ sull'altro:

E nel browser:

L'output di MyEclipse ha però introdotto un 'problema' e cioè che il file è stato prelevato più volte (esattamente tre) dal server ftp, in sostanza è stato fatto un polling cioè una lettura ciclica temporale (3 secondi) prima di terminare la lettura dal sever ftp, vedremo dopo come si può risolvere questa situazione per ora andiamo avanti.

Clicchiamo sul pulsante 'browse':

E nella schermata successiva:

Su uno dei messageID:

Apparirà la seguente schermata che fornisce tutta una serie di informazioni:

Il passo successivo è di inserire un Processore, che ci permetterà di creare e modificare un messaggio. Vediamo prima un semplice esempio, inseriamo un processore che stampa il nome del file, inseriamo il seguente metodo nella nostra classe:

private Processor executeFirstProcessor()
{
return new Processor() {
@Override
public void process(Exchange exchange)
{
System.out.println("We just downloaded : "+ exchange.getIn().getHeader("CamelFileName"))
}
};
}

Il metodo restituisce un oggetto 'org.apache.camel.Processor' e nel metodo creaiamo in return una classe anonima implementanto tramite overriding il metodo process(Exchange).

Ovviamente tramite exchange possiamo accedere sia al Message di input e di output sia all'header. Proprio quest'ultimo ci permette di recuperare la properties 'camelFileName' dove nel nostro caso sarà il nome del file prelevato via ftp.

Ora quello che resta da fare è inserire il processore nel flusso di elaborazione del messaggio. Lo facciamo ovviamente via DSL. Ecco come:

Questo l'output:

In pratica cotruiremo una 'pipeline' di entità costituita da Consumer, Producer, Processor
etc. Che creano/elaborano/modificano il messaggio.

Velocemente aggiungiamo alla nostra classe la seguente variabile:

private boolean loaded = false;

E al nostro processore:

Infine sostituiamo l'istruzione Thread.sleep(3000); con la seguente:

while (loaded!=true) {
System.out.println("in attesan");
}

Ora rilanciamo l'applicazione:

E nel browser, il file sarà caricato una sola volta:

Supponiamo ora di voler prelevare il contenuto del file e trasformarlo in XML prima di scriverlo sulla queue come fare? Aggiungiamo un altro processore, aggiungiamo alla nostra classe prima il metodo seguente per trasformare una semplice stringa in XML:

private String buildXmlFilecontent(String message)
{
return "<?xml version="1.0" encoding="UTF-8"?><messaggio>" + message + "</messaggio>";
}

Questo invece il codice per aggiungere il nuovo processore:

private Processor processFileContentInXml()
{
return new Processor()
{
@Override
public void process(Exchange exchange) throws Exception
{
try {
RemoteFile rf = (RemoteFile) exchange.getIn().getBody();
System.out.println("msg:" + rf.getBody());
exchange.getOut().setBody(buildXmlFilecontent(rf.getBody().toString()));
} catch (Exception ex) {
System.out.println(ex.getMessage());
}
}
};
}

Il metodo in sostanza ricava il file come oggetto (Camel) RemoteFile e ne preleva il contenuto quindi lo trasforma in xml e ne scrive il risultato nell'OutputMessage.

Ovviamente va aggiunto alla pipeline:

L'ordine dei processori può essere cambiato secondo le necessità e possibilità.

Conclusioni

Apache Camel viene ampiamente utilizzato per l'integrazione, molti motori ESB si basano su Camel (come Apache serviceMix), lo stesso Camel viene spesso indicato come ESB Light Container.

Questo articolo è utile a capire la potenza di questo strumento, basti pensare a come realizzare quanto implementato utilizzando solo Java o il proprio linguaggio di programmazione.