xnTwitterStream.class.php

<?php
/**
 * xnTwitterStream.class.php
 *  Twitter Streaming APIでのデータの取得用クラス
 **/
class xnTwitterStrream implements Iterator
{
  const READ_NULL_REPEAT_MAX = 30;
  /**
   *  @var  $dispather sfEventDispatcher
   **/
  private $dispatcher = null;
  /**
   *  @var  $eof_flag bool
   **/
  private $eof_flag = true;
  /**
   *  @var  $read_handle resource  読み取りファイルハンドル
   **/
  private $read_handle = null;
  /**
   *  @var  $response_header  string
   **/
  private $response_header = '';
  /**
   *  @var  $buffer string    読み取ったJSONオブジェクト用のバッファ
   **/
  private $buffer = '';

  /**
   * コンストラクタ
   *
   * @param sfEventDispatcher $dispatcher
   */
  public function  __construct(sfEventDispatcher $dispatcher = null)
  {
    $this->dispatcher = $dispatcher;
  }

  /**
   * connect
   * 指定したURLのAPIに接続してストリーム接続を開始する。
   *
   * @param   string    $api_url          接続するAPIのURL
   * @param   string    $account          接続に使用するアカウント
   * @param   string    $password         接続に使用するパスワード
   * @param   string    $method
   * @param   array     $params
   * @param   int       $timeout
   * @param   int       $retry
   * @param   int       $retry_interval
   *
   * @return  bool      接続に失敗した場合はfalse。リクエストに成功した場合はtrue。
   **/
  public function connect(
      $api_url,
      $account,
      $password,
      $method = 'get',
      $params = array(),
      $timeout = 10,
      $retry = 3,
      $retry_interval = 10)
  {
    $url_info = parse_url($api_url);
    $target_server = $url_info['host'];
    $target_path = $url_info['path'];

    $error_no;
    $error_str;

    $fp = null;
    while ($retry--)
    {
      if($fp = fsockopen($target_server, 80, $error_no, $error_str, $timeout))
      {
        //  接続成功
        break;
      }
      else
      {
        //  接続できなかった
        if ($retry)
        {
          //  再試行時間待機する。
          sleep($retry_interval);
        }
        else
        {
          return false;
        }
      }
    }

    //  パラメータを準備する
    $parameter_str = http_build_query($params);
    /*
    foreach ($params as $key=>$value)
    {
      $parameter_str .= $key . '=' . $value . '&';
    }
     */

    $requests = array();
    switch ($method = strtolower($method))
    {
      case 'get':
        $requests[] = sprintf('GET %s HTTP/1.1', $target_path);
        $requests[] = sprintf('Host: %s', $target_server);
        $requests[] = sprintf('Authorization: Basic %s', base64_encode(sprintf('%s:%s', $account, $password)));
        $requests[] = sprintf('User-Agent: PHP %s', __CLASS__);
        $requests[] = sprintf('Content-Type: text/html');
        $requests[] = "\r\n";
        break;
      case 'post':
      default:
        $requests[] = sprintf('POST %s HTTP/1.1', $target_path);
        $requests[] = sprintf('Host: %s', $target_server);
        $requests[] = sprintf('Authorization: Basic %s', base64_encode(sprintf('%s:%s', $account, $password)));
        $requests[] = sprintf('User-Agent: PHP %s', __CLASS__);
        $requests[] = sprintf('Content-Type: application/x-www-form-urlencoded');
        $requests[] = sprintf('Content-Length: %s', strlen($parameter_str));
        $requests[] = '';
        $requests[] = $parameter_str;
        break;
    }
    fputs($fp, $request_data = implode("\r\n", $requests));
    if ($this->dispatcher)
    {
      $this->dispatcher->notify(new sfEvent($this, 'command.log', array($request_data)));
    }

    //  初回レスポンスヘッダを取得する。
    $response_header = '';
    while (($buffer = fgets($fp)) !== "\r\n")
    {
      $response_header .= $buffer;
    }
    $this->response_header = $response_header;
    if ($this->dispatcher)
    {
      $this->dispatcher->notify(new sfEvent($this, 'command.log', array($response_header)));
    }

    $this->read_handle = $fp;
    $this->eof_flag = false;
    return true;
  }

  /**
   * key
   *
   * @return int
   */
  public function key()
  {
    return 0;
  }

  /**
   * current
   *
   * @return  JsonObject
   **/
  public function current()
  {
    return $this->buffer;
  }

  /**
   * next
   *
   * @throws  Exception   空行の連続が制限回数を超えると例外
   **/
  public function next()
  {
    return $this->next_rewind_body();
  }
  public function rewind()
  {
    return $this->next_rewind_body();
  }

  protected function next_rewind_body()
  {
    if (!feof($this->read_handle))
    {
      $this->eof_flag = false;
      $retry = self::READ_NULL_REPEAT_MAX;
      while (strpos($buffer = trim($raw = fgets($this->read_handle)),'{') === false)
      {
        if (--$retry < 0)
        {
          fclose($this->read_handle);
          throw new Exception('stream repeat null error');
        }
      }
      $this->buffer = json_decode($buffer);
    }
    else
    {
      $this->eof_flag = true;
      fclose($this->read_handle);
    }
  }

  public function valid()
  {
    return !$this->eof_flag;
  }
}