integrating mq protocols with wso2 esb 4.9.0 (rabbitmq, mqtt, kafka)
TRANSCRIPT
Integrating MQ Protocols WSO2 Enterprise Service Bus 4.9.0
Jagath AriyarathneTechnical Lead
Kevin RathnasekaraSoftware Engineer
Agenda
o Enterprise Messaging
o Messaging support in WSO2 ESB
o What’s new in ESB 4.9.0 with MQ protocols
o RabbitMQ
o Kafka
o MQTT
o Demo on MQTT Integration
Enterprise Messaging
o What is messaging?Communication between two applications with high-speed, asynchronousand reliable delivery
o Why we need messagingo Remote communicationo Platform/language integrationo Asynchronous communicationo Reliable communication
Enterprise Messaging cont...
o Achieved by message queues
o Enterprise messaging middleware systemso ActiveMQo IBM WebSphereo RabbitMQo JBoss Messaging
Producer (Sender)
Message Queue
Consumer(Receiver)
Enterprise Messaging with WSO2 ESB
o RabbitMQ, Kafka and MQTT transports
RabbitMQ with WSO2 ESB
o In-built RabbitMQ transport, easy to configure
o Automatic connection recovery
o All content-types are now supported
o Synchronized request-response support
o Introduced the new feature rich inbound endpoint for RabbitMQ
o Added the SSL support for RabbitMQ inbound/transport
RabbitMQ with WSO2 ESB cont...
o RabbitMQ as a transport in ESB 4.9.0
RabbitMQ with WSO2 ESB cont...
o Enable RabbitMQ Transport (axis2.xml)o Transport Receiver
<transportReceiver name="rabbitmq" class="org.apache.axis2.transport.rabbitmq.RabbitMQListener">
<parameter name="AMQPConnectionFactory" locked="false">
<parameter name="rabbitmq.server.host.name" locked="false">192.168.0.3</parameter>
<parameter name="rabbitmq.server.port" locked="false">5672</parameter>
<parameter name="rabbitmq.server.user.name" locked="false">user</parameter>
<parameter name="rabbitmq.server.password" locked="false">abc123</parameter>
</parameter>
</transportReceiver>
o Transport Sender
<transportSender name="rabbitmq" class="org.apache.axis2.transport.rabbitmq.RabbitMQSender"/>
RabbitMQ with WSO2 ESB cont...
o Use RabbitMQ transport in a Proxy<proxy xmlns="http://ws.apache.org/ns/synapse" name="AMQPProxy" transports="rabbitmq" statistics="disable" trace="disable"
startOnLoad="true">
<target>
<inSequence>
<log level="full"/>
<property name="OUT_ONLY" value="true"/>
<property name="FORCE_SC_ACCEPTED" value="true" scope="axis2"/>
</inSequence>
<endpoint>
<address
uri="rabbitmq:/AMQPProxy?rabbitmq.server.host.name=192.168.0.3&rabbitmq.server.port=5672&rabbitmq.server.user.
name=user&rabbitmq.server.password=abc123&rabbitmq.queue.name=queue2&rabbitmq.exchange.name=exchange2"/>
</endpoint>
</target>
<parameter name="rabbitmq.queue.name">queue1</parameter>
<parameter name="rabbitmq.exchange.name">exchange1</parameter>
<parameter name="rabbitmq.connection.factory">AMQPConnectionFactory</parameter>
<description></description>
</proxy>
RabbitMQ with WSO2 ESB cont...
o Sample configuration for RabbitMQ Inbound<inboundEndpoint xmlns="http://ws.apache.org/ns/synapse" name="RabbitMQConsumer" sequence="amqpSeq" onError="amqpErrorSeq" protocol="
rabbitmq" suspend="false">
<parameters>
<parameter name="sequential">true</parameter>
<parameter name="coordination">true</parameter>
<parameter name="rabbitmq.connection.factory">AMQPConnectionFactory</parameter>
<parameter name="rabbitmq.server.host.name">localhost</parameter>
<parameter name="rabbitmq.server.port">5672</parameter>
<parameter name="rabbitmq.server.user.name">guest</parameter>
<parameter name="rabbitmq.server.password">guest</parameter>
<parameter name="rabbitmq.queue.name">queue</parameter>
<parameter name="rabbitmq.exchange.name">exchange</parameter>
<parameter name="rabbitmq.connection.ssl.enabled">false</parameter>
</parameters>
</inboundEndpoint>
RabbitMQ with WSO2 ESB cont...
Kafka with WSO2 ESB
o Kafka inbound endpoint to connect to inbound connections<inboundEndpoint xmlns="http://ws.apache.org/ns/synapse"
name="KafkaListenerEP"
sequence="requestHandlerSeq"
onError="inFaulte"
protocol="kafka"
suspend="false">
<parameters>
<parameter name="interval">100</parameter>
<parameter name="coordination">true</parameter>
<parameter name="sequential">true</parameter>
<parameter name="zookeeper.connect">localhost:2181</parameter>
<parameter name="consumer.type">highlevel</parameter>
<parameter name="content.type">application/xml</parameter>
<parameter name="topics">test,sampletest</parameter>
<parameter name="group.id">test-group</parameter>
</parameters>
</inboundEndpoint>
Kafka with WSO2 ESB cont...
o Kafka connector for outbound communication<proxy xmlns="http://ws.apache.org/ns/synapse"
name="KafkaTransport"
transports="https http"
startOnLoad="true"
trace="disable">
<description/>
<target>
<inSequence>
<kafkaTransport.init>
<brokerList>localhost:9092</brokerList>
</kafkaTransport.init>
<kafkaTransport.publishMessages>
<topic>test</topic>
</kafkaTransport.publishMessages>
</inSequence>
</target>
</proxy>
Kafka with WSO2 ESB cont...
MQTT with WSO2 ESB
o Improved MQTT transport(axis2 transport)
o transport sender
o transport listener
o MQTT inbound endpoint for more versatile integrations
MQTT with WSO2 ESB cont...
MQTT with WSO2 ESB
o Inbound communication with Axis2 transport<transportReceiver name="mqtt" class="org.apache.axis2.transport.mqtt.MqttListener">
<parameter locked="false" name="mqttConFactory">
<parameter locked="false" name="mqtt.server.host.name">localhost</parameter>
<parameter name="mqtt.connection.factory">mqttConFactory</parameter>
<parameter locked="false" name="mqtt.server.port">1883</parameter>
<parameter locked="false" name="mqtt.client.id">client-id-1234</parameter>
<parameter locked="false" name="mqtt.topic.name">esb.test</parameter>
</parameter>
</transportReceiver>
<proxy xmlns="http://ws.apache.org/ns/synapse"
name="SampleProxy"
transports="mqtt"
startOnLoad="true"
trace="disable">
MQTT with WSO2 ESB cont...
Outbound communication with Axis2 transport
o Axis2.xml configuration
<transportSender name="mqtt" class="org.apache.axis2.transport.mqtt.MqttSender"/>
o Sample Sequence
<sequence name="dispatchSeq" onError="fault">
<log level="full"/>
<send>
<endpoint>
<address uri="mqtt:/sender?mqtt.server.host.name=localhost&mqtt.server.port=1884&mqtt.client.id=mqtt.sender.id&mqtt.topic.name=esb.publish.topic&mqtt.subscription.qos=2&mqtt.blocking.sender=true"/>
</endpoint>
</send>
<drop/>
</sequence>
Demo
A
B
References
o WSO2 Documentationo https://docs.wso2.com/display/ESB490/RabbitMQ+AMQP+Transport
o https://docs.wso2.com/display/ESB490/Kafka+Inbound+Protocol
o https://docs.wso2.com/display/ESB490/MQTT+Transport
o Blogso http://blog.maheeka.me/2015/09/esb-490-enhanced-rabbitmq-support.html
o http://rkathees.blogspot.com/2015/09/wso2-esb-490-kafka-support.html
o http://elilsivanesan.blogspot.com/2015/04/mqtt-v31-introduction.html
o Other○ http://www.eclipse.org/paho/files/javadoc/index.html
