ΕΝΟΤΗΤΑ 3 APACHE CAMEL - EXCHANGE AND PROCESSOR

Στο σημερινό δωρεάν μάθημα Apache Camel θα εξηγήσουμε τι ακριβώς είναι το Exchange και τα Exchange patterns και πως χρησιμοποιούμε το Exchange σε συνδυασμό με τον Processor. Αυτές είναι δύο καινούργιες έννοιες που πρέπει να προσθέσουμε στην θεωρία μας για να κατανοήσουμε καλύτερα την ανταλλαγή μηνυμάτων όπως και την δυνατότητα παρέμβασης σε αυτά.

Όπως ήδη έχουμε αναλύσει από την προηγούμενη ενότητα, η επικοινωνία ανάμεσα στα components όπως και ανάμεσα σε διαφορετικά συστήματα γίνεται μέσω μηνυμάτων. Ένα τέτοιο μήνυμα στείλαμε και σε ένα εξωτερικό broker αφού πρώτα διαβάσαμε το περιεχόμενο από ένα text αρχείο.

Το Apache Camel δεν μεταφέρει άμεσα το μήνυμα. Ο κύριος λόγος είναι ότι ένα μήνυμα ρέει μόνο προς μια κατεύθυνση. Όταν ασχολούμαστε με μηνύματα, υπάρχουν πολλά μοτίβα ανταλλαγής μηνυμάτων (Message Exchange Patterns ή MEP) που μπορούμε να χρησιμοποιήσουμε.

Ανάλογα με τις περιπτώσεις χρήσης, μπορούμε να στείλουμε ένα μήνυμα χωρίς να περιμένουμε απάντηση από τον τελικό προορισμό – αυτό το μοτίβο ονομάζεται event message και χρησιμοποιεί το InOnly MEP. Για παράδειγμα, όταν διαβάζετε ένα αρχείο, όπως κάναμε στην προηγούμενη ενότητα, επεξεργαζόμαστε απλώς το περιεχόμενο του αρχείου χωρίς να περιμένουμε κάποια απάντηση πίσω από το File component που διαβάζει το αρχείο. Σε αυτή την περίπτωση, το component που είναι υπεύθυνο για την ανάγνωση του αρχείου θα καθορίσει ένα InOnly MEP.

Από την άλλη, μπορεί να θέλετε να εφαρμόσετε ένα μοτίβο αιτήματος-απάντησης όπου ένα μήνυμα απόκρισης πρέπει να επιστραφεί στον αποστολέα του μηνύματος. Σε αυτή την περίπτωση θα χρησιμοποιήσουμε ένα InOut MEP. Για παράδειγμα, σκεφτείτε ένα αίτημα σε ένα Web Service όπου ένα response θα πρέπει να επιστραφεί πίσω σε εκείνον που έκανε το αίτημα.

Αν θέλετε να βρείτε περισσότερες πληροφορίες για το message exchange μπορείτε να δείτε το Apache Camel documention:

https://camel.apache.org/manual/latest/exchange.html

https://camel.apache.org/manual/latest/exchange-pattern.html

Καθώς λοιπόν το μήνυμα ρέει προς μία μόνο κατεύθυνση, προκειμένου το Apache Camel να υποστηρίξει διαφορετικά MEPs, μας προσφέρει δύο μηνύματα:

- Το πρώτο μήνυμα είναι υποχρεωτικό καθώς είναι το in μήνυμα.

- Το δεύτερο μήνυμα είναι προαιρετικό (ανάλογα με το MEP) καθώς είναι το out μήνυμα.

Για αυτό ακριβώς το λόγο το Apache Camel τυλίγει όλα τα μηνύματα σε ένα Exchange αντικείμενο. Στην ουσία, το πραγματικό αντικείμενο που μεταφέρεται είναι το Exchange, ενεργώντας σαν container μηνυμάτων με όλα τα meta-data που απαιτούνται για την δρομολόγηση.

Βασικά μια ανταλλαγή μηνυμάτων περιέχει τα ακόλουθα:

- Exchange ID: Αναγνωριστικό ανταλλαγής ως String. Αυτό είναι ένα μοναδικό αναγνωριστικό που δημιουργεί αυτόματα το Apache Camel για εμάς.

- MEP: Το μοτίβο ανταλλαγής μηνυμάτων (MEP) καθορίζει το μοτίβο ανταλλαγής

- Exception: Αποθηκεύει την τρέχουσα αιτία μιας αποτυχημένης ανταλλαγής. Εάν παρουσιαστεί σφάλμα ανά πάσα στιγμή κατά τη δρομολόγηση, τότε θα αποθηκευτεί σε αυτό το πεδίο.

-Properties: Οι ιδιότητες είναι ένα Map που υπάρχουν κατά τη διάρκεια ολόκληρης της εκτέλεσης ανταλλαγής άσχετα αν το μήνυμα μπορεί να αλλάξει κατά τη δρομολόγηση. Αυτό είναι και η διαφορά τους με τα message headers που περιορίζονται μόνο στην διάρκεια ζωής του μηνύματος.

- In Message: Το in message είναι υποχρεωτικό και είναι πάντα ενεργοποιημένο. Είναι το μόνο μήνυμα που συμπληρώνεται στην ανταλλαγή με το InOnly MEP.

- Out Message: Το Out Message είναι προαιρετικό και χρησιμοποιείται μόνο με το InOut MEP.

Τώρα που γνωρίζουμε τι ακριβώς είναι Exchange αντικείμενο ας δούμε πως μπορούμε να το χρησιμοποιήσουμε μέσω ενός processor. Αλλά τι είναι ένας processor?

Ο processor είναι ένα βασικό στοιχείο του Apache Camel που αντιπροσωπεύει ένα node που μπορεί να χρησιμοποιήσει, να δημιουργήσει ή να τροποποίηση μια εισερχόμενη ανταλλαγή. Κατά τη διάρκεια εκτέλεσης ενός route, exchanges ρέουν από έναν processor σε έναν άλλο.

Ένα route ξεκινάει πρώτα από τον consumer που δέχεται ένα event (στο δικό μας παράδειγμα είναι ένα αρχείο) και δημιουργεί ένα Exchange. Το route μεταφέρει αυτό το Exchange από τον ένα processor στον άλλον έως το τελικό σημείο ή component. Είναι πιθανόν το route να τελειώνει σε έναν processor που επιστρέφει το Exchange πίσω στον consumer εάν έχει οριστεί σαν InOut ή σταματάει στο τελικό σημείο εάν οριστεί InOnly. Καταλαβαίνουμε λοιπόν ότι ένας processor, αφού λάβει το Exchange το τροποποιεί και τελικά το ενημερώνει ή το αναβαθμίζει με πληροφορίες. Στην πραγματικότητα όλα τα Exchange Integration Patterns (EIPS) του Apache Camel υλοποιούνται χρησιμοποιώντας processors.

Ο processor περιγράφεται από το org.apache.camel.Processor interface το οποίο περιέχει μια και μοναδική μέθοδο την process.

https://www.javadoc.io/doc/org.apache.camel/camel-api/latest/org/apache/camel/Processor.html

Για να υλοποιήσουμε έναν processor έχουμε δύο επιλογές σύμφωνα με το documentation. Μπορείτε να γράψουμε μια δική μας κλάση που κάνει implements το interface Processor

https://camel.apache.org/manual/latest/processor.html

Ή μπορούμε να χρησιμοποιήσουμε έναν processor στο route δημιουργώντας μια anonymous inner class.

https://camel.apache.org/manual/latest/processor.html

Στα δικά μας παραδείγματα θα χρησιμοποιήσουμε τον δεύτερο τρόπο. Ας δούμε λοιπόν πως μπορούμε να προσθέσουμε έναν processor στο Camel Route που είχαμε δημιουργήσει στην προηγούμενη ενότητα.

Στο παράδειγμα μας, στέλναμε ένα text μήνυμα σε ένα εξωτερικό broker που είχαμε λάβει διαβάζοντας ένα text αρχείο που βρισκόταν είτε μέσα στο φάκελλο testdata1 είτε μέσα στο φάκελλο testdata2. Τώρα θέλουμε, πριν στείλουμε το μήνυμα στο direct component, να παρέμβουμε και να προσθέσουμε επίσης ένα ενημερωτικό μήνυμα όπως “Message from testdata1” έτσι ώστε να γνωρίζουμε από που προήλθε το μήνυμα. Ο κώδικας λοιπόν θα είναι ο εξής:

MyRouteBuilder.java

import javax.inject.Inject;
import org.apache.activemq.artemis.jms.client.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.sjms.SjmsComponent;

public class MyRouteBuilder extends RouteBuilder {

    @Inject
    private CamelContext camel;

    SjmsComponent component = new SjmsComponent();

    @Override
    public void configure() {
        component.setConnectionFactory(new ActiveMQConnectionFactory("tcp://localhost:61616"));
        camel.addComponent("sjms", component);

        from("file:C:\\testdata1").
            process(new Processor() {
            public void process(Exchange exchange) throws Exception {
                String inBody = exchange.getIn().getBody(String.class);
                inBody = "Message from testdata1 : " + inBody;
                exchange.getIn().setBody(inBody);
            }
            }).
        to("direct:start");
        
        from("file:C:\\testdata2").process(new Processor() {
            public void process(Exchange exchange) throws Exception {
                String inBody = exchange.getIn().getBody(String.class);
                inBody = "Message from testdata2 : " + inBody;
                exchange.getIn().setBody(inBody);
            }
        }).to("direct:start");

        from("direct:start").to("sjms:queue:myqueue");
    }
}

		

Το POM.xml αρχείο παραμένει το ίδιο.

Το exchange αντικείμενο έχει αρκετές μεθόδους διαθέσιμες. Μια από αυτές είναι η getIn( ) η οποία μας επιστρέφει το μήνυμα.

https://www.javadoc.io/doc/org.apache.camel/camel-api/latest/org/apache/camel/Exchange.html

Μετά ορίζουμε ότι το μήνυμα θα περιέχει String πληροφορία. Τέλος, αφού προσθέσουμε την έξτρα πληροφορία που θέλουμε στο αρχικό μήνυμα, με την μέθοδο setBody( ) ορίζουμε ποιο θα είναι το νέο μήνυμα που θα σταλθεί στο direct component.

Αν βάλουμε ένα text αρχείο μέσα σε έναν από τους δύο φακέλλους, το μήνυμα που θα βρείτε μέσα στο broker θα περιέχει την αρχική πληροφορία συν την πληροφορία που έχουμε προσθέσει μέσω του processor.