/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.beam.sdk.extensions.euphoria.core.translate;

import javax.annotation.Nullable;
import org.apache.beam.sdk.extensions.euphoria.core.client.functional.UnaryFunction;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TypeDescriptors;

/**
 * A transform for extracting stamp and applying a user-defined transform on the extracted
 * collection.
 *
 * @param <InputT> input type
 * @param <OutputT> output type
 */
public class TimestampExtractTransform<InputT, OutputT>
    extends PTransform<PCollection<InputT>, PCollection<OutputT>> {

  /**
   * Create transformation working on KVs with timestamps as key.
   *
   * @param <InputT> type of input elements
   * @param <OutputT> type of output elements
   * @param transform the transform applied on elements with timestamp
   * @return the transform
   */
  public static <InputT, OutputT> TimestampExtractTransform<InputT, OutputT> of(
      PCollectionTransform<InputT, OutputT> transform) {

    return new TimestampExtractTransform<>(null, transform);
  }

  /**
   * Create transformation working on KVs with timestamps as key.
   *
   * @param <InputT> type of input elements
   * @param <OutputT> type of output elements
   * @param name name of the transform
   * @param transform the transform applied on elements with timestamp
   * @return the transform
   */
  public static <InputT, OutputT> TimestampExtractTransform<InputT, OutputT> of(
      String name, PCollectionTransform<InputT, OutputT> transform) {

    return new TimestampExtractTransform<>(name, transform);
  }

  /**
   * Apply user defined transform(s) to input {@link PCollection} and return output one.
   *
   * @param <InputT> input type
   * @param <OutputT> output type
   */
  @FunctionalInterface
  public interface PCollectionTransform<InputT, OutputT>
      extends UnaryFunction<PCollection<KV<Long, InputT>>, PCollection<OutputT>> {}

  private static class Wrap<T> extends DoFn<T, KV<Long, T>> {

    @ProcessElement
    public void processElement(ProcessContext ctx) {
      ctx.output(KV.of(ctx.timestamp().getMillis(), ctx.element()));
    }
  }

  private static class Unwrap<T> extends DoFn<KV<Long, T>, T> {

    @ProcessElement
    public void processElement(ProcessContext ctx) {
      ctx.output(ctx.element().getValue());
    }
  }

  private final PCollectionTransform<InputT, OutputT> timestampedTransform;

  private TimestampExtractTransform(
      @Nullable String name, PCollectionTransform<InputT, OutputT> timestampedTransform) {

    super(name);
    this.timestampedTransform = timestampedTransform;
  }

  @Override
  public PCollection<OutputT> expand(PCollection<InputT> input) {
    PCollection<KV<Long, InputT>> in;
    in = input.apply(getName("wrap"), ParDo.of(new Wrap<>()));
    if (input.getTypeDescriptor() != null) {
      in =
          in.setTypeDescriptor(
              TypeDescriptors.kvs(TypeDescriptors.longs(), input.getTypeDescriptor()));
    }
    return timestampedTransform.apply(in);
  }

  private String getName(String suffix) {
    return name != null ? name + "::" + suffix : suffix;
  }
}
