beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dhalp...@apache.org
Subject [1/2] incubator-beam git commit: Add NestedValueProvider
Date Tue, 01 Nov 2016 01:38:33 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/master 8bc68c59b -> b2fa59b36


Add NestedValueProvider


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/0594741e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/0594741e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/0594741e

Branch: refs/heads/master
Commit: 0594741e8e56f3194f8807eb4eabf9c309274113
Parents: 8bc68c5
Author: sammcveety <sam.mcveety@gmail.com>
Authored: Thu Oct 20 12:29:04 2016 -0400
Committer: Dan Halperin <dhalperi@google.com>
Committed: Mon Oct 31 18:37:58 2016 -0700

----------------------------------------------------------------------
 .../apache/beam/sdk/options/ValueProvider.java  | 36 +++++++++++++++++++-
 .../beam/sdk/options/ValueProviderTest.java     | 33 ++++++++++++++++++
 2 files changed, 68 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/0594741e/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ValueProvider.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ValueProvider.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ValueProvider.java
index 3f35a18..d0949ef 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ValueProvider.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ValueProvider.java
@@ -33,7 +33,6 @@ import com.fasterxml.jackson.databind.SerializerProvider;
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import com.fasterxml.jackson.databind.deser.ContextualDeserializer;
-
 import java.io.IOException;
 import java.io.Serializable;
 import java.lang.reflect.InvocationHandler;
@@ -41,6 +40,7 @@ import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
 import java.util.concurrent.ConcurrentHashMap;
 import javax.annotation.Nullable;
+import org.apache.beam.sdk.transforms.SerializableFunction;
 
 /**
  * {@link ValueProvider} is an interface which abstracts the notion of
@@ -94,6 +94,40 @@ public interface ValueProvider<T> {
   }
 
   /**
+   * {@link NestedValueProvider} is an implementation of {@link ValueProvider} that
+   * allows for wrapping another {@link ValueProvider} object.
+   */
+  static class NestedValueProvider<T, X> implements ValueProvider<T>, Serializable
{
+
+    private final ValueProvider<X> value;
+    private final SerializableFunction<X, T> translator;
+
+    NestedValueProvider(ValueProvider<X> value, SerializableFunction<X, T> translator)
{
+      this.value = checkNotNull(value);
+      this.translator = checkNotNull(translator);
+    }
+
+    /**
+     * Creates a {@link NestedValueProvider} that wraps the provided value.
+     */
+    public static <T, X> NestedValueProvider<T, X> of(
+        ValueProvider<X> value, SerializableFunction<X, T> translator) {
+      NestedValueProvider<T, X> factory = new NestedValueProvider<T, X>(value,
translator);
+      return factory;
+    }
+
+    @Override
+    public T get() {
+      return translator.apply(value.get());
+    }
+
+    @Override
+    public boolean isAccessible() {
+      return value.isAccessible();
+    }
+  }
+
+  /**
    * {@link RuntimeValueProvider} is an implementation of {@link ValueProvider} that
    * allows for a value to be provided at execution time rather than at graph
    * construction time.

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/0594741e/sdks/java/core/src/test/java/org/apache/beam/sdk/options/ValueProviderTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/ValueProviderTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/ValueProviderTest.java
index ed7a37a..660543c 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/ValueProviderTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/ValueProviderTest.java
@@ -25,8 +25,10 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import java.util.List;
+import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
 import org.apache.beam.sdk.options.ValueProvider.RuntimeValueProvider;
 import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider;
+import org.apache.beam.sdk.transforms.SerializableFunction;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
@@ -210,4 +212,35 @@ public class ValueProviderTest {
     assertTrue(vp.isAccessible());
     assertEquals("quux", vp.get());
   }
+
+  @Test
+  public void testNestedValueProviderStatic() throws Exception {
+    ValueProvider<String> svp = StaticValueProvider.of("foo");
+    ValueProvider<String> nvp = NestedValueProvider.of(
+      svp, new SerializableFunction<String, String>() {
+        @Override
+        public String apply(String from) {
+          return from + "bar";
+        }
+      });
+    assertTrue(nvp.isAccessible());
+    assertEquals("foobar", nvp.get());
+  }
+
+  @Test
+  public void testNestedValueProviderRuntime() throws Exception {
+    TestOptions options = PipelineOptionsFactory.as(TestOptions.class);
+    ValueProvider<String> rvp = options.getBar();
+    ValueProvider<String> nvp = NestedValueProvider.of(
+      rvp, new SerializableFunction<String, String>() {
+        @Override
+        public String apply(String from) {
+          return from + "bar";
+        }
+      });
+    assertFalse(nvp.isAccessible());
+    expectedException.expect(RuntimeException.class);
+    expectedException.expectMessage("Not called from a runtime context");
+    nvp.get();
+  }
 }


Mime
View raw message