<commit_before><?php

namespace Frosas\Http;

class Clients {
    
    private $connectionsPending = array();
    private $connectionsDone = array();

    function __construct($clients = null) {
        foreach ((array) $clients as $client) $this->add($client);
    }

    function add($client) {
        if (! $client instanceof \Zend\Http\Client) {
            $client = new \Zend\Http\Client($client);
        }

        $client->setAdapter(new Adapter\Writer);
        $client->send();

        $socket = $client->getAdapter()->socket();
        stream_set_blocking($socket, 0);

        $rawRequest = $client->getAdapter()->rawRequest();
        $client->setAdapter(new Adapter\Reader($rawRequest));

        $this->connectionsPending[(int) $socket] = array(
            'client' => $client,
            'socket' => $socket
        );

        return $client;
    }

    function next() {
        while ($this->connectionsPending && ! $this->connectionsDone) {
            $sockets = $this->sockets();
            $null = null;
            // TODO Handle timeouts
            stream_select($sockets, $null, $null, 9999);
            foreach ($sockets as $socket) {
                $connection = $this->connectionsPending[(int) $socket];
                $client = $connection['client'];
                $data = stream_get_contents($socket);
                if ($data === false) throw new \Exception;
                $client->getAdapter()->appendToResponse($data);
                if ($data === '') { // End of response
                    unset($this->connectionsPending[(int) $socket]);
                    $this->connectionsDone[] = $connection;
                }
            }
        }
        
        if ($connection = array_shift($this->connectionsDone)) {
            $client = $connection['client'];
            $client->send();
            return $client;
        }
    }
    
    private function sockets() {
        return array_map(function($client) {
            return $client['socket'];
        }, $this->connectionsPending);        
    }
}<commit_msg>Limit active connections<commit_after><?php

namespace Frosas\Http;

use Zend\Http\Client;

class Clients {

    private $connectionsQueued = array();
    private $connectionsActive = array();
    private $connectionsDone = array();
    private $maxActiveConnections = 10;

    function __construct($clients = null) {
        foreach ((array) $clients as $client) $this->add($client);
    }

    function add($client) {
        if (! $client instanceof Client) $client = new Client($client);
        $this->connectionsQueued[] = array('client' => $client);
        $this->fillActiveConnectionsQueue();
        return $client;
    }

    function next() {
        while ($this->connectionsActive && ! $this->connectionsDone) {
            $sockets = $this->sockets();
            $null = null;
            // TODO Handle timeouts
            stream_select($sockets, $null, $null, 9999);
            foreach ($sockets as $socket) {
                $connection = $this->connectionsActive[(int) $socket];
                $client = $connection['client'];
                $data = stream_get_contents($socket);
                if ($data === false) throw new \Exception;
                $client->getAdapter()->appendToResponse($data);
                if ($data === '') { // End of response
                    unset($this->connectionsActive[(int) $socket]);
                    $this->connectionsDone[] = $connection;
                    $this->fillActiveConnectionsQueue(); // Add another one
                }
            }
        }
        
        if ($connection = array_shift($this->connectionsDone)) {
            $client = $connection['client'];
            $client->send();
            return $client;
        }
    }

    function setMaxActiveConnections($max) {
        $this->maxActiveConnections = $max;
    }
    
    private function sockets() {
        return array_map(function($client) {
            return $client['socket'];
        }, $this->connectionsActive);        
    }

    private function fillActiveConnectionsQueue() {
        while ($this->connectionsQueued && count($this->connectionsActive) < $this->maxActiveConnections) {
            $connection = array_shift($this->connectionsQueued);

            // Connect + send request
            $client = $connection['client'];
            $client->setAdapter(new Adapter\Writer);
            $client->send();

            // Prepare client for receiving response
            $socket = $client->getAdapter()->socket();
            $connection['socket'] = $socket;
            stream_set_blocking($socket, 0);
            $rawRequest = $client->getAdapter()->rawRequest();
            $client->setAdapter(new Adapter\Reader($rawRequest));

            $this->connectionsActive[(int) $socket] = $connection;
        }
    }
}
{'fraction_non_alphanumeric': 0.14279843846311896, 'fraction_numerical': 0.002465584549003493, 'mean_word_length': 2.008652657601978, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 54, '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': '2238736', 'n_tokens_mistral': 1292, 'n_tokens_neox': 1224, 'n_words': 255}