<commit_before><?php
/**
 * This file is part of the Floodgate library.
 *
 * @author     <NAME> <<EMAIL>>
 * @copyright  2015
 *
 * For the full copyright and license information,
 * please view the LICENSE file that was distributed
 * with this source code.
 */

namespace Impensavel\Floodgate;

use Closure;

use GuzzleHttp\Client;
use GuzzleHttp\Message\ResponseInterface;
use GuzzleHttp\Stream\Utils;
use GuzzleHttp\Subscriber\Oauth\Oauth1;

class Floodgate
{
    /**
     * Twitter Streaming API URLs
     */
    const STREAM_URL      = 'https://stream.twitter.com/1.1/statuses/';
    const SITE_STREAM_URL = 'https://sitestream.twitter.com/1.1/';
    const USER_STREAM_URL = 'https://userstream.twitter.com/1.1/';

    /**
     * HTTP Client object
     *
     * @access  private
     * @var     \GuzzleHttp\Client
     */
    private $http;

    /**
     * Floodgate constructor
     *
     * @access  public
     * @param   \GuzzleHttp\Client                  $http
     * @param   \GuzzleHttp\Subscriber\Oauth\Oauth1 $oauth
     * @return  Floodgate
     */
    public function __construct(Client $http, Oauth1 $oauth)
    {
        $this->http = $http;

        $this->http->getEmitter()->attach($oauth);
    }

    /**
     * Create a Floodgate
     *
     * @static
     * @access  public
     * @param   array  $config Twitter OAuth configuration
     * @return  Floodgate
     */
    public static function create(array $config)
    {
        $config = array_merge([
            'consumer_key'    => '',
            'consumer_secret' => '',
            'token'           => '',
            'token_secret'    => '',
        ], $config);

        $http = new Client([
            'base_url' => static::STREAM_URL,
        ]);

        $oauth = new Oauth1($config);

        return new static($http, $oauth);
    }

    /**
     * Response processor
     *
     * @access  private
     * @param   Closure                               $callback
     * @param   \GuzzleHttp\Message\ResponseInterface $response
     * @throws  FloodgateException
     * @return  void
     */
    private function processor(Closure $callback, ResponseInterface $response)
    {
        $status = $response->getStatusCode();

        if ($status != 200) {
            throw new FloodgateException($response->getReasonPhrase(), $status);
        }

        $stream = $response->getBody();

        // read stream continuously and pass each line to the callback
        while (! $stream->eof()) {
            $line = Utils::readline($stream);

            $data = json_decode($line);

            $callback($data);
        }
    }

    /**
     * Open the Floodgate
     *
     * @access  public
     * @param   string $endpoint   Streaming API endpoint
     * @param   array  $parameters Streaming API parameters
     * @param   string $method     HTTP method
     * @return  \GuzzleHttp\Message\ResponseInterface
     */
    public function open($endpoint, array $parameters = [], $method = 'GET')
    {
        $options = [
            'exceptions' => false,
            'stream'     => true,
            'auth'       => 'oauth',
        ];

        // set option name according to the HTTP method
        $name = (strtolower($method) == 'post') ? 'body' : 'query';

        foreach ($parameters as $field => $value) {
            $options[$name][$field] = is_array($value) ? implode(',', $value) : $value;
        }

        $response = $this->http->createRequest($method, $endpoint, $options);

        return $this->http->send($response);
    }

    /**
     * Streaming API Sample endpoint
     *
     * @access  public
     * @param   Closure $callback
     * @param   array   $parameters Streaming API parameters
     * @throws  FloodgateException
     * @return  void
     */
    public function sample(Closure $callback, array $parameters = [])
    {
        $response = $this->open('filter.json', $parameters);

        $this->processor($callback, $response);
    }

    /**
     * Streaming API Filter endpoint
     *
     * @access  public
     * @param   Closure $callback
     * @param   array   $parameters Streaming API parameters
     * @throws  FloodgateException
     * @return  void
     */
    public function filter(Closure $callback, array $parameters = [])
    {
        $response = $this->open('filter.json', $parameters, 'POST');

        $this->processor($callback, $response);
    }

    /**
     * Streaming API Firehose endpoint
     *
     * @access  public
     * @param   Closure $callback   Callback to
     * @param   array   $parameters Streaming API parameters
     * @throws  FloodgateException
     * @return  void
     */
    public function firehose(Closure $callback, array $parameters = [])
    {
        $response = $this->open('firehose.json', $parameters);

        $this->processor($callback, $response);
    }
}
<commit_msg>chore(*): implemented main consumption loop, allowing for reconnections<commit_after><?php
/**
 * This file is part of the Floodgate library.
 *
 * @author     <NAME> <<EMAIL>>
 * @copyright  2015
 *
 * For the full copyright and license information,
 * please view the LICENSE file that was distributed
 * with this source code.
 */

namespace Impensavel\Floodgate;

use Closure;

use GuzzleHttp\Client;
use GuzzleHttp\Stream\StreamInterface;
use GuzzleHttp\Stream\Utils;
use GuzzleHttp\Subscriber\Oauth\Oauth1;

class Floodgate
{
    /**
     * Twitter Streaming API URL
     */
    const STREAM_URL = 'https://stream.twitter.com/1.1/statuses/';

    /**
     * HTTP sleep intervals
     *
     * @static
     * @access  private
     * @var     array
     */
    private static $intervals = [
        200 => 0,  // OK
        420 => 60, // too many reconnects
        504 => 5,  // server unavailable
    ];

    /**
     * Last connection timestamp
     *
     * @access  private
     * @var     int
     */
    private $lastConnection = 0;

    /**
     * Last data received timestamp
     *
     * @access  private
     * @var     int
     */
    private $lastReception = 0;

    /**
     * Last HTTP status
     *
     * @access  private
     * @var     int
     */
    private $lastStatus = 200;

    /**
     * Reconnection attempts
     *
     * @access  private
     * @var     int
     */
    private $attempts = 1;

    /**
     * Enable connection persistence
     *
     * @access  private
     * @var     bool
     */
    private $persist = true;

    /**
     * HTTP Client object
     *
     * @access  private
     * @var     \GuzzleHttp\Client
     */
    private $http;

    /**
     * Floodgate constructor
     *
     * @access  public
     * @param   \GuzzleHttp\Client                  $http
     * @param   \GuzzleHttp\Subscriber\Oauth\Oauth1 $oauth
     * @return  Floodgate
     */
    public function __construct(Client $http, Oauth1 $oauth)
    {
        $this->http = $http;

        $this->http->getEmitter()->attach($oauth);
    }

    /**
     * Create a Floodgate
     *
     * @static
     * @access  public
     * @param   array  $config Twitter OAuth configuration
     * @return  Floodgate
     */
    public static function create(array $config)
    {
        $config = array_merge([
            'consumer_key'    => '',
            'consumer_secret' => '',
            'token'           => '',
            'token_secret'    => '',
        ], $config);

        $http = new Client([
            'base_url' => static::STREAM_URL,
        ]);

        $oauth = new Oauth1($config);

        return new static($http, $oauth);
    }

    /**
     * Check if the connection is stalled
     *
     * @access  private
     * @return  bool
     */
    private function isStalled()
    {
        // Twitter considers a connection stalled
        // when 90 seconds (3 cycles) have passed
        // since the last data was received
        return (time() - $this->lastReception) > 90;
    }

    /**
     * Consumption loop
     *
     * @access  private
     * @param   string  $endpoint   Streaming API endpoint
     * @param   Closure $callback
     * @param   array   $parameters Streaming API parameters
     * @param   string  $method     HTTP method
     * @return  void
     */
    private function loop($endpoint, Closure $callback, array $parameters = [], $method = 'GET')
    {
        do {
            $response = $this->open($endpoint, $parameters, $method);

            if ($response) {
                $this->processor($callback, $response);
            }

        } while ($this->persist);
    }

    /**
     * Stream processor
     *
     * @access  private
     * @param   Closure                            $callback
     * @param   \GuzzleHttp\Stream\StreamInterface $stream
     * @throws  FloodgateException
     * @return  void
     */
    private function processor(Closure $callback, StreamInterface $stream)
    {
        // (re)set last received data timestamp
        $this->lastReception = time();

        while (! $stream->eof()) {
            $line = Utils::readline($stream);

            if ($this->isStalled()) {
                break;
            }

            // pass each line to the callback
            $callback(json_decode($line));

            // update last data received timestamp
            $this->lastReception = time();
        }
    }

    /**
     * Open the Floodgate
     *
     * @access  public
     * @param   string $endpoint   Streaming API endpoint
     * @param   array  $parameters Streaming API parameters
     * @param   string $method     HTTP method
     * @return  \GuzzleHttp\Stream\StreamInterface|bool
     */
    public function open($endpoint, array $parameters = [], $method = 'GET')
    {
        $options = [
            'exceptions' => false,
            'stream'     => true,
            'auth'       => 'oauth',
        ];

        // set option name according to the HTTP method in use
        $name = (strtolower($method) == 'post') ? 'body' : 'query';

        // set endpoint parameters
        foreach ($parameters as $field => $value) {
            $options[$name][$field] = is_array($value) ? implode(',', $value) : $value;
        }

        $request = $this->http->createRequest($method, $endpoint, $options);

        // back off N seconds, according to each specific HTTP status
        sleep(static::$intervals[$this->lastStatus] * $this->attempts++);

        $response = $this->http->send($request);

        // save latest HTTP status code
        $this->lastStatus = $response->getStatusCode();

        // (re)set last connection timestamp
        $this->lastConnection = time();

        // act according to HTTP status
        switch ($this->lastStatus) {
            case 200: // OK
                $this->attempts = 1;
                return $response->getBody();

            case 420: // too many reconnects
            case 503: // server unavailable
                if ($this->attempts > 10) {
                    $this->persist = false;
                }

                return false;

            // assume everything else is unrecoverable
            default:
                throw new FloodgateException($response->getReasonPhrase(), $this->lastStatus);
        }
    }

    /**
     * Streaming API Sample endpoint
     *
     * @access  public
     * @param   Closure $callback
     * @param   array   $parameters Streaming API parameters
     * @throws  FloodgateException
     * @return  void
     */
    public function sample(Closure $callback, array $parameters = [])
    {
        $this->loop('sample.json', $callback, $parameters);
    }

    /**
     * Streaming API Filter endpoint
     *
     * @access  public
     * @param   Closure $callback
     * @param   array   $parameters Streaming API parameters
     * @throws  FloodgateException
     * @return  void
     */
    public function filter(Closure $callback, array $parameters = [])
    {
        $this->loop('filter.json', $callback, $parameters, 'POST');
    }

    /**
     * Streaming API Firehose endpoint
     *
     * @access  public
     * @param   Closure $callback   Callback to
     * @param   array   $parameters Streaming API parameters
     * @throws  FloodgateException
     * @return  void
     */
    public function firehose(Closure $callback, array $parameters = [])
    {
        $this->loop('firehose.json', $callback, $parameters);
    }
}

{'fraction_non_alphanumeric': 0.12210150802659316, 'fraction_numerical': 0.00510783200908059, 'mean_word_length': 1.8612850846671305, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 81, 'https://': 4, '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': '5171077', 'n_tokens_mistral': 3620, 'n_tokens_neox': 3443, 'n_words': 891}