<|start_filename|>src/io/iteratorx/Function.java<|end_filename|>
package io.iteratorx;

import java.io.Serializable;

public interface Function<T, R> extends Serializable {

	R apply(T t) throws Exception;
}

<|start_filename|>src/io/iteratorx/parallels/Threads.java<|end_filename|>
package io.iteratorx.parallels;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Logger;

import io.iteratorx.Consumer;

public class Threads<T> implements Parallels {
	private static final Logger logger = Logger.getLogger(Threads.class.getName());

	public static <T> long forEach(final int parallelism, final Iterator<T> iterator, final Consumer<T> onNext)
			throws Exception {

		// for back pressure
		final BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(parallelism * 3);
		final ExecutorService executorService = new ThreadPoolExecutor(parallelism, parallelism, 0L,
				TimeUnit.MILLISECONDS, queue);

		// submit runnable
		long count = 0;
		while (iterator.hasNext()) {
			final T item = iterator.next();
			count++;

			// for back pressure
			while (queue.size() >= parallelism * 3) {
				Thread.sleep(10);
			}

			// submit
			executorService.submit(new Runnable() {
				@Override
				public void run() {
					try {
						onNext.accept(item);

					} catch (final Exception e) {
						throw new RuntimeException(e);
					}
				}
			});
		}

		// shutdown, still executing
		executorService.shutdown();

		// waiting for executing finished
		try {
			if (!executorService.awaitTermination(1, TimeUnit.DAYS)) {
				executorService.shutdownNow();
			}
		} catch (final InterruptedException e) {

			// shutdown, stop executing
			executorService.shutdownNow();

			throw e;
		}

		return count;
	}

	public static <T> Threads<T> from(final Iterator<T> iterator) throws Exception {
		return new Threads<T>(iterator);
	}

	public static <T> Threads<T> from(final Iterator<T> iterator, final int parallelism) throws Exception {
		return new Threads<T>(iterator, DefaultParallelism);
	}

	public static <T> Threads<T> from(final Iterable<T> iterable) throws Exception {
		return new Threads<T>(iterable);
	}

	public static <T> Threads<T> from(final Iterable<T> iterable, final int parallelism) throws Exception {
		return new Threads<T>(iterable, DefaultParallelism);
	}

	private final Iterator<T> iterator;
	private final int parallelism;

	public Threads(final Iterator<T> iterator) {
		this(iterator, DefaultParallelism);
	}

	public Threads(final Iterator<T> iterator, final int parallelism) {
		this.iterator = iterator;
		this.parallelism = parallelism;
	}

	public Threads(final Iterable<T> iterable) {
		this(iterable.iterator());
	}

	public Threads(final Iterable<T> iterable, final int parallelism) {
		this(iterable.iterator(), parallelism);
	}

	public long forEach(final Consumer<T> onNext) throws Exception {
		final long count = forEach(this.parallelism, this.iterator, onNext);

		logger.info("OK finished! count = " + count);
		return count;
	}

	public long forBatch(final Consumer<List<T>> onNext) throws Exception {
		return forBatch(DefaultBatchSize, onNext);
	}

	public long forBatch(final int batchSize, final Consumer<List<T>> onNext) throws Exception {
		final AtomicLong count = new AtomicLong();
		final Iterator<List<T>> groupIterator = new Iterator<List<T>>() {

			final List<T> group = new ArrayList<>();

			@Override
			public boolean hasNext() {
				return Threads.this.iterator.hasNext();
			}

			@Override
			public List<T> next() {
				do {
					final T item = Threads.this.iterator.next();
					group.add(item);
					count.incrementAndGet();

					if (group.size() >= batchSize) {
						try {
							return new ArrayList<>(group);
						} finally {
							group.clear();
						}
					}

				} while (Threads.this.iterator.hasNext());

				if (group.size() > 0) {
					return new ArrayList<>(group);
				} else {
					return null;
				}
			}
		};

		final long group = forEach(this.parallelism, groupIterator, onNext);
		logger.info("OK finished! count = " + count + "; group = " + group);
		return count.get();
	}
}


<|start_filename|>src/io/iteratorx/Consumer.java<|end_filename|>
package io.iteratorx;

import java.io.Serializable;

public interface Consumer<T> extends Serializable {

	void accept(T t) throws Exception;
}

<|start_filename|>src/io/iteratorx/parallels/Parallels.java<|end_filename|>
package io.iteratorx.parallels;

public interface Parallels {

	public static int DefaultParallelism = 3 * Runtime.getRuntime().availableProcessors();

	public static int DefaultBatchSize = 1000;

}


<|start_filename|>src/io/iteratorx/samples/JdbcWriterUsage.java<|end_filename|>
package io.iteratorx.samples;

import com.alibaba.fastjson.JSONObject;

import io.iteratorx.reader.DataSourceBuilder.JdbcDataSourceBuilder;
import io.iteratorx.reader.JdbcWriter;

public class JdbcWriterUsage {

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

		// create jdbc Writer
		final JdbcWriter jdbcWriter = new JdbcWriter(
				new JdbcDataSourceBuilder().setUrl("jdbc:postgresql://10.23.112.2:3333/dbname").setUser("username")
						.setPassword("password").build());

		// write
		jdbcWriter.execute("insert into tablename values(?, ?)", "name", "type");

		// write batch by iterable
		final Iterable<JSONObject> iter = jdbcWriter.asJdbcReader().read("select name, type from tablename");
		jdbcWriter.executeBatch("insert into tablename(name, type) values(?, ?)", iter);

	}
}


<|start_filename|>src/io/iteratorx/samples/FileReaderUsage.java<|end_filename|>
package io.iteratorx.samples;

import java.io.File;
import java.util.Collection;

import com.alibaba.fastjson.JSONObject;

import io.iteratorx.reader.FileReader;

public class FileReaderUsage {

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

		// create file reader
		final FileReader fileReader = new FileReader();

		// fetch by iterable
		for (final JSONObject item : fileReader.read(new File("data.json"), "utf-8")) {
			System.err.println(item);
		}

		// fetch all into one collection
		final Collection<JSONObject> items = fileReader.readAll(new File("data.json"), "utf-8");
		for (final JSONObject item : items) {
			System.err.println(item);
		}

	}
}


<|start_filename|>src/io/iteratorx/reader/JdbcReader.java<|end_filename|>
package io.iteratorx.reader;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.logging.Logger;

import javax.sql.DataSource;

import org.junit.Assert;

import com.alibaba.fastjson.JSONObject;

import io.iteratorx.AutoCloseableIterator;

public class JdbcReader {
	private static final Logger logger = Logger.getLogger(JdbcReader.class.getName());

	private final DataSource dataSource;

	protected int batchSize = 1000;
	protected int queryTimeout = 60 * 60;

	private JSONObject columnMetaData = null;

	private Connection conn = null;
	private PreparedStatement ps = null;
	private ResultSet rs = null;

	public JdbcReader(final DataSource dataSource) {
		Assert.assertNotNull(dataSource);
		this.dataSource = dataSource;
	}

	public JdbcReader setBatchSize(final int batchSize) {
		this.batchSize = batchSize;
		return this;
	}

	public JdbcReader setQueryTimeout(final int seconds) {
		this.queryTimeout = seconds;
		return this;
	}

	public JSONObject getColumnMetaData() {
		return this.columnMetaData;
	}

	public List<JSONObject> readAll(final String sql, final Object... parameters) {
		final List<JSONObject> results = new ArrayList<>();
		for (final JSONObject one : read(sql, parameters)) {
			results.add(one);
		}
		return results;
	}

	/**
	 * Read each jdbc Table Row into JSONObject iteratively.
	 * 
	 * NOTICE: thread-unsafe: do not use result Iterable in multi-threads!
	 * 
	 */
	public Iterable<JSONObject> read(final String sql, final Object... parameters) {

		// query batch
		try {
			// query
			conn = dataSource.getConnection();
			conn.setAutoCommit(false);
			ps = conn.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
			for (int i = 0; i < parameters.length; i++) {
				ps.setObject(i + 1, parameters[i]);
			}
			ps.setFetchSize(batchSize);
			ps.setQueryTimeout(queryTimeout);
			rs = ps.executeQuery();

			// get column metadata
			columnMetaData = parseColumnMetaData(rs.getMetaData());

		} catch (final SQLException e) {
			IOUtils.close(rs, ps, conn);

			throw new RuntimeException(e);
		}

		// return iterator
		return new Iterable<JSONObject>() {

			@Override
			public Iterator<JSONObject> iterator() {
				return new AutoCloseableIterator<JSONObject>() {
					private static final long serialVersionUID = -7178131452993168194L;

					boolean hasNext = true;

					@Override
					public boolean hasNext() {
						if (hasNext == false) {
							// already closed

							return false;
						}

						try {
							hasNext = rs.next();
							if (hasNext == false) {
								close();
							}

							return hasNext;

						} catch (final SQLException e) {
							close();

							throw new RuntimeException(e);
						}
					}

					@Override
					public JSONObject next() {
						try {
							final JSONObject data = new JSONObject(true);
							for (final String column : getColumnMetaData().keySet()) {
								data.put(column, rs.getObject(column));
							}

							return data;

						} catch (final SQLException e) {
							close();

							throw new RuntimeException(e);
						}
					}

					@Override
					public void close() {
						IOUtils.close(rs, ps, conn);
						hasNext = false;
					}
				};

			}
		};

	}

	protected JSONObject parseColumnMetaData(final ResultSetMetaData resultSetMetaData) throws SQLException {
		final JSONObject metadata = new JSONObject(true);

		for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
			final JSONObject columnMetaData = new JSONObject();
			metadata.put(resultSetMetaData.getColumnName(i), columnMetaData);

			columnMetaData.put("ColumnType", resultSetMetaData.getColumnType(i));
			columnMetaData.put("ColumnTypeName", resultSetMetaData.getColumnTypeName(i));
			columnMetaData.put("ColumnClassName", resultSetMetaData.getColumnClassName(i));
		}
		return metadata;
	}

	public JdbcWriter asJdbcWriter() {
		return new JdbcWriter(this.dataSource);
	}
}


<|start_filename|>src/io/iteratorx/samples/JdbcReaderUsage.java<|end_filename|>
package io.iteratorx.samples;

import java.util.Collection;

import com.alibaba.fastjson.JSONObject;

import io.iteratorx.reader.DataSourceBuilder.JdbcDataSourceBuilder;
import io.iteratorx.reader.JdbcReader;

public class JdbcReaderUsage {

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

		// create jdbc reader
		final JdbcReader jdbcReader = new JdbcReader(
				new JdbcDataSourceBuilder().setUrl("jdbc:postgresql://10.23.112.2:3333/dbname").setUser("username")
						.setPassword("password").build());

		// fetch by iterable
		for (final JSONObject item : jdbcReader.read("select * from tablename")) {
			System.err.println(item);
		}

		// fetch all into one collection
		final Collection<JSONObject> items = jdbcReader.readAll("select * from tablename where id = ?", "param");
		for (final JSONObject item : items) {
			System.err.println(item);
		}

	}
}


<|start_filename|>src/io/iteratorx/reader/DataSourceBuilder.java<|end_filename|>
package io.iteratorx.reader;

import java.io.PrintWriter;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.SQLFeatureNotSupportedException;
import java.util.Properties;
import java.util.logging.Logger;

import javax.sql.DataSource;

public abstract class DataSourceBuilder {

	public static class JdbcDataSourceBuilder extends DataSourceBuilder {

		@Override
		public DataSource build() {
			return new DataSource() {

				@Override
				public <T> T unwrap(final Class<T> iface) throws SQLException {
					throw new UnsupportedOperationException();
				}

				@Override
				public boolean isWrapperFor(final Class<?> iface) throws SQLException {
					throw new UnsupportedOperationException();
				}

				@Override
				public Logger getParentLogger() throws SQLFeatureNotSupportedException {
					throw new UnsupportedOperationException();
				}

				@Override
				public void setLoginTimeout(final int seconds) throws SQLException {
					DriverManager.setLoginTimeout(seconds);
				}

				@Override
				public void setLogWriter(final PrintWriter out) throws SQLException {
					DriverManager.setLogWriter(out);
				}

				@Override
				public int getLoginTimeout() throws SQLException {
					return DriverManager.getLoginTimeout();
				}

				@Override
				public PrintWriter getLogWriter() throws SQLException {
					return DriverManager.getLogWriter();
				}

				@Override
				public Connection getConnection() throws SQLException {
					return getConnection(properties.getProperty("user"), properties.getProperty("password"));
				}

				@Override
				public Connection getConnection(final String user, final String password) throws SQLException {

					properties.setProperty("user", user);
					properties.setProperty("password", password);

					try {
						final String url = properties.getProperty("url");

						final String driver = IOUtils.getDriver(url, properties.getProperty("driver"));
						Class.forName(driver);

						final int loginTimeout = properties.get("loginTimeout") == null ? 60
								: (int) properties.get("loginTimeout");
						DriverManager.setLoginTimeout(loginTimeout);

						final Connection conn = DriverManager.getConnection(url, properties);
						if (conn.isClosed()) {
							throw new SQLException("connection closed!");
						}
						return conn;

					} catch (final ClassNotFoundException e) {
						throw new RuntimeException(e);
					}
				}

			};
		}
	}

	protected final Properties properties = new Properties();

	protected DataSourceBuilder() {
	}

	public DataSourceBuilder setDriver(final String driver) {
		return set("driver", driver);
	}

	public DataSourceBuilder setUrl(final String url) {
		return set("url", url);
	}

	public DataSourceBuilder setUser(final String user) {
		return set("user", user);
	}

	public DataSourceBuilder setPassword(final String password) {
		return set("password", password);
	}

	public DataSourceBuilder setLoginTimeout(final int seconds) {
		return set("loginTimeout", seconds);
	}

	public DataSourceBuilder set(final String key, final Object value) {
		properties.put(key, value);
		return this;
	}

	public abstract DataSource build();

}


<|start_filename|>src/io/iteratorx/parallels/Flink.java<|end_filename|>
package io.iteratorx.parallels;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.IteratorInputFormat;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;

import io.iteratorx.AutoCloseableIterator;
import io.iteratorx.Consumer;

public class Flink implements Parallels {

	public static class IteratorInputFormatProxy extends IteratorInputFormat<JSONObject> {
		private static final long serialVersionUID = 6007638560942548638L;

		private static final Map<UUID, Iterator<JSONObject>> iterators = new HashMap<>();

		public static IteratorInputFormatProxy create(final Iterator<JSONObject> iterator) {
			final UUID uuid = UUID.randomUUID();
			iterators.put(uuid, iterator);
			return new IteratorInputFormatProxy(new AutoCloseableIterator<JSONObject>() {
				private static final long serialVersionUID = 4641584376245923845L;

				@Override
				public boolean hasNext() {
					return iterators.get(uuid).hasNext();
				}

				@Override
				public JSONObject next() {
					return iterators.get(uuid).next();
				}

				@Override
				public void close() throws Exception {
					final Iterator<JSONObject> iterator = iterators.get(uuid);
					if (iterator instanceof AutoCloseable) {
						((AutoCloseable) iterator).close();
					}
				}
			});
		}

		private IteratorInputFormatProxy(final Iterator<JSONObject> iterator) {
			super(iterator);
		}

	}

	public static <T> long forEach(final DataSet<T> dataSet, final Consumer<T> onNext) throws Exception {
		return dataSet.filter(t -> {
			onNext.accept(t);
			return true;
		}).count();
	}

	public static Flink from(final Iterator<JSONObject> iterator) throws Exception {
		return new Flink(iterator);
	}

	public static Flink from(final Iterator<JSONObject> iterator, final int parallelism) throws Exception {
		return new Flink(iterator, DefaultParallelism);
	}

	public static Flink from(final Iterable<JSONObject> iterable) throws Exception {
		return new Flink(iterable);
	}

	public static Flink from(final Iterable<JSONObject> iterable, final int parallelism) throws Exception {
		return new Flink(iterable, DefaultParallelism);
	}

	private final DataSet<JSONObject> dataSet;

	public Flink(final Iterator<JSONObject> iterator) {
		this(iterator, DefaultParallelism);
	}

	public Flink(final Iterator<JSONObject> iterator, final int parallelism) {
		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(parallelism);

		this.dataSet = env.createInput(IteratorInputFormatProxy.create(iterator));
	}

	public Flink(final Iterable<JSONObject> iterable) {
		this(iterable.iterator());
	}

	public Flink(final Iterable<JSONObject> iterable, final int parallelism) {
		this(iterable.iterator(), parallelism);
	}

	public DataSet<JSONObject> dataSet() {
		return this.dataSet;
	}

	public long forEach(final Consumer<JSONObject> onNext) throws Exception {
		return forEach(this.dataSet, onNext);
	}

	public long forBatch(final Consumer<List<JSONObject>> onNext) throws Exception {
		return forBatch(DefaultBatchSize, onNext);
	}

	public long forBatch(final int batchSize, final Consumer<List<JSONObject>> onNext) throws Exception {
		final DataSet<List<JSONObject>> groupDataSet = dataSet
				.reduceGroup(new GroupReduceFunction<JSONObject, List<JSONObject>>() {
					private static final long serialVersionUID = -1748951891205342549L;

					private final AtomicInteger count = new AtomicInteger();
					private final List<JSONObject> group = new ArrayList<>();

					@Override
					public void reduce(final Iterable<JSONObject> values, final Collector<List<JSONObject>> out)
							throws Exception {
						for (final JSONObject one : values) {
							group.add(one);
							count.getAndIncrement();

							if (group.size() >= batchSize) {
								out.collect(new ArrayList<>(group));
								group.clear();
							}
						}

						if (group.size() > 0) {
							out.collect(new ArrayList<>(group));
							group.clear();
						}
					}

				});

		return forEach(groupDataSet, onNext);
	}
}


<|start_filename|>src/io/iteratorx/reader/IOUtils.java<|end_filename|>
package io.iteratorx.reader;

import java.io.IOException;
import java.io.Reader;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.logging.Logger;

import com.google.common.base.Throwables;

public class IOUtils {
	private static final Logger logger = Logger.getLogger(IOUtils.class.getName());

	public static String getDriver(String url, final String driver) {
		if (driver != null) {
			return driver;
		}

		url = url.toLowerCase();
		if (url.startsWith("jdbc:postgresql")) {
			return "org.postgresql.Driver";
		} else if (url.startsWith("jdbc:mysql")) {
			return "com.mysql.jdbc.Driver";
		} else if (url.startsWith("jdbc:oracle")) {
			return "oracle.jdbc.OracleDriver";
		} else if (url.startsWith("jdbc:sqlserver")) {
			return "com.microsoft.sqlserver.jdbc.SQLServerDriver";
		} else if (url.startsWith("jdbc:db2")) {
			return "com.ibm.db2.jcc.DB2Driver";
		} else if (url.startsWith("jdbc:ignite")) {
			return "org.apache.ignite.IgniteJdbcThinDriver";
		} else if (url.startsWith("jdbc:derby")) {
			return "org.apache.derby.jdbc.EmbeddedDriver";
		} else if (url.startsWith("jdbc:sqlite")) {
			return "org.sqlite.JDBC";
		} else if (url.startsWith("jdbc:hive2")) {
			return "org.apache.hive.jdbc.HiveDriver";
		} else {
			throw new IllegalStateException();
		}
	}

	public static void close(final ResultSet rs, final Statement stmt, final Connection conn) {
		if (rs != null) {
			try {
				rs.close();
			} catch (final SQLException e) {
				logger.warning(Throwables.getStackTraceAsString(e));
			}
		}
		if (stmt != null) {
			try {
				stmt.close();
			} catch (final SQLException e) {
				logger.warning(Throwables.getStackTraceAsString(e));
			}
		}
		if (conn != null) {
			try {
				conn.close();
			} catch (final SQLException e) {
				logger.warning(Throwables.getStackTraceAsString(e));
			}
		}
	}

	public static void close(final Reader reader) {
		try {
			reader.close();
		} catch (final IOException e) {
			logger.warning(Throwables.getStackTraceAsString(e));
		}
	}
}


<|start_filename|>src/io/iteratorx/samples/ParallelsUsage.java<|end_filename|>
package io.iteratorx.samples;

import com.alibaba.fastjson.JSONObject;

import io.iteratorx.parallels.Flink;
import io.iteratorx.parallels.RxJava;
import io.iteratorx.parallels.Threads;
import io.iteratorx.reader.DataSourceBuilder.JdbcDataSourceBuilder;
import io.iteratorx.reader.JdbcReader;

public class ParallelsUsage {

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

		// create jdbc reader
		final JdbcReader jdbcReader = new JdbcReader(
				new JdbcDataSourceBuilder().setUrl("jdbc:postgresql://10.23.112.2:3333/dbname").setUser("username")
						.setPassword("password").build());

		// by Threads
		{
			// process each item parallelly using thread pool
			Threads.from(jdbcReader.read("select * from tablename")).forEach(item -> {
				System.err.println(item);
			});

			// process batch data parallelly
			Threads.from(jdbcReader.read("select * from tablename")).forBatch(items -> {
				for (final JSONObject item : items) {
					System.err.println(item);
				}
			});
		}

		// by Flink
		{
			// process each item parallelly using Flink engine
			Flink.from(jdbcReader.read("select * from tablename")).forEach(item -> {
				System.err.println(item);
			});

			// process batch data parallelly
			Flink.from(jdbcReader.read("select * from tablename")).forBatch(items -> {
				for (final JSONObject item : items) {
					System.err.println(item);
				}
			});

			// use DataSet directly to enable all Flink power
			Flink.from(jdbcReader.read("select * from tablename")).dataSet().distinct().count();
		}

		// by RxJava
		{
			// process each item parallely using RxJava engine
			RxJava.from(jdbcReader.read("select * from tablename")).forEach(item -> {
				System.err.println(item);
			});

			// process batch data parallely
			RxJava.from(jdbcReader.read("select * from tablename")).forBatch(items -> {
				for (final JSONObject item : items) {
					System.err.println(item);
				}
			});

			// use Observable directly
			RxJava.from(jdbcReader.read("select * from tablename")).observable().distinct().count();
		}
	}
}


<|start_filename|>src/io/iteratorx/reader/FileReader.java<|end_filename|>
package io.iteratorx.reader;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import com.alibaba.fastjson.JSONObject;

import io.iteratorx.AutoCloseableIterator;
import io.iteratorx.Function;

public class FileReader {

	public <T> List<T> readAll(final File file, final String charset, final Function<String, T> function) {
		final List<T> results = new ArrayList<>();
		for (final T one : read(file, charset, function)) {
			results.add(one);
		}
		return results;
	}

	/**
	 * Read each file content line into JSONObject iteratively.
	 * 
	 * NOTICE: thread-unsafe: do not use result Iterable in multi-threads!
	 * 
	 */
	public <T> Iterable<T> read(final File file, final String charset, final Function<String, T> function) {

		// open buffered reader
		final BufferedReader bufferedReader;
		{
			try {
				bufferedReader = new BufferedReader(new InputStreamReader(new FileInputStream(file), charset));
			} catch (final FileNotFoundException | UnsupportedEncodingException e) {
				throw new RuntimeException(e);
			}
		}

		// return iterator
		return new Iterable<T>() {

			@Override
			public Iterator<T> iterator() {
				return new AutoCloseableIterator<T>() {
					private static final long serialVersionUID = 5244479534484878112L;

					String nextData = doNext();

					@Override
					public boolean hasNext() {
						return nextData != null;
					}

					@Override
					public T next() {
						final String result = nextData;
						if (nextData != null) {
							nextData = doNext();
						}
						try {
							return function.apply(result);
						} catch (final Exception e) {
							return null;
						}
					}

					private String doNext() {
						try {
							final String lineContent = bufferedReader.readLine();
							if (lineContent == null) {
								close();
							}

							return lineContent;

						} catch (final IOException e) {
							close();

							throw new RuntimeException(e);
						}
					}

					@Override
					public void close() {
						IOUtils.close(bufferedReader);
						nextData = null;
					}
				};
			}
		};
	}

	public List<JSONObject> readAll(final File file, final String charset) {
		final List<JSONObject> results = new ArrayList<>();
		for (final JSONObject one : read(file, charset)) {
			results.add(one);
		}
		return results;
	}

	/**
	 * Read each file content line into JSONObject iteratively.
	 * 
	 * NOTICE: thread-unsafe: do not use result Iterable in multi-threads!
	 * 
	 */
	public Iterable<JSONObject> read(final File file, final String charset) {
		return read(file, charset, (s) -> {
			return JSONObject.parseObject(s);
		});
	}
}

<|start_filename|>src/io/iteratorx/parallels/RxJava.java<|end_filename|>
package io.iteratorx.parallels;

import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.logging.Logger;

import io.iteratorx.Consumer;
import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;

public class RxJava<T> implements Parallels {
	private static final Logger logger = Logger.getLogger(RxJava.class.getName());

	public static <T> void forEach(final Observable<T> observable, final Consumer<T> onNext) {
		observable.subscribe(t -> {
			onNext.accept(t);
		});

		logger.info("OK finished!");
	}

	public static <T> RxJava<T> from(final Iterator<T> iterator) throws Exception {
		return new RxJava<T>(iterator);
	}

	public static <T> RxJava<T> from(final Iterator<T> iterator, final int parallelism) throws Exception {
		return new RxJava<T>(iterator, DefaultParallelism);
	}

	public static <T> RxJava<T> from(final Iterable<T> iterable) throws Exception {
		return new RxJava<T>(iterable);
	}

	public static <T> RxJava<T> from(final Iterable<T> iterable, final int parallelism) throws Exception {
		return new RxJava<T>(iterable, DefaultParallelism);
	}

	private final Observable<T> observable;

	public RxJava(final Iterator<T> iterator) {
		this(iterator, DefaultParallelism);
	}

	public RxJava(final Iterator<T> iterator, final int parallelism) {
		this(new Iterable<T>() {

			@Override
			public Iterator<T> iterator() {
				return iterator;
			}
		}, parallelism);
	}

	public RxJava(final Iterable<T> iterable) {
		this(iterable, DefaultParallelism);
	}

	public RxJava(final Iterable<T> iterable, final int parallelism) {
		this.observable = Observable.fromIterable(iterable)
				.observeOn(Schedulers.from(Executors.newFixedThreadPool(parallelism)))
				.subscribeOn(Schedulers.from(Executors.newFixedThreadPool(parallelism)));
	}

	public Observable<T> observable() {
		return this.observable;
	}

	public void forEach(final Consumer<T> onNext) {
		forEach(this.observable, onNext);
	}

	public void forBatch(final Consumer<List<T>> onNext) {
		forBatch(DefaultBatchSize, onNext);
	}

	public void forBatch(final int batchSize, final Consumer<List<T>> onNext) {
		forEach(this.observable.buffer(batchSize), onNext);
	}

}


<|start_filename|>src/io/iteratorx/AutoCloseableIterator.java<|end_filename|>
package io.iteratorx;

import java.io.Serializable;
import java.util.Iterator;

public interface AutoCloseableIterator<T> extends Iterator<T>, AutoCloseable, Serializable {

}
