Construindo Com a API do Twitter: Usando Fluxos de Tempo Real

() translation by (you can also view the original English article)

Final product image — What You'll Be Creating

Enquanto a API REST do Twitter é ótima para várias aplicações, se quiser atualizações imediatas e acesso a um conjunto maior de notificações, a API de Fluxos do Twitter (Twitter Streaming API) é essencial. Por exemplo, somente a API de fluxos dirá que outro usuário favoritou um de seus tweets.

Para usar a API de Fluxos é preciso uma conexão ativa e persistente (keep-alive) entre seu servidor e o Twitter. Esse tipo de implementação pode ser bem estranha para vários programadores PHP. Assim que um tweet é realizado, o Twitter notifica seu servidor em tempo real, permitindo você guardá-lo em sua base de dados sem o atraso da fila da API REST. O uso da API de fluxos também não está sujeito ao limite de uso da API do Twitter.

Eis uma visualização de como o processo funciona:

Há três variações da API de Fluxos do Twitter:

O Fluxo Público. Essa daqui permite que sua aplicação monitore dados públicos no Twitter, como tweets públicos, filtros de tags, etc.;
O Fluxo do Usuário. Essa permite você rastrear o fluxo de tweets de um usuário, em tempo real. A parte três desta série focará no fluxo do usuário.
Fluxos de Site. Fluxos de site permitem sua aplicação monitorar os feeds de Twitter de vários usuários, em tempo real.

O trabalho da sua implementação de fluxo é registrar os eventos que chegam tão logo eles apareçam e processá-los em plano de fundo, usando a API REST quando necessário, para obter mais dados. Fluxos de Site requerem aprovação prévia do Twitter, que, geralmente, é reservada para grandes empresas e desenvolvedores.

Felizmente, há uma pequena biblioteca de código aberto, chamada Phirehose, que implementa a maioria dos requerimentos da API de fluxos. Esse tutorial descreverá como integrar a Phirehose em sua aplicação de código aberto, Birdcage.

A Biblioteca Phirehose

A Phirehose é uma implementação fantástica em PHP e de código aberto dos requerimentos da API de Fluxos do Twitter, escrita por Fenn Bailey. Como descrito por ele, a Phirehose serve para:

prover uma interface simples para a API de Fluxos do Twitter, para aplicações em PHP;
obedecer as recomendações de manipulação de erros, reconexão, etc., da API de Fluxos;
encorajar bons comportamentos de clientes da API de Fluxos
operar independente de extensões PHP (por exemplo: shared memory, PCNTL, etc.)

Pelo meu uso, ela opera praticamente sem erros. Veja a documentação da Phirehose aqui.

Ela foi criada para manter a conexão com o Twitter e responder ao feed de dados do Twitter enquanto opera de forma indefinida, sem interrupções. Não foi criada para processar dados minuciosos de tweets ou hidratação de dados, que foram descritos na parte dois desta série. Isso pode ser feito separadamente.

Executando a Phirehose Indefinidamente

Geralmente, você não pode executar uma tarefa cron da web como uma operação persistente por tempo indefinido. É melhor criar um daemon de linha de comando.

Uma das poderosas funcionalidades que o framework Yii oferece é a possibilidade de executar aplicações de linha de comando. Isso nos permitirá executar aplicações de linha de comando persistentes, criadas por nós, e que utilizam o framework PHP Birdcage e MySQL.

Construindo uma Comando para Linha de Comando no Yii

No diretório /app/, fora da pasta raiz acessível pela web, adicionaremos um arquivo stream.php que executará nosso comando de fluxo com o Phirehose, na linha de comando:

<?php
defined('YII_DEBUG') or define('YII_DEBUG',true);
$yii=dirname(__FILE__).'/../framework/yii.php';
$config=dirname(__FILE__).'/protected/config/main.php';
require_once($yii);
Yii::createConsoleApplication($config)->run();

O próximo passo é construirmo o arquivo de comando, StreamCommand.php, no diretório /app/protected/commands:

<?php
class StreamCommand extends CConsoleCommand
{
    // teste com o Fluxo ./app/stream.php
    public function run($args)
    {
        // obtem as chaves de usuário do Twitter
        $result = Account::model()->findByPk(1);
        $c = new Consumer($result['oauth_token'],$result['oauth_token_secret'],Phirehose::METHOD_USER);
        // carrega as chaves do app do Twitter
        $app = UserSetting::model()->loadPrimarySettings();
        $c->consumerKey = $app['twitter_key'];
        $c->consumerSecret = $app['twitter_secret'];
        $c->consume();
    }
}

Ele lançará o processo do Phirehose, o Consumer, usando nossas chaves de aplicação e usuário do Twitter.

Nota: Para o exemplo do fluxo do Birdcage, assumiremos que apenas uma conta de Twitter foi registrado e está fixa no código (account_id = 1).

Integrando o Phirehose

Para integrar o Phirehose no Birdcage, movi os arquivos OAuthPhirehose.php e UserstreamPhirehose.php para o diretório /app/protected/components. No meu arquivo de configuração, main.php, adicionei o phirehose à lista de componentes carregados:

'preload'=>array(
    'log',
    'bootstrap',
    'mailgun',
    'phirehose',
      'advanced'
    ),

Então, criei uma migração para a base de dados, criando uma tabela para gravar os dados originais do fluxo do Twitter:

class m140919_193106_create_stream_table extends CDbMigration
{
 protected $MySqlOptions = 'ENGINE=InnoDB CHARSET=utf8 COLLATE=utf8_unicode_ci';
     public $tablePrefix;
     public $tableName;

     public function before() {
       $this->tablePrefix = Yii::app()->getDb()->tablePrefix;
       if ($this->tablePrefix <> '')
         $this->tableName = $this->tablePrefix.'stream';
     }

       public function safeUp()
    {
      $this->before();
      $this->createTable($this->tableName, array(
               'id' => 'pk',
               'tweet_id' => 'bigint(20) unsigned NOT NULL',
               'code' => 'text  NULL',
               'is_processed' => 'tinyint default 0',
               'created_at' => 'DATETIME NOT NULL DEFAULT 0',
               'modified_at' => 'TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP',
                 ), $this->MySqlOptions);
                 $this->createIndex('tweet_id', $this->tableName , 'tweet_id', true);

    }

Também criei um modelo chamado Consumer.php que estende OauthPhirehose com o método enqueueStatus obrigatório.

Queremos minimizar a quantidade de processamento que a resposta em tempo real precisa realizar. Essencialmente, apenas queremos guardar os dados recebidos do Twitter em nossa base de dados e nada mais. Também podemos realizar outros processamentos em nossas próprias tarefas de plano de fundo sem atrapalhar a conexão do Phirehose. Minha função apenas pega os dados do tweet e o guarda na tabela stream:

<?php
  class Consumer extends OauthPhirehose
  {
// Essa função é chamada automaticamente pela classe Phirehose
    // Quando um novo tweet é recebido com os dados JSON em $status
    public function enqueueStatus($status) {
      $stream_item = json_decode($status);
      if (!(isset($stream_item->id_str))) { return;}
      $s = new Stream;
      $s->tweet_id = $stream_item->id_str;
      $s->code = base64_encode(serialize($stream_item));
      $s->is_processed=0;
      $s->created_at = new CDbExpression('NOW()');
      $s->modified_at =new CDbExpression('NOW()');
      $s->save();
      var_dump($stream_item);
    }
  }
  ?>

Nós dependeremos das tarefas de plano de fundo executadas pelo nosso DaemonController para processar os dados no modelo de Tweet do Bircdcage. Isso será descrito logo abaixo.

Ativando o Phirehose

Você pode testar o Pirehose usando o terminal de comandos do PHP:

php ./app/stream.php Stream

O Twitter enviará o fluxo informação de seguidores da conta do usuário, seguida pelos dados em tempo real, assim que chegar.

Para ativar o Phirehose como um comando de terminal sempre ativo e intermitente, usaremos o comando nohup, por exemplo sem espera e redirecionamento do retorno para dev/null:

nohup php ./app/stream.php Stream > /dev/null 2>&1&

O Ubuntu retornará o id do seu processo para futuros monitoramento e finalização:

[1] 9768

Se quiser verificar se o processo está em execução, escaneie a lista de tarefas pelo id do processo:

ps -e all | grep 9768

Você deve ver algo mais ou menos assim:

0 1000 9768 9743 20 0 273112 16916 poll_s S pts/0 0:00 php ./app/stream.php Stream

E você também pode finalizar o Phirehose eliminando-o através do id do processo:

kill 9768

Pela minha experiência, o Phirehose funcionou perfeitamente com essa técnica, operando sem interrupções durante as duas últimas semanas.

Processando os Dados do Fluxo

Também precisamos criar um processo de plano de fundo no Birdcage que processará os dados do fluxo das tabelas Tweet, Mention, URL e Hashtag—como se tivessem vindo da API REST.

Altere seu arquivo de configuração twitter.ini para que ele use os fluxos:

1	twitter_stream = true

E podemos usar a mesma tarefa cron da parte dois, para executar essa operação:

# Para definir o temp, você proverá valores reais para
# minutos (m), horas (h), dia do mês (dom), mês (mon),
# e dia da semana (dow) ou usar '*' nesses campos (para 'qualquer um').#
# Perceba que as tarefas serão inicializadas de acordo a noção de tempo e
# fuso horários do sistema de daemon.
#
# Por exemplo, você pode executar um backup de todas as contas de usuário
# às 5 da manhã a cada semana:
# 0 5 * * 1 tar -zcf /var/backups/home.tgz /home/
#
# m h  dom mon dow   command
*/5 * * * * wget -O /dev/null http://birdcage.yourdomain.com/daemon/index

Assim, quando o DaemonController for chamado, ele ativará o método de processo model() de Stream:

public function actionIndex() {
    // Se não usar fluxos do twitter, processaremos pela API REST
    if (!Yii::app()->params['twitter_stream']) {
      Tweet::model()->getStreams();
    } else {
      Stream::model()->process();
    }
  }

O método do processo descompacta os dados de fluxo codificados e analisa cada entrada, da mesma forma que fizemos com a API REST:

public function process() {
    // obtem tweets não processados da máquina de fluxos
    $account_id = 1;
    $items = Stream::model()->unprocessed()->findAll();
    foreach ($items as $i) {
      $tweet = unserialize(base64_decode($i['code']));
      Tweet::model()->parse($account_id,$tweet);
      $this->setStatus($i['id'],self::STREAM_PROCESSED);
    }
  }

Atualmente, o Birdcage ignora os dados do fluxo que não seja um tweet, por exemplo, notificações, mensagens diretas, etc. Deixarei essa parte como tarefa para você—ou você pode verificar minha aplicação expandida, Birdhouse.

Finalizando

Espero que tenha achado essa série em três partes sobre a API do Twitter bem informativa e útil. Você deve ter aprendido sobre OAuth, a API REST, a API de Fluxos, a construir uma base de dados para o Twitter, processar a linha do tempo com ambos os tipos de API, contar corretamente a quantidade de caracteres em um tweet e publicá-lo, e mais.

Deixe quaisquer comentários, correções ou ideias a mais. Você pode navegar todos os meus outros tutoriais no Tuts+ em minha página daqui ou seguir-me no Twitter, em @reifman.