jruby processor with activemq-5.5.0

Activemq is already in bundle with camel. The camel config file is in $ACTIVEMQ_HOME/conf/camel.xml.

But you need something more to use a jruby processor with activemq:
- add lang namespace to camel.xml;
- add a aopalliance.jar and jruby.jar to classpath (and more, if you need logback);
- write and register jruby bean.

Add namespace to camel.xml: just replace header with:

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

Add a aopalliance.jar to classpath: download springsource-2.5 with all dependencies. Unzip it and get aopalliance.jar from the lib/aopalliance directory.

Activemq $CLASSPATH is defined as $ACTIVEMQ_HOME/conf, so create a ruby folder in conf and add a jruby bean named ruby_processor.rb:

java_import "org.apache.camel.Exchange"
java_import "org.apache.camel.Processor"
java_import "org.slf4j.Logger"
java_import "org.slf4j.LoggerFactory"
java_import "ch.qos.logback.classic.LoggerContext"

java_package "com.test.camel"

class RubyProcessor
  include Processor

  def initialize
    @logger = LoggerFactory.getLogger("com.test.camel.#{File.basename(__FILE__, ".rb")}");
  end  

  java_signature "void process(Exchange exchange)"
  def process(exchange)
    test = exchange.in.headers['test']
    @logger.info("RubyProcessor: header test is: #{test}")
    exchange.in.set_header("Issuer", "rubyProcessor")
  end
end

Please note this bean does very few: just dump the value of a header to a log file, then sets a new header. It is just e proof-of-concept bean.

In camel.xml add a test route like:

<from uri="activemq:example.A"/>
     <bean ref="rubyProcessor" />
            <to uri="activemq:example.B" />

 <lang:jruby id="rubyProcessor"
              script-interfaces="org.apache.camel.Processor"
              script-source="classpath:ruby/ruby_processor.rb">
  </lang:jruby>

The ticky part.

To have logback work first you need to include in lib folder logback-classic and logback-core then I had to upgrade slf4j to 1.6.2 which means copying a couple of jars both in lib and in lib/optional.

The logback.xml can be created in conf, logfile from logback are written in $ACTIVEMQ_HOME.

Than I noticed the ruby scripting fails to work in the routes. I had to replace the <ruby> tags with <simple>.

Send a mail with ruby1.8, ruby1.9 and jruby

I lost the best part of this morning to send a mail with ruby with the constrain it should work with ruby1.8, ruby1.9 and jruby.

This is my best effort:

#!/usr/bin/env ruby
#ENCODING: UTF-8

require 'rubygems'
require 'net/smtp'
require 'mail'

TO = 'you@test.com'
FROM = 'me@test.com'

mail = Mail.new do
  from    FROM
  to      TO
  subject 'This is a test email'
  body    "Test\n\nNow is the time #{Time.now}\n"
end

Net::SMTP.start('localhost') do |smtp|
  smtp.send_message(mail.to_s, TO, FROM)
end

I hope it could save someone's time!

Monitor activemq queue-size with jruby jmx

The aim is to know when a certain threshold of queued messages is reached on certain destinations. For example, if new messages show up in DLQ (dead letter queue) it might be a hint of some problems.

First of all enable jmx for activemq on a free port (11099, for example). For activemq-5.5.0 modify /etc/default/activemq as follows:

ACTIVEMQ_SUNJMX_START="-Dcom.sun.management.jmxremote.port=11099 
ACTIVEMQ_SUNJMX_START="$ACTIVEMQ_SUNJMX_START -Dcom.sun.management.jmxremote.authenticate=false"
ACTIVEMQ_SUNJMX_START="$ACTIVEMQ_SUNJMX_START -Dcom.sun.management.jmxremote.ssl=false"
ACTIVEMQ_SUNJMX_START="$ACTIVEMQ_SUNJMX_START -Dcom.sun.management.jmxremote"

Restart activemq and confirm something is listening on port 11099:

$ sudo lsof -i :11099
COMMAND   PID     USER   FD   TYPE    DEVICE SIZE NODE NAME
java    23206 activemq   10u  IPv6 110495619       TCP *:11099 (LISTEN)

Install jmx gem:

sudo jruby -S gem install jmx

Then a simple script like this can query the size of the queue ActiveMQ.DLQ for further actions:

require 'java'
require 'rubygems'
require 'jmx'

client = JMX.connect(:port => 11099)

queue = client["org.apache.activemq:BrokerName=bacedifo,Type=Queue,Destination=ActiveMQ.DLQ"]
#puts queue.attributes

puts queue["QueueSize"]