ISPN-6300 Add missing bindings to dist remote exec

* Added cacheManager and context parameters as bindings for distributed
  remote task execution.
* Marshaller not currently passed since the marshaller in the context
  can't easily be initialised in the distributed tasks.
* Added tests to verify that bindings are correctly set for normal
  remote execution and typed remote execution.

client/hotrod-client/src/test/java/org/infinispan/client/hotrod/ExecTest.java

@@ -9,16 +9,19 @@
 import org.infinispan.scripting.ScriptingManager;
 import org.infinispan.scripting.utils.ScriptingUtils;
 import org.infinispan.test.TestingUtil;
+import org.testng.AssertJUnit;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.function.BiConsumer;
 
 import static org.infinispan.client.hotrod.test.HotRodClientTestingUtil.loadScript;
 import static org.infinispan.client.hotrod.test.HotRodClientTestingUtil.withScript;
@@ -169,9 +172,40 @@ public void testExecReturnNull(String cacheName) throws IOException {
       });
    }
 
+   @Test(dataProvider = "CacheNameProvider")
+   public void testLocalExecPutGet(String cacheName) {
+      execPutGet(cacheName, "/test-put-get.js", ExecMode.LOCAL, "local-key", "local-value");
+   }
+
+   @Test(dataProvider = "CacheNameProvider")
+   public void testDistExecPutGet(String cacheName) {
+      execPutGet(cacheName, "/test-put-get-dist.js", ExecMode.DIST, "dist-key", "dist-value");
+   }
+
+   private void execPutGet(String cacheName, String path, ExecMode mode, String key, String value) {
+      withScript(manager(0), path, scriptName -> {
+         Map<String, String> params = new HashMap<>();
+         params.put("k", key);
+         params.put("v", value);
+         Object results =  clients.get(0).getCache(cacheName).execute(scriptName, params);
+         mode.assertResult.accept(value, results);
+      });
+   }
+
    @DataProvider(name = "CacheNameProvider")
    private static Object[][] provideCacheMode() {
       return new Object[][] {{REPL_CACHE}, {DIST_CACHE}};
    }
 
+   enum ExecMode {
+      LOCAL(AssertJUnit::assertEquals),
+      DIST((v, r) -> assertEquals(Arrays.asList(v, v), r));
+
+      final BiConsumer<String, Object> assertResult;
+
+      ExecMode(BiConsumer<String, Object> assertResult) {
+         this.assertResult = assertResult;
+      }
+   }
+
 }

client/hotrod-client/src/test/java/org/infinispan/client/hotrod/ExecTypedTest.java

@@ -6,11 +6,13 @@
 import org.infinispan.commons.marshall.StringMarshaller;
 import org.infinispan.configuration.cache.CacheMode;
 import org.infinispan.configuration.cache.ConfigurationBuilder;
+import org.testng.AssertJUnit;
 import org.testng.annotations.Test;
 
 import java.nio.charset.Charset;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.function.BiConsumer;
 
 import static org.infinispan.client.hotrod.test.HotRodClientTestingUtil.withScript;
 import static org.testng.AssertJUnit.assertEquals;
@@ -53,27 +55,15 @@ private RemoteCacheManager createAddScriptClient() {
       return new InternalRemoteCacheManager(clientBuilder.build());
    }
 
-   public void testRemoteTypedScriptPutGetExecute() throws Exception {
-      withScript(addScriptClient.getCache(SCRIPT_CACHE), "/typed-put-get.js", scriptName -> {
-         Map<String, String> params = new HashMap<>();
-         params.put("k", "typed-key");
-         params.put("v", "typed-value");
-         String result = execClient.getCache().execute(scriptName, params);
-         assertEquals("typed-value", result);
-      });
+   public void testLocalTypedExecPutGet() {
+      execPutGet("/typed-put-get.js", ExecMode.LOCAL, "local-typed-key", "local-typed-value");
    }
 
-   public void testRemoteTypedScriptPutGetExecuteCyrillic() throws Exception {
-      withScript(addScriptClient.getCache(SCRIPT_CACHE), "/typed-put-get.js", scriptName -> {
-         Map<String, String> params = new HashMap<>();
-         params.put("k", "բարև");
-         params.put("v", "привет");
-         String result = execClient.getCache().execute(scriptName, params);
-         assertEquals("привет", result);
-      });
+   public void testLocalTypedExecPutGetCyrillic() {
+      execPutGet("/typed-put-get.js", ExecMode.LOCAL, "բարև", "привет");
    }
 
-   public void testPutGetEmptyString() throws Exception {
+   public void testLocalTypedExecPutGetEmptyString() {
       withScript(addScriptClient.getCache(SCRIPT_CACHE), "/typed-put-get.js", scriptName -> {
          Map<String, String> params = new HashMap<>();
          params.put("k", "empty-key");
@@ -83,26 +73,51 @@ public void testPutGetEmptyString() throws Exception {
       });
    }
 
-   public void testRemoteTypedScriptSizeExecute() throws Exception {
+   public void testLocalTypedExecSize() {
       withScript(addScriptClient.getCache(SCRIPT_CACHE), "/typed-size.js", scriptName -> {
          clients.get(0).getCache().clear();
          String result = execClient.getCache().execute(scriptName, new HashMap<>());
          assertEquals("0", result);
       });
    }
 
-   public void testRemoteTypedCacheManagerScriptExecute() throws Exception {
+   public void testLocalTypedExecWithCacheManager() {
       withScript(addScriptClient.getCache(SCRIPT_CACHE), "/typed-cachemanager-put-get.js", scriptName -> {
          String result = execClient.getCache().execute(scriptName, new HashMap<>());
          assertEquals("a", result);
       });
    }
 
-   public void testRemoteTypedScriptNullReturnExecute() throws Exception {
+   public void testLocalTypedExecNullReturn() {
       withScript(addScriptClient.getCache(SCRIPT_CACHE), "/typed-null-return.js", scriptName -> {
          String result = clients.get(0).getCache().execute(scriptName, new HashMap<>());
          assertEquals(null, result);
       });
    }
 
+   public void testDistTypedExecPutGet() {
+      execPutGet("/typed-put-get-dist.js", ExecMode.DIST, "dist-typed-key", "dist-typed-value");
+   }
+
+   private void execPutGet(String path, ExecMode mode, String key, String value) {
+      withScript(addScriptClient.getCache(SCRIPT_CACHE), path, scriptName -> {
+         Map<String, String> params = new HashMap<>();
+         params.put("k", key);
+         params.put("v", value);
+         String result = execClient.getCache().execute(scriptName, params);
+         mode.assertResult.accept(value, result);
+      });
+   }
+
+   enum ExecMode {
+      LOCAL(AssertJUnit::assertEquals),
+      DIST((v, r) -> assertEquals(String.format("[\"%1$s\", \"%1$s\"]", v), r));
+
+      final BiConsumer<String, String> assertResult;
+
+      ExecMode(BiConsumer<String, String> assertResult) {
+         this.assertResult = assertResult;
+      }
+   }
+
 }

client/hotrod-client/src/test/resources/typed-put-get-dist.js

@@ -0,0 +1,4 @@
+// mode=distributed,language=javascript,parameters=[k, v],datatype='text/plain; charset=utf-8'
+var cache = cacheManager.getCache();
+cache.put(k, v);
+cache.get(k);

scripting/src/main/java/org/infinispan/scripting/impl/DataType.java

@@ -5,6 +5,7 @@
 import org.infinispan.commons.util.Immutables;
 
 import java.nio.charset.Charset;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -47,11 +48,15 @@ static final class Utf8Transformer implements Transformer {
       }
 
       @Override
+      @SuppressWarnings("unchecked")
       public Object fromDataType(Object obj, Optional<Marshaller> marshaller) {
-         return Objects.isNull(obj)
-               ? null : obj instanceof String
-               ? ((String) obj).getBytes(CHARSET_UTF8)
-               : obj.toString().getBytes(CHARSET_UTF8);
+         if (obj instanceof List)
+            return ((List) obj).stream()
+                  .map(Object::toString)
+                  .collect(Collectors.joining("\", \"", "[\"", "\"]"))
+                  .toString().getBytes(CHARSET_UTF8);
+
+         return Objects.isNull(obj) ? null : obj.toString().getBytes(CHARSET_UTF8);
       }
 
       private static String asString(Object v) {

scripting/src/main/java/org/infinispan/scripting/impl/DistributedRunner.java

@@ -1,10 +1,10 @@
 package org.infinispan.scripting.impl;
 
 import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Future;
 
 import org.infinispan.Cache;
 import org.infinispan.distexec.DefaultExecutorService;
@@ -33,11 +33,19 @@ public <T> CompletableFuture<T> runScript(ScriptingManagerImpl scriptManager, Sc
       }
       DefaultExecutorService des = new DefaultExecutorService(masterCacheNode);
       try {
-         List<CompletableFuture<T>> tasks = des.submitEverywhere(new DistributedScript<T>(metadata));
+         Map<String, Object> ctxParams = extractContextParams(metadata, binding);
+         List<CompletableFuture<T>> tasks = des.submitEverywhere(new DistributedScript<T>(metadata, ctxParams));
 
          return (CompletableFuture<T>) CompletableFutures.sequence(tasks);
       } finally {
          des.shutdown();
       }
    }
+
+   private Map<String, Object> extractContextParams(ScriptMetadata metadata, CacheScriptBindings binding) {
+      Map<String, Object> params = new HashMap<>();
+      metadata.parameters().stream().forEach(paramName -> params.put(paramName, binding.get(paramName)));
+      return params;
+   }
+
 }

scripting/src/main/java/org/infinispan/scripting/impl/DistributedScript.java

@@ -1,6 +1,7 @@
 package org.infinispan.scripting.impl;
 
 import java.io.Serializable;
+import java.util.Map;
 import java.util.Set;
 
 import javax.script.Bindings;
@@ -18,11 +19,13 @@
  */
 class DistributedScript<T> implements DistributedCallable<Object, Object, T>, Serializable {
    private final ScriptMetadata metadata;
+   private final Map<String, ?> ctxParams;
    private transient ScriptingManagerImpl scriptManager;
    private transient Bindings bindings;
 
-   DistributedScript(ScriptMetadata metadata) {
+   DistributedScript(ScriptMetadata metadata, Map<String, ?> ctxParams) {
       this.metadata = metadata;
+      this.ctxParams = ctxParams;
    }
 
    @Override
@@ -36,5 +39,7 @@ public void setEnvironment(Cache<Object, Object> cache, Set<Object> inputKeys) {
       bindings = new SimpleBindings();
       bindings.put("inputKeys", inputKeys);
       bindings.put("cache", cache);
+      bindings.put("cacheManager", cache.getCacheManager());
+      ctxParams.entrySet().stream().forEach(e -> bindings.put(e.getKey(), e.getValue()));
    }
 }

scripting/src/test/resources/test-put-get-dist.js

@@ -0,0 +1,4 @@
+// mode=distributed,language=javascript,parameters=[k, v]
+var cache = cacheManager.getCache();
+cache.put(k, v);
+cache.get(k);

scripting/src/test/resources/test-put-get.js

@@ -0,0 +1,4 @@
+// mode=local,language=javascript,parameters=[k, v]
+var cache = cacheManager.getCache();
+cache.put(k, v);
+cache.get(k);