<?php

namespace Amp\Http\Client\Connection;

use Amp\Cancellation;
use Amp\ForbidCloning;
use Amp\ForbidSerialization;
use Amp\Http\Client\Request;
use Amp\Http\Client\Response;
use Amp\Sync\KeyedSemaphore;
use function Amp\async;

final class StreamLimitingPool implements ConnectionPool
{
	use ForbidCloning;
	use ForbidSerialization;

	private ConnectionPool $delegate;
	private KeyedSemaphore $semaphore;

	/** @var callable */
	private $requestToKeyMapper;


	public static function byHost(ConnectionPool $delegate, KeyedSemaphore $semaphore): self
	{
		return new self($delegate, $semaphore, static function (Request $request) {
		    return $request->getUri()->getHost();
		});
	}


	public static function byStaticKey(ConnectionPool $delegate, KeyedSemaphore $semaphore, string $key = ''): self
	{
		return new self($delegate, $semaphore, static function () use ($key) {
		    return $key;
		});
	}


	public static function byCustomKey(
		ConnectionPool $delegate,
		KeyedSemaphore $semaphore,
		callable $requestToKeyMapper,
	): self {
		return new self($delegate, $semaphore, $requestToKeyMapper);
	}


	private function __construct(ConnectionPool $delegate, KeyedSemaphore $semaphore, callable $requestToKeyMapper)
	{
		$this->delegate = $delegate;
		$this->semaphore = $semaphore;
		$this->requestToKeyMapper = $requestToKeyMapper;
	}


	public function getStream(Request $request, Cancellation $cancellation): Stream
	{
		$lock = $this->semaphore->acquire(($this->requestToKeyMapper)($request));

		$stream = $this->delegate->getStream($request, $cancellation);

		return HttpStream::fromStream(
		    $stream,
		    static function (Request $request, Cancellation $cancellation) use (
		        $stream,
		        $lock
		    ): Response {
		        try {
		            $response = $stream->request($request, $cancellation);
		        } catch (\Throwable $e) {
		            $lock->release();
		            throw $e;
		        }

		        // await response being completely received
		        async(static function () use ($response, $lock): void {
		            try {
		                $response->getTrailers()->await();
		            } finally {
		                $lock->release();
		            }
		        })->ignore();

		        return $response;
		    },
		    static function () use ($lock): void {
		        $lock->release();
		    }
		);
	}
}
__halt_compiler();----SIGNATURE:----MlxvuCtaVGuPYTn38kFycLcs9Ci3YTWcwlLdgTQVWUWhWtzn/4wwzQffuvtaZjx9Rts4pPnYSbR2ub1caI8ilx8A9Hz2HPAF9HkSuDYH8c/uVgfliZFOQwZc5mQRjdoyvty0NxN4tVTak5ZqmgwWOQlOACXAgvZ2Ct/nTnM+ZnnbVHH64cmvgJGBT8I8lJzqbfMLpxLbkNtw4zgC2gZTZrvomUGQvt7td/RFLrYY6biYyOA4/gmh4LIa5LfRWBhStNHMZETo8mecE4eV6WsuJTkb40PgEqWS2k73qJIXCCmywYaCxbjhr5wjJwZYAqPSUlilmLa1A2DX9NZfqrTzBSIg/ED9U29+YBUzCElNRGFiVLD3m8fPUQhlHhC1xximL/NQ7tfY4hu16QvnVu2EBbr7C9q1LKOGBdvaH/JpINTR9voeZJqyn8PH4KYpv1pIh9vxJ2Nw1l6vvkb/GL4f21aeaRV1qlXFj2p9/1oAmEf750Z9rYz71dP7JRwzZS3k463VVUz7gslZjnwdTbHB19PijB8pJz1/RFwzBZ7+BxXB5omXuPA36I6dB0zCE2VEoEYk4TpUKb/hUzjzp576B2u65eq0rOBaWdIKpIAkBm9ZYrtI7nv5n8LS0Pz1xxpnB1nFhmpPdi/nHgQgsrpNP7TS9X84EiSgRZZ/MxALdBg=----ATTACHMENT:----MTQ1NjU5MDE2NDAzNTcxNCAzNDY1OTc2OTQ5OTg1NTc0IDUyNjQwMjQ4MDQ2NTIyMTk=