Reactor: Reactive Programming Java 8

One of the more interesting features in the upcoming release of Java 9 is the support for Reactive Programming in Java. At a high-level Reactive programming deals with asynchronous data streams. Reactive streams provides a standard for asynchronous stream processing. Java 9 supports Reactive programming implementing the Reactive Streams specification through the Flow API. But before the release of Java 9, there are ways to implement Reactive programming in Java 8. Reactor and RxJava are a couple of APIs to implement Reactive streams in Java 8. Reactor is used internally by the Spring Framework to it's own Reactive support in version 5. In this post, we will look into Reactor library.

Mybatis Spring Integration

This post will give a quick introduction into integrating MyBatis and Spring with a very basic application. I will describe only the integration using XML based configuration. As always with MyBatis, this starts with setting up SqlSessionFactory from Spring. Instead of configuring the datasource and mapping in mybatis-config.xml, all configuration will be setup in Spring configuration. The only additional configuration will be the Mapper.xml file.

Spring Standalone Application Setup with Gradle

This post is a continuation of the Spring Standalone Application post. In this post I will go over the application setup and the Gradle build file used for building the project. The assumption here is that Gradle is already installed on the machine and ready to use. For instructions on how to install Gradle please go the Gradle installation page.

Continue to full post ...

Application Setup

Following is the directory structure I had for this project. In case of the Annotation based approach, you don't need the resources folder.

+---SpringStandalone
    |   build.gradle
    |               
    \---src
        +---main
        |   \---java
        |       +---main
        |       |       SpringStandaloneTest.java
        |       |       
        |       \---service
        |               TestService.java
        |   \---resources
        |           applicationContext.xml
        |               
        \---test
            \---java

Gradle Build File: Gradle Application Plugin

The Gradle Application plugin is very useful for creating Standalone applications that can be run using gradle run. By default, the Gradle Application plugin applies the java plugin and Distribution plugin. You can configure the main class and any JVM arguments that you may want to pass to the application. To pass the main class name, you can use the mainClassName parameter like below

mainClassName = "main.SpringStandaloneTest"

And to pass any JVM arguments to the application, you can use applicationDefaultJvmArgs as shown below

applicationDefaultJvmArgs = ["-Dprop=val"]

And here is the full build.gradle file I used for the spring standalone application

apply plugin:'application'
mainClassName = "main.SpringStandaloneTest"
applicationName = 'SpringStandalone'

repositories {
    jcenter()
}

dependencies {
  compile group: 'org.springframework', name: 'spring-core', version: '4.3.9.RELEASE'
 compile group: 'org.springframework', name: 'spring-context', version: '4.3.9.RELEASE'
 compile group: 'org.springframework', name: 'spring-beans', version: '4.3.9.RELEASE'
 compile group: 'org.springframework', name: 'spring-tx', version: '4.3.9.RELEASE'
 compile group: 'org.springframework', name: 'spring-orm', version: '4.3.9.RELEASE'

    // Use JUnit test framework
    compile 'junit:junit:4.12'
}


applicationDefaultJvmArgs = ["-Dprop=val"]

Standalone Spring Application

Having a basic spring standalone application ready always helps when you want to do a quick POC or test of new libraries or tools you plan to use. This post shows a couple of ways in which you can setup a basic Spring standalone application.

• Spring Standalone application with ApplicationContext XML Configuration file.
• Spring Standalone application with Annotations

Continue to full post...

For this application I will use a simple service class TestService that has a single method echo(), which returns the Upper case version of any string passed as a parameter. The TestService class will be used a component to be injected into the main application class.

Spring Standalone application with ApplicationContext XML Configuration file

In this model we have 4 files

1. SpringStandaloneTest.java
2. TestService.java
3. applicationContext.xml
4. build.gradle (will go over the gradle build in the next post)

Following are the files used for this application

SpringStandaloneTest.java

package main;

import org.springframework.context.support.ClassPathXmlApplicationContext;

import service.TestService;

public class SpringStandaloneTest {

 public static void main(String[] args) {
  ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext("applicationContext.xml");
  SpringStandaloneTest test = ctx.getBean("springStandalone", SpringStandaloneTest.class);
  test.callService();

 }

 private TestService testService = null;

 private void callService() {
  System.out.println(testService.echo("Hello"));

 }

 public TestService getTestService() {
  return testService;
 }

 public void setTestService(TestService testService) {
  this.testService = testService;
 }

}

TestService.java

package service;

public class TestService {
 
 public String echo(String str) {
  return str.toUpperCase();
 }

}

applicationContext.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xsi:schemaLocation="
        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">

 <bean name="testService" class="service.TestService" />
 <bean name="springStandalone" class="main.SpringStandaloneTest">
  <property name="testService" ref="testService"></property>
 </bean>

</beans>

Spring Standalone application with Annotations

In this model, we will setup the standalone application with annotations instead of the applicationContext.xml file. The annotation @Component is used to define a class as a Spring component, so that Spring can detect it as a Spring bean. The @ComponentScan is used to direct Spring where to look for Components. It is similar to the directive when using Spring XML configuration. The following is the code used to implement the Standalone Spring application using annotations.

SpringStandaloneTest.java

package main;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.ComponentScan;

import service.TestService;

@ComponentScan(basePackages= {"main", "service"})
public class SpringStandaloneTest {

 public static void main(String[] args) {
  ApplicationContext context = new AnnotationConfigApplicationContext(SpringStandaloneTest.class);

  SpringStandaloneTest test = context.getBean(SpringStandaloneTest.class);
  test.callService();
  
 }

 @Autowired
 private TestService service = null;

 private void callService() {
  System.out.println(service .echo("Hello"));

  
 }

}

TestService.java

package service;

import org.springframework.stereotype.Component;

@Component
public class TestService {
 
 public String echo(String str) {
  return str.toUpperCase();
 }

}

In the next post I will describe the Gradle build setup used for this standalone application.

AsynchronousFileChannel

Java 7 AsynchronousFileChannel is an asynchronous channel for reading and writing to files asynchronously. In this post, we will look into a few operations on AsynchronousFileChannel.

• Read from a file using Future
• Read from File using CompletionHandler
• Write to File using Future
• Write to file using CompletionHandler

One thing common to all the operations will be the position within the file. Since we are working asynchronously, there is no current position, and the position where you want to read or write has to be sent with each call.

Continue to full post...

An AsynchronousFileChannel can be created using the open method. This static method takes the Path and the type of operations (READ, WRITE etc.) as paramters

Reading Data with Future

Following is the read() method that returns a Future.

Future operation = fileChannel.read(buffer, 0);

The data from the file is read into the ByteBuffer parameter starting at the beginning of the file. Since this is an asynchronous call, the read() method will return immediately, even if the read is not completed. To check if the read wheter the read operation completed, you can call the isDone() on the returned Future object.
In the following program, we try to read from a file and wait till the read is complete. Although this is not an ideal use case, it will demonstrate how to use the "Read with Future" feature.

package nio;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class AsyncFileChannelTest {
    public static void main(String[] args) {
        Path path = Paths.get("C:/test/data.csv");
        AsynchronousFileChannel fileChannel = null;
        try {
            fileChannel = AsynchronousFileChannel.open(path, StandardOpenOption.READ);
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            long position = 0;

            StringBuffer fileData = new StringBuffer();

            Future operation = fileChannel.read(buffer, position);
            // Future returns -1 if End of File is reached.
            while (operation.get() > 0) {
                while (!operation.isDone())
                    ;
                // Switch ByteBuffer from read to write mode
                buffer.flip();

                byte[] data = new byte[buffer.limit()];
                buffer.get(data);

                fileData.append(new String(data));

                buffer.clear();
                // Update to new read position.
                position = position + operation.get();
                operation = fileChannel.read(buffer, position);
            }
            System.out.println(fileData);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

}

---

Reading with CompletionHandler

This version of the read() operation takes a CompletionHandler as a parameter.

package nio;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.CompletionHandler;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;

public class AsyncFileChannelTest {
    public static class MyCompletionHandler 
            implements CompletionHandler {
        @Override
        public void failed(Throwable exc, ByteBuffer attachment) {
            // TODO Auto-generated method stub

        }

        @Override
        public void completed(Integer result, ByteBuffer buffer) {
            System.out.println("result = " + result);
            bytesRead = result.intValue();
            if (bytesRead < 0)
                return;
            // Switch ByteBuffer from read to write mode
            buffer.flip();

            byte[] data = new byte[buffer.limit()];
            buffer.get(data);

            fileData.append(new String(data));

            buffer.clear();
            position = position + bytesRead;
            fileChannel.read(buffer, position, buffer, this);

        }

    }

    private static int bytesRead = 0;
    private static StringBuffer fileData = new StringBuffer();
    private static long position = 0;
    private static AsynchronousFileChannel fileChannel = null;

    public static void main(String[] args) {
        Path path = Paths.get("C:/test/data.csv");

        try {
            fileChannel = AsynchronousFileChannel.open(path, StandardOpenOption.READ);
            ByteBuffer buffer = ByteBuffer.allocate(1024);

            MyCompletionHandler myCompletionHandler = new MyCompletionHandler();
            fileChannel.read(buffer, position, buffer, myCompletionHandler);
            // read() returns -1 if End of File is reached.
            while (bytesRead > 0) {

                // Update to new read position.
                position = position + bytesRead;
                fileChannel.read(buffer, position, buffer, myCompletionHandler);
            }

            // Main thread waits till the asynchronous operations are complete
            try {
                Thread.currentThread().sleep(10000);
            } catch (InterruptedException e) {

                e.printStackTrace();
            }
            System.out.println(fileData);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

}

Once the read is finished the completed() method of CompletionHandler is called with the number of Bytes read as a paramter. The second parameter can be any object, and in our case, I passed the same ByteBuffer object as a parameter to help read the entire file. The failed() method is called if the read operation fails.

Writing Data using Future

Writing to a file with Future follows the same model as read. The differnces are that you open the channel in "WRITE" mode instead of "READ" and call the write() method. When the write is done, the number bytes written to the file will be set in the Future. As with read() we wait till the read is complete by calling isDone() method.

package nio;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.CompletionHandler;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class AsyncFileChannelTest {

    public static void main(String[] args) {
        Path path = Paths.get("C:/test/data.csv");
        AsynchronousFileChannel fileChannel = null;

        try {
            fileChannel = AsynchronousFileChannel.open(path, StandardOpenOption.WRITE);
            ByteBuffer buffer = ByteBuffer.allocate(100);

            buffer.put("Hello World".getBytes());
            buffer.flip();

            Future operation = fileChannel.write(buffer, 0);
            buffer.clear();

            while(!operation.isDone());

            System.out.println("Number of Bytes Written : " + operation.get());

            
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

       
    }

}

Writing with CompletionHandler

Writing to a file with CompletionHandler follows the same model as a read. The differnces are that you open the channel in "WRITE" mode instead of "READ" and call the write() method.

package nio;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.CompletionHandler;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;

public class AsyncFileChannelTest {
    public static class MyCompletionHandler implements CompletionHandler {

        @Override
        public void completed(Integer result, ByteBuffer attachment) {
            System.out.println("Number of bytes written : " + result);

        }

        @Override
        public void failed(Throwable exc, ByteBuffer attachment) {
            // TODO Auto-generated method stub

        }

    }

    public static void main(String[] args) {
        Path path = Paths.get("C:/test/data.csv");
        AsynchronousFileChannel fileChannel = null;

        try {
            fileChannel = AsynchronousFileChannel.open(path, StandardOpenOption.WRITE);
            ByteBuffer buffer = ByteBuffer.allocate(100);

            MyCompletionHandler myCompletionHandler = new MyCompletionHandler();

            buffer.put("Hello World".getBytes());
            buffer.flip();
            fileChannel.write(buffer, 0, buffer, myCompletionHandler);
        } catch (IOException e) {
            e.printStackTrace();
        }

        // Main thread waits till the asynchronous operations are complete
        try {
            Thread.currentThread().sleep(10000);
        } catch (InterruptedException e) {

            e.printStackTrace();
        }

    }

}