<commit_before>﻿//-----------------------------------------------------------------------
// <copyright file="Retry.cs" company="Akka.NET Project">
//     Copyright (C) 2015-2017 Lightbend Inc. <http://www.lightbend.com>
//     Copyright (C) 2013-2017 Akka.NET project <https://github.com/akkadotnet/akka.net>
// </copyright>
//-----------------------------------------------------------------------

using System;
using System.Collections.Generic;
using System.Linq;
using Akka.Pattern;
using Akka.Streams.Stage;
using Akka.Util;

namespace Akka.Streams.Dsl
{
    public static class Retry
    {
        /// <summary>
        /// EXPERIMENTAL API
        /// <para>
        /// Retry flow factory. given a flow that produces <see cref="Result{T}"/>s, this wrapping flow may be used to try
        /// and pass failed elements through the flow again. More accurately, the given flow consumes a tuple
        /// of `input` and `state`, and produces a tuple of <see cref="Result{T}"/> of `output` and `state`.
        /// If the flow emits a failed element (i.e. <see cref="Result{T}.IsSuccess"/> is false), the <paramref name="retryWith"/>
        /// function is fed with the `state` of the failed element, and may produce a new input-state tuple to pass through
        /// the original flow. The function may also yield `null` instead of `(input, state)`, which means not to retry a failed element.
        /// </para>
        /// <para>
        /// IMPORTANT CAVEAT:
        /// The given flow must not change the number of elements passing through it (i.e. it should output
        /// exactly one element for every received element). Ignoring this, will have an unpredicted result,
        /// and may result in a deadlock.
        /// </para> 
        /// </summary>
        /// <param name="flow">the flow to retry</param>
        /// <param name="retryWith">if output was failure, we can optionaly recover from it,
        /// and retry with a new pair of input and new state we get from this function.</param>
        /// <typeparam name="TI">input elements type</typeparam>
        /// <typeparam name="TS">state to create a new `(I,S)` to retry with</typeparam>
        /// <typeparam name="TO">output elements type</typeparam>
        /// <typeparam name="TM">materialized value type</typeparam>
        public static IGraph<FlowShape<Tuple<TI, TS>, Tuple<Result<TO>, TS>>, TM> Create<TI, TS, TO, TM>(
            IGraph<FlowShape<Tuple<TI, TS>, Tuple<Result<TO>, TS>>, TM> flow, Func<TS, Tuple<TI, TS>> retryWith)
        {
            return GraphDsl.Create(flow, (b, origFlow) =>
            {
                var retry = b.Add(new RetryCoordinator<TI, TS, TO>(retryWith));

                b.From(retry.Outlet2).Via(origFlow).To(retry.Inlet2);

                return new FlowShape<Tuple<TI, TS>, Tuple<Result<TO>, TS>>(retry.Inlet1, retry.Outlet1);
            });
        }

        /// <summary>
        /// EXPERIMENTAL API
        /// <para>
        /// Factory for multiple retries flow. similar to the simple retry, but this will allow to
        /// break down a "heavy" element which failed into multiple "thin" elements, that may succeed individually.
        /// Since it's easy to inflate elements in retry cycle, there's also a limit parameter, that will limit the
        /// amount of generated elements by the `retryWith` function, and will fail the stage if that limit is exceeded.
        /// </para>
        /// <para>
        /// Passing `null` is valid, and will result in filtering out the failure quietly, without
        /// emitting a failed <see cref="Result{T}"/> element.
        /// </para>
        /// <para>
        /// IMPORTANT CAVEAT:
        /// The given flow must not change the number of elements passing through it (i.e. it should output
        ///     exactly one element for every received element). Ignoring this, will have an unpredicted result,
        /// and may result in a deadlock.
        /// </para>
        /// </summary>
        /// <param name="limit">since every retry can generate more elements, the inner queue can get too big.
        /// if the limit is reached, the stage will fail.</param>
        /// <param name="flow">the flow to retry</param>
        /// <param name="retryWith">if output was failure, we can optionaly recover from it, and retry with
        /// a sequence of input and new state pairs we get from this function.</param>
        /// <typeparam name="TI">input elements type</typeparam>
        /// <typeparam name="TS">state to create a new `(I,S)` to retry with</typeparam>
        /// <typeparam name="TO">output elements type</typeparam>
        /// <typeparam name="TM">materialized value type</typeparam>
        /// <returns></returns>
        public static IGraph<FlowShape<Tuple<TI, TS>, Tuple<Result<TO>, TS>>, TM> Concat<TI, TS, TO, TM>(long limit,
            IGraph<FlowShape<Tuple<TI, TS>, Tuple<Result<TO>, TS>>, TM> flow, Func<TS, IEnumerable<Tuple<TI, TS>>> retryWith)
        {
            return GraphDsl.Create(flow, (b, origFlow) =>
            {
                var retry = b.Add(new RetryConcatCoordinator<TI, TS, TO>(limit, retryWith));

                b.From(retry.Outlet2).Via(origFlow).To(retry.Inlet2);

                return new FlowShape<Tuple<TI, TS>, Tuple<Result<TO>, TS>>(retry.Inlet1, retry.Outlet1);<commit_msg>add experimental api attribute<commit_after>﻿//-----------------------------------------------------------------------
// <copyright file="Retry.cs" company="Akka.NET Project">
//     Copyright (C) 2015-2017 Lightbend Inc. <http://www.lightbend.com>
//     Copyright (C) 2013-2017 Akka.NET project <https://github.com/akkadotnet/akka.net>
// </copyright>
//-----------------------------------------------------------------------

using System;
using System.Collections.Generic;
using System.Linq;
using Akka.Annotations;
using Akka.Pattern;
using Akka.Streams.Stage;
using Akka.Util;

namespace Akka.Streams.Dsl
{
    public static class Retry
    {
        /// <summary>
        /// EXPERIMENTAL API
        /// <para>
        /// Retry flow factory. given a flow that produces <see cref="Result{T}"/>s, this wrapping flow may be used to try
        /// and pass failed elements through the flow again. More accurately, the given flow consumes a tuple
        /// of `input` and `state`, and produces a tuple of <see cref="Result{T}"/> of `output` and `state`.
        /// If the flow emits a failed element (i.e. <see cref="Result{T}.IsSuccess"/> is false), the <paramref name="retryWith"/>
        /// function is fed with the `state` of the failed element, and may produce a new input-state tuple to pass through
        /// the original flow. The function may also yield `null` instead of `(input, state)`, which means not to retry a failed element.
        /// </para>
        /// <para>
        /// IMPORTANT CAVEAT:
        /// The given flow must not change the number of elements passing through it (i.e. it should output
        /// exactly one element for every received element). Ignoring this, will have an unpredicted result,
        /// and may result in a deadlock.
        /// </para> 
        /// </summary>
        /// <param name="flow">the flow to retry</param>
        /// <param name="retryWith">if output was failure, we can optionaly recover from it,
        /// and retry with a new pair of input and new state we get from this function.</param>
        /// <typeparam name="TI">input elements type</typeparam>
        /// <typeparam name="TS">state to create a new `(I,S)` to retry with</typeparam>
        /// <typeparam name="TO">output elements type</typeparam>
        /// <typeparam name="TM">materialized value type</typeparam>
        [ApiMayChange]
        public static IGraph<FlowShape<Tuple<TI, TS>, Tuple<Result<TO>, TS>>, TM> Create<TI, TS, TO, TM>(
            IGraph<FlowShape<Tuple<TI, TS>, Tuple<Result<TO>, TS>>, TM> flow, Func<TS, Tuple<TI, TS>> retryWith)
        {
            return GraphDsl.Create(flow, (b, origFlow) =>
            {
                var retry = b.Add(new RetryCoordinator<TI, TS, TO>(retryWith));

                b.From(retry.Outlet2).Via(origFlow).To(retry.Inlet2);

                return new FlowShape<Tuple<TI, TS>, Tuple<Result<TO>, TS>>(retry.Inlet1, retry.Outlet1);
            });
        }

        /// <summary>
        /// EXPERIMENTAL API
        /// <para>
        /// Factory for multiple retries flow. similar to the simple retry, but this will allow to
        /// break down a "heavy" element which failed into multiple "thin" elements, that may succeed individually.
        /// Since it's easy to inflate elements in retry cycle, there's also a limit parameter, that will limit the
        /// amount of generated elements by the `retryWith` function, and will fail the stage if that limit is exceeded.
        /// </para>
        /// <para>
        /// Passing `null` is valid, and will result in filtering out the failure quietly, without
        /// emitting a failed <see cref="Result{T}"/> element.
        /// </para>
        /// <para>
        /// IMPORTANT CAVEAT:
        /// The given flow must not change the number of elements passing through it (i.e. it should output
        ///     exactly one element for every received element). Ignoring this, will have an unpredicted result,
        /// and may result in a deadlock.
        /// </para>
        /// </summary>
        /// <param name="limit">since every retry can generate more elements, the inner queue can get too big.
        /// if the limit is reached, the stage will fail.</param>
        /// <param name="flow">the flow to retry</param>
        /// <param name="retryWith">if output was failure, we can optionaly recover from it, and retry with
        /// a sequence of input and new state pairs we get from this function.</param>
        /// <typeparam name="TI">input elements type</typeparam>
        /// <typeparam name="TS">state to create a new `(I,S)` to retry with</typeparam>
        /// <typeparam name="TO">output elements type</typeparam>
        /// <typeparam name="TM">materialized value type</typeparam>
        [ApiMayChange]
        public static IGraph<FlowShape<Tuple<TI, TS>, Tuple<Result<TO>, TS>>, TM> Concat<TI, TS, TO, TM>(long limit,
            IGraph<FlowShape<Tuple<TI, TS>, Tuple<Result<TO>, TS>>, TM> flow, Func<TS, IEnumerable<Tuple<TI, TS>>> retryWith)
        {
            return GraphDsl.Create(flow, (b, origFlow) =>
            {
                var retry = b.Add(new RetryConcatCoordinator<TI, TS, TO>(limit, retryWith));

                b.From(retry.Outlet2).Via(origFlow).To(retry.Inlet2);

                return new FlowShape<Tuple<TI, TS>, Tuple<Result<TO>, TS>>(retry.Inlet1, retry.Outlet1);
{'fraction_non_alphanumeric': 0.1557064459120082, 'fraction_numerical': 0.004464700958050414, 'mean_word_length': 2.1801242236024843, 'pattern_counts': {'":': 0, '<': 177, '<?xml version=': 0, '>': 181, 'https://': 2, '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': '2536722', 'n_tokens_mistral': 3042, 'n_tokens_neox': 2949, 'n_words': 1077}