The Skip Locked feature in Postgres 9.5 (message queue)

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

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

Dynamic Webservice registration on ServiceMix

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

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

If-less programming (java) Part №2

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

If-less programming (java) Part №1

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

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

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

ServiceMix Atomikos XA transaction JDBC JMS

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

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

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

WebSocket server java. Best way

Итак, когда мы разобрались как реализовать 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);
    }
}

NIO2.0 Asynchronous I/O. WebSocket chat server

Для лучшего понимания 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

Understanding NIO2.0 Asynchronous I/O. EchoServer

Посмотрим внимательно на 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.

Client notification using ZeroMQ

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

Authentication and authorization ActiveMQ

Итак, если вам надо управлять доступом к очередям на 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

В этой статье я решил поделится всеми своими рекомендациями при разработке с применением 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