Эксперименты и исследования

The Skip Locked feature in Postgres 9.5 (message queue)

2016-10-14T21:49:00.000+09:00

В углу что-то мяукают по ActiveMQ. А между тем в postgress запилили крутую фичу. for update skip locked позволяет пропустить в select строки заблокированные в другой транзакции. К сожалению такое не прокатит в Oracle. Но тем не менее для мессажинга можно использовать уже сейчас. Из коробки имеем:

транзакции
персистенцию сообщений
клиент(обычный jdbc)
кластер(обычные средства предоставляемые postgres)

package com.bssys.blog.messaging;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;

import javax.sql.DataSource;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceUtils;
import org.springframework.jdbc.datasource.SingleConnectionDataSource;
import org.springframework.stereotype.Component;

@Component("consumer")
public class Consumer implements Runnable {

 private String select = "select id, message from skiptable where status = 'TODO' order by priority asc limit 1 for update skip locked";
 private String delete = "delete from skiptable where id = ? ";

 @Autowired
 private DataSource dataSource;

 public void run() {

  while (!Thread.currentThread().isInterrupted()) {
   Connection con = DataSourceUtils.getConnection(dataSource);
   SingleConnectionDataSource singleDs = new SingleConnectionDataSource(con, true);
   try {
    JdbcTemplate jdbcTemplate = new JdbcTemplate(singleDs);
    List<Map<String, Object>> queryForList = jdbcTemplate.queryForList(select);
    if (!queryForList.isEmpty()) {
     for (Map<String, Object> map : queryForList) {
      String id = (String) map.get("id");
      String message = (String) map.get("message");
      jdbcTemplate.update(delete, new Object[] { id });
      System.out.println(Thread.currentThread().getName() + ":" + message);
     }
     con.commit();
    } else {
     System.out.println(Thread.currentThread().getName() + ": no mass");
    }
   } catch (Exception e) {
    try {
     con.rollback();
    } catch (SQLException e1) {
     e1.printStackTrace();
    }
   } finally {
    singleDs.destroy();
   }
  }
 }
}

Dynamic Webservice registration on ServiceMix

2016-10-14T00:48:00.001+09:00

Рассмотрим как можно динамически зарегистрировать webservice на ServiceMix. В общем смысле подобным образом можно динамически регистрировать любые сервисы. Для примера реализуем простейший сервис c использованием Apache Camel. Сервис будет слушать файловую папку, в которую будет выгружаться файлы со ссылками на wsdl. Эти сервисы и будем динамически регистрировать.

Общий смысл такой. Создадим bean с методом registy который будет принимать ссылку на wsdl, создавать соответствующую конечную точку(endpoint) и динамически добавлять route, принимающий сообщения из нее.

package com.bssys.dynamic.cxf_example;

import javax.xml.namespace.QName;

import org.apache.camel.CamelContext;
import org.apache.camel.component.cxf.CxfEndpoint;
import org.apache.camel.component.cxf.DataFormat;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component("dynamicLogic")
public class DynamicLogic {
 
 @Autowired
 private CamelContext context;
 
 public void registry(String url) throws Exception{
  
  CxfEndpoint cxfEndpoint = new CxfEndpoint();
  cxfEndpoint.setDataFormat(DataFormat.RAW);
  cxfEndpoint.setWsdlURL(url);
  
  QName serviceName = cxfEndpoint.getServiceName();
  
  cxfEndpoint.setAddress("http://127.0.0.1:8383/"+serviceName.getLocalPart());
  
  context.addEndpoint(serviceName.getLocalPart(), cxfEndpoint);
  DynamicRouteBuilder builder = new DynamicRouteBuilder(serviceName.getLocalPart());
  context.addRoutes(builder);
  
 }
 
}

Динамический route принимающий сообщения из зарегистрированной конечной точки

package com.bssys.dynamic.cxf_example;

import org.apache.camel.builder.RouteBuilder;

public class DynamicRouteBuilder extends RouteBuilder {

 private String uri;

 public DynamicRouteBuilder(String uri) {
  this.uri = uri;
 }

 @Override
 public void configure() throws Exception {
  from(uri).to("direct:template");
 }

}

Маршрут Apache Camel

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:camel="http://camel.apache.org/schema/spring"
 xmlns:osgi="http://www.springframework.org/schema/osgi" xmlns:context="http://www.springframework.org/schema/context"
 xsi:schemaLocation="
       http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

 <context:annotation-config />
 <context:component-scan base-package="com.bssys.dynamic.cxf_example" />

 <camelContext id="mainContext" xmlns="http://camel.apache.org/schema/spring">
  <route>
   <from uri="file://C:/data/registryws" />
   <camel:bean ref="dynamicLogic" method="registry" />
  </route>
  <route>
   <from uri="direct:template" />
   <log message="The message contains ${body}" />
  </route>
 </camelContext>

</beans>

If-less programming (java) Part №2

2016-10-07T17:33:00.001+09:00

В предыдущей части мы рассмотрели как избавиться от if в паттерне "фабричный метод". Но есть паттерны специально придуманные чтобы избавить вас от этой проблемы. Один из них паттерн "Состояние".

Рассмотрим небольшой пример. У нас есть класс Tractor, у которого есть состояние: Ориентация по стороне света и координаты. У трактора есть возможность повернуться(метод turn) и двигаться в этом направлении (метод move).

package com.bssys.blog;

public class Tractor {

 public Orientation currentOrientation;
 public Coordinate currentCoordinate;

 public Tractor() {
  currentOrientation = Orientation.North;
  currentCoordinate = new Coordinate(0, 0);
 }

 public Tractor(Orientation currentOrientation, Coordinate currentCoordinate) {
  this.currentOrientation = currentOrientation;
  this.currentCoordinate = currentCoordinate;
 }

 public void move() {

  if (currentOrientation == Orientation.North) {
   currentCoordinate.y = currentCoordinate.y + 1;
  } else if (currentOrientation == Orientation.South) {
   currentCoordinate.y = currentCoordinate.y - 1;
  } else if (currentOrientation == Orientation.West) {
   currentCoordinate.x = currentCoordinate.x - 1;
  } else if (currentOrientation == Orientation.East) {
   currentCoordinate.x = currentCoordinate.x + 1;
  }
 }

 public void turn() {
  if (currentOrientation == Orientation.North) {
   currentOrientation = Orientation.East;
  } else if (currentOrientation == Orientation.South) {
   currentOrientation = Orientation.West;
  } else if (currentOrientation == Orientation.West) {
   currentOrientation = Orientation.North;
  } else if (currentOrientation == Orientation.East) {
   currentOrientation = Orientation.South;
  }
 }

 public Orientation getCurrentOrientation() {
  return currentOrientation;
 }

 public Coordinate getCurrentCoordinate() {
  return currentCoordinate;
 }

 @Override
 public String toString() {
  return "" + currentOrientation + " " + currentCoordinate;
 }
}

class Coordinate {

 public int x;
 public int y;

 public Coordinate(int x, int y) {
  this.x = x;
  this.y = y;
 }

 @Override
 public String toString() {
  return "x=" + x + " " + "y=" + y;
 }
}

enum Orientation {
 North, South, West, East
}

Обычно решение в лоб вырождается в портянку if-else в зависимости от некоторого состояния. В нашем случае состояние представляет собой направление движения, т.е. ориентацию по стороне света. Для того чтобы решить эту проблему выделим интерфейс состояния. Он будет представлять собой следующее:

interface State {
 public void turn(Tractor tractor);
 public void move();
}

Реализуем этот интерфейс для всех наших возможных состояний, и добавим в класс Tractor поле хранящее текущее состояние согласно нашему паттерну:

class EastState implements State {
 
 private Coordinate coordinate;
 
 public EastState(Coordinate coordinate) {
  this.coordinate = coordinate;
 }

 public void turn(Tractor tractor) {
  tractor.setCurrentState(new SouthState(coordinate));
 }
 
 public void move() {
  coordinate.x = coordinate.x + 1;
 }

 @Override
 public String toString() {
  return "East:" + coordinate;
 }
}

class NorthState implements State {
 
 private Coordinate coordinate;
 
 public NorthState(Coordinate coordinate) {
  this.coordinate = coordinate;
 }

 public void turn(Tractor tractor) {
  tractor.setCurrentState(new EastState(coordinate));
 }
 
 public void move() {
  coordinate.y = coordinate.y + 1;
 }

 @Override
 public String toString() {
  return "North:" + coordinate;
 }
}

class SouthState implements State {
 
 private Coordinate coordinate;
 
 public SouthState(Coordinate coordinate) {
  this.coordinate = coordinate;
 }

 public void turn(Tractor tractor) {
  tractor.setCurrentState(new WestState(coordinate));
 }

 public void move() {
  coordinate.y = coordinate.y - 1;
 }

 @Override
 public String toString() {
  return "South:" + coordinate;
 }
}

class WestState implements State {
 
 private Coordinate coordinate;
 
 public WestState(Coordinate coordinate) {
  this.coordinate = coordinate;
 }

 public void turn(Tractor tractor) {
  tractor.setCurrentState(new NorthState(coordinate));
 }

 public void move() {
  coordinate.x = coordinate.x - 1;
 }

 @Override
 public String toString() {
  return "West:" + coordinate;
 }
}

Смотрите, логика переходов по состояниям у нас теперь находится в специальных классах. И класс Tractor теперь просто должен делегировать свои действия текущему состоянию. Обратите внимание, что меняет состояние только метод turn. Таким образом итоговый код класса Tractor вырождается в следующее:

public class Tractor {

 private State currentState;

 public Tractor() {
  currentState = new NorthState(new Coordinate(0, 0));
 }

 public void move() {
  currentState.move();
 }

 public void turn() {
  currentState.turn(this);
 }

 public void setCurrentState(State state) {
  this.currentState = state;
 }

 @Override
 public String toString() {
  return currentState.toString();
 }

 public static void main(String[] args) {
  Tractor tractor = new Tractor();
  System.out.println(tractor);
  tractor.turn();
  tractor.move();
  tractor.move();
  tractor.move();
  System.out.println(tractor);
  tractor.turn();
  tractor.move();
  tractor.move();
  tractor.move();
  System.out.println(tractor);
 }
}

If-less programming (java) Part №1

2016-10-06T22:20:00.001+09:00

Господа, всем известно что нагромождение конструкций if-else есть признак плохого тона. Попробуем обойтись без их использования в некоторых ситуациях.

If-else в фабричном методе.

Для примера есть стандартная реализация паттерна "Фабричный метод", создающая объекты в зависимости от некоторого ключа.

package com.bssys.blog.example;


public class App {
 private Factory factory;

 public App() {
  factory = new Factory();
 }

 public void onMessage(String message) {
  Product product = factory.createProduct(message);
  product.action();
 }

 public static void main(String[] args) {
  App app = new App();
  app.onMessage("A");
  app.onMessage("B");
 }
}

interface Product {
 public void action();
}

class ConcreteProductA implements Product {

 public void action() {
  System.out.println("ConcreteProductA");
 }

}

class ConcreteProductB implements Product {

 public void action() {
  System.out.println("ConcreteProductB");
 }

}

class Factory {
 Product createProduct(String key) {
  Product product = null;
  if ("A".equals(key)) {
   product = new ConcreteProductA();
  } else if ("B".equals(key)) {
   product = new ConcreteProductB();
  }
  return product;
 }
}

Тут присутcnвует громоздкая конструкция if-else. Как избавиться от нее?

Использовать HashMap.

Фактически введем динамический if. Таким образом класс Factory будет выглядеть следующим образом:

class Factory {
 
 private Map<String, Product> factoryMap;
 
 public Factory() {
  factoryMap = new HashMap<String, Product>();
  factoryMap.put("A", new ConcreteProductA());
  factoryMap.put("B", new ConcreteProductB());
 }
 
 Product createProduct(String key) {
  Product product = factoryMap.get(key);
  return product;
 }
}

Тут мы немного сломали ожидаемое поведение, так как фабричный метод будет возвращать одни и теже объекты. Иногда это даже хорошо. Но тем не менее чтобы ожидаемое поведение было одинаковым введем интерфейс фабрики которая будет создавать конкретные продукты. И отрефакторим код:

package com.bssys.blog.example;

import java.util.HashMap;
import java.util.Map;

public class App {
 private CommonFactory factory;

 public App() {
  factory = new CommonFactory();
 }

 public void onMessage(String message) {
  Product product = factory.createProduct(message);
  product.action();
 }

 public static void main(String[] args) {
  App app = new App();
  app.onMessage("A");
  app.onMessage("B");
 }
}

interface Product {
 public void action();
}

class ConcreteProductA implements Product {

 public void action() {
  System.out.println("ConcreteProductA");
 }

}

class ConcreteProductB implements Product {

 public void action() {
  System.out.println("ConcreteProductB");
 }

}

interface Factory{
 Product createProduct();
}

class ConcreteProductAFactory implements Factory{

 public Product createProduct() {
  return new ConcreteProductA();
 }
}

class ConcreteProductBFactory implements Factory{

 public Product createProduct() {
  return new ConcreteProductB();
 }
}

class CommonFactory {
 
 private Map<String, Factory> factoryMap;
 
 public CommonFactory() {
  factoryMap = new HashMap<String, Factory>();
  factoryMap.put("A", new ConcreteProductAFactory());
  factoryMap.put("B", new ConcreteProductBFactory());
 }
 
 Product createProduct(String key) {
  Factory factory = factoryMap.get(key);
  return factory.createProduct();
 }
}

Уже лучше. Мы выделили интерфейс фабрики. Теперь у нас есть класс CommonFactory, который по сути является реестром всех объектов. Теперь если подключить мощь Spring мы могли бы вынести наш HashMap в конфиг, и управлять поведением не перекомпилируя наш код. Как вы видите мы нагенерили много кода. И хотя это правильный подход в java, есть немного другой способ избавиться от if.

Использовать enum с абстрактным методом.

enum Type {
 A {
  @Override
  public Product createProduct() {
   return new ConcreteProductA();
  }
 },
 B {
  @Override
  public Product createProduct() {
   return new ConcreteProductB();
  }
 };

 public abstract Product createProduct();
}

В таком случае наш фабричный метод переродится в следующее:

class Factory {
 
 public Product createProduct(String key) {
  Type valueOf = Type.valueOf(key);
  return valueOf.createProduct();
 }
}

Тут конечно тоже есть неявный динамический if при выполнении операции valueOf, но все равно выглядит элегантно.

Есть еще один способ обойтись без if используя reflection, но я не приветствую такой подход. Если хотите динамичности то берите наверное другой язык, а не java.

ServiceMix Atomikos XA transaction JDBC JMS

2016-04-26T18:13:00.003+09:00

Снова вернемся к важному вопросу о XA транзакциях. Рассмотрим реализацию XA транзакций с использованием популярного менеджера распределенных транзакций Atomikos. Это широко используемый диспетчер транзакций JTA с открытым исходным кодом, который предназначен для сред отличных от J2EE. Это как раз наш случай.
В данном примере будет представлена готовая конфигурация транзакций.
Приложение разделено на модули, которые разделяют настройку ресурсов, от бизнес логики. Фактически вы можете установить модули в ServiceMix и заняться разработкой, не разгребая как во многих других примерах системную и бизнес логику.
Итак модули:

ts - поднимает транзакционные артефакты JtaTransactionManager, UserTransactionManager, UserTransactionImp
ds - поднимает транзакционный источник данных на примере БД Oracle OracleXADataSource, AtomikosDataSourceBean
amq - поднимает транзакционные артефакты для работы с JMS на примере ActiveMQ, ActiveMQXAConnectionFactory, AtomikosConnectionFactoryBean
testXA - приложение для ServiceMix демонстрируещее работающие транзакции

Контекст модуля ts определяет два бина txMgr и userTransaction. Классы реализации предоставляются Atomikos и реализуют интерфейсы TransactionManager и UserTransaction из стандарта J2EE. После этого определяется бин jtaTxMgr с классом реализации JtaTransactionManager и внедряются два бина транзакций предоставляемые Atomikos. Это информирует Spring о необходимости использования Atomikos JTA для управления транзакциями. Далее сконфигурированный менеджер транзакций выставляется как osgi-сервис, чтобы позволить другим модулям его переиспользовать.

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:osgi="http://www.springframework.org/schema/osgi"
       xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

   <bean id="jtaTxMgr" class="org.springframework.transaction.jta.JtaTransactionManager">
  <property name="transactionManager" ref="txMgr" />
  <property name="userTransaction" ref="userTransaction" />
 </bean>

 <bean id="txMgr" class="com.atomikos.icatch.jta.UserTransactionManager" init-method="init" destroy-method="close">
  <property name="forceShutdown" value="false" />
 </bean>

 <bean id="userTransaction" class="com.atomikos.icatch.jta.UserTransactionImp">
  <property name="transactionTimeout" value="120" />
 </bean>
 
 <osgi:service interface="org.springframework.transaction.PlatformTransactionManager" ref="jtaTxMgr">
  <osgi:service-properties>
   <entry key="transaction.manager.name" value="atomikos" />
  </osgi:service-properties>
 </osgi:service>
</beans>

Контекст модуля ds определяет источник данных OracleXADataSource на примере БД Oracle , с использованием AtomikosDataSourceBean, который поддерживает XA источники данных. Он также предоставляет пул соединений. OracleXADataSource несмотря на название не является пулом, и при работе каждый раз поднимает соединение с базой. Не знаю почему во многих примерах его используют напрямую. При большой нагрузке его использование будет "безбожно жрать коннекты". Сконфигурированный источник данных выставляется как osgi-сервис. Обратите внимание что интерфейс для сервиса определяется как javax.sql.DataSource, что позволит в дальнейшем использовать удобный JdbcTemplate.

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:osgi="http://www.springframework.org/schema/osgi"
 xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

 <bean id="atomikosXaDataSourceBean" class="com.atomikos.jdbc.AtomikosDataSourceBean">
  <property name="uniqueResourceName" value="oracleXA" />
  <property name="testQuery" value="select * from dual" />
  <property name="xaDataSource" ref="dataSourceXA" />
  <property name="minPoolSize" value="1" />
  <property name="maxPoolSize" value="5" />
  <property name="reapTimeout" value="0" />
 </bean>

 <bean id="dataSourceXA" class="oracle.jdbc.xa.client.OracleXADataSource">
  <property name="URL" value="jdbc:oracle:thin:@host:1521:sid" />
  <property name="user" value="123" />
  <property name="password" value="123" />
  <property name="connectionCachingEnabled" value="false" />
 </bean>

 <osgi:service interface="javax.sql.DataSource" ref="atomikosXaDataSourceBean">
  <osgi:service-properties>
   <entry key="osgi.jndi.service.name" value="jdbc/oracleXA" />
   <entry key="datasource.name" value="oracleXA" />
  </osgi:service-properties>
 </osgi:service>
</beans>

Аналогичным образом определяем нужные бины для работы с ActiveMQ с помощью AtomikosConnectionFactoryBean

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:osgi="http://www.springframework.org/schema/osgi"
       xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

  <bean id="atomikosXaConnectionFactoryBean" class="com.atomikos.jms.AtomikosConnectionFactoryBean"
  init-method="init" destroy-method="close">
  <property name="uniqueResourceName" value="amq1" />
  <property name="xaConnectionFactory" ref="jmsXaConnectionFactory" />
  <property name="localTransactionMode" value="false" />
  <property name="minPoolSize" value="1" />
  <property name="maxPoolSize" value="5" />
  <property name="reapTimeout" value="0" />
 </bean>

 <bean id="jmsXaConnectionFactory" class="org.apache.activemq.ActiveMQXAConnectionFactory">
  <property name="brokerURL" value="failover:(tcp://localhost:61616)" />
  <property name="redeliveryPolicy">
   <bean class="org.apache.activemq.RedeliveryPolicy">
    <property name="maximumRedeliveries" value="1" />
   </bean>
  </property>
  <property name="userName" value="smx" />
  <property name="password" value="smx" />
 </bean>
 
 <osgi:service interface="javax.jms.ConnectionFactory" ref="atomikosXaConnectionFactoryBean">
  <osgi:service-properties>
   <entry key="osgi.jndi.service.name" value="jms/connectionFactory" />
   <entry key="connection.factory.name" value="activemqXA" />
  </osgi:service-properties>
 </osgi:service>
</beans>

Пример демонстрирующий использование транзакций разделен на два контекста.
system-context.xml получающий по ссылке нужные системные артефакты для работы с транзакциями и camel-context.xml содержащий бизнес логику

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:osgi="http://www.springframework.org/schema/osgi"
 xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

 <osgi:reference id="jtaTxMgr" interface="org.springframework.transaction.PlatformTransactionManager" filter="(transaction.manager.name=atomikos)"/>
 <osgi:reference id="atomikosXaDataSourceBean" interface="javax.sql.DataSource" filter="(datasource.name=oracleXA)"/>
 <osgi:reference id="atomikosXaConnectionFactoryBean" interface="javax.jms.ConnectionFactory" filter="(connection.factory.name=activemqXA)"/>
 
 <bean id="REQUIRED" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
  <property name="transactionManager" ref="jtaTxMgr" />
  <property name="propagationBehaviorName" value="PROPAGATION_REQUIRED" />
 </bean>

 <bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
  <property name="connectionFactory" ref="atomikosXaConnectionFactoryBean" />
  <property name="transacted" value="true" />
  <property name="transactionManager" ref="jtaTxMgr" />
  <property name="maxConcurrentConsumers" value="1" />
  <property name="cacheLevelName" value="CACHE_NONE" />
 </bean>

 <bean id="activemqXA" class="org.apache.activemq.camel.component.ActiveMQComponent">
  <property name="configuration" ref="jmsConfig" />
 </bean>
 
 <bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">
        <property name="dataSource" ref="atomikosXaDataSourceBean"/>
    </bean>
</beans>

Бизнес логика следующая:

Получить jms сообщение
Выгрузить сообщение в промежуточную очередь
Выполнить обновление БД (успешное)
Выполнить обновление БД приводящее к исключению
Выгрузить сообщение в итоговую очередь

Как вы можете убедиться, запустив этот пример, что сообщение в результате обработки данным роутом будет выгружено в очередь ошибок, при этом ни в одном промежуточном ресурсе данные не сохранятся. Это то что там надо.

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:osgi="http://www.springframework.org/schema/osgi"
 xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
 
 <bean id="hello" class="com.bssys.nio2.testXA.HelloBean">
  <constructor-arg ref="jdbcTemplate"/>
 </bean>

 <camelContext xmlns="http://camel.apache.org/schema/spring">
  <route>
   <from uri="activemqXA:queue:SQL_IN" />
   <transacted  ref="REQUIRED"/>
   <to uri="activemqXA:queue:SQL_MIDLE" />
   <bean ref="hello" method="goodlogic"/>
   <log message="REDELIVERED: ${header.JMSRedelivered}" />
   <bean ref="hello" method="badlogic"/>
   <to uri="activemqXA:queue:SQL_OUT" />
  </route>
 </camelContext>

</beans>

Чтобы сообщение выгружалось в очередь DLQ надо настроить немного ActiveMQ прописав в activemq.xml следующее

<policyEntry queue=">">
          <deadLetterStrategy>
            <!--
              Use the prefix 'DLQ.' for the destination name, and make
              the DLQ a queue rather than a topic
            -->
            <individualDeadLetterStrategy
              queuePrefix="DLQ." useQueueForQueueMessages="true" />
          </deadLetterStrategy>
 </policyEntry>

Таким образом для каждом очереди будет создаваться своя очередь DLQ. Очень удобно в дальнейшем разгребать конфликты. Ну и зависимости для Servicemix'а для того чтобы модули взлетели. Предварительно надо соответствующим образом установить драйвер для Oracle, которого нет в центральном репозитории Maven. Будут вопросы пишите в коментариях, я отвечу.

<?xml version="1.0" encoding="UTF-8"?>
<features xmlns="http://karaf.apache.org/xmlns/features/v1.0.0"
 name="egg">
 <feature name="egg-features" version="1.0">
  <feature version="3.2.14.RELEASE_1">spring-jdbc</feature>
  <feature>camel-sql</feature>
  <feature>camel-jdbc</feature>
  <feature>jndi</feature>
  <feature>war</feature>
  <feature>activemq</feature>
  <feature>activemq-web-console</feature>
  <feature>activemq-broker</feature>
  <feature>transaction</feature>
  <bundle>wrap:mvn:com.oracle/ojdbc6/12.1.0.2.0</bundle>
  <bundle>wrap:mvn:commons-dbcp/commons-dbcp/1.4</bundle>
  <bundle>mvn:com.atomikos/transactions-osgi/4.0.2</bundle>
 </feature>
</features>

Исходники на GitHib https://github.com/Hibernate2009/servicemix-atomikos-xa-jms-jdbc.git

WebSocket server java. Best way

2016-03-11T22:25:00.002+09:00

Итак, когда мы разобрались как реализовать WebSocket server с использованием NIO2.0. Я покажу лучший способ это сделать используя JSR 356, Java API for WebSocket. Обратите внимание насколько все просто. Достаточно обвесить обычный POJO следующими анатациями и упаковать как веб-приложение.

@ServerEndpoint
@OnOpen
@OnMessage
@OnClose
@OnError

Все! Это настолько просто, что даже не требует пояснений.
Код как обычно на https://github.com/Hibernate2009/websocket-jsr356.git

package com.bssys.ws;

import java.util.logging.Logger;

import javax.websocket.CloseReason;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;


@ServerEndpoint("/echo")
public class WsChatServlet {
 private Logger logger = Logger.getLogger(this.getClass().getName());
  
    @OnOpen
    public void onOpen(Session session) {
        logger.info("Connected ... " + session.getId());
        SessionManager.getInstance().addSession(session);
    }
 
    @OnMessage
    public void onMessage(Session session, String message ) {
     logger.info(String.format("Client %s send message %s", session.getId(), message));
     SessionManager.getInstance().writeMessageToClients(session, message);
    }
 
    @OnClose
    public void onClose(Session session, CloseReason closeReason) {
        logger.info(String.format("Session %s closed because of %s", session.getId(), closeReason));
        SessionManager.getInstance().removeSession(session);
    }
    
    @OnError
    public void onError(Session session, Throwable throwable){
     logger.info(String.format("Session %s error ", session.getId()));
     SessionManager.getInstance().removeSession(session);
    }
}

Вспомогательный класс для управления подключениями

package com.bssys.ws;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import javax.websocket.Session;

public class SessionManager {

 private final List<Session> sessions = Collections.synchronizedList(new ArrayList<Session>());

 private static SessionManager instance;

 private SessionManager() {
 }

 public static SessionManager getInstance() {
  if (instance == null) {
   instance = new SessionManager();
  }
  return instance;
 }

 public void addSession(Session session) {
  sessions.add(session);
 }

 public void removeSession(Session session) {
  sessions.remove(session);
 }

 public void writeMessageToClients(Session session, String message) {
  synchronized (session) {
   for (Session clientSession : sessions) {
    if (session != clientSession) {
     try {
      clientSession.getBasicRemote().sendText(message);
     } catch (IOException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
     }
    }
   }
  }
 }
}

NIO2.0 Asynchronous I/O. WebSocket chat server

2016-03-11T00:09:00.002+09:00

Для лучшего понимания NIO2.0 есть очень подходящая задача. А именно попробуем реализовать WebSocket chat server. В качестве клиента будет выступать браузер. Если кратко с помощью WebSocket можно создать двунаправленый канал связи, в частности между браузером и сервером. Клиент в общем может быть любым, главное чтобы он поддерживал нужный протокол.

При установлении соединения браузер и сервер обмениваются так называемым рукопожатием.
По сути клиент посылает обычный GET запрос примерного вида

GET / HTTP/1.1
Upgrade: websocket
Connection: Upgrade
Host: 127.0.0.1
Origin: null
Sec-WebSocket-Key: tR+4GeaXR7PuLsQtvlsMTw==
Sec-WebSocket-Version: 13
Sec-WebSocket-Extensions: x-webkit-deflate-frame

Если сервер поддерживает websocket ответит следующим образом

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: ELY6l2B3+S/6vjfrkzFMurz/hhQ=

Браузер при этом оставляет открытым соединение. Что позволяет отправлять и получать данные в любое время.
Важную часть при установлении рукопожатия составляет обмен ключами. Особой магии тут нет.

Клиент посылает ключ в заголовке Sec-WebSocket-Key закодированный в base64.
Серверу нужно взять этот ключ добавить "магическую строку" 258EAFA5-E914-47DA-95CA-C5AB0DC85B11 и построить по итоговой строке SHA-1 хеш, и передать его в заголовке ответа Sec-WebSocket-Accept

После этого сервер и клиент могут обмениваться сообщениями в специфическом формате, так называемыми фреймами.

0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 +-+-+-+-+-------+-+-------------+-------------------------------+ |F|R|R|R| opcode|M| Payload len | Extended payload length | |I|S|S|S| (4) |A| (7) | (16/64) | |N|V|V|V| |S| | (if payload len==126/127) | | |1|2|3| |K| | | +-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - + | Extended payload length continued, if payload len == 127 | + - - - - - - - - - - - - - - - +-------------------------------+ | |Masking-key, if MASK set to 1 | +-------------------------------+-------------------------------+ | Masking-key (continued) | Payload Data | +-------------------------------- - - - - - - - - - - - - - - - + : Payload Data continued ... : + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + | Payload Data continued ... | +---------------------------------------------------------------+

На самом деле полностью для нашего примера реализовывать парсинг не обязательно. На просторах интернета я нашел готовую реализацию парсера фреймов, которая нам подойдет. В любом случае в продакшен такую поделку тащить ни в коем случае нельзя. Для этого лучше подойдут уже готовые имплементации.(Jetty, Spring и тп).

Итак для реализации чата на websocket нам понадобятся следующие CompletedHandler

AcceptCompletionHandler - принимающий подключение от браузера
HandShakeReadCompletionHandler - читающий handshake от клиента
HandShakeWriteCompletionHandler - handshake ответ сервера
ReadCompletionHandler - принимающий сообщение от сервера
WriteCompletionHandler - отправляющий сообщение клиентам
WebSocketServer - main класс

WebSocketServer - запускает основной поток на прием подключенией через AcceptCompletionHandler, хранит все подключения клиентов и позволяет отправить сообщение всем подключеным клиентам.

package com.bssys.ws_nio2;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class WebSocketServer {

 private final List<AsynchronousSocketChannel> connections = Collections
   .synchronizedList(new ArrayList<AsynchronousSocketChannel>());

 public void run() throws IOException {
  ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(10);
  AsynchronousChannelGroup group = AsynchronousChannelGroup.withThreadPool(newFixedThreadPool);

  final AsynchronousServerSocketChannel listener = AsynchronousServerSocketChannel.open(group);
  InetSocketAddress address = new InetSocketAddress("localhost", 3333);
  listener.bind(address);

  AcceptCompletionHandler acceptCompletionHandler = new AcceptCompletionHandler(listener, this);

  listener.accept(null, acceptCompletionHandler);
 }

 public void addClient(AsynchronousSocketChannel client) {
  connections.add(client);
 }

 public void removeClient(AsynchronousSocketChannel channel) {
  connections.remove(channel);
 }

 public void writeMessageToClients(AsynchronousSocketChannel channel, String message) {
  synchronized (connections) {
   for (AsynchronousSocketChannel clientConnection : connections) {
    if (clientConnection != channel) {
     try {
      byte[] doWrite = WriteFrame.doWrite(message);
      ByteBuffer outputBuffer = ByteBuffer.wrap(doWrite);
      SessionState sessionState = new SessionState();
      clientConnection.write(outputBuffer, sessionState, new WriteCompletionHandler());
     } catch (IOException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
     }
    }
   }
  }
 }

 public static void main(String[] args) throws IOException {
  System.out.println("WebSocket server start...");
  new WebSocketServer().run();
  try {
   Thread.currentThread().join();
  } catch (InterruptedException e) {
   e.printStackTrace();
  }
 }
}

AcceptCompletionHandler - при успешном подключении клиента, запускает асинхронный обмен рукопожатием через HandShakeReadCompletionHandler

package com.bssys.ws_nio2;

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;

public class AcceptCompletionHandler implements CompletionHandler<AsynchronousSocketChannel, Void> {

 private AsynchronousServerSocketChannel listener;
 private WebSocketServer server;

 public AcceptCompletionHandler(AsynchronousServerSocketChannel listener, WebSocketServer server) {
  this.listener = listener;
  this.server = server;
 }

 public void completed(AsynchronousSocketChannel socketChannel, Void arg1) {
  System.out.println("client connected: " + socketChannel);

  listener.accept(null, this);

  ByteBuffer inputBuffer = ByteBuffer.allocate(2048);
  HandShakeReadCompletionHandler readCompletionHandler = new HandShakeReadCompletionHandler(socketChannel,
    inputBuffer, server);
  SessionState sessionState = new SessionState();
  socketChannel.read(inputBuffer, sessionState, readCompletionHandler);
 }

 public void failed(Throwable arg0, Void arg1) {
 }
}

HandShakeReadCompletionHandler - читает переданное сообщение от клиента на установление рукопожатия. Генерирует ключ, подготавливает ответное сообщение handshake и оправляет его асинхронно с помощью HandShakeWriteCompletionHandler

package com.bssys.ws_nio2;

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.security.NoSuchAlgorithmException;
import java.util.Map;

public class HandShakeReadCompletionHandler implements CompletionHandler<Integer, SessionState> {

 private AsynchronousSocketChannel socketChannel;
 private ByteBuffer inputBuffer;
 private WebSocketServer server;

 public HandShakeReadCompletionHandler(AsynchronousSocketChannel socketChannel, ByteBuffer inputBuffer,
   WebSocketServer server) {
  this.socketChannel = socketChannel;
  this.inputBuffer = inputBuffer;
  this.server = server;
 }

 public void completed(Integer bytesRead, SessionState sessionState) {

  byte[] buffer = new byte[bytesRead];
  inputBuffer.rewind();

  inputBuffer.get(buffer);
  String message = new String(buffer);

  System.out.println("Received message from client : " + message);

  HttpRequest httpRequest = new HttpRequest(message);
  Map<String, String> headers = httpRequest.getHeaders();

  inputBuffer.clear();

  String websocketKey = headers.get("Sec-WebSocket-Key").trim();
  try {
   String websocketAcceptKey = KeyGeneration.getKey(websocketKey);
   String response = String.format("HTTP/1.1 101 Switching Protocols\r\n" + "Upgrade: websocket\r\n"
     + "Connection: Upgrade\r\n" + "Sec-WebSocket-Accept: %s\r\n\r\n", websocketAcceptKey);
   ByteBuffer wrap = ByteBuffer.wrap(response.getBytes());
   HandShakeWriteCompletionHandler handShakeWriteCompletionHandler = new HandShakeWriteCompletionHandler(
     socketChannel, server);
   socketChannel.write(wrap, sessionState, handShakeWriteCompletionHandler);

  } catch (NoSuchAlgorithmException e) {
   e.printStackTrace();
  }
 }

 public void failed(Throwable exc, SessionState sessionState) {
 }
}

HandShakeWriteCompletionHandler -при успешной отправке handshake, сохраняет поключение, и запускает асинхронное чтение канала клиента. через ReadCompletionHandler

package com.bssys.ws_nio2;

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;

public class HandShakeWriteCompletionHandler implements CompletionHandler<Integer, SessionState> {

 private AsynchronousSocketChannel socketChannel;
 private WebSocketServer server;

 public HandShakeWriteCompletionHandler(AsynchronousSocketChannel socketChannel, WebSocketServer server) {
  this.socketChannel = socketChannel;
  this.server = server;
 }

 public void completed(Integer result, SessionState sessionState) {
  server.addClient(socketChannel);
  ByteBuffer inputBuffer = ByteBuffer.allocate(1024);
  socketChannel.read(inputBuffer, sessionState, new ReadCompletionHandler(socketChannel, inputBuffer, server));
 }

 public void failed(Throwable exc, SessionState sessionState) {
 }
}

ReadCompletionHandler - Читает сообщение, декодирует его, отправляет его всем подключеным клиентам, и снова запускает чтение из канала, используя самого себя в качестве CompletionHandler. Там образом организуется цикл чтения сообщения от клиента. Все работает асинхронно.

package com.bssys.ws_nio2;

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;

public class ReadCompletionHandler implements CompletionHandler<Integer, SessionState> {

 private AsynchronousSocketChannel channel;
 private ByteBuffer inputBuffer;
 private WebSocketServer server;

 public ReadCompletionHandler(AsynchronousSocketChannel channel, ByteBuffer inputBuffer, WebSocketServer server) {
  this.channel = channel;
  this.inputBuffer = inputBuffer;
  this.server = server;
 }

 public void completed(Integer bytesRead, SessionState session) {
  if (bytesRead < 1) {
   System.out.println("Closing connection to " + channel);
   server.removeClient(channel);
  } else {

   byte[] buffer = new byte[bytesRead];
   inputBuffer.rewind();
   inputBuffer.get(buffer);

   byte[] decodeFrame = RawParse.parse(buffer).payload;
   String message = new String(decodeFrame);

   System.out.println("Received message from " + ":" + message);
   inputBuffer.clear();

   channel.read(inputBuffer, session, this);

   server.writeMessageToClients(channel, message);
  }
 }

 public void failed(Throwable exc, SessionState attachment) {
  server.removeClient(channel);
 }
}

Весь код вместе с клиентом на js на gihub
https://github.com/Hibernate2009/nio-websocket-server.git

Ссылки которые мне очень помогли самому разобраться с websocket'ами.

https://websockets.wordpress.com/the-lightweight-websockets-java-server/
http://stackoverflow.com/questions/18368130/how-to-parse-and-validate-a-websocket-frame-in-java
http://learn.javascript.ru/websockets

Understanding NIO2.0 Asynchronous I/O. EchoServer

2016-03-03T18:48:00.000+09:00

Посмотрим внимательно на NIO2.0 асинхронную сокетную модель появившуюся еще в Java 7 и напишем полностью асинхронный сервер и клиент по типу request/reply.
Асинхронный IO для сокетного взаимодействия представлен следующими классами:

AsynchronousSocketChannel
AsynchronousServerSocketChannel

Есть два подхода работы с ними. Используя:

Future
CompletionHandler

Использование Future предполагает следующие действия

Использование блокирующего вызова get(), который сразу вернет результат
```
AsynchronousSocketChannel worker = future.get();
```

Проверка результата выполнения

while(!future.isDone()) {
 // do something
        Thread.sleep(100);
}

Работа с Future моделью в любом случае предполагает блокировку потока выполнения. Я вообще не вижу смысла в таком случае в использовании асинхронного IO. Какой в смысл, если в результате у нас все равно возникают блокировки, и при этом нужно будет выносить процессинг в отдельный поток. Такое может понадобиться только в самых простых случаях.

Очевидно что для полного асинхронного взаимодействия надо использовать CompletionHandler. При этом надо действовать аккуратно чтобы не свалиться calback hell.

Интерфейс CompletionHandler следующий:

interface CompletionHandler<V,A> {
 void completed(V result, A attachment);
 void failed(Throwable exc, A attachment);
}

V = тип результирующего значения
A = тип объекта вложения при выполнении операции. Используется для того чтобы передавать контекст. Фактически это сессия текущего запроса.

Для реализации сервера нам в общем случае понадобится 3 реализации CompletionHandler

для обработки подключения клиента
CompletionHandler<AsynchronousSocketChannel, ? super A>
для чтения запроса клиента
CompletionHandler<Integer, ? super A>
для отправки ответа клиенту
CompletionHandler<Integer, ? super A>

Для клиента аналогично

для подключения к серверу
CompletionHandler<Void, ? super A>
для отправки запроса на сервер
CompletionHandler<Integer, ? super A>
для чтения ответа серверу
CompletionHandler<Integer, ? super A>

Итак вот шаблон реализации асинхронного сервера calback hell исполнении. Инициализацию и поднятие сервера опустим, ее вы можете увидеть в итоговой реализации.

listener.accept(state, new CompletionHandler<AsynchronousSocketChannel, SessionState>() {
 @Override
 public void completed(final AsynchronousSocketChannel socketChannel, SessionState attachment) {
  ByteBuffer inputBuffer = ByteBuffer.allocate(1024);
  socketChannel.read(inputBuffer, attachment, new CompletionHandler<Integer, SessionState>(){
   @Override
   public void completed(Integer result, SessionState attachment) {
    ByteBuffer outputBuffer = ByteBuffer.allocate(1024);;
    socketChannel.write(outputBuffer, attachment,new CompletionHandler<Integer, SessionState>(){
     @Override
     public void completed(Integer result, SessionState attachment) {
      
     }
     @Override
     public void failed(Throwable exc, SessionState attachment) {
      
     }
    });
   }
   @Override
   public void failed(Throwable exc, SessionState attachment) {
    
   }
  });
 }
 @Override
 public void failed(Throwable exc, SessionState attachment) {
  
 }
});

Конечно код по такому шаблону неудобно сопровождать. Лучше всего разнести реализацию CompletionHandler в отдельные классы. Итак вот итоговый результат EchoServer. Исходный код https://github.com/Hibernate2009/nio-echoserver.git
EchoServer

package com.bssys.nio2.server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.bssys.nio2.SessionState;

public class EchoServer {

 public void run() throws IOException {
  ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(10);
  AsynchronousChannelGroup group = AsynchronousChannelGroup.withThreadPool(newFixedThreadPool);

  final AsynchronousServerSocketChannel listener = AsynchronousServerSocketChannel.open(group);
  InetSocketAddress address = new InetSocketAddress("localhost", 3333);
  listener.bind(address);

  AcceptCompletionHandler acceptCompletionHandler = new AcceptCompletionHandler(listener);

  SessionState state = new SessionState();
  listener.accept(state, acceptCompletionHandler);
  
 }

 public static void main(String[] args) throws IOException {
  System.out.println("Async server started");
  new EchoServer().run();
  try {
   Thread.currentThread().join();
  } catch (InterruptedException e) {
   e.printStackTrace();
  }
 }
}

AcceptCompletionHandler

package com.bssys.nio2.server;

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;

import com.bssys.nio2.SessionState;

public class AcceptCompletionHandler implements CompletionHandler<AsynchronousSocketChannel, SessionState> {

 private AsynchronousServerSocketChannel listener;

 public AcceptCompletionHandler(AsynchronousServerSocketChannel listener) {
  this.listener = listener;
 }

 @Override
 public void completed(AsynchronousSocketChannel socketChannel, SessionState sessionState) {
  // accept the next connection
  SessionState newSessionState = new SessionState();
  listener.accept(newSessionState, this);

  // handle this connection
  ByteBuffer inputBuffer = ByteBuffer.allocate(2048);
  ServerReadCompletionHandler readCompletionHandler = new ServerReadCompletionHandler(socketChannel, inputBuffer);
  socketChannel.read(inputBuffer, sessionState, readCompletionHandler);
 }

 @Override
 public void failed(Throwable exc, SessionState sessionState) {
  // Handle connection failure...
 }
}

ServerReadCompletionHandler

package com.bssys.nio2.server;

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;

import com.bssys.nio2.SessionState;

public class ServerReadCompletionHandler implements CompletionHandler<Integer, SessionState> {

 private AsynchronousSocketChannel socketChannel;
 private ByteBuffer inputBuffer;

 public ServerReadCompletionHandler(AsynchronousSocketChannel socketChannel, ByteBuffer inputBuffer) {
  this.socketChannel = socketChannel;
  this.inputBuffer = inputBuffer;
 }

 @Override
 public void completed(Integer bytesRead, SessionState sessionState) {

  byte[] buffer = new byte[bytesRead];
  inputBuffer.rewind();

  inputBuffer.get(buffer);
  String message = new String(buffer);

  System.out.println("Received message from client : " + message);
  
  ServerWriteCompletionHandler writeCompletionHandler = new ServerWriteCompletionHandler(socketChannel);
  
  ByteBuffer outputBuffer = ByteBuffer.wrap(buffer);
  socketChannel.write(outputBuffer, sessionState, writeCompletionHandler);
 }

 @Override
 public void failed(Throwable exc, SessionState attachment) {
  // Handle read failure.....
 }
}

ServerWriteCompletionHandler

package com.bssys.nio2.server;

import java.io.IOException;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;

import com.bssys.nio2.SessionState;

public class ServerWriteCompletionHandler implements CompletionHandler<Integer, SessionState> {

 private AsynchronousSocketChannel socketChannel;
 

 public ServerWriteCompletionHandler(AsynchronousSocketChannel socketChannel) {
  this.socketChannel = socketChannel;
 }

 @Override
 public void completed(Integer bytesWritten, SessionState sessionState) {
  try {
   socketChannel.close();
  } catch (IOException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }

 @Override
 public void failed(Throwable exc, SessionState attachment) {
  // Handle write failure.....
 }
}

Неплохая презентация по работе с буферами http://www.kgeorgiy.info/courses/java-advanced/slides/nio.xhtml#(1)

Client notification using ZeroMQ

2016-02-25T20:51:00.001+09:00

Итак нам понадобилось реализовать механизм рассылки сообщений. Дополнительных требований таких как гарантированная доставка не предъявляется. Фактически нам подойдет примитивное сокетное взаимодействие.
На практике обычно стоит обратная задача, когда надо реализовать сервер выдерживающий большую нагрузку. Есть множество решений как ее решать, используя традиционные блокирующие сокеты, или что будет лучшим выбором, используя асинхронное взаимодействие NIO в экосистеме java. Например можно взять готовый клиент-серверный фремворк netty.
У нас же стоит обратная задача, сервер должен уведомлять большое количество клиентов.
В enterpise системах, для этой цели обычно используются сиcтемы MQ. Но в нашем pet-project можно использовать кое-что поинтереснее.

Используя для нашей задачи традиционные сокеты, нам бы потребовалось заводить в отдельный поток каждый сокет клиента, и при наступлении события перебирать их в цикле, решая в том числе проблемы с дисконектами. А сколько при этом потребуется ресурсов, если клиентов будет хотя бы больше 1000.
И вот для этого нам идеально подойдет библиотека ZeroMQ.
Подробно рассказывать о ней я не буду, информации в интернетах достаточно. Отмечу просто что кроме традиционного взаимодействия REQUEST/REPLY библиотека предоставляет возможность взаимодействия по типу PUBLISH/SUBSCRIBE.

Это то что нам нужно.
Итак архитектура механизма доставки следующая:

PubServer.java - сервер доставки получает сообщение из ActiveMQ в примитивном формате из которого получает имя топик и само сообщение. И собственно отправляет его в топик нужному клиенту.
SubClient.java- принимает имя топика из которого будет получать сообщения устанавливает соединение с сервером и подписывается на топик.

Реализация ZeroMQ предоставлена официально поддерживамой библиотекой jeromq для java.

package com.bssys.zeromq;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

public class PubServer {

 private Socket publisher = null;

 public PubServer() throws Exception {
  initJMSConnection();
  initZMQ();
 }

 private void initZMQ() {
  Context context = ZMQ.context(1);
  publisher = context.socket(ZMQ.PUB);
  publisher.bind("tcp://*:5563");

  Runtime.getRuntime().addShutdownHook(new Thread() {
   public void run() {
    System.out.println("ShutDown");
    if (publisher != null) {
     publisher.close();
    }
    if (context != null) {
     context.term();
    }
   }
  });
 }

 private void initJMSConnection() {
  Connection connection = null;
  Session session = null;
  MessageConsumer consumer = null;
  try {
   ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
   connection = connectionFactory.createConnection();
   connection.start();
   session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   Destination adminQueue = session.createQueue("TEST_QUEUE");
   consumer = session.createConsumer(adminQueue);
   for (int i = 0; i <= 10; i++) {
    consumer.setMessageListener(new ConsumerMessageListener());
   }

  } catch (Exception e) {
   if (consumer != null) {
    try {
     consumer.close();
    } catch (JMSException e1) {
     e1.printStackTrace();
    }
   }
   if (session != null) {
    try {
     session.close();
    } catch (JMSException e1) {
     e1.printStackTrace();
    }
   }
   if (connection != null) {
    try {
     connection.close();
    } catch (JMSException e1) {
     e1.printStackTrace();
    }
   }
   throw new IllegalStateException("JMS connection fail");
  }
 }

 public synchronized void send(String topic, String msg) {
  publisher.sendMore(topic);
  publisher.send(msg);
 }

 class ConsumerMessageListener implements MessageListener {

  @Override
  public void onMessage(Message jmsMessage) {
   System.out.println(Thread.currentThread().getName()+" recive message");
   if (jmsMessage instanceof TextMessage) {
    TextMessage textMessage = (TextMessage) jmsMessage;
    try {
     String text = textMessage.getText();
     String[] split = text.split(":");
     if (split != null && split.length > 1) {
      String topic = split[0];
      String msg = split[1];
      send(topic, msg);
     }
    } catch (JMSException e) {
     e.printStackTrace();
    }
   }
  }
 }

 public static void main(String arg[]) {
  try {
   new PubServer();
  } catch (Exception e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }
}

package com.bssys.zeromq;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

public class SubClient implements Runnable {

 private Context context;
 private Socket subscriber;
 
 private String topic;

 public SubClient(String topic) {
  this.topic = topic;
  context = ZMQ.context(1);
  subscriber = context.socket(ZMQ.SUB);
  
  ExecutorService newCachedThreadPool = Executors.newCachedThreadPool();
  newCachedThreadPool.submit(this);
  
  Runtime.getRuntime().addShutdownHook(new Thread() {
   public void run() {
    System.out.println("ShutDown");
    subscriber.close();
    context.term();
   }
  });

 }

 @Override
 public void run() {
  subscriber.connect("tcp://localhost:5563");
  subscriber.subscribe(topic.getBytes());
  while (!Thread.currentThread().isInterrupted()) {
   // Read envelope with address
   String address = subscriber.recvStr();
   // Read message contents
   String contents = subscriber.recvStr();
   System.out.println(address + " : " + contents);
  }
 }
 
 public static void main(String args[]){
  if (args!=null && args.length>0){
   new SubClient(args[0]);
  }else {
   System.out.println("java SubClient A");
  }
 }

}

Authentication and authorization ActiveMQ

2016-02-09T01:00:00.001+09:00

Итак, если вам надо управлять доступом к очередям на ActiveMQ, предоставляю пример как это можно сделать. Может кому пригодится. В файле activemq.xml делаем следующее

<plugins>
 <simpleAuthenticationPlugin>
  <users>
   <authenticationUser username="admin" password="admin" groups="producers,consumers,admins" />
   <authenticationUser username="producer" password="producer" groups="producers" />
   <authenticationUser username="consumer" password="consumer" groups="consumers" />
  </users>
 </simpleAuthenticationPlugin>
 <authorizationPlugin>
  <map>
   <authorizationMap>
    <authorizationEntries>
     <authorizationEntry queue=">" write="producers" read="consumers" admin="admins" />
     <authorizationEntry topic="ActiveMQ.Advisory.>" write="producers,consumers" read="producers,consumers" admin="admins" />
    </authorizationEntries>
    <tempDestinationAuthorizationEntry>
     <tempDestinationAuthorizationEntry write="producers" read="consumers" admin="admins" />
    </tempDestinationAuthorizationEntry>
   </authorizationMap>
  </map>
 </authorizationPlugin>
</plugins>

Теперь положить сообщение в очередь может пользователь producer, а прочитать consumer

Best practices ServiceMix and Apache Camel

2016-01-12T20:34:00.000+09:00

В этой статье я решил поделится всеми своими рекомендациями при разработке с применением ServiceMix и Apache Camel. По мере развития статья будет дополняться.

Используйте Spring DSL вместо Java DSL. Это позволит в некоторых случаях при необходимости вести разработку в обычном текстовом редакторе, избавит вас от format-hell. Так же в силу древовидности XML вам не придется использовать костыли по типу end(), которые вы вынуждены вставлять чтобы контролировать блоки кода.
При подключении к ActiveMQ всегда используйте failover (failover:tcp://127.0.0.1:61616)

При использовании файлового адаптера читающего файлы из директории, всегда подключайте фильтр на проверку пустого файла.

import org.apache.camel.component.file.GenericFile;
import org.apache.camel.component.file.GenericFileFilter;
import org.apache.log4j.Logger;

public class EmptyFilter<T> implements GenericFileFilter<T> {
 private static Logger log = Logger.getLogger(EmptyFilter.class);

 public boolean accept(GenericFile<T> file) {
  boolean res = false;
  if (!file.isDirectory()) {
   if (file.getFileLength() > 0) {
    res = true;
   }
  }
  return res;
 }
}

Иcпользование:

file:{{inbox}}?filter=#emptyFilter

Итерирование по коллекциям удобно делать через http://camel.apache.org/splitter.html из коробки предоставляющий возможность параллелизации
```
<split parallelProcessing="true">
 <xpath>/items/*</xpath>
        <to uri="direct:foo" />
</split>
```

Все используемые библиотеке выносите в отдельный файл feature. И никогда не носите библиотеки внутри разрабатываемых вами bundle. Любой jar можно установить используя wrap.

<?xml version="1.0" encoding="UTF-8"?>
<features>
  <feature name='tir_config' version='1.0' install='auto'>
    <feature version="3.2.11.RELEASE_1">spring-jdbc</feature>
    <feature>jdbc</feature>
    <feature>jetty</feature>
    <feature>camel-quartz</feature>
    <feature>activemq-broker</feature>
    <feature>camel-jetty</feature>
    <feature>camel-jackson</feature>
    <feature>camel-urlrewrite</feature>
    <feature>camel-saxon</feature>
    <feature>activemq-web-console</feature>
    <bundle>wrap:mvn:org.apache.httpcomponents/httpcore/4.3.3</bundle>
    <bundle>wrap:mvn:org.apache.httpcomponents/httpclient/4.3.6</bundle>
    <bundle>wrap:mvn:org.apache.commons/commons-pool2/2.4.2</bundle>
    <bundle>mvn:org.apache.servicemix.bundles/org.apache.servicemix.bundles.jedis/2.7.3_1</bundle>
  </feature>
</features>

JDBC-источники данных(DataSource) удобно деплоить как blueprint и не городить отдельный bundle, который будет доступен как osgi-сервис. Также всегда добавляйте свойство osgi.jndi.service.name, это позволит обращаться к источнику также через JNDI. Файл достаточно положить в папку deploy. Это представляет настройки в чистом виде, без необходимости выводить их в отдельный настроечный файл

<?xml version="1.0" encoding="UTF-8"?>
<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:cm="http://aries.apache.org/blueprint/xmlns/blueprint-cm/v1.0.0">
 <bean id="postgresPoolingDS" class="org.apache.commons.dbcp.BasicDataSource">
  <property name="driverClassName" value="org.postgresql.Driver" />
  <property name="url" value="jdbc:postgresql://10.1.6.66/postgres" />
  <property name="username" value="postgres" />
  <property name="password" value="postgres" />
  <property name="defaultAutoCommit" value="true" />
  <property name="testWhileIdle" value="false" />
  <property name="testOnBorrow" value="false" />
  <property name="testOnReturn" value="false" />
  <property name="timeBetweenEvictionRunsMillis" value="5000" />
  <property name="maxActive" value="10" />
  <property name="maxIdle" value="10" />
  <property name="minIdle" value="0" />
  <property name="maxWait" value="10000" />
  <property name="initialSize" value="1" />
  <property name="removeAbandonedTimeout" value="10" />
  <property name="removeAbandoned" value="true" />
  <property name="logAbandoned" value="false" />
  <property name="minEvictableIdleTimeMillis" value="30000" />
 </bean>
 <service interface="javax.sql.DataSource" ref="postgresPoolingDS" >
  <service-properties>
   <entry key="datasource.name" value="postgres" />
   <entry key="osgi.jndi.service.name" value="jdbc/postgresds" />
  </service-properties>
 </service>
</blueprint>

Связи между модулями реализуйте через osgi-сервисы. Зависимостями можно управлять через настроечные файлы. Таким образом можно например настраивать используемый источник данных (DataSource)
```
<osgi:reference id="dataSource" interface="javax.sql.DataSource" filter="(datasource.name=${dbType})" />
```
Использование osgi-сервисов избавит вас от головной боли связанной с очередностью запуска модулей. Они сами будут разруливать свои зависимости, и ждать нужного osgi-сервиса(Модуль будет находиться в состоянии WAIT). В отличии от использования кода напрямую, когда модуль просто свалится (FAIL), и даже появление нужной зависимости не заставит его подняться. Необходимо будет делать restart вручную. Если какая-то часть кода просится в библиотеку, выносите его в отдельный jar и подключайте в файле features, про который мы уже сказали выше.

ActiveMQ Master/Slave and load balancer cluster

2014-04-28T19:25:00.003+10:00

Пришло время объединить все технологии кластеризации ActiveMQ вместе, и продемонстрировать систему устойчивую к сбоям и позволяющую распределять нагрузку. Для примера построим архитектуру кластера на базе 8 инстансов ActiveMQ. Схема выглядит следующим образом.

Рассмотрим детально:

Инстансы связаные общим хранилищем Master/Slave 1-2, 3-4, 5-6, 7-8. По идее каждая пара разворачивается на отдельной машине. Для возможности развертывания на отдельной машине, все инстансы разнесены по портам.
С помощью Network Broker каждая пара с связана с каждой парой.
Для каждой пары на этом же хосте, с помощью Apache Camel поднимается HTTP точка, принимающая запросы от внешнего балансировщика и прокидывающая запрос в очередь.
Все инстансы из каждой группы могут делиться сообщения с другими группами
Отдельные клиенты разворачиваются каждый на своем хосте и читают сообщения и очереди.
Балансировщик для простоты тоже реализован с помощью Apache Camel

Ниже можно взглянуть на настройки каждого инстанса и исходный код.

instance1:

<networkConnectors>
 <networkConnector name="instance1-instance3-instance4" uri="masterslave:(tcp://localhost:61619,tcp://localhost:61620)" />
 <networkConnector name="instance1-instance5-instance6" uri="masterslave:(tcp://localhost:61621,tcp://localhost:61622)" />
 <networkConnector name="instance1-instance7-instance8" uri="masterslave:(tcp://localhost:61623,tcp://localhost:61624)" />
</networkConnectors>

<persistenceAdapter>
 <kahaDB directory="${activemq.data}/../../../kahadb1" />
</persistenceAdapter>

instance2:

<networkConnectors>
 <networkConnector name="instance2-instance3-instance4" uri="masterslave:(tcp://localhost:61619,tcp://localhost:61620)" />
 <networkConnector name="instance2-instance5-instance6" uri="masterslave:(tcp://localhost:61621,tcp://localhost:61622)" />
 <networkConnector name="instance2-instance7-instance8" uri="masterslave:(tcp://localhost:61623,tcp://localhost:61624)" />
</networkConnectors>

<persistenceAdapter>
 <kahaDB directory="${activemq.data}/../../../kahadb1" />
</persistenceAdapter>

instance3:

<networkConnectors>
 <networkConnector name="instance3-instance1-instance2" uri="masterslave:(tcp://localhost:61617,tcp://localhost:61618)" />
 <networkConnector name="instance3-instance5-instance6" uri="masterslave:(tcp://localhost:61621,tcp://localhost:61622)" />
 <networkConnector name="instance3-instance7-instance8" uri="masterslave:(tcp://localhost:61623,tcp://localhost:61624)" />
</networkConnectors>

<persistenceAdapter>
 <kahaDB directory="${activemq.data}/../../../kahadb2" />
</persistenceAdapter>

instance4:

<networkConnectors>
 <networkConnector name="instance4-instance1-instance2" uri="masterslave:(tcp://localhost:61617,tcp://localhost:61618)" />
 <networkConnector name="instance4-instance5-instance6" uri="masterslave:(tcp://localhost:61621,tcp://localhost:61622)" />
 <networkConnector name="instance4-instance7-instance8" uri="masterslave:(tcp://localhost:61623,tcp://localhost:61624)" />
</networkConnectors>

<persistenceAdapter>
 <kahaDB directory="${activemq.data}/../../../kahadb2" />
</persistenceAdapter>

instance5:

<networkConnectors>
 <networkConnector name="instance5-instance1-instance2" uri="masterslave:(tcp://localhost:61617,tcp://localhost:61618)" />
 <networkConnector name="instance5-instance3-instance4" uri="masterslave:(tcp://localhost:61619,tcp://localhost:61620)" />
 <networkConnector name="instance5-instance7-instance8" uri="masterslave:(tcp://localhost:61623,tcp://localhost:61624)" />
</networkConnectors>

<persistenceAdapter>
 <kahaDB directory="${activemq.data}/../../../kahadb3" />
</persistenceAdapter>

instance6:

<networkConnectors>
 <networkConnector name="instance6-instance1-instance2" uri="masterslave:(tcp://localhost:61617,tcp://localhost:61618)" />
 <networkConnector name="instance6-instance3-instance4" uri="masterslave:(tcp://localhost:61619,tcp://localhost:61620)" />
 <networkConnector name="instance6-instance7-instance8" uri="masterslave:(tcp://localhost:61623,tcp://localhost:61624)" />
</networkConnectors>

<persistenceAdapter>
 <kahaDB directory="${activemq.data}/../../../kahadb3" />
</persistenceAdapter>

instance7:

<networkConnectors>
 <networkConnector name="instance7-instance1-instance2" uri="masterslave:(tcp://localhost:61617,tcp://localhost:61618)" />
 <networkConnector name="instance7-instance3-instance4" uri="masterslave:(tcp://localhost:61619,tcp://localhost:61620)" />
 <networkConnector name="instance7-instance5-instance6" uri="masterslave:(tcp://localhost:61621,tcp://localhost:61622)" />
</networkConnectors>

<persistenceAdapter>
 <kahaDB directory="${activemq.data}/../../../kahadb4" />
</persistenceAdapter>

instance8:

<networkConnectors>
 <networkConnector name="instance8-instance1-instance2" uri="masterslave:(tcp://localhost:61617,tcp://localhost:61618)" />
 <networkConnector name="instance8-instance3-instance4" uri="masterslave:(tcp://localhost:61619,tcp://localhost:61620)" />
 <networkConnector name="instance8-instance5-instance6" uri="masterslave:(tcp://localhost:61621,tcp://localhost:61622)" />
</networkConnectors>

<persistenceAdapter>
 <kahaDB directory="${activemq.data}/../../../kahadb4" />
</persistenceAdapter>

Настройки ресурсов для клиента, я объединил в один модуль. В реальном случае настройки будут разнесены.

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:osgi="http://www.springframework.org/schema/osgi"
 xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd">

 <osgi:reference id="platformTransactionManager"
  interface="org.springframework.transaction.PlatformTransactionManager" />


 <bean id="instanceConnectionFactory1" class="org.apache.activemq.ActiveMQConnectionFactory">
  <property name="brokerURL"
   value="failover://(tcp://localhost:61617,tcp://localhost:61618)?initialReconnectDelay=100" />
  <property name="userName" value="karaf" />
  <property name="password" value="karaf" />
  <property name="redeliveryPolicy">
   <bean class="org.apache.activemq.RedeliveryPolicy">
    <property name="maximumRedeliveries" value="0" />
   </bean>
  </property>
 </bean>

 <bean id="instanceConnectionFactory2" class="org.apache.activemq.ActiveMQConnectionFactory">
  <property name="brokerURL"
   value="failover://(tcp://localhost:61619,tcp://localhost:61620)?initialReconnectDelay=100" />
  <property name="userName" value="karaf" />
  <property name="password" value="karaf" />
  <property name="redeliveryPolicy">
   <bean class="org.apache.activemq.RedeliveryPolicy">
    <property name="maximumRedeliveries" value="0" />
   </bean>
  </property>
 </bean>

 <bean id="instanceConnectionFactory3" class="org.apache.activemq.ActiveMQConnectionFactory">
  <property name="brokerURL"
   value="failover://(tcp://localhost:61621,tcp://localhost:61622)?initialReconnectDelay=100" />
  <property name="userName" value="karaf" />
  <property name="password" value="karaf" />
  <property name="redeliveryPolicy">
   <bean class="org.apache.activemq.RedeliveryPolicy">
    <property name="maximumRedeliveries" value="0" />
   </bean>
  </property>
 </bean>

 <bean id="instanceConnectionFactory4" class="org.apache.activemq.ActiveMQConnectionFactory">
  <property name="brokerURL"
   value="failover://(tcp://localhost:61623,tcp://localhost:61624)?initialReconnectDelay=100" />
  <property name="userName" value="karaf" />
  <property name="password" value="karaf" />
  <property name="redeliveryPolicy">
   <bean class="org.apache.activemq.RedeliveryPolicy">
    <property name="maximumRedeliveries" value="0" />
   </bean>
  </property>
 </bean>

 <bean id="jmsConfig1"
  class="org.apache.activemq.camel.component.ActiveMQConfiguration">
  <property name="connectionFactory" ref="instanceConnectionFactory1" />
  <property name="transactionManager" ref="platformTransactionManager" />
  <property name="transacted" value="true" />
 </bean>

 <bean id="jmsConfig2"
  class="org.apache.activemq.camel.component.ActiveMQConfiguration">
  <property name="connectionFactory" ref="instanceConnectionFactory2" />
  <property name="transactionManager" ref="platformTransactionManager" />
  <property name="transacted" value="true" />
 </bean>

 <bean id="jmsConfig3"
  class="org.apache.activemq.camel.component.ActiveMQConfiguration">
  <property name="connectionFactory" ref="instanceConnectionFactory3" />
  <property name="transactionManager" ref="platformTransactionManager" />
  <property name="transacted" value="true" />
 </bean>

 <bean id="jmsConfig4"
  class="org.apache.activemq.camel.component.ActiveMQConfiguration">
  <property name="connectionFactory" ref="instanceConnectionFactory4" />
  <property name="transactionManager" ref="platformTransactionManager" />
  <property name="transacted" value="true" />
 </bean>

 <bean id="instance1" class="org.apache.activemq.camel.component.ActiveMQComponent">
  <constructor-arg ref="jmsConfig1" />
 </bean>
 <bean id="instance2" class="org.apache.activemq.camel.component.ActiveMQComponent">
  <constructor-arg ref="jmsConfig2" />
 </bean>
 <bean id="instance3" class="org.apache.activemq.camel.component.ActiveMQComponent">
  <constructor-arg ref="jmsConfig3" />
 </bean>
 <bean id="instance4" class="org.apache.activemq.camel.component.ActiveMQComponent">
  <constructor-arg ref="jmsConfig4" />
 </bean>

 <bean id="PROPAGATION_REQUIRED" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
  <property name="transactionManager" ref="platformTransactionManager" />
  <property name="propagationBehaviorName" value="PROPAGATION_REQUIRED" />
 </bean>
 <bean id="PROPAGATION_REQUIRES_NEW" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
  <property name="transactionManager" ref="platformTransactionManager" />
  <property name="propagationBehaviorName" value="PROPAGATION_REQUIRES_NEW" />
 </bean>

</beans>

Непосредственно код клиента на Apache Camel

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:camel="http://camel.apache.org/schema/spring"
 xmlns:osgi="http://www.springframework.org/schema/osgi"
 xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

 <camelContext xmlns="http://camel.apache.org/schema/spring">
  <route>
   <from uri="jetty:http://localhost:8191/myapp/myservice" />
   <to uri="instance1:queue:notify" />
  </route>
  <route>
   <from uri="jetty:http://localhost:8192/myapp/myservice" />
   <to uri="instance2:queue:notify" />
  </route>
  <route>
   <from uri="jetty:http://localhost:8193/myapp/myservice" />
   <to uri="instance3:queue:notify" />
  </route>
  <route>
   <from uri="jetty:http://localhost:8194/myapp/myservice" />
   <to uri="instance4:queue:notify" />
  </route>
  
  <route>
   <from uri="instance1:queue:notify" />
   <log message="instance1:${body}" />
   <to uri="file:C:/data/kraken/instance1" />
  </route>
  <route>
   <from uri="instance2:queue:notify" />
   <log message="instance2:${body}" />
   <to uri="file:C:/data/kraken/instance2" />
  </route>
  <route>
   <from uri="instance3:queue:notify" />
   <log message="instance3:${body}" />
   <to uri="file:C:/data/kraken/instance3" />
  </route>
  <route>
   <from uri="instance4:queue:notify" />
   <log message="instance4:${body}" />
   <to uri="file:C:/data/kraken/instance4" />
  </route>
 </camelContext>
</beans>

И код балансировщика, который отправляет 1000 сообщений, также на Apache Camel.

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:camel="http://camel.apache.org/schema/spring"
 xmlns:osgi="http://www.springframework.org/schema/osgi"
 xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

 <bean id="myDataSet" class="org.apache.camel.component.dataset.SimpleDataSet">
  <property name="size" value="1000" />
 </bean>

 <camelContext xmlns="http://camel.apache.org/schema/spring">
  <route>
   <from uri="dataset:myDataSet?produceDelay=-1" />
   <loadBalance>
    <roundRobin/>
    <to uri="http://localhost:8191/myapp/myservice" />
    <to uri="http://localhost:8192/myapp/myservice" />
    <to uri="http://localhost:8193/myapp/myservice" />
    <to uri="http://localhost:8194/myapp/myservice" />
   </loadBalance>
  </route>
 </camelContext>
 
</beans>

ActiveMQ Master/Slave and Broker Networks cluster

2014-04-24T17:57:00.002+10:00

В предыдущей статье мы рассмотрели высоко-нагруженый кластер ActiveMQ для распределения нагрузки. Однако эта конфигурация все таки обладает следующим недостатком. При выходе из строя одного из узлов ActiveMQ, сообщения размещенные в нем становятся более недоступны, и будут вычитаны только после того как конкретный инстанс ActiveMQ будет снова поднят. Проблема может быть решена с помощью конфигурации Master/Slave, в которой два, или больше инстанса ActiveMQ имеют общее хранилище. Пара Master/Slave может быть связана с аналогичной парой, размещенной на другом хосте, и связаны с помощью Broker Networks. Схематично это выглядит следующим образом.
Для простоты развертывания все производится на одной машине. Инстансы ActiveMQ инстансы разделены по портам.

Ниже будут представлены настройки каждого из узлов кластера

instance1:

<networkConnectors>
 <networkConnector name="instance1-instance3-instance4" uri="masterslave:(tcp://localhost:61619,tcp://localhost:61620)" />
</networkConnectors>

<persistenceAdapter>
 <kahaDB directory="${activemq.data}/../../../kahadb1" />
</persistenceAdapter>

instance2:

<networkConnectors>
 <networkConnector name="instance2-instance3-instance4" uri="masterslave:(tcp://localhost:61619,tcp://localhost:61620)" />
</networkConnectors>

<persistenceAdapter>
 <kahaDB directory="${activemq.data}/../../../kahadb1" />
</persistenceAdapter>

instance3:

<networkConnectors>
 <networkConnector name="instance3-instance1-instance2" uri="masterslave:(tcp://localhost:61617,tcp://localhost:61618)" />
</networkConnectors>

<persistenceAdapter>
 <kahaDB directory="${activemq.data}/../../../kahadb2" />
</persistenceAdapter>

instance4:

<networkConnectors>
 <networkConnector name="instance4-instance1-instance2" uri="masterslave:(tcp://localhost:61617,tcp://localhost:61618)" />
</networkConnectors>

<persistenceAdapter>
 <kahaDB directory="${activemq.data}/../../../kahadb2" />
</persistenceAdapter>

Обратите внимание:

instance1 и instance2 настроены на одну директорию kahadb1, и связаны с помощью networkConnectors с instance3 и instance4.
instance3 и instance4 настроены на одну директорию kahadb2, и связаны с помощью networkConnectors с instance1 и instance2.

клиентский url в таком случае будет выглядеть так:

failover:(tcp://machine1:masterAport,tcp://machine2:masterBport,tcp://machine1:slaveBport,tcp://machine2:slaveAport)?randomize=false

т.е. в нашем случае:

failover:(tcp://localhost:61617,tcp://localhost:61619,tcp://localhost:61618,tcp://localhost:61620)?randomize=false

ActiveMQ high availability load balancer cluster

2014-04-14T22:54:00.000+10:00

Рассмотрим актуальную задачу Load Balancer между инстансами ActiveMQ. Балансировка в таком случае строиться с помощью networkConnectors. Для простоты развертывания инстансы ActiveMQ развернем на одной машине, разделение будет строиться на базе портов. Вот схематичное представление кластера:

Конфигурация каждого инстанса ActiveMQ представлена ниже:

instance1:

<networkConnectors>
 <!-- broker1 -> broker2 -->
 <networkConnector name="instance1-instance2" uri="static:(tcp://localhost:61618)" />
 <!-- broker1 -> broker3 -->
 <networkConnector name="instance1-instance3" uri="static:(tcp://localhost:61619)" />
 <!-- broker1 -> broker4 -->
 <networkConnector name="instance1-instance4" uri="static:(tcp://localhost:61620)" />
</networkConnectors>

instance2:

<networkConnectors>
 <!-- broker2 -> broker1 -->
 <networkConnector name="instance2-instance1" uri="static:(tcp://localhost:61617)" />
 <!-- broker2 -> broker3 -->
 <networkConnector name="instance2-instance3" uri="static:(tcp://localhost:61619)" />
 <!-- broker2 -> broker4 -->
 <networkConnector name="instance2-instance4" uri="static:(tcp://localhost:61620)" />
</networkConnectors>

instance3:

<networkConnectors>
 <!-- broker3 -> broker1 -->
 <networkConnector name="instance3-instance1" uri="static:(tcp://localhost:61617)" />
 <!-- broker3 -> broker2 -->
 <networkConnector name="instance3-instance2" uri="static:(tcp://localhost:61618)" />
 <!-- broker3 -> broker4 -->
 <networkConnector name="instance3-instance4" uri="static:(tcp://localhost:61620)" />
</networkConnectors>

instance4:

<networkConnectors>
 <!-- broker4 -> broker2 -->
 <networkConnector name="instance4-instance1" uri="static:(tcp://localhost:61617)" />
 <!-- broker4 -> broker3 -->
 <networkConnector name="instance4-instance2" uri="static:(tcp://localhost:61618)" />
 <!-- broker4 -> broker4 -->
 <networkConnector name="instance4-instance3" uri="static:(tcp://localhost:61619)" />
</networkConnectors>

URL для consumer и producer будет выглядеть следующим образом:

failover://(tcp://localhost:61617,tcp://localhost:61619,tcp://localhost:61618,tcp://localhost:61620)?randomize=true

Transactional JMS Client

2014-04-03T20:29:00.002+10:00

Разберемся детально с локальными транзакциями на примере взаимодействия JMS очередей (ActiveMQ). В документации по этому поводу говорится, что для того чтобы с route был транзакционным достаточно в начале объявить его с помощью инструкции <transacted/> . Но как будет выглядеть транзакционное поведение умалчивается. Обычно в качестве примера демонстрируется подобный route.

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">

  <camelContext xmlns="http://camel.apache.org/schema/spring">
    <route>
      <from uri="jmstx:queue:okay"/>
      <transacted/>
      <to uri="jmstx:queue:final"/>
    </route>
  </camelContext>
</beans>

Который не демонстрирует ровным счетом ничего.

Модифицируем код следующим образом

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">

  <camelContext xmlns="http://camel.apache.org/schema/spring">
    <route>
      <from uri="jmstx:queue:okay"/>
      <transacted/>
      <to uri="jmstx:queue:final"/>
      <bean ref="businessLogic" method="business">
    </route>
  </camelContext>
</beans>

Поставим посередине вызов bean содержащей бизнес-логику. Сама бизнес-логика простейшая
package com.bssys.transaction_client;

package com.bssys.transaction_client;

import org.apache.log4j.Logger;

public class BusinessLogic {
 private Logger log = Logger.getLogger(BusinessLogic.class);
 
 public void business(String message){
  log.info(message);
  if ("exception".equals(message)){
   log.info("Rollback " + message);
   throw new IllegalArgumentException("This is forced");
  }
 }
}

Задумка такова, что при отправке сообщения содержащего строку "exception" поднимается исключение, которое должно вызвать откат транзакции. Для того чтобы вся эта магия заработала необходимо настроить транзакции. Все транзакционные артефакты я описал в отдельном spring-контексте.

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:osgi="http://www.springframework.org/schema/osgi"
 xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd">

 <osgi:reference id="connectionFactory" interface="javax.jms.ConnectionFactory" />
 <osgi:reference id="platformTransactionManager" interface="org.springframework.transaction.PlatformTransactionManager" />

 <bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
  <property name="connectionFactory" ref="connectionFactory" />
  <property name="transactionManager" ref="platformTransactionManager" />
  <property name="transacted" value="true" />
 </bean>

 <bean id="jmstx" class="org.apache.camel.component.jms.JmsComponent">
  <constructor-arg ref="jmsConfig" />
 </bean>
 
 <bean id="PROPAGATION_REQUIRED" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
  <property name="transactionManager" ref="platformTransactionManager" />
  <property name="propagationBehaviorName" value="PROPAGATION_REQUIRED" />
 </bean>
 
 <bean id="PROPAGATION_REQUIRES_NEW" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
  <property name="transactionManager" ref="platformTransactionManager" />
  <property name="propagationBehaviorName" value="PROPAGATION_REQUIRES_NEW" />
 </bean>

</beans>

Попробуем запустить и проверить. Модуль я запускаю на ServiceMix 5.0.0 последней на данный момент версии. Отправив сообщение "exception" в очередь okay. В логах видим следующее сообщение, повторяющееся некоторое количество раз.

12:51:52,683 | INFO  | msConsumer[okay] | BusinessLogic                    | 205 - transaction-client - 0.0.1 | exception
12:51:52,684 | INFO  | msConsumer[okay] | BusinessLogic                    | 205 - transaction-client - 0.0.1 | Rollback exception
12:51:52,685 | ERROR | msConsumer[okay] | DefaultErrorHandler              | 110 - org.apache.camel.camel-core - 2.12.3 | Failed delivery for (MessageId: ID:NovA-63543-1396509662689-1:3:1:1:2 on ExchangeId: ID-NovA-63522-1396509447471-9-1). Exhausted after delivery attempt: 1 caught: java.lang.IllegalArgumentException: This is forced

Message History
---------------------------------------------------------------------------------------------------------------------------------------
RouteId              ProcessorId          Processor                                                                        Elapsed (ms)
[route15           ] [route15           ] [jmstx://queue:okay                                                            ] [         3]
[route15           ] [transacted15      ] [transacted[]                                                                  ] [         3]
[route15           ] [to21              ] [jmstx:queue:final                                                             ] [         1]
[route15           ] [bean8             ] [bean[ref:businessLogic method: business]                                      ] [         2]

Exchange
---------------------------------------------------------------------------------------------------------------------------------------
Exchange[
 Id                  ID-NovA-63522-1396509447471-9-1
 ExchangePattern     InOnly
 Headers             {breadcrumbId=ID:NovA-63543-1396509662689-1:3:1:1:2, CamelRedelivered=false, CamelRedeliveryCounter=0, JMSCorrelationID=, JMSDeliveryMode=1, JMSDestination=queue://okay, JMSExpiration=0, JMSMessageID=ID:NovA-63543-1396509662689-1:3:1:1:2, JMSPriority=0, JMSRedelivered=false, JMSReplyTo=null, JMSTimestamp=1396515112680, JMSType=, JMSXGroupID=null, JMSXUserID=null}
 BodyType            String
 Body                exception
]

Сообщение действительно не попало в очередь final. Но это ли мы действительно хотели? В результате сообщение не находится ни в одной из очередей. Т.е. сообщение пропало! Ищите в логах. Это просто не допустимо. Итого корректное поведение, которого я хочу добиться следующее.
При возникновении исключения сообщение должно откатываться во всех системах, и для того чтобы оно не потерялось перекладываться в специальную очередь для ошибок. Иными словами, нам необходимо реализовать специльный паттерн для таких случаев "Dead letter chanel". Судя по документации есть несколько путей чтобы реализовать подобное.
Например в некоторых местах документации предлагают такой вариант:

// Java
import org.apache.camel.spring.SpringRouteBuilder;

public class MyRouteBuilder extends SpringRouteBuilder {
    ...
    public void configure() {
        onException(IllegalArgumentException.class).
            maximumRedeliveries(1)
            .handled(true)
            .to("file:target/messages?fileName=deadLetters.xml&fileExist=Append")
            .markRollbackOnly();  // NB: Must come *after* the dead letter endpoint.

        from("file:src/data?noop=true")
            .transacted()
            .beanRef("accountService","credit")
            .beanRef("accountService","debit")
            .beanRef("accountService","dumpTable")
            .to("file:target/messages");
    }
}

Попробуем адаптировать этот пример ближе к нам. Таким образом это будет выглядеть следующим образом:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:camel="http://camel.apache.org/schema/spring"
 xmlns:osgi="http://www.springframework.org/schema/osgi"
 xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

 <bean id="businessLogic" class="com.bssys.transaction_client.BusinessLogic" />

 <camelContext xmlns="http://camel.apache.org/schema/spring">

  <route>
   <!-- 1: from the jms queue -->
   <from uri="jmstx:queue:okay" />
   <onException>
    <exception>java.lang.IllegalArgumentException</exception>
    <redeliveryPolicy maximumRedeliveries="1"/>
    <handled>
     <constant>true</constant>
    </handled>
    <log message="${exception.message}" />
    <to uri="file:target/messages?fileName=deadLetters.xml&amp;fileExist=Append" />
    <rollback markRollbackOnly="true" />
   </onException>
   <transacted/>
   <to uri="jmstx:queue:final" />
   <bean ref="businessLogic" method="business" />
  </route>
  
 </camelContext>

</beans>

Да, действительно это похоже на то что нам надо, но только с одной оговоркой. в качестве dead letter chanel, сообщение скидывается в текстовый файл. Но нам то надо перекладывать его в отдельную очередь. Адаптируем пример, где в качестве dead letter queue, будет использоваться отдельная очередь.

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:camel="http://camel.apache.org/schema/spring"
 xmlns:osgi="http://www.springframework.org/schema/osgi"
 xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

 <bean id="businessLogic" class="com.bssys.transaction_client.BusinessLogic" />

 <camelContext xmlns="http://camel.apache.org/schema/spring">

  <route>
   <!-- 1: from the jms queue -->
   <from uri="jmstx:queue:okay" />
   <onException>
    <exception>java.lang.IllegalArgumentException</exception>
    <camel:redeliveryPolicy maximumRedeliveries="1"/>
    <handled>
     <constant>true</constant>
    </handled>
    <log message="${exception.message}" />
    <to uri="jmstx:queue:bad" />
    <rollback markRollbackOnly="true" />
   </onException>
   <!-- <transacted ref="PROPAGATION_REQUIRED" /> -->
   <transacted/>
   <to uri="jmstx:queue:final" />
   <bean ref="businessLogic" method="business" />
  </route>
 </camelContext>

</beans>

Запустив мы увидим в логах следующее

13:46:09,301 | INFO  | msConsumer[okay] | BusinessLogic                    | 207 - transaction-client - 0.0.1 | exception
13:46:09,302 | INFO  | msConsumer[okay] | BusinessLogic                    | 207 - transaction-client - 0.0.1 | Rollback exception
13:46:09,302 | INFO  | msConsumer[okay] | BusinessLogic                    | 207 - transaction-client - 0.0.1 | exception
13:46:09,303 | INFO  | msConsumer[okay] | BusinessLogic                    | 207 - transaction-client - 0.0.1 | Rollback exception
13:46:09,303 | INFO  | msConsumer[okay] | route5                           | 110 - org.apache.camel.camel-core - 2.12.3 | This is forced
13:46:09,305 | WARN  | msConsumer[okay] | DefaultErrorHandler              | 110 - org.apache.camel.camel-core - 2.12.3 | Rollback (MessageId: ID:NovA-64390-1396518019146-1:1:1:1:3 on ExchangeId: ID-NovA-64373-1396518004181-2-1) due: This is forced

Message History
---------------------------------------------------------------------------------------------------------------------------------------
RouteId              ProcessorId          Processor                                                                        Elapsed (ms)
[route5            ] [route5            ] [jmstx://queue:okay                                                            ] [         5]
[route5            ] [transacted5       ] [transacted[]                                                                  ] [         5]
[route5            ] [to8               ] [jmstx:queue:final                                                             ] [         1]
[route5            ] [bean3             ] [bean[ref:businessLogic method: business]                                      ] [         4]
[route5            ] [log3              ] [log                                                                           ] [         0]
[route5            ] [to7               ] [jmstx:queue:bad                                                               ] [         1]
[route5            ] [rollback3         ] [rollback                                                                      ] [         0]

Exchange
---------------------------------------------------------------------------------------------------------------------------------------
Exchange[
 Id                  ID-NovA-64373-1396518004181-2-1
 ExchangePattern     InOnly
 Headers             {breadcrumbId=ID:NovA-64390-1396518019146-1:1:1:1:3, JMSCorrelationID=, JMSDeliveryMode=1, JMSDestination=queue://okay, JMSExpiration=0, JMSMessageID=ID:NovA-64390-1396518019146-1:1:1:1:3, JMSPriority=0, JMSRedelivered=false, JMSReplyTo=null, JMSTimestamp=1396518369297, JMSType=, JMSXGroupID=null, JMSXUserID=null}
 BodyType            String
 Body                exception
]

13:46:09,305 | WARN  | msConsumer[okay] | TransactionErrorHandler          | 110 - org.apache.camel.camel-core - 2.12.3 | Transaction rollback (0x2b0ef227) redelivered(false) for (MessageId: ID:NovA-64390-1396518019146-1:1:1:1:3 on ExchangeId: ID-NovA-64373-1396518004181-2-1) due exchange was marked for rollbackOnly
13:46:09,305 | WARN  | msConsumer[okay] | EndpointMessageListener          | 110 - org.apache.camel.camel-core - 2.12.3 | Execution of JMS message listener failed. Caused by: [org.apache.camel.RuntimeCamelException - org.apache.camel.RollbackExchangeException: Intended rollback. Exchange[JmsMessage[JmsMessageID: ID:NovA-64390-1396518019146-1:1:1:1:3]]]

По факту сообщение попало в очередь jmstx:queue:bad, но было отменено по rollback! Это совсем никуда не годится! Для того чтобы сообщение не откатывалось из очереди ошибок, отправку необходимо выполнять в отдельной транзакции. При этом надо настроить транзакции так, чтобы не было двойной переотправки. Дело в том что помимо повторений на стороне Camel, переотправкой сообщений будет также заниматься transactionManager.

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:osgi="http://www.springframework.org/schema/osgi"
 xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd">

 <osgi:reference id="platformTransactionManager"
  interface="org.springframework.transaction.PlatformTransactionManager" />

  <bean id="activemqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
  <property name="brokerURL" value="tcp://localhost:61616" />
  <property name="userName" value="karaf" />
  <property name="password" value="karaf" />
  <property name="redeliveryPolicy">
   <bean class="org.apache.activemq.RedeliveryPolicy">
    <property name="maximumRedeliveries" value="0" />
   </bean>
  </property>
 </bean>
 
 <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
        <property name="maxConnections" value="8"/>
        <property name="connectionFactory" ref="activemqConnectionFactory"/>
    </bean>

 <bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
  <property name="connectionFactory" ref="pooledConnectionFactory" />
  <property name="transactionManager" ref="platformTransactionManager" />
  <property name="transacted" value="true" />
 </bean>

 <bean id="jmstx" class="org.apache.camel.component.jms.JmsComponent">
  <constructor-arg ref="jmsConfig" />
 </bean>

 <bean id="PROPAGATION_REQUIRED" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
  <property name="transactionManager" ref="platformTransactionManager" />
  <property name="propagationBehaviorName" value="PROPAGATION_REQUIRED" />
 </bean>
 <bean id="PROPAGATION_REQUIRES_NEW" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
  <property name="transactionManager" ref="platformTransactionManager" />
  <property name="propagationBehaviorName" value="PROPAGATION_REQUIRES_NEW" />
 </bean>

</beans>

Итак вот итоговый результат!

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:camel="http://camel.apache.org/schema/spring"
 xmlns:osgi="http://www.springframework.org/schema/osgi"
 xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

 <bean id="businessLogic" class="com.bssys.transaction_client.BusinessLogic" />

 <camelContext xmlns="http://camel.apache.org/schema/spring">

  <route>
   <!-- 1: from the jms queue -->
   <from uri="jmstx:queue:okay" />
   <onException>
    <exception>java.lang.IllegalArgumentException</exception>
    <camel:redeliveryPolicy maximumRedeliveries="3" redeliveryDelay="1000"/>
    <handled>
     <constant>true</constant>
    </handled>
    <log message="${exception.message}" />
    <to uri="direct:dead" />
    <rollback markRollbackOnly="true" />
   </onException>
   <transacted/>
   <to uri="jmstx:queue:final" />
   <bean ref="businessLogic" method="business" />
  </route>

  <route>
   <camel:from uri="direct:dead" />
   <transacted ref="PROPAGATION_REQUIRES_NEW" />
   <to uri="jmstx:queue:bad" />
  </route>
 </camelContext>

</beans>

В примере делается 3 попытки с интервалом 1сек, и в конечном итоге сообщение попадает в очередь ошибок.

14:07:36,765 | INFO  | msConsumer[okay] | BusinessLogic                    | 207 - transaction-client - 0.0.1 | exception
14:07:36,766 | INFO  | msConsumer[okay] | BusinessLogic                    | 207 - transaction-client - 0.0.1 | Rollback exception
14:07:37,766 | INFO  | msConsumer[okay] | BusinessLogic                    | 207 - transaction-client - 0.0.1 | exception
14:07:37,766 | INFO  | msConsumer[okay] | BusinessLogic                    | 207 - transaction-client - 0.0.1 | Rollback exception
14:07:38,767 | INFO  | msConsumer[okay] | BusinessLogic                    | 207 - transaction-client - 0.0.1 | exception
14:07:38,767 | INFO  | msConsumer[okay] | BusinessLogic                    | 207 - transaction-client - 0.0.1 | Rollback exception
14:07:39,767 | INFO  | msConsumer[okay] | BusinessLogic                    | 207 - transaction-client - 0.0.1 | exception
14:07:39,767 | INFO  | msConsumer[okay] | BusinessLogic                    | 207 - transaction-client - 0.0.1 | Rollback exception
14:07:39,768 | INFO  | msConsumer[okay] | route9                           | 110 - org.apache.camel.camel-core - 2.12.3 | This is forced
14:07:39,826 | WARN  | msConsumer[okay] | DefaultErrorHandler              | 110 - org.apache.camel.camel-core - 2.12.3 | Rollback (MessageId: ID:NovA-64390-1396518019146-1:2:1:1:2 on ExchangeId: ID-NovA-64373-1396518004181-4-1) due: This is forced

Message History
---------------------------------------------------------------------------------------------------------------------------------------
RouteId              ProcessorId          Processor                                                                        Elapsed (ms)
[route9            ] [route9            ] [jmstx://queue:okay                                                            ] [      3062]
[route9            ] [transacted9       ] [transacted[]                                                                  ] [      3062]
[route9            ] [to14              ] [jmstx:queue:final                                                             ] [         1]
[route9            ] [bean5             ] [bean[ref:businessLogic method: business]                                      ] [      3061]
[route9            ] [log5              ] [log                                                                           ] [         0]
[route9            ] [to13              ] [direct:dead                                                                   ] [        57]
[route10           ] [transacted10      ] [transacted[ref:PROPAGATION_REQUIRES_NEW]                                      ] [        57]
[route10           ] [to15              ] [jmstx:queue:bad                                                               ] [         3]
[route9            ] [rollback5         ] [rollback                                                                      ] [         0]

Exchange
---------------------------------------------------------------------------------------------------------------------------------------
Exchange[
 Id                  ID-NovA-64373-1396518004181-4-1
 ExchangePattern     InOnly
 Headers             {breadcrumbId=ID:NovA-64390-1396518019146-1:2:1:1:2, JMSCorrelationID=, JMSDeliveryMode=1, JMSDestination=queue://okay, JMSExpiration=0, JMSMessageID=ID:NovA-64390-1396518019146-1:2:1:1:2, JMSPriority=0, JMSRedelivered=false, JMSReplyTo=null, JMSTimestamp=1396519656761, JMSType=, JMSXGroupID=null, JMSXUserID=null}
 BodyType            String
 Body                exception
]

14:07:39,826 | WARN  | msConsumer[okay] | TransactionErrorHandler          | 110 - org.apache.camel.camel-core - 2.12.3 | Transaction rollback (0x2b75718c) redelivered(false) for (MessageId: ID:NovA-64390-1396518019146-1:2:1:1:2 on ExchangeId: ID-NovA-64373-1396518004181-4-1) due exchange was marked for rollbackOnly
14:07:39,827 | WARN  | msConsumer[okay] | EndpointMessageListener          | 110 - org.apache.camel.camel-core - 2.12.3 | Execution of JMS message listener failed. Caused by: [org.apache.camel.RuntimeCamelException - org.apache.camel.RollbackExchangeException: Intended rollback. Exchange[JmsMessage[JmsMessageID: ID:NovA-64390-1396518019146-1:2:1:1:2]]]

Весь код тут: git clone https://Hibernate2009@bitbucket.org/Hibernate2009/transaction-jms-client.git

RESTFul with Apache Camel

2014-03-17T22:48:00.002+10:00

После длительного перерыва, рассмотрим как реализовать с помощью Apache Camel простейший REST-сервис. Реализовать подобное с помощью Apache Camel можно навскидку 3-мя способами. Мне как раз был необходим сервис по хранению и управлению настройками в формате key=value. REST необходим чтобы в дальнейшем сделать на AngularJS страницу для управления настройками.
Итак Apache Camel предоставляет следующие возможности для реализации REST сервиса

CXF Bean Component и CXFRS Component очень похожи друг на друга, и строятся на основе отдельного java-bean помеченного специальными аннотациями, и отличаются лишь обвязкой и инициализацией.Лично меня привлек CXF Bean Component который поддерживает из коробки одновременно два стандарта JAXRS, JAXWS. Что позволяет управлять выводом XML-JSON, плюс есть возможность использовать разработанный код для реализации SOAP WebService.
Сам код на Apache Camel очень лаконичен:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:camel="http://camel.apache.org/schema/spring"
 xmlns:osgi="http://www.springframework.org/schema/osgi" xmlns:context="http://www.springframework.org/schema/context"
 xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
       http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">

 <context:component-scan base-package="com.bssys.*" />
 
 <bean id="itemRestService" class="com.bssys.rest_service.ItemRestService" />
 
 <camelContext xmlns="http://camel.apache.org/schema/spring">
  <route>
   <from uri="jetty:http://localhost:8787?matchOnUriPrefix=true" />
   <to uri="cxfbean:itemRestService" />
  </route>
 </camelContext>
</beans>

Детали скрываются в реализации ItemRestService. Ниже я опишу полную последовательность разработки
Сам сервис я попытался сделать захватывающим разные аспекты связывания данных.
Общий путь к REST-сервису:
http://localhost:8787/example
GET http://localhost:8787/example/getAllItem
GET http://localhost:8787/example/getItem/{key}
PUT http://localhost:8787/example/updateItem
PUT http://localhost:8787/example/addItem
DELETE http://localhost:8787/example/removeItem/{key}

getAllItem- GET запрос без параметров
getItem/{key}- GET запрос с запросе указывающий имя получаемого свойства
updateItem - PUT запрос обновления свойства. В теле запроса передается JSON
addItem - PUT запрос добавления свойства. В теле запроса передается JSON
removeItem/{key} - DELETE запрос на удаление свойства.

Изначально, для прозрачной работы с данными нам понадобится модель данных. У меня она предоставлена в виде XSD-схемы.

<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns:xsd="http://www.w3.org/2001/XMLSchema"  targetNamespace="http://bssys.com/item"
 xmlns:tns="http://bssys.com/item" elementFormDefault="qualified">
 
 <xsd:element name="itemRequest" type="tns:itemRequest"/>
 <xsd:element name="itemResponse" type="tns:itemResponse"/>
 
 <xsd:complexType name="item">
  <xsd:sequence>
   <xsd:element name="key" type="xsd:string"/>
   <xsd:element name="value" type="xsd:string"/>
  </xsd:sequence>
 </xsd:complexType>
 
 <xsd:complexType name="itemRequest">
  <xsd:sequence>
   <xsd:element name="item" type="tns:item"/>
  </xsd:sequence>
 </xsd:complexType>
 
 <xsd:complexType name="itemResponse">
  <xsd:sequence>
   <xsd:element name="item" minOccurs="0" maxOccurs="unbounded" type="tns:item"/>
   <xsd:element name="errorMessage" type="xsd:string"/>
  </xsd:sequence>
 </xsd:complexType>
</xsd:schema>

На основе который мы сгенерируем JAXB классы. Именно знакомый нам JAXB отвечает за "marshaling" из разных форматов. На самом деле, без этого этапа можно запросто обойтись. XSD-схема далее применяться не будет. Конечно можно разработать нужные классы вручную, но в таком подходе сервис получается стройнее, и появляется возможность в дальнейшем переработать его в SOAP Webservice.

После генерации имеем следующие классы:

Item
ItemRequest
ItemResponse
ObjectFactory

Классы уже анотированы атонациями JAXB. Для возможности использования в составе REST сервиса, необходимо только добавить анотацию @XmlRootElement в классы ItemRequest ItemResponse.
На их основе разработаем интерфейс нашего сервиса:

package com.bssys.service;

import org.springframework.stereotype.Service;
import com.bssys.jaxb.model.ItemRequest;
import com.bssys.jaxb.model.ItemResponse;

@Service
public interface ItemService {
 public ItemResponse addItem(ItemRequest request);
 public ItemResponse updateItem(ItemRequest request);
 public ItemResponse getItem(String key);
 public ItemResponse getAllItem();
 public ItemResponse removeItem(String key);
 public ItemRequest stubRequest();
}

Слой DAO:

package com.bssys.dao;

import java.util.List;

import org.springframework.stereotype.Service;

import com.bssys.jaxb.model.Item;

@Service
public interface ItemManagerDao {
 public Item getItem(String name);
 public void deleteItem(String name);
 public List getAllItems();
 public void insertItem(Item item);
 public void updateItem(Item item);
}

Далее простейшая реализация "in memory":

package com.bssys.dao;

import java.util.ArrayList;
import java.util.List;

import org.springframework.stereotype.Component;

import com.bssys.jaxb.model.Item;


@Component
public class ItemManagerMemoryDao implements ItemManagerDao {
 
 private List items = new ArrayList();

 @Override
 public Item getItem(String name) {
  // TODO Auto-generated method stub
  for (Item item: items){
   String key = item.getKey();
   if (key.equals(name)){
    return item;
   }
  }
  throw new RuntimeException("Item Not Found: " + name);
 }

 @Override
 public List getAllItems() {
  // TODO Auto-generated method stub
  return items;
 }

 @Override
 public void insertItem(Item item) {
  // TODO Auto-generated method stub
  items.add(item);
 }

 @Override
 public void updateItem(Item item) {
  // TODO Auto-generated method stub
  Item storeItem = getItem(item.getKey());
  storeItem.setValue(item.getValue());
 }

 @Override
 public void deleteItem(String name) {
  // TODO Auto-generated method stub
  Item storeItem = getItem(name);
  items.remove(storeItem);
 }
}

И сам REST-сервис:

package com.bssys.rest_service;

import java.util.List;

import javax.ws.rs.Consumes;
import javax.ws.rs.DELETE;
import javax.ws.rs.GET;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;

import org.springframework.beans.factory.annotation.Autowired;

import com.bssys.dao.ItemManagerMemoryDao;
import com.bssys.jaxb.model.Item;
import com.bssys.jaxb.model.ItemRequest;
import com.bssys.jaxb.model.ItemResponse;
import com.bssys.service.ItemService;

@Path("/example")
public class ItemRestService implements ItemService{
 @Autowired
 private ItemManagerMemoryDao dao;

 @PUT
 @Path("addItem")
 @Consumes("application/json")
 @Produces("application/json")
 public ItemResponse addItem(ItemRequest request) {
  // TODO Auto-generated method stub
  ItemResponse userResponse = new ItemResponse();
  try{
   dao.insertItem(request.getItem());
   userResponse.setErrorMessage("Ok");
  }catch(Exception e){
   userResponse.setErrorMessage("Error: "+e.getMessage());
  }
  return userResponse;
 }

 @PUT
 @Path("updateItem")
 @Consumes("application/json")
 @Produces("application/json")
 public ItemResponse updateItem(ItemRequest request) {
  // TODO Auto-generated method stub
  ItemResponse userResponse = new ItemResponse();
  try{
   dao.updateItem(request.getItem());
   userResponse.setErrorMessage("Ok");
  }catch(Exception e){
   userResponse.setErrorMessage("Error: "+e.getMessage());
  }
  return userResponse;
 }

 @GET
 @Path("getItem/{key}")
 @Produces("application/json")
 public ItemResponse getItem(@PathParam("key") String key) {
  // TODO Auto-generated method stub
  ItemResponse userResponse = new ItemResponse();
  try{
   Item item = dao.getItem(key);
   userResponse.getItem().add(item);
   userResponse.setErrorMessage("Ok");
  }catch(Exception e){
   userResponse.setErrorMessage("Error: "+e.getMessage());
  }
  return userResponse;
 }

 @GET
 @Path("getAllItem")
 @Produces("application/json")
 public ItemResponse getAllItem() {
  // TODO Auto-generated method stub
  ItemResponse userResponse = new ItemResponse();
  try{
   List allItems = dao.getAllItems();
   List item = userResponse.getItem();
   item.addAll(allItems);
   userResponse.setErrorMessage("Ok");
  }catch(Exception e){
   userResponse.setErrorMessage("Error: "+e.getMessage());
  }
  return userResponse;
 }

 @DELETE
 @Path("removeItem/{key}")
 @Consumes("application/json")
 @Produces("application/json")
 public ItemResponse removeItem(@PathParam("key") String key) {
  // TODO Auto-generated method stub
  ItemResponse userResponse = new ItemResponse();
  try{
   dao.deleteItem(key);
   userResponse.setErrorMessage("Ok");
  }catch(Exception e){
   userResponse.setErrorMessage("Error: "+e.getMessage());
  }
  return userResponse;
 }

 @Override
 @GET
 @Path("stubRequest")
 @Produces("application/json")
 public ItemRequest stubRequest() {
  // TODO Auto-generated method stub
  ItemRequest itemRequest = new ItemRequest();
  Item item = new Item();
  item.setKey("key");
  item.setValue("value");
  itemRequest.setItem(item);
  return itemRequest;
 }
}

Все это собирается как OSGI-модуль готовый к развертыванию в ServiceMi. Исходники на BitBucket https://Hibernate2009@bitbucket.org/Hibernate2009/restful-with-apache-camel.git

Apache Camel on ServiceMix XA Transaction with database and JMS

2013-01-17T22:09:00.000+10:00

Перейдем к одной из самых серьезных задач, без которой не может обойтись ни одна серьезная интеграционная платформа, к XA транзакциям. Для примера возьмем ситуацию близкую к реальности описанную как маршрут Apache Camel.

<camelContext xmlns="http://camel.apache.org/schema/spring">
 <!-- Transactional route -->
 <route>
  <from uri="jmstx:queue:giro" />
  <to uri="jmstx:queue:statusLog" />
  <bean ref="accountService" method="credit" />
  <bean ref="accountService" method="debit" />
 </route>
 <!-- Feeder route -->
 <route>
  <from uri="jmstx:queue:statusLog" />
  <to uri="file:Success" />
 </route>
</camelContext>

В этом маршруте описывается следующее:

Принимает сообщение из JMS-очереди giro
Перенаправляет сообщение в JMS-очередь statusLog, в которой должны быть только успешные сообщения
Меняем состояние базы данных

Второй маршрут описывает следующее:

Принимает сообщение из JMS-очереди statusLog
Сохраняет сообщение в папке Success

В ходе работы в итоговой папке Success должны находиться только успешные сообщения. При возникновении исключения, состояние в базе не меняется, и файла с транзакцией не оказывается в папке успешных сообщений. В этом примере участвуют 3 транзакционные системы JMS, database и filesystem. Как это заставить работать на ServiceMix и Apache Camel?...

Для начала необходимо ознакомиться с основной документацией по транзакциям на ServiceMix и его коммерческой версииhttp://fusesource.com/docs/esbent/7.1/camel_tx/front.html. Ниже изложенный материал полностью основан на стандартной документации.
Итак для начала как обычно создаем проект в Eclipse на основе архетипа camel-archetype-spring-dm. Мы не будем изобретать логику и воспользуемся примером из документации. Создаем класс AccountService, который будет работать с базой данных. Вот прямая ссылка на исходник AccountService. В классе разработаны методы debit и credit которые меняют состояние базы.

Для начала определим DataSource. В предыдущей статье мы как раз опубликовали XADataSource к базе данных Oracle. В конфиге получаем доступ к нашему DataSource

<!-- Reference JDBC DataSource -->
<osgi:reference id="oracleXADataSource" interface="javax.sql.DataSource"
 filter="(oracle.xa.datasource.name=oracleXA)" />

В качестве фильтра используется выражение, которые мы определили при его создании. Теперь можно определить наш бин и передать ему полученный DataSource

<!-- bean for account business logic -->
<bean id="accountService" class="com.bssys.ts2.AccountService">
 <property name="dataSource" ref="oracleXADataSource" />
</bean>

Так настраивается соединение с базой.
Настройки JMS ресурсов чуть более сложны. Первым делом необходимо получить доступ к TransactionManager предоставляемых ServiceMix

<!-- OSGi TM Service -->
<!-- access through Spring's PlatformTransactionManager -->
<osgi:reference id="osgiPlatformTransactionManager"
 interface="org.springframework.transaction.PlatformTransactionManager" />
<!-- access through PlatformTransactionManager -->
<osgi:reference id="osgiJtaTransactionManager"
 interface="javax.transaction.TransactionManager" />

Имея их настраиваем ConnectionFactory для JMS

<!-- JMS TX endpoint configuration -->
<bean id="jmstx" class="org.apache.activemq.camel.component.ActiveMQComponent">
 <property name="configuration" ref="jmsTxConfig" />
</bean>

<bean id="jmsTxConfig" class="org.apache.camel.component.jms.JmsConfiguration">
 <property name="connectionFactory" ref="jmsXaPoolConnectionFactory" />
 <property name="transactionManager" ref="osgiPlatformTransactionManager" />
 <property name="transacted" value="false" />
 <property name="cacheLevelName" value="CACHE_CONNECTION" />
</bean>

<!-- connection factory wrapper to support auto-enlisting of XA resource -->
<bean id="jmsXaPoolConnectionFactory" class="org.apache.activemq.pool.XaPooledConnectionFactory">
 <property name="maxConnections" value="1" />
 <property name="connectionFactory" ref="jmsXaConnectionFactory" />
 <property name="transactionManager" ref="osgiJtaTransactionManager" />
</bean>

<bean id="jmsXaConnectionFactory" class="org.apache.activemq.ActiveMQXAConnectionFactory">
 <property name="brokerURL" value="vm:local" />
 <property name="redeliveryPolicy">
  <bean class="org.apache.activemq.RedeliveryPolicy">
   <property name="maximumRedeliveries" value="0" />
  </bean>
 </property>
</bean>

Осталось сформировать базу по примеру из документации. Собрать и закинуть наш модуль в папку deploy.

ij> CREATE TABLE accounts (name VARCHAR(50), amount INT);

ij> INSERT INTO accounts (name,amount) VALUES ('Major Clanger',2000);

ij> INSERT INTO accounts (name,amount) VALUES ('Tiny Clanger',100);

Теперь кидая сообщения вида мы будем видеть их в папке Success, и смотреть как поменялись данные в таблице Account

<transaction>
  <transfer>
    <sender>Major Clanger</sender>
    <receiver>Tiny Clanger</receiver>
    <amount>90</amount>
  </transfer>
</transaction>

Для того чтобы спровоцировать откат транзакции достаточно кинуть сообщение с полем amount превышающим 100.

<transaction>
  <transfer>
    <sender>Major Clanger</sender>
    <receiver>Tiny Clanger</receiver>
    <amount>150</amount>
  </transfer>
</transaction>

В этом случае метод debit выкинет исключение которое вызовет rollbak для транзакции.

public void debit(
            @XPath("/transaction/transfer/sender/text()") String name,
            @XPath("/transaction/transfer/amount/text()") String amount
            )
    {
        log.info("debit() called with args name = " + name + " and amount = " + amount);
        int iamount = Integer.parseInt(amount);
        if (iamount > 100) {
            throw new IllegalArgumentException("Debit limit is 100");
        }
        int origAmount = jdbc.queryForInt(
                "select amount from accounts where name = ?",
                new Object[]{name}
        );
        int newAmount = origAmount - Integer.parseInt(amount);
        if (newAmount < 0) {
            throw new IllegalArgumentException("Not enough in account");
        }
        
        jdbc.update(
                "update accounts set amount = ? where name = ?",
                new Object[] {newAmount, name}
        );
    }

JDBC XA DataSource as OSGI Service

2013-01-16T20:29:00.001+10:00

В данной заметке я расскажу как использовать базу данных в связке Apache Camel и Apache ServiceMix. Да, конечно можно описать все необходимы и бины и источники данных в конфиге Camel, но тут мы заглянем дальше. Описанные ресурсы будут участвовать в XA-транзации, и все это добро будет крутиться на ServiceMix. Это первая часть большого примера.

Итак, берем последнюю версию ServiceMix, на момент написания статьи это версия 4.4.2. В этой версии ServiceMix использует Apache Camel 2.8.5, версия старая, но для нас абсолютно не критично.

Для того чтобы устанавливать проложения под ServiceMix, необходим правильный способ сборки. Создаем в Eclipse проект Maven на основе архетипа camel-archetype-spring-dm, в результате будет создан проект, который собирается в osgi bundle и после сборки полностью готовый к установке на ServiceMix. Тут важно при создании проекта установить версию Apаche Camel 2.8.5, которая соответствует версии в ServiceMix, иначе у нас возникнут проблемы при запуске нашего модуля. Отличие от архетипа, которым мы пользовались до этого сосредоточены в pom.xml:

Установлен тип сборки bundle

<packaging>bundle</packaging>

Добавлен новый плагин

<plugin>
 <groupId>org.apache.felix</groupId>
 <artifactId>maven-bundle-plugin</artifactId>
 <version>2.3.4</version>
 <extensions>true</extensions>
 <configuration>
  <instructions>
   <Bundle-SymbolicName>datasources</Bundle-SymbolicName>
   <Private-Package>com.bssys.datasources.*</Private-Package>
   <Import-Package>*,org.apache.camel.osgi</Import-Package>
  </instructions>
 </configuration>
</plugin>

После команды mvn build в папке target будет сгенерирован модуль готовый к развертыванию.
На самом деле в этом моделе Camel нам и не нужен. В файле camel-context.xml удаляем дефолтный маршрут, и вместо него описываем нужный нам DataSource. У себя в проекте в качестве базы данных используется Oracle, поэтому для примера возьмем именно его.

<?xml version="1.0" encoding="UTF-8"?>
<!-- Licensed to the Apache Software Foundation (ASF) under one or more contributor 
 license agreements. See the NOTICE file distributed with this work for additional 
 information regarding copyright ownership. The ASF licenses this file to 
 You under the Apache License, Version 2.0 (the "License"); you may not use 
 this file except in compliance with the License. You may obtain a copy of 
 the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required 
 by applicable law or agreed to in writing, software distributed under the 
 License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS 
 OF ANY KIND, either express or implied. See the License for the specific 
 language governing permissions and limitations under the License. -->

<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:camel="http://camel.apache.org/schema/spring"
 xmlns:osgi="http://www.springframework.org/schema/osgi"
 xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

 <bean id="oracleXADataSource" class="oracle.jdbc.xa.client.OracleXADataSource">
  <property name="URL" value="jdbc:oracle:thin:@127.0.0.1:1521:orcl" />
  <property name="user" value="user" />
  <property name="password" value="password" />
 </bean>

 <osgi:service id="xaDataSource" ref="oracleXADataSource"
  interface="javax.sql.XADataSource">
  <osgi:service-properties>
   <entry key="oracle.xa.datasource.name" value="oracleXA" />
  </osgi:service-properties>
 </osgi:service>

 <bean id="helloBean" class="com.bssys.datasources.HelloBean">
  <property name="say" value="Hi from Camel" />
 </bean>
</beans>

Самая интересная часть это описание OSGI-сервиса, который ссылается на описанный OracleXADataSource и выставляет свойство oracle.xa.datasource.name. Свойства могут быть любыми, и нужны там нам чтобы в дальнейшем выбирать нужный OSGI-сервис. Как это сделать будет описано далее.
Выставленный как OSGI-сервис XA-DataSource в ServiceMIx автоматически оборачивает в обертку Apache Aries Auto-Enlisting XA Wrapper. Более подробно можно почитать в документации http://fusesource.com/docs/esbent/7.0/camel_tx/XaJdbc-AutoEnlist.html. После этого выполняем команду:

mvn build

Забираем собранный модуль из папки проекта target, и закидываем его в папку deploy в ServiceMix.
Если все прошло успешно, то в ServiceMix должен быть наш установленный модуль.
Можно проверить набрав в консоли:

osgi:list

[ 265] [Active     ] [            ] [Started] [   60] DataSource bundle (0.0.1.SNAPSHOT)

Видим наш установленный модуль с id=265
Проверим что наш OSGI-сервис успешно обернут:

osgi:ls 265

Среди прочего можно увидеть:

aries.xa.aware = true
Bundle-SymbolicName = datasources
Bundle-Version = 0.0.1.SNAPSHOT
objectClass = javax.sql.DataSource
oracle.xa.datasource.name = oracleXA
org.springframework.osgi.bean.name = oracleXADataSource
service.id = 261

Теперь можно получать наш DataSource по OSGI-ссылке. Как это сделать будет рассказано в моей следующей заметке.

Parse HTML-page and follow the link with Apache Camel

2012-08-30T00:00:00.000+10:00

Как распарсить HTML-страницу и пройти по ссылкам с помощью Ahace Camel? Если кратко-легко! Выполним эту задачу не написав ни одной строчки кода.
На этот раз воспользуемся Spring-DSL, так как в данном случае он более лаконичен.

Для начала необходимо подключить необходимые зависимости

<dependency>
 <groupId>org.apache.camel</groupId>
 <artifactId>camel-http</artifactId>
 <version>2.9.1</version>
</dependency>
<dependency>
 <groupId>org.apache.camel</groupId>
 <artifactId>camel-tagsoup</artifactId>
 <version>2.9.1</version>
</dependency>

camel-tagsoup необходим для того чтобы распарсить HTML-страницу. Ниже сам маршрут на Spring-DSL выполняющий данную задачу.

<camel:route>
 <camel:from uri="direct:bot" /><!-- В качестве сообщения будет поступать url -->
 <camel:setHeader headerName="original"><!--Сохраним url для дальнейшего использования -->
  <camel:simple>${body}</camel:simple>
 </camel:setHeader>
 <camel:recipientList><!-- Динамически вызовем url -->
  <camel:simple>${body}/b/opp</camel:simple>
 </camel:recipientList>
 <camel:unmarshal>
  <camel:tidyMarkup /><!-- Привести HTML к удобочитаемому виду -->
 </camel:unmarshal>
 <camel:setBody>
  <camel:xpath>//a</camel:xpath><!-- С помощью XPath отбираем только ссылки -->
 </camel:setBody>
 <camel:split>
  <camel:tokenize token="a" xml="true" /><!-- С помощью сплиттера разделяем полученые ссылки -->
  <camel:setHeader headerName="href">
   <camel:xpath>//@href</camel:xpath><!-- Получаем url-ссылки -->
  </camel:setHeader>
  <camel:log message="${in.header.original}${in.header.href}"></camel:log>
  <camel:recipientList>
   <camel:simple>${in.header.original}${in.header.href}</camel:simple><!--Формируем итоговый url и проходим по ссылке -->
  </camel:recipientList>
 </camel:split>
</camel:route>

Само собой ссылки в HTML относительные. Удачи!

Calling Web Services with Apache Camel

2012-08-08T23:20:00.000+10:00

Казалось бы, такая простая задача для интеграционной системы как вызов внешнего webservice, для Apache Camel вызывает затруднения. Прольем немного света. Вызов внешнего webservice на Apache Camel реально занимает одну строчку! Возьмем доступный всем открытый сервис погоды:
http://www.webservicex.com/globalweather.asmx?WSDL
Вот как вызвать этот сервис на Camel:

from("file:c:/data/inbox3").to("cxf:bean:calculatorWSEndpoint?defaultOperationName=GetWeather").to("file:c:/data/outbox");

Подробности под катом!

Подготовим spring-config.xml, в нем нам необходимо зарегистрировать cxf-endpoint для вызываемого webservice

<?xml version="1.0" encoding="UTF-8"?>
<!-- Licensed to the Apache Software Foundation (ASF) under one or more contributor 
 license agreements. See the NOTICE file distributed with this work for additional 
 information regarding copyright ownership. The ASF licenses this file to 
 You under the Apache License, Version 2.0 (the "License"); you may not use 
 this file except in compliance with the License. You may obtain a copy of 
 the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required 
 by applicable law or agreed to in writing, software distributed under the 
 License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS 
 OF ANY KIND, either express or implied. See the License for the specific 
 language governing permissions and limitations under the License. -->

<!-- Configures the Camel Context -->

<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:cxf="http://camel.apache.org/schema/cxf"
 xsi:schemaLocation="
       http://camel.apache.org/schema/cxf http://camel.apache.org/schema/cxf/camel-cxf.xsd
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
 
 <cxf:cxfEndpoint xmlns:tns="http://www.webserviceX.NET" 
  id="calculatorWSEndpoint"
  endpointName="tns:GlobalWeatherSoap"
  address="http://www.webservicex.com/globalweather.asmx"
  wsdlURL="http://www.webservicex.com/globalweather.asmx?WSDL">
  <cxf:properties>
   <entry key="dataFormat" value="PAYLOAD" />
  </cxf:properties>
 </cxf:cxfEndpoint>

 <camelContext xmlns="http://camel.apache.org/schema/spring">
  <package>com.bssys.fileid</package>
 </camelContext>

</beans>

Напомню что взамодествие с webservice может происходить в двух режимах

MESSAGE
PAYLOAD

Для получения информации о погоде будем использовать объявленый метод GetWeather

В режиме MESSAGE сообщение представляет из себя полный SOAP-запрос.
В режиме PAYLOAD сообщение из body SOAP-запроса.

Запрос в PAYLOAD-режиме

<ns1:GetWeather xmlns:ns1="http://www.webserviceX.NET">
 <ns1:CityName>Moscow</ns1:CityName>
 <ns1:CountryName/>
</ns1:GetWeather>

Запрос в MESSAGE-режиме

<s11:Envelope xmlns:s11="http://schemas.xmlsoap.org/soap/envelope/">
  <s11:Body>
    <ns1:GetWeather xmlns:ns1="http://www.webserviceX.NET">
      <ns1:CityName>Moscow</ns1:CityName>
      <ns1:CountryName/>
    </ns1:GetWeather>
  </s11:Body>
</s11:Envelope>

Запустив наш маршрут увидим в папке ответ от webservice:

<GetWeatherResponse xmlns="http://www.webserviceX.NET" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"><GetWeatherResult>&lt;?xml version="1.0" encoding="utf-16"?&gt;
&lt;CurrentWeather&gt;
  &lt;Location&gt;Moscow / Vnukovo , Russia (UUWW) 55-39N 037-16E&lt;/Location&gt;
  &lt;Time&gt;Aug 08, 2012 - 04:00 AM EDT / 2012.08.08 0800 UTC&lt;/Time&gt;
  &lt;Wind&gt; from the W (270 degrees) at 9 MPH (8 KT):0&lt;/Wind&gt;
  &lt;Visibility&gt; greater than 7 mile(s):0&lt;/Visibility&gt;
  &lt;SkyConditions&gt; mostly clear&lt;/SkyConditions&gt;
  &lt;Temperature&gt; 73 F (23 C)&lt;/Temperature&gt;
  &lt;DewPoint&gt; 50 F (10 C)&lt;/DewPoint&gt;
  &lt;RelativeHumidity&gt; 43%&lt;/RelativeHumidity&gt;
  &lt;Pressure&gt; 29.94 in. Hg (1014 hPa)&lt;/Pressure&gt;
  &lt;Status&gt;Success&lt;/Status&gt;
&lt;/CurrentWeather&gt;</GetWeatherResult></GetWeatherResponse>

Apache Camel Message Processing Actions (Like OSB)

2012-08-06T23:40:00.002+10:00

Работая в проекте с OSB, я удивился как легко и просто там работать с xml. Модификация сообщения на Oracle Service Bus действительно делается легко и непринужденно c помощью инструменов Assign, Delete, Insert. Как же наш Apache Camel? А вот в нем не все так просто. Но все необходимое у нас есть, попробуем сделать что-то подобное!

Интерфейс Exchange предоставляет нам метод

void setProperty(String name, Object value)
Sets a property on the exchange

Это позволяет нам сохранять переменные в контексте всего процесса обработки сообщения.
Вот что представляет собой элемент Assign в OSB:

Данный элемент сохраняет выражение в переменной.Выражение может быть результатом XQuery, XSLT преобразования или просто строго определенным xml.
Аналогом этого элемента является следющее выражение Java DSL:

from("file:c:/data/inbox").setProperty("original").simple("${body}");

В данном случае мы в переменную "original" сохранили тело запроса, помимо simple, Camel позволяет позволяет выполнить XPath выражение, XQuery и XSLT, и многое другое. В данном случае, чтобы сохранить тело оригинального запроса нам достаточно "simple".

C элементами Delete и Insert не все так просто.
Вот интерфейс элементов OSB
Delete:

XPath, выражение результат которого будет удален из переменной "original"
Для Apache Camel это будет реализовано с помощью процессора:

beanRef("delete","execute('original', '//record[@recordUUID=1]')");

"delete" -имя java-bean заданное в spring-context.xml
'execute'- имя метода
'original' - имя переменной из которой будет произведено удаление
'//record[@recordUUID=1]')' - XPath выражение

Insert:

beanRef("insert","execute('raw','//records', 'original')")

"insert" - имя java-bean заданное в spring-context.xml
'raw' - имя переменной значение которой будет вставляться
'//records' - Xpath выражение, в переменной original, обозначает место в которое будет произведена вставка.
'original' - имя переменной которое будет изменено.

На Apache Camel нам потребуется разработать два процессора:

DeleteProcessor:

package com.bssys.fileid.message;

import org.apache.camel.Exchange;
import org.apache.camel.builder.xml.XPathBuilder;
import org.w3c.dom.Document;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;

public class DeleteProcessor {
 
 public void execute(String invariable, String xpath, Exchange exchange ) {
  
  Document docVariable = exchange.getProperty(invariable, Document.class);
  
  XPathBuilder xpathBuilder = XPathBuilder.xpath(xpath);
  NodeList list = xpathBuilder.evaluate(exchange.getContext(), docVariable, NodeList.class);
  
  if (list!=null){
   for (int size = list.getLength(), i = 0; i < size; i++) {
    Node child = list.item(i);
    Node parentNode = child.getParentNode();
    parentNode.removeChild(child);
   }
  }
  exchange.setProperty(invariable, docVariable);
 }
}

InsertProcessor

package com.bssys.fileid.message;

import org.apache.camel.Exchange;
import org.apache.camel.builder.xml.XPathBuilder;
import org.w3c.dom.Document;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;

public class InsertProcessor {

 public void execute(String variable, String xpath, String invariable, Exchange exchange) {

  Document docVariable = exchange.getProperty(variable, Document.class);
  Document docInVariable = exchange.getProperty(invariable, Document.class);

  Node rootNode = docVariable.getFirstChild();
  
  if (rootNode != null) {

   XPathBuilder xpathBuilder = XPathBuilder.xpath(xpath);
   NodeList list = xpathBuilder.evaluate(exchange.getContext(), docInVariable, NodeList.class);
   
   for (int i = 0; i < list.getLength(); i++) {
    Node item = list.item(i);
    Node importNode = docInVariable.importNode(rootNode, true);
    item.appendChild(importNode);
   }
  }
  exchange.setProperty(invariable, docInVariable);
 }
}

Теперь воспользуемся разработаными инструментами и построим следующий маршрут.
На входе у нас файловый адаптер, который принимает xml-файлы:

<?xml version="1.0" encoding="UTF-8"?>
<storeRegistryCardsRequest>
 <records>
  <record recordUUID="1">
   <cardHolderIdentifiers>cardHolderIdentifiers</cardHolderIdentifiers>
   <card>card</card>
  </record>
  <record recordUUID="2">
   <cardHolderIdentifiers>cardHolderIdentifiers</cardHolderIdentifiers>
   <card>card</card>
  </record>
  <record recordUUID="3">
   <cardHolderIdentifiers>cardHolderIdentifiers</cardHolderIdentifiers>
   <card>card</card>
  </record>
 </records>
</storeRegistryCardsRequest>

Удаляет record с UUID=1, и добавляет record c UUID=4 Вот такой маршрут у нас получился в итоге.

from("file:c:/data/inbox3").
  setProperty("original").simple("${body}").
  beanRef("delete","execute('original', '//record[@recordUUID=1]')").
  setProperty("raw").simple("<record recordUUID=\"4\"><cardHolderIdentifiers>cardHolderIdentifiers</cardHolderIdentifiers><card>card</card></record>").
  beanRef("insert","execute('raw','//records', 'original')").
  setBody().simple("${property.original}").
  to("file:c:/data/outbox");

Результат обработки:

<storeRegistryCardsRequest>
 <records>

  <record recordUUID="2">
   <cardHolderIdentifiers>cardHolderIdentifiers</cardHolderIdentifiers>
   <card>card</card>
  </record>
  <record recordUUID="3">
   <cardHolderIdentifiers>cardHolderIdentifiers</cardHolderIdentifiers>
   <card>card</card>
  </record>
  <record recordUUID="4">
   <cardHolderIdentifiers>cardHolderIdentifiers</cardHolderIdentifiers>
   <card>card</card>
  </record>
 </records>
</storeRegistryCardsRequest>

Удачи!

Minitroopers Train Bot

2012-07-23T23:59:00.002+10:00

Для тех кто хочет немного отвлечься, есть замечательная браузерка minitroopers.com
Описание и прочее выкладывать не буду, в интернете информации море. Для тех кто увлекся могут почитать http://minitroopers.wikia.com/wiki/Mini_Troopers_Wiki
Чтобы не тратить драгоценное время на прокачку твинков своей основной армии, я написал простого бота, который отыгрывает за вас.

Ежедневные бои. Выбирает первого кандидата из списка. По опыту первый самый слабый, но хотя если и проиграете не страшно.
Миссия из трех попыток. Если она у вас открыта.
5 шагов рейда. Если рейд у твинка больше 5 боев, то на это стоит поглядеть!

Для запуска необходима Java.

java -jar minitroopers.jar

Имена твинков заносятся в файлик minitroopers.property в следующем формате:

URL^login^password

Например:

http://dd-dealer.minitroopers.com^dd-dealer^ddpassword

http://izilla.minitroopers.com

Если ваш твинк запаролен то логин в 99% совпадает с URL, но если у вас это не так, можете посмотреть в браузере какой логин отправляется и записать его в файлик.

Ссылка на бинарник:

Minitroopers Bot

Файл с твинками:

minitroopers.property

Ну и по традиции подобных игр. Добавляйтесь в мою армию!
http://hibernate.minitroopers.com/
Удачи в прокачке!

JDBC Polling Adapter c помощью Apache Camel

2012-05-30T22:45:00.000+10:00

В качестве wiki and issue tracking system у нас используется TRAC http://trac.edgewall.org/ Появилась задача мониторинга изменения задач в этой системе. TRAC написан на Python и поддерживает из коробки только один вид нотификации-email. Активность по тикетам в системе должна выгружаться в xml-файл для обработки другой системой. Идеальная задача для Camel ;)

Первоначальная идея решения была основана на работе с email. Однако по причине ее ущербности она была отброшена.
Быстро окинув взглядом что нам предоставляет Camel я к своему удивлению не нашел такой простой вещи как JDBC-адаптер. Адаптер, в полном смысле этого понятия, с помощью которого возможна передача сообщений из БД в интеграционное решение. Как пример что это такое можно посмотреть на JDBC-адаптер от Websphere ESB JDBC Adapter.

Основной принцип двунаправленной коммуникации заключается в следующем. В БД которую необходимо мониторить заводится служебная табличка, которую с некоторым интервалом опрашивает адаптер, забирает данные и передает в интеграционное решение. Данные в служебную табличку складываются с помощью тригера.

Основные компоненты у на сбудут следующие:

Таймер, инициализирующий событие мониторинга служебной таблички:
Процессор, который выбирает из таблички сообщения и асинронно передает их на обработку.
Процессор, непосредственно обрабатывающий сообщения.

Далее нам протребуется механизм пуллинга, позволяющий отправлять сообщения напрямую из процессора. Для этого в Apache Camel предусмотрен ProducerTemplate.

Итак вот каркас нашего приложения: camel-context.xml Внутри camelContext инстанцируется Producer, и далее по ID инжектируется в JDBCAdapter


<?xml version="1.0" encoding="UTF-8"?>
<!-- Licensed to the Apache Software Foundation (ASF) under one or more contributor 
 license agreements. See the NOTICE file distributed with this work for additional 
 information regarding copyright ownership. The ASF licenses this file to 
 You under the Apache License, Version 2.0 (the "License"); you may not use 
 this file except in compliance with the License. You may obtain a copy of 
 the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required 
 by applicable law or agreed to in writing, software distributed under the 
 License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS 
 OF ANY KIND, either express or implied. See the License for the specific 
 language governing permissions and limitations under the License. -->

<!-- Configures the Camel Context -->

<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

 <bean id="JDBCAdapter" class="foo.JDBCAdapter">
  <property name="producer" ref="change" />
 </bean>

 <camelContext xmlns="http://camel.apache.org/schema/spring">
  <package>foo</package>
  <template id="change" defaultEndpoint="seda:change" />
 </camelContext>

</beans>


package foo;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.ProducerTemplate;

public class JDBCAdapter implements Processor {

 private ProducerTemplate producer;

 public void setProducer(ProducerTemplate producer) {
  this.producer = producer;
 }

 @Override
 public void process(Exchange arg0) throws Exception {
  // TODO Auto-generated method stub
  producer.sendBody("Ok");
 }
}

Итоговый роутер:

package foo;

import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.spring.Main;

import static org.apache.camel.builder.xml.XPathBuilder.xpath;

/**
 * A Camel Router
 */
public class MyRouteBuilder extends RouteBuilder {

    /**
     * A main() so we can easily run these routing rules in our IDE
     */
    public static void main(String... args) throws Exception {
        Main.main(args);
    }

    /**
     * Let's configure the Camel routing rules using Java code...
     */
    public void configure() {
     from("timer://foo?fixedRate=true&period=1000").processRef("JDBCAdapter");
     from("seda:change").log("Yes baby!${body}");
    }
}

Обратите внимание что endpoint у нас определяется как seda вместо direct, что обеспечивает нужную нам асинхронность! Проверим что все работает как было задумано.

[                          main] MainSupport                    INFO  Apache Camel 2.9.2 starting
[                          main] AnnotationTypeConverterLoader  INFO  Found 3 packages with 15 @Converter classes to load
[                          main] DefaultTypeConverter           INFO  Loaded 170 core type converters (total 170 type converters)
[                          main] DefaultTypeConverter           INFO  Loaded additional 0 type converters (total 170 type converters) in 0.000 seconds
[                          main] SpringCamelContext             INFO  Apache Camel 2.9.2 (CamelContext: camel-1) is starting
[                          main] ManagementStrategyFactory      INFO  JMX enabled. Using ManagedManagementStrategy.
[                          main] ultManagementLifecycleStrategy INFO  StatisticsLevel at All so enabling load performance statistics
[                          main] SpringCamelContext             INFO  Route: route1 started and consuming from: Endpoint[timer://foo?fixedRate=true&period=1000]
[                          main] SpringCamelContext             INFO  Route: route2 started and consuming from: Endpoint[seda://change]
[                          main] SpringCamelContext             INFO  Total 2 routes, of which 2 is started.
[                          main] SpringCamelContext             INFO  Apache Camel 2.9.2 (CamelContext: camel-1) started in 0.234 seconds
[l-1) thread #2 - seda://change] route2                         INFO  Yes baby!Ok
[l-1) thread #2 - seda://change] route2                         INFO  Yes baby!Ok
[l-1) thread #2 - seda://change] route2                         INFO  Yes baby!Ok
[l-1) thread #2 - seda://change] route2                         INFO  Yes baby!Ok

Как можно догадаться вся логика обработки сообщений будет определятся роутингом:

from("seda:change").log("Yes baby!${body}");

Вернемся к нашей задаче. Trac хранит все изменения по тикетам в табличке ticket_change следующей структуры Создадим служебную табличку, которая полностью копирует структуру ticket_change, у меня она называется ticket_change_audit Тригер, который срабатывает при добавлении данных:

CREATE OR REPLACE FUNCTION etk."notify"()
  RETURNS trigger AS
$BODY$BEGIN
insert into etk.ticket_change_audit (ticket, time, author, field, oldvalue, newvalue) values(NEW.ticket, NEW.time, NEW.author, NEW.field, NEW.oldvalue, NEW.newvalue);
END;$BODY$
  LANGUAGE plpgsql VOLATILE
  COST 100;
ALTER FUNCTION etk."notify"() OWNER TO tracuser;
COMMENT ON FUNCTION etk."notify"() IS 'dsd';

Теперь нам осталось доработать логику процессора JDBCAdapter'а. Создаем следующий бизнес-объект для нашей таблички(JPA и другие ORM не использую специально чтобы не усложнять наш пример)

package com.bssys.trac.bo;

import java.io.Serializable;

public class TicketChange implements Serializable {

 private Integer ticket;

 private Long time;

 private String author;

 private String field;

 private String oldValue;

 private String newValue;

 public Integer getTicket() {
  return ticket;
 }

 public String getField() {
  return field;
 }

 public void setField(String field) {
  this.field = field;
 }

 public void setTicket(Integer ticket) {
  this.ticket = ticket;
 }

 public Long getTime() {
  return time;
 }

 public void setTime(Long time) {
  this.time = time;
 }

 public String getAuthor() {
  return author;
 }

 public void setAuthor(String author) {
  this.author = author;
 }

 public String getOldValue() {
  return oldValue;
 }

 public void setOldValue(String oldValue) {
  this.oldValue = oldValue;
 }

 public String getNewValue() {
  return newValue;
 }

 public void setNewValue(String newValue) {
  this.newValue = newValue;
 }
}

И добавляем логику работы с БД с помощью спрингового JDBCTemplate. Далее необходимо будет просто реализвать нужную логику обработки сообщения, в теле message Exchange, будет прилетать объект TicketChange Вот что у нас получилось в итоге:

package com.bssys.trac;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.ProducerTemplate;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;

import com.bssys.trac.bo.TicketChange;

public class JDBCAdapter implements Processor {

 private JdbcTemplate template;
 private ProducerTemplate producer;

 public void setTemplate(JdbcTemplate template) {
  this.template = template;
 }

 public void setProducer(ProducerTemplate producer) {
  this.producer = producer;
 }

 @Override
 public void process(Exchange exchange) throws Exception {
  // TODO Auto-generated method stub

  List res = template.query("select * from etk.ticket_change_audit", new RowMapper() {

   @Override
   public TicketChange mapRow(ResultSet rs, int rowNum) throws SQLException {
    // TODO Auto-generated method stub
    TicketChange ticketChange = new TicketChange();
    ticketChange.setTicket(rs.getInt("ticket"));
    ticketChange.setTime(rs.getLong("time"));
    ticketChange.setAuthor(rs.getString("author"));
    ticketChange.setField(rs.getString("field"));
    ticketChange.setOldValue(rs.getString("oldvalue"));
    ticketChange.setNewValue(rs.getString("newvalue"));

    return ticketChange;
   }
  });

  for (TicketChange change : res) {

   TicketChangeSendProcessor sendProcessor = new TicketChangeSendProcessor(change);
   producer.send(sendProcessor);

   String delete = "delete from etk.ticket_change_audit WHERE ticket=? and time=? and author=? and field=? and oldvalue=? and newvalue=?";
   template.update(delete, change.getTicket(), change.getTime(), change.getAuthor(), change.getField(), change.getOldValue(), change.getNewValue());

  }

  exchange.getOut().setBody(res);
 }
}

Блеск и нищета Apache Camel

2012-03-30T01:44:00.000+10:00

В этой статье я расскажу про фееричный фейл Apache Camel , на реальной бизнес-задаче.
Возникла задача по обработке 5000000 zip-файлов (O_o). Каждый файлик необходимо распаковать, достать из архива XML файл, распарсить его, и некоторую информацию дописать в лог в определенном формате.

Идеальная задача для Apache Camel подумал я и бросился в бой!

Один роутинг и один процессор, вот что у меня вышло.

Роутинг:

public void configure() {
 from("file:c://data/inbox?noop=true").
  log("${file:name}").
  convertBodyTo(InputStream.class).
  processRef("ZipProcess").
 to("file:c://data/outbox/?fileExist=Append&fileName=report.txt");
}

Процессор:

package com.bssys.zoom.zipbot;

import java.io.BufferedInputStream;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;

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

public class ZipProcess implements Processor {

 @Override
 public void process(Exchange exchange) throws Exception {
  // TODO Auto-generated method stub
  InputStream is = exchange.getIn().getBody(InputStream.class);
  ZipInputStream zipInputStream = new ZipInputStream(new BufferedInputStream(is));

  String res = null;
  ZipEntry zipEntry = null;
  while ((zipEntry = zipInputStream.getNextEntry()) != null) {
   String fileName = zipEntry.getName();
   if (fileName.endsWith(".XML")) {
    ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
    byte buf[] = new byte[1024];
    int count = -1;
    while ((count = zipInputStream.read(buf)) != -1) {
     byteArrayOutputStream.write(buf, 0, count);
    }

    String xml = new String(byteArrayOutputStream.toByteArray());
    //... Тут парсим XML и выбираем нужные данные
    res = String.format("%s:\r\n", fileName);
   }
   zipInputStream.closeEntry();
  }
  zipInputStream.close();
  exchange.getOut().setBody(res);
 }
}

Эта штука еле взлетела отожрав 5гб оперативной памяти. Скорость работы вполне приличная 1000 файлов за 300ms. Производительность вполне достаточная, если учесть что парсер можно оптимизировать или совсем отказаться от него, используя indexOf. Через пару часов работы, заглядывая в результирующий файлик, всплыло неладное. Данные повторялись! Т.е. Camel брал один и тотже файл по нескольку раз. На моем тестовом экземпляре в 1000 файлов все работало без нареканий. Заглянув документацию заметил следующий комментарий по параметру noop:

If true, the file is not moved or deleted in any way. This option is good for readonly data, or for ETL type requirements. If noop=true, Camel will set idempotent=true as well, to avoid consuming the same files over and over again.

Этот параметр я проставляю для конечной точки, что бы Camel не удалял и не перемещал файлы из исходной папки. Если проставляется noop=true то автоматически проставляется также параметр idempotent=true

Option to use the Idempotent Consumer EIP pattern to let Camel skip already processed files. Will by default use a memory based LRUCache that holds 1000 entries. If noop=true then idempotent will be enabled as well to avoid consuming the same files over and over again.

Все стало ясно. Camel в режиме мониторинга устанавливает кэш в 1000 файлов, чтобы не забирать уже обработанные файлы. И просто не предназначен для такой ситуации с 5000000 файлов. Можно не устанавливать флаг noop, но при этом сильно режется производительность. Вот этого я никак не ожидал. Хотя по логике ничего работа Camel Не вызывает нареканий. В режиме мониторинга мы упираемся в этуже проблему. В результате без использования Camel, была написана программка, которая и выполнила эту непосильную задачу, заняв при этом меньшее количество памяти. Привожу часть просто для интереса.

public void go(String inbox, String outbox) throws Exception {

  File inboxDir = new File(inbox);

  if (inboxDir.isDirectory()) {
   String[] listFiles = inboxDir.list();
   System.out.println("Count files: " + listFiles.length);
   BufferedOutputStream bufferedOutputStream = null;
   try {
    bufferedOutputStream = new BufferedOutputStream(new FileOutputStream(new File(outbox + "/report.txt")));
    long count = 0;
    for (String fileName : listFiles) {
     count++;
     System.out.println(count + ":" + fileName);
     FileInputStream fileInputStream = new FileInputStream(new File(inbox + "/" + fileName));
     try{
      String res = process(fileInputStream);
      bufferedOutputStream.write(res.getBytes());
     }catch(Exception e){
      FileOutputStream fileOutputStream = new FileOutputStream(new File(outbox + "/fail.txt"), true);
      fileOutputStream.write((fileName+"\r\n").getBytes());
      fileInputStream.close();
     }
    }
    bufferedOutputStream.flush();

   } finally {
    if (bufferedOutputStream != null) {
     bufferedOutputStream.close();
    }
   }

  }
 }

Метод process, собственно на 99% схож с кодом процессора. Удачи!

Файловый интерфейс для Base64 кодирования/декодирования с помощью Apache Camel

2012-03-30T00:24:00.002+10:00

Чем мне нравится Apache Camel, так это готовыми компонентами конечных точек(endpoints) на все случаи жизни. В моей жизни встретилась задача, по Base64 кодированию/декодированию множества файлов.

Схема работы приложения следующая:
В соответствующие папки будут кидаться файлики, результат будет формироваться в папке output:

c://data/encode
c://data/decode

c://data/output

Идеальная задача для Apache Camel!

Как обычно создаем Maven Project на основе уже известного нам camel-archetype-java.
Добавляем в зависимости к проекту сommon-codec, с помощью которого будет проводиться кодирование/декодирование

 
<dependency>
 <groupId>commons-codec</groupId>
 <artifactId>commons-codec</artifactId>
 <version>1.6</version>
</dependency>

Создаем два процессора которые будут кодировать и раскодировать поступающие сообщения.

Декодировщик:

package com.bssys.base64.convert;

import java.io.BufferedInputStream;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.commons.codec.binary.Base64;

public class Base64DecodeProcessor implements Processor {

 @Override
 public void process(Exchange exchange) throws Exception {
  // TODO Auto-generated method stub
  InputStream is = exchange.getIn().getBody(InputStream.class);
  String fileName = (String)exchange.getIn().getHeader(Exchange.FILE_NAME);
  BufferedInputStream bufferedInputStream = new BufferedInputStream(is);
  ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
  byte buf[] = new byte[1024];
  int pos=-1;
  while((pos= bufferedInputStream.read(buf))!=-1){
   byteArrayOutputStream.write(buf, 0, pos);
  }
  bufferedInputStream.close();
  byte[] encodeBytes = Base64.decodeBase64(byteArrayOutputStream.toByteArray());
  exchange.getOut().setHeader(Exchange.FILE_NAME, fileName);
  exchange.getOut().setBody(encodeBytes);
 }
}

Кодировщик:

package com.bssys.base64.convert;

import java.io.BufferedInputStream;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.commons.codec.binary.Base64;

public class Base64EncodeProcessor implements Processor {

 @Override
 public void process(Exchange exchange) throws Exception {
  // TODO Auto-generated method stub
  InputStream is = exchange.getIn().getBody(InputStream.class);
  String fileName = (String)exchange.getIn().getHeader(Exchange.FILE_NAME);
  BufferedInputStream bufferedInputStream = new BufferedInputStream(is);
  ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
  byte buf[] = new byte[1024];
  int pos = -1;
  while ((pos = bufferedInputStream.read(buf)) != -1) {
   byteArrayOutputStream.write(buf, 0, pos);
  }
  bufferedInputStream.close();
  byte[] encodeBytes = Base64.encodeBase64(byteArrayOutputStream
    .toByteArray());
  exchange.getOut().setHeader(Exchange.FILE_NAME, fileName);
  exchange.getOut().setBody(encodeBytes);

 }
}

Добавляем наши 2 процессора в spring-контекст

<bean id="Base64Encode" class="com.bssys.base64.convert.Base64EncodeProcessor"/>     
<bean id="Base64Decode" class="com.bssys.base64.convert.Base64DecodeProcessor"/>

И модифицируем MyRouteBuilder.java добавив в него 2 роутера:

public void configure() {
    // TODO create Camel routes here.
    from("file:c://data/encode").convertBodyTo(InputStream.class).to("Base64Encode").to("file:c://data/outbox");
    from("file:c://data/decode").convertBodyTo(InputStream.class).to("Base64Decode").to("file:c://data/outbox");
}

Запускаем и наслаждаемся. Apache Camel слушает заданные папки, и при поступлении файликов выполняет за нас всю черную работу.