WSO2 MI: Envío y Recepción con ActiveMQ

Buenas, hoy veremos un ejemplo completo de como interacturar desde WSO2 con una cola ActiveMQ. Mostraremos como configurar el Micro Integrator, MI, escribir y leer de la cola y consultar cuantos registros tiene. 

Para el ejemplo utilizaremos las siguientes versiones:

• Micro Integrator 4.0.0
• Active MQ 5.8.14. 

    1. Docker Compose

Por un lado debemos incluir las librerías y dependencias de ActiveMQ en el docker compose base que tenemos:

• activemq-broker, activemq-client y activemq-openwire-legacy 5.14.0
• geronimo-j2ee-management_1_1_spec-1.0.1 y geronimo-jms_1.1_spec-1.1.1
• hawtbu-1.11

He incluir la instancia de ActiveMQ:

  activemq:
    image: rmohr/activemq:5.14.5
    mem_limit: 1G
    hostname: activemq
    container_name: activemq
    ports:
      - 8161:8161   #UI
      - 61616:61616 #JMS
    networks:
      wso2-net:
        ipv4_address: 172.16.238.20

Si no configuramos nada más, cuando intentemos conectarnos con ActiveMQ, recibiremos el siguiente error:

Unexpected error during sending message out org.apache.axis2.AxisFault: 
The system cannot infer the transport information from the jms:/.....

Por tanto deberemos modificar el fichero de configuración deployment.toml para añadir:

[[transport.jms.sender]]
name = "myQueueSender"
parameter.initial_naming_factory = "org.apache.activemq.jndi.ActiveMQInitialContextFactory"
parameter.provider_url = "tcp://activemq:61616"
parameter.connection_factory_name = "QueueConnectionFactory"
parameter.connection_factory_type = "queue"
parameter.cache_level = "producer"

Y para poder leer de una cola necesitamos configurar un JMS Listener también en el fichero de configuración deployment.toml

[[transport.jms.listener]]
name = "myQueueConnectionFactory"
parameter.initial_naming_factory = "org.apache.activemq.jndi.ActiveMQInitialContextFactory"
parameter.provider_url = "failover:tcp://activemq:61616"
parameter.connection_factory_name = "QueueConnectionFactory"
parameter.connection_factory_type = "queue"
parameter.cache_level = "consumer"

[[transport.jms.listener]]
name = "default"
parameter.initial_naming_factory = "org.apache.activemq.jndi.ActiveMQInitialContextFactory"
parameter.provider_url = "failover:tcp://activemq:61616"
parameter.connection_factory_name = "QueueConnectionFactory"
parameter.connection_factory_type = "queue"

En esta configuración es muy importante indicar la configuración de failover. Puesto que de lo contrario si se cayese el nodo de ActiveMQ, no se podría volver a conectar con el mismo a menos que se reiniciase el MI. 

    2. Envío de mensajes

Ahora procederemos a crear el primer recurso de una API, con el cual enviaremos información al activeMQ. Pero previamente transformaremos el mensaje, simplemente por añadir una pequeña lógica al ejemplo.

<resource methods="POST" uri-template="/activeMQ/">
    <inSequence>
        <payloadFactory media-type="xml">
            <format>
                <m:getQuote xmlns:m="http://services.samples">
                    <m:request>
                        <m:book>
                            <m:isbn>$1</m:isbn>
                            <m:release>$2</m:release>
                        </m:book>
                    </m:request>
                </m:getQuote>
            </format>
            <args>
                <arg expression="$.isbn" evaluator="json"/>
                <arg expression="$.year" evaluator="json"/>
            </args>
        </payloadFactory>
        <property name="OUT_ONLY" value="true"/>
        <call>
        <endpoint>
            <address uri="jms:/myqueue?transport.jms.ConnectionFactoryJNDIName=QueueConnectionFactory

                   &java.naming.provider.url=failover:(tcp://activemq:61616)

                   &java.naming.factory.initial=org.apache.activemq.jndi.ActiveMQInitialContextFactory

                   &transport.jms.DestinationType=queue&redeliveryPolicy.redeliveryDelay=30000

                   &maximumRedeliveries=10" />
        </endpoint>
        </call>
        <respond />
    </inSequence>
    <outSequence/>
    <faultSequence/>
</resource>

En este código hay dos apartados interesantes:

• La configuración de la propiedad OUT_ONLY que nos permite indicarle al MI que no se quede a la espera de una respuesta por parte del ActiveMQ. 
• La configuración de la invocación al ActiveMQ, donde aparte de los valores básicos para la configuración,  podemos incluir la política de reintentos a través de las propiedades de redelivery o aumentar la resiliencia con la configuración de failover, donde podemos incluir varios nodos de ActiveMQ.

Si hubiésemos configurado con valores más específicos, no necesitaríamos un address endpoint con tanto detalle. Solo necesitaríamos indicar la cola y el transport sender a utilizar. 

<address uri="jms:/myqueue?transport.jms.ConnectionFactory=myCustomSender" />

    3. Lectura de mensajes en la cola

Desde la versión 5.8 ActiveMQ incluye una API a través de la cual podemos enviar y consultar mensajes. Además con la ayuda de Jolokia pone a disposición de los usuarios una API de gestión. Esta  API será la que utilicemos para ver cuantos mensajes hay disponibles en la cola.   

La lógica de nuestro recurso será sencillo: 

    - Creamos una cabecera de autorización básica para la invocación de la API de activeMQ.

    - En la llamada a la API indicamos el broker y la cola de la cual deseamos los datos. Los cuales obtenemos desde nuestro propio recurso.

    - Leemos la respuesta y en caso de ser correcta en base al valor del campo status. (a tener en cuenta que siempre devuelve HTTP Status Code a 200 y Content-Type a text/plain). Leemos el campo value y lo devolvemos.  

<resource methods="GET" uri-template="/activeMQ/{broker}/{queue}/count">
    <inSequence>
        <property name="Authorization" expression="fn:concat('Basic ', base64Encode('admin:admin'))" scope="transport"/>  
        <send>
            <endpoint xmlns="http://ws.apache.org/ns/synapse" name="mock_v1_activemqCount_httpEp">
                <http method="GET" uri-template="http://activemq:8161/api/jolokia/read/org.apache.activemq:type=Broker,brokerName={uri.var.broker},destinationType=Queue,destinationName={uri.var.queue}/QueueSize" />
            </endpoint>
        </send>
    </inSequence>
    <outSequence>
        <property name="ContentType" scope="axis2" type="STRING" value="application/json"/>
        <property name="messageType" scope="axis2" type="STRING" value="application/json"/>
        <switch source="json-eval($.status)">
            <case regex="200">
                <payloadFactory media-type="json">
                    <format> { "count": $1 } </format>
                    <args> <arg evaluator="json" expression="json-eval($.value)"/> </args>
                </payloadFactory>
            </case>
            <default>
                <property name="HTTP_SC" value="500" scope="axis2" />
                <payloadFactory media-type="json">
                    <format>{ "msg": "Unexpected error"}</format>
                    <args />
                </payloadFactory>
            </default>
        </switch>
        <respond/>
    </outSequence>
    <faultSequence/>
</resource>

    4. Lectura de mensajes 

Por último crearemos un Proxy Service que nos permitirá la lectura de esa misma cola. En este caso es importante también devolverle una respuesta a la cola, puesto que de lo contrarío reenviaría el mensaje a la cola de Dead Letter Channel, ActiveMQ.DLQ. 

<proxy xmlns="http://ws.apache.org/ns/synapse" name="activemqExample_v1_consumer_pr" transports="jms" statistics="disable" trace="disable" startOnLoad="true">
    <target>
        <inSequence>
            <log level="full">
                <property name="MSG" value="Receiving messages from Queue" />
            </log>
            <property action="set" name="OUT_ONLY" value="true"/>
            <drop />
        </inSequence>
        <faultSequence />
    </target>
    <parameter name="redeliveryPolicy.maximumRedeliveries">1</parameter>
    <parameter name="transport.jms.DestinationType">queue</parameter>
    <parameter name="transport.jms.SessionTransacted">true</parameter>
    <parameter name="transport.jms.Destination">myqueue</parameter>
    <parameter name="redeliveryPolicy.redeliveryDelay">2000</parameter>
    <parameter name="transport.jms.CacheLevel">consumer</parameter>
    <description />
</proxy>

Ahora podremos realizar la siguiente invocación y comprobar como por un lado enviamos el mensaje y lo leemos por el otro. 

curl --request POST 'https://localhost:8253/mock/activeMQ' \
--header 'Content-Type: application/json' \
--data-raw '{
    "author":"Alfred Bester",
    "name":"the stars my destination",
    "isbn":"1-85798-814-0",
    "year": 1956
}'

Con esto hemos visto un completo ejemplo sobre que podemos hacer para leer y escribir en una cola de ActiveMQ con WSO2. Si nos adentramos un poco más en la configuración podremos realizar tareas más complejas. Como siempre, todo el código lo podemos ver aquí.

WSO2 Micro Integrator : Arrancar y Desplegar

Hoy vamos a ver un post básico sobre cómo lanzar y desplegar una aplicación en un WSO2 Micro Integrator, MI. La idea es crear un ejemplo básico sobre el cual podremos construir ejemplos más complejos en un futuros posts. 

El Micro Integrator es la evolución del Enterprise Integrator de WSO2. Es la solución que plantean para acercar el mundo de los Enterprise Service Bus hacía los microservicios. Permitiendo el despliegue de integraciones con contenedores en un tiempo mínimo, una huella de memoria baja y de forma ágil. 

Para el ejemplo utilizaremos:

• Micro Integrator 4.0.0

Lo primero será crear nuestra aplicación que para este ejemplo será un simple mock. Si no tienes mucha experiencia con WSO2 lo mejor es utilizar el Integration Studio. El cual es un eclipse embebido pero con múltiples cualidades:

• Motor gráfico que te ayuda en la creación de integraciones. 
• Templates para realizar determinadas acciones. 
• Guías paso a paso para realizar la configuración de determinados componentes.
• Permite seleccionar los artefactos a desplegar de forma sencilla. 

Mi opinión personal sobre el Integration Studio es que ha mejorado mucho con los años y que si no tienes experiencia te puede ayudar muchísimo. Pero si tienes experiencia y un buen arquetipo Maven, que puedes hacerte tú mismo, con un simple editor de texto puedes conseguir lo mismo de forma más rápida y ágil. 

Es por eso que cuando pongamos código relativo a integraciones WSO2 lo haremos siempre mostrando el código XML y no los gráficos propios del Integration Studio. 

Con el vamos a crear una simple API que tenga dos recursos que nos permitan devolver siempre el mismo resultado:

<?xml version="1.0" encoding="UTF-8"?>
<api context="/book" name="basicExample_v1_book_api" xmlns="http://ws.apache.org/ns/synapse">
    <resource methods="GET">
        <inSequence/>
        <outSequence/>
        <faultSequence/>
    </resource>
    <resource methods="GET" uri-template="/mock/">
        <inSequence>
            <log level="custom">
                <property name="MSG" value="basicExample_v1_book_api - get - /mock/ - init"/>
            </log>
            <payloadFactory media-type="json">
                <format>
		    [{
		        "id" : 1, "author":"Frank Herbert", "name":"Dune"
		    },{
		        "id" : 2, "author":"Alfred Bester", "name":"The starts my destination"
		    },{
		        "id" : 3, "author":"Frederik Pohl","name":"Gateway"
		    }]
		</format>
                <args/>
            </payloadFactory>
            <log level="custom">
                <property name="MSG" value="basicExample_v1_book_api - get - /mock/ - end"/>
            </log>
            <respond/>
        </inSequence>
        <outSequence/>
        <faultSequence/>
    </resource>
    <resource methods="GET" uri-template="/mock/{id}">
        <inSequence>
            <log level="custom">
                <property name="MSG" value="basicExample_v1_book_api - get - /mock/{id} - init"/>
            </log>
            <payloadFactory media-type="json">
                <format>
		    { "id" : $1, "author":"Frank Herbert", "name":"Dune"}
		</format>
                <args>
                    <arg evaluator="xml" expression="$ctx:uri.var.id"/>
                </args>
            </payloadFactory>
            <log level="custom">
                <property name="MSG" value="basicExample_v1_book_api - get - /mock/{id} - end"/>
            </log>
            <respond/>
        </inSequence>
        <outSequence/>
        <faultSequence/>
    </resource>
</api>

Esta API, la incluiremos en un CAR que es el formato de paquete para las integraciones de WSO2. 

A continuación veremos cómo desplegarlo en una imagen docker. Pero para poder bajarnos la imagen del MI, necesitaremos una cuenta de WSO2, gratuita, y logarnos en su docker hub. 

docker login docker.wso2.com -u youraccount@example.com

Esta imagen la incluiremos en un docker-compose básico en el cual podremos incluirle otras dependencias en futuros posts. 

version: '3.7'

networks:
  wso2-net:
    ipam:
      driver: default
      config:
        - subnet: "172.16.238.0/24"

services:
  wso2mi:
    mem_limit: 1G
    image: wso2/wso2mi:4.0.0
    hostname: wso2mi
    container_name: wso2mi
    ports:
      - 8290:8290
      - 8253:8253
      - 9164:9164
    networks:
      wso2-net:
        ipv4_address: 172.16.238.10
    environment:
      JAVA_OPTS: "-DenableManagementApi=true -Duser.timezone=Europe/Madrid"
    volumes:
       - ./conf/deployment.toml:/home/wso2carbon/wso2mi-4.0.0/conf/deployment.toml

Un par de apuntes sobre este fichero:

• Con el parámetro enableManagementApi podemos habilitar la API de gestión.
• El fichero deployment.toml es el fichero de configuración del Micro Integrator. Nosotros para este caso no hemos hecho ningún cambio sobre el fichero de por defecto, pero es normal modificarlo y por eso lo hemos añadido en un volumen. 

Ahora con este docker-compose veremos distintas formas de poder desplegar el CAR. 

1. Configurar un volumen donde se encuentren el desplegable

Nos permite indicar en el momento del arranque las integraciones que queremos desplegar. 

    volumes:
       - ./conf/deployment.toml:/home/wso2carbon/wso2mi-4.0.0/conf/deployment.toml
       - ./apps/wso2-basic-example_1.0.0.car:/home/wso2carbon/wso2mi-4.0.0/repository/deployment/server/carbonapps/wso2-basic-example_1.0.0.car

2. Arrancar el MI y pasarle el desplegable

Similar al paso anterior, pero nos permite desplegar aplicaciones con el MI ya en marcha. Esto es especialmente util cuando estamos desarrollando. 

docker cp wso2-basic-example_1.0.0.car wso2mi:/home/wso2carbon/wso2mi-4.0.0/repository/deployment/server/carbonapps/

3. Crear un Dockerfile con el CAR. 

Esta opción será la ideal cuando tengamos nuestra integración estable y la querremos desplegar en entornos productivos. 

    Primero construimos el Dockerfile

FROM wso2/wso2mi:4.0.0
COPY wso2-basic-example_1.0.0.car /home/wso2carbon/wso2mi-4.0.0/repository/deployment/server/carbonapps/

    Luego construimos la imagen

docker build -t deesebc/book-api-mi .

    Y por último arrancamos un contenedor

docker run -it -p 8290:8290 -p 8253:8253 -p 9164:9164 --name micro-integrator deesebc/book-api-mi

4. API Management

A pesar de estar indicado en la documentación como una posibilidad, a 03/2022 es solo posible como actualización de pago del MI. 

Una vez que lo hemos arrancado ya podremos consultar nuestra API

curl --location --request GET 'http://localhost:8290/book/mock'
curl --location --request GET 'https://localhost:8253/book/mock'

El otro puerto habilitado, 9164, será el que nos da acceso a través de HTTPS a la API de gestión. 

Espero que haya sido de utilidad, y como siempre si queréis el código completo podéis verlo aquí.