DbParallelTest.java

package org.easybatch.jdbc;

import org.easybatch.core.api.Record;
import org.easybatch.core.api.RecordMapper;
import org.easybatch.core.api.RecordProcessor;
import org.easybatch.core.dispatcher.PoisonRecordBroadcaster;
import org.easybatch.core.dispatcher.RoundRobinRecordDispatcher;
import org.easybatch.core.filter.PoisonRecordFilter;
import org.easybatch.core.impl.Engine;
import org.easybatch.core.mapper.GenericRecordMapper;
import org.easybatch.core.reader.QueueRecordReader;
import org.easybatch.core.record.GenericRecord;
import org.junit.Before;
import org.junit.Test;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.util.Arrays;
import java.util.concurrent.*;

import static org.easybatch.core.impl.EngineBuilder.aNewEngine;

public class DbParallelTest {

    private static final int THREAD_POOL_SIZE = 3;

    private Connection connection;

    private String query;

    @Before
    public void setUp() throws Exception {
        Class.forName("com.mysql.jdbc.Driver");
        connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?user=root&password=root");
        query = "select * from tweet;";
    }

    @Test
    public void testDatabaseParallelProcessing() throws Exception {

        BlockingQueue<Record> queue1 = new LinkedBlockingQueue<Record>();
        BlockingQueue<Record> queue2 = new LinkedBlockingQueue<Record>();

        RoundRobinRecordDispatcher roundRobinRecordDispatcher = new RoundRobinRecordDispatcher(Arrays.asList(queue1, queue2));

        Engine masterEngine = aNewEngine()
                .reader(new JdbcRecordReader(connection, query))
                .mapper(new CustomJdbcMapper())
                .processor(roundRobinRecordDispatcher)
                .batchProcessEventListener(new PoisonRecordBroadcaster(roundRobinRecordDispatcher))
                .build();

        // Build worker engines
        Engine workerEngine1 = buildWorkerEngine(queue1);
        Engine workerEngine2 = buildWorkerEngine(queue2);

        // Create a thread pool to call master and worker engines in parallel
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);

        // Submit workers to executor service
        executorService.invokeAll(Arrays.asList(masterEngine, workerEngine1, workerEngine2));

        executorService.shutdown();

    }

    public Engine buildWorkerEngine(BlockingQueue<Record> queue) {
        return aNewEngine()
                .reader(new QueueRecordReader(queue))
                .filter(new PoisonRecordFilter())
                .mapper(new GenericRecordMapper())
                .processor(new TweetProcessor())
                .build();
    }

    private class TweetProcessor implements RecordProcessor<Tweet, Tweet> {

        @Override
        public Tweet processRecord(Tweet tweet) throws Exception {
            System.out.println(Thread.currentThread().getName() + ": tweet = " + tweet);
            return tweet;
        }
    }

    private class CustomJdbcMapper implements RecordMapper<GenericRecord<Tweet>> {

        public GenericRecord<Tweet> mapRecord(Record record) throws Exception {
            JdbcRecord jdbcRecord = (JdbcRecord) record;
            ResultSet resultSet = jdbcRecord.getPayload();
            Tweet tweet = new Tweet();
            tweet.setId((resultSet.getInt("id")));
            tweet.setUser(resultSet.getString("user"));
            tweet.setMessage(resultSet.getString("message"));
            return new GenericRecord<Tweet>(record.getHeader(), tweet);
        }

    }
}

Tweet.java

/*
 * The MIT License
 *
 *  Copyright (c) 2015, Mahmoud Ben Hassine ([email protected])
 *
 *  Permission is hereby granted, free of charge, to any person obtaining a copy
 *  of this software and associated documentation files (the "Software"), to deal
 *  in the Software without restriction, including without limitation the rights
 *  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 *  copies of the Software, and to permit persons to whom the Software is
 *  furnished to do so, subject to the following conditions:
 *
 *  The above copyright notice and this permission notice shall be included in
 *  all copies or substantial portions of the Software.
 *
 *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 *  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 *  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 *  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 *  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 *  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 *  THE SOFTWARE.
 */

package org.easybatch.jdbc;

/**
 * Java bean representing a tweet.
 *
 * @author Mahmoud Ben Hassine ([email protected])
 */
public class Tweet {

    private int id;

    private String user;

    private String message;

    public Tweet() {
    }

    public Tweet(int id, String user, String message) {
        this.id = id;
        this.user = user;
        this.message = message;
    }

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getUser() {
        return user;
    }

    public void setUser(String user) {
        this.user = user;
    }

    public String getMessage() {
        return message;
    }

    public void setMessage(String message) {
        this.message = message;
    }

    @Override
    public String toString() {
        final StringBuilder sb = new StringBuilder("Tweet{");
        sb.append("id=").append(id);
        sb.append(", user='").append(user).append('\'');
        sb.append(", message='").append(message).append('\'');
        sb.append('}');
        return sb.toString();
    }

}