<commit_before>/*******************************************************************************
 * Copyright 2015, The IKANOW Open Source Project.
 * 
 * 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.ikanow.aleph2.data_model.utils;

import java.util.Collection;
import java.util.Collections;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.BiFunction;
import java.util.function.Function;

import scala.Tuple2;
import scala.concurrent.Await;
import scala.concurrent.duration.Duration;

import com.ikanow.aleph2.data_model.objects.shared.BasicMessageBean;
 
/** Utilities relating to the specific use of futures within Aleph2
 * @author acp
 *
 */
public class FutureUtils {

	/** Wraps a scala Future in a completable future 
	 * @param f the scala Future
	 * @return the CompletableFuture
	 */
	@SuppressWarnings("unchecked")
	public static <T> CompletableFuture<T> wrap(final scala.concurrent.Future<Object> f) {
		// Note the beginnings of a better way are here: http://onoffswitch.net/converting-akka-scala-futures-java-futures/
		// but needs completing (basically: register a callback with the scala future, when that completes call complete/completeExceptionally)
		// In the meantime this works but clogs up one of the threads from the common pool
		
		return CompletableFuture.supplyAsync(() -> {
			try {
				if (f.isCompleted()) {
					return (T)f.value().get().get();
				}
				else {
					try {
						return (T) Await.result(f, Duration.Inf());
					} catch (Exception e) {
						throw new ExecutionException(e);
					}
				}
			}
			catch (Exception e) {
				throw new RuntimeException(e);
			}
		});
	}
	
	/** Creates a trivial management future with no side channel - ie equivalent to the input future itself
	 * @return the management future (just containing the base future)
	 */
	public static <T> ManagementFuture<T> createManagementFuture(final CompletableFuture<T> base_future) {
		return new ManagementFuture<T>(base_future, Optional.empty());
	}
	
	/** Creates a trivial management future with the specified side channel - the future interface describes the primary action, the future returned by getManagementResults describes secondary results generated by actions spawned by the primary actions
	 * @return the management future (just containing the base future)
	 */
	public static <T> ManagementFuture<T> createManagementFuture(final CompletableFuture<T> base_future, 
															final CompletableFuture<Collection<BasicMessageBean>> secondary_results)
	{
		return new ManagementFuture<T>(base_future, Optional.of(secondary_results));
	}
	
	/** A wrapper for a future that enables decoration with a management side channel
	 * @author acp
	 *
	 * @param <T> - the contents of the original wrapper
	 */
	public static class ManagementFuture<T> extends CompletableFuture<T> {

		/** User constructor
		 * @param delegate - the result of the primary action
		 * @param side_channel - a ist of message beans generated by the management systems while perform the primary action
		 */
		protected ManagementFuture(final CompletableFuture<T> delegate, final Optional<CompletableFuture<Collection<BasicMessageBean>>> side_channel) {
			_delegate = delegate.whenComplete((value, exception) -> {
				if (null != exception) this.completeExceptionally(exception);
				else this.complete(value);
			});
			_management_side_channel = side_channel;
		}
		
		/** Internal function to handle user error handling
		 * @param error_handler - an error handler provided hte main channel's throwable + the side channels
		 * @param channel - the side channel
		 * @return - a function compatible with standard completable futures
		 */
		private static <X> Function<Throwable,? extends X> errorHandling(
				final BiFunction<Throwable, CompletableFuture<Collection<BasicMessageBean>>, ? extends X> error_handler,
				final Optional<CompletableFuture<Collection<BasicMessageBean>>> channel)
		{
			return err -> error_handler.apply(err, channel
					.orElse(CompletableFuture.completedFuture(Collections.<BasicMessageBean>emptyList())));
		}
		
		/** Returns a version of the management future that provides an error handling routine that provides the side channel also
		 * @param error_handler - a user error handler that also gets the side channel
		 * @return - the future stripped of its side channel 
		 */
		public  CompletableFuture<T> exceptionallyWithSideChannel(final BiFunction<Throwable, CompletableFuture<Collection<BasicMessageBean>>, ? extends T> error_handler) {
			return this.exceptionally(ManagementFuture.<T>errorHandling(error_handler, _management_side_channel));
		}
		/** Returns a version of the management future that provides an error handling routine that provides the side channel also, AND ALSO the future is combined with its side channel
		 * @param error_handler - an optional user error handler that also gets the side channel
		 * @return A version of the future complete only when both the main result and the side channel is complete, and providing both
		 */
		public CompletableFuture<Tuple2<T, Collection<BasicMessageBean>>> combineWithSideChannel(final Optional<
				BiFunction<Throwable, CompletableFuture<Collection<BasicMessageBean>>, Tuple2<T, Collection<BasicMessageBean>>>> error_handler)
		{
			final CompletableFuture<Tuple2<T, Collection<BasicMessageBean>>> intermediate = 
				this.<Collection<BasicMessageBean>, Tuple2<T, Collection<BasicMessageBean>>>thenCombine(this._management_side_channel
						.orElse(CompletableFuture.completedFuture(Collections.<BasicMessageBean>emptyList())),
						(t, msgs) -> Tuples._2T(t, msgs))					
						;
			return error_handler.isPresent() 
					? intermediate.exceptionally(ManagementFuture.errorHandling(error_handler.get(), _management_side_channel))
					: intermediate;
		}
		
		protected final CompletableFuture<T> _delegate;
		protected final Optional<CompletableFuture<Collection<BasicMessageBean>>> _management_side_channel;
		
		/* (non-Javadoc)
		 * @see java.util.concurrent.Future#cancel(boolean)
		 */
		@Override
		public boolean cancel(boolean mayInterruptIfRunning) {
			return _delegate.cancel(mayInterruptIfRunning);
		}

		/* (non-Javadoc)
		 * @see java.util.concurrent.Future#isCancelled()
		 */
		@Override
		public boolean isCancelled() {
			return _delegate.isCancelled();
		}

		/** If this future has a management side channel, returns that (otherwise an immediately completable future to an empty list)
		 * @return the management side channel future
		 */
		public CompletableFuture<Collection<BasicMessageBean>> getManagementResults() {
			return _management_side_channel.orElse(CompletableFuture.completedFuture(Collections.emptyList()));
		}		
	}

	/** De-nests a completablefuture<managementfuture> back into a management future
	 * @return the de-nested management future
	 */
	public static <T> ManagementFuture<T> denestManagementFuture(final CompletableFuture<ManagementFuture<T>> nested_mgmt_future) {
		return createManagementFuture(
				nested_mgmt_future.<T>thenApply(mf -> mf.join()), 
				nested_mgmt_future.thenApply(mf -> mf.getManagementResults().join()));
	}
	
	/** Generates a future that will error as soon as it's touched
	 * @param e - the underlying exception
	 * @return a future that errors when touched
	 */
	public static <T> CompletableFuture<T> returnError(final Exception e) {
		return new CompletableFuture<T>() {
			{
				this.completeExceptionally(e);
			}
		};		
	}	
}<commit_msg>ALEPH-19 Tweaked akka completion converter to be purely scala for akka-independent data model<commit_after>/*******************************************************************************
 * Copyright 2015, The IKANOW Open Source Project.
 * 
 * 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.
 * 
 * Note contains code from http://onoffswitch.net/converting-akka-scala-futures-java-futures/
 * Listed below, not covered by this license
 * 
 ******************************************************************************/
package com.ikanow.aleph2.data_model.utils;

import java.util.Collection;

import java.util.Collections;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.BiFunction;
import java.util.function.Function;

import scala.Function1;
import scala.Tuple2;
import scala.concurrent.Await;
import scala.concurrent.ExecutionContext;
import scala.concurrent.Future;
import scala.concurrent.duration.Duration;
import scala.runtime.AbstractFunction1;
import scala.runtime.BoxedUnit;
import scala.util.Try;

import com.ikanow.aleph2.data_model.objects.shared.BasicMessageBean;
 
/** Utilities relating to the specific use of futures within Aleph2
 * Contains code fragments from http://onoffswitch.net/converting-akka-scala-futures-java-futures/
 * @author acp
 *
 */
public class FutureUtils {

	@SuppressWarnings("unchecked")
	public static <T> CompletableFuture<T> wrap(final scala.concurrent.Future<Object> f) {
		// Note the beginnings of a better way are here: http://onoffswitch.net/converting-akka-scala-futures-java-futures/
		// but needs completing (basically: register a callback with the scala future, when that completes call complete/completeExceptionally)
		// In the meantime this works but clogs up one of the threads from the common pool
		
		return CompletableFuture.supplyAsync(() -> {
			try {
				if (f.isCompleted()) {
					return (T)f.value().get().get();
				}
				else {
					try {
						return (T) Await.result(f, Duration.Inf());
					} catch (Exception e) {
						throw new ExecutionException(e);
					}
				}
			}
			catch (Exception e) {
				throw new RuntimeException(e);
			}
		});
	}

	/** Wraps a scala Future in a completable future efficiently using the underlying scala infrastructure
	 * Code from http://onoffswitch.net/converting-akka-scala-futures-java-futures/ 
	 * @param f the scala Future
	 * @return the CompletableFuture
	 */
	@SuppressWarnings("unchecked")
	public static <T> CompletableFuture<T> efficientWrap(final scala.concurrent.Future<Object> f, final ExecutionContext execution_context) {
		return (CompletableFuture<T>) new FromScalaFuture<>(f).executeOn(execution_context);
	}	
	
	/** Creates a trivial management future with no side channel - ie equivalent to the input future itself
	 * @return the management future (just containing the base future)
	 */
	public static <T> ManagementFuture<T> createManagementFuture(final CompletableFuture<T> base_future) {
		return new ManagementFuture<T>(base_future, Optional.empty());
	}
	
	/** Creates a trivial management future with the specified side channel - the future interface describes the primary action, the future returned by getManagementResults describes secondary results generated by actions spawned by the primary actions
	 * @return the management future (just containing the base future)
	 */
	public static <T> ManagementFuture<T> createManagementFuture(final CompletableFuture<T> base_future, 
															final CompletableFuture<Collection<BasicMessageBean>> secondary_results)
	{
		return new ManagementFuture<T>(base_future, Optional.of(secondary_results));
	}
	
	/** A wrapper for a future that enables decoration with a management side channel
	 * @author acp
	 *
	 * @param <T> - the contents of the original wrapper
	 */
	public static class ManagementFuture<T> extends CompletableFuture<T> {

		/** User constructor
		 * @param delegate - the result of the primary action
		 * @param side_channel - a ist of message beans generated by the management systems while perform the primary action
		 */
		protected ManagementFuture(final CompletableFuture<T> delegate, final Optional<CompletableFuture<Collection<BasicMessageBean>>> side_channel) {
			_delegate = delegate.whenComplete((value, exception) -> {
				if (null != exception) this.completeExceptionally(exception);
				else this.complete(value);
			});
			_management_side_channel = side_channel;
		}
		
		/** Internal function to handle user error handling
		 * @param error_handler - an error handler provided hte main channel's throwable + the side channels
		 * @param channel - the side channel
		 * @return - a function compatible with standard completable futures
		 */
		private static <X> Function<Throwable,? extends X> errorHandling(
				final BiFunction<Throwable, CompletableFuture<Collection<BasicMessageBean>>, ? extends X> error_handler,
				final Optional<CompletableFuture<Collection<BasicMessageBean>>> channel)
		{
			return err -> error_handler.apply(err, channel
					.orElse(CompletableFuture.completedFuture(Collections.<BasicMessageBean>emptyList())));
		}
		
		/** Returns a version of the management future that provides an error handling routine that provides the side channel also
		 * @param error_handler - a user error handler that also gets the side channel
		 * @return - the future stripped of its side channel 
		 */
		public  CompletableFuture<T> exceptionallyWithSideChannel(final BiFunction<Throwable, CompletableFuture<Collection<BasicMessageBean>>, ? extends T> error_handler) {
			return this.exceptionally(ManagementFuture.<T>errorHandling(error_handler, _management_side_channel));
		}
		/** Returns a version of the management future that provides an error handling routine that provides the side channel also, AND ALSO the future is combined with its side channel
		 * @param error_handler - an optional user error handler that also gets the side channel
		 * @return A version of the future complete only when both the main result and the side channel is complete, and providing both
		 */
		public CompletableFuture<Tuple2<T, Collection<BasicMessageBean>>> combineWithSideChannel(final Optional<
				BiFunction<Throwable, CompletableFuture<Collection<BasicMessageBean>>, Tuple2<T, Collection<BasicMessageBean>>>> error_handler)
		{
			final CompletableFuture<Tuple2<T, Collection<BasicMessageBean>>> intermediate = 
				this.<Collection<BasicMessageBean>, Tuple2<T, Collection<BasicMessageBean>>>thenCombine(this._management_side_channel
						.orElse(CompletableFuture.completedFuture(Collections.<BasicMessageBean>emptyList())),
						(t, msgs) -> Tuples._2T(t, msgs))					
						;
			return error_handler.isPresent() 
					? intermediate.exceptionally(ManagementFuture.errorHandling(error_handler.get(), _management_side_channel))
					: intermediate;
		}
		
		protected final CompletableFuture<T> _delegate;
		protected final Optional<CompletableFuture<Collection<BasicMessageBean>>> _management_side_channel;
		
		/* (non-Javadoc)
		 * @see java.util.concurrent.Future#cancel(boolean)
		 */
		@Override
		public boolean cancel(boolean mayInterruptIfRunning) {
			return _delegate.cancel(mayInterruptIfRunning);
		}

		/* (non-Javadoc)
		 * @see java.util.concurrent.Future#isCancelled()
		 */
		@Override
		public boolean isCancelled() {
			return _delegate.isCancelled();
		}

		/** If this future has a management side channel, returns that (otherwise an immediately completable future to an empty list)
		 * @return the management side channel future
		 */
		public CompletableFuture<Collection<BasicMessageBean>> getManagementResults() {
			return _management_side_channel.orElse(CompletableFuture.completedFuture(Collections.emptyList()));
		}		
	}

	/** De-nests a completablefuture<managementfuture> back into a management future
	 * @return the de-nested management future
	 */
	public static <T> ManagementFuture<T> denestManagementFuture(final CompletableFuture<ManagementFuture<T>> nested_mgmt_future) {
		return createManagementFuture(
				nested_mgmt_future.<T>thenApply(mf -> mf.join()), 
				nested_mgmt_future.thenApply(mf -> mf.getManagementResults().join()));
	}
	
	/** Generates a future that will error as soon as it's touched
	 * @param e - the underlying exception
	 * @return a future that errors when touched
	 */
	public static <T> CompletableFuture<T> returnError(final Exception e) {
		return new CompletableFuture<T>() {
			{
				this.completeExceptionally(e);
			}
		};		
	}	
	////////////////////////////////////////////
	
	// SCALA CONVERSION UTILS
	// Code from http://onoffswitch.net/converting-akka-scala-futures-java-futures/
	
	/**Code from http://onoffswitch.net/converting-akka-scala-futures-java-futures/
	 */
	protected static class FromScalaFuture<T> {
		 
	    private final Future<T> future;
	 
	    public FromScalaFuture(Future<T> future) {
	        this.future = future;
	    }
	 
	    public CompletableFuture<T> executeOn(ExecutionContext context) {
	        final CompletableFuture<T> completableFuture = new CompletableFuture<>();
	 
	        Function1<Try<T>, BoxedUnit> f = new AbstractFunction1<Try<T>, BoxedUnit>() {
	        	public BoxedUnit apply(Try<T> in) {
	        		if (in.isSuccess()) {
		                completableFuture.complete(in.get());	        			
	        		}
	        		else {
	        			completableFuture.completeExceptionally(in.failed().get());
	        		}	        		
	        		return BoxedUnit.UNIT;
	        	}
	        	
	        };	        
	        future.onComplete(f, context);	 
	        return completableFuture;
	    }
	}		
}
{'fraction_non_alphanumeric': 0.10741408512265749, 'fraction_numerical': 0.0021854340818445064, 'mean_word_length': 4.344233576642336, 'pattern_counts': {'":': 0, '<': 151, '<?xml version=': 0, '>': 163, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6616514', 'n_tokens_mistral': 5459, 'n_tokens_neox': 4794, 'n_words': 1505}