[beam] branch master updated: Add named variant of PTransform#compose()

[sc...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

scott pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new 65c8d51  Add named variant of PTransform#compose()
     new 1991965  Merge pull request #7437: [BEAM-6386] Add named variant of PTransform::compose()
65c8d51 is described below

commit 65c8d510d6e867c7da1320405c1f4e5e18d9c90d
Author: Jeff Klukas <je...@klukas.net>
AuthorDate: Tue Jan 8 13:16:13 2019 -0500

    Add named variant of PTransform#compose()
---
 .../main/java/org/apache/beam/sdk/transforms/PTransform.java | 12 ++++++++++++
 .../java/org/apache/beam/sdk/transforms/PTransformTest.java  |  8 ++++++++
 2 files changed, 20 insertions(+)

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/PTransform.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/PTransform.java
index 49c40f7..0c6b4d2 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/PTransform.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/PTransform.java
@@ -319,4 +319,16 @@ public abstract class PTransform<InputT extends PInput, OutputT extends POutput>
       }
     };
   }
+
+  /** Like {@link #compose(SerializableFunction)}, but with a custom name. */
+  @Experimental
+  public static <InputT extends PInput, OutputT extends POutput>
+      PTransform<InputT, OutputT> compose(String name, SerializableFunction<InputT, OutputT> fn) {
+    return new PTransform<InputT, OutputT>(name) {
+      @Override
+      public OutputT expand(InputT input) {
+        return fn.apply(input);
+      }
+    };
+  }
 }
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PTransformTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PTransformTest.java
index 35d18d9..ae63d5b 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PTransformTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PTransformTest.java
@@ -19,6 +19,7 @@ package org.apache.beam.sdk.transforms;
 
 import static org.apache.beam.sdk.values.TypeDescriptors.integers;
 import static org.hamcrest.Matchers.empty;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThat;
 
 import java.io.Serializable;
@@ -54,6 +55,13 @@ public class PTransformTest implements Serializable {
   }
 
   @Test
+  public void testNamedCompose() {
+    PTransform<PCollection<Integer>, PCollection<Integer>> composed =
+        PTransform.compose("MyName", (PCollection<Integer> numbers) -> numbers);
+    assertEquals("MyName", composed.name);
+  }
+
+  @Test
   @Category(NeedsRunner.class)
   public void testComposeBasicSerializableFunction() throws Exception {
     PCollection<Integer> output =