[FLINK-3075] Change Either creation method names and expose Right/Left classes

Closes #1402

flink-java/src/main/java/org/apache/flink/api/java/typeutils/Either.java

@@ -19,39 +19,45 @@
 package org.apache.flink.api.java.typeutils;
 
 /**
- * This type represents a value of one two possible types, Left or Right
- * (a disjoint union), inspired by Scala's Either type.
+ * This type represents a value of one two possible types, Left or Right (a
+ * disjoint union), inspired by Scala's Either type.
  *
- * @param <L> the type of Left
- * @param <R> the type of Right
+ * @param <L>
+ *            the type of Left
+ * @param <R>
+ *            the type of Right
  */
 public abstract class Either<L, R> {
 
 	/**
 	 * Create a Left value of Either
 	 */
-	public static <L, R> Either<L, R> left(L value) {
+	public static <L, R> Either<L, R> Left(L value) {
 		return new Left<L, R>(value);
 	}
 
 	/**
 	 * Create a Right value of Either
 	 */
-	public static <L, R> Either<L, R> right(R value) {
+	public static <L, R> Either<L, R> Right(R value) {
 		return new Right<L, R>(value);
 	}
 
 	/**
 	 * Retrieve the Left value of Either.
+	 * 
 	 * @return the Left value
-	 * @throws IllegalStateException if called on a Right
+	 * @throws IllegalStateException
+	 *             if called on a Right
 	 */
 	public abstract L left() throws IllegalStateException;
 
 	/**
 	 * Retrieve the Right value of Either.
+	 * 
 	 * @return the Right value
-	 * @throws IllegalStateException if called on a Left
+	 * @throws IllegalStateException
+	 *             if called on a Left
 	 */
 	public abstract R right() throws IllegalStateException;
 
@@ -71,7 +77,15 @@ public abstract class Either<L, R> {
 		return getClass() == Right.class;
 	}
 
-	private static class Left<L, R> extends Either<L, R> {
+	/**
+	 * A left value of {@link Either}
+	 *
+	 * @param <L>
+	 *            the type of Left
+	 * @param <R>
+	 *            the type of Right
+	 */
+	public static class Left<L, R> extends Either<L, R> {
 		private final L value;
 
 		public Left(L value) {
@@ -106,9 +120,25 @@ public abstract class Either<L, R> {
 		public String toString() {
 			return "Left(" + value.toString() + ")";
 		}
+
+		/**
+		 * Creates a left value of {@link Either}
+		 * 
+		 */
+		public static <L, R> Left<L, R> of(L left) {
+			return new Left<L, R>(left);
+		}
 	}
 
-	private static class Right<L, R> extends Either<L, R> {
+	/**
+	 * A right value of {@link Either}
+	 *
+	 * @param <L>
+	 *            the type of Left
+	 * @param <R>
+	 *            the type of Right
+	 */
+	public static class Right<L, R> extends Either<L, R> {
 		private final R value;
 
 		public Right(R value) {
@@ -143,5 +173,13 @@ public abstract class Either<L, R> {
 		public String toString() {
 			return "Right(" + value.toString() + ")";
 		}
+
+		/**
+		 * Creates a right value of {@link Either}
+		 * 
+		 */
+		public static <L, R> Right<L, R> of(R right) {
+			return new Right<L, R>(right);
+		}
 	}
 }

flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializer.java

@@ -18,6 +18,9 @@
 
 package org.apache.flink.api.java.typeutils.runtime;
 
+import static org.apache.flink.api.java.typeutils.Either.Left;
+import static org.apache.flink.api.java.typeutils.Either.Right;
+
 import java.io.IOException;
 
 import org.apache.flink.api.common.typeutils.TypeSerializer;
@@ -67,7 +70,7 @@ public class EitherSerializer<L, R> extends TypeSerializer<Either<L, R>> {
 	@Override
 	public Either<L, R> createInstance() {
 		// We arbitrarily always create a Right value instance.
-		return Either.right(rightSerializer.createInstance());
+		return Right(rightSerializer.createInstance());
 	}
 
 	@Override
@@ -75,12 +78,12 @@ public class EitherSerializer<L, R> extends TypeSerializer<Either<L, R>> {
 		if (from.isLeft()) {
 			L left = from.left();
 			L copyLeft = leftSerializer.copy(left);
-			return Either.left(copyLeft);
+			return Left(copyLeft);
 		}
 		else {
 			R right = from.right();
 			R copyRight = rightSerializer.copy(right);
-			return Either.right(copyRight);
+			return Right(copyRight);
 		}
 	}
 
@@ -90,19 +93,19 @@ public class EitherSerializer<L, R> extends TypeSerializer<Either<L, R>> {
 			final R right = from.right();
 			if (reuse.isRight()) {
 				R copyRight = rightSerializer.copy(right, reuse.right());
-				return Either.right(copyRight);
+				return Right(copyRight);
 			}
 			else {
 				// if the reuse record isn't a right value, we cannot reuse
 				R copyRight = rightSerializer.copy(right);
-				return Either.right(copyRight);
+				return Right(copyRight);
 			}
 		}
 		else {
 			L left = from.left();
 			// reuse record is never a left value because we always create a right instance
 			L copyLeft = leftSerializer.copy(left);
-			return Either.left(copyLeft);
+			return Left(copyLeft);
 		}
 	}
 
@@ -127,10 +130,10 @@ public class EitherSerializer<L, R> extends TypeSerializer<Either<L, R>> {
 	public Either<L, R> deserialize(DataInputView source) throws IOException {
 		boolean isLeft = source.readBoolean();
 		if (isLeft) {
-			return Either.left(leftSerializer.deserialize(source));
+			return Left(leftSerializer.deserialize(source));
 		}
 		else {
-			return Either.right(rightSerializer.deserialize(source));
+			return Right(rightSerializer.deserialize(source));
 		}
 	}
 
@@ -139,16 +142,16 @@ public class EitherSerializer<L, R> extends TypeSerializer<Either<L, R>> {
 		boolean isLeft = source.readBoolean();
 		if (!isLeft) {
 			if (reuse.isRight()) {
-				return Either.right(rightSerializer.deserialize(reuse.right(), source));
+				return Right(rightSerializer.deserialize(reuse.right(), source));
 			}
 			else {
 				// if the reuse record isn't a right value, we cannot reuse
-				return Either.right(rightSerializer.deserialize(source));
+				return Right(rightSerializer.deserialize(source));
 			}
 		}
 		else {
 			// reuse record is never a left value because we always create a right instance
-			return Either.left(leftSerializer.deserialize(source));
+			return Left(leftSerializer.deserialize(source));
 		}
 	}
 

flink-java/src/test/java/org/apache/flink/api/java/type/extractor/TypeExtractorTest.java

@@ -1903,7 +1903,7 @@ public class TypeExtractorTest {
 
 	@Test(expected=InvalidTypesException.class)
 	public void testEitherFromObjectException() {
-		Either<String, Tuple1<Integer>> either = Either.left("test");
+		Either<String, Tuple1<Integer>> either = Either.Left("test");
 		TypeExtractor.getForObject(either);
 	}
 }

flink-java/src/test/java/org/apache/flink/api/java/typeutils/EitherTypeInfoTest.java

@@ -18,18 +18,20 @@
 
 package org.apache.flink.api.java.typeutils;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
 import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.Either.Right;
 import org.apache.flink.util.TestLogger;
 import org.junit.Test;
 
-import static org.junit.Assert.*;
-
 public class EitherTypeInfoTest extends TestLogger {
 
-	Either<Integer, String> intEither = Either.left(1);
-	Either<Integer, String> stringEither = Either.right("boo");
-	Either<Integer, Tuple2<Double, Long>> tuple2Either = Either.right(new Tuple2<Double, Long>(42.0, 2l));
+	Either<Integer, String> intEither = Either.Left(1);
+	Either<Integer, String> stringEither = Either.Right("boo");
+	Either<Integer, Tuple2<Double, Long>> tuple2Either = new Right<>(new Tuple2<Double, Long>(42.0, 2l));
 
 	@Test
 	public void testEitherTypeEquality() {

flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/EitherSerializerTest.java

@@ -20,6 +20,8 @@ package org.apache.flink.api.java.typeutils.runtime;
 
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.fail;
+import static org.apache.flink.api.java.typeutils.Either.Left;
+import static org.apache.flink.api.java.typeutils.Either.Right;
 
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
@@ -38,11 +40,11 @@ public class EitherSerializerTest {
 	public void testStringDoubleEither() {
 
 	Either<String, Double>[] testData = new Either[] {
-			Either.left("banana"),
-			Either.left(""),
-			Either.right(32.0),
-			Either.right(Double.MIN_VALUE),
-			Either.right(Double.MAX_VALUE)};
+			Left("banana"),
+			Left(""),
+			Right(32.0),
+			Right(Double.MIN_VALUE),
+			Right(Double.MAX_VALUE)};
 
 	EitherTypeInfo<String, Double> eitherTypeInfo = (EitherTypeInfo<String, Double>) new EitherTypeInfo<String, Double>(
 			BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.DOUBLE_TYPE_INFO);
@@ -58,11 +60,11 @@ public class EitherSerializerTest {
 	public void testEitherWithTuple() {
 
 	Either<Tuple2<Long, Long>, Double>[] testData = new Either[] {
-			Either.left(new Tuple2<>(2l, 9l)),
-			Either.left(new Tuple2<>(Long.MIN_VALUE, Long.MAX_VALUE)),
-			Either.right(32.0),
-			Either.right(Double.MIN_VALUE),
-			Either.right(Double.MAX_VALUE)};
+			Either.Left(new Tuple2<>(2l, 9l)),
+			new Left<>(new Tuple2<>(Long.MIN_VALUE, Long.MAX_VALUE)),
+			new Right<>(32.0),
+			Right(Double.MIN_VALUE),
+			Right(Double.MAX_VALUE)};
 
 	EitherTypeInfo<Tuple2<Long, Long>, Double> eitherTypeInfo = (EitherTypeInfo<Tuple2<Long, Long>, Double>)
 			new EitherTypeInfo<Tuple2<Long, Long>, Double>(