Added @NotNull sproc parameter validation.

Author: Pedro Ribeiro

database/test/21_shard1/21_schema1/04_tables/30_basic_table.sql

@@ -0,0 +1,6 @@
+CREATE TABLE ztest_schema1.basic_table (
+ bt_id serial primary key,
+ bt_key text,
+ bt_value text );
+
+insert into ztest_schema1.basic_table ( bt_key, bt_value ) select 'key1','value1' union all select 'key2','value2';

database/test/21_shard1/21_schema1/05_stored_procedures/20_get_empty_list.sql

@@ -0,0 +1,10 @@
+CREATE FUNCTION get_empty_list(p_objs example_domain_object_with_inner_object[]) RETURNS setof example_domain_object_with_inner_object AS
+$$
+BEGIN
+  IF array_length(p_objs, 1) <> 0 THEN
+    RAISE EXCEPTION 'Array length is not 0';
+  END IF;
+
+  RETURN;
+END;
+$$ LANGUAGE plpgsql SECURITY DEFINER;

database/test/21_shard1/21_schema1/05_stored_procedures/20_test_validations.sql

@@ -70,4 +70,16 @@ BEGIN
   a.a := null;
   return a;
 END;
-$$ LANGUAGE plpgsql SECURITY DEFINER;
+$$ LANGUAGE plpgsql SECURITY DEFINER;
+
+CREATE OR REPLACE FUNCTION test_sproc_call_with_multiple_parameters_validation (
+    a example_domain_object_with_validation,
+    parameter0 text,
+    parameter1 text,
+    parameter2 text)
+RETURNS example_domain_object_with_validation AS
+$$
+BEGIN
+  return a;
+END;
+$$ LANGUAGE plpgsql SECURITY DEFINER;

pom.xml

@@ -4,7 +4,7 @@
 
     <groupId>de.zalando</groupId>
     <artifactId>zalando-sprocwrapper</artifactId>
-    <version>1.0.5</version>
+    <version>1.0.6</version>
     <packaging>jar</packaging>
     <name>Stored Procedure Wrapper</name>
     <description>Library to make PostgreSQL stored procedures available through simple Java "*SProcService" interfaces including automatic object serialization and deserialization (using typemapper and convention-over-configuration). Supports sharding, advisory locking, statement timeouts and PostgreSQL types such as enums and hstore.</description>

src/main/java/de/zalando/sprocwrapper/proxy/InvocationContext.java

@@ -0,0 +1,46 @@
+package de.zalando.sprocwrapper.proxy;
+
+import java.lang.reflect.Method;
+
+import java.util.Arrays;
+
+import com.google.common.base.Preconditions;
+
+public class InvocationContext {
+
+    private final Object proxy;
+    private final Method method;
+    private final Object[] args;
+
+    public InvocationContext(final Object proxy, final Method method, final Object[] args) {
+        this.proxy = Preconditions.checkNotNull(proxy, "proxy");
+        this.method = Preconditions.checkNotNull(method, "method");
+        this.args = args;
+    }
+
+    public Object getProxy() {
+        return proxy;
+    }
+
+    public Method getMethod() {
+        return method;
+    }
+
+    public Object[] getArgs() {
+        return args;
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder builder = new StringBuilder();
+        builder.append("InvocationContext [proxy=");
+        builder.append(proxy);
+        builder.append(", method=");
+        builder.append(method);
+        builder.append(", args=");
+        builder.append(Arrays.toString(args));
+        builder.append("]");
+        return builder.toString();
+    }
+
+}

src/main/java/de/zalando/sprocwrapper/proxy/SProcProxy.java

@@ -45,6 +45,6 @@ public Object invoke(final Object proxy, final Method m, final Object[] args) {
             return null;
         }
 
-        return p.execute(dataSourceProvider, args);
+        return p.execute(dataSourceProvider, new InvocationContext(proxy, m, args));
     }
 }

src/main/java/de/zalando/sprocwrapper/proxy/StoredProcedure.java

@@ -339,6 +339,10 @@ private Map<Integer, Object[]> partitionArguments(final DataSourceProvider dataS
 
         // TODO: currently only implemented for single shardKey argument as first argument!
         final List<Object> originalArgument = (List<Object>) args[0];
+        if (originalArgument == null || originalArgument.isEmpty()) {
+            throw new IllegalArgumentException("ShardKey (first argument) of sproc '" + name + "' not defined");
+        }
+
         List<Object> partitionedArgument = null;
         Object[] partitionedArguments = null;
         int shardId;
@@ -383,28 +387,27 @@ private static class Call implements Callable<Object> {
         private final StoredProcedure sproc;
         private final DataSource shardDs;
         private final Object[] params;
-        private final Object[] originalArgs;
+        private final InvocationContext invocation;
 
         public Call(final StoredProcedure sproc, final DataSource shardDs, final Object[] params,
-                final Object[] originalArgs) {
+                final InvocationContext invocation) {
             this.sproc = sproc;
             this.shardDs = shardDs;
             this.params = params;
-            this.originalArgs = originalArgs;
+            this.invocation = invocation;
         }
 
         @Override
         public Object call() throws Exception {
-            return sproc.executor.executeSProc(shardDs, sproc.getQuery(), params, sproc.getTypes(), originalArgs,
+            return sproc.executor.executeSProc(shardDs, sproc.getQuery(), params, sproc.getTypes(), invocation,
                     sproc.returnType);
         }
 
     }
 
     private static ExecutorService parallelThreadPool = Executors.newCachedThreadPool();
 
-    @SuppressWarnings({ "unchecked", "rawtypes" })
-    public Object execute(final DataSourceProvider dp, final Object[] args) {
+    public Object execute(final DataSourceProvider dp, final InvocationContext invocation) {
 
         List<Integer> shardIds = null;
         Map<Integer, Object[]> partitionedArguments = null;
@@ -413,17 +416,17 @@ public Object execute(final DataSourceProvider dp, final Object[] args) {
             shardIds = dp.getDistinctShardIds();
         } else {
             if (autoPartition) {
-                partitionedArguments = partitionArguments(dp, args);
+                partitionedArguments = partitionArguments(dp, invocation.getArgs());
                 shardIds = Lists.newArrayList(partitionedArguments.keySet());
             } else {
-                shardIds = Lists.newArrayList(getShardId(args));
+                shardIds = Lists.newArrayList(getShardId(invocation.getArgs()));
             }
         }
 
         if (partitionedArguments == null) {
             partitionedArguments = Maps.newHashMap();
             for (final int shardId : shardIds) {
-                partitionedArguments.put(shardId, args);
+                partitionedArguments.put(shardId, invocation.getArgs());
             }
         }
 
@@ -434,7 +437,7 @@ public Object execute(final DataSourceProvider dp, final Object[] args) {
 
         } catch (final SQLException e) {
             throw new CannotGetJdbcConnectionException("Failed to acquire connection for virtual shard "
-                    + shardIds.get(0) + " translates to" + dp.getDataSourceId(shardIds.get(0)) + " for " + name, e);
+                    + shardIds.get(0) + " for " + name, e);
         }
 
         final List<Object[]> paramValues = Lists.newArrayList();
@@ -459,7 +462,7 @@ public Object execute(final DataSourceProvider dp, final Object[] args) {
             }
 
             // most common case: only one shard and no argument partitioning
-            return executor.executeSProc(firstDs, getQuery(), paramValues.get(0), getTypes(), args, returnType);
+            return executor.executeSProc(firstDs, getQuery(), paramValues.get(0), getTypes(), invocation, returnType);
         } else {
             Map<Integer, SameConnectionDatasource> transactionalDatasources = null;
             try {
@@ -471,11 +474,11 @@ public Object execute(final DataSourceProvider dp, final Object[] args) {
                 Object sprocResult = null;
                 final long start = System.currentTimeMillis();
                 if (parallel) {
-                    sprocResult = executeInParallel(dp, args, shardIds, paramValues, transactionalDatasources, results,
-                            sprocResult);
+                    sprocResult = executeInParallel(dp, invocation, shardIds, paramValues, transactionalDatasources,
+                            results, sprocResult);
                 } else {
-                    sprocResult = executeSequential(dp, args, shardIds, paramValues, transactionalDatasources, results,
-                            sprocResult);
+                    sprocResult = executeSequential(dp, invocation, shardIds, paramValues, transactionalDatasources,
+                            results, sprocResult);
                 }
 
                 if (LOG.isTraceEnabled()) {
@@ -528,9 +531,10 @@ public Object execute(final DataSourceProvider dp, final Object[] args) {
     }
 
     @SuppressWarnings({ "rawtypes", "unchecked" })
-    private Object executeSequential(final DataSourceProvider dp, final Object[] args, final List<Integer> shardIds,
-            final List<Object[]> paramValues, final Map<Integer, SameConnectionDatasource> transactionalDatasources,
-            final List<?> results, Object sprocResult) {
+    private Object executeSequential(final DataSourceProvider dp, final InvocationContext invocation,
+            final List<Integer> shardIds, final List<Object[]> paramValues,
+            final Map<Integer, SameConnectionDatasource> transactionalDatasources, final List<?> results,
+            Object sprocResult) {
         DataSource shardDs;
         int i = 0;
         final List<String> exceptions = Lists.newArrayList();
@@ -543,7 +547,7 @@ private Object executeSequential(final DataSourceProvider dp, final Object[] arg
 
             sprocResult = null;
             try {
-                sprocResult = executor.executeSProc(shardDs, getQuery(), paramValues.get(i), getTypes(), args,
+                sprocResult = executor.executeSProc(shardDs, getQuery(), paramValues.get(i), getTypes(), invocation,
                         returnType);
             } catch (final Exception e) {
 
@@ -568,9 +572,10 @@ private Object executeSequential(final DataSourceProvider dp, final Object[] arg
     }
 
     @SuppressWarnings({ "rawtypes", "unchecked" })
-    private Object executeInParallel(final DataSourceProvider dp, final Object[] args, final List<Integer> shardIds,
-            final List<Object[]> paramValues, final Map<Integer, SameConnectionDatasource> transactionalDatasources,
-            final List<?> results, Object sprocResult) {
+    private Object executeInParallel(final DataSourceProvider dp, final InvocationContext invocation,
+            final List<Integer> shardIds, final List<Object[]> paramValues,
+            final Map<Integer, SameConnectionDatasource> transactionalDatasources, final List<?> results,
+            Object sprocResult) {
         DataSource shardDs;
         final Map<Integer, FutureTask<Object>> tasks = Maps.newHashMapWithExpectedSize(shardIds.size());
         FutureTask<Object> task;
@@ -582,7 +587,7 @@ private Object executeInParallel(final DataSourceProvider dp, final Object[] arg
                 LOG.debug(getDebugLog(paramValues.get(i)));
             }
 
-            task = new FutureTask<Object>(new Call(this, shardDs, paramValues.get(i), args));
+            task = new FutureTask<Object>(new Call(this, shardDs, paramValues.get(i), invocation));
             tasks.put(shardId, task);
             parallelThreadPool.execute(task);
             i++;

src/main/java/de/zalando/sprocwrapper/proxy/executors/Executor.java

@@ -2,9 +2,12 @@
 
 import javax.sql.DataSource;
 
+import de.zalando.sprocwrapper.proxy.InvocationContext;
+
 /**
  * @author  jmussler
  */
 public interface Executor {
-    Object executeSProc(DataSource ds, String sql, Object[] args, int[] types, Object[] originalArgs, Class returnType);
+    Object executeSProc(DataSource ds, String sql, Object[] args, int[] types, InvocationContext invocationContext,
+            Class<?> returnType);
 }

src/main/java/de/zalando/sprocwrapper/proxy/executors/ExecutorWrapper.java

@@ -12,6 +12,7 @@
 
 import de.zalando.sprocwrapper.SProcCall.AdvisoryLock;
 import de.zalando.sprocwrapper.dsprovider.SameConnectionDatasource;
+import de.zalando.sprocwrapper.proxy.InvocationContext;
 
 /**
  * This Executor wraps stored procedure calls that use advisory locks and / or need different statement timeouts set.
@@ -96,7 +97,7 @@ private boolean unlockAdvisoryLock(final Connection conn) throws SQLException {
 
     @Override
     public Object executeSProc(final DataSource ds, final String sql, final Object[] args, final int[] types,
-            final Object[] originalArgs, final Class returnType) {
+            final InvocationContext invocationContext, final Class<?> returnType) {
 
         SameConnectionDatasource sameConnDs = null;
 
@@ -110,7 +111,7 @@ public Object executeSProc(final DataSource ds, final String sql, final Object[]
                 throw new RuntimeException("Could not acquire AdvisoryLock " + lock.name());
             }
 
-            return executor.executeSProc(sameConnDs, sql, args, types, originalArgs, returnType);
+            return executor.executeSProc(sameConnDs, sql, args, types, invocationContext, returnType);
 
         } catch (final SQLException e) {
 

src/main/java/de/zalando/sprocwrapper/proxy/executors/GlobalTransformerExecutorWrapper.java

@@ -10,6 +10,8 @@
 
 import com.google.common.collect.Lists;
 
+import de.zalando.sprocwrapper.proxy.InvocationContext;
+
 import de.zalando.typemapper.core.ValueTransformer;
 import de.zalando.typemapper.core.fieldMapper.GlobalValueTransformerRegistry;
 
@@ -31,8 +33,8 @@ public GlobalTransformerExecutorWrapper(final Executor e) {
     @SuppressWarnings({ "rawtypes", "unchecked" })
     @Override
     public Object executeSProc(final DataSource ds, final String sql, final Object[] args, final int[] types,
-            final Object[] originalArgs, final Class returnType) {
-        final Object result = originalExecutor.executeSProc(ds, sql, args, types, originalArgs, String.class);
+            final InvocationContext invocationContext, final Class<?> returnType) {
+        final Object result = originalExecutor.executeSProc(ds, sql, args, types, invocationContext, String.class);
         if (result == null) {
             return null;
         }