<commit_before>require 'concurrent/promises'

module Concurrent
  module Promises
    module FactoryMethods
      # only proof of concept
      # @return [Future]
      def select(*channels)
        # TODO (pitr-ch 26-Mar-2016): redo, has to be non-blocking
        future do
          # noinspection RubyArgCount
          Channel.select do |s|
            channels.each do |ch|
              s.take(ch) { |value| [value, ch] }
            end
          end
        end
      end
    end

    class Future < Event
      # Zips with selected value form the suplied channels
      # @return [Future]
      def then_select(*channels)
        ZipFuturesPromise.new([self, Concurrent::Promises.select(*channels)], @DefaultExecutor).future
      end

      # @note may block
      # @note only proof of concept
      def then_put(channel)
        on_success(:io) { |value| channel.put value }
      end

      # Asks the actor with its value.
      # @return [Future] new future with the response form the actor
      def then_ask(actor)
        self.then { |v| actor.ask(v) }.flat
      end

      # TODO (pitr-ch 14-Mar-2016): document, and move to core
      def run(terminated = Promises.future)
        on_completion do |success, value, reason|
          if success
            if value.is_a?(Future)
              value.run terminated
            else
              terminated.success value
            end
          else
            terminated.fail reason
          end
        end
      end

      include Enumerable

      def each(&block)
        each_body self.value, &block
      end

      def each!(&block)
        each_body self.value!, &block
      end

      private

      def each_body(value, &block)
        (value.nil? ? [nil] : Array(value)).each(&block)
      end

    end
  end

  # inspired by https://msdn.microsoft.com/en-us/library/dd537607(v=vs.110).aspx
  class Cancellation < Synchronization::Object
    safe_initialization!

    def self.create(future_or_event = Promises.completable_event, *complete_args)
      [(i = new(future_or_event, *complete_args)), i.token]
    end

    private_class_method :new

    def initialize(future, *complete_args)
      raise ArgumentError, 'future is not Completable' unless future.is_a?(Promises::Completable)
      @Cancel       = future
      @Token        = Token.new @Cancel.with_hidden_completable
      @CompleteArgs = complete_args
    end

    def token
      @Token
    end

    def cancel
      try_cancel or raise MultipleAssignmentError, 'cannot cancel twice'
    end

    def try_cancel
      !!@Cancel.complete(false)
    end

    def canceled?
      @Cancel.complete?
    end

    class Token < Synchronization::Object
      safe_initialization!

      def initialize(cancel)
        @Cancel = cancel
      end

      def event
        @Cancel
      end

      alias_method :future, :event

      def on_cancellation(*args, &block)
        @Cancel.on_completion *args, &block
      end

      def then(*args, &block)
        @Cancel.chain *args, &block
      end<commit_msg>Cancellation unify cancel methods<commit_after>require 'concurrent/promises'

module Concurrent
  module Promises
    module FactoryMethods
      # only proof of concept
      # @return [Future]
      def select(*channels)
        # TODO (pitr-ch 26-Mar-2016): redo, has to be non-blocking
        future do
          # noinspection RubyArgCount
          Channel.select do |s|
            channels.each do |ch|
              s.take(ch) { |value| [value, ch] }
            end
          end
        end
      end
    end

    class Future < Event
      # Zips with selected value form the suplied channels
      # @return [Future]
      def then_select(*channels)
        ZipFuturesPromise.new([self, Concurrent::Promises.select(*channels)], @DefaultExecutor).future
      end

      # @note may block
      # @note only proof of concept
      def then_put(channel)
        on_success(:io) { |value| channel.put value }
      end

      # Asks the actor with its value.
      # @return [Future] new future with the response form the actor
      def then_ask(actor)
        self.then { |v| actor.ask(v) }.flat
      end

      # TODO (pitr-ch 14-Mar-2016): document, and move to core
      def run(terminated = Promises.future)
        on_completion do |success, value, reason|
          if success
            if value.is_a?(Future)
              value.run terminated
            else
              terminated.success value
            end
          else
            terminated.fail reason
          end
        end
      end

      include Enumerable

      def each(&block)
        each_body self.value, &block
      end

      def each!(&block)
        each_body self.value!, &block
      end

      private

      def each_body(value, &block)
        (value.nil? ? [nil] : Array(value)).each(&block)
      end

    end
  end

  # inspired by https://msdn.microsoft.com/en-us/library/dd537607(v=vs.110).aspx
  class Cancellation < Synchronization::Object
    safe_initialization!

    def self.create(future_or_event = Promises.completable_event, *complete_args)
      [(i = new(future_or_event, *complete_args)), i.token]
    end

    private_class_method :new

    def initialize(future, *complete_args)
      raise ArgumentError, 'future is not Completable' unless future.is_a?(Promises::Completable)
      @Cancel       = future
      @Token        = Token.new @Cancel.with_hidden_completable
      @CompleteArgs = complete_args
    end

    def token
      @Token
    end

    def cancel(raise_on_repeated_call = true)
      !!@Cancel.complete(*@CompleteArgs, raise_on_repeated_call)
    end

    def canceled?
      @Cancel.complete?
    end

    class Token < Synchronization::Object
      safe_initialization!

      def initialize(cancel)
        @Cancel = cancel
      end

      def event
        @Cancel
      end

      alias_method :future, :event

      def on_cancellation(*args, &block)
        @Cancel.on_completion *args, &block
      end

      def then(*args, &block)
        @Cancel.chain *args, &block
      end
{'fraction_non_alphanumeric': 0.08455097685109178, 'fraction_numerical': 0.006895419471351174, 'mean_word_length': 2.0338645418326693, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, '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': '6936432', 'n_tokens_mistral': 1840, 'n_tokens_neox': 1737, 'n_words': 485}