Asynchronous Streaming Request Processing in Spring MVC 4.2 + Spring Boot

With the release of Spring 4.2 version, Three new classes have been introduced to handle Requests Asynchronously of the Servlet Thread. Which are;

1. ResponseBodyEmitter
2. SseEmitter
3. StreamingResponseBody

ResponseBodyEmitter enables to send DeferredResult with a compatible HttpMessageConverter. Results can be emitted from threads which are not necessarily the Servlet Request Thread of the Servlet Container.

SseEmitter is used to send Server Sent Events to the Client. Server Sent Events has a fixed format and the response type for the result will be text/event-stream.

StreamingResponseBody is used to send raw unformatted data such as bytes to the client asynchronously of the Servlet Thread.

ResponseBodyEmitter and SseEmitter has a method named complete to mark its completion and StreamingResponseBody will complete when there is no more data to send. 

All three options will be keeping alive a connection to the endpoint until the end of the request.

StreamingResponseBody is particularly useful for streaming large files such as Media Files as writing of the bytes to the Response's OutputStream will be done asynchronously. StreamingResponseBody has a writeTo(OutputStream os) call back method which needs to be overridden inorder to support streaming.

I wrote a small Spring Boot Application to showcase the StreamingResponseBody capabilities in terms of Streaming large files. The application source code can be found at www.github.com/shazin/itube. Below is a screen shot of the application.

In order to send the Video files streaming to the Projekktor player in the web page following code snippet is used.

And a Custom Web Configuration to over ride default timeout behavior to no timeout and finally configuring an AsyncTaskExecutor

References

1. Spring MVC 4.2: New and Noteworthy - Rossen Stoyanchev
2. Projekktor Player
3. Videos Downloaded From Videezy

The New Buzz Word in Information Technology Space - Microservices

Microservices, a word which is buzzing around information technology space almost everyday. What is this means? How did we get here?

Well to start off we need to go back to good old days to see how enterprise applications were developed. Most of the application developed at that time where Monolithic. Monolithic applications are collection of functionalities grouped together which will eventually be compiled and packaged into one single application. In Java terms a Jar/War/Ear. In a monolithic application if an application had a Web Tier, User Registration and Account Registration functionalities. All these will be packaged and deployed as a single unit. Let's see the pros and cons of this approach.

Pros of Monolithic Applications

1. Easy steps to build (can be automated).
2. Ease of monitoring.
3. Centralized code repository.
4. Code is monoglot.

Cons of Monolithic Applications

1. As the Application becomes complex and large building can take significantly more time.
2. During deployment of newer versions there are down times for the application.
3. Less Flexible for change.
4. Difficulty in scaling out. 

These cons are blockers for High Availability/Low Latency, Rapidly Evolving applications. In order to address these Software Architects tried Modular/Component programming where each functionality is developed as a separate module/component. This in fact managed to reduce the build time drastically. But the applications were still monoliths. Deploying a newer version of a module/component meant that the whole application had to stop and restarted as they are run in one process.

Enter Microservices, According to Martin Fowler microservices means;

The term "Microservice Architecture" has sprung up over the last few years to describe a particular way of designing software applications as suites of independently deployable services. While there is no precise definition of this architectural style, there are certain common characteristics around organization around business capability, automated deployment, intelligence in the endpoints, and decentralized control of languages and data.

His illustrations clearly distinguishes a Monolithic application from a Microservices architecture.

Having a Microservices based architecture means that your applications can;

• Evolve rapidly and independently.
• Deploy and test a service easily.
• Scale out or in for demand.
• Be polyglot (Each Microservice can be developed using separate Programming Language/Database).
• Be developed by Small, Independent, Easy to Manage Teams

Lets have a look at Hypothetical Microservices Application. The company MS have business cases for User Registration and Account Registration. User registration has sub use cases (Persisting User, Sending Email Account Verification, Verifying Email Account) of their own which are complex and time consuming but must be flexible for change. Account Registration means maintaining the financial aspect of the User. Furthermore there is a REST API web tier open to users and partners. The following microservices architecture can be used to implement MS company application.

For implementation we can use Spring Boot as it allows to create both Web Container based as well as Stand Alone Production Ready applications using minimal coding. There are more alternatives if you are not a fan of Spring at https://github.com/mfornos/awesome-microservices

When Rabbit MQ is installed successfully, this microservices based application is ready to run. The applications are so independent there is no order to start the microservices. Just running the Main methods of all three projects would spin up the required service.

When a Create New User request is sent to the Web REST API it is enqueued in ms-user queue which will be dequeued and processed by User Registration microservice, which after finishing processing will enqueue another message in ms-account queue. Those messages will be handled by the Account Registration microservice. 

Request to REST API

Message Dequeued from Rabbit MQ queue ms-user

Message Dequeued from Rabbit MQ queue ms-account

All the microservices run on their own process and can be modified, scaled out and/or in transparently. The implementation source code can be found at https://github.com/shazin/ms

References

• Microservices by Martin Fowler - http://martinfowler.com/articles/microservices.html
• Spring Boot - http://projects.spring.io/spring-boot/
• Rabbit MQ - https://www.rabbitmq.com/

RxJava and Spring Boot

It has been a really long time (more than six months) since I wrote my last blog post. Life was a bit busy with coming down from Malaysia and finding a new job in Sri Lanka and settling down. It doesn't mean I was not working with new technologies. Got the chance to work with alot of new technologies and frameworks.

Ok having said that while going through the usual weekly dose of InfoQ videos I was bumped on Simon Baslé and Laurent Doguin's talk on Introducing RxJava into a Spring Boot REST API. 

I have blogged about Reactive programming previously in this post. Which is all about writing non blocking, concurrent, functional code. Two main aspects of Reactive manifesto is;

1. Reactive Streams (Observable in RxJava)
2. Reactive Extensions (Observer/Subscriber in RxJava)

But this time I wanted to go beyond and learn the cores of Reactive Programming. I build a small application which has a Service which returns an Observable (Non blocking) which can be used to asynchronously perform many tasks such filter, zip, map etc.

This Service can be used in many ways to produces many results as we need and I have used findAll() method in findByName() method to filter records which are only matching the given name. Likewise this gives many more options.

And in the Controller this Service can be used to retrieve the data we want and finally collect those as a List or any other way and pass it back.


Spring a return type named DeferredResult which will send a response asynchronously when it is finished processing. Either by returning a DeferredResult from the controller method or by writing a custom ReturnValueHandler for Observable we can send asynchronous results.

The complete source code can be found at https://github.com/shazin/reactiveapp and I strongly advice to go through the presentation in the references section.

References

1. Introducing RxJava into a Spring Boot REST API - http://www.infoq.com/presentations/rxjava-spring-boot
2. Introduction to Reactive Programming - http://www.infoq.com/presentations/rxjava-reactor

Trackbacks/Pings

1. https://www.techtalks.lk/blog/rxjava-spring-boot

Simple Twitter Sentiment Notification System with ESP8266 Wifi + Arduino

I am a huge fan of Cricket and die hard supporter of Sri Lankan Cricket Team. The World Cup was concluded recently with Australia emerging as champions. New Zealand were runner up and had a great tournament even though they lost the final. With cricket comes a lot of social media interaction. Trolls, Complements, Griefs, Celebrations, you name it all happens in social media platforms. 

Out of this twitter stands out because in twitter you can directly interact with players involved and large audience of cricket followers. I wanted to know how twitter users are feeling and talking about cricket matches during the world cup so I developed a small, simple twitter sentiment notification system with Arduino and a recently bought ESP8266 Wifi.

Requirement

The requirement was to just illuminate a red or green light (out, not out in cricket :) ) based on the sentiment score of a particular tweet instead of reading the whole tweet. The idea was to keep everything as simple as possible because this is just a small project with limited usage.

Implementation Decision

For this I used Spring XD which is built for Extreme Data Processing by creating an engine to source , transform and sink from multiple data points. Furthermore I used a Redis list act as the sink from Spring XD and source for the arduino via ESP8266 over wifi. 

The sentiment analysis is done by a custom written Processor module for Spring XD which just does negative word count and determines whether the tweet is negative or positive, simple but works for this simple project.

A small java TCP server act as the intermediate between arduino via ESP8266 over wifi and Redis

Architecture

Implementation

Hardware

Software

The source codes are available at https://github.com/shazin/tweetsentiments

Spring XD 1.1.0 was used with Redis 2.8.7, for this to work Spring XD must be started in Single Node mode with the following command

SPRING_XD_HOME/xd/bin/xd-singlenode

Redis server must be started with the following command

REDIS_HOME/src/redis-server

A shell must be opened to Spring XD to execute commands

SPRING_XD_HOME/xd/bin/xd-shell

in the shell following commands must be executed

This registers the custom sentiment processor module in Spring XD 

module upload --file /Path/To/sent-processor-0.0.1-SNAPSHOT.jar --name sent-processor --type processor

This creates the stream which sources tweets from twitter, analysis and calculates their sentiment score and sinks in redis. twittersearch --query could be anything you want to search. In this case I have added spring to focus on the word spring.

stream create --name tweetsentiments --definition "twittersearch --query=Spring --consumerKey= --consumerSecret= | sent-processor | redis --queue=sentiments" --deploy

Finally the TCP server must be started and arduino sketch ESPWifi must be loaded into arduino.

java -cp jedis-2.6.2.jar:./ Server

Demonstration

References

• ESP8266 Arduino Library - https://github.com/itead/ITEADLIB_Arduino_WeeESP8266
• Spring XD - http://docs.spring.io/spring-xd/docs/current-SNAPSHOT/reference/html/
• Redis - http://redis.io/

Observable in Reactive Programming, Hybrid of Publisher/Subscriber Pattern and Iterator Pattern

As Programmers we are advised to use Design Patterns as much as possible. But making use of Design Patterns doesn't guarantee that our code becomes Flexible, Maintainable and Bug Free. But it is certainly a step towards achieving those goals.

When I was in University I was told by lecturers to read the book Design Patterns: Elements of Reusable Object-Oriented Software by authors Erich Gamma, Richard Helm, Ralph Johnson and John Vlissides famously known as the Gang of Four. The authors went a long way to find the Design Patterns that are used by Industry professionals and managed to document all those in detail in a nice and understandable book.

In that book under Behavioral Patterns there were two patterns which could be taught as most widely used patterns which are Publisher/Subscriber pattern and Iterator pattern.

In brief Publisher/Subscriber pattern enables to decouple data production and consumption clearly by having Producers notify Subscribed consumers as and when data is available. And Iterator pattern enables to loop through a collection transparently without knowing underlying implementation. 

At first the two patterns looks totally different but as Jafar Husain points out in the presentation video in the references section, any problem that can be solved by Publisher/Subscriber pattern can also be solved by Iterator pattern and vice-versa. 

Yet both were distinguishable from each other because with Publisher/Subscriber pattern there is no way of finding whether the Producer is finished producing data and with Iterator pattern one could do it. 

Reactive Programming is a topic which is catching up in the industry and fueled by Netflix is actually an hybrid of both Publisher/Subscriber pattern and Iterator pattern. This model enables separation of Publisher/Subscriber logic along with Iterator pattern's feature of being notified of completion and exceptions. RxJava provides this model programming to be performed in Java language and it enables Declarative style of programming while producing concise code. Below is an example which shows both Declarative/Functional (RxJava based implementation) and Imperative (Core Java based implementation) doing the same task. 

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.security.MessageDigest;
import java.util.Arrays;
import java.util.List;

import rx.Observable;
import rx.Scheduler;
import rx.schedulers.Schedulers;

/**
 * RxJava Observable vs Imperative Java Code
 * 
 * @author shazin
 *
 */
public class Main {

    public static void main(String[] args) {
        List<String> urls = Arrays.asList("http://www.google.com", "http://www.lkbotics.com", "http://www.yahoo.com");
        
        System.out.println("============================= Declarative/Functional ==========================================");
        
        Observable.from(urls)
                //.subscribeOn(Schedulers.io()) // Uncommenting this line will enable concurrent execution
                .flatMap(url -> Observable.just(getHtml(url)))
                .filter(html -> html != null && html.toString().length() > 0)
                .doOnNext(html -> saveHtml(html))
                .map(html -> getTitle(html))
                .doOnError(t -> System.out.println("Error : "+t))
                .doOnCompleted(() -> {System.out.println("Completed!");})
                .subscribe(title -> System.out.println(title));
        
        System.out.println("=================================== Imperative ================================================");
        
        for(String url:urls) {
            try {
                String html = getHtml(url);
                if(html != null && html.length() > 0) { 
                    saveHtml(html);
                    System.out.println(getTitle(html));
                }
            } catch(Exception e) {
                System.out.println("Error : "+e);
                break;
            }
        }
        System.out.println("Completed!");
    }

    public static String getHtml(String url) {
        System.out.printf("Retrieving Html for url : %s\n", url);
        StringBuilder out = new StringBuilder();
        HttpURLConnection httpURLConnection = null;
        try {
            httpURLConnection = (HttpURLConnection) new URL(
                    url).openConnection();
            try (BufferedReader br = new BufferedReader(new InputStreamReader(
                    httpURLConnection.getInputStream()))) {

                String line = null;
                while ((line = br.readLine()) != null) {
                    out.append(line).append("\n");
                }
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            if(httpURLConnection != null) {
                httpURLConnection.disconnect();
            }
        }

        return out.toString();
    }
    
    public static String getTitle(String html) {
        String content = html.toLowerCase();
        return html.substring(content.indexOf("<title>") + 7, content.indexOf("</title>"));
    }
    
    public static void saveHtml(String html) {
        try {
            MessageDigest digest = MessageDigest.getInstance("MD5");
            byte[] data = digest.digest(html.getBytes("UTF-8"));
            
            System.out.printf("Saving html for %s : %s\n", getTitle(html), hex(data));
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    
    public static String hex(byte[] data) {
        StringBuilder hex = new StringBuilder();
        for(byte d:data) {
            hex.append(String.format("%x", d & 0xFF));
        }
        return hex.toString();
    }

}

The real power and flexibility of Reactive Programming against Imperative Programming is its ability scale and produce low response time. Just for example to run the tasks concurrently, un-commenting the //.subscribeOn(Schedulers.io()) line would suffice whereas running the imperative tasks concurrently would require a lot more  careful coding.

There are plenty of more benefits of Reactive Programming and this post doesn't cover all of them, Please go through the References section to learn more.

References

• Asynchronous Programming at Netflix - Jafar Husain
• ReactiveX - http://reactivex.io/intro.html
• RxJava Tutorial - http://blog.danlew.net/2014/09/15/grokking-rxjava-part-1/

Anti Pattern - ThreadLocal variables with Thread Pool

In a previous post I wrote the usage and benefits of ThreadLocal based instance variables in concurrent applications. This seemingly innocent and fail proof implementation will provide clear data separation and visibility between threads in multi threaded applications UNTIL, you use Thread Pooling.

ThreadLocal variables as the name suggest is local to the thread, til the thread is alive the ThreadLocal instance variable can not be Garbage Collected. This post explains the theory clearly.

To see it in action I wrote small program which infinitely creates Tasks in one implementation submitting tasks to a Thread Pool and in another instance creating ad-hoc Threads which are no pooled. And I used jVisualVM to monitor the VM.

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


public class ThreadLocalTest {

    private static ExecutorService executor = Executors.newFixedThreadPool(100);
    
    public static void main(String[] args) {
        while(true) {
            // Thread Pooled based implementation
            executor.execute(new SimpleThread());
            // Ad-hoc Thread based implementation
            new SimpleThread().start();
        }
    }
    
    public static class SimpleThread extends Thread {
        
        private ThreadLocal<Integer> no = new ThreadLocal<Integer>() {

            @Override
            protected Integer initialValue() {
                return 0;
            }
            
        };
        
        public void run() {
            while(no.get() < 100) {
                no.set(no.get() + 1);
            }
            System.out.println(String.format("Thread : %s Finished", Thread.currentThread().getName()));
            no.remove();
        }
    }
}

In the Thread pooled implementation one thread throwing an Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "Thread Name" is just a matter of time.

Where as the adhoc Thread creation seems to keep on running with good VM Garbage Collection but the Thread pool implementation fills up the heap space within minutes and eventually runs out of memory. 

Lesson

Always keep an eye on the Thread classes you use in your application, make sure they don't have ThreadLocal variables when you are using Thread Pooling.

Give special care when you get pre-compiled third party libraries which might have ThreadLocal so always read javadocs thoroughly when you use Thread pooling.

IoT with MQTT + Apache Kafka (Arduino + Raspberry Pi)

Motivation

Internet of Things always fascinated me because of the sheer no of people talking about it and the no of projects coming up related to it. So I thought I should give it a go myself and put up a small demonstration to learn the concepts of IoT projects.

This project is about implementing a IoT system to measure illumination inside of a house and act upon when there is low illumination by illuminating a light.

Design

Architecture

Arduino + Photoresistor Schematic

Raspberry Pi + LED Schematic

Implementation Decisions

I could have easily connected the Arduino and Raspberry Pi directly and get the thing done but I wanted to learn the correct and hard way to do as that is the whole purpose of this project. After some research on the internet I found that the best way to connect CPU and Memory constraint devices to IoT system is by using a protocol called MQTT. MQTT is a pub/sub protocol with a very small (2 byte) header which is ideal for small devices such as Arduino and reduces bandwidth usage when sending data across networks. 

But my aim was not fulfilled with this, what I wanted to implement is a system where more and more devices can be connected both as publishers of parameters (Illumination, Temperature, Humidity, etc.) and as consumers of those parameters in a scalable, decoupled manner. So after some more research I found that there are messaging queues used to enable this decoupling and scalability concerns. During my research I found that there is a upcoming tool named Apache Kafka which is capable of delivering what I am expecting in a more secure fault tolerant manner, so I decided to have it a go and see.

Implementation

Hardware

1. 1 Arduino UNO Development Board
2. 1 Raspberry Pi Model B
3. 1 Ethernet Shield
4. 1 Wifi Adaptor (Dongle)
5. 1 Photoresistor 
6. 1 LED
7. 1 Resistors
8. 1 Wireless Router with LAN Port

Software

All Software used in this project is available at https://github.com/shazin/iot

Prerequisites

Apache Kafka must be downloaded and configured to be started on a PC/Laptop, MQTT Server mosquitto must also be running.

For Apache Kafka refer this url Kafka Getting Started
For Mosquitto refer this url Mosquitto

Demonstraton

The Terminal Tabs in the video have the following started;

1. Mosquitto Server 
2. Zookeeper Server for Apache Kafka
3. Apache Kafka Server
4. Apache Kafka Consumer consuming on topic illumination
5. SSH session to raspberry pi which is running illumination.py python script

And the Spring Source IDE is running Mqtt Kafka Bridge.

When the Photoresistor in Arduino is cover the illumination parameter drops below 300 which will trigger the LED in Raspberry Pi to illuminate.

Lessons Learnt

The approach I tried is very responsive and is easy to scale because of the decoupling I have done by using Mosquitto and Apache Kafka. I didn't see any lags in Raspberry Pi consumer or Arduino Producer. This is may be required for System with lower latency SLAs. 

Future Work

As this is my first attempt at IoT systems, I have left out security and device identification aside. But these are highly required in a fully fledged IoT system. Planning to work on them in the future.

References

1. Samza in LinkedIn: How LinkedIn Processes Billions of Events Everyday in Real-time - Neha Narkhede
2. Federated Identity for IoT with OAuth - Paul Fremantle
3. Python Kafka
4. Kafka Java API 
5. MQTT Java API
6. Arduino Pub/Sub Client

Trackbacks/Pings

1. InfoQ China under Technology Point 2 : http://www.infoq.com/cn/news/2016/06/iot-weekly-46