Java 8: JDBC ResultSet to Stream

Streams API is a gift.
But JDBC is still the old-school one.
Let's wrap it!

package org.example.jdbc.stream;

import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

public final class StreamHelper {
    public static class Record {
        private final Map<String, Object> fields = new HashMap<>(16);
        private final long count;

        private Record(final ResultSet resultSet) throws SQLException {
            final ResultSetMetaData metaData = resultSet.getMetaData();
            count = metaData.getColumnCount();
            for (int i = 1; i <= count; ++i) {
                fields.put(metaData.getColumnName(i), resultSet.getObject(i));
            }
        }

        /**
         * Is there a column named like this?
         *
         * @param columnName is the column name in the query.
         * @return True if found.
         */
        public boolean contains(final String columnName) {
            return fields.containsKey(columnName);
        }

        /**
         * Number of columns.
         *
         * @return Numer of columns.
         */
        public long count() {
            return count;
        }

        /**
         * Get value casted to the requested type.
         * <p>
         * No type checking happens inside. It is your job to know the datatype in the database.
         * <p>
         * Example:<br>
         * {@code record.get("COLUMN1", Long.class); // returns a Long}
         *
         * @param columnName is the column name in the query.
         * @param type is Java type of the column.
         * @return The value casted to the Java type.
         */
        public <T> T get(final String columnName, final Class<T> type) {
            return type.cast(getObject(columnName));
        }

        /**
         * Get columns in the record.
         *
         * @return Collection of the column names.
         */
        public Set<String> getColumns() {
            return Collections.unmodifiableSet(fields.keySet());
        }

        /**
         * Get value as an object.
         *
         * @param columnName is the column name in the query.
         * @return The value.
         */
        public Object getObject(final String columnName) {
            return fields.get(columnName);
        }

        /**
         * Get value as string.
         *
         * @param columnName is the column name in the query.
         * @return Value as string.
         */
        public String getString(final String columnName) {
            return Objects.toString(fields.get(columnName));
        }

        /**
         * Is the given cell null?
         *
         * @param columnName is the column name in the query.
         * @return True if null.
         */
        public boolean isNull(final String columnName) {
            return getObject(columnName) == null;
        }

        @Override
        public String toString() {
            return fields.entrySet().stream().map(e -> e.getKey() + ": " + e.getValue())
                    .collect(Collectors.joining(", "));
        }
    }

    /**
     * Wrap a ResultSet in a Stream.
     * <p>
     * The wrapper consumes the result set. The caller must close the result set after the stream
     * processing was finished.
     *
     * @param resultSet is the open result set to streamline.
     * @return A stream of rows.
     */
    public static Stream<Record> asStream(final ResultSet resultSet) {
        // "est = Long.MAX_VALUE if infinite, unknown, or too expensive to compute."
        return StreamSupport.stream(new Spliterators.AbstractSpliterator<Record>(Long.MAX_VALUE,
                Spliterator.NONNULL | Spliterator.IMMUTABLE) {
            @Override
            public boolean tryAdvance(final Consumer<? super Record> action) {
                try {
                    if (!resultSet.next()) {
                        return false;
                    }
                } catch (@SuppressWarnings("unused") final SQLException e) {
                    return false;
                }
                try {
                    action.accept(new Record(resultSet));
                } catch (@SuppressWarnings("unused") final SQLException e) {
                    return false;
                }
                return true;
            }
        }, true).parallel();
    }

    private StreamHelper() { /* Hidden. */ }
}

Usage example:

import org.example.jdbc.stream.StreamHelper;
import org.example.jdbc.stream.StreamHelper.Record;
// ...
    @GET
    @Path("test")
    public Response test() {
        try (Connection connection = Database.connect();
                Statement statement = connection.createStatement();
                ResultSet resultSet = statement.executeQuery("SELECT * FROM MY_TABLE")) {
            return Response.ok(StreamHelper.asStream(resultSet).map(Record::toString)
                    .collect(Collectors.joining(", "))).build();
        } catch (final SQLException | NamingException e) {
            return Response.serverError().entity(e.toString()).build();
        }
    }

bash: homemade timeout replacement

So I was young and reckless, and didn't know there is a command called timeout in coreutils. This is how I managed to do it:

#!/bin/sh
{
    ./time_consuming_binary -a param -a notherparam --pleaseblock
} &
CHILDPID=$!
# Kill it after 30 sec
sleep 30
kill -9 $CHILDPID 2>&1 /dev/null

How to check if it had to be killed or not? Measure the wall time of the execution :) I've used it in a Nagios service check, the thing it watched either returned under 5 sec or blocked indefinitely (thanks, NFS), hence the 30 secs.

Java: Running any type of Callable with any kind of constructors

Java 1.7

import java.lang.reflect.InvocationTargetException;
import java.util.concurrent.*;

public class ParallelCaller {

    public static final int NUM_CORES = 2 * Runtime.getRuntime().availableProcessors();
    private ExecutorService eservice = Executors.newFixedThreadPool(NUM_CORES);
    ;
    private CompletionService<Object> cservice = new ExecutorCompletionService<>(eservice);

    public void Parallel(int numtasks, int iterations, Class clazz, Object... args) throws NoSuchMethodException, InstantiationException, IllegalAccessException, IllegalArgumentException, InvocationTargetException {
        long begTest = new java.util.Date().getTime();

        for (int index = 1; index <= numtasks; index++) {
            cservice.submit((Callable<Object>) clazz.getConstructors()[0].newInstance(args));
        }

        eservice.shutdown();

        Object taskResult;
        for (int index = 0; index < numtasks; index++) {
            try {
                taskResult = cservice.take().get();
                System.out.println("result " + taskResult);
            } catch (ExecutionException e) {
                System.out.println(e.getMessage());
            } catch (InterruptedException e) {
                System.out.println(e.getMessage());
                // (Re-)Cancel if current thread also interrupted
                eservice.shutdownNow();
                // Preserve interrupt status
                Thread.currentThread().interrupt();
            }
        }

        Double secs = new Double((new java.util.Date().getTime() - begTest) * 0.001);
        System.out.println("run time " + secs + " secs");
    }
}

Usage:

public static void main(String[] args) throws NoSuchMethodException, InstantiationException, IllegalAccessException, IllegalArgumentException, InvocationTargetException {
        int parallel = args.length > 0 ? Integer.parseInt(args[0]) : 5;
        int iterations = args.length > 1 ? Integer.parseInt(args[1]) : 100;

        new ParallelCaller().Parallel(parallel, iterations, LoginTest.class, "http://somethingtotest.com", "testuser02", "testuser02", iterations, LoginTest.KnownDrivers.Firefox);
    }

The referred class should implement Callable.

Java: Parallel.For

This piece is from StackOverflow:

public class Parallel {
    private static final int NUM_CORES = Runtime.getRuntime().availableProcessors();

    private static final ExecutorService forPool = Executors.newFixedThreadPool(NUM_CORES  * 2);

    public static void For(final Iterable<T> elements, final Operation<T> operation) {
        try {
            // invokeAll blocks for us until all submitted tasks in the call complete
            forPool.invokeAll(createCallables(elements, operation));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static Collection<Callable<Void>> createCallables(final Iterable<T> elements, final Operation<T> operation) {
        List<Callable<Void>> callables = new LinkedList<Callable<Void>>();
        for (final T elem : elements) {
            callables.add(new Callable<Void>() {
                @Override
                public Void call() {
                    operation.perform(elem);
                    return null;
                }
            });
        }

        return callables;
    }

    public static interface Operation<T> {
        public void perform(T pParameter);
    }
}

Usage:

// Collection of items to process in parallel
Collection<Integer> elems = new LinkedList<Integer>();
for (int i = 0; i < 40; ++i) {
    elems.add(i);
}
Parallel.For(elems, 
 // The operation to perform with each item
 new Parallel.Operation<Integer>() {
    public void perform(Integer param) {
        System.out.println(param);
    };
});