Abstract class Libs_Multihttp

<?php
/**
 * Libs_Multihttp
 *
 * @access public
 */
abstract class Libs_Multihttp {
     
    /**
     * Фабрика
     *
     * @access public
     * @var object
     */
    public static function factory(array $uris) {
         
        return new Multihttp($uris);
    }
     
    /**
     * Конструктор Multihttp
     *
     * @access public
     * @return  void
     */   
      public function __construct($uris) {
         
        $this->_uris = $uris;
    }
     
    /**
     * URIs
     *
     * @access protected
     * @var array
     */   
    protected $_uris = array();
     
    /**
     * Ожидание ответа от сервера в сек
     *
     * @access protected
     * @var int
     */   
    protected $_request_timeout = 5;    
     
    /**
     * Число попыток сединения
     *
     * @access protected
     * @var int
     */   
    protected $_request_attempts = 5;
     
    /**
     * Задержка между соединениями в сек
     *
     * @access protected
     * @var int
     */   
    protected $_request_delay = 1;
     
    /**
     * Полученные ответы
     *
     * @access protected
     * @var array
     */   
    protected $_responses = array();
     
    /**
     * Установка настроек запроса
     *
     * @access  public
     * @param   array
     * @return  object
     */
    public function setRequestSettings($settings) {
         
        if ( isset($settings['timeout']) && is_int($settings['timeout']) ) {
            $this->_request_timeout = $settings['timeout'];
        }
         
        if ( isset($settings['attempts']) && is_int($settings['attempts']) ) {
            $this->_request_attempts = $settings['timeout'];
        }        
         
        if ( isset($settings['delay']) && is_int($settings['delay']) ) {
            $this->_request_delay = $settings['delay'];
        }        
         
        return $this;
    }
     
    /**
     * Получение настроек запроса
     *
     * @access  public
     * @return  array
     */
    public function getRequestSettings() {
         
        return array(
            'timeout'  => $this->_request_timeout,
            'attempts' => $this->_request_attempts,
            'delay'    => $this->_request_delay
        );
    }
     
    /**
     * Запуск роутера
     *
     * @access  public
     * @param   string $route
     * @return  object
     */
    public function sendRequests() {
         
        // инициализируем "контейнер" для отдельных соединений (мультикурл)
        $cmh = curl_multi_init();
 
        // массив заданий для мультикурла
        $tasks = array();
         
        // перебираем uri
        foreach ( $this->_uris as $uri ) {
             
            $uri_parts = parse_url($uri);
 
            $scheme   = isset($uri_parts['scheme'])   ? $uri_parts['scheme']   : NULL;
            $host     = isset($uri_parts['host'])     ? $uri_parts['host']     : NULL;
            $port     = isset($uri_parts['port'])     ? $uri_parts['port']     : NULL;
            $user     = isset($uri_parts['user'])     ? $uri_parts['user']     : NULL;
            $pass     = isset($uri_parts['pass'])     ? $uri_parts['pass']     : NULL;
            $path     = isset($uri_parts['path'])     ? $uri_parts['path']     : NULL;
            $query    = isset($uri_parts['query'])    ? $uri_parts['query']    : NULL;
            $fragment = isset($uri_parts['fragment']) ? $uri_parts['fragment'] : NULL;
         
            // инициализируем отдельное соединение (поток)
            $ch = curl_init($scheme . '://' . $host . ':' . $port . $path . ($query ? '?'.$query : ''));
 
            // порт соединения
            curl_setopt($ch, CURLOPT_PORT, $port);
         
            // если будет редирект - перейти по нему
            curl_setopt($ch, CURLOPT_FOLLOWLOCATION, TRUE);
             
            // возвращать результат в качетсве строки
            curl_setopt($ch, CURLOPT_RETURNTRANSFER, TRUE);
             
            // время ожидания ответа от сервера
            curl_setopt($ch, CURLOPT_TIMEOUT, $this->_request_timeout);            
             
            // не возвращать http-заголовок
            curl_setopt($ch, CURLOPT_HEADER, FALSE);
             
            // добавляем дескриптор потока в массив заданий
            $tasks[$uri] = $ch;
             
            // добавляем дескриптор потока в мультикурл
            curl_multi_add_handle($cmh, $ch);
        }
         
        $active = null;
        // запускаем выполнение потоков
        do {
            $mrc = curl_multi_exec($cmh, $active);
        }
        while ( $mrc == CURLM_CALL_MULTI_PERFORM );
          
        // выполняем, пока есть активные потоки
        while ( $active && ($mrc == CURLM_OK) ) {
            
            // если какой-либо поток готов к действиям
            if ( curl_multi_select($cmh) != -1 ) {
                 
                // ждем, пока что-нибудь изменится
                do {
                    $mrc = curl_multi_exec($cmh, $active);
                     
                    // получаем информацию о потоке
 
                    $msgs_in_queue = 1;
 
                    while ( $msgs_in_queue ) {
                        $msgs_in_queue = $this->_setResponses($tasks, $cmh);
                    }
 
                } while ( $mrc == CURLM_CALL_MULTI_PERFORM );
            }
        }
         
        // закрываем мультикурл
        curl_multi_close($cmh);
 
        return $this;
         
    }
 
 
     
    /**
     * Сохранение результатов ответа
     *
     * @access  private
     * @param   array $tasks
     * @param   object $cmh
     * @return  int
     */
    private function _setResponses($tasks, $cmh) {
         
        $returnValue = 0;
        $info = curl_multi_info_read($cmh, $returnValue);
 
        // если поток завершился
        if ( $info['msg'] == CURLMSG_DONE ) {
 
            $ch = $info['handle'];
 
            // ищем урл страницы по дескриптору потока в массиве заданий
            $uri = array_search($ch, $tasks);
 
            // забираем содержимое
            $this->_responses[$uri] = curl_multi_getcontent($ch);
 
            // удаляем поток из мультикурла
            curl_multi_remove_handle($cmh, $ch);
 
            // закрываем отдельное соединение (поток)
            curl_close($ch);
        }
         
        return $returnValue;
 
    }
 
     
    /**
     * Получение ответа с преобразованием данных в массив
     *
     * @access  public
     * @param   string $route
     * @return  mixed
     */
    public function getResponses(array $formats, array $params=array()) {
         
        $returnValue = NULL;
 
        foreach( $this->_uris as $key=>$uri ) {
 
            if ( !empty($this->_responses[$uri]) ) {
                 
                switch ( strtolower($formats[$key]) ) {
                     
                    case 'xml':
                        $returnValue[$key] = simplexml_load_string($this->_responses[$uri]);
                        break;
                     
                    case 'json':
                        $returnValue[$key] = Json::decode($this->_responses[$uri], TRUE);
                        break;                
                     
                    case 'csv':
                        $csv = array();
                         
                        $results = explode("\n", $this->_responses[$uri]);
                        foreach ( $results as $result ) {
                            $csv[] = str_getcsv($result, $params[$key]['delimiter'], $params[$key]['enclosure'], $params[$key]['escape']);
                        }
                        $returnValue[$key] = $csv;
                        break;
                         
                    default:
                        $returnValue[$key] = $this->_responses[$uri];
                        break;
                }
            }
        }
         
        return $returnValue;
    }
}
?>