You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by dh...@apache.org on 2016/03/24 03:48:12 UTC
[48/67] [partial] incubator-beam git commit: Directory reorganization
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/CoderRegistry.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/CoderRegistry.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/CoderRegistry.java
deleted file mode 100644
index 00982e6..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/CoderRegistry.java
+++ /dev/null
@@ -1,843 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.coders;
-
-import com.google.api.services.bigquery.model.TableRow;
-import com.google.cloud.dataflow.sdk.coders.CannotProvideCoderException.ReasonCode;
-import com.google.cloud.dataflow.sdk.coders.protobuf.ProtoCoder;
-import com.google.cloud.dataflow.sdk.transforms.SerializableFunction;
-import com.google.cloud.dataflow.sdk.util.CoderUtils;
-import com.google.cloud.dataflow.sdk.values.KV;
-import com.google.cloud.dataflow.sdk.values.TimestampedValue;
-import com.google.cloud.dataflow.sdk.values.TypeDescriptor;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Maps;
-import com.google.protobuf.ByteString;
-
-import org.joda.time.Instant;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.lang.reflect.ParameterizedType;
-import java.lang.reflect.Type;
-import java.lang.reflect.TypeVariable;
-import java.lang.reflect.WildcardType;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import javax.annotation.Nullable;
-
-/**
- * A {@link CoderRegistry} allows registering the default {@link Coder} to use for a Java class,
- * and looking up and instantiating the default {@link Coder} for a Java type.
- *
- * <p>{@link CoderRegistry} uses the following mechanisms to determine a default {@link Coder} for a
- * Java class, in order of precedence:
- * <ol>
- * <li>Registration:
- * <ul>
- * <li>A {@link CoderFactory} can be registered to handle a particular class via
- * {@link #registerCoder(Class, CoderFactory)}.</li>
- * <li>A {@link Coder} class with the static methods to satisfy
- * {@link CoderFactories#fromStaticMethods} can be registered via
- * {@link #registerCoder(Class, Class)}.</li>
- * <li>Built-in types are registered via
- * {@link #registerStandardCoders()}.</li>
- * </ul>
- * <li>Annotations: {@link DefaultCoder} can be used to annotate a type with
- * the default {@code Coder} type. The {@link Coder} class must satisfy the requirements
- * of {@link CoderProviders#fromStaticMethods}.
- * <li>Fallback: A fallback {@link CoderProvider} is used to attempt to provide a {@link Coder}
- * for any type. By default, this is {@link SerializableCoder#PROVIDER}, which can provide
- * a {@link Coder} for any type that is serializable via Java serialization. The fallback
- * {@link CoderProvider} can be get and set via {@link #getFallbackCoderProvider()}
- * and {@link #setFallbackCoderProvider}. Multiple fallbacks can be chained together using
- * {@link CoderProviders#firstOf}.
- * </ol>
- */
-public class CoderRegistry implements CoderProvider {
-
- private static final Logger LOG = LoggerFactory.getLogger(CoderRegistry.class);
-
- public CoderRegistry() {
- setFallbackCoderProvider(
- CoderProviders.firstOf(ProtoCoder.coderProvider(), SerializableCoder.PROVIDER));
- }
-
- /**
- * Registers standard Coders with this CoderRegistry.
- */
- public void registerStandardCoders() {
- registerCoder(Byte.class, ByteCoder.class);
- registerCoder(ByteString.class, ByteStringCoder.class);
- registerCoder(Double.class, DoubleCoder.class);
- registerCoder(Instant.class, InstantCoder.class);
- registerCoder(Integer.class, VarIntCoder.class);
- registerCoder(Iterable.class, IterableCoder.class);
- registerCoder(KV.class, KvCoder.class);
- registerCoder(List.class, ListCoder.class);
- registerCoder(Long.class, VarLongCoder.class);
- registerCoder(Map.class, MapCoder.class);
- registerCoder(Set.class, SetCoder.class);
- registerCoder(String.class, StringUtf8Coder.class);
- registerCoder(TableRow.class, TableRowJsonCoder.class);
- registerCoder(TimestampedValue.class, TimestampedValue.TimestampedValueCoder.class);
- registerCoder(Void.class, VoidCoder.class);
- registerCoder(byte[].class, ByteArrayCoder.class);
- }
-
- /**
- * Registers {@code coderClazz} as the default {@link Coder} class to handle encoding and
- * decoding instances of {@code clazz}, overriding prior registrations if any exist.
- *
- * <p>Supposing {@code T} is the static type corresponding to the {@code clazz}, then
- * {@code coderClazz} should have a static factory method with the following signature:
- *
- * <pre> {@code
- * public static Coder<T> of(Coder<X> argCoder1, Coder<Y> argCoder2, ...)
- * } </pre>
- *
- * <p>This method will be called to create instances of {@code Coder<T>} for values of type
- * {@code T}, passing Coders for each of the generic type parameters of {@code T}. If {@code T}
- * takes no generic type parameters, then the {@code of()} factory method should have no
- * arguments.
- *
- * <p>If {@code T} is a parameterized type, then it should additionally have a method with the
- * following signature:
- *
- * <pre> {@code
- * public static List<Object> getInstanceComponents(T exampleValue);
- * } </pre>
- *
- * <p>This method will be called to decompose a value during the {@link Coder} inference process,
- * to automatically choose {@link Coder Coders} for the components.
- *
- * @param clazz the class of objects to be encoded
- * @param coderClazz a class with static factory methods to provide {@link Coder Coders}
- */
- public void registerCoder(Class<?> clazz, Class<?> coderClazz) {
- registerCoder(clazz, CoderFactories.fromStaticMethods(coderClazz));
- }
-
- /**
- * Registers {@code coderFactory} as the default {@link CoderFactory} to produce {@code Coder}
- * instances to decode and encode instances of {@code clazz}. This will override prior
- * registrations if any exist.
- */
- public void registerCoder(Class<?> clazz, CoderFactory coderFactory) {
- coderFactoryMap.put(clazz, coderFactory);
- }
-
- /**
- * Register the provided {@link Coder} for encoding all values of the specified {@code Class}.
- * This will override prior registrations if any exist.
- *
- * <p>Not for use with generic rawtypes. Instead, register a {@link CoderFactory} via
- * {@link #registerCoder(Class, CoderFactory)} or ensure your {@code Coder} class has the
- * appropriate static methods and register it directly via {@link #registerCoder(Class, Class)}.
- */
- public <T> void registerCoder(Class<T> rawClazz, Coder<T> coder) {
- Preconditions.checkArgument(
- rawClazz.getTypeParameters().length == 0,
- "CoderRegistry.registerCoder(Class<T>, Coder<T>) may not be used "
- + "with unspecialized generic classes");
-
- CoderFactory factory = CoderFactories.forCoder(coder);
- registerCoder(rawClazz, factory);
- }
-
- /**
- * Returns the {@link Coder} to use by default for values of the given type.
- *
- * @throws CannotProvideCoderException if there is no default Coder.
- */
- public <T> Coder<T> getDefaultCoder(TypeDescriptor<T> typeDescriptor)
- throws CannotProvideCoderException {
- return getDefaultCoder(typeDescriptor, Collections.<Type, Coder<?>>emptyMap());
- }
-
- /**
- * See {@link #getDefaultCoder(TypeDescriptor)}.
- */
- @Override
- public <T> Coder<T> getCoder(TypeDescriptor<T> typeDescriptor)
- throws CannotProvideCoderException {
- return getDefaultCoder(typeDescriptor);
- }
-
- /**
- * Returns the {@link Coder} to use by default for values of the given type, where the given input
- * type uses the given {@link Coder}.
- *
- * @throws CannotProvideCoderException if there is no default Coder.
- */
- public <InputT, OutputT> Coder<OutputT> getDefaultCoder(
- TypeDescriptor<OutputT> typeDescriptor,
- TypeDescriptor<InputT> inputTypeDescriptor,
- Coder<InputT> inputCoder)
- throws CannotProvideCoderException {
- return getDefaultCoder(
- typeDescriptor, getTypeToCoderBindings(inputTypeDescriptor.getType(), inputCoder));
- }
-
- /**
- * Returns the {@link Coder} to use on elements produced by this function, given the {@link Coder}
- * used for its input elements.
- */
- public <InputT, OutputT> Coder<OutputT> getDefaultOutputCoder(
- SerializableFunction<InputT, OutputT> fn, Coder<InputT> inputCoder)
- throws CannotProvideCoderException {
-
- ParameterizedType fnType = (ParameterizedType)
- TypeDescriptor.of(fn.getClass()).getSupertype(SerializableFunction.class).getType();
-
- return getDefaultCoder(
- fn.getClass(),
- SerializableFunction.class,
- ImmutableMap.of(fnType.getActualTypeArguments()[0], inputCoder),
- SerializableFunction.class.getTypeParameters()[1]);
- }
-
- /**
- * Returns the {@link Coder} to use for the specified type parameter specialization of the
- * subclass, given {@link Coder Coders} to use for all other type parameters (if any).
- *
- * @throws CannotProvideCoderException if there is no default Coder.
- */
- public <T, OutputT> Coder<OutputT> getDefaultCoder(
- Class<? extends T> subClass,
- Class<T> baseClass,
- Map<Type, ? extends Coder<?>> knownCoders,
- TypeVariable<?> param)
- throws CannotProvideCoderException {
-
- Map<Type, Coder<?>> inferredCoders = getDefaultCoders(subClass, baseClass, knownCoders);
-
- @SuppressWarnings("unchecked")
- Coder<OutputT> paramCoderOrNull = (Coder<OutputT>) inferredCoders.get(param);
- if (paramCoderOrNull != null) {
- return paramCoderOrNull;
- } else {
- throw new CannotProvideCoderException(
- "Cannot infer coder for type parameter " + param.getName());
- }
- }
-
- /**
- * Returns the {@link Coder} to use for the provided example value, if it can be determined.
- *
- * @throws CannotProvideCoderException if there is no default {@link Coder} or
- * more than one {@link Coder} matches
- */
- public <T> Coder<T> getDefaultCoder(T exampleValue) throws CannotProvideCoderException {
- Class<?> clazz = exampleValue == null ? Void.class : exampleValue.getClass();
-
- if (clazz.getTypeParameters().length == 0) {
- // Trust that getDefaultCoder returns a valid
- // Coder<T> for non-generic clazz.
- @SuppressWarnings("unchecked")
- Coder<T> coder = (Coder<T>) getDefaultCoder(clazz);
- return coder;
- } else {
- CoderFactory factory = getDefaultCoderFactory(clazz);
-
- List<Object> components = factory.getInstanceComponents(exampleValue);
- if (components == null) {
- throw new CannotProvideCoderException(String.format(
- "Cannot provide coder based on value with class %s: The registered CoderFactory with "
- + "class %s failed to decompose the value, which is required in order to provide "
- + "Coders for the components.",
- clazz.getCanonicalName(), factory.getClass().getCanonicalName()));
- }
-
- // componentcoders = components.map(this.getDefaultCoder)
- List<Coder<?>> componentCoders = new ArrayList<>();
- for (Object component : components) {
- try {
- Coder<?> componentCoder = getDefaultCoder(component);
- componentCoders.add(componentCoder);
- } catch (CannotProvideCoderException exc) {
- throw new CannotProvideCoderException(
- String.format("Cannot provide coder based on value with class %s",
- clazz.getCanonicalName()),
- exc);
- }
- }
-
- // Trust that factory.create maps from valid component Coders
- // to a valid Coder<T>.
- @SuppressWarnings("unchecked")
- Coder<T> coder = (Coder<T>) factory.create(componentCoders);
- return coder;
- }
- }
-
- /**
- * Returns the {@link Coder} to use by default for values of the given class. The following three
- * sources for a {@link Coder} will be attempted, in order:
- *
- * <ol>
- * <li>A {@link Coder} class registered explicitly via a call to {@link #registerCoder},
- * <li>A {@link DefaultCoder} annotation on the class,
- * <li>This registry's fallback {@link CoderProvider}, which may be able to generate a
- * {@link Coder} for an arbitrary class.
- * </ol>
- *
- * @throws CannotProvideCoderException if a {@link Coder} cannot be provided
- */
- public <T> Coder<T> getDefaultCoder(Class<T> clazz) throws CannotProvideCoderException {
-
- CannotProvideCoderException factoryException;
- try {
- CoderFactory coderFactory = getDefaultCoderFactory(clazz);
- LOG.debug("Default coder for {} found by factory", clazz);
- @SuppressWarnings("unchecked")
- Coder<T> coder = (Coder<T>) coderFactory.create(Collections.<Coder<?>>emptyList());
- return coder;
- } catch (CannotProvideCoderException exc) {
- factoryException = exc;
- }
-
- CannotProvideCoderException annotationException;
- try {
- return getDefaultCoderFromAnnotation(clazz);
- } catch (CannotProvideCoderException exc) {
- annotationException = exc;
- }
-
- CannotProvideCoderException fallbackException;
- if (getFallbackCoderProvider() != null) {
- try {
- return getFallbackCoderProvider().getCoder(TypeDescriptor.<T>of(clazz));
- } catch (CannotProvideCoderException exc) {
- fallbackException = exc;
- }
- } else {
- fallbackException = new CannotProvideCoderException("no fallback CoderProvider configured");
- }
-
- // Build up the error message and list of causes.
- StringBuilder messageBuilder = new StringBuilder()
- .append("Unable to provide a default Coder for ").append(clazz.getCanonicalName())
- .append(". Correct one of the following root causes:");
-
- messageBuilder
- .append("\n Building a Coder using a registered CoderFactory failed: ")
- .append(factoryException.getMessage());
-
- messageBuilder
- .append("\n Building a Coder from the @DefaultCoder annotation failed: ")
- .append(annotationException.getMessage());
-
- messageBuilder
- .append("\n Building a Coder from the fallback CoderProvider failed: ")
- .append(fallbackException.getMessage());
-
- throw new CannotProvideCoderException(messageBuilder.toString());
- }
-
- /**
- * Sets the fallback {@link CoderProvider} for this registry. If no other method succeeds in
- * providing a {@code Coder<T>} for a type {@code T}, then the registry will attempt to create
- * a {@link Coder} using this {@link CoderProvider}.
- *
- * <p>By default, this is set to {@link SerializableCoder#PROVIDER}.
- *
- * <p>See {@link #getFallbackCoderProvider}.
- */
- public void setFallbackCoderProvider(CoderProvider coderProvider) {
- fallbackCoderProvider = coderProvider;
- }
-
- /**
- * Returns the fallback {@link CoderProvider} for this registry.
- *
- * <p>See {@link #setFallbackCoderProvider}.
- */
- public CoderProvider getFallbackCoderProvider() {
- return fallbackCoderProvider;
- }
-
- /**
- * Returns a {@code Map} from each of {@code baseClass}'s type parameters to the {@link Coder} to
- * use by default for it, in the context of {@code subClass}'s specialization of
- * {@code baseClass}.
- *
- * <p>If no {@link Coder} can be inferred for a particular type parameter, then that type variable
- * will be absent from the returned {@code Map}.
- *
- * <p>For example, if {@code baseClass} is {@code Map.class}, where {@code Map<K, V>} has type
- * parameters {@code K} and {@code V}, and {@code subClass} extends {@code Map<String, Integer>}
- * then the result will map the type variable {@code K} to a {@code Coder<String>} and the
- * type variable {@code V} to a {@code Coder<Integer>}.
- *
- * <p>The {@code knownCoders} parameter can be used to provide known {@link Coder Coders} for any
- * of the parameters; these will be used to infer the others.
- *
- * <p>Note that inference is attempted for every type variable. For a type
- * {@code MyType<One, Two, Three>} inference will be attempted for all of {@code One},
- * {@code Two}, {@code Three}, even if the requester only wants a {@link Coder} for {@code Two}.
- *
- * <p>For this reason {@code getDefaultCoders} (plural) does not throw an exception if a
- * {@link Coder} for a particular type variable cannot be inferred, but merely omits the entry
- * from the returned {@code Map}. It is the responsibility of the caller (usually
- * {@link #getDefaultCoder} to extract the desired coder or throw a
- * {@link CannotProvideCoderException} when appropriate.
- *
- * @param subClass the concrete type whose specializations are being inferred
- * @param baseClass the base type, a parameterized class
- * @param knownCoders a map corresponding to the set of known {@link Coder Coders} indexed by
- * parameter name
- *
- * @deprecated this method is not part of the public interface and will be made private
- */
- @Deprecated
- public <T> Map<Type, Coder<?>> getDefaultCoders(
- Class<? extends T> subClass,
- Class<T> baseClass,
- Map<Type, ? extends Coder<?>> knownCoders) {
- TypeVariable<Class<T>>[] typeParams = baseClass.getTypeParameters();
- Coder<?>[] knownCodersArray = new Coder<?>[typeParams.length];
- for (int i = 0; i < typeParams.length; i++) {
- knownCodersArray[i] = knownCoders.get(typeParams[i]);
- }
- Coder<?>[] resultArray = getDefaultCoders(
- subClass, baseClass, knownCodersArray);
- Map<Type, Coder<?>> result = new HashMap<>();
- for (int i = 0; i < typeParams.length; i++) {
- if (resultArray[i] != null) {
- result.put(typeParams[i], resultArray[i]);
- }
- }
- return result;
- }
-
- /**
- * Returns an array listing, for each of {@code baseClass}'s type parameters, the {@link Coder} to
- * use by default for it, in the context of {@code subClass}'s specialization of
- * {@code baseClass}.
- *
- * <p>If a {@link Coder} cannot be inferred for a type variable, its slot in the resulting array
- * will be {@code null}.
- *
- * <p>For example, if {@code baseClass} is {@code Map.class}, where {@code Map<K, V>} has type
- * parameters {@code K} and {@code V} in that order, and {@code subClass} extends
- * {@code Map<String, Integer>} then the result will contain a {@code Coder<String>} and a
- * {@code Coder<Integer>}, in that order.
- *
- * <p>The {@code knownCoders} parameter can be used to provide known {@link Coder Coders} for any
- * of the type parameters. These will be used to infer the others. If non-null, the length of this
- * array must match the number of type parameters of {@code baseClass}, and simply be filled with
- * {@code null} values for each type parameters without a known {@link Coder}.
- *
- * <p>Note that inference is attempted for every type variable. For a type
- * {@code MyType<One, Two, Three>} inference will will be attempted for all of {@code One},
- * {@code Two}, {@code Three}, even if the requester only wants a {@link Coder} for {@code Two}.
- *
- * <p>For this reason {@code getDefaultCoders} (plural) does not throw an exception if a
- * {@link Coder} for a particular type variable cannot be inferred. Instead, it results in a
- * {@code null} in the array. It is the responsibility of the caller (usually
- * {@link #getDefaultCoder} to extract the desired coder or throw a
- * {@link CannotProvideCoderException} when appropriate.
- *
- * @param subClass the concrete type whose specializations are being inferred
- * @param baseClass the base type, a parameterized class
- * @param knownCoders an array corresponding to the set of base class type parameters. Each entry
- * can be either a {@link Coder} (in which case it will be used for inference) or
- * {@code null} (in which case it will be inferred). May be {@code null} to indicate the
- * entire set of parameters should be inferred.
- * @throws IllegalArgumentException if baseClass doesn't have type parameters or if the length of
- * {@code knownCoders} is not equal to the number of type parameters of {@code baseClass}.
- */
- private <T> Coder<?>[] getDefaultCoders(
- Class<? extends T> subClass,
- Class<T> baseClass,
- @Nullable Coder<?>[] knownCoders) {
- Type type = TypeDescriptor.of(subClass).getSupertype(baseClass).getType();
- if (!(type instanceof ParameterizedType)) {
- throw new IllegalArgumentException(type + " is not a ParameterizedType");
- }
- ParameterizedType parameterizedType = (ParameterizedType) type;
- Type[] typeArgs = parameterizedType.getActualTypeArguments();
- if (knownCoders == null) {
- knownCoders = new Coder<?>[typeArgs.length];
- } else if (typeArgs.length != knownCoders.length) {
- throw new IllegalArgumentException(
- String.format("Class %s has %d parameters, but %d coders are requested.",
- baseClass.getCanonicalName(), typeArgs.length, knownCoders.length));
- }
-
- Map<Type, Coder<?>> context = new HashMap<>();
- for (int i = 0; i < knownCoders.length; i++) {
- if (knownCoders[i] != null) {
- try {
- verifyCompatible(knownCoders[i], typeArgs[i]);
- } catch (IncompatibleCoderException exn) {
- throw new IllegalArgumentException(
- String.format("Provided coders for type arguments of %s contain incompatibilities:"
- + " Cannot encode elements of type %s with coder %s",
- baseClass,
- typeArgs[i], knownCoders[i]),
- exn);
- }
- context.putAll(getTypeToCoderBindings(typeArgs[i], knownCoders[i]));
- }
- }
-
- Coder<?>[] result = new Coder<?>[typeArgs.length];
- for (int i = 0; i < knownCoders.length; i++) {
- if (knownCoders[i] != null) {
- result[i] = knownCoders[i];
- } else {
- try {
- result[i] = getDefaultCoder(typeArgs[i], context);
- } catch (CannotProvideCoderException exc) {
- result[i] = null;
- }
- }
- }
- return result;
- }
-
-
- /////////////////////////////////////////////////////////////////////////////
-
- /**
- * Thrown when a {@link Coder} cannot possibly encode a type, yet has been proposed as a
- * {@link Coder} for that type.
- */
- @VisibleForTesting static class IncompatibleCoderException extends RuntimeException {
- private Coder<?> coder;
- private Type type;
-
- public IncompatibleCoderException(String message, Coder<?> coder, Type type) {
- super(message);
- this.coder = coder;
- this.type = type;
- }
-
- public IncompatibleCoderException(String message, Coder<?> coder, Type type, Throwable cause) {
- super(message, cause);
- this.coder = coder;
- this.type = type;
- }
-
- public Coder<?> getCoder() {
- return coder;
- }
-
- public Type getType() {
- return type;
- }
- }
-
- /**
- * Returns {@code true} if the given {@link Coder} can possibly encode elements
- * of the given type.
- */
- @VisibleForTesting static <T, CoderT extends Coder<T>, CandidateT>
- void verifyCompatible(CoderT coder, Type candidateType) throws IncompatibleCoderException {
-
- // Various representations of the coder's class
- @SuppressWarnings("unchecked")
- Class<CoderT> coderClass = (Class<CoderT>) coder.getClass();
- TypeDescriptor<CoderT> coderDescriptor = TypeDescriptor.of(coderClass);
-
- // Various representations of the actual coded type
- @SuppressWarnings("unchecked")
- TypeDescriptor<T> codedDescriptor = CoderUtils.getCodedType(coderDescriptor);
- @SuppressWarnings("unchecked")
- Class<T> codedClass = (Class<T>) codedDescriptor.getRawType();
- Type codedType = codedDescriptor.getType();
-
- // Various representations of the candidate type
- @SuppressWarnings("unchecked")
- TypeDescriptor<CandidateT> candidateDescriptor =
- (TypeDescriptor<CandidateT>) TypeDescriptor.of(candidateType);
- @SuppressWarnings("unchecked")
- Class<CandidateT> candidateClass = (Class<CandidateT>) candidateDescriptor.getRawType();
-
- // If coder has type Coder<T> where the actual value of T is lost
- // to erasure, then we cannot rule it out.
- if (candidateType instanceof TypeVariable) {
- return;
- }
-
- // If the raw types are not compatible, we can certainly rule out
- // coder compatibility
- if (!codedClass.isAssignableFrom(candidateClass)) {
- throw new IncompatibleCoderException(
- String.format("Cannot encode elements of type %s with coder %s because the"
- + " coded type %s is not assignable from %s",
- candidateType, coder, codedClass, candidateType),
- coder, candidateType);
- }
- // we have established that this is a covariant upcast... though
- // coders are invariant, we are just checking one direction
- @SuppressWarnings("unchecked")
- TypeDescriptor<T> candidateOkDescriptor = (TypeDescriptor<T>) candidateDescriptor;
-
- // If the coded type is a parameterized type where any of the actual
- // type parameters are not compatible, then the whole thing is certainly not
- // compatible.
- if ((codedType instanceof ParameterizedType) && !isNullOrEmpty(coder.getCoderArguments())) {
- ParameterizedType parameterizedSupertype = ((ParameterizedType)
- candidateOkDescriptor.getSupertype(codedClass).getType());
- Type[] typeArguments = parameterizedSupertype.getActualTypeArguments();
- List<? extends Coder<?>> typeArgumentCoders = coder.getCoderArguments();
- if (typeArguments.length < typeArgumentCoders.size()) {
- throw new IncompatibleCoderException(
- String.format("Cannot encode elements of type %s with coder %s:"
- + " the generic supertype %s has %s type parameters, which is less than the"
- + " number of coder arguments %s has (%s).",
- candidateOkDescriptor, coder,
- parameterizedSupertype, typeArguments.length,
- coder, typeArgumentCoders.size()),
- coder, candidateOkDescriptor.getType());
- }
- for (int i = 0; i < typeArgumentCoders.size(); i++) {
- try {
- verifyCompatible(
- typeArgumentCoders.get(i),
- candidateDescriptor.resolveType(typeArguments[i]).getType());
- } catch (IncompatibleCoderException exn) {
- throw new IncompatibleCoderException(
- String.format("Cannot encode elements of type %s with coder %s"
- + " because some component coder is incompatible",
- candidateType, coder),
- coder, candidateType, exn);
- }
- }
- }
- }
-
- private static boolean isNullOrEmpty(Collection<?> c) {
- return c == null || c.size() == 0;
- }
-
- /**
- * The map of classes to the CoderFactories to use to create their
- * default Coders.
- */
- private Map<Class<?>, CoderFactory> coderFactoryMap = new HashMap<>();
-
- /**
- * A provider of coders for types where no coder is registered.
- */
- private CoderProvider fallbackCoderProvider;
-
- /**
- * Returns the {@link CoderFactory} to use to create default {@link Coder Coders} for instances of
- * the given class, or {@code null} if there is no default {@link CoderFactory} registered.
- */
- private CoderFactory getDefaultCoderFactory(Class<?> clazz) throws CannotProvideCoderException {
- CoderFactory coderFactoryOrNull = coderFactoryMap.get(clazz);
- if (coderFactoryOrNull != null) {
- return coderFactoryOrNull;
- } else {
- throw new CannotProvideCoderException(
- String.format("Cannot provide coder based on value with class %s: No CoderFactory has "
- + "been registered for the class.", clazz.getCanonicalName()));
- }
- }
-
- /**
- * Returns the {@link Coder} returned according to the {@link CoderProvider} from any
- * {@link DefaultCoder} annotation on the given class.
- */
- private <T> Coder<T> getDefaultCoderFromAnnotation(Class<T> clazz)
- throws CannotProvideCoderException {
- DefaultCoder defaultAnnotation = clazz.getAnnotation(DefaultCoder.class);
- if (defaultAnnotation == null) {
- throw new CannotProvideCoderException(
- String.format("Class %s does not have a @DefaultCoder annotation.",
- clazz.getCanonicalName()));
- }
-
- LOG.debug("DefaultCoder annotation found for {} with value {}",
- clazz, defaultAnnotation.value());
- CoderProvider coderProvider = CoderProviders.fromStaticMethods(defaultAnnotation.value());
- return coderProvider.getCoder(TypeDescriptor.of(clazz));
- }
-
- /**
- * Returns the {@link Coder} to use by default for values of the given type,
- * in a context where the given types use the given coders.
- *
- * @throws CannotProvideCoderException if a coder cannot be provided
- */
- private <T> Coder<T> getDefaultCoder(
- TypeDescriptor<T> typeDescriptor,
- Map<Type, Coder<?>> typeCoderBindings)
- throws CannotProvideCoderException {
-
- Coder<?> defaultCoder = getDefaultCoder(typeDescriptor.getType(), typeCoderBindings);
- LOG.debug("Default coder for {}: {}", typeDescriptor, defaultCoder);
- @SuppressWarnings("unchecked")
- Coder<T> result = (Coder<T>) defaultCoder;
- return result;
- }
-
- /**
- * Returns the {@link Coder} to use by default for values of the given type,
- * in a context where the given types use the given coders.
- *
- * @throws CannotProvideCoderException if a coder cannot be provided
- */
- private Coder<?> getDefaultCoder(Type type, Map<Type, Coder<?>> typeCoderBindings)
- throws CannotProvideCoderException {
- Coder<?> coder = typeCoderBindings.get(type);
- if (coder != null) {
- return coder;
- }
- if (type instanceof Class<?>) {
- Class<?> clazz = (Class<?>) type;
- return getDefaultCoder(clazz);
- } else if (type instanceof ParameterizedType) {
- return getDefaultCoder((ParameterizedType) type, typeCoderBindings);
- } else if (type instanceof TypeVariable || type instanceof WildcardType) {
- // No default coder for an unknown generic type.
- throw new CannotProvideCoderException(
- String.format("Cannot provide a coder for type variable %s"
- + " (declared by %s) because the actual type is unknown due to erasure.",
- type,
- ((TypeVariable<?>) type).getGenericDeclaration()),
- ReasonCode.TYPE_ERASURE);
- } else {
- throw new RuntimeException(
- "Internal error: unexpected kind of Type: " + type);
- }
- }
-
- /**
- * Returns the {@link Coder} to use by default for values of the given
- * parameterized type, in a context where the given types use the
- * given {@link Coder Coders}.
- *
- * @throws CannotProvideCoderException if no coder can be provided
- */
- private Coder<?> getDefaultCoder(
- ParameterizedType type,
- Map<Type, Coder<?>> typeCoderBindings)
- throws CannotProvideCoderException {
-
- CannotProvideCoderException factoryException;
- try {
- return getDefaultCoderFromFactory(type, typeCoderBindings);
- } catch (CannotProvideCoderException exc) {
- factoryException = exc;
- }
-
- CannotProvideCoderException annotationException;
- try {
- Class<?> rawClazz = (Class<?>) type.getRawType();
- return getDefaultCoderFromAnnotation(rawClazz);
- } catch (CannotProvideCoderException exc) {
- annotationException = exc;
- }
-
- // Build up the error message and list of causes.
- StringBuilder messageBuilder = new StringBuilder()
- .append("Unable to provide a default Coder for ").append(type)
- .append(". Correct one of the following root causes:");
-
- messageBuilder
- .append("\n Building a Coder using a registered CoderFactory failed: ")
- .append(factoryException.getMessage());
-
- messageBuilder
- .append("\n Building a Coder from the @DefaultCoder annotation failed: ")
- .append(annotationException.getMessage());
-
- throw new CannotProvideCoderException(messageBuilder.toString());
- }
-
- private Coder<?> getDefaultCoderFromFactory(
- ParameterizedType type,
- Map<Type, Coder<?>> typeCoderBindings)
- throws CannotProvideCoderException {
- Class<?> rawClazz = (Class<?>) type.getRawType();
- CoderFactory coderFactory = getDefaultCoderFactory(rawClazz);
- List<Coder<?>> typeArgumentCoders = new ArrayList<>();
- for (Type typeArgument : type.getActualTypeArguments()) {
- try {
- Coder<?> typeArgumentCoder = getDefaultCoder(typeArgument,
- typeCoderBindings);
- typeArgumentCoders.add(typeArgumentCoder);
- } catch (CannotProvideCoderException exc) {
- throw new CannotProvideCoderException(
- String.format("Cannot provide coder for parameterized type %s: %s",
- type,
- exc.getMessage()),
- exc);
- }
- }
- return coderFactory.create(typeArgumentCoders);
- }
-
- /**
- * Returns an immutable {@code Map} from each of the type variables
- * embedded in the given type to the corresponding types
- * in the given {@link Coder}.
- */
- private Map<Type, Coder<?>> getTypeToCoderBindings(Type type, Coder<?> coder) {
- if (type instanceof TypeVariable || type instanceof Class) {
- return ImmutableMap.<Type, Coder<?>>of(type, coder);
- } else if (type instanceof ParameterizedType) {
- return getTypeToCoderBindings((ParameterizedType) type, coder);
- } else {
- return ImmutableMap.of();
- }
- }
-
- /**
- * Returns an immutable {@code Map} from the type arguments of the parameterized type to their
- * corresponding {@link Coder Coders}, and so on recursively for their type parameters.
- *
- * <p>This method is simply a specialization to break out the most
- * elaborate case of {@link #getTypeToCoderBindings(Type, Coder)}.
- */
- private Map<Type, Coder<?>> getTypeToCoderBindings(ParameterizedType type, Coder<?> coder) {
- List<Type> typeArguments = Arrays.asList(type.getActualTypeArguments());
- List<? extends Coder<?>> coderArguments = coder.getCoderArguments();
-
- if ((coderArguments == null) || (typeArguments.size() != coderArguments.size())) {
- return ImmutableMap.of();
- } else {
- Map<Type, Coder<?>> typeToCoder = Maps.newHashMap();
-
- typeToCoder.put(type, coder);
-
- for (int i = 0; i < typeArguments.size(); i++) {
- Type typeArgument = typeArguments.get(i);
- Coder<?> coderArgument = coderArguments.get(i);
- typeToCoder.putAll(getTypeToCoderBindings(typeArgument, coderArgument));
- }
-
- return ImmutableMap.<Type, Coder<?>>builder().putAll(typeToCoder).build();
- }
-
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/CollectionCoder.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/CollectionCoder.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/CollectionCoder.java
deleted file mode 100644
index a028317..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/CollectionCoder.java
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.coders;
-
-import com.google.cloud.dataflow.sdk.util.PropertyNames;
-import com.google.common.base.Preconditions;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-import java.util.Collection;
-import java.util.List;
-
-/**
- * A {@link CollectionCoder} encodes {@link Collection Collections} in the format
- * of {@link IterableLikeCoder}.
- */
-public class CollectionCoder<T> extends IterableLikeCoder<T, Collection<T>> {
-
- public static <T> CollectionCoder<T> of(Coder<T> elemCoder) {
- return new CollectionCoder<>(elemCoder);
- }
-
- /////////////////////////////////////////////////////////////////////////////
- // Internal operations below here.
-
- /**
- * {@inheritDoc}
- *
- * @return the decoded elements directly, since {@link List} is a subtype of
- * {@link Collection}.
- */
- @Override
- protected final Collection<T> decodeToIterable(List<T> decodedElements) {
- return decodedElements;
- }
-
- @JsonCreator
- public static CollectionCoder<?> of(
- @JsonProperty(PropertyNames.COMPONENT_ENCODINGS)
- List<Object> components) {
- Preconditions.checkArgument(components.size() == 1,
- "Expecting 1 component, got " + components.size());
- return of((Coder<?>) components.get(0));
- }
-
- /**
- * Returns the first element in this collection if it is non-empty,
- * otherwise returns {@code null}.
- */
- public static <T> List<Object> getInstanceComponents(
- Collection<T> exampleValue) {
- return getInstanceComponentsHelper(exampleValue);
- }
-
- protected CollectionCoder(Coder<T> elemCoder) {
- super(elemCoder, "Collection");
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/CustomCoder.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/CustomCoder.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/CustomCoder.java
deleted file mode 100644
index b34ef8c..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/CustomCoder.java
+++ /dev/null
@@ -1,137 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.coders;
-
-import static com.google.cloud.dataflow.sdk.util.Structs.addString;
-import static com.google.cloud.dataflow.sdk.util.Structs.addStringList;
-import static com.google.common.base.Preconditions.checkNotNull;
-
-import com.google.cloud.dataflow.sdk.util.CloudObject;
-import com.google.cloud.dataflow.sdk.util.PropertyNames;
-import com.google.cloud.dataflow.sdk.util.SerializableUtils;
-import com.google.cloud.dataflow.sdk.util.StringUtils;
-import com.google.common.collect.Lists;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-import java.io.Serializable;
-import java.util.Collection;
-
-/**
- * An abstract base class for writing a {@link Coder} class that encodes itself via Java
- * serialization.
- *
- * <p>To complete an implementation, subclasses must implement {@link Coder#encode}
- * and {@link Coder#decode} methods. Anonymous subclasses must furthermore override
- * {@link #getEncodingId}.
- *
- * <p>Not to be confused with {@link SerializableCoder} that encodes objects that implement the
- * {@link Serializable} interface.
- *
- * @param <T> the type of elements handled by this coder
- */
-public abstract class CustomCoder<T> extends AtomicCoder<T>
- implements Serializable {
- @JsonCreator
- public static CustomCoder<?> of(
- // N.B. typeId is a required parameter here, since a field named "@type"
- // is presented to the deserializer as an input.
- //
- // If this method did not consume the field, Jackson2 would observe an
- // unconsumed field and a returned value of a derived type. So Jackson2
- // would attempt to update the returned value with the unconsumed field
- // data, The standard JsonDeserializer does not implement a mechanism for
- // updating constructed values, so it would throw an exception, causing
- // deserialization to fail.
- @JsonProperty(value = "@type", required = false) String typeId,
- @JsonProperty(value = "encoding_id", required = false) String encodingId,
- @JsonProperty("type") String type,
- @JsonProperty("serialized_coder") String serializedCoder) {
- return (CustomCoder<?>) SerializableUtils.deserializeFromByteArray(
- StringUtils.jsonStringToByteArray(serializedCoder),
- type);
- }
-
- /**
- * {@inheritDoc}
- *
- * @return A thin {@link CloudObject} wrapping of the Java serialization of {@code this}.
- */
- @Override
- public CloudObject asCloudObject() {
- // N.B. We use the CustomCoder class, not the derived class, since during
- // deserialization we will be using the CustomCoder's static factory method
- // to construct an instance of the derived class.
- CloudObject result = CloudObject.forClass(CustomCoder.class);
- addString(result, "type", getClass().getName());
- addString(result, "serialized_coder",
- StringUtils.byteArrayToJsonString(
- SerializableUtils.serializeToByteArray(this)));
-
- String encodingId = getEncodingId();
- checkNotNull(encodingId, "Coder.getEncodingId() must not return null.");
- if (!encodingId.isEmpty()) {
- addString(result, PropertyNames.ENCODING_ID, encodingId);
- }
-
- Collection<String> allowedEncodings = getAllowedEncodings();
- if (!allowedEncodings.isEmpty()) {
- addStringList(result, PropertyNames.ALLOWED_ENCODINGS, Lists.newArrayList(allowedEncodings));
- }
-
- return result;
- }
-
- /**
- * {@inheritDoc}
- *
- * @throws NonDeterministicException a {@link CustomCoder} is presumed
- * nondeterministic.
- */
- @Override
- public void verifyDeterministic() throws NonDeterministicException {
- throw new NonDeterministicException(this,
- "CustomCoder implementations must override verifyDeterministic,"
- + " or they are presumed nondeterministic.");
- }
-
- /**
- * {@inheritDoc}
- *
- * @return The canonical class name for this coder. For stable data formats that are independent
- * of class name, it is recommended to override this method.
- *
- * @throws UnsupportedOperationException when an anonymous class is used, since they do not have
- * a stable canonical class name.
- */
- @Override
- public String getEncodingId() {
- if (getClass().isAnonymousClass()) {
- throw new UnsupportedOperationException(
- String.format("Anonymous CustomCoder subclass %s must override getEncodingId()."
- + " Otherwise, convert to a named class and getEncodingId() will be automatically"
- + " generated from the fully qualified class name.",
- getClass()));
- }
- return getClass().getCanonicalName();
- }
-
- // This coder inherits isRegisterByteSizeObserverCheap,
- // getEncodedElementByteSize and registerByteSizeObserver
- // from StandardCoder. Override if we can do better.
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DefaultCoder.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DefaultCoder.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DefaultCoder.java
deleted file mode 100644
index 110579b..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DefaultCoder.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.coders;
-
-import com.google.cloud.dataflow.sdk.values.PCollection;
-
-import java.lang.annotation.Documented;
-import java.lang.annotation.ElementType;
-import java.lang.annotation.Retention;
-import java.lang.annotation.RetentionPolicy;
-import java.lang.annotation.Target;
-
-/**
- * The {@link DefaultCoder} annotation
- * specifies a default {@link Coder} class to handle encoding and decoding
- * instances of the annotated class.
- *
- * <p>The specified {@link Coder} must satisfy the requirements of
- * {@link CoderProviders#fromStaticMethods}. Two classes provided by the SDK that
- * are intended for use with this annotation include {@link SerializableCoder}
- * and {@link AvroCoder}.
- *
- * <p>To configure the use of Java serialization as the default
- * for a class, annotate the class to use
- * {@link SerializableCoder} as follows:
- *
- * <pre><code>{@literal @}DefaultCoder(SerializableCoder.class)
- * public class MyCustomDataType implements Serializable {
- * // ...
- * }</code></pre>
- *
- * <p>Similarly, to configure the use of
- * {@link AvroCoder} as the default:
- * <pre><code>{@literal @}DefaultCoder(AvroCoder.class)
- * public class MyCustomDataType {
- * public MyCustomDataType() {} // Avro requires an empty constructor.
- * // ...
- * }</code></pre>
- *
- * <p>Coders specified explicitly via
- * {@link PCollection#setCoder}
- * take precedence, followed by Coders registered at runtime via
- * {@link CoderRegistry#registerCoder}. See {@link CoderRegistry} for a more detailed discussion
- * of the precedence rules.
- */
-@Documented
-@Retention(RetentionPolicy.RUNTIME)
-@Target(ElementType.TYPE)
-@SuppressWarnings("rawtypes")
-public @interface DefaultCoder {
- Class<? extends Coder> value();
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DelegateCoder.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DelegateCoder.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DelegateCoder.java
deleted file mode 100644
index cdd882b..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DelegateCoder.java
+++ /dev/null
@@ -1,164 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.coders;
-
-import com.google.common.collect.Lists;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.io.Serializable;
-import java.util.Collection;
-import java.util.List;
-
-/**
- * A {@code DelegateCoder<T, IntermediateT>} wraps a {@link Coder} for {@code IntermediateT} and
- * encodes/decodes values of type {@code T} by converting
- * to/from {@code IntermediateT} and then encoding/decoding using the underlying
- * {@code Coder<IntermediateT>}.
- *
- * <p>The conversions from {@code T} to {@code IntermediateT} and vice versa
- * must be supplied as {@link CodingFunction}, a serializable
- * function that may throw any {@code Exception}. If a thrown
- * exception is an instance of {@link CoderException} or
- * {@link IOException}, it will be re-thrown, otherwise it will be wrapped as
- * a {@link CoderException}.
- *
- * @param <T> The type of objects coded by this Coder.
- * @param <IntermediateT> The type of objects a {@code T} will be converted to for coding.
- */
-public class DelegateCoder<T, IntermediateT> extends CustomCoder<T> {
- /**
- * A {@link DelegateCoder.CodingFunction CodingFunction<InputT, OutputT>} is a serializable
- * function from {@code InputT} to {@code OutputT} that may throw any {@link Exception}.
- */
- public static interface CodingFunction<InputT, OutputT> extends Serializable {
- public abstract OutputT apply(InputT input) throws Exception;
- }
-
- public static <T, IntermediateT> DelegateCoder<T, IntermediateT> of(Coder<IntermediateT> coder,
- CodingFunction<T, IntermediateT> toFn,
- CodingFunction<IntermediateT, T> fromFn) {
- return new DelegateCoder<T, IntermediateT>(coder, toFn, fromFn);
- }
-
- @Override
- public void encode(T value, OutputStream outStream, Context context)
- throws CoderException, IOException {
- coder.encode(applyAndWrapExceptions(toFn, value), outStream, context);
- }
-
- @Override
- public T decode(InputStream inStream, Context context) throws CoderException, IOException {
- return applyAndWrapExceptions(fromFn, coder.decode(inStream, context));
- }
-
- /**
- * Returns the coder used to encode/decode the intermediate values produced/consumed by the
- * coding functions of this {@code DelegateCoder}.
- */
- public Coder<IntermediateT> getCoder() {
- return coder;
- }
-
- /**
- * {@inheritDoc}
- *
- * @throws NonDeterministicException when the underlying coder's {@code verifyDeterministic()}
- * throws a {@link Coder.NonDeterministicException}. For this to be safe, the
- * intermediate {@code CodingFunction<T, IntermediateT>} must also be deterministic.
- */
- @Override
- public void verifyDeterministic() throws NonDeterministicException {
- coder.verifyDeterministic();
- }
-
- /**
- * {@inheritDoc}
- *
- * @return a structural for a value of type {@code T} obtained by first converting to
- * {@code IntermediateT} and then obtaining a structural value according to the underlying
- * coder.
- */
- @Override
- public Object structuralValue(T value) throws Exception {
- return coder.structuralValue(toFn.apply(value));
- }
-
- @Override
- public String toString() {
- return "DelegateCoder(" + coder + ")";
- }
-
- /**
- * {@inheritDoc}
- *
- * @return a {@link String} composed from the underlying coder class name and its encoding id.
- * Note that this omits any description of the coding functions. These should be modified
- * with care.
- */
- @Override
- public String getEncodingId() {
- return delegateEncodingId(coder.getClass(), coder.getEncodingId());
- }
-
- /**
- * {@inheritDoc}
- *
- * @return allowed encodings which are composed from the underlying coder class and its allowed
- * encoding ids. Note that this omits any description of the coding functions. These
- * should be modified with care.
- */
- @Override
- public Collection<String> getAllowedEncodings() {
- List<String> allowedEncodings = Lists.newArrayList();
- for (String allowedEncoding : coder.getAllowedEncodings()) {
- allowedEncodings.add(delegateEncodingId(coder.getClass(), allowedEncoding));
- }
- return allowedEncodings;
- }
-
- private String delegateEncodingId(Class<?> delegateClass, String encodingId) {
- return String.format("%s:%s", delegateClass.getName(), encodingId);
- }
-
- /////////////////////////////////////////////////////////////////////////////
-
- private <InputT, OutputT> OutputT applyAndWrapExceptions(
- CodingFunction<InputT, OutputT> fn,
- InputT input) throws CoderException, IOException {
- try {
- return fn.apply(input);
- } catch (IOException exc) {
- throw exc;
- } catch (Exception exc) {
- throw new CoderException(exc);
- }
- }
-
- private final Coder<IntermediateT> coder;
- private final CodingFunction<T, IntermediateT> toFn;
- private final CodingFunction<IntermediateT, T> fromFn;
-
- protected DelegateCoder(Coder<IntermediateT> coder,
- CodingFunction<T, IntermediateT> toFn,
- CodingFunction<IntermediateT, T> fromFn) {
- this.coder = coder;
- this.fromFn = fromFn;
- this.toFn = toFn;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DeterministicStandardCoder.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DeterministicStandardCoder.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DeterministicStandardCoder.java
deleted file mode 100644
index 0e0018a..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DeterministicStandardCoder.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.coders;
-
-/**
- * A {@link DeterministicStandardCoder} is a {@link StandardCoder} that is
- * deterministic, in the sense that for objects considered equal
- * according to {@link Object#equals(Object)}, the encoded bytes are
- * also equal.
- *
- * @param <T> the type of the values being transcoded
- */
-public abstract class DeterministicStandardCoder<T> extends StandardCoder<T> {
- protected DeterministicStandardCoder() {}
-
- /**
- * {@inheritDoc}
- *
- * @throws NonDeterministicException never, unless overridden. A
- * {@link DeterministicStandardCoder} is presumed deterministic.
- */
- @Override
- public void verifyDeterministic() throws NonDeterministicException { }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DoubleCoder.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DoubleCoder.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DoubleCoder.java
deleted file mode 100644
index 68d58df..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DoubleCoder.java
+++ /dev/null
@@ -1,113 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.coders;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.EOFException;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.io.UTFDataFormatException;
-
-/**
- * A {@link DoubleCoder} encodes {@link Double} values in 8 bytes using Java serialization.
- */
-public class DoubleCoder extends AtomicCoder<Double> {
-
- @JsonCreator
- public static DoubleCoder of() {
- return INSTANCE;
- }
-
- /////////////////////////////////////////////////////////////////////////////
-
- private static final DoubleCoder INSTANCE = new DoubleCoder();
-
- private DoubleCoder() {}
-
- @Override
- public void encode(Double value, OutputStream outStream, Context context)
- throws IOException, CoderException {
- if (value == null) {
- throw new CoderException("cannot encode a null Double");
- }
- new DataOutputStream(outStream).writeDouble(value);
- }
-
- @Override
- public Double decode(InputStream inStream, Context context)
- throws IOException, CoderException {
- try {
- return new DataInputStream(inStream).readDouble();
- } catch (EOFException | UTFDataFormatException exn) {
- // These exceptions correspond to decoding problems, so change
- // what kind of exception they're branded as.
- throw new CoderException(exn);
- }
- }
-
- /**
- * {@inheritDoc}
- *
- * @throws NonDeterministicException always.
- * Floating-point operations are not guaranteed to be deterministic, even
- * if the storage format might be, so floating point representations are not
- * recommended for use in operations that require deterministic inputs.
- */
- @Override
- public void verifyDeterministic() throws NonDeterministicException {
- throw new NonDeterministicException(this,
- "Floating point encodings are not guaranteed to be deterministic.");
- }
-
- /**
- * {@inheritDoc}
- *
- * @return {@code true}. This coder is injective.
- */
- @Override
- public boolean consistentWithEquals() {
- return true;
- }
-
- /**
- * {@inheritDoc}
- *
- * @return {@code true}. {@link DoubleCoder#getEncodedElementByteSize} returns a constant.
- */
- @Override
- public boolean isRegisterByteSizeObserverCheap(Double value, Context context) {
- return true;
- }
-
- /**
- * {@inheritDoc}
- *
- * @return {@code 8}, the byte size of a {@link Double} encoded using Java serialization.
- */
- @Override
- protected long getEncodedElementByteSize(Double value, Context context)
- throws Exception {
- if (value == null) {
- throw new CoderException("cannot encode a null Double");
- }
- return 8;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DurationCoder.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DurationCoder.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DurationCoder.java
deleted file mode 100644
index 25527f0..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/DurationCoder.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.coders;
-
-import com.google.cloud.dataflow.sdk.util.common.ElementByteSizeObserver;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-
-import org.joda.time.Duration;
-import org.joda.time.ReadableDuration;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-
-/**
- * A {@link Coder} that encodes a joda {@link Duration} as a {@link Long} using the format of
- * {@link VarLongCoder}.
- */
-public class DurationCoder extends AtomicCoder<ReadableDuration> {
-
- @JsonCreator
- public static DurationCoder of() {
- return INSTANCE;
- }
-
- /////////////////////////////////////////////////////////////////////////////
-
- private static final DurationCoder INSTANCE = new DurationCoder();
-
- private final VarLongCoder longCoder = VarLongCoder.of();
-
- private DurationCoder() {}
-
- private Long toLong(ReadableDuration value) {
- return value.getMillis();
- }
-
- private ReadableDuration fromLong(Long decoded) {
- return Duration.millis(decoded);
- }
-
- @Override
- public void encode(ReadableDuration value, OutputStream outStream, Context context)
- throws CoderException, IOException {
- if (value == null) {
- throw new CoderException("cannot encode a null ReadableDuration");
- }
- longCoder.encode(toLong(value), outStream, context);
- }
-
- @Override
- public ReadableDuration decode(InputStream inStream, Context context)
- throws CoderException, IOException {
- return fromLong(longCoder.decode(inStream, context));
- }
-
- /**
- * {@inheritDoc}
- *
- * @return {@code true}. This coder is injective.
- */
- @Override
- public boolean consistentWithEquals() {
- return true;
- }
-
- /**
- * {@inheritDoc}
- *
- * @return {@code true}, because it is cheap to ascertain the byte size of a long.
- */
- @Override
- public boolean isRegisterByteSizeObserverCheap(ReadableDuration value, Context context) {
- return longCoder.isRegisterByteSizeObserverCheap(toLong(value), context);
- }
-
- @Override
- public void registerByteSizeObserver(
- ReadableDuration value, ElementByteSizeObserver observer, Context context) throws Exception {
- longCoder.registerByteSizeObserver(toLong(value), observer, context);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/EntityCoder.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/EntityCoder.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/EntityCoder.java
deleted file mode 100644
index 3ae857f..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/EntityCoder.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.coders;
-
-import com.google.api.services.datastore.DatastoreV1.Entity;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-
-/**
- * A {@link Coder} for {@link Entity} objects based on their encoded Protocol Buffer form.
- */
-public class EntityCoder extends AtomicCoder<Entity> {
-
- @JsonCreator
- public static EntityCoder of() {
- return INSTANCE;
- }
-
- /***************************/
-
- private static final EntityCoder INSTANCE = new EntityCoder();
-
- private EntityCoder() {}
-
- @Override
- public void encode(Entity value, OutputStream outStream, Context context)
- throws IOException, CoderException {
- if (value == null) {
- throw new CoderException("cannot encode a null Entity");
- }
-
- // Since Entity implements com.google.protobuf.MessageLite,
- // we could directly use writeTo to write to a OutputStream object
- outStream.write(java.nio.ByteBuffer.allocate(4).putInt(value.getSerializedSize()).array());
- value.writeTo(outStream);
- outStream.flush();
- }
-
- @Override
- public Entity decode(InputStream inStream, Context context)
- throws IOException {
- byte[] entitySize = new byte[4];
- inStream.read(entitySize, 0, 4);
- int size = java.nio.ByteBuffer.wrap(entitySize).getInt();
- byte[] data = new byte[size];
- inStream.read(data, 0, size);
- return Entity.parseFrom(data);
- }
-
- @Override
- protected long getEncodedElementByteSize(Entity value, Context context)
- throws Exception {
- return value.getSerializedSize();
- }
-
- /**
- * {@inheritDoc}
- *
- * @throws NonDeterministicException always.
- * A datastore kind can hold arbitrary {@link Object} instances, which
- * makes the encoding non-deterministic.
- */
- @Override
- public void verifyDeterministic() throws NonDeterministicException {
- throw new NonDeterministicException(this,
- "Datastore encodings can hold arbitrary Object instances");
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/InstantCoder.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/InstantCoder.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/InstantCoder.java
deleted file mode 100644
index 99b58ce..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/InstantCoder.java
+++ /dev/null
@@ -1,113 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.coders;
-
-import com.google.cloud.dataflow.sdk.util.common.ElementByteSizeObserver;
-import com.google.common.base.Converter;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-
-import org.joda.time.Instant;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-
-/**
- * A {@link Coder} for joda {@link Instant} that encodes it as a big endian {@link Long}
- * shifted such that lexicographic ordering of the bytes corresponds to chronological order.
- */
-public class InstantCoder extends AtomicCoder<Instant> {
-
- @JsonCreator
- public static InstantCoder of() {
- return INSTANCE;
- }
-
- /////////////////////////////////////////////////////////////////////////////
-
- private static final InstantCoder INSTANCE = new InstantCoder();
-
- private final BigEndianLongCoder longCoder = BigEndianLongCoder.of();
-
- private InstantCoder() {}
-
- /**
- * Converts {@link Instant} to a {@code Long} representing its millis-since-epoch,
- * but shifted so that the byte representation of negative values are lexicographically
- * ordered before the byte representation of positive values.
- *
- * <p>This deliberately utilizes the well-defined overflow for {@code Long} values.
- * See http://docs.oracle.com/javase/specs/jls/se7/html/jls-15.html#jls-15.18.2
- */
- private static final Converter<Instant, Long> ORDER_PRESERVING_CONVERTER =
- new Converter<Instant, Long>() {
-
- @Override
- protected Long doForward(Instant instant) {
- return instant.getMillis() - Long.MIN_VALUE;
- }
-
- @Override
- protected Instant doBackward(Long shiftedMillis) {
- return new Instant(shiftedMillis + Long.MIN_VALUE);
- }
- };
-
- @Override
- public void encode(Instant value, OutputStream outStream, Context context)
- throws CoderException, IOException {
- if (value == null) {
- throw new CoderException("cannot encode a null Instant");
- }
- longCoder.encode(ORDER_PRESERVING_CONVERTER.convert(value), outStream, context);
- }
-
- @Override
- public Instant decode(InputStream inStream, Context context)
- throws CoderException, IOException {
- return ORDER_PRESERVING_CONVERTER.reverse().convert(longCoder.decode(inStream, context));
- }
-
- /**
- * {@inheritDoc}
- *
- * @return {@code true}. This coder is injective.
- */
- @Override
- public boolean consistentWithEquals() {
- return true;
- }
-
- /**
- * {@inheritDoc}
- *
- * @return {@code true}. The byte size for a big endian long is a constant.
- */
- @Override
- public boolean isRegisterByteSizeObserverCheap(Instant value, Context context) {
- return longCoder.isRegisterByteSizeObserverCheap(
- ORDER_PRESERVING_CONVERTER.convert(value), context);
- }
-
- @Override
- public void registerByteSizeObserver(
- Instant value, ElementByteSizeObserver observer, Context context) throws Exception {
- longCoder.registerByteSizeObserver(
- ORDER_PRESERVING_CONVERTER.convert(value), observer, context);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/IterableCoder.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/IterableCoder.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/IterableCoder.java
deleted file mode 100644
index 70dcd84..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/IterableCoder.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.coders;
-
-import static com.google.cloud.dataflow.sdk.util.Structs.addBoolean;
-
-import com.google.cloud.dataflow.sdk.util.CloudObject;
-import com.google.cloud.dataflow.sdk.util.PropertyNames;
-import com.google.common.base.Preconditions;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-import java.util.List;
-
-/**
- * An {@link IterableCoder} encodes any {@link Iterable} in the format
- * of {@link IterableLikeCoder}.
- *
- * @param <T> the type of the elements of the iterables being transcoded
- */
-public class IterableCoder<T> extends IterableLikeCoder<T, Iterable<T>> {
-
- public static <T> IterableCoder<T> of(Coder<T> elemCoder) {
- return new IterableCoder<>(elemCoder);
- }
-
- /////////////////////////////////////////////////////////////////////////////
- // Internal operations below here.
-
- @Override
- protected final Iterable<T> decodeToIterable(List<T> decodedElements) {
- return decodedElements;
- }
-
- @JsonCreator
- public static IterableCoder<?> of(
- @JsonProperty(PropertyNames.COMPONENT_ENCODINGS)
- List<Coder<?>> components) {
- Preconditions.checkArgument(components.size() == 1,
- "Expecting 1 component, got " + components.size());
- return of(components.get(0));
- }
-
- /**
- * Returns the first element in this iterable if it is non-empty,
- * otherwise returns {@code null}.
- */
- public static <T> List<Object> getInstanceComponents(
- Iterable<T> exampleValue) {
- return getInstanceComponentsHelper(exampleValue);
- }
-
- protected IterableCoder(Coder<T> elemCoder) {
- super(elemCoder, "Iterable");
- }
-
- @Override
- public CloudObject asCloudObject() {
- CloudObject result = super.asCloudObject();
- addBoolean(result, PropertyNames.IS_STREAM_LIKE, true);
- return result;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/IterableLikeCoder.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/IterableLikeCoder.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/IterableLikeCoder.java
deleted file mode 100644
index 7fb573a..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/IterableLikeCoder.java
+++ /dev/null
@@ -1,278 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.coders;
-
-import com.google.cloud.dataflow.sdk.util.BufferedElementCountingOutputStream;
-import com.google.cloud.dataflow.sdk.util.VarInt;
-import com.google.cloud.dataflow.sdk.util.common.ElementByteSizeObservableIterable;
-import com.google.cloud.dataflow.sdk.util.common.ElementByteSizeObserver;
-import com.google.common.base.Preconditions;
-
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.List;
-import java.util.Observable;
-import java.util.Observer;
-
-/**
- * An abstract base class with functionality for assembling a
- * {@link Coder} for a class that implements {@code Iterable}.
- *
- * <p>To complete a subclass, implement the {@link #decodeToIterable} method. This superclass
- * will decode the elements in the input stream into a {@link List} and then pass them to that
- * method to be converted into the appropriate iterable type. Note that this means the input
- * iterables must fit into memory.
- *
- * <p>The format of this coder is as follows:
- *
- * <ul>
- * <li>If the input {@link Iterable} has a known and finite size, then the size is written to the
- * output stream in big endian format, followed by all of the encoded elements.</li>
- * <li>If the input {@link Iterable} is not known to have a finite size, then each element
- * of the input is preceded by {@code true} encoded as a byte (indicating "more data")
- * followed by the encoded element, and terminated by {@code false} encoded as a byte.</li>
- * </ul>
- *
- * @param <T> the type of the elements of the {@code Iterable}s being transcoded
- * @param <IterableT> the type of the Iterables being transcoded
- */
-public abstract class IterableLikeCoder<T, IterableT extends Iterable<T>>
- extends StandardCoder<IterableT> {
- public Coder<T> getElemCoder() {
- return elementCoder;
- }
-
- /**
- * Builds an instance of {@code IterableT}, this coder's associated {@link Iterable}-like
- * subtype, from a list of decoded elements.
- */
- protected abstract IterableT decodeToIterable(List<T> decodedElements);
-
- /////////////////////////////////////////////////////////////////////////////
- // Internal operations below here.
-
- private final Coder<T> elementCoder;
- private final String iterableName;
-
- /**
- * Returns the first element in the iterable-like {@code exampleValue} if it is non-empty,
- * otherwise returns {@code null}.
- */
- protected static <T, IterableT extends Iterable<T>>
- List<Object> getInstanceComponentsHelper(IterableT exampleValue) {
- for (T value : exampleValue) {
- return Arrays.<Object>asList(value);
- }
- return null;
- }
-
- protected IterableLikeCoder(Coder<T> elementCoder, String iterableName) {
- Preconditions.checkArgument(elementCoder != null,
- "element Coder for IterableLikeCoder must not be null");
- Preconditions.checkArgument(iterableName != null,
- "iterable name for IterableLikeCoder must not be null");
- this.elementCoder = elementCoder;
- this.iterableName = iterableName;
- }
-
- @Override
- public void encode(
- IterableT iterable, OutputStream outStream, Context context)
- throws IOException, CoderException {
- if (iterable == null) {
- throw new CoderException("cannot encode a null " + iterableName);
- }
- Context nestedContext = context.nested();
- DataOutputStream dataOutStream = new DataOutputStream(outStream);
- if (iterable instanceof Collection) {
- // We can know the size of the Iterable. Use an encoding with a
- // leading size field, followed by that many elements.
- Collection<T> collection = (Collection<T>) iterable;
- dataOutStream.writeInt(collection.size());
- for (T elem : collection) {
- elementCoder.encode(elem, dataOutStream, nestedContext);
- }
- } else {
- // We don't know the size without traversing it so use a fixed size buffer
- // and encode as many elements as possible into it before outputting the size followed
- // by the elements.
- dataOutStream.writeInt(-1);
- BufferedElementCountingOutputStream countingOutputStream =
- new BufferedElementCountingOutputStream(dataOutStream);
- for (T elem : iterable) {
- countingOutputStream.markElementStart();
- elementCoder.encode(elem, countingOutputStream, nestedContext);
- }
- countingOutputStream.finish();
- }
- // Make sure all our output gets pushed to the underlying outStream.
- dataOutStream.flush();
- }
-
- @Override
- public IterableT decode(InputStream inStream, Context context)
- throws IOException, CoderException {
- Context nestedContext = context.nested();
- DataInputStream dataInStream = new DataInputStream(inStream);
- int size = dataInStream.readInt();
- if (size >= 0) {
- List<T> elements = new ArrayList<>(size);
- for (int i = 0; i < size; i++) {
- elements.add(elementCoder.decode(dataInStream, nestedContext));
- }
- return decodeToIterable(elements);
- } else {
- List<T> elements = new ArrayList<>();
- long count;
- // We don't know the size a priori. Check if we're done with
- // each block of elements.
- while ((count = VarInt.decodeLong(dataInStream)) > 0) {
- while (count > 0) {
- elements.add(elementCoder.decode(dataInStream, nestedContext));
- count -= 1;
- }
- }
- return decodeToIterable(elements);
- }
- }
-
- @Override
- public List<? extends Coder<?>> getCoderArguments() {
- return Arrays.asList(elementCoder);
- }
-
- /**
- * {@inheritDoc}
- *
- * @throws NonDeterministicException always.
- * Encoding is not deterministic for the general {@link Iterable} case, as it depends
- * upon the type of iterable. This may allow two objects to compare as equal
- * while the encoding differs.
- */
- @Override
- public void verifyDeterministic() throws NonDeterministicException {
- throw new NonDeterministicException(this,
- "IterableLikeCoder can not guarantee deterministic ordering.");
- }
-
- /**
- * {@inheritDoc}
- *
- * @return {@code true} if the iterable is of a known class that supports lazy counting
- * of byte size, since that requires minimal extra computation.
- */
- @Override
- public boolean isRegisterByteSizeObserverCheap(
- IterableT iterable, Context context) {
- return iterable instanceof ElementByteSizeObservableIterable;
- }
-
- @Override
- public void registerByteSizeObserver(
- IterableT iterable, ElementByteSizeObserver observer, Context context)
- throws Exception {
- if (iterable == null) {
- throw new CoderException("cannot encode a null Iterable");
- }
- Context nestedContext = context.nested();
-
- if (iterable instanceof ElementByteSizeObservableIterable) {
- observer.setLazy();
- ElementByteSizeObservableIterable<?, ?> observableIterable =
- (ElementByteSizeObservableIterable<?, ?>) iterable;
- observableIterable.addObserver(
- new IteratorObserver(observer, iterable instanceof Collection));
- } else {
- if (iterable instanceof Collection) {
- // We can know the size of the Iterable. Use an encoding with a
- // leading size field, followed by that many elements.
- Collection<T> collection = (Collection<T>) iterable;
- observer.update(4L);
- for (T elem : collection) {
- elementCoder.registerByteSizeObserver(elem, observer, nestedContext);
- }
- } else {
- // TODO: Update to use an accurate count depending on size and count, currently we
- // are under estimating the size by up to 10 bytes per block of data since we are
- // not encoding the count prefix which occurs at most once per 64k of data and is upto
- // 10 bytes long. Since we include the total count we can upper bound the underestimate
- // to be 10 / 65536 ~= 0.0153% of the actual size.
- observer.update(4L);
- long count = 0;
- for (T elem : iterable) {
- count += 1;
- elementCoder.registerByteSizeObserver(elem, observer, nestedContext);
- }
- if (count > 0) {
- // Update the length based upon the number of counted elements, this helps
- // eliminate the case where all the elements are encoded in the first block and
- // it is quite short (e.g. Long.MAX_VALUE nulls encoded with VoidCoder).
- observer.update(VarInt.getLength(count));
- }
- // Update with the terminator byte.
- observer.update(1L);
- }
- }
- }
-
- /**
- * An observer that gets notified when an observable iterator
- * returns a new value. This observer just notifies an outerObserver
- * about this event. Additionally, the outerObserver is notified
- * about additional separators that are transparently added by this
- * coder.
- */
- private class IteratorObserver implements Observer {
- private final ElementByteSizeObserver outerObserver;
- private final boolean countable;
-
- public IteratorObserver(ElementByteSizeObserver outerObserver,
- boolean countable) {
- this.outerObserver = outerObserver;
- this.countable = countable;
-
- if (countable) {
- // Additional 4 bytes are due to size.
- outerObserver.update(4L);
- } else {
- // Additional 5 bytes are due to size = -1 (4 bytes) and
- // hasNext = false (1 byte).
- outerObserver.update(5L);
- }
- }
-
- @Override
- public void update(Observable obs, Object obj) {
- if (!(obj instanceof Long)) {
- throw new AssertionError("unexpected parameter object");
- }
-
- if (countable) {
- outerObserver.update(obs, obj);
- } else {
- // Additional 1 byte is due to hasNext = true flag.
- outerObserver.update(obs, 1 + (long) obj);
- }
- }
- }
-}