/*
 * 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.tsfile.external.commons.io.function;

import org.apache.tsfile.external.commons.io.IOExceptionList;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.Optional;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiFunction;
import java.util.function.IntFunction;
import java.util.function.ToDoubleFunction;
import java.util.function.ToIntFunction;
import java.util.function.ToLongFunction;
import java.util.function.UnaryOperator;
import java.util.stream.Collector;
import java.util.stream.DoubleStream;
import java.util.stream.IntStream;
import java.util.stream.LongStream;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

/**
 * Like {@link Stream} but throws {@link IOException}.
 *
 * @param <T> the type of the stream elements.
 * @since 2.12.0
 */
public interface IOStream<T> extends IOBaseStream<T, IOStream<T>, Stream<T>> {

  /**
   * Constructs a new IOStream for the given Stream.
   *
   * @param <T> the type of the stream elements.
   * @param stream The stream to delegate.
   * @return a new IOStream.
   */
  static <T> IOStream<T> adapt(final Stream<T> stream) {
    return IOStreamAdapter.adapt(stream);
  }

  /**
   * This class' version of {@link Stream#empty()}.
   *
   * @param <T> the type of the stream elements
   * @return an empty sequential {@code IOStreamImpl}.
   * @see Stream#empty()
   */
  static <T> IOStream<T> empty() {
    return IOStreamAdapter.adapt(Stream.empty());
  }

  /**
   * Like {@link Stream#iterate(Object, UnaryOperator)} but for IO.
   *
   * @param <T> the type of stream elements.
   * @param seed the initial element.
   * @param f a function to be applied to the previous element to produce a new element.
   * @return a new sequential {@code IOStream}.
   */
  static <T> IOStream<T> iterate(final T seed, final IOUnaryOperator<T> f) {
    Objects.requireNonNull(f);
    final Iterator<T> iterator =
        new Iterator<T>() {
          @SuppressWarnings("unchecked")
          T t = (T) IOStreams.NONE;

          @Override
          public boolean hasNext() {
            return true;
          }

          @Override
          public T next() throws NoSuchElementException {
            try {
              return t = t == IOStreams.NONE ? seed : f.apply(t);
            } catch (final IOException e) {
              final NoSuchElementException nsee = new NoSuchElementException();
              nsee.initCause(e);
              throw nsee;
            }
          }
        };
    return adapt(
        StreamSupport.stream(
            Spliterators.spliteratorUnknownSize(
                iterator, Spliterator.ORDERED | Spliterator.IMMUTABLE),
            false));
  }

  /**
   * Null-safe version of {@link StreamSupport#stream(java.util.Spliterator, boolean)}.
   *
   * <p>Copied from Apache Commons Lang.
   *
   * @param <T> the type of stream elements.
   * @param values the elements of the new stream, may be {@code null}.
   * @return the new stream on {@code values} or {@link Stream#empty()}.
   */
  static <T> IOStream<T> of(final Iterable<T> values) {
    return values == null ? empty() : adapt(StreamSupport.stream(values.spliterator(), false));
  }

  /**
   * Null-safe version of {@link Stream#of(Object[])} for an IO stream.
   *
   * @param <T> the type of stream elements.
   * @param values the elements of the new stream, may be {@code null}.
   * @return the new stream on {@code values} or {@link Stream#empty()}.
   */
  @SafeVarargs // Creating a stream from an array is safe
  static <T> IOStream<T> of(final T... values) {
    return values == null || values.length == 0 ? empty() : adapt(Arrays.stream(values));
  }

  /**
   * Returns a sequential {@code IOStreamImpl} containing a single element.
   *
   * @param t the single element
   * @param <T> the type of stream elements
   * @return a singleton sequential stream
   */
  static <T> IOStream<T> of(final T t) {
    return adapt(Stream.of(t));
  }

  /**
   * Like {@link Stream#allMatch(java.util.function.Predicate)} but throws {@link IOException}.
   *
   * @param predicate {@link Stream#allMatch(java.util.function.Predicate)}.
   * @return Like {@link Stream#allMatch(java.util.function.Predicate)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default boolean allMatch(final IOPredicate<? super T> predicate) throws IOException {
    return unwrap().allMatch(t -> Erase.test(predicate, t));
  }

  /**
   * Like {@link Stream#anyMatch(java.util.function.Predicate)} but throws {@link IOException}.
   *
   * @param predicate {@link Stream#anyMatch(java.util.function.Predicate)}.
   * @return Like {@link Stream#anyMatch(java.util.function.Predicate)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default boolean anyMatch(final IOPredicate<? super T> predicate) throws IOException {
    return unwrap().anyMatch(t -> Erase.test(predicate, t));
  }

  /**
   * TODO Package-private for now, needs IOCollector?
   *
   * <p>Adding this method now and an IO version later is an issue because call sites would have to
   * type-cast to pick one. It would be ideal to have only one.
   *
   * <p>Like {@link Stream#collect(Collector)}.
   *
   * <p>Package private for now.
   *
   * @param <R> Like {@link Stream#collect(Collector)}.
   * @param <A> Like {@link Stream#collect(Collector)}.
   * @param collector Like {@link Stream#collect(Collector)}.
   * @return Like {@link Stream#collect(Collector)}.
   */
  default <R, A> R collect(final Collector<? super T, A, R> collector) {
    return unwrap().collect(collector);
  }

  /**
   * Like {@link Stream#collect(java.util.function.Supplier, java.util.function.BiConsumer,
   * java.util.function.BiConsumer)}.
   *
   * @param <R> Like {@link Stream#collect(java.util.function.Supplier,
   *     java.util.function.BiConsumer, java.util.function.BiConsumer)}.
   * @param supplier Like {@link Stream#collect(java.util.function.Supplier,
   *     java.util.function.BiConsumer, java.util.function.BiConsumer)}.
   * @param accumulator Like {@link Stream#collect(java.util.function.Supplier,
   *     java.util.function.BiConsumer, java.util.function.BiConsumer)}.
   * @param combiner Like {@link Stream#collect(java.util.function.Supplier,
   *     java.util.function.BiConsumer, java.util.function.BiConsumer)}.
   * @return Like {@link Stream#collect(java.util.function.Supplier, java.util.function.BiConsumer,
   *     java.util.function.BiConsumer)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default <R> R collect(
      final IOSupplier<R> supplier,
      final IOBiConsumer<R, ? super T> accumulator,
      final IOBiConsumer<R, R> combiner)
      throws IOException {
    return unwrap()
        .collect(
            () -> Erase.get(supplier),
            (t, u) -> Erase.accept(accumulator, t, u),
            (t, u) -> Erase.accept(combiner, t, u));
  }

  /**
   * Like {@link Stream#count()}.
   *
   * @return Like {@link Stream#count()}.
   */
  default long count() {
    return unwrap().count();
  }

  /**
   * Like {@link Stream#distinct()}.
   *
   * @return Like {@link Stream#distinct()}.
   */
  default IOStream<T> distinct() {
    return adapt(unwrap().distinct());
  }

  /**
   * Like {@link Stream#filter(java.util.function.Predicate)}.
   *
   * @param predicate Like {@link Stream#filter(java.util.function.Predicate)}.
   * @return Like {@link Stream#filter(java.util.function.Predicate)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default IOStream<T> filter(final IOPredicate<? super T> predicate) throws IOException {
    return adapt(unwrap().filter(t -> Erase.test(predicate, t)));
  }

  /**
   * Like {@link Stream#findAny()}.
   *
   * @return Like {@link Stream#findAny()}.
   */
  default Optional<T> findAny() {
    return unwrap().findAny();
  }

  /**
   * Like {@link Stream#findFirst()}.
   *
   * @return Like {@link Stream#findFirst()}.
   */
  default Optional<T> findFirst() {
    return unwrap().findFirst();
  }

  /**
   * Like {@link Stream#flatMap(java.util.function.Function)}.
   *
   * @param <R> Like {@link Stream#flatMap(java.util.function.Function)}.
   * @param mapper Like {@link Stream#flatMap(java.util.function.Function)}.
   * @return Like {@link Stream#flatMap(java.util.function.Function)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings({"unused", "resource"}) // thrown by Erase; resource closed by caller.
  default <R> IOStream<R> flatMap(
      final IOFunction<? super T, ? extends IOStream<? extends R>> mapper) throws IOException {
    return adapt(unwrap().flatMap(t -> Erase.apply(mapper, t).unwrap()));
  }

  /**
   * TODO Package-private for now, needs IODoubleStream?
   *
   * <p>Adding this method now and an IO version later is an issue because call sites would have to
   * type-cast to pick one. It would be ideal to have only one.
   *
   * <p>Like {@link Stream#flatMapToDouble(java.util.function.Function)}.
   *
   * @param mapper Like {@link Stream#flatMapToDouble(java.util.function.Function)}.
   * @return Like {@link Stream#flatMapToDouble(java.util.function.Function)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default DoubleStream flatMapToDouble(final IOFunction<? super T, ? extends DoubleStream> mapper)
      throws IOException {
    return unwrap().flatMapToDouble(t -> Erase.apply(mapper, t));
  }

  /**
   * TODO Package-private for now, needs IOIntStream?
   *
   * <p>Adding this method now and an IO version later is an issue because call sites would have to
   * type-cast to pick one. It would be ideal to have only one.
   *
   * <p>Like {@link Stream#flatMapToInt(java.util.function.Function)}.
   *
   * @param mapper Like {@link Stream#flatMapToInt(java.util.function.Function)}.
   * @return Like {@link Stream#flatMapToInt(java.util.function.Function)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default IntStream flatMapToInt(final IOFunction<? super T, ? extends IntStream> mapper)
      throws IOException {
    return unwrap().flatMapToInt(t -> Erase.apply(mapper, t));
  }

  /**
   * TODO Package-private for now, needs IOLongStream?
   *
   * <p>Adding this method now and an IO version later is an issue because call sites would have to
   * type-cast to pick one. It would be ideal to have only one.
   *
   * <p>Like {@link Stream#flatMapToLong(java.util.function.Function)}.
   *
   * @param mapper Like {@link Stream#flatMapToLong(java.util.function.Function)}.
   * @return Like {@link Stream#flatMapToLong(java.util.function.Function)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default LongStream flatMapToLong(final IOFunction<? super T, ? extends LongStream> mapper)
      throws IOException {
    return unwrap().flatMapToLong(t -> Erase.apply(mapper, t));
  }

  /**
   * Performs an action for each element gathering any exceptions.
   *
   * @param action The action to apply to each element.
   * @throws IOExceptionList if any I/O errors occur.
   */
  default void forAll(final IOConsumer<T> action) throws IOExceptionList {
    forAll(action, (i, e) -> e);
  }

  /**
   * Performs an action for each element gathering any exceptions.
   *
   * @param action The action to apply to each element.
   * @param exSupplier The exception supplier.
   * @throws IOExceptionList if any I/O errors occur.
   */
  default void forAll(
      final IOConsumer<T> action, final BiFunction<Integer, IOException, IOException> exSupplier)
      throws IOExceptionList {
    final AtomicReference<List<IOException>> causeList = new AtomicReference<>();
    final AtomicInteger index = new AtomicInteger();
    final IOConsumer<T> safeAction = IOStreams.toIOConsumer(action);
    unwrap()
        .forEach(
            e -> {
              try {
                safeAction.accept(e);
              } catch (final IOException innerEx) {
                if (causeList.get() == null) {
                  // Only allocate if required
                  causeList.set(new ArrayList<>());
                }
                if (exSupplier != null) {
                  causeList.get().add(exSupplier.apply(index.get(), innerEx));
                }
              }
              index.incrementAndGet();
            });
    IOExceptionList.checkEmpty(causeList.get(), null);
  }

  /**
   * Like {@link Stream#forEach(java.util.function.Consumer)} but throws {@link IOException}.
   *
   * @param action Like {@link Stream#forEach(java.util.function.Consumer)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default void forEach(final IOConsumer<? super T> action) throws IOException {
    unwrap().forEach(e -> Erase.accept(action, e));
  }

  /**
   * Like {@link Stream#forEachOrdered(java.util.function.Consumer)}.
   *
   * @param action Like {@link Stream#forEachOrdered(java.util.function.Consumer)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default void forEachOrdered(final IOConsumer<? super T> action) throws IOException {
    unwrap().forEachOrdered(e -> Erase.accept(action, e));
  }

  /**
   * Like {@link Stream#limit(long)}.
   *
   * @param maxSize Like {@link Stream#limit(long)}.
   * @return Like {@link Stream#limit(long)}.
   */
  default IOStream<T> limit(final long maxSize) {
    return adapt(unwrap().limit(maxSize));
  }

  /**
   * Like {@link Stream#map(java.util.function.Function)}.
   *
   * @param <R> Like {@link Stream#map(java.util.function.Function)}.
   * @param mapper Like {@link Stream#map(java.util.function.Function)}.
   * @return Like {@link Stream#map(java.util.function.Function)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default <R> IOStream<R> map(final IOFunction<? super T, ? extends R> mapper) throws IOException {
    return adapt(unwrap().map(t -> Erase.apply(mapper, t)));
  }

  /**
   * TODO Package-private for now, needs IOToDoubleFunction?
   *
   * <p>Adding this method now and an IO version later is an issue because call sites would have to
   * type-cast to pick one. It would be ideal to have only one.
   *
   * <p>Like {@link Stream#mapToDouble(ToDoubleFunction)}.
   *
   * <p>Package private for now.
   *
   * @param mapper Like {@link Stream#mapToDouble(ToDoubleFunction)}.
   * @return Like {@link Stream#mapToDouble(ToDoubleFunction)}.
   */
  default DoubleStream mapToDouble(final ToDoubleFunction<? super T> mapper) {
    return unwrap().mapToDouble(mapper);
  }

  /**
   * TODO Package-private for now, needs IOToIntFunction?
   *
   * <p>Adding this method now and an IO version later is an issue because call sites would have to
   * type-cast to pick one. It would be ideal to have only one.
   *
   * <p>Like {@link Stream#mapToInt(ToIntFunction)}.
   *
   * <p>Package private for now.
   *
   * @param mapper Like {@link Stream#mapToInt(ToIntFunction)}.
   * @return Like {@link Stream#mapToInt(ToIntFunction)}.
   */
  default IntStream mapToInt(final ToIntFunction<? super T> mapper) {
    return unwrap().mapToInt(mapper);
  }

  /**
   * TODO Package-private for now, needs IOToLongFunction?
   *
   * <p>Adding this method now and an IO version later is an issue because call sites would have to
   * type-cast to pick one. It would be ideal to have only one.
   *
   * <p>Like {@link Stream#mapToLong(ToLongFunction)}.
   *
   * <p>Package private for now.
   *
   * @param mapper Like {@link Stream#mapToLong(ToLongFunction)}.
   * @return Like {@link Stream#mapToLong(ToLongFunction)}.
   */
  default LongStream mapToLong(final ToLongFunction<? super T> mapper) {
    return unwrap().mapToLong(mapper);
  }

  /**
   * Like {@link Stream#max(java.util.Comparator)}.
   *
   * @param comparator Like {@link Stream#max(java.util.Comparator)}.
   * @return Like {@link Stream#max(java.util.Comparator)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default Optional<T> max(final IOComparator<? super T> comparator) throws IOException {
    return unwrap().max((t, u) -> Erase.compare(comparator, t, u));
  }

  /**
   * Like {@link Stream#min(java.util.Comparator)}.
   *
   * @param comparator Like {@link Stream#min(java.util.Comparator)}.
   * @return Like {@link Stream#min(java.util.Comparator)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default Optional<T> min(final IOComparator<? super T> comparator) throws IOException {
    return unwrap().min((t, u) -> Erase.compare(comparator, t, u));
  }

  /**
   * Like {@link Stream#noneMatch(java.util.function.Predicate)}.
   *
   * @param predicate Like {@link Stream#noneMatch(java.util.function.Predicate)}.
   * @return Like {@link Stream#noneMatch(java.util.function.Predicate)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default boolean noneMatch(final IOPredicate<? super T> predicate) throws IOException {
    return unwrap().noneMatch(t -> Erase.test(predicate, t));
  }

  /**
   * Like {@link Stream#peek(java.util.function.Consumer)}.
   *
   * @param action Like {@link Stream#peek(java.util.function.Consumer)}.
   * @return Like {@link Stream#peek(java.util.function.Consumer)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default IOStream<T> peek(final IOConsumer<? super T> action) throws IOException {
    return adapt(unwrap().peek(t -> Erase.accept(action, t)));
  }

  /**
   * Like {@link Stream#reduce(java.util.function.BinaryOperator)}.
   *
   * @param accumulator Like {@link Stream#reduce(java.util.function.BinaryOperator)}.
   * @return Like {@link Stream#reduce(java.util.function.BinaryOperator)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default Optional<T> reduce(final IOBinaryOperator<T> accumulator) throws IOException {
    return unwrap().reduce((t, u) -> Erase.apply(accumulator, t, u));
  }

  /**
   * Like {@link Stream#reduce(Object, java.util.function.BinaryOperator)}.
   *
   * @param identity Like {@link Stream#reduce(Object, java.util.function.BinaryOperator)}.
   * @param accumulator Like {@link Stream#reduce(Object, java.util.function.BinaryOperator)}.
   * @return Like {@link Stream#reduce(Object, java.util.function.BinaryOperator)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default T reduce(final T identity, final IOBinaryOperator<T> accumulator) throws IOException {
    return unwrap().reduce(identity, (t, u) -> Erase.apply(accumulator, t, u));
  }

  /**
   * Like {@link Stream#reduce(Object, BiFunction, java.util.function.BinaryOperator)}.
   *
   * @param <U> Like {@link Stream#reduce(Object, BiFunction, java.util.function.BinaryOperator)}.
   * @param identity Like {@link Stream#reduce(Object, BiFunction,
   *     java.util.function.BinaryOperator)}.
   * @param accumulator Like {@link Stream#reduce(Object, BiFunction,
   *     java.util.function.BinaryOperator)}.
   * @param combiner Like {@link Stream#reduce(Object, BiFunction,
   *     java.util.function.BinaryOperator)}.
   * @return Like {@link Stream#reduce(Object, BiFunction, java.util.function.BinaryOperator)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default <U> U reduce(
      final U identity,
      final IOBiFunction<U, ? super T, U> accumulator,
      final IOBinaryOperator<U> combiner)
      throws IOException {
    return unwrap()
        .reduce(
            identity,
            (t, u) -> Erase.apply(accumulator, t, u),
            (t, u) -> Erase.apply(combiner, t, u));
  }

  /**
   * Like {@link Stream#skip(long)}.
   *
   * @param n Like {@link Stream#skip(long)}.
   * @return Like {@link Stream#skip(long)}.
   */
  default IOStream<T> skip(final long n) {
    return adapt(unwrap().skip(n));
  }

  /**
   * Like {@link Stream#sorted()}.
   *
   * @return Like {@link Stream#sorted()}.
   */
  default IOStream<T> sorted() {
    return adapt(unwrap().sorted());
  }

  /**
   * Like {@link Stream#sorted(java.util.Comparator)}.
   *
   * @param comparator Like {@link Stream#sorted(java.util.Comparator)}.
   * @return Like {@link Stream#sorted(java.util.Comparator)}.
   * @throws IOException if an I/O error occurs.
   */
  @SuppressWarnings("unused") // thrown by Erase.
  default IOStream<T> sorted(final IOComparator<? super T> comparator) throws IOException {
    return adapt(unwrap().sorted((t, u) -> Erase.compare(comparator, t, u)));
  }

  /**
   * Like {@link Stream#toArray()}.
   *
   * @return {@link Stream#toArray()}.
   */
  default Object[] toArray() {
    return unwrap().toArray();
  }

  /**
   * TODO Package-private for now, needs IOIntFunction?
   *
   * <p>Adding this method now and an IO version later is an issue because call sites would have to
   * type-cast to pick one. It would be ideal to have only one.
   *
   * <p>Like {@link Stream#toArray(IntFunction)}.
   *
   * <p>Package private for now.
   *
   * @param <A> Like {@link Stream#toArray(IntFunction)}.
   * @param generator Like {@link Stream#toArray(IntFunction)}.
   * @return Like {@link Stream#toArray(IntFunction)}.
   */
  default <A> A[] toArray(final IntFunction<A[]> generator) {
    return unwrap().toArray(generator);
  }
}
