Доброго времени суток, %username%.
Скажу сразу, на 99% данный пост про Java EE Connector Architecture, с примерами кода. Откуда взялся 1% про Fidonet вы поймете в самом конце.
Резюме для ленивых
JMS и JCA — родственники, входящие принимает MessageDrivenBean, исходящие отправляются через ConnectionFactory.
Минимальный пакет для входящего соединения — 4 класса, для исходящего — 8 классов и настройка адаптера на стороне сервера приложений.
Дальше — только подробности и боль
Минимальный пакет для входящего соединения — 4 класса, для исходящего — 8 классов и настройка адаптера на стороне сервера приложений.
Дальше — только подробности и боль
Для начала — история вопроса и решение бизнес-задачи.
Постановка задачи
Мне поставили задачу об интеграции существующей бизнес-системы («система А» ) с другой системой, которая была разработана много лет назад и понимает только собственный протокол передачи данных («система B»). Модифицировать чужие системы нельзя, соответственно задача свелась к написанию некой шины/прокси. Интеграция состоит в передаче туда-сюда сообщений с конвертацией их в процессе из одного формата в другой.
Система «А» имела много современных механизмов интеграции, самым простым для использования были признаны веб-сервисы. Под это дело был оперативно запилен стандартный интеграционный скелет для JEE — JAX-WS+EJB+JMS для гарантированной доставки сообщения.
А вот для работы с системой «B» стандартных средств не было. Робкие попытки поработать с сетью из контекста EJB успехом не увенчались, гугл подсказал два варианта решения проблемы: костылить сервлеты для работы с non-http или написать JCA-адаптер. Понятно, что был выбран второй путь — с JCA я до этого не работал, а узнать что-то новое всегда интересно.
Исследование
Начав копать гугл, я достаточно сильно обломался. Везде писали, ЧТО именно нужно сделать ( коннектор, менеджер, адаптер итд ), но почти нигде не писали, КАК это сделать. Стандартный способ «посмотреть чужой код и понять процесс» дал сбой — чужого кода был такой мизер, что понять что-то мне не представлялось возможным.
Спасли меня две вещи: JSR 322 и единственный нагугленный адаптер на google code. Собственно, это и стало отправной точкой — задеплоив примеры из jca-sockets и открыв pdf, я начал разбираться и путем научного тыка понимать, как оно собственно работает.
Потратив около 16 часов на исследование и эксперименты, я выяснил следующее:
JCA-модуль имеет внутри себя две независимых части: «Входящие» и «Исходящие». Эти части могут быть как вместе, так и по-отдельности. Более того, их может быть несколько. Сам модуль регистритуется классом, реализующим javax.resource.spi.ResourceAdapter и указанным в META-INF/ra.xml, при этом ResourceAdapter нужен в первую очередь для работы с Входящими; Для Исходящих адаптер ничего не делает и его скелет можно даже не заполнять.
Входящие
Входящий канал биндится к MessageEndpoint'у ( обычно это @MessageDrivenBean; да-да, JCA это кишки JMS ) и активируется ActivationSpec'ом.
META-INF/ra.xml — описание ResourceAdapter'а и inbound потоков
ra.xml
<connector xmlns="http://xmlns.jcp.org/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee
http://xmlns.jcp.org/xml/ns/javaee/connector_1_7.xsd"
version="1.7" metadata-complete="true">
<vendor-name>xxx-services</vendor-name>
<eis-type>FidoNet</eis-type>
<resourceadapter-version>2.5</resourceadapter-version>
<resourceadapter>
<!-- Класс, который реализует javax.resource.spi.ResourceAdapter; config-property - поля, должны быть доступны через геттеры/сеттеры -->
<resourceadapter-class>in.fidonode.binkp.ra.BinkpServerResourceAdapter</resourceadapter-class>
<config-property>
<config-property-name>version</config-property-name>
<config-property-type>java.lang.String</config-property-type>
<config-property-value>jnode-jee 2.5 binkp/1.1</config-property-value>
</config-property>
<!-- Описание входящего потока -->
<inbound-resourceadapter>
<messageadapter>
<messagelistener>
<!-- Интерфейс, который должен реализовать @MessageDrivenBean для того, чтоб получать сообщения с этого адаптера -->
<messagelistener-type>in.fidonode.binkp.ra.BinkpMessageListener</messagelistener-type>
<activationspec>
<!-- Класс-холдер параметров, которая передается через @ActivationConfigProperty, должны быть доступны геттеры и сеттеры параметров -->
<activationspec-class>in.fidonode.binkp.ra.BinkpActivationSpec</activationspec-class>
<!-- Список обязательных параметров -->
<required-config-property>
<config-property-name>listenPort</config-property-name>
</required-config-property>
<!-- Описание параметров параметров -->
<config-property>
<config-property-name>listenPort</config-property-name>
<config-property-type>java.lang.Integer</config-property-type>
<config-property-value>24554</config-property-value>
</config-property>
</activationspec>
</messagelistener>
</messageadapter>
</inbound-resourceadapter>
</resourceadapter>
</connector>
Интерфейс BinkpMessageListener — для клиентов и должен быть в classpath;
Приведу его тут:
public interface BinkpMessageListener {
public void onMessage(FidoMessage message);
}
Теперь рассмотрим простейшую реализацию ResourceAdapter
BinkpServerResourceAdapter.java
public class BinkpServerResourceAdapter implements ResourceAdapter, Serializable {
private static final long serialVersionUID = 1L;
private static Logger log = Logger.getLogger(BinkpServerResourceAdapter.class
.getName());
private ConcurrentHashMap<BinkpActivationSpec, BinkpEndpoint> activationMap =
new ConcurrentHashMap<BinkpActivationSpec, BinkpEndpoint>();
private BootstrapContext ctx;
private String version;
@Override
public void endpointActivation(MessageEndpointFactory endpointFactory,
ActivationSpec spec) throws ResourceException {
BinkpEndpoint activation = new BinkpEndpoint(ctx.getWorkManager(),
(BinkpActivationSpec) spec, endpointFactory);
activationMap.put((BinkpActivationSpec) spec, activation);
activation.start();
log.info("endpointActivation(" + activation + ")");
}
@Override
public void endpointDeactivation(MessageEndpointFactory endpointFactory,
ActivationSpec spec) {
BinkpEndpoint activation = activationMap.remove(spec);
if (activation != null)
activation.stop();
log.info("endpointDeactivation(" + activation + ")");
}
@Override
public void start(BootstrapContext ctx)
throws ResourceAdapterInternalException {
this.ctx = ctx;
log.info("start()");
}
@Override
public void stop() {
for (BinkpEndpoint act : activationMap.values()) {
act.stop();
}
activationMap.clear();
log.info("stop()");
}
@Override
public XAResource[] getXAResources(ActivationSpec[] arg0)
throws ResourceException {
return null;
}
public String getVersion() {
return version;
}
public void setVersion(String version) {
this.version = version;
}
}
Что тут происходит? При загрузке JCA-модуля создается экземпляр класса BinkpServerResourceAdapter, у него заполняются параметры ( в данном случае — поле version) и вызывается метод start().
На самом деле, внутри метода start() можно делать много всего, но в данном примере мы просто сохраняем контекст для получения из него в дальнейшем WorkManager'а.
Когда сервер приложений находит @MessageDrivenBean, он пытается найти адаптер, который отправляет сообщения на тот интерфейс, который реализует бин. Для JMS это MessageListener, у нас это BinkpMessageListener. Создается ActivationSpec ( у нас это BinkpActivationSpec, реализующий javax.resource.spi.ActivationSpec), поля в котором заполняются согласно данным в activationConfig, создается MessageEndpointFactory и вызывается ResourceAdapter.endpointActivation(). В этой функции необходимо создать тот «сервер», который будет принимать входящие соединения, будь то tcp/ip сервер или поток для работы с unix-socket, создать на основе того конфига который был в MDB. Класс BinkpEndpoint — это и есть тот самый «сервер».
BinkpEndpoint.java
public class BinkpEndpoint implements Work, FidoMessageListener {
private static final Logger logger = Logger.getLogger(BinkpEndpoint.class
.getName());
private BinkpServer server;
private final WorkManager workManager;
private final MessageEndpointFactory messageEndpointFactory;
public BinkpEndpoint(WorkManager workManager,
BinkpActivationSpec activationSpec,
MessageEndpointFactory messageEndpointFactory) {
this.workManager = workManager;
this.messageEndpointFactory = messageEndpointFactory;
server = new BinkpServer(activationSpec.getListenPort(), this);
}
public void start() throws ResourceException {
workManager.scheduleWork(this);
}
public void stop() {
if (server != null) {
server.stop();
}
}
/** из FidoMessageListener **/
@Override
public Message incomingMessage(FidoMessage message) {
String message = msg.encode();
BinkpMessageListener listener = (BinkpMessageListener) messageEndpointFactory
.createEndpoint(null);
listener.onMessage(message);
}
/** из Work **/
@Override
public void run() {
server.start();
}
/** из Work **/
@Override
public void release() {
stop();
}
}
Можно заметить, что везде фигурируют некие endpoint'ы. У меня был с этим некоторый затык, поэтому расшифрую:
Endpoint — это то, что слушает «входящий» поток. Именно к нему относятся функции endpointActication
MessageEndpoint — экземпляр MDB, который обрабатывает то или иное сообщение. Получается вызовом MessageEndpointFactory.createEndpoint() ( Эту функцию нельзя вызывать из основного треда ). Он легко кастится к интерфейсу MDB.
Собственно, все. Реализацию BinkpServer за ненадобностью опущу, но принцип должен быть понятен, минимальный «Входящий» JCA делается из четырех классов ( ResourceAdapter, MessageListener, ActivationSpec, Endpoint )
Создание Endpoint'а и обработка входящих:
@MessageDriven(messageListenerInterface = BinkpMessageListener.class,
activationConfig = { @ActivationConfigProperty(propertyName = "listenPort", propertyValue = "24554") })
public class ReceiveMessageBean implements BinkpMessageListener {
@Override
public void onMessage(FidoMessage msg) {
// do smth with mesaage
}
}
Исходящие
А вот тут — все веселее, минимальный «Исходящий» JCA делается аж из 8 классов, что в 2 раза больше чем «Входящий». Но давайте по-порядку.
META-INF/ra.xml — описание ResourceAdapter'а и outbound потоков
ra.xml
<connector xmlns="http://xmlns.jcp.org/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee
http://xmlns.jcp.org/xml/ns/javaee/connector_1_7.xsd"
version="1.7" metadata-complete="true">
<vendor-name>xxx-services</vendor-name>
<eis-type>FidoNet</eis-type>
<resourceadapter-version>2.5</resourceadapter-version>
<resourceadapter>
<!-- Класс, который реализует javax.resource.spi.ResourceAdapter; config-property - поля, должны быть доступны через геттеры/сеттеры -->
<resourceadapter-class>in.fidonode.binkp.ra.BinkpServerResourceAdapter</resourceadapter-class>
<config-property>
<config-property-name>version</config-property-name>
<config-property-type>java.lang.String</config-property-type>
<config-property-value>jnode-jee 2.5 binkp/1.1</config-property-value>
</config-property>
<!-- Описание исходящего потока. Всегда ПЕРЕД входящим -->
<outbound-resourceadapter>
<connection-definition>
<!-- Фабрика с которой будет работать JEE-сервер, она создает фабрику для создания соединений -->
<managedconnectionfactory-class>in.fidonode.binkp.ra.ManagedConnectionFactory</managedconnectionfactory-class>
<!-- Фабика соединений, которая будет отдаваться по запросу. Интерфейс должен быть у клиента в classpath -->
<connectionfactory-interface>in.fidonode.binkp.ra.ConnectionFactory</connectionfactory-interface>
<connectionfactory-impl-class>in.fidonode.binkp.ra.ConnectionFactoryImpl</connectionfactory-impl-class>
<!-- Соединение, которое будет отдавать фабрика соединений. Интерфейс должен быть у клиента в classpath -->
<connection-interface>in.fidonode.binkp.ra.Connection</connection-interface>
<connection-impl-class>in.fidonode.binkp.ra.ConnectionImpl</connection-impl-class>
</connection-definition>
<!-- Про транзакции и аутентификацию есть отдельный талмуд, я этот момент пропустил -->
<transaction-support>NoTransaction</transaction-support>
<reauthentication-support>false</reauthentication-support>
</outbound-resourceadapter>
<!-- Описание входящего потока -->
<inbound-resourceadapter>
<!-- ... -->
</inbound-resourceadapter>
</resourceadapter>
</connector>
Интерфейсы Connection и ConnectionFactory — для клиентов и должны быть в classpath. Сразу приведу их тут, там ничего интересного.
BinkpClient приводить не буду :-)
public interface Connection {
public BinkpClient connect(String hostname, int port);
}
public interface ConnectionFactory {
public Connection createConnection();
}
Соединения бывают Managed и Unmanaged. Первые — со свистелками, listener'ами и другим, вторые — без.
Класс, реализующий ManagedConnectionFactory, должен уметь создавать оба типа соединений.
ManagedConnectionFactory.java
public class ManagedConnectionFactory implements
javax.resource.spi.ManagedConnectionFactory {
private PrintWriter logwriter;
private static final long serialVersionUID = 1L;
/**
* Создание фабрики для unmanaged-соединений
*/
@Override
public Object createConnectionFactory() throws ResourceException {
return new ConnectionFactoryImpl();
}
/**
* Создание managed-фабрики для managed-connection
*/
@Override
public Object createConnectionFactory(ConnectionManager cxManager)
throws ResourceException {
return new ManagedConnectionFactoryImpl(this, cxManager);
}
/**
* Создание managed-соединения
*/
@Override
public ManagedConnection createManagedConnection(Subject subject,
ConnectionRequestInfo cxRequestInfo) throws ResourceException {
return new in.fidonode.binkp.ra.ManagedConnection();
}
@Override
public PrintWriter getLogWriter() throws ResourceException {
return logwriter;
}
@SuppressWarnings("rawtypes")
@Override
public ManagedConnection matchManagedConnections(Set connectionSet,
Subject subject, ConnectionRequestInfo cxRequestInfo)
throws ResourceException {
ManagedConnection result = null;
Iterator it = connectionSet.iterator();
while (result == null && it.hasNext()) {
ManagedConnection mc = (ManagedConnection) it.next();
if (mc instanceof in.fidonode.binkp.ra.ManagedConnection) {
result = mc;
}
}
return result;
}
@Override
public void setLogWriter(PrintWriter out) throws ResourceException {
logwriter = out;
}
}
Когда приложение запрашивает у JEE-сервера тот или иной коннектор, сервер приложений просит ManagedConnectionFactory создать ConnectionFactory и отдает его приложению.
Как можно заметить, ConnectionFactory тоже бывает Managed и Unmanaged. В принципе все это можно свести к одному классу, но это сильно зависит от того, что именно и как мы передаем, есть ли там транзакции итд.
ConnectionFactoryIml просто делает new ConnectionImpl(), а вот ManagedConnectionFactoryImpl чуть посложнее:
ManagedConnectionFactoryImpl.java
public class ManagedConnectionFactoryImpl implements ConnectionFactory {
private ManagedConnectionFactory factory;
private ConnectionManager manager;
public ManagedConnectionFactoryImpl(ManagedConnectionFactory factory,
ConnectionManager manager) {
super();
this.factory = factory;
this.manager = manager;
}
/** создает managed-соединение через родителя-ManagedConnectionFactory **/
@Override
public Connection createConnection() {
try {
return (Connection) manager.allocateConnection(factory, null);
} catch (ResourceException e) {
return null;
}
}
}
ManagedConnection, реализующий javax.resource.spi.ManagedConnection — это обертка для интерфейса Connection, которая как-раз добавляет свистелок и listener'ов. Именно этот класс возвращает ManagedConnectionFactory.createManagedConnection(), которую мы вызываем при создании соединения из ManagedConnectionFactoryImpl.createConnection() через ConnectionManager.allocateConnection()
ManagedConnection.java
public class ManagedConnection implements javax.resource.spi.ManagedConnection {
private PrintWriter logWriter;
private Connection connection;
private List<ConnectionEventListener> listeners;
public ManagedConnection() {
listeners = Collections
.synchronizedList(new ArrayList<ConnectionEventListener>());
}
@Override
public void associateConnection(Object connection) throws ResourceException {
if (connection != null && connection instanceof Connection) {
this.connection = (Connection) connection;
}
}
@Override
public Object getConnection(Subject subject,
ConnectionRequestInfo cxRequestInfo) throws ResourceException {
if (connection == null) {
connection = new ManagedConnectionImpl();
}
return connection;
}
@Override
public void cleanup() throws ResourceException {
}
@Override
public void destroy() throws ResourceException {
}
@Override
public PrintWriter getLogWriter() throws ResourceException {
return logWriter;
}
@Override
public ManagedConnectionMetaData getMetaData() throws ResourceException {
throw new NotSupportedException();
}
@Override
public XAResource getXAResource() throws ResourceException {
throw new NotSupportedException();
}
@Override
public LocalTransaction getLocalTransaction() throws ResourceException {
return null;
}
@Override
public void setLogWriter(PrintWriter out) throws ResourceException {
logWriter = out;
}
@Override
public void addConnectionEventListener(ConnectionEventListener listener) {
if (listener != null) {
listeners.add(listener);
}
}
@Override
public void removeConnectionEventListener(ConnectionEventListener listener) {
if (listener != null) {
listeners.remove(listener);
}
}
}
Ну вот, теперь мы подошли к самому простому — реализации соединения :-)
public class ConnectionImpl implements Connection {
@Override
public BinkpClient connect(String hostname, int port) {
return new BinkpClient(hostname, port);
}
}
Итоговая цепочка вызовов для установления исходящего соединения
ManagedConnectionFactory.createConnectionFactory()
->ManagedConnectionFactoryImpl.createConnection()
-->СonnectionManager.allocateConnection()
--->ManagedConnectionFactory.createManagedConnection()
---->ManagedConnection.getConnection()
----->ManagedConnectionImpl.connect()
Ну и не забудьте настроить сервер приложений для работы с этим адаптером, ну и jndi указать.
Код для вызова:
private BinkpClient createBinkpClient(String host, int port) {
ConnectionFactory cf = ((ConnectionFactory) new InitialContext().lookup("java:eis/BinkpConnectionFactory"));
Connection conn = cf.getConnection();
return conn.connect(host, port);
}
А причем тут Фидо?
А почти непричем. Дело в том, что изначальная задача была вовсе не о binkp, но она была рабочей, а значит попадала под NDA. Поэтому, разобравшись с JCA и решив что нужно написать статью на Хабре ( кстати, оглядываясь назад я начинаю понимать, почему никто такую статью еще не написал. И это еще без транзакций! ), я разморозил старую идею — форк jnode для JEE-серверов, для запуска ноды в виде одного ear. В свое время именно знаний JCA мне не хватило для того, чтобы запустить проект :-)
Под это все я написал вышеприведенные примеры, и они даже заработали. Так что если вы хотите потренироваться в java ee вообще и рефакторинге из java se в частности — пишите письма и коммитьте код. Да, пойнтов все еще беру.
Спасибо за внимание, оставайтесь с нами. Опечатки можно писать в комментариях, я даже не сомневаюсь что их тут десятки.
