beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From k...@apache.org
Subject [1/2] beam git commit: Add wrapping of lambda in a SimpleFunction
Date Tue, 07 Feb 2017 16:32:05 GMT
Repository: beam
Updated Branches:
  refs/heads/master ed7b82e7e -> 144bffd40


Add wrapping of lambda in a SimpleFunction


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

Branch: refs/heads/master
Commit: 23152178f81e635db65a7aae71f47fa67b3dc065
Parents: ed7b82e
Author: Kenneth Knowles <klk@google.com>
Authored: Thu Jan 26 11:19:42 2017 -0800
Committer: Kenneth Knowles <klk@google.com>
Committed: Tue Feb 7 08:16:39 2017 -0800

----------------------------------------------------------------------
 .../beam/sdk/transforms/SimpleFunction.java     | 44 +++++++++++--
 .../beam/sdk/transforms/SimpleFunctionTest.java | 43 ++++++++++++
 .../sdk/transforms/MapElementsJava8Test.java    | 24 ++++++-
 .../sdk/transforms/SimpleFunctionJava8Test.java | 69 ++++++++++++++++++++
 4 files changed, 170 insertions(+), 10 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/23152178/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/SimpleFunction.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/SimpleFunction.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/SimpleFunction.java
index 8604659..db44380 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/SimpleFunction.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/SimpleFunction.java
@@ -17,6 +17,8 @@
  */
 package org.apache.beam.sdk.transforms;
 
+import java.lang.reflect.Method;
+import javax.annotation.Nullable;
 import org.apache.beam.sdk.transforms.display.DisplayData;
 import org.apache.beam.sdk.transforms.display.HasDisplayData;
 import org.apache.beam.sdk.values.TypeDescriptor;
@@ -29,6 +31,40 @@ import org.apache.beam.sdk.values.TypeDescriptor;
 public abstract class SimpleFunction<InputT, OutputT>
     implements SerializableFunction<InputT, OutputT>, HasDisplayData {
 
+  @Nullable
+  private final SerializableFunction<InputT, OutputT> fn;
+
+  protected SimpleFunction() {
+    this.fn = null;
+    // A subclass must override apply if using this constructor. Check that via
+    // reflection.
+    try {
+      Method methodThatMustBeOverridden =
+          SimpleFunction.class.getDeclaredMethod("apply", new Class[] {Object.class});
+      Method methodOnSubclass =
+          getClass().getMethod("apply", new Class[] {Object.class});
+
+      if (methodOnSubclass.equals(methodThatMustBeOverridden)) {
+        throw new IllegalStateException(
+            "Subclass of SimpleFunction must override 'apply' method"
+                + " or pass a SerializableFunction to the constructor,"
+                + " usually via a lambda or method reference.");
+      }
+
+    } catch (NoSuchMethodException exc) {
+      throw new RuntimeException("Impossible state: missing 'apply' method entirely", exc);
+    }
+  }
+
+  protected SimpleFunction(SerializableFunction<InputT, OutputT> fn) {
+    this.fn = fn;
+  }
+
+  @Override
+  public OutputT apply(InputT input) {
+    return fn.apply(input);
+  }
+
   public static <InputT, OutputT>
       SimpleFunction<InputT, OutputT> fromSerializableFunctionWithOutputType(
           SerializableFunction<InputT, OutputT> fn, TypeDescriptor<OutputT> outputType)
{
@@ -77,23 +113,17 @@ public abstract class SimpleFunction<InputT, OutputT>
   private static class SimpleFunctionWithOutputType<InputT, OutputT>
       extends SimpleFunction<InputT, OutputT> {
 
-    private final SerializableFunction<InputT, OutputT> fn;
     private final TypeDescriptor<OutputT> outputType;
 
     public SimpleFunctionWithOutputType(
         SerializableFunction<InputT, OutputT> fn,
         TypeDescriptor<OutputT> outputType) {
-      this.fn = fn;
+      super(fn);
       this.outputType = outputType;
     }
 
 
     @Override
-    public OutputT apply(InputT input) {
-      return fn.apply(input);
-    }
-
-    @Override
     public TypeDescriptor<OutputT> getOutputTypeDescriptor() {
       return outputType;
     }

http://git-wip-us.apache.org/repos/asf/beam/blob/23152178/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SimpleFunctionTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SimpleFunctionTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SimpleFunctionTest.java
new file mode 100644
index 0000000..bcfb558
--- /dev/null
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SimpleFunctionTest.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Tests for {@link SimpleFunction}.
+ */
+@RunWith(JUnit4.class)
+public class SimpleFunctionTest {
+
+  @Rule
+  public ExpectedException thrown = ExpectedException.none();
+
+  @Test
+  public void testFailureIfNotOverridden() {
+    thrown.expect(IllegalStateException.class);
+    thrown.expectMessage("must override");
+    thrown.expectMessage("apply");
+
+    SimpleFunction<Integer, Integer> broken = new SimpleFunction<Integer, Integer>()
{};
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/23152178/sdks/java/java8tests/src/test/java/org/apache/beam/sdk/transforms/MapElementsJava8Test.java
----------------------------------------------------------------------
diff --git a/sdks/java/java8tests/src/test/java/org/apache/beam/sdk/transforms/MapElementsJava8Test.java
b/sdks/java/java8tests/src/test/java/org/apache/beam/sdk/transforms/MapElementsJava8Test.java
index ce0f111..7e63a7d 100644
--- a/sdks/java/java8tests/src/test/java/org/apache/beam/sdk/transforms/MapElementsJava8Test.java
+++ b/sdks/java/java8tests/src/test/java/org/apache/beam/sdk/transforms/MapElementsJava8Test.java
@@ -37,11 +37,11 @@ public class MapElementsJava8Test implements Serializable {
   public final transient TestPipeline pipeline = TestPipeline.create();
 
   /**
-   * Basic test of {@link MapElements} with a lambda (which is instantiated as a
-   * {@link SerializableFunction}).
+   * Basic test of {@link MapElements} with a lambda (which is instantiated as a {@link
+   * SerializableFunction}).
    */
   @Test
-  public void testMapBasic() throws Exception {
+  public void testMapLambda() throws Exception {
 
     PCollection<Integer> output = pipeline
         .apply(Create.of(1, 2, 3))
@@ -55,6 +55,24 @@ public class MapElementsJava8Test implements Serializable {
   }
 
   /**
+   * Basic test of {@link MapElements} with a lambda wrapped into a {@link SimpleFunction}
to
+   * remember its type.
+   */
+  @Test
+  public void testMapWrappedLambda() throws Exception {
+
+    PCollection<Integer> output =
+        pipeline
+            .apply(Create.of(1, 2, 3))
+            .apply(
+                MapElements
+                    .via(new SimpleFunction<Integer, Integer>((Integer i) -> i *
2) {}));
+
+    PAssert.that(output).containsInAnyOrder(6, 2, 4);
+    pipeline.run();
+  }
+
+  /**
    * Basic test of {@link MapElements} with a method reference.
    */
   @Test

http://git-wip-us.apache.org/repos/asf/beam/blob/23152178/sdks/java/java8tests/src/test/java/org/apache/beam/sdk/transforms/SimpleFunctionJava8Test.java
----------------------------------------------------------------------
diff --git a/sdks/java/java8tests/src/test/java/org/apache/beam/sdk/transforms/SimpleFunctionJava8Test.java
b/sdks/java/java8tests/src/test/java/org/apache/beam/sdk/transforms/SimpleFunctionJava8Test.java
new file mode 100644
index 0000000..9beab34
--- /dev/null
+++ b/sdks/java/java8tests/src/test/java/org/apache/beam/sdk/transforms/SimpleFunctionJava8Test.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+import java.io.Serializable;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Java 8 tests for {@link SimpleFunction}.
+ */
+@RunWith(JUnit4.class)
+public class SimpleFunctionJava8Test implements Serializable {
+
+  @Rule
+  public final transient TestPipeline pipeline = TestPipeline.create();
+
+  /**
+   * Basic test of {@link MapElements} with a lambda (which is instantiated as a {@link
+   * SerializableFunction}).
+   */
+  @Test
+  public void testGoodTypeForLambda() throws Exception {
+    SimpleFunction<Integer, String> fn =
+        new SimpleFunction<Integer, String>((Integer i) -> i.toString()) {};
+
+    assertThat(fn.getInputTypeDescriptor(), equalTo(TypeDescriptors.integers()));
+    assertThat(fn.getOutputTypeDescriptor(), equalTo(TypeDescriptors.strings()));
+  }
+
+  /**
+   * Basic test of {@link MapElements} with a lambda wrapped into a {@link SimpleFunction}
to
+   * remember its type.
+   */
+  @Test
+  public void testGoodTypeForMethodRef() throws Exception {
+    SimpleFunction<Integer, String> fn =
+        new SimpleFunction<Integer, String>(SimpleFunctionJava8Test::toStringThisThing)
{};
+
+    assertThat(fn.getInputTypeDescriptor(), equalTo(TypeDescriptors.integers()));
+    assertThat(fn.getOutputTypeDescriptor(), equalTo(TypeDescriptors.strings()));
+  }
+
+  private static String toStringThisThing(Integer i) {
+    return i.toString();
+  }
+}


Mime
View raw message