Search tweets

Author: WyriHaximus

examples/search-tweets-async.php

@@ -0,0 +1,39 @@
+<?php
+
+use ApiClients\Client\Twitter\AsyncClient;
+use React\EventLoop\Factory;
+use function ApiClients\Foundation\resource_pretty_print;
+
+require dirname(__DIR__) . DIRECTORY_SEPARATOR . 'vendor/autoload.php';
+$config = require 'resolve_config.php';
+
+$loop = Factory::create();
+$client = (new AsyncClient(
+    $config['consumer']['key'],
+    $config['consumer']['secret'],
+    $loop
+))->withAccessToken(
+    $config['access_token']['token'],
+    $config['access_token']['secret']
+)->stream();
+
+$hashtags = [];
+
+if (count($argv) > 1) {
+    unset($argv[0]);
+    foreach ($argv as $hashtag) {
+        $hashtags[] = '#' . $hashtag;
+    }
+}
+
+$hashtags = array_unique($hashtags);
+
+$client->searchTweets([
+    // Any parameters from: https://developer.twitter.com/en/docs/tweets/search/api-reference/get-search-tweets.html
+    'q' => 'nasa',
+    'result_type' => 'popular',
+])->subscribe(function ($document) {
+    resource_pretty_print($document);
+});
+
+$loop->run();

src/AsyncStreamingClient.php

@@ -4,14 +4,19 @@
 
 use ApiClients\Foundation\Client;
 use ApiClients\Foundation\Hydrator\CommandBus\Command\HydrateCommand;
+use ApiClients\Foundation\Transport\CommandBus\Command\RequestCommand;
 use ApiClients\Foundation\Transport\CommandBus\Command\StreamingRequestCommand;
+use ApiClients\Foundation\Transport\ParsedContentsInterface;
 use GuzzleHttp\Psr7\Request;
 use Psr\Http\Message\RequestInterface;
+use Psr\Http\Message\ResponseInterface;
 use React\EventLoop\LoopInterface;
 use Rx\Observable;
 use Rx\Operator\CutOperator;
 use Rx\React\Promise;
 use Rx\Scheduler\ImmediateScheduler;
+use function ApiClients\Tools\Rx\observableFromArray;
+use function RingCentral\Psr7\build_query;
 
 final class AsyncStreamingClient implements AsyncStreamingClientInterface
 {
@@ -56,6 +61,28 @@ public function filtered(array $filter = []): Observable
         );
     }
 
+    public function searchTweets(array $filter = []): Observable
+    {
+        $query = build_query($filter);
+
+        return Promise::toObservable($this->client->handle(new RequestCommand(
+            new Request(
+                'GET',
+                'https://api.twitter.com/1.1/search/tweets.json?' . $query,
+                []
+            )
+        ))->then(function (ResponseInterface $response) {
+            /** @var ParsedContentsInterface $body */
+            $body = $response->getBody();
+
+            return $body->getParsedContents()['statuses'];
+        }))->flatMap(function (array $statuses) {
+            return observableFromArray($statuses);
+        })->flatMap(function (array $document) {
+            return Promise::toObservable($this->client->handle(new HydrateCommand('Tweet', $document)));
+        });
+    }
+
     protected function stream(RequestInterface $request): Observable
     {
         return Promise::toObservable($this->client->handle(new StreamingRequestCommand(

src/AsyncStreamingClientInterface.php

@@ -9,4 +9,6 @@ interface AsyncStreamingClientInterface
     public function sample(): Observable;
 
     public function filtered(array $filter = []): Observable;
+
+    public function searchTweets(array $filter): Observable;
 }