/*
 * Copyright 2016-2020 the original author or authors.
 *
 * 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
 *
 *      https://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.springframework.data.util;

import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.stream.Collector;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

import org.springframework.util.Assert;

Simple interface to ease streamability of Iterables. 
Author:
Oliver Gierke, Christoph Strobl
Since:
2.0/**
 * Simple interface to ease streamability of {@link Iterable}s.
 *
 * @author Oliver Gierke
 * @author Christoph Strobl
 * @since 2.0
 */
@FunctionalInterface
public interface Streamable<T> extends Iterable<T>, Supplier<Stream<T>> {

	
Returns an empty Streamable. 
Returns:
will never be null./**
	 * Returns an empty {@link Streamable}.
	 *
	 * @return will never be {@literal null}.
	 */
	static <T> Streamable<T> empty() {
		return Collections::emptyIterator;
	}

	
Returns a Streamable with the given elements. 
Params:
t – the elements to return.
Returns:
/**
	 * Returns a {@link Streamable} with the given elements.
	 *
	 * @param t the elements to return.
	 * @return
	 */
	@SafeVarargs
	static <T> Streamable<T> of(T... t) {
		return () -> Arrays.asList(t).iterator();
	}

	
Returns a Streamable for the given Iterable. 
Params:
iterable – must not be null.
Returns:
/**
	 * Returns a {@link Streamable} for the given {@link Iterable}.
	 *
	 * @param iterable must not be {@literal null}.
	 * @return
	 */
	static <T> Streamable<T> of(Iterable<T> iterable) {

		Assert.notNull(iterable, "Iterable must not be null!");

		return iterable::iterator;
	}

	static <T> Streamable<T> of(Supplier<? extends Stream<T>> supplier) {
		return LazyStreamable.of(supplier);
	}

	
Creates a non-parallel Stream of the underlying Iterable. 
Returns:
will never be null./**
	 * Creates a non-parallel {@link Stream} of the underlying {@link Iterable}.
	 *
	 * @return will never be {@literal null}.
	 */
	default Stream<T> stream() {
		return StreamSupport.stream(spliterator(), false);
	}

	
Returns a new Streamable that will apply the given Function to the current one. 
Params:
mapper – must not be null.
See Also:
Stream.map(Function)
Returns:
/**
	 * Returns a new {@link Streamable} that will apply the given {@link Function} to the current one.
	 *
	 * @param mapper must not be {@literal null}.
	 * @return
	 * @see Stream#map(Function)
	 */
	default <R> Streamable<R> map(Function<? super T, ? extends R> mapper) {

		Assert.notNull(mapper, "Mapping function must not be null!");

		return Streamable.of(() -> stream().map(mapper));
	}

	
Returns a new Streamable that will apply the given Function to the current one. 
Params:
mapper – must not be null.
See Also:
Stream.flatMap(Function)
Returns:
/**
	 * Returns a new {@link Streamable} that will apply the given {@link Function} to the current one.
	 *
	 * @param mapper must not be {@literal null}.
	 * @return
	 * @see Stream#flatMap(Function)
	 */
	default <R> Streamable<R> flatMap(Function<? super T, ? extends Stream<? extends R>> mapper) {

		Assert.notNull(mapper, "Mapping function must not be null!");

		return Streamable.of(() -> stream().flatMap(mapper));
	}

	
Returns a new Streamable that will apply the given filter Predicate to the current one. 
Params:
predicate – must not be null.
See Also:
Stream.filter(Predicate)
Returns:
/**
	 * Returns a new {@link Streamable} that will apply the given filter {@link Predicate} to the current one.
	 *
	 * @param predicate must not be {@literal null}.
	 * @return
	 * @see Stream#filter(Predicate)
	 */
	default Streamable<T> filter(Predicate<? super T> predicate) {

		Assert.notNull(predicate, "Filter predicate must not be null!");

		return Streamable.of(() -> stream().filter(predicate));
	}

	
Returns whether the current Streamable is empty. 
Returns:
/**
	 * Returns whether the current {@link Streamable} is empty.
	 * 
	 * @return
	 */
	default boolean isEmpty() {
		return !iterator().hasNext();
	}

	
Creates a new Streamable from the current one and the given Stream concatenated. 
Params:
stream – must not be null.
Returns:
Since:
2.1/**
	 * Creates a new {@link Streamable} from the current one and the given {@link Stream} concatenated.
	 * 
	 * @param stream must not be {@literal null}.
	 * @return
	 * @since 2.1
	 */
	default Streamable<T> and(Supplier<? extends Stream<? extends T>> stream) {

		Assert.notNull(stream, "Stream must not be null!");

		return Streamable.of(() -> Stream.concat(this.stream(), stream.get()));
	}

	
Creates a new Streamable from the current one and the given values concatenated. 
Params:
others – must not be null.
Returns:
will never be null.
Since:
2.2/**
	 * Creates a new {@link Streamable} from the current one and the given values concatenated.
	 * 
	 * @param others must not be {@literal null}.
	 * @return will never be {@literal null}.
	 * @since 2.2
	 */
	@SuppressWarnings("unchecked")
	default Streamable<T> and(T... others) {

		Assert.notNull(others, "Other values must not be null!");

		return Streamable.of(() -> Stream.concat(this.stream(), Arrays.stream(others)));
	}

	
Creates a new Streamable from the current one and the given Iterable concatenated. 
Params:
iterable – must not be null.
Returns:
will never be null.
Since:
2.2/**
	 * Creates a new {@link Streamable} from the current one and the given {@link Iterable} concatenated.
	 * 
	 * @param iterable must not be {@literal null}.
	 * @return will never be {@literal null}.
	 * @since 2.2
	 */
	default Streamable<T> and(Iterable<? extends T> iterable) {

		Assert.notNull(iterable, "Iterable must not be null!");

		return Streamable.of(() -> Stream.concat(this.stream(), StreamSupport.stream(iterable.spliterator(), false)));
	}

	
Convenience method to allow adding a Streamable directly as otherwise the invocation is ambiguous between and(Iterable) and and(Supplier). 
Params:
streamable – must not be null.
Returns:
will never be null.
Since:
2.2/**
	 * Convenience method to allow adding a {@link Streamable} directly as otherwise the invocation is ambiguous between
	 * {@link #and(Iterable)} and {@link #and(Supplier)}.
	 * 
	 * @param streamable must not be {@literal null}.
	 * @return will never be {@literal null}.
	 * @since 2.2
	 */
	default Streamable<T> and(Streamable<? extends T> streamable) {
		return and((Supplier<? extends Stream<? extends T>>) streamable);
	}

	
Creates a new, unmodifiable List. 
Returns:
will never be null.
Since:
2.2/**
	 * Creates a new, unmodifiable {@link List}.
	 * 
	 * @return will never be {@literal null}.
	 * @since 2.2
	 */
	default List<T> toList() {
		return stream().collect(StreamUtils.toUnmodifiableList());
	}

	
Creates a new, unmodifiable Set. 
Returns:
will never be null.
Since:
2.2/**
	 * Creates a new, unmodifiable {@link Set}.
	 * 
	 * @return will never be {@literal null}.
	 * @since 2.2
	 */
	default Set<T> toSet() {
		return stream().collect(StreamUtils.toUnmodifiableSet());
	}

	/*
	 * (non-Javadoc)
	 * @see java.util.function.Supplier#get()
	 */
	default Stream<T> get() {
		return stream();
	}

	
A collector to easily produce a Streamable from a Stream using Collectors.toList as intermediate collector. 
See Also:
toStreamable(Collector<Object,?,Iterable<Object>>)
Returns:
Since:
2.2/**
	 * A collector to easily produce a {@link Streamable} from a {@link Stream} using {@link Collectors#toList} as
	 * intermediate collector.
	 * 
	 * @return
	 * @see #toStreamable(Collector)
	 * @since 2.2
	 */
	public static <S> Collector<S, ?, Streamable<S>> toStreamable() {
		return toStreamable(Collectors.toList());
	}

	
A collector to easily produce a Streamable from a Stream and the given intermediate collector. 
Returns:
Since:
2.2/**
	 * A collector to easily produce a {@link Streamable} from a {@link Stream} and the given intermediate collector.
	 * 
	 * @return
	 * @since 2.2
	 */
	@SuppressWarnings("unchecked")
	public static <S, T extends Iterable<S>> Collector<S, ?, Streamable<S>> toStreamable(
			Collector<S, ?, T> intermediate) {

		return Collector.of( //
				(Supplier<T>) intermediate.supplier(), //
				(BiConsumer<T, S>) intermediate.accumulator(), //
				(BinaryOperator<T>) intermediate.combiner(), //
				Streamable::of);
	}
}