Pattern Based Development with ServiceMix

Sky-Tiger

Implement the insurance broker with ServiceMix: the request part

When working with a JBI container like ServiceMix it’s a best practice to start with a SA diagram before we start with the implementation of the several BC and SE configuration files and implementation classes. With the SA diagram, consisting of the SUs, we get a good overview of what we actually need to implement. The SA diagram of the request part of the insurance broker example is shown in figure 8.

Figure 8 An overview of the SA and SUs for the request part of the insurance broker example.

There is quite a bit of work to do, as you can see in figure 8. With the numbers in the circles, the ordering of the message flow is shown. The message flow starts by consuming a message from the insurance.in queue with the JMS BC. Let’s look at the JMS BC configuration in listing 7.

Sky-Tiger

Listing 7 Consume the insurance request message from the JMS queue.

<beans xmlns:jms="http://servicemix.apache.org/jms/1.0"
       xmlns:esb="http://esbinaction.com/insurance">
  
  <classpath>
    <location>.</location>
    <location>bcel.jar</location>
    <location>jibx-bind.jar</location>
    <location>jibx-extras.jar</location>
    <location>jibx-run.jar</location>
    <location>qdox-1.6.1.jar</location>
    <location>stax-api.jar</location>
    <location>wstx-asl.jar</location>
    <location>xmlpull_1_1_4.jar</location>
    <location>xpp3.jar</location>
  </classpath>
  
  <jms:consumer service="esb:insuranceReceiver"
       endpoint="jmsEndpoint"
       targetService="esb:insuranceDSLRouter"
       destinationName="insurance.in"
       connectionFactory="#connectionFactory"
       marshaler="#InsuranceJMSMarshaler"/>

  <bean id="InsuranceJMSMarshaler"
      class="esb.dzone.servicemix.util.InsuranceJMSMarshaler"/>
  
  <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="tcp://localhost:61616" />
  </bean>

</beans>

The JMS BC configuration as shown in listing 7 is quite similar to the configuration we saw earlier in the hello world example. The main difference is that we use an in-only MEP in this implementation and the message is forwarded to a content-based router implemented with Apache Camel. We also have defined a marshaler in this configuration, because we sent a CarInsuranceRequest or TravelInsuranceRequest message to the insurance.in queue. The marshaler needs to transform these messages into an XML message, before it can be sent to the Camel content-based router. We use JiBX to transform the Java objects to XML and vice versa. Because we use JiBX in the implementation of the InsuranceJMSMarshaler, we have to include the JiBX jars on the JMS BC class path as show in listing 7. The implementation of the marshaler is shown in listing 8.

Listing 8 Implementation of the JMS marshaler that uses JiBX for XML transformation.

public class InsuranceJMSMarshaler extends DefaultConsumerMarshaler {

  protected void populateMessage(Message message,
        NormalizedMessage normalizedMessage) throws Exception {
    if (message instanceof ObjectMessage) {
      ObjectMessage objectMessage = (ObjectMessage) message;
      Object payload = objectMessage.getObject();
      Source source = JiBXUtil.marshalDocument(payload, "UTF-8");
      normalizedMessage.setContent(source);
    } else {
      throw new UnsupportedOperationException("JMS message is not a ObjectMessage");
    }
  }
}

When a JMS ObjectMessage is consumed by the JMS BC, the Java object payload is transformed to XML with a JiBXUtil class. In the source code of this article you can find more details about the JiBX transformation and JiBXUtil class implementation.

Sky-Tiger

Now let’s move on the content-based router implementation with Apache Camel. In addition to using the EIP SE, the ServiceMix project recommends to take a look at the Camel SE for more complex integration logic implementations. Apache Camel is a sub-project of the Apache ActiveMQ project and provides an implementation for a lot of the enterprise integration patterns of the Hohpe and Woolf book. Camel provides a XML based configuration and a Java domain specific language (DSL) to implement integration logic. The configuration of the Camel SE and the content-based router implementation is shown in listing 9.

Listing 9a Send the car insurance request to the two car insurance services.

<!-- camel-context.xml Camel SU configuration -->
<beans xmlns="http://www.springframework.org/schema/beans">
  <camelContext id="camel" xmlns="http://activemq.apache.org/camel/schema/spring">
    <package>esb.dzone.servicemix.camel</package>
  </camelContext>
</beans>

Listing 9b

/ The Camel content-based router implementation
public class InsuranceRouter extends RouteBuilder {
  private final static String JBI_SERVICE = "jbi:service:";
  private final static String NAMESPACE = "http://esbinaction.com/insurance";
  private final static String SERVICE_IN = JBI_SERVICE + NAMESPACE +
       "/insuranceDSLRouter";
  private final static String LUXURY_CAR_OUT = JBI_SERVICE + NAMESPACE +
       "/luxuryCarSender";
  private final static String BUDGET_CAR_OUT = JBI_SERVICE + NAMESPACE +
       "/budgetCarSender";
  private final static String TRAVEL_OUT = JBI_SERVICE + NAMESPACE +
       "/travelPipeline";
  private final static String FAILURE_OUT = JBI_SERVICE + NAMESPACE +
       "/insuranceFailureSender";

  public void configure() {
    from(SERVICE_IN)
      .convertBodyTo(DOMSource.class)
      .choice()
        .when(xpath("//ins:TravelInsuranceRequest")
            .namespace("ins", "http://dzone.com/insurance"))
          .to(TRAVEL_OUT)
        .when(xpath("//ins:CarInsuranceRequest")
            .namespace("ins", "http://dzone.com/insurance"))
          .to(LUXURY_CAR_OUT, BUDGET_CAR_OUT)
        .otherwise()
          .to(FAILURE_OUT);
  }
}

Listing 9 consists of two files, a camel-context.xml file which is used to configure the Camel SU and the InsuranceRouter Java class that implements the content-based routing logic. The camel-context.xml file defines a package that will be inspected at runtime to look for Camel classes. The Camel runtime will find the InsuranceRouter class, because it extends the RouteBuilder Camel framework class.

In the InsuranceRouter we have to implement the configure method with the content-based routing logic. First, we have to define a message channel to receive messages, the SERVICE_IN attribute, which corresponds with the targetService configuration of the JMS BC of listing 7. With this message channel definition, the routing logic is registered as a JBI service endpoint in the ServiceMix container. Then we define routing logic with the choice and when and otherwise methods. When the incoming XML message is a car insurance request it’s routed on to the budget and luxury car insurance endpoints and for a travel insurance request, the message is routed to an EIP pipeline.

Sky-Tiger

Let’s first look at the budget car and luxury car insurance endpoints in listing 10. Note that this listing consists of a JMS BC configuration and a File BC configuration. The JMS BC configuration is actually part of the same xbean.xml file as shown in listing 7 as you can see in the article’s source code.

Listing 10 The JMS BC and File BC configurations for the car insurance endpoints.

<!-- JMS BC SU configuration -->
<jmsrovider service="esb:luxuryCarSender"
    endpoint="jmsEndpoint"
    destinationName="luxurycar.send"
    connectionFactory="#connectionFactory"/>

<!-- File BC SU configuration -->
<beans xmlns:file="http://servicemix.apache.org/file/1.0"
     xmlns:esb="http://esbinaction.com/insurance">
       
  <classpath>
    <location>.</location>
    <location>bcel.jar</location>
    <location>jibx-bind.jar</location>
    <location>jibx-extras.jar</location>
    <location>jibx-run.jar</location>
    <location>qdox-1.6.1.jar</location>
    <location>stax-api.jar</location>
    <location>wstx-asl.jar</location>
    <location>xmlpull_1_1_4.jar</location>
    <location>xpp3.jar</location>
  </classpath>

  <file:sender service="esb:budgetCarSender"
      endpoint="fileEndpoint"
      directory="file:budgetCarIn"
      marshaler="#InsuranceFileMarshaler"/>
               
  <bean id="InsuranceFileMarshaler"
       class="esb.dzone.servicemix.util.InsuranceFileMarshaler"/>       
</beans>

The JMS configuration is pretty standard as it just sends the message to a JMS queue named luxury.send. The File BC configuration defines a file sender, which sends the message from the content-based router to a file directory named budgetCarIn. But, because we want the message to be formatted as a CSV file, we have defined another marshaler here. The file marshaler uses JiBX to transform the XML message to a CarInsuranceRequest Java object, which is transformed to a CSV message with similar transformation logic as the Mule example implementation of the previous article. Let’s look at the file marshaler in listing 11.

Listing 11 The file marshaler implementation.

public class InsuranceFileMarshaler extends DefaultFileMarshaler {

  protected void writeMessageContent(MessageExchange exchange,
      NormalizedMessage message,  OutputStream out, String path)
            throws MessagingException {
    Source src = message.getContent();
    if (src == null) {
      throw new NoMessageContentAvailableException(exchange);
    }
    try {
      CarInsuranceRequest request = (CarInsuranceRequest)
          JiBXUtil.unmarshalDocument(src, CarInsuranceRequest.class);
      String csvMessage = getInsuranceCSV(request);
      OutputStreamWriter writer = new OutputStreamWriter(out);
      writer.write(csvMessage);
      writer.flush();
    } catch (Exception e) {
      throw new MessagingException(e);
    }
  }
   
  private String getInsuranceCSV(CarInsuranceRequest request) {
            return new StringBuffer()
                        .append(request.getRequestID())
                        .append(","
                        .append(request.getNumberPlate())
                        .append(","
                        .append(request.getCarType())
                        .append(","
                        .append(request.getBuildYear())
                        .append(","
                        .append(new SimpleDateFormat().format(request.getStartDate()))
                        .toString();
  }
}

The file marshaler just transforms the CarInsuranceRequest to a CSV message and writes it to a file with an OutputStreamWriter. This completes the car insurance request implementation, so now let’s move on to the travel insurance request handling.

Before we can send the travel insurance request to the webservice, we first have to transform the XML to the target format defined by the webservice WSDL. To reduce the lines of code in this article a bit (there’s enough code already to my opinion), we will focus on the ServiceMix configuration. Because the ServiceMix Saxon SE, which provides transformation functionality based on XSLT, requires an in-out MEP, we first have to transform the message exchange to in-out with an EIP pipeline as already explained in figure 4. And because the CXF BC, which provides webservice functionality, also requires an in-out MEP we need to define two pipelines as shown in listing 12.

Listing 12 The EIP pipeline definitions to be able to invoke the travel webservice.

<beans xmlns:eip="http://servicemix.apache.org/eip/1.0"
    xmlns:esb="http://esbinaction.com/insurance"
    xmlns:tri="http://dzone.com/travelInsurance">
       
  <eipipeline service="esb:travelPipeline" endpoint="routingEndpoint">
    <eip:transformer>
      <eip:exchange-target service="esb:transformTravelRequest"/>
    </eip:transformer>
    <eip:target>
      <eip:exchange-target service="esb:travelServicePipeline" />
    </eip:target>
  </eipipeline>
       
  <eipipeline service="esb:travelServicePipeline" endpoint="routingEndpoint">
    <eip:transformer>
      <eip:exchange-target service="tri:TravelInsuranceServiceImplService"
          operation="tri:getTravelInsurance"/>
    </eip:transformer>
    <eip:target>
      <eip:exchange-target service="esb:insuranceSender" />
    </eip:target>
  </eipipeline>
</beans>

The first pipeline definition is invoked from the Camel content-based router implementation shown in listing 9. In the first pipeline we invoke the Saxon SE, which transforms the XML message to the webservice message format. Then the first pipeline invokes the second pipeline to invoke the webservice with an in-out MEP. So let’s quickly look at the Saxon SE configuration in listing 13.

Listing 13 The Saxon SE configuration, which transforms the XML message to the travel web service message format.

<beans xmlns:saxon="http://servicemix.apache.org/saxon/1.0"
    xmlns:esb="http://esbinaction.com/insurance">
          
  <saxon:xslt service="esb:transformTravelRequest" endpoint="xsltEndpoint"
        resource="classpath:TravelRequest.xslt" />       
</beans>

Sky-Tiger

Well, this is a short configuration. We define the XSLT file to transform the incoming XML message and we define the Saxon JBI service endpoint names. So let’s move on to the CXF BC configuration in listing 14, which performs the travel webservice call.

Listing 14 The CXF BC configuration, which invokes the travel webservice.

<beans xmlns:cxfbc="http://servicemix.apache.org/cxfbc/1.0"
        xmlns:tri="http://dzone.com/travelInsurance">

  <classpath>
    <location>.</location>
  </classpath>
  
  <cxfbcrovider wsdl="classpath:travelinsurance.wsdl"
       locationURI="http://localhost:9090/hello"
            endpoint="TravelInsuranceServiceImplPort"
            service="tri:TravelInsuranceServiceImplService"/>
</beans>

With the travelinsurance.wsdl file included in the CXF BC SU, the CXF BC is able to invoke the travel webservice on the location specified with the locationURI attribute.

Well that wraps up the request part. We’ve shown you so far how we can use a content based router implemented with Apache Camel to determine the target endpoint of the insurance request message. We’ve used a recipient list to send a car insurance request to two insurance endpoints using two different transports, file and JMS, in two different formats, CSV and XML. We’ve also shown you how a travel insurance webservice can be invoked using the CXF BC, and how the response of this webservice is routed back to a JMS queue where the website is listening on. So we’ve already done a very small section of the response part, since we already routed the response from the webservice back to the queue the website is listening on.

Now let’s look at how we can implement the responses of the car insurance companies.

Sky-Tiger

Implement the insurance broker with ServiceMix: the response part

As said, we already defined most of the insurance broker example configuration. Let’s first look at the remaining pieces of the response part as part of a SA diagram in figure 9.

Figure 9 An overview of the SA and SUs for the response part of the insurance broker example.

We already defined the CXF provider and pipeline configuration of the travel webservice invocation. So the only part we have to discuss for the response part is the aggregator implementation. Let’s look at the aggregator configuration in listing 15.

Listing 15 The definition of the aggregator, which aggregates two car insurance response messages in one result message.

<beans xmlns:eip="http://servicemix.apache.org/eip/1.0"
    xmlns:esb="http://esbinaction.com/insurance"
    xmlns:tri="http://dzone.com/travelInsurance">
       
  <eip:split-aggregator service="esb:insuranceAggregator" endpoint="routingEndpoint">
    <eip:target>
      <eip:exchange-target service="esb:insuranceSender" />
    </eip:target>
  </eip:split-aggregator>
</beans>

This is not too difficult, isn’t it? We define the aggregator JBI service endpoint name and the target service where the aggregated message is sent. To be able to use the out-of-the-box aggregation functionality of the EIP aggregator we do have to define some message properties, such as a correlation identifier and the number of messages to be aggregated. In the JMS and file marshalers we discussed in listing 8 and 11 we can add these message properties. In listing 16 the implementation of the file marshaler is shown.

Sky-Tiger

Listing 16. Overview of the file marshaler read method which adds aggregation message properties.

public class InsuranceFileMarshaler extends DefaultFileMarshaler {
  
  public static final String FILE_NAME_PROPERTY = "org.apache.servicemix.file.name";
  public static final String FILE_PATH_PROPERTY = "org.apache.servicemix.file.path";

  public void readMessage(MessageExchange exchange, NormalizedMessage message,
        InputStream in, String path) throws IOException, JBIException {
           
            BufferedReader br = new BufferedReader(new InputStreamReader(in));
            StringBuilder sb = new StringBuilder();
            String line = null;
     while ((line = br.readLine()) != null) {
       sb.append(line);
     }
     br.close();
            String csvString = sb.toString();
     String[] elements = csvString.split(",");
     InsuranceResponse insuranceResponse = new InsuranceResponse();
     insuranceResponse.setResponseID(elements[0]);
     insuranceResponse.setRequestID(elements[1]);
     insuranceResponse.setInsuranceCompanyName(elements[2]);
     insuranceResponse.setPrice(Float.parseFloat(elements[3]));
     Source source = null;
     try {
       source = JiBXUtil.marshalDocument(insuranceResponse, "UTF-8");
     } catch(Exception e) {
       throw new JBIException(e);
     }
     message.setContent(source);
     message.setProperty(FILE_NAME_PROPERTY, new File(path).getName());
     message.setProperty(FILE_PATH_PROPERTY, path);
     message.setProperty("org.apache.servicemix.eip.splitter.corrid",
         insuranceResponse.getRequestID());
     message.setProperty("org.apache.servicemix.eip.splitter.index", 0);
     message.setProperty("org.apache.servicemix.eip.splitter.count", new Integer(2));
  }
}

As shown in listing 16, the file poller first transforms the CSV message to an InsuranceResponse Java object. Then the InsuranceResponse object is transformed to an XML message with JiBX. Before the XML message is sent further on in the ServiceMix container, we add the EIP aggregator message properties. The first property is the correlation identifier, then the message index number and the number of messages to be aggregated. In the JMS marshaler we implemented similar functionality, but the index message property is set to 1.

And that’s it for the response part!
Test the insurance broker with ServiceMix

To really grasp the ServiceMix configurations shown in the insurance broker example, you can use the source code to get the example to run. We provided an Ant build script, build.xml, to make it a little bit easier for you. With the start target you can start the ServiceMix container and with the deploy-insurance target you can deploy the complete insurance broker service assembly to the ServiceMix container.

Then you can use the InsuranceTest JUnit test to send a JMS message to trigger the insurance broker message flow. With the LuxuryCarTest JUnit test you can sent the luxury car response message and there is a response.csv file available in the file SU directory to send a response for the budget car insurance company. The InsuranceTest also includes code to simulate the travel web service.
Conclusion

In this article we’ve shown how you can implement an integration case study with ServiceMix. There are quite a bit of differences between the Mule and ServiceMix insurance broker case study implementations as you can see, but the configuration also has some similarities. Both Mule and ServiceMix use Spring extensively to configure the integration logic and both ESBs use an XML based configuration style.

The main differences are related to JBI foundation of ServiceMix and the Mule specific architecture of Mule. JBI uses SUs and SAs and Mule defines everything as part of a mule-config.xml file. Another difference is the use of XML messages or normalized messages within ServiceMix (according to the JBI specification) and Java objects within Mule. Mule accepts Java objects as message payload, but ServiceMix requires the use of an XML payload. A third difference is the hotdeploy feature of ServiceMix that’s lacking in the Mule implementation. With ServiceMix you can easily deploy new versions of a SA, while ServiceMix keeps running. With Mule you will have to restart the container, before a new Mule configuration can be loaded.

We hope you enjoyed this introduction into ServiceMix 3.2.1 and the pattern based integration development approach. We hope we were able to show the functionality of Mule and ServiceMix with a small case study.
Additional resources

1. ServiceMix – http://servicemix.apache.org
2. ServiceMix 4 – http://servicemix.apache.org/SMX4/index.html
3. Enterprise Integration Patterns – http://www.enterpriseintegrationpatterns.com
4. Apache Camel – http://activemq.apache.org/camel
5. Download article source code – http://www.esbinaction.com/files/dzoneservicemix.zip

jieforest

好文，就是太长了点。

justforregister

nice job

justforregister

nice job