ΕΝΟΤΗΤΑ 5 APACHE CAMEL - CONTENT-BASED ROUTER

Στο σημερινό δωρεάν μάθημα Apache Camel θα δούμε πως να ενσωματώσουμε ένα content-based router σε ένα Apache Camel route.

Πριν ξεκινήσουμε να δούμε τον κώδικα, θα πρέπει να αναφερθούμε στο γεγονός ότι ο content-based router είναι μέλος των messaging patterns όπως περιγράφεται στην Enterprise Integration Patterns ιστοσελίδα.

https://www.enterpriseintegrationpatterns.com/patterns/messaging/ContentBasedRouter.html

Μπορούμε να βρούμε και το αντίστοιχο Apache Camel documentation εδώ:

https://camel.apache.org/components/latest/eips/content-based-router-eip.html

Το Content-Based Router εξετάζει το περιεχόμενο του μηνύματος και δρομολογεί το μήνυμα σε εκείνο το κανάλι που δέχεται μόνο το συγκεκριμένο είδος μηνυμάτων. Η δρομολόγηση μπορεί να βασίζεται σε μια σειρά κριτηρίων όπως συγκεκριμένες τιμές πεδίων, συγκεκριμένες λέξεις που περιέχονται στο μήνυμα κτλ.

Στο δικό μας παράδειγμα, το οποίο ακολουθεί την ροή του διαγράμματος από την Enterprise Integration Patterns ιστοσελίδα, θέλουμε να φιλτράρουμε την εισερχόμενη παραγγελία και με βάση το περιεχόμενο του μηνύματος, να την στείλουμε είτε σε ένα queue που είναι για widget παραγγελίες είτε σε ένα queue για gadget παραγγελίες.

Θα δημιουργήσουμε λοιπόν ένα JAX-RS web service όπου θα δέχεται μια απλή παραγγελία σε JSON μορφή. Η παραγγελία περιγράφεται από μια απλή κλάση με το όνομα Order. Όταν το web service λάβει την παραγγελία, θα την στείλει στο direct:start component έτσι ώστε να ενεργοποιηθεί το route. Ας δούμε πρώτα όμως τον κώδικα του web service πριν προχωρήσουμε στο Content-Based Router.

Όπως ήδη αναφέραμε θα πρέπει να δημιουργήσουμε μια κλάση η οποία θα περιγράφει την παραγγελία. Για να κρατήσουμε τον κώδικα όσο πιο απλό γίνεται, την κάθε παραγγελία θα την περιγράφουμε με δύο μόνο μεταβλητές – name και amount.

Order.java

package com.example;


public class Order {

	private String name;
	private int amount;

	public String getName() {
	return name;
	}

	public void setName(String name) {
	this.name = name;
	}

	public int getAmount() {
	return amount;
	}

	public void setAmount(int amount) {
	this.amount = amount;
	}

	public Order() {
	}

	public Order(String name, int amount) {
	this.name = name;
	this.amount = amount;
	}
}

Για να γίνει bootstrap ένα web service, δηλαδή για να ενεργοποιηθεί αυτόματα και να είναι έτοιμο προς χρήση αμέσως μετά από την εκκίνηση του application server, θα πρέπει πρώτα να δημιουργήσουμε μια κλάση η οποία θα κληρονομεί από την Application.

JAXRSApplication.java

package com.example;

import javax.ws.rs.ApplicationPath;
import javax.ws.rs.core.Application;

@ApplicationPath("resources")
public class JAXRSApplication extends Application {
}

Τώρα δημιουργούμε το web service, που όπως είπαμε είναι πολύ απλό και έχει σαν κύριο σκοπό εκτός από το να είναι το κεντρικό σημείο υποδοχής του μηνύματος της παραγγελία, να στέλνει το μήνυμα και σε ένα Apache Camel route. Αυτό το καταφέρνουμε με το αντικείμενο CamelContext και την μέθοδο createProducerTemplate. To component στο οποίο στέλνουμε την πληροφορία είναι το direct:start.

OrderResource.java

package com.example;

import javax.enterprise.context.ApplicationScoped;
import javax.json.bind.Jsonb;
import javax.json.bind.JsonbBuilder;
import javax.ws.rs.*;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.inject.Inject;
import org.apache.camel.CamelContext;
import org.apache.camel.ProducerTemplate;

@ApplicationScoped
@Path("/orders")
public class OrderResource {

	String customerString;


	@Inject
	private CamelContext camelctx;

	@POST
	@Consumes(MediaType.APPLICATION_JSON)
	public Response createOrder(String input) {

	ProducerTemplate producer = camelctx.createProducerTemplate();
	producer.sendBody("direct:start", input);

	Jsonb jsonb = JsonbBuilder.create();
	Order order = new Order();
	order = jsonb.fromJson(input, Order.class);

	System.out.println(order.toString());
	return Response.ok(order).build();

	}
}

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

MyRouteBuilder.java

package com.example;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;

public class MyRouteBuilder extends RouteBuilder {


		@Override
		public void configure() {


		from("direct:start").to("vm:incomingOrders");

		from("vm:incomingOrders")
		.choice()
		.when(body().contains("widget"))
		.to("vm:widgetOrders")
		.when(body().contains("gadget"))
		.to("vm:gadgetOrders")
		.otherwise()
		.to("vm:badOrders");

		// test that our route is working
		from("vm:widgetOrders")
		.process(new Processor() {
		public void process(Exchange exchange) throws Exception {
		System.out.println("=========================");
		System.out.println("=Received a Widget Order=");
		System.out.println("=========================");
		String inBody = exchange.getIn().getBody(String.class);
		System.out.println(inBody);
		}
		});

		from("vm:gadgetOrders")
		.process(new Processor() {
		public void process(Exchange exchange) throws Exception {
		System.out.println("=========================");
		System.out.println("=Received a Gadget Order=");
		System.out.println("=========================");
		String inBody = exchange.getIn().getBody(String.class);
		System.out.println(inBody);
		}
		});

		from("vm:badOrders")
		.process(new Processor() {
		public void process(Exchange exchange) throws Exception {
		System.out.println("==========================");
		System.out.println("===Received a Bad Order===");
		System.out.println("==========================");
		String inBody = exchange.getIn().getBody(String.class);
		System.out.println(inBody);
		}
		});

		}
}

Το route λοιπόν ξεκινάει από το direct:start και στέλνει τα εισερχόμενα μηνύματα σε ένα εσωτερικό queue που το έχουμε ονομάσει vm:incomingOrders. Όταν τα μηνύματα φτάσουν εκεί, τα διαβάζουμε και ξεκινάμε μια λογική παρόμοια με εκείνη του if-else. Με την λέξη κλειδί choice ειδοποιούμε το route, και κατα συνέπεια το context, ότι θα ακολουθήσει έλεγχος στο περιεχόμενο του μηνύματος.

Με την λέξη κλειδί when ξεκινάει ο πρώτος έλεγχος και ο κάθε έλεγχος που θέλουμε να κάνουμε. Εδώ ψάχνουμε μέσα στο μήνυμα για την λέξη widget. Αν υπάρχει, τότε το μήνυμα θα δρομολογηθεί σε ένα εσωτερικό queue με το όνομα vm:widgetOrders. Επειδή αυτά τα queues μας τα προσφέρει το Apache Camel για testing σκοπούς χρειαζόμαστε έναν τρόπο να τα διαβάσουμε και να βεβαιωθούμε ότι το σωστό μήνυμα έφτασε στο queue. Η εύκολη λύση σε αυτό το πρόβλημα είναι να δημιουργήσουμε ένα Processor ο οποίος θα διαβάζει το μήνυμα από το queue και θα μας το δείχνει στο terminal.

Αν τώρα το μήνυμα δεν περιέχει την λέξη κλειδί widget αλλά περιέχει την λέξη gadget τότε θα δρομολογηθεί σε ένα διαφορετικό queue με το όνομα vm:gadgetOrders. Και ο δεύτερος έλεγχος ξεκινάει με την λέξη κλειδί when. Χρησιμοποιούμε έναν ακόμα Processor που θα διαβάσει το μήνυμα από το queue και θα μας το δείξει στο terminal.

Τέλος, για να καλύψουμε όλες τις πιθανότητες, αν για κάποιο λόγο το μήνυμα δεν περιέχει την λέξη widget ούτε την λέξη gadget τότε θα δρομολογηθεί σε ένα queue που το έχουμε ονομάσει vm:badOrders. Σε αυτή την συστοιχία από when όταν θέλουμε σαν τελευταία επιλογή να καλύψουμε οποιαδήποτε άλλη περίπτωση πέραν από αυτές που έχουμε καλύψει στα when τότε χρησιμοποιούμε την λέξη otherwise. Εδώ δεν χρειάζεται να ελέγξουμε για κάτι, απλά στέλνουμε το μήνυμα στο vm:badOrders.

Ας τρέξουμε το πρόγραμμα μας και ας στείλουμε τρία διαφορετικά μηνύματα – ένα που να περιέχει την λέξη widget, ένα που να περιέχει την λέξη gadget και ένα τελευταίο με μια τυχαία λέξη.

Πριν κλείσουμε το σημερινό δωρεάν μάθημα Apache Camel, ας προσθέσουμε ακόμα μια ιδιότητα στο Apache Camel Route. Ας υποθέσουμε λοιπόν ότι όλες οι παραγγελίες που έρχονται και είναι είτε widget είτε gadget πρέπει να προχωρήσουν σε ένα τελευταίο queue που το ονομάζουμε OrderProcessing. Ενώ οι παραγγελιές που είναι αποθηκευμένες στο vm:badOrders θέλουμε να παραμείνουν εκεί και να μην προχωρήσουν παρακάτω. Πως μπορούμε να το κάνουμε αυτό?

Βασικά πρέπει να προσθέσουμε δύο ακόμα λέξεις-κλειδιά στον κώδικα μας. Η πρώτη λέξη είναι η stop η οποία γράφεται αμέσως μετά το vm:badOrders δηλώνοντας έτσι ότι αυτός είναι ο τελικός προορισμός των μηνυμάτων που ανήκουν στο συγκεκριμένο queue. Η δεύτερη λέξη είναι το end μετά το stop για να δηλώσει το τέλος του choice. Αμέσως μετά ορίζουμε το καινούργιο queue στο οποίο θα συνεχίσουν τα μηνύματα. Αν δεν είχαμε δηλώσει το stop τότε όλα τα μηνύματα θα προχωρούσαν στο επόμενο queue ενώ αν δεν κλείναμε το choice με το end τότε θα έπρεπε να γράφαμε πολλαπλές φορές το orderProcessing queue για να δηλώναμε την συνέχεια του μηνύματος.

Ο κώδικας λοιπόν του MyRouteBuilder αλλάζει ο εξής:

MyRouteBuilder.java

package com.example;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;

public class MyRouteBuilder extends RouteBuilder {


		@Override
		public void configure() {


		from("direct:start").to("vm:incomingOrders");

		from("vm:incomingOrders")
		.choice()
		.when(body().contains("widget"))
		.to("vm:widgetOrders")
		.when(body().contains("gadget"))
		.to("vm:gadgetOrders")
		.otherwise()
		.to("vm:badOrders").stop()
		.end()
		.to("vm:orderProcessing");

		// test that our route is working
		from("vm:widgetOrders")
		.process(new Processor() {
		public void process(Exchange exchange) throws Exception {
		System.out.println("=========================");
		System.out.println("=Received a Widget Order=");
		System.out.println("=========================");
		String inBody = exchange.getIn().getBody(String.class);
		System.out.println(inBody);
		}
		});

		from("vm:gadgetOrders")
		.process(new Processor() {
		public void process(Exchange exchange) throws Exception {
		System.out.println("=========================");
		System.out.println("=Received a Gadget Order=");
		System.out.println("=========================");
		String inBody = exchange.getIn().getBody(String.class);
		System.out.println(inBody);
		}
		});

		from("vm:badOrders")
		.process(new Processor() {
		public void process(Exchange exchange) throws Exception {
		System.out.println("==========================");
		System.out.println("===Received a Bad Order===");
		System.out.println("==========================");
		String inBody = exchange.getIn().getBody(String.class);
		System.out.println(inBody);
		}
		});

		from("vm:orderProcessing")
		.process(new Processor() {
		public void process(Exchange exchange) throws Exception {
		System.out.println("==========================");
		System.out.println("==Received a Final Order==");
		System.out.println("==========================");
		String inBody = exchange.getIn().getBody(String.class);
		System.out.println(inBody);
		}
		});

}
}

Αν στείλουμε ένα badOrder μήνυμα θα το δούμε μόνο μέσα στο badOrders queue, ενώ για το widget και το gadget θα δούμε δύο μηνύματα στο terminal επειδή αυτά προχώρησαν στο τελικό queue.