NodeJs-Java-Pub-Sub.md

How to set up a system system using Redis as a pub/sub queue to push data between a node.js application and a java application running on the same system.

To Configure this, we will need jedis, a java driver for jedis and node-redis a node driver.

1. create a new node project
2. Install the node driver using npm install node_redis
3. Create the node app file with the following contents.

Node.js app.js

var redis = require('redis'), //npm install node_redis
subscriber = redis.createClient(),
publisher = redis.createClient();

/*
subscriber.on("message", function (channel, message) {
	console.log("Message '" + message + "' on channel '" + channel + "' arrived!")
});
*/
subscriber.subscribe("commonChannel");

publisher.publish("commonChannel", "haaaaai");
publisher.publish("commonChannel", "kthxbai");

We can then start the nodejs app using node app.js or nodemon if you prefer that.

Create a java project using the code below. What is important is the MainActivity for starting the application and the Subscriber.java class which listens for messages from redis.

MainActivity.java

import redis.clients.jedis.Jedis
import redis.clients.jedis.JedisPool
import redis.clients.jedis.JedisPoolConfig

/**
 * Created: 2/13/14 12:49 PM
 */
class MainActivity {
    public static final String CHANNEL_NAME = "commonChannel";

    //private static Logger logger = Logger.getLogger(MainActivity.class);

    public static void main(String[] args) throws Exception {

        JedisPoolConfig poolConfig = new JedisPoolConfig();

        JedisPool jedisPool = new JedisPool(poolConfig, "localhost", 6379, 0);

        final Jedis subscriberJedis = jedisPool.getResource();

        final Subscriber subscriber = new Subscriber()
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    println("Subscribing to \"commonChannel\". This thread will be blocked.");
                    subscriberJedis.subscribe(subscriber, CHANNEL_NAME);
                    println("Subscription ended.");
                } catch (Exception e) {
                    //logger.error("Subscribing failed.", e);
                }
            }
        }).start();

        Jedis publisherJedis = jedisPool.getResource();

        new Publisher(publisherJedis, CHANNEL_NAME).start();

        subscriber.unsubscribe();
        jedisPool.returnResource(subscriberJedis);
        jedisPool.returnResource(publisherJedis);
    }
}

In the Subscriber.java class we can parse the message received in the onMessage method, and proceed to do all the work needed to be done.

Subscriber.java

import redis.clients.jedis.JedisPubSub

/**
 * User: u0165547
 * Created: 2/13/14 12:49 PM
 */
class Subscriber  extends JedisPubSub{
   // private static Logger logger = Logger.getLogger(Subscriber.class);

    @Override
    public void onMessage(String channel, String message) {
        println "Message received. Channel: ${channel}, Msg: ${message}";
    }

    @Override
    public void onPMessage(String pattern, String channel, String message) {

    }

    @Override
    public void onSubscribe(String channel, int subscribedChannels) {
          println "onSubcribe ${channel}"
    }

    @Override
    public void onUnsubscribe(String channel, int subscribedChannels) {
        println "onUnSubcribe ${channel}"
    }

    @Override
    public void onPUnsubscribe(String pattern, int subscribedChannels) {

    }

    @Override
    public void onPSubscribe(String pattern, int subscribedChannels) {

    }
}

Publisher.java

package com.richard

import redis.clients.jedis.Jedis

/**
 * User: u0165547
 * Created: 2/13/14 12:49 PM
 */
class Publisher {
    //private static final Logger logger = Logger.getLogger(Publisher.class);

    private final Jedis publisherJedis;

    private final String channel;

    public Publisher(Jedis publisherJedis, String channel) {
        this.publisherJedis = publisherJedis;
        this.channel = channel;
    }

    public void start() {
        println "Type your message (quit for terminate)"
        try {
            BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));

            while (true) {
                String line = reader.readLine();

                if (!"quit".equals(line)) {
                    publisherJedis.publish(channel, line);
                } else {
                    break;
                }
            }

        } catch (IOException e) {
            //logger.error("IO failure while reading input, e");
        }
    }
}

All the java/groovy class can be managed by the following simple gradle build file. gradle build file

apply plugin: 'java'
apply plugin: 'idea'
apply plugin: 'groovy'

sourceCompatibility = 1.7

repositories {
    mavenCentral()
}

version = '1.0'
jar {
    manifest {
        attributes 'Implementation-Title': 'Gradle Quickstart', 'Implementation-Version': version
    }
}

dependencies {
    compile 'redis.clients:jedis:2.2.1'
    compile 'org.codehaus.groovy:groovy-all:2.2.1'
    //compile 'log4j:log4j:1.2.17'
}