### ----------------------------------------------------
### -- Cro::HTTP
### -- Licenses: Artistic-2.0
### -- Authors: Jonathan Worthington <jnthn@jnthn.net>
### -- File: lib/Cro/HTTP/Client.rakumod
### ----------------------------------------------------
### -- Chunk 2 of 2
 && $http ne '1.1'),
                    ((self ?? (self.ca // %$ca) !! %$ca) // Empty);
        }
        my $in = Supplier::Preserving.new;
        my $out = $version-decision
            ?? establish($connector, $in.Supply, $log-connection, :nodelay, :$host, :$port, :$conn-timeout, |%tls-config)
            !! do {
                my $s = Supplier::Preserving.new;
                establish($connector, $in.Supply, $log-connection, :nodelay, :$host, :$port, :$conn-timeout, |%tls-config).tap:
                    { $s.emit($_) },
                    done => { $s.done },
                    quit => {
                        try $version-decision.break($_);
                        $s.quit($_);
                    };
                $s.Supply
            };
        $version-decision.then: -> $version {
            $version.result eq '2'
                ?? Pipeline2.new(:$secure, :$host, :$port, :$in, :$out)
                !! Pipeline.new(:$secure, :$host, :$port, :$in, :$out)
        }
    }

    sub establish(Cro::Connector $connector, Supply $incoming, $log-connection, :$conn-timeout, *%options) {
        my $connection-obtained = False;
        supply {
            whenever Promise.in($conn-timeout) {
                die X::Cro::HTTP::Client::Timeout.new(phase => 'connection', uri => %options<host>) unless $connection-obtained;
            }

            my Promise $connection = $connector.connect(|%options);
            $connection.then({
                $connection-obtained = True;
                $log-connection.end;
            });
            whenever $connection -> Cro::Transform $transform {
                whenever $transform.transformer($incoming) -> $msg {
                    emit $msg;
                    LAST done;
                }
            }
        }
    }

    sub calculate-timeout($value, Cro::Policy::Timeout $timeout-policy is rw) {
        my $curr-timeout;
        given $value {
            when Map:D {
                my %phases;
                for $_.kv -> $k, $v {
                    %phases{$k} = Real($v);
                }
                $curr-timeout = Cro::HTTP::Client::Policy::Timeout.new(total => Inf, |%phases);
            }
            when Real:D {
                $curr-timeout = Cro::HTTP::Client::Policy::Timeout.new(total => Real($_));
            }
            when Cro::Policy::Timeout:D {
                $curr-timeout = $_;
            }
            default {
                die X::Cro::HTTP::Client::InvalidTimeoutFormat.new(bad => $_);
            }
        }
        with $curr-timeout {
            $timeout-policy = $_;
        }
    }

    method !assemble-request(Str $method, Cro::Uri $base-url, Cro::Uri $proxy-url, %options, Cro::Policy::Timeout $timeout-policy is rw, --> Cro::HTTP::Request) {
        # Add any query string parameters.
        my $url;
        with %options<query> -> $query {
            my Cro::Uri::HTTP $http-uri = $base-url ~~ Cro::Uri::HTTP
                ?? $base-url
                !! Cro::Uri::HTTP.parse($base-url.Str);
            $url = $http-uri.add-query($query.list);
        }
        else {
            $url = $base-url;
        }

        # Form target and request object.
        # If we have a proxy URL, include the host along with the path.
        my $target = ($proxy-url ?? ~$url !! $url.path) || '/';
        $target ~= "?{$url.query}" if $url.query;
        my $request-uri = $proxy-url // $url;
        my $request = Cro::HTTP::Request.new(:$method, :$target, :$request-uri);
        my $port = $url.port;
        $request.append-header('Host', $url.host ~
            ($port && $port != 80 | 443 ?? ":$port" !! ""));

        # Add defaults from the instance, if we have one.
        if self {
            self!set-headers($request, @.headers.List);
            $request.append-header('content-type', $.content-type) if $.content-type;
            $.cookie-jar.add-to-request($request, $url) if $.cookie-jar;
            if %!auth && !(%options<auth>:exists) {
                self!form-authentication($request, %!auth, %options<if-asked>:exists);
            }
        }

        # Process options.
        my Bool $body-set = False;
        for %options.kv -> $_, $value {
            when 'body' {
                if !$body-set {
                    $request.set-body($value);
                    $body-set = True;
                } else {
                    die X::Cro::HTTP::Client::BodyAlreadySet.new;
                }
            }
            when 'body-byte-stream' {
                if !$body-set {
                    $request.set-body-byte-stream($value);
                    $body-set = True;
                } else {
                    die X::Cro::HTTP::Client::BodyAlreadySet.new;
                }
            }
            when 'content-type' {
                $request.append-header('content-type', $value)
            }
            when 'headers' {
                self!set-headers($request, $value.List) if $value ~~ Iterable;
            }
            when 'auth' {
                self!form-authentication: $request, %$value, %$value<if-asked>:exists;
            }
            when 'cookies' {
                for $value.list {
                    when Cro::HTTP::Cookie {
                        $request.add-cookie($_);
                    }
                    when Pair {
                        $request.add-cookie(.key, .value);
                    }
                    default {
                        die X::Cro::HTTP::Client::InvalidCookie.new(bad => $_);
                    }
                }
            }
            when 'timeout' {
                calculate-timeout($value, $timeout-policy);
            }
        }

        my $default-timeout = Cro::HTTP::Client::Policy::Timeout.new(:total(Inf));
        ($timeout-policy = self ?? $!timeout-policy // $default-timeout !! $default-timeout) without $timeout-policy;

        # Set User-agent, check if wasn't set already for us
        unless $request.has-header('User-agent') {
            # These checks are required to skip header setting
            # if it is set to Nil or empty string by the user
            if %options<user-agent>:exists {
                with %options<user-agent> {
                    $request.append-header('User-agent', $_) if $_;
                }
            } elsif self && $!user-agent {
                $request.append-header('User-agent', $!user-agent);
            } else {
                $request.append-header('User-agent', 'Cro');
            }
        }

        return $request;
    }

    method !form-authentication($request, %data, Bool $skip) {
        return if $skip;
        if %data<username>:exists {
            my $hash = encode-base64("{%data<username>}:{%data<password>}", :str);
            $request.append-header('Authorization', "Basic $hash");
        } else {
            $request.append-header('Authorization', "Bearer {%data<bearer>}");
        }
    }

    method !set-headers($request, @headers) {
        for @headers {
            when Pair | Cro::HTTP::Header {
                $request.append-header($_)
            }
            default {
                die X::Cro::HTTP::Client::IncorrectHeaderType.new(what => $_);
            }
        }
    }

    #| Returns the underlying connector used by the HTTP client in order to make
    #| a TCP or TLS condition. Can be overridden to customize the transport used;
    #| for example, Cro::HTTP::Test uses this to fake a connection.
    method choose-connector($secure) {
        $secure ?? Cro::TLS::Connector !! Cro::TCP::Connector
    }
}


