Recentemente ho incontrato il compito di attuare il Twitter Api di flusso di processo di Feed Twitter (Tweet) in tempo reale. Più specificamente, l’applicazione da costruire dovuto essere in grado di ascoltare i tweet che vanno dal più argomenti e utenti (faremo riferimento a loro come tracce), e dovrebbe essere in grado di hot-swap queste tracce quasi istantaneamente.

Non molto tempo prima di lavorare su questa applicazione, ho dovuto eseguire la stessa operazione in un contesto di Facebook. Contrariamente a Twitter, che ne sfrutta le Api di flusso per fornire applicazioni con aggiornamenti in tempo reale, Facebook utilizza invece di Webhooks. Webhooks, essendo un HTTP richiamata, di fornire agli sviluppatori un livello di sofisticazione che è molto più facile da affrontare. Streaming Api fornite da Twitter, tuttavia, richiede l’applicazione di mantenere una connessione HTTP permanente con uno di Twitter molti streaming endpoint. La scrittura di un’applicazione che si basa su una connessione permanente per il funzionamento richiede di pensare in un modo leggermente differente rispetto a quello che stiamo presumibilmente utilizzato per la costruzione di una applicazione web tradizionale.

L’obiettivo di questo articolo potrà essere lo sviluppo di tale applicazione.

Twitter Streaming Api

Se si da un’occhiata a Twitter lo Sviluppatore di Documentazione, vedrai che ci forniscono Streaming Api. Naturalmente, essi ci forniscono il RESTO e Annunci di Api, ma per i nostri scopi, possiamo solo concentrandosi sui loro Streaming Api.

Così che cosa esattamente sono le Api di flusso? Per dirla in termini semplici, utilizzando le Api di flusso, l’applicazione è in grado di stabilire una bassa latenza di connessione con Twitter, streaming di endpoint. Una volta che la connessione è stata stabilita con successo, Twitter inoltrare in tempo reale nutrire la vostra applicazione ha richiesto per questa connessione pipeline quasi istantaneamente. Secondo la documentazione:

Streaming Api consentono agli sviluppatori di bassa latenza di accesso a Twitter globale del flusso di Tweet di dati. Una corretta implementazione di un client streaming sarà spinto i messaggi che indicano Tweet e altri eventi si sono verificati, senza l’overhead associato con il polling di un endpoint REST.

Streaming Api richiede il collegamento a essere persistente (essenzialmente non finisce mai) in natura. In sostanza, è simile al download di un infinitamente di file di grandi dimensioni — sistema continuerà a ricevere i dati a meno che non si terminare manualmente la connessione.

Il tipo di feed l’applicazione riceve dipenderà dal tipo di flussi che si intende utilizzare. Streaming Api consente agli sviluppatori di tre diversi tipi di flussi:

• Flussi di Pubblico – i Flussi di dati pubblici che scorre attraverso Twitter. Adatto per i seguenti utenti specifici o argomenti, e di data mining.
• Utente Flussi – per Singolo utente flussi, contenente circa tutti i dati con un singolo utente visualizza Twitter.
• Sito Torrenti – Il multi-utente versione di utente flussi. Sito flussi sono destinati per i server che devono connettersi a Twitter a nome di molti utenti.

Come ho detto in precedenza, il feed riceviamo dipenderà dal tipo di flussi di noi. Quindi, prima di andare avanti, diamo uno sguardo ai requisiti di applicazione, a seconda che ci si stia utilizzando Pubblici e / o corsi di Utente Flussi (Sito, i Flussi sono in closed beta).

La nostra applicazione deve soddisfare i seguenti requisiti:

• Essere in grado di elaborare ricevuto tweet senza interrompere la connessione con lo streaming endpoint
• Mantenere questi requisiti in mente, ora abbiamo a stabilirsi sui flussi di noi. Utente flussi, come affermato in precedenza, fornisce i dati da un singolo punto di vista dell’utente. Secondo Twitter documentazione per l’Utente Flussi:

  Utente Flussi di fornire un flusso di dati e di eventi specifici per l’utente autenticato. Nota che l’Utente Flussi non sono destinati a server-to-server connessioni. Se avete bisogno di fare dei collegamenti per conto di più utenti della stessa macchina, considerare l’utilizzo di flussi di sito.

  Ridurre al minimo il numero di connessioni in un’applicazione per l’Utente di Flussi. Ogni account di Twitter è limitato solo a qualche Utente simultanee Flussi di connessioni per OAuth applicazione, indipendentemente dall’IP. Una volta che l’applicazione per applicazione limite viene superato, il più antico collegamento viene interrotto. Un account di registrazione da molte istanze della stessa OAuth applicazione del ciclo di connessioni come l’applicazione delle istanze di ricollegare e scollegare tutti gli altri.

  Dal momento che siamo più concentrati su essere in grado di ricevere i tweet da numerosi utenti, non è chiaro come molti utenti che saranno. Andando dai documenti, l’Utente Flussi non è tagliato per la nostra applicazione, in caso d’uso. Così adesso siamo di sinistra a fare con Flussi di Pubblico.

  anziché Utente Flussi, che ci consente di ricevere i feed in contesto su un numero limitato di singoli utenti, Pubblici Flussi di fornire applicazioni con un flusso di dati pubblici che scorre attraverso Twitter. Parlando in contesto di tweet in particolare, se si utilizza mezzi di flussi elegantemente, la nostra applicazione non essere vincolato all’ambito di applicazione limitato a singoli utenti, in contrasto con l’Utente Flussi. Dal momento che non sono necessariamente interessati privato flusso di dati, siamo in grado di passare lungo Maniglie Twitter (@nomeutente) corrispondenti ai singoli Utenti di Twitter lungo le piste prima di stabilire una connessione con il pubblico in streaming endpoint. Questo consentirà la nostra applicazione per ascoltare dal vivo pubblici tweet da parte di più utenti. Un breve punto di preavviso prima di andare avanti, come per Twitter Documentazione per la Pubblica Torrenti, la nostra applicazione può stabilire solo una singola connessione con un pubblico di flussi in qualsiasi momento.

   

  Comunicare con l’API di flusso: Costruire la nostra applicazione

  Ora che abbiamo finito con tutti i introduttivo bit, possiamo iniziare a costruire la nostra applicazione. Sarò sviluppare l’applicazione in PHP che esegue Linux, utilizzando il popolare Framework Symfony e Compositore come gestore di pacchetti. Detto questo, si dovrebbe essere in grado di applicare i concetti fondamentali e realizzare il progetto in qualsiasi lingua di vostra scelta.

  io, comunque, l’utilizzo di Phirehose da Fennb, una libreria di terze parti per facilitare il processo di comunicazione con il Twitter Api di flusso. Se sei in via di sviluppo su piattaforma diversa da PHP, ecco una lunga lista di le librerie di terze parti si può fare uso di. È possibile aggiungere Phirehose per il vostro progetto tramite Compositore utilizzando il seguente comando:

  php composer require “fennb/phirehose”

   

  Impostazione della nostra applicazione:

  Creare un nuovo progetto Symfony (chiamiamolo twitterFeeds), eseguendo il seguente comando:

  symfony nuovo twitterFeeds

  Prima di andare avanti con l’articolo, assicurarsi di aver installato la libreria di terze parti, come sopra menzionato. Successivamente, creare un nuovo bundle (chiamiamola TwitterBundle), eseguendo il seguente comando:

  php bin/console generate:bundle

  Seguire con qualunque configurazione dei dettagli è richiesto da voi. A quel punto, andare a TwitterBundle directory del progetto e creare un nuovo Servizi directory. In questa directory, creare un nuovo file chiamato TwitterService.php che si utilizza. Assicurarsi di registrare questo servizio per la vostra applicazione.

   

  Stabilire una connessione permanente:

  Come ho detto prima, la scrittura di un’applicazione che si basa su una connessione permanente per il funzionamento richiede di pensare in un modo leggermente differente. Diversamente, nel caso di utilizzo di Api REST, dove possiamo effettuare chiamate API, o Webhooks, dove un callback è realizzato con i nostri server, con lo Streaming Api, stiamo stabilendo una connessione persistente con un flusso di endpoint che poi vive per sempre (idealmente parlando), di processo e di qualsiasi tipo di dati che riceviamo sul nostro terminale attraverso questo mezzo. Chiudiamo questa connessione solo nei casi in cui abbiamo bisogno di riavviare per aggiornare le tracce. Quindi, come facciamo a stabilire tale collegamento? Un modo è quello di fare uso di Processi in Background.

  Permette di dire che dobbiamo ascoltare live tweet, abbiamo quindi bisogno di stabilire una connessione permanente con pubblica di Twitter in streaming endpoint. Per farlo, siamo in grado di eseguire un processo in background tramite il terminale. Questo processo in background verrà eseguito come un processo individuale e di stabilire l’ha detto una connessione persistente nostra applicazione richiede. Possiamo, quindi, di monitorare il ID di processo (PID) di questo processo in background se abbiamo bisogno di modificare in qualsiasi modo.

  Per rendere il nostro processo in background, dobbiamo creare un comando da console che poi siamo in grado di eseguire da terminale nella directory principale del nostro progetto. Per rendere il nostro comando, è possibile utilizzare il Componente Console fornito dal Framework Symfony. Prima di creare un Comando directory TwitterBundle, e quindi creare TwitterCommand.php file in quella directory.

  /** * Webkul Software. * * @category Webkul, Uvdesk * @package Webkul_UVDesk_TF * @author Akshay Kumar * @copyright Copyright (c) 2010-2016 Webkul Software Private Limited (https://webkul.com) * @license https://store.webkul.com/license.html */ namespace TwitterBundle\Command; use Symfony\Component\Console\Command\Command; use Symfony\Component\Console\Input\InputInterface; use Symfony\Component\Console\Output\OutputInterface; use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand; use Symfony\Component\Routing\Generator\UrlGeneratorInterface; class TwitterCommand extends ContainerAwareCommand { protected function configure() { $this->setName('twitter:stream')->setDescription('Listen for Live Twitter Feeds.'); } protected function execute(InputInterface $input, OutputInterface $output) { $twitterTracks = ['#ReasonsToLoveMe']; // Define your tracks you wish to receive updates for $this->getContainer()->get('twitter.service')->listenForTweets($twitterTracks); // This will start the streaming process } }

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31

  /** * Webkul Software. * * @category Webkul, Uvdesk * @package Webkul_UVDesk_TF * @author Akshay Kumar * @copyright Copyright (c) 2010-2016 Webkul Software Private Limited (https://webkul.com) * @license https://store.webkul.com/license.html */   namespace TwitterBundle\Command;   use Symfony\Component\Console\Command\Command; use Symfony\Component\Console\Input\InputInterface; use Symfony\Component\Console\Output\OutputInterface; use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand; use Symfony\Component\Routing\Generator\UrlGeneratorInterface;   class TwitterCommand extends ContainerAwareCommand { protected function configure() { $this->setName('twitter:stream')->setDescription('Listen for Live Twitter Feeds.'); }   protected function execute(InputInterface $input, OutputInterface $output) { $twitterTracks = ['#ReasonsToLoveMe']; // Define your tracks you wish to receive updates for $this->getContainer()->get('twitter.service')->listenForTweets($twitterTracks); // This will start the streaming process } }

  Dalla directory principale del nostro progetto, ora siamo in grado di eseguire il codice scritto all’interno di un execute() funzione utilizzando il seguente comando nel terminale:

  php bin/console twitter:stream

  Come si può vedere, il TwitterCommand::execute() funzione a sua volta chiamata la TwitterCommand::listenForTweets() funzione definita in TwitterService, la costituzione di una nuova connessione persistente con Twitter, streaming di endpoint. Quindi cerchiamo di codice di questo accanto. Nel TwitterService.php che abbiamo creato in precedenza,

  /** * Webkul Software. * * @category Webkul, Uvdesk * @package Webkul_UVDesk_TF * @author Akshay Kumar * @copyright Copyright (c) 2010-2016 Webkul Software Private Limited (https://webkul.com) * @license https://store.webkul.com/license.html */ namespace TwitterBundle\Services; const CONSUMER_KEY = YOUR_APP_CONSUMER_KEY; const CONSUMER_SECRET = YOUR_APP_CONSUMER_SECRET; const APPLICATION_TOKEN = YOUR_APP_ACCESS_TOKEN; const APPLICATION_TOKEN_SECRET = YOUR_APP_ACCESS_TOKEN_SECRET; // Include the third-party library require_once __DIR__ . '/../../../vendor/fennb/phirehose/lib/OauthPhirehose.php'; class TwitterListener extends \OauthPhirehose { public function __construct() { parent::__construct(APPLICATION_TOKEN, APPLICATION_TOKEN_SECRET, \Phirehose::METHOD_FILTER); $this->consumerKey = CONSUMER_KEY; $this->consumerSecret = CONSUMER_SECRET; } public function enqueueStatus($status) { $twitterFeed = json_decode($status, true); if (is_array($twitterFeed) && isset($twitterFeed['user']['screen_name'])) { $twitterData = array('feed' => $twitterFeed); dump('New Feed Recieved'); dump($twitterData); } } } class TwitterService { /** * This function will establish a persistent connection with one of the twitter's streaming endpoint. * @param array $twitterTrackCollection Collection of tracks which we'll receive updates for */ public function listenForTweets(array $twitterTrackCollection = []) { // Terminate the process if there are no active tracks if (empty($twitterTrackCollection)) { echo 'No active tracks. Terminating Process...' . PHP_EOL . PHP_EOL; exit(0); } $twitterListener = new TwitterListener(); $twitterListener->setTrack($twitterTrackCollection); $twitterListener->consume(); } }

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58

  /** * Webkul Software. * * @category Webkul, Uvdesk * @package Webkul_UVDesk_TF * @author Akshay Kumar * @copyright Copyright (c) 2010-2016 Webkul Software Private Limited (https://webkul.com) * @license https://store.webkul.com/license.html */   namespace TwitterBundle\Services;   const CONSUMER_KEY = YOUR_APP_CONSUMER_KEY; const CONSUMER_SECRET = YOUR_APP_CONSUMER_SECRET; const APPLICATION_TOKEN = YOUR_APP_ACCESS_TOKEN; const APPLICATION_TOKEN_SECRET = YOUR_APP_ACCESS_TOKEN_SECRET;   // Include the third-party library require_once __DIR__ . '/../../../vendor/fennb/phirehose/lib/OauthPhirehose.php';   class TwitterListener extends \OauthPhirehose {     public function __construct() {         parent::__construct(APPLICATION_TOKEN, APPLICATION_TOKEN_SECRET, \Phirehose::METHOD_FILTER);         $this->consumerKey = CONSUMER_KEY;         $this->consumerSecret = CONSUMER_SECRET;     }       public function enqueueStatus($status)     {           $twitterFeed = json_decode($status, true);         if (is_array($twitterFeed) && isset($twitterFeed['user']['screen_name'])) {             $twitterData = array('feed' => $twitterFeed);             dump('New Feed Recieved');             dump($twitterData);         }     } }   class TwitterService { /** * This function will establish a persistent connection with one of the twitter's streaming endpoint. * @param  array  $twitterTrackCollection Collection of tracks which we'll receive updates for */ public function listenForTweets(array $twitterTrackCollection = [])     {      // Terminate the process if there are no active tracks      if (empty($twitterTrackCollection)) {      echo 'No active tracks. Terminating Process...' . PHP_EOL . PHP_EOL;      exit(0);      }        $twitterListener = new TwitterListener();         $twitterListener->setTrack($twitterTrackCollection);         $twitterListener->consume();     } }

  Ora, se si esegue il comando che abbiamo appena creato il vostro terminale, una nuova connessione con Twitter API sarà istituito e inizierai a ricevere pubblico tweet per le tracce che avete definito. Al momento della scrittura di questo articolo, “#ReasonsToLoveMe” è stato trending modo che ho usato come esempio di traccia. Ora ogni volta che un tweet che consiste in uno qualsiasi dei nostri specificato tracce (in questo caso, “#ReasonsToLoveMe”), che i tweet saranno passati lungo il collegamento di condutture. Questo tweet saranno poi passati alla TwitterListener::enqueueStatus() funzione, dove è possibile elaborare ricevuto feed. Tuttavia, si dovrebbe evitare di fare ogni volta l’operazione intensivo a questo punto. Ragione di essere che se l’applicazione è in grado di elaborare questo flusso di dati è abbastanza veloce, in contrasto con il tasso che si sta ricevendo la alimenta, si corre il rischio che il collegamento è terminato, che per la maggior parte di noi sarebbe auspicabile.

  

  in Tempo Reale Feed di Twitter

  Fino ad ora, siamo stati avviare manualmente la connessione andando sul nostro terminale ed eseguire il comando. Ma dal momento che applicazioni del mondo reale sono molto diversi automatizzare questo processo di avvio e chiusura della connessione. Nel nostro DefaultController, scriviamo un’azione gestore (che chiameremo startStreamAction), che quando viene eseguito volontà di avviare il processo di streaming. Assicurati di aver configurato il percorso del gestore del corrispondente routing.yml file. In DefaultController.php,

  /** * Webkul Software. * * @category Webkul, Uvdesk * @package Webkul_UVDesk_TF * @author Akshay Kumar * @copyright Copyright (c) 2010-2016 Webkul Software Private Limited (https://webkul.com) * @license https://store.webkul.com/license.html */ namespace TwitterBundle\Controller; use Symfony\Bundle\FrameworkBundle\Controller\Controller; class DefaultController extends Controller { // ..... /** * This handler will switch to the root directory and start a new background process */ public function startStreamAction(Request $request) { chdir('../'); $command = 'php bin/console twitter:listen'; exec($command); exit(0); } // ..... }

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31

  /** * Webkul Software. * * @category Webkul, Uvdesk * @package Webkul_UVDesk_TF * @author Akshay Kumar * @copyright Copyright (c) 2010-2016 Webkul Software Private Limited (https://webkul.com) * @license https://store.webkul.com/license.html */   namespace TwitterBundle\Controller;   use Symfony\Bundle\FrameworkBundle\Controller\Controller;   class DefaultController extends Controller {     // .....       /**      * This handler will switch to the root directory and start a new background process      */     public function startStreamAction(Request $request)     {         chdir('../');         $command = 'php bin/console twitter:listen';         exec($command);         exit(0);     }       // ..... }

  Ora, ogni volta che si visita l’itinerario di questo gestore, questo comando verrà eseguito separatamente in un nuovo thread.

  Con Flussi di Pubblico, abbiamo solo ricevere feed pubblici per le tracce che abbiamo definito prima di stabilire una connessione persistente con lo streaming endpoint. Quindi, se ne presenta la necessità di aggiornare le nostre tracce, abbiamo bisogno di terminare la connessione precedente, ri-definire le nostre tracce, e poi ri-stabilire una nuova connessione con lo streaming endpoint. Questo processo è molto critica in quanto si può avere solo una connessione con lo streaming endpoint (rischiamo di avere il nostro IP bannato con connessioni multiple). Quindi abbiamo bisogno di assicurarsi che prima di tentare di creare una nuova connessione, il tutto tenuto in precedenza le connessioni vengono chiuse.

  Per chiudere una connessione con lo streaming endpoint, siamo in grado di terminare il processo in background corrispondente alla connessione. Per uccidere il processo, siamo in grado di fare uso di esso ID di processo (PID), o cercare utilizzando il nome e gli argomenti di il processo in background. Utilizzando l’ID di processo, ogni volta che si stabilisce una nuova connessione, abbiamo bisogno di log (database, file di testo, ecc…) l’ID di processo corrispondente al processo in background. Utilizzando questo registro, si può quindi terminare precedentemente detenute processi in background prima di iniziare uno nuovo. Aggiorniamo il nostro startStreamAction gestore per registrare l’ID di processo ogni volta che viene effettuata una connessione, e nel nostro DefaultController, creare una nuova azione gestore (consente di chiamare questo stopStreamAction) per terminare precedentemente detenute processi in background. Possiamo quindi chiamare il gestore prima di iniziare un nuovo processo in background per garantire che una sola istanza di una connessione persistente con lo streaming endpoint è attivo in un momento.

  /** * Webkul Software. * * @category Webkul, Uvdesk * @package Webkul_UVDesk_TF * @author Akshay Kumar * @copyright Copyright (c) 2010-2016 Webkul Software Private Limited (https://webkul.com) * @license https://store.webkul.com/license.html */ namespace TwitterBundle\Controller; use Symfony\Component\HttpFoundation\Request; use Symfony\Component\HttpFoundation\Response; use Symfony\Bundle\FrameworkBundle\Controller\Controller; use Symfony\Component\Routing\Generator\UrlGeneratorInterface; class DefaultController extends Controller { // ...... /** * This handler will first terminate any previously held background process * by firing the stopStreamAction() handler. Then it'll create a new background * process which establish a new connection with a streaming endpoint. */ public function startStreamAction(Request $request) { // Execute the stopStreamAction() handler $curlHandler = curl_init(); $stopStreamPath = $this->generateUrl('twitter_stream_stop', array(), UrlGeneratorInterface::ABSOLUTE_URL); curl_setopt($curlHandler, CURLOPT_URL, $stopStreamPath); curl_setopt($curlHandler, CURLOPT_RETURNTRANSFER, 1); curl_exec($curlHandler); // Create a new background process chdir('../'); $output = array(); $command = 'nohup php bin/console twitter:listen > /dev/null 2>&1 & echo $!'; exec($command, $output); $processId = (int) $output[0]; // Process ID of the newly created background process. Log this. exit(0); } /** * This handler will fetch a log of Process IDs corresponding to any background * process and terminate them. */ public function stopStreamAction(Request $request) { $activeProcesses = ARRAY_COLLECTION_OF_PROCESS_IDS_TO_TERMINATE; if (!empty($activeProcesses)) { foreach ($activeProcesses as $processId) { $command = 'kill ' . $processId . ' 2>&1'; exec($command, $output); $this->get('logger')->info('Twitter Stream Process Terminated: ' . $processId); } } exit(0); } // ...... }

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64

  /** * Webkul Software. * * @category Webkul, Uvdesk * @package Webkul_UVDesk_TF * @author Akshay Kumar * @copyright Copyright (c) 2010-2016 Webkul Software Private Limited (https://webkul.com) * @license https://store.webkul.com/license.html */   namespace TwitterBundle\Controller;   use Symfony\Component\HttpFoundation\Request; use Symfony\Component\HttpFoundation\Response; use Symfony\Bundle\FrameworkBundle\Controller\Controller; use Symfony\Component\Routing\Generator\UrlGeneratorInterface;   class DefaultController extends Controller {     // ......       /**      * This handler will first terminate any previously held background process        * by firing the stopStreamAction() handler. Then it'll create a new background        * process which establish a new connection with a streaming endpoint.      */     public function startStreamAction(Request $request)     {      // Execute the stopStreamAction() handler         $curlHandler = curl_init();         $stopStreamPath = $this->generateUrl('twitter_stream_stop', array(), UrlGeneratorInterface::ABSOLUTE_URL);         curl_setopt($curlHandler, CURLOPT_URL, $stopStreamPath);         curl_setopt($curlHandler, CURLOPT_RETURNTRANSFER, 1);         curl_exec($curlHandler);           // Create a new background process         chdir('../');         $output = array();         $command = 'nohup php bin/console twitter:listen > /dev/null 2>&1 & echo $!';         exec($command, $output);         $processId = (int) $output[0]; // Process ID of the newly created background process. Log this.         exit(0);     }       /**      * This handler will fetch a log of Process IDs corresponding to any background      * process and terminate them.      */     public function stopStreamAction(Request $request)     {         $activeProcesses = ARRAY_COLLECTION_OF_PROCESS_IDS_TO_TERMINATE;           if (!empty($activeProcesses)) {             foreach ($activeProcesses as $processId) {                 $command = 'kill ' . $processId . ' 2>&1';                 exec($command, $output);                 $this->get('logger')->info('Twitter Stream Process Terminated: ' . $processId);             }         }         exit(0);     }       // ...... }

  Se si guarda attentamente, abbiamo modificato il comando che stiamo per eseguire da terminale tramite il exec() funzione.

  nohup php bin/console twitter:ascoltare > /dev/null 2>&1 & echo $!

  Tutti i bit e pezzi di restituire l’ID del processo il processo in background creato tramite il comando exec() funzione. Ora, ogni volta che un nuovo processo in background è creato a questo tipo di approccio, tutti i nostri precedentemente detenute sfondo processo verrà terminato (se qualsiasi). Questo dovrebbe essere più che sufficiente per garantire che solo una è attiva una connessione alla volta, purché correttamente registro il loro Id di processo. Si può migliorare questa funzionalità in base alle vostre esigenze.

  Tutto ciò che è a sinistra, ora è a processo il flusso di dati che vi troverete a ricevere attraverso la connessione pipeline. Per farlo, facciamo il miglioramento del nostro Servizio di Twitter (TwitterService.php) in primo luogo, e quindi reindirizzare ogni tweet che la nostra applicazione riceve un gestore (chiamiamola processStreamAction) che nel processo di questi tweet. In TwitterService.php,

  /** * Webkul Software. * * @category Webkul, Uvdesk * @package Webkul_UVDesk_TF * @author Akshay Kumar * @copyright Copyright (c) 2010-2016 Webkul Software Private Limited (https://webkul.com) * @license https://store.webkul.com/license.html */ namespace TwitterBundle\Services; const CONSUMER_KEY = YOUR_APP_CONSUMER_KEY; const CONSUMER_SECRET = YOUR_APP_CONSUMER_SECRET; const APPLICATION_TOKEN = YOUR_APP_ACCESS_TOKEN; const APPLICATION_TOKEN_SECRET = YOUR_APP_ACCESS_TOKEN_SECRET; require_once __DIR__ . '/../../../vendor/fennb/phirehose/lib/OauthPhirehose.php'; class TwitterListener extends \OauthPhirehose { private $curlHandler = null; private $processStreamPath = null; public function __construct() { parent::__construct(APPLICATION_TOKEN, APPLICATION_TOKEN_SECRET, \Phirehose::METHOD_FILTER); $this->processStreamPath = URL_TO_PROCESS_STREAM_ACTION_HANDLER; $this->consumerKey = CONSUMER_KEY; $this->consumerSecret = CONSUMER_SECRET; $this->configureCURL(); } private function configureCURL() { $this->curlHandler = curl_init(); $headers = array('Content-type: multipart/form-data'); curl_setopt($this->curlHandler, CURLOPT_POST, true); curl_setopt($this->curlHandler, CURLOPT_URL, $this->processStreamPath); curl_setopt($this->curlHandler, CURLOPT_RETURNTRANSFER, 1); } public function enqueueStatus($status) { $twitterFeed = json_decode($status, true); if (is_array($twitterFeed) && isset($twitterFeed['user']['screen_name'])) { $twitterData = array('feed' => $twitterFeed); curl_setopt($this->curlHandler, CURLOPT_POSTFIELDS, http_build_query($twitterData)); curl_exec($this->curlHandler); } } } class TwitterService { /** * This function will establish a persistent connection with one of the twitter's streaming endpoint. * @param array $twitterTrackCollection Collection of tracks which we'll receive updates for */ public function listenForTweets(array $twitterTrackCollection = []) { // Terminate the process if there are no active tracks if (empty($twitterTrackCollection)) { echo 'No active tracks. Terminating Process...' . PHP_EOL . PHP_EOL; exit(0); } $twitterListener = new TwitterListener(); $twitterListener->setTrack($twitterTrackCollection); $twitterListener->consume(); } }

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71

  /** * Webkul Software. * * @category Webkul, Uvdesk * @package Webkul_UVDesk_TF * @author Akshay Kumar * @copyright Copyright (c) 2010-2016 Webkul Software Private Limited (https://webkul.com) * @license https://store.webkul.com/license.html */   namespace TwitterBundle\Services;   const CONSUMER_KEY = YOUR_APP_CONSUMER_KEY; const CONSUMER_SECRET = YOUR_APP_CONSUMER_SECRET; const APPLICATION_TOKEN = YOUR_APP_ACCESS_TOKEN; const APPLICATION_TOKEN_SECRET = YOUR_APP_ACCESS_TOKEN_SECRET;   require_once __DIR__ . '/../../../vendor/fennb/phirehose/lib/OauthPhirehose.php';   class TwitterListener extends \OauthPhirehose {     private $curlHandler = null;     private $processStreamPath = null;       public function __construct() {         parent::__construct(APPLICATION_TOKEN, APPLICATION_TOKEN_SECRET, \Phirehose::METHOD_FILTER);         $this->processStreamPath = URL_TO_PROCESS_STREAM_ACTION_HANDLER;         $this->consumerKey = CONSUMER_KEY;         $this->consumerSecret = CONSUMER_SECRET;         $this->configureCURL();     }       private function configureCURL()     {         $this->curlHandler = curl_init();         $headers = array('Content-type: multipart/form-data');         curl_setopt($this->curlHandler, CURLOPT_POST, true);         curl_setopt($this->curlHandler, CURLOPT_URL, $this->processStreamPath);         curl_setopt($this->curlHandler, CURLOPT_RETURNTRANSFER, 1);     }       public function enqueueStatus($status)     {           $twitterFeed = json_decode($status, true);         if (is_array($twitterFeed) && isset($twitterFeed['user']['screen_name'])) {             $twitterData = array('feed' => $twitterFeed);             curl_setopt($this->curlHandler, CURLOPT_POSTFIELDS, http_build_query($twitterData));             curl_exec($this->curlHandler);         }     } }   class TwitterService { /** * This function will establish a persistent connection with one of the twitter's streaming endpoint. * @param  array  $twitterTrackCollection Collection of tracks which we'll receive updates for */ public function listenForTweets(array $twitterTrackCollection = [])     {      // Terminate the process if there are no active tracks      if (empty($twitterTrackCollection)) {      echo 'No active tracks. Terminating Process...' . PHP_EOL . PHP_EOL;      exit(0);      }        $twitterListener = new TwitterListener();         $twitterListener->setTrack($twitterTrackCollection);         $twitterListener->consume();     } }

  Ora, ogni volta che un feed twitter è ricevuto, TwitterListener::enqueueStatus() di trasmettere il feed per il percorso che abbiamo specificato utilizzando CURL facendo una richiesta POST. In sostanza, questo è esattamente come un Webhook se ci pensate. Proviamo ora a scrivere il processStreamAction() gestore nel nostro DefaultController. In DefaultController.php,

  /** * Webkul Software. * * @category Webkul, Uvdesk * @package Webkul_UVDesk_TF * @author Akshay Kumar * @copyright Copyright (c) 2010-2016 Webkul Software Private Limited (https://webkul.com) * @license https://store.webkul.com/license.html */ namespace TwitterBundle\Controller; use Symfony\Component\HttpFoundation\Request; use Symfony\Component\HttpFoundation\Response; use Symfony\Bundle\FrameworkBundle\Controller\Controller; use Symfony\Component\Routing\Generator\UrlGeneratorInterface; class DefaultController extends Controller { /** * This handler will first terminate any previously held background process * by firing the stopStreamAction() handler. Then it'll create a new background * process which establish a new connection with a streaming endpoint. */ public function startStreamAction(Request $request) { // Execute the stopStreamAction() handler $curlHandler = curl_init(); $stopStreamPath = $this->generateUrl('twitter_stream_stop', array(), UrlGeneratorInterface::ABSOLUTE_URL); curl_setopt($curlHandler, CURLOPT_URL, $stopStreamPath); curl_setopt($curlHandler, CURLOPT_RETURNTRANSFER, 1); curl_exec($curlHandler); // Create a new background process chdir('../'); $output = array(); $command = 'nohup php bin/console twitter:listen > /dev/null 2>&1 & echo $!'; exec($command, $output); $processId = (int) $output[0]; // Process ID of the newly created background process. Log this. exit(0); } /** * This handler will fetch a log of Process IDs corresponding to any background * process and terminate them. */ public function stopStreamAction(Request $request) { $activeProcesses = ARRAY_COLLECTION_OF_PROCESS_IDS_TO_TERMINATE; if (!empty($activeProcesses)) { foreach ($activeProcesses as $processId) { $command = 'kill ' . $processId . ' 2>&1'; exec($command, $output); $this->get('logger')->info('Twitter Stream Process Terminated: ' . $processId); } } exit(0); } /** * A POST request will be made to this handler whenever our application receives * a feed. You can write your implementation here. */ public function processStreamAction(Request $request) { if (isset($_POST)) { if (array_key_exists('feed', $_POST)) { // Real-time Twitter Feed. Write your own implementation over here. $streamContent = $_POST['feed']; } } else { // Request method not supported. } exit(0); } }

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77

  /** * Webkul Software. * * @category Webkul, Uvdesk * @package Webkul_UVDesk_TF * @author Akshay Kumar * @copyright Copyright (c) 2010-2016 Webkul Software Private Limited (https://webkul.com) * @license https://store.webkul.com/license.html */   namespace TwitterBundle\Controller;   use Symfony\Component\HttpFoundation\Request; use Symfony\Component\HttpFoundation\Response; use Symfony\Bundle\FrameworkBundle\Controller\Controller; use Symfony\Component\Routing\Generator\UrlGeneratorInterface;   class DefaultController extends Controller {     /**      * This handler will first terminate any previously held background process        * by firing the stopStreamAction() handler. Then it'll create a new background        * process which establish a new connection with a streaming endpoint.      */     public function startStreamAction(Request $request)     {      // Execute the stopStreamAction() handler         $curlHandler = curl_init();         $stopStreamPath = $this->generateUrl('twitter_stream_stop', array(), UrlGeneratorInterface::ABSOLUTE_URL);         curl_setopt($curlHandler, CURLOPT_URL, $stopStreamPath);         curl_setopt($curlHandler, CURLOPT_RETURNTRANSFER, 1);         curl_exec($curlHandler);           // Create a new background process         chdir('../');         $output = array();         $command = 'nohup php bin/console twitter:listen > /dev/null 2>&1 & echo $!';         exec($command, $output);         $processId = (int) $output[0]; // Process ID of the newly created background process. Log this.         exit(0);     }       /**      * This handler will fetch a log of Process IDs corresponding to any background      * process and terminate them.      */     public function stopStreamAction(Request $request)     {         $activeProcesses = ARRAY_COLLECTION_OF_PROCESS_IDS_TO_TERMINATE;           if (!empty($activeProcesses)) {             foreach ($activeProcesses as $processId) {                 $command = 'kill ' . $processId . ' 2>&1';                 exec($command, $output);                 $this->get('logger')->info('Twitter Stream Process Terminated: ' . $processId);             }         }         exit(0);     }       /**      * A POST request will be made to this handler whenever our application receives      * a feed. You can write your implementation here.      */     public function processStreamAction(Request $request)     {         if (isset($_POST)) {             if (array_key_exists('feed', $_POST)) {              // Real-time Twitter Feed. Write your own implementation over here.                 $streamContent = $_POST['feed'];             }         } else {             // Request method not supported.         }         exit(0);     } }

  Ci si va! Ora abbiamo una cucina completamente funzionale applicazione che consente non solo di ascoltare in diretta i feed di Twitter per l’specificato tracce, ma ci permette anche di aggiornare le nostre tracce quasi istantaneamente, facendo in modo che solo una connessione con lo streaming endpoint è attivo al momento. È possibile estendere l’applicazione a proprio piacimento e renderlo più robusto e a prova di errore. Qui a UVDesk, seguiamo un approccio simile a fornire ai nostri Utenti di Twitter con una migliore user-experience e connettività sociale.