<commit_before>my class Proc::Async { ... }

my role X::Proc::Async is Exception {
    has Proc::Async $.proc;
}

my class X::Proc::Async::TapBeforeSpawn does X::Proc::Async {
    has $.handle;
    method message() {
        "To avoid data races, you must tap $!handle before running the process"
    }
}

my class X::Proc::Async::SupplyOrStd does X::Proc::Async {
    method message() {
        "Using .Supply on a Proc::Async implies merging stdout and stderr; .stdout " ~
            "and .stderr cannot therefore be used in combination with it"
    }
}

my class X::Proc::Async::BindOrUse does X::Proc::Async {
    has $.handle;
    has $.use;
    method message() {
        "Cannot both bind $.handle to a handle and also $.use"
    }
}

my class X::Proc::Async::CharsOrBytes does X::Proc::Async {
    has $.handle;
    method message() {
        "Can only tap one of chars or bytes supply for $!handle"
    }
}

my class X::Proc::Async::AlreadyStarted does X::Proc::Async {
    method message() {
        "Process has already been started"
    }
}

my class X::Proc::Async::MustBeStarted does X::Proc::Async {
    has $.method;
    method message() {
        "Process must be started first before calling '$!method'"
    }
}

my class X::Proc::Async::OpenForWriting does X::Proc::Async {
    has $.method;
    method message() {
        "Process must be opened for writing with :w to call '$!method'"
    }
}

my class Proc::Async {
    # An asynchornous process output pipe is a Supply that also can provide
    # the native descriptor of the underlying pipe.
    class Pipe is Supply {
        my class PermitOnTap does Tappable {
            has Tappable $.delegate;
            has &.on-tap;
            method tap(|c) {
                &!on-tap();
                $!delegate.tap(|c)
            }
            method live() { self.delegate.live }
            method serial() { self.delegate.serial }
            method sane() { self.delegate.sane }
        }

        has Promise $.native-descriptor;
        has &!on-nd-used;

        submethod BUILD(:$!native-descriptor!, :&!on-nd-used) {}

        method native-descriptor() {
            &!on-nd-used();
            $!native-descriptor
        }

        method new($delegate, $native-descriptor, &on-tap, &on-nd-used) {
            self.bless(
                tappable => PermitOnTap.bless(:$delegate, :&on-tap),
                :$native-descriptor, :&on-nd-used
            )
        }
    }

    my class ProcessCancellation is repr('AsyncTask') { }
    my enum  CharsOrBytes ( :Bytes(0), :Chars(1) );

    has $!ready_promise = Promise.new;
    has $!ready_vow = $!ready_promise.vow;
    has $!handle_available_promise = Promise.new;
    has $!stdout_descriptor_vow;
    has $!stderr_descriptor_vow;
    has $!stdout_descriptor_used = Promise.new;
    has $!stderr_descriptor_used = Promise.new;
    has $.path;
    has @.args;
    has $.w;
    has $.enc = 'utf8';
    has $.translate-nl = True;
    has Bool $.started = False;
    has $!stdout_supply;
    has CharsOrBytes $!stdout_type;
    has $!stderr_supply;
    has CharsOrBytes $!stderr_type;
    has $!merge_supply;
    has CharsOrBytes $!merge_type;
    has $!stdin-fd;
    has $!stdout-fd;
    has $!stderr-fd;
    has $!process_handle;
    has $!exit_promise;
    has @!promises;
    has $!encoder;
    has @!close-after-exit;

    method new($path, *@args) { self.bless(:$path, :@args, |%_) }

    submethod TWEAK(--> Nil) {
        $!encoder := Encoding::Registry.find($!enc).encoder(:$!translate-nl);
    }

    method !pipe-cbs(\channel) {
        -> { $!handle_available_promise.then({ nqp::permit($!process_handle, channel, -1) }) },
        -> { (channel == 1 ?? $!stdout_descriptor_used !! $!stderr_descriptor_used).keep(True) }
    }

    method !pipe(\what, \the-supply, \type, \value, \fd-vow, \permit-channel) {
        X::Proc::Async::TapBeforeSpawn.new(handle => what, proc => self).throw
          if $!started;
        X::Proc::Async::CharsOrBytes.new(handle => what, proc => self).throw
          if the-supply and type != value;

        type         = value;
        the-supply //= Supplier::Preserving.new;

        if nqp::iscont(fd-vow) {<commit_msg>Revert "Minimally invasive streamline Proc::Async.new"<commit_after>my class Proc::Async { ... }

my role X::Proc::Async is Exception {
    has Proc::Async $.proc;
}

my class X::Proc::Async::TapBeforeSpawn does X::Proc::Async {
    has $.handle;
    method message() {
        "To avoid data races, you must tap $!handle before running the process"
    }
}

my class X::Proc::Async::SupplyOrStd does X::Proc::Async {
    method message() {
        "Using .Supply on a Proc::Async implies merging stdout and stderr; .stdout " ~
            "and .stderr cannot therefore be used in combination with it"
    }
}

my class X::Proc::Async::BindOrUse does X::Proc::Async {
    has $.handle;
    has $.use;
    method message() {
        "Cannot both bind $.handle to a handle and also $.use"
    }
}

my class X::Proc::Async::CharsOrBytes does X::Proc::Async {
    has $.handle;
    method message() {
        "Can only tap one of chars or bytes supply for $!handle"
    }
}

my class X::Proc::Async::AlreadyStarted does X::Proc::Async {
    method message() {
        "Process has already been started"
    }
}

my class X::Proc::Async::MustBeStarted does X::Proc::Async {
    has $.method;
    method message() {
        "Process must be started first before calling '$!method'"
    }
}

my class X::Proc::Async::OpenForWriting does X::Proc::Async {
    has $.method;
    method message() {
        "Process must be opened for writing with :w to call '$!method'"
    }
}

my class Proc::Async {
    # An asynchornous process output pipe is a Supply that also can provide
    # the native descriptor of the underlying pipe.
    class Pipe is Supply {
        my class PermitOnTap does Tappable {
            has Tappable $.delegate;
            has &.on-tap;
            method tap(|c) {
                &!on-tap();
                $!delegate.tap(|c)
            }
            method live() { self.delegate.live }
            method serial() { self.delegate.serial }
            method sane() { self.delegate.sane }
        }

        has Promise $.native-descriptor;
        has &!on-nd-used;

        submethod BUILD(:$!native-descriptor!, :&!on-nd-used) {}

        method native-descriptor() {
            &!on-nd-used();
            $!native-descriptor
        }

        method new($delegate, $native-descriptor, &on-tap, &on-nd-used) {
            self.bless(
                tappable => PermitOnTap.bless(:$delegate, :&on-tap),
                :$native-descriptor, :&on-nd-used
            )
        }
    }

    my class ProcessCancellation is repr('AsyncTask') { }
    my enum  CharsOrBytes ( :Bytes(0), :Chars(1) );

    has $!ready_promise = Promise.new;
    has $!ready_vow = $!ready_promise.vow;
    has $!handle_available_promise = Promise.new;
    has $!stdout_descriptor_vow;
    has $!stderr_descriptor_vow;
    has $!stdout_descriptor_used = Promise.new;
    has $!stderr_descriptor_used = Promise.new;
    has $.path;
    has @.args;
    has $.w;
    has $.enc = 'utf8';
    has $.translate-nl = True;
    has Bool $.started = False;
    has $!stdout_supply;
    has CharsOrBytes $!stdout_type;
    has $!stderr_supply;
    has CharsOrBytes $!stderr_type;
    has $!merge_supply;
    has CharsOrBytes $!merge_type;
    has $!stdin-fd;
    has $!stdout-fd;
    has $!stderr-fd;
    has $!process_handle;
    has $!exit_promise;
    has @!promises;
    has $!encoder;
    has @!close-after-exit;

    proto method new(|) { * }
    multi method new(*@args where .so) {
        my $path = @args.shift;
        self.bless(:$path, :@args, |%_)
    }

    submethod TWEAK(--> Nil) {
        $!encoder := Encoding::Registry.find($!enc).encoder(:$!translate-nl);
    }

    method !pipe-cbs(\channel) {
        -> { $!handle_available_promise.then({ nqp::permit($!process_handle, channel, -1) }) },
        -> { (channel == 1 ?? $!stdout_descriptor_used !! $!stderr_descriptor_used).keep(True) }
    }

    method !pipe(\what, \the-supply, \type, \value, \fd-vow, \permit-channel) {
        X::Proc::Async::TapBeforeSpawn.new(handle => what, proc => self).throw
          if $!started;
        X::Proc::Async::CharsOrBytes.new(handle => what, proc => self).throw
          if the-supply and type != value;

        type         = value;
        the-supply //= Supplier::Preserving.new;

        if nqp::iscont(fd-vow) {
{'fraction_non_alphanumeric': 0.14340994371482177, 'fraction_numerical': 0.0011726078799249532, 'mean_word_length': 2.4116, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 19, 'https://': 0, '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': '1269965', 'n_tokens_mistral': 2810, 'n_tokens_neox': 2721, 'n_words': 708}