Event driven programming using Spring Boot and Reactor Example

Core concepts

Reactor:

Reactor is a framework to make event driven programming much easier and it is based on Reactor Design Pattern. Reactor is good for asynchronous applications on the JVM, it is an event gateway where event consumers are registered with a notification key.

Selector:

Selector is an abstraction to find consumer by invoking event.

Consumers and Event:

Consumers and Events as core module, Consumer is event consumer which needs to be notified for the event.

Producer:

Producer produces the Events and publish.

Here is one example which use the Reactor pattern to Message produce and consumes.

1. Create a Maven project and add below dependencies

<properties>
        <java.version>1.8</java.version>
        <version.reactor>2.0.6.RELEASE</version.reactor>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-web</artifactId>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-bus</artifactId>
            <version>${version.reactor}</version>
        </dependency>
        <dependency>
            <groupId>org.reactivestreams</groupId>
            <artifactId>reactive-streams</artifactId>
            <version>1.0.0.final</version>
        </dependency>

        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-core</artifactId>
            <version>${version.reactor}</version>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-stream</artifactId>
            <version>${version.reactor}</version>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-groovy</artifactId>
            <version>${version.reactor}</version>
        </dependency>
        <dependency>
            <groupId>io.projectreactor.spring</groupId>
            <artifactId>reactor-spring-core</artifactId>
            <version>${version.reactor}</version>
        </dependency>
        <dependency>
            <groupId>io.projectreactor.spring</groupId>
            <artifactId>reactor-spring-context</artifactId>
            <version>${version.reactor}</version>
        </dependency>

    </dependencies>

2. Create Customer POJO

package com.vinod.test;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
@JsonIgnoreProperties(ignoreUnknown = true)
public class Customer {
    private String name;
    private String address;
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public String getAddress() {
        return address;
    }
    public void setAddress(String address) {
        this.address = address;
    }
    @Override
    public String toString() {
        return "Customer [name=" + name + ", address=" + address + "]";
    }

}

3. Create a Consumer

package com.vinod.test;
import org.springframework.stereotype.Service;
import reactor.bus.Event;
import reactor.fn.Consumer;

@Service
class CustomerReceiver implements Consumer<Event<Customer>> {

    public void accept(Event<Customer> ev) {
        System.out.println("Customer " + ev.getData());
    }

}
 

4. Create a publisher

package com.vinod.test;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.bus.Event;
import reactor.bus.EventBus;

@Service
public class CustomerPublisher {

    @Autowired
    EventBus eventBus;

    public void publishCustomerDetails() throws InterruptedException {
        Customer customer = new Customer();
        customer.setName("vinod");
        customer.setAddress("Sasi area");
        eventBus.notify("customer", Event.wrap(customer));
        System.out.println("Message sent");
    }

}

5. Spring boot Main class

package com.vinod.test;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import reactor.Environment;
import reactor.bus.EventBus;

import static reactor.bus.selector.Selectors.$;

@Configuration
@EnableAutoConfiguration
@ComponentScan
public class Application implements CommandLineRunner {

    @Bean
    Environment env() {
        return Environment.initializeIfEmpty().assignErrorJournal();
    }

    @Bean
    EventBus createEventBus(Environment env) {
        return EventBus.create(env, Environment.THREAD_POOL);
    }

    @Autowired
    private EventBus eventBus;
    @Autowired
    private CustomerReceiver customerReceiver;
    @Autowired
    private CustomerPublisher customerPublisher;
    @Autowired
    private AdminReceiver adminReceiver;

    public void run(String... args) throws Exception {
        eventBus.on($("customer"), customerReceiver);
        customerPublisher.publishCustomerDetails();
    }

    public static void main(String[] args) throws InterruptedException {
        SpringApplication.run(Application.class, args);

    }

}

6. Run the program (Application.java)

Here we can see the publisher sends the message and consumer is consuming the same.

Message sent
Customer Customer [name=vinod, address=Sasi area]

7. Download Example

https://github.com/kkvinodkumaran/spring

Java Drools Example

Drools is a open source Business Logic integration Platform based on java from JBoss and RedHat. Here is one simple example to execute business logics based on rules. see more about drools

1. Create a maven project and add below dependencies

 
  <dependencies>
        <dependency>
            <groupId>org.drools</groupId>
            <artifactId>drools-compiler</artifactId>
            <version>6.0.0.CR1</version>
        </dependency>
        <dependency>
            <groupId>org.drools</groupId>
            <artifactId>drools-core</artifactId>
            <version>6.0.0.CR1</version>
        </dependency>
    </dependencies>

2. Create a .drl file to add our business logic (eligibility.drl)

import com.vinod.test.Customer
 
dialect "mvel"
 
rule "Customer eligibility rule"
    when
        $customer : Customer(age>=21)              
    then
        System.out.println($customer.name" Eligibile for benefits");
end

3. Create Customer pojo class

package com.vinod.test;

public class Customer {
    private String name;
    private int age;
    public Customer() {
    }
    public Customer(String name, int age) {
        this.name = name;
        this.age = age;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public int getAge() {
        return age;
    }
    public void setAge(int age) {
        this.age = age;
    }
}
 

4. Create a main class to load and fire rules

package com.vinod.test;

import java.io.File;
import java.io.FilenameFilter;
import java.util.ArrayList;
import java.util.List;

import org.kie.api.KieServices;
import org.kie.api.builder.KieBuilder;
import org.kie.api.builder.KieFileSystem;
import org.kie.api.builder.KieRepository;
import org.kie.api.io.KieResources;
import org.kie.api.io.Resource;
import org.kie.api.runtime.KieContainer;
import org.kie.api.runtime.KieSession;

public class CustomerEligibilityRuleTest {

    public static void main(String[] args) {
        List<File> rules = new ArrayList<File>();
        File currentdir = new File(System.getProperty("user.dir"));
        File[] listFiles = currentdir.listFiles(new FilenameFilter() {
            public boolean accept(File dir, String name) {
                return (name.endsWith(".drl"));
            }
        });
        for (File listFile : listFiles) {
            System.out.println(listFile.getName());
            rules.add(listFile);
        }
        List<Customer> customers = new ArrayList<Customer>();
        customers.add(new Customer("Vinod k kumaran", 21));
        executeRules(rules, customers);
    }

    public static void executeRules(List<File> files, List<Customer> customers) {
        try {
            KieServices kieServices = KieServices.Factory.get();
            KieResources kieResources = kieServices.getResources();
            KieFileSystem kieFileSystem = kieServices.newKieFileSystem();
            KieRepository kieRepository = kieServices.getRepository();
            for (File ruleFile : files) {
                Resource resource = kieResources.newFileSystemResource(ruleFile);
                kieFileSystem.write(resource);
            }
            KieBuilder kb = kieServices.newKieBuilder(kieFileSystem);
            kb.buildAll();
            KieContainer kContainer = kieServices.newKieContainer(kieRepository.getDefaultReleaseId());
            KieSession kSession = kContainer.newKieSession();
            kSession.addEventListener(new CustomerRuleTracker());
            for (Object customer : customers) {
                kSession.insert(customer);
            }
            kSession.fireAllRules();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
 

5. Add rule tracker class to listen all stages of firing rules 

This class is added in the KieSession already in the above main class

package com.vinod.test;

import org.drools.core.event.BeforeActivationFiredEvent;
import org.kie.api.event.rule.AfterMatchFiredEvent;
import org.kie.api.event.rule.BeforeMatchFiredEvent;
import org.kie.api.event.rule.DefaultAgendaEventListener;
import org.kie.api.event.rule.MatchCancelledEvent;
import org.kie.api.event.rule.MatchCreatedEvent;
import org.kie.api.event.rule.MatchEvent;
import org.kie.api.event.rule.RuleFlowGroupActivatedEvent;

public class CustomerRuleTracker extends DefaultAgendaEventListener {

    @Override
    public void afterMatchFired(AfterMatchFiredEvent event) {
        System.out.println("Executing afterMatchFired ");
    }
    @Override
    public void beforeRuleFlowGroupActivated(RuleFlowGroupActivatedEvent event) {
        System.out.println("Executing beforeRuleFlowGroupActivated ");

    }
    @Override
    public void beforeMatchFired(BeforeMatchFiredEvent event) {
        System.out.println("Executing beforeMatchFired ");
    }
    public void beforeActivationFired(BeforeActivationFiredEvent event) {
        System.out.println("Executing beforeActivationFired ");

    }
    @Override
    public void matchCreated(MatchCreatedEvent event) {
        System.out.println("Executing matchCreated ");
        registerActions(event);
    }
    @Override
    public void matchCancelled(MatchCancelledEvent event) {
        System.out.println("Executing matchCancelled");

    }
    private void registerActions(MatchEvent event) {
        System.out.println("Executing registerActions");
    }
}

6. Run the main program.

We can see the customer name + eligible for benefits printed as per our rules..

eligibility.drl

Executing matchCreated
Executing registerActions
Executing beforeMatchFired
Vinod k kumaran Eligibile for benefits
Executing afterMatchFired

7. Download Example

https://github.com/kkvinodkumaran/drools

PGP Encryption and Decryption using Apache Camel

In this example we will see how do to the file encryption and decryption using Apache Camel using pgp

1. Generate pgp keys

In order to do the encryption/decryption we required the pgp public and private keys, in this example we will use below portal to generate the keys. Once we created the files we need to place it in to our source resource folder

https://www.igolder.com/pgp/generate-key/

 

2. Create a maven project and add below dependencies

<properties>
        <camel.version>2.13.2</camel.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-ftp</artifactId>
            <version>2.14.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-core</artifactId>
            <version>${camel.version}</version>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-crypto</artifactId>
            <version>${camel.version}</version>

        </dependency>
        <dependency>
            <groupId>bouncycastle</groupId>
            <artifactId>bcpg-jdk13</artifactId>
            <version>121</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.5</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.5</version>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
        </dependency>
    </dependencies>

3. Create a Camel Route to start encryption

 

package com.vinod.test;

import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.converter.crypto.PGPDataFormat;
import org.apache.camel.main.Main;

public class EncryptionTest {

    /**
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        Main main = new Main();
        main.enableHangupSupport();
        main.addRouteBuilder(new MyRouteBuilder());
        main.run(args);

    }

}

class MyRouteBuilder extends RouteBuilder {

    @Override
    public void configure() {

        try {
            System.out.println("My Encryption/Decryption route started");
            // Encryption
            PGPDataFormat encrypt = new PGPDataFormat();
            encrypt.setKeyFileName("vinodpublickeyfile.pgp");
            encrypt.setKeyUserid("vinod");

            from("file:tobeencrypt?noop=true;delete=true").marshal(encrypt).to("file:encrypted");

            // Decryption

            PGPDataFormat decrypt = new PGPDataFormat();
            decrypt.setKeyFileName("vinodprivatekeyfile.pgp");

            decrypt.setKeyUserid("vinod");
            decrypt.setPassword("vinod@123");

            from("file:tobedecrypt").unmarshal(decrypt).to("file:decrypted");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

5. Run the program

After running the above main program camel route will start and we can see the two directories are created and we can place the files which we needs to be encrypted.

Once we placed the file in to tobeencrypt directory Camel route will process that file and place in to encrypted directory after encryption

 

 

 

 

 

 

 

 

 

 

 

 

Download example

https://github.com/kkvinodkumaran/camel 

 

 

 

 

Servlet RequestDispatcher

The javax.servlet.RequestDispatcher interface class help your servlet to call another servlet, JSP file, or HTML file from inside another servlet.

There are two main method in this interfaces

1) forward(ServletRequest request, ServletResponse response)

                  —>> forwards a request from a servlet to another resource (servlet, JSP file, or HTML file) on the server

2) include(ServletRequest request, ServletResponse response)
                  —>> includes the content of a resource (servlet, JSP page, HTML file) in the response

Example

Servlet class

This servlet class will do the include or forward based on the action from the user interface

package com.vinod;

import java.io.IOException;

import javax.servlet.RequestDispatcher;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

/**
 * Servlet implementation class MyServletDispatcher
 */
public class MyServletDispatcher extends HttpServlet {
    private static final long serialVersionUID = 1L;

    /**
     * @see HttpServlet#HttpServlet()
     */
    public MyServletDispatcher() {
        super();
        // TODO Auto-generated constructor stub
    }

    /**
     * @see HttpServlet#doGet(HttpServletRequest request, HttpServletResponse
     *      response)
     */
    protected void doGet(HttpServletRequest request, HttpServletResponse response)
            throws ServletException, IOException {
        RequestDispatcher rd = request.getRequestDispatcher("index.jsp");
        if (request.getParameter("action") != null) {
            String action = request.getParameter("action");
            if (action.equalsIgnoreCase("include")) {
                rd.include(request, response);
            } else if (action.equalsIgnoreCase("forward")) {
                rd.forward(request, response);
            }
        }
    }

    /**
     * @see HttpServlet#doPost(HttpServletRequest request, HttpServletResponse
     *      response)
     */
    protected void doPost(HttpServletRequest request, HttpServletResponse response)
            throws ServletException, IOException {
        // TODO Auto-generated method stub
        doGet(request, response);
    }

}

Test jsp (Dispatcher.jsp)

<%@ page language="java" contentType="text/html; charset=UTF-8"
    pageEncoding="UTF-8"%>
<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
<html>
<head>
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
<title>Insert title here</title>
</head>
<body>
    <form name="input" action="http://localhost:8081/MyServletDispatcher"
        method="get">
        Enter Action: <input type="text" name="action"><input
            type="submit" value="Submit">
    </form>
</body>
</html>

index.JSP

 

<html>
<body>
<h2>Hello World!</h2>
</body>
</html>

Deploy and run the Dispatcher.jsp ( In this example using jetty server, so used jetty:run maven command to start the jetty server)

http://localhost:8081/DispatcherTest.jsp

 

 

Download complete Example

https://github.com/kkvinodkumaran/myj2ee

Java Blocking Queues

java.util.concurrent.BlockingQueue is a Queue that supports operations that wait for the queue to become non-empty when retrieving and removing an element, and wait for space to become available in the queue when adding an element.

1. Thread safe
2. Allow duplicates
3. Not allowed null values

   

Types

• ArrayBlockingQueue
• DelayQueue
• LinkedBlockingQueue
• PriorityBlockingQueue
• SynchronousQueue

Example

package com.vinod.test;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class BlockingQueueExamples {
   public static void main(String[] args) throws InterruptedException, ExecutionException {
      BlockingQueue<String> bq = new ArrayBlockingQueue<String>(1000);
      ExecutorService exec = Executors.newFixedThreadPool(2);
      exec.submit(new ConsumerThread(bq));
      exec.submit(new ProducerThread(bq));

   }

}

class ProducerThread implements Callable<Object> {
   private BlockingQueue<String> blockingQueue;

   public ProducerThread(BlockingQueue<String> blockingQueue) {
      this.blockingQueue = blockingQueue;
   }

   public Object call() throws Exception {
      System.out.println("Producer Started:");
      Thread.sleep(1000);
      blockingQueue.put("vinod");
      Thread.sleep(1000);
      blockingQueue.put("vinod1");
      Thread.sleep(1000);
      blockingQueue.put("vinod2");
      return null;
   }

}

class ConsumerThread implements Callable<Object> {
   private BlockingQueue<String> blockingQueue;

   public ConsumerThread(BlockingQueue<String> blockingQueue) {
      this.blockingQueue = blockingQueue;
   }

   public Object call() throws Exception {
      System.out.println("Consumer Started:");
      System.out.println(blockingQueue.take());
      System.out.println(blockingQueue.take());
      System.out.println(blockingQueue.take());

      return null;
   }

}

Output

Consumer Started:

Producer Started:

vinod

vinod1

vinod2