Camel integration trail

Apache Camel ist ein Routing und Mediation Engine. Camel unterstützt eine Vielzahl von Komponenten zur Lösung von Integrationsproblemen. Camel nutzt Domain Specific Languages (DSLs) für die Definition des Nachrichtenflusses und die Transformation von Nachrichten (Typ- und Formattransformationen). Die Architektur von Camel ist modular und erweiterbar. Camel kann als eigenständige, leichtgewichtige Laufzeitumgebung oder in ESB-Umgebungen betrieben werden. Apache Service Mix (ESB und JBI-Container) und die JBoss Enterprise SOA Platform 5.1(beinhaltet ein Camel Gateway als Technical Preview) können als Laufzeitumgebung für Camel dienen. Für andere ESB- und Laufzeitumgebungen wurden spezielle Integrationsszenarion, wie beispielsweise das Deployment des Camel Frameworks, als WAR definiert. Camel ist nicht als eigenständiger Integrationsserver oder ESB designed. Die Einbettung von Camel in  zahlreiche Plattformen (Web, Spring, Java EE, JBI Container, OSGi, etc.) hingegen ist ein primäres Ziel des Camel-Teams.

Documentation of  65 EIPs

Apache Camel basiert auf den von Gregor Hohpe und Bobby Woolf definierten  Enterprise Integration Patterns (EIPs). Die beiden Autoren haben in ihrem Fachbuch: "Enterprise Integration Patterns" die wesentlichen immer wiederkehrenden Integrationsmuster publiziert. Das Buch ist als Lektüre für Integrationsspezialisten empfehlenswert. Die EIPs sind nicht nur für die Lösung von Integrationsproblemen nützlich, sondern definieren ein einheitliches Vokabular für die Kommunikation und Notation von Integrationslösungen.

Apache Camel ist sehr einfach zu nutzen. Das Paradigma Convention over Configuration ist komfortabel für die Integrationsentwickler umgesetzt. Integrationsentwickler können sich sehr gezielt auf die Lösung eines fachlichen Integrationsproblems konzentrieren. Komplexe Konfigurationen und der Aufbau der Integrationsinfrastruktur zur Problemlösung sind nicht mehr durch den Entwickler vorzunehmen.Camel erlaubt die Lösung eines Integrationsproblems im kleinen Rahmen mit der Möglichkeit die gefundene Lösung in der Folge in eine komplexe ESB-Umgebung zu deployen. 

Best Practice: Develop single route and deploy in ESB environment

Camel beinhaltet ein umfangreiches Test-Kit (inklusive Mocks für Endpoints) und unterstützt dadurch die TDD Strategie. Ein Endpoint wird über eine URI definiert und ist mit der URI-Notation parametrierbar.  Zahlreiche Komponenten für gängige Protokolle und Schnittstellen für die Verwendung der gängigen Technologien sind bereits in Camel integriert. Typkonverter, eine einfache Beanschnittstelle, ein ausgefeiltes Fehlerhandling, DSLs (Java, Spring, Scala), Akka-Integration, Transaktionen, ein definiertes Threading-Modell und eine Expression Language sind weitere Highlights von Camel. Rasche Produktivität und zeitnahes Feedback ist bei der Anwendung von Apache Camel deshalb nicht nur ein Versprechen, sondern durch die vielfältigen Funktionalitäten bei gleichzeitiger Leichtgewichtigkeit des Frameworks garantiert.

Szenario Camel Route mit Auditfunktionalität:

Camel route with audit functionality

In dem Szenario werden aus einer Dateiquelle XML- und CSV-Dateien mit Auftragsdaten gelesen, prozessiert und über ein Wire Tap im Rahmen der Auditierung an den Service Manager 2.0 zur Visualisierung gesendet.

Szenario Camel Route Quellcode (Java DSL):

import javax.jms.ConnectionFactory;
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;
import com.service.manager.gateway.api.application.camel.audit.OrderAuditService;
import com.service.manager.gateway.api.application.camel.processor.OrderProcessorService;

public class CamelRouteXmlAndCsvOrders {
  
    public static void main(String args[]) throws Exception {

        final CamelContext context = new DefaultCamelContext();
        final OrderProcessorService processorService = OrderProcessorService.newInstance();
      
        final ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");
        context.addComponent("jms", JmsComponent.jmsComponentAutoAcknowledge(connectionFactory));

        context.addRoutes(new RouteBuilder() {

            @Override
            public void configure() {

                from("file:data?noop=true").to("jms:incomingChannel");

                from("jms:incomingChannel").
                    wireTap("jms:serviceManagerAuditChannel").
                        choice().
                            when(header("CamelFileName").endsWith(".xml")).
                                to("jms:xmlDataChannel").
                            when(header("CamelFileName").regex("^.*(csv|csl)$")).                      
                                to("jms:csvDataChannel").
                            otherwise().
                                to("jms:badDataChannel");
              
                from("jms:xmlDataChannel").process(new Processor() {
                  
                    public void process(Exchange exchange) throws Exception {
                  
                        processorService.processXmlData(exchange.getIn().getBody());
                    }                  
                });  
              
                from("jms:csvDataChannel").process(new Processor() {
                  
                    public void process(Exchange exchange) throws Exception {
                  
                        processorService.processCsvData(exchange.getIn().getBody());
                    }
                  
                });  
              
                from("jms:serviceManagerAuditChannel").

                     bean(OrderAuditService.class, "auditOrderMessage");
            }
        });

        context.start();
        Thread.sleep(2000);
        context.stop();
    }
}

Der Aufruf der Methode bean(OrderAuditService.class, "auditOrderMessage") delegiert die Verarbeitung der Audit-Message zu dem Audit-Service. Das Service Activator Pattern ist dabei die Grundlage für den Service-Aufruf. Der Service Activator dient als Mediator zwischen einem Requester und einem Service. Das Service Activator Pattern wird in Java EE Applikationen häufig in Message Driven Beans (MDBs) genutzt. Die Anwendung des Patterns in nicht Messaging-Szenarien ist allerdings ebenso möglich.

Szenario Camel Route Audit Service:

import org.apache.log4j.Logger;
import com.service.manager.api.gateway.annotation.ServiceManagerGateway;
import com.service.manager.api.gateway.context.ServiceManagerGatewayContext;
import com.service.manager.gateway.api.application.camel.helper.DataConverterHelper;

@ServiceManagerGateway(servicePlanId="SERVICEPLAN:12f9d7f03a413639340c8172e32354246e1",
                                        serviceNodeId="SERVICE:12f9d8074be1363e84b43082e32351810c22")
public class OrderAuditService {

    static {
      
        ServiceManagerGatewayContext.newContext(OrderAuditService.class);
    }
  
    private static Logger log = Logger.getLogger(OrderAuditService.class);

    public void auditOrderMessage(final String body) {
      
        final DataConverterHelper dataConverter = DataConverterHelper.newInstance();
      
        if(dataConverter.isXml(body)) {
          
            log.info(dataConverter.convertXmlMessage(body));
        }
        else {
      
            log.info(dataConverter.convertCsvMessage(body));
        }
    }
}

Der Camel Route Audit Service sendet die Audit-Messages über das Service Manager 2.0 Gateway zu der Servicemanagementanwendung zur Visualisierung. Der Service Manager 2.0 eignet sich perfekt für die Anzeige von Audit- und Alarmmeldungen in SOA-Infrastrukturen, die mittels Camel und/oder ESBs betrieben werden.

Visualisation of audit-messages in Service Manager 2.0

Der Service Manager 2.0 erlaubt es, Camel Routing-Schemata abzubilden und über einen eindeutigen Identifier zu verknüpfen, sodass jeder einzelne Schritt im Routing-Prozess tracebar ist. Spezielle Camel Event Notifier zur Übermittlung von Camel Routing und Transformationsdaten können registriert und mit dem Service Manager 2.0 verbunden werden. Alarmfälle, die durch Fehler verursacht werden, sind dabei zeitnah visualisierbar und unterstützen schnelle Reaktionszeiten des Support-Teams zur Einhaltung der Service Level Agreements (SLAs) während der Betriebszeit einer service-orientierten Applikation.

Der Rechtshinweis des Java Blog für Clean Code Developer ist bei der Verwendung und Weiterentwicklung des Quellcodes des Blogeintrages zu beachten.

Pack the bag

Ein Bag wird als Multiset Datentyp bezeichnet. Der Multiset Datentyp ist keine Collection Klasse des Java SE Frameworks. Die Collection Klassen von Google hingegen beinhalten eine Multiset Klasse, die in Java-Anwendungen einsetzbar ist. Die in dem Blogbeitrag implementierte Bag-Klasse ist während einer Code Kata entstanden. Die Code Kata hat nur den Speicheralgorithmus des Bags beschrieben, ohne einen detaillierten Hinweis auf die Schnittstelle und Implementierung des Bags zu geben.

Textauszug der Kata-Beschreibung:

Eine Bag-Klasse speichert zu einem Objekt die Anzahl der Vorkommnisse. Ein Bag mit den Einträgen {"firstObj", "firstObj", "secondObj", "thirdObj"} liefert für einen getCount(„firstObj“) Aufruf den Wert zwei. Der Aufruf von uniqueEntrySet() hingegen liefert {"firstObj", "secondObj", "thirdObj"}.

Hinweis:

Ein Bag eignet sich zur Speicherplatzreduzierung, weil nicht jedes einzelne Objekt verwaltet wird. Ein ähnliches Verhalten ist von dem Flyweight Pattern bekannt. Das Flyweight Pattern ist wie der Bag ebenfalls sparsam im Umgang mit dem verfügbaren Speicherplatz.

UML-Diagramm:

Bag-Schnittstelle:

import java.util.Collection;
import java.util.Map;

public interface Bag<T> {

    public boolean put(final T element);
    public boolean put(final T element, final int occurrences);
    public int remove(final T element);
    public int remove(final T element, final int occurrences);
    public int count(final T element);
    public Collection<Map.Entry<T, Integer>> uniqueSet();
}

Bag-Implementierung:

import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class HashBag<T> implements Bag<T> {

    private final Map<T, Integer> bagMap = new ConcurrentHashMap<T, Integer>();

    @Override
    public boolean put(final T element) {
      
        assertElementParameterNotNull(element);
      
        final int elementCount = count(element);
        if (0 == elementCount) {

            bagMap.put(element, 1);          
        }
        else {

            bagMap.put(element, elementCount + 1);
        }

        return true;
    }

    @Override
    public boolean put(final T element, final int occurrences) {
      
        assertElementParameterNotNull(element);
        assertOccurrencesNotUnderOrOverflowed(occurrences);
      
        final int elementCount = count(element);
        if (0 == elementCount) {

            bagMap.put(element, occurrences);          
        }
        else {

            bagMap.put(element, elementCount + occurrences);
        }

        return true;
    }

    @Override
    public int remove(final T element) {

        assertElementParameterNotNull(element);
        assertElementAvailableInBag(element);

        return bagMap.remove(element);
    }

    @Override
    public int remove(final T element, final int occurrences) {

        assertElementParameterNotNull(element);
        assertElementAvailableInBag(element);
        assertOccurrencesNotUnderOrOverflowed(occurrences);
        assertOccurrencesNotGreaterThanElementCount(occurrences, element);

        final int elementCount = count(element);
        if (occurrences == elementCount) {

            return remove(element);
        }

        return bagMap.put(element, elementCount - occurrences);
    }

    @Override
    public Collection<Map.Entry<T, Integer>> uniqueSet() {

        return Collections.unmodifiableCollection(bagMap.entrySet());
    }

    @Override
    public int count(final T element) {
      
        assertElementParameterNotNull(element);
      
        final Integer elementCount = bagMap.get(element);
        if (null == elementCount) {

            return 0;
        }

        return elementCount;
    }

    private void assertElementParameterNotNull(final T element) {
      
        if (null == element) {

            throw new NullPointerException("Element parameter must not be null!");
        }
    }
  
    private void assertElementAvailableInBag(final T element) {

        final int elementCount = count(element);
        if (0 == elementCount) {

            final String errorMessage = String.format(
                    "Element (%s) not available in bag!", element);
          
            throw new IllegalArgumentException(errorMessage);
        }
    }

    private void assertOccurrencesNotGreaterThanElementCount(final int occurrences, final T element) {

        final int elementCount = count(element);
        if (occurrences > elementCount) {

            final String errorMessage = String.format(
                    "Bag occurrences (%d) to remove greater than element count (%d) in bag!",
                    occurrences, elementCount);
          
            throw new IllegalArgumentException(errorMessage);
        }
    }
  
    private void assertOccurrencesNotUnderOrOverflowed(final int occurrences) {

        if(occurrences < 0) {
          
            throw new IllegalArgumentException(String.format("Underflow (occurrences (%d) < 0)!", occurrences));
        }  
        else if(occurrences > Integer.MAX_VALUE) {
          
            throw new IllegalArgumentException(String.format("Overflow (occurrences (%d) > max. int)! ", occurrences));
        }
    }
}

Bag-Testfälle:

import static org.junit.Assert.*;
import java.util.Collection;
import java.util.Map;
import org.junit.Before;
import org.junit.Test;

public class BagTest {

    private Bag<String> bag;

    private String firstBagElement;
    private String secondBagElement;

    @Before
    public void setUp() {

        bag = new HashBag<String>();

        firstBagElement = "firstBagElement";
        secondBagElement = "secondBagElement";
    }

    @Test
    public void testPutSingleElement() {

        assertEquals(true, bag.put(firstBagElement));
        assertEquals(1, bag.count(firstBagElement));
    }
  
    @Test
    public void testPutMultipleElements() {

        assertEquals(true, bag.put(firstBagElement, 20));
        assertEquals(20, bag.count(firstBagElement));
    }

    @Test(expected=NullPointerException.class)
    public void testPutSingleNullElement() {

        assertEquals(true, bag.put(null));
    }
  
    @Test(expected=NullPointerException.class)
    public void testPutMultipleNullElements() {

        assertEquals(true, bag.put(null, 2));
    }
  
    @Test(expected = IllegalArgumentException.class)
    public void testPutMultipleElementsWithUnderflow() {

        bag.put(firstBagElement, 20);
        bag.put(firstBagElement, -20);
      
        assertEquals(20, bag.count(firstBagElement));
    }

    @Test(expected = IllegalArgumentException.class)
    public void testPutMultipleElementsWithOverflow() {

        bag.put(firstBagElement, 20);
        bag.put(firstBagElement, Integer.MAX_VALUE + 1);
      
        assertEquals(20, bag.count(firstBagElement));
    }
  
    @Test
    public void testPutSingleAndMultipleElements() {

        bag.put(firstBagElement);
        bag.put(firstBagElement, 20);
        bag.put(firstBagElement);

        assertEquals(22, bag.count(firstBagElement));
    }

    @Test
    public void testPutDifferentElements() {

        bag.put(firstBagElement);
        bag.put(secondBagElement);

        assertDifferentBagElements(bag.uniqueSet(), 2, 1, 1);
    }

    @Test
    public void testPutMultipleDifferentElements() {

        bag.put(firstBagElement, 20);
        bag.put(secondBagElement, 10);

        assertDifferentBagElements(bag.uniqueSet(), 2, 20, 10);
    }

    @Test
    public void testPutMultipleAndSingleDifferentElements() {

        bag.put(firstBagElement, 20);
        bag.put(firstBagElement);
        bag.put(secondBagElement, 10);
        bag.put(secondBagElement, 10);

        assertDifferentBagElements(bag.uniqueSet(), 2, 21, 20);
    }
  
    @Test
    public void testRemoveSingleElement() {

        bag.put(firstBagElement);
        bag.put(firstBagElement);
      
        assertEquals(2, bag.remove(firstBagElement));
        assertEquals(0, bag.count(firstBagElement));
    }

    @Test(expected=NullPointerException.class)
    public void testRemoveSingleNullElement() {

        assertEquals(0, bag.remove(null));
    }
  
    @Test(expected = IllegalArgumentException.class)
    public void testRemoveWhileElementIsNotAvailable() {

        bag.put(firstBagElement);
        bag.put(firstBagElement);
        bag.remove(secondBagElement);

        assertEquals(0, bag.count(secondBagElement));
    }

    @Test(expected = IllegalArgumentException.class)
    public void testRemoveMoreOccurencesThanAvailable() {

        bag.put(firstBagElement);
        bag.put(firstBagElement);
        bag.remove(firstBagElement,3);

        assertEquals(0, bag.count(firstBagElement));
    }
  
    @Test
    public void testRemoveMultipleSingleElements() {
      
        bag.put(firstBagElement, 30);
      
        assertEquals(30, bag.remove(firstBagElement, 30));      
        assertEquals(0, bag.count(firstBagElement));
    }

    @Test(expected=NullPointerException.class)
    public void testRemoveMultipleNullElements() {

        assertEquals(0, bag.remove(null, 2));
    }
  
    @Test(expected = IllegalArgumentException.class)
    public void testRemoveMultipleSingleElementsWithUnderflow() {
      
        bag.put(firstBagElement, 30);
        bag.remove(firstBagElement, -1);
      
        assertEquals(0, bag.count(firstBagElement));
    }
  
    @Test(expected = IllegalArgumentException.class)
    public void testRemoveMultipleSingleElementsWithOverflow() {
      
        bag.put(firstBagElement, 30);
        bag.remove(firstBagElement, Integer.MAX_VALUE + 1);
      
        assertEquals(0, bag.count(firstBagElement));
    }
  
    @Test
    public void testRemoveDifferentSingleElements() {

        bag.put(firstBagElement);
        bag.put(secondBagElement);
  
        assertEquals(1, bag.remove(firstBagElement));
        assertEquals(1, bag.remove(secondBagElement));      
        assertDifferentBagElements(bag.uniqueSet(), 0, 1, 1);
    }
  
    @Test
    public void testRemoveDifferentMultipleElements() {

        bag.put(firstBagElement);
        bag.put(firstBagElement);
        bag.put(secondBagElement);
        bag.put(secondBagElement);
      
        assertEquals(2, bag.remove(firstBagElement,1));
        assertEquals(2, bag.remove(secondBagElement,1));
      
        assertDifferentBagElements(bag.uniqueSet(), 2, 1, 1);
    }
  
    @Test
    public void testRemoveMultipleSingleElementsWithRest() {

        bag.put(firstBagElement, 30);
        bag.remove(firstBagElement, 10);

        assertEquals(20, bag.count(firstBagElement));
    }

    @Test(expected = IllegalArgumentException.class)
    public void testRemoveWithOverflow() {

        bag.put(firstBagElement, 10);
        bag.remove(firstBagElement, 20);

        assertEquals(0, bag.count(firstBagElement));
    }

    @Test(expected=NullPointerException.class)
    public void testNullElementCount() {

        assertEquals(0, bag.count(null));
    }
  
    @Test
    public void testUniqueElementSet() {

        bag.put(firstBagElement);
        bag.put(firstBagElement);
        bag.put(secondBagElement);

        assertDifferentBagElements(bag.uniqueSet(), 2, 2, 1);
    }

    private void assertDifferentBagElements(final Collection<Map.Entry<String, Integer>> ElementSet,
                                            int ElementSetElementCount, int firstExpectedElementCount,
                                            int secondExpectedElementCount) {

        assertEquals(ElementSetElementCount, ElementSet.size());

        for (Map.Entry<String, Integer> element : ElementSet) {

            if (element.getKey().equals("firstBagElement")) {

                assertEquals(firstExpectedElementCount, element.getValue().intValue());
              
            }
            else if (element.getKey().equals("secondBagElement")) {

                assertEquals(secondExpectedElementCount, element.getValue().intValue());
            }
        }
    }
}

Der aufmerksame Leser wird sofort erkennen, dass in den Testfällen Magic Numbers verwendet werden. Die Lesbarkeit des Quellcodes wird durch Magic Numbers verschlechtert. Ein reiner Zahlenwert hat keine semantische Aussagekraft. Änderungen an den Zahlenwerten sind überall im Quellcode nachzuziehen. Diese Änderungen sind aufwendig und fehleranfällig. Im produktiven Quellcode sind Magic Numbers deshalb nicht tolerierbar. Aufgrund der Einfachheit des Testfalls im Rahmen der Code Kata aber pragmatisch. Den Verstoß gegen das Gebot: "Schreibe die Testfälle mit der gleichen Sorgfalt wie den produktiven Quellcode" nehme ich in diesem einen Fall in Kauf.

Der Rechtshinweis des Java Blog für Clean Code Developer ist bei der Verwendung und Weiterentwicklung des Quellcodes des Blogeintrages zu beachten.