Skip to content

Commit 0331ab7

Browse files
WyriHaximusWyriHaximus
committed
Split off the streaming from the clients into streaming clients
1 parent 74f34ea commit 0331ab7

22 files changed

+396
-60
lines changed

‎composer.lock

+4-4
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎examples/streaming-filter-hashtag-async.php

+4-2
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@
1616
))->withAccessToken(
1717
$config['access_token']['token'],
1818
$config['access_token']['secret']
19-
);
19+
)->stream();
2020

2121
$hashtags = [];
2222

@@ -27,7 +27,9 @@
2727
}
2828
}
2929

30-
$client->filteredStream([
30+
$hashtags = array_unique($hashtags);
31+
32+
$client->filtered([
3133
'track' => implode(',', $hashtags),
3234
])->subscribeCallback(function ($document) {
3335
resource_pretty_print($document);

‎examples/streaming-filter-hashtag.php

+36
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
<?php
2+
3+
use ApiClients\Client\Twitter\Client;
4+
use function ApiClients\Foundation\resource_pretty_print;
5+
use function React\Promise\all;
6+
7+
require dirname(__DIR__) . DIRECTORY_SEPARATOR . 'vendor/autoload.php';
8+
$config = require 'resolve_config.php';
9+
10+
$client = (new Client(
11+
$config['consumer']['key'],
12+
$config['consumer']['secret']
13+
))->withAccessToken(
14+
$config['access_token']['token'],
15+
$config['access_token']['secret']
16+
)->stream();
17+
18+
$hashtags = [];
19+
20+
if (count($argv) > 1) {
21+
unset($argv[0]);
22+
foreach ($argv as $hashtag) {
23+
$hashtags[] = '#' . $hashtag;
24+
}
25+
}
26+
27+
$hashtags = array_unique($hashtags);
28+
29+
$client->filtered(
30+
function ($document) {
31+
resource_pretty_print($document);
32+
},
33+
[
34+
'track' => implode(',', $hashtags),
35+
]
36+
);

‎examples/streaming-filter-track-async.php

+4-2
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@
1717
))->withAccessToken(
1818
$config['access_token']['token'],
1919
$config['access_token']['secret']
20-
);
20+
)->stream();
2121

2222
$users = [
2323
'WyriHaximus',
@@ -30,7 +30,9 @@
3030
}
3131
}
3232

33-
$client->filteredStream([
33+
$tracks = array_unique($tracks);
34+
35+
$client->filtered([
3436
'track' => implode(',', $tracks),
3537
])->subscribeCallback(function ($document) {
3638
resource_pretty_print($document);

‎examples/streaming-filter-track.php

+38
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
<?php
2+
3+
use ApiClients\Client\Twitter\Client;
4+
use function ApiClients\Foundation\resource_pretty_print;
5+
use function React\Promise\all;
6+
7+
require dirname(__DIR__) . DIRECTORY_SEPARATOR . 'vendor/autoload.php';
8+
$config = require 'resolve_config.php';
9+
10+
$client = (new Client(
11+
$config['consumer']['key'],
12+
$config['consumer']['secret']
13+
))->withAccessToken(
14+
$config['access_token']['token'],
15+
$config['access_token']['secret']
16+
)->stream();
17+
18+
$users = [
19+
'WyriHaximus',
20+
];
21+
22+
if (count($argv) > 1) {
23+
unset($argv[0]);
24+
foreach ($argv as $track) {
25+
$tracks[] = $track;
26+
}
27+
}
28+
29+
$tracks = array_unique($tracks);
30+
31+
$client->filtered(
32+
function ($document) {
33+
resource_pretty_print($document);
34+
},
35+
[
36+
'track' => implode(',', $tracks),
37+
]
38+
);

‎examples/streaming-filter-users-async.php

+3-2
Original file line numberDiff line numberDiff line change
@@ -30,8 +30,9 @@
3030
}
3131
}
3232

33-
$promises = [];
33+
$users = array_unique($users);
3434

35+
$promises = [];
3536
foreach ($users as $user) {
3637
$promises[] = $client->user($user);
3738
}
@@ -43,7 +44,7 @@
4344
$userIds[] = $user->idStr();
4445
}
4546

46-
$client->filteredStream([
47+
$client->stream()->filtered([
4748
'follow' => implode(',', $userIds),
4849
])->subscribeCallback(function ($document) {
4950
resource_pretty_print($document);

‎examples/streaming-filter-users.php

+43
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
<?php
2+
3+
use ApiClients\Client\Twitter\Client;
4+
use function ApiClients\Foundation\resource_pretty_print;
5+
use function React\Promise\all;
6+
7+
require dirname(__DIR__) . DIRECTORY_SEPARATOR . 'vendor/autoload.php';
8+
$config = require 'resolve_config.php';
9+
10+
$client = (new Client(
11+
$config['consumer']['key'],
12+
$config['consumer']['secret']
13+
))->withAccessToken(
14+
$config['access_token']['token'],
15+
$config['access_token']['secret']
16+
);
17+
18+
$users = [
19+
'WyriHaximus',
20+
];
21+
22+
if (count($argv) > 1) {
23+
unset($argv[0]);
24+
foreach ($argv as $user) {
25+
$users[] = $user;
26+
}
27+
}
28+
29+
$users = array_unique($users);
30+
31+
$userIds = [];
32+
foreach ($users as $user) {
33+
$userIds[] = $client->user($user)->idStr();
34+
}
35+
36+
$client->stream()->filtered(
37+
function ($document) {
38+
resource_pretty_print($document);
39+
},
40+
[
41+
'follow' => implode(',', $userIds),
42+
]
43+
);

‎examples/streaming-sample-async.php

+2-2
Original file line numberDiff line numberDiff line change
@@ -15,9 +15,9 @@
1515
))->withAccessToken(
1616
$config['access_token']['token'],
1717
$config['access_token']['secret']
18-
);
18+
)->stream();
1919

20-
$client->sampleStream()->subscribeCallback(function ($document) {
20+
$client->sample()->subscribeCallback(function ($document) {
2121
resource_pretty_print($document);
2222
});
2323

‎examples/streaming-sample-counters-async.php

+2-2
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818
))->withAccessToken(
1919
$config['access_token']['token'],
2020
$config['access_token']['secret']
21-
);
21+
)->stream();
2222

2323
$counters = [
2424
DeletedTweet::class => 0,
@@ -43,7 +43,7 @@
4343
}
4444
});
4545

46-
$client->sampleStream()->subscribeCallback(function ($document) use(&$counters) {
46+
$client->sample()->subscribeCallback(function ($document) use(&$counters) {
4747
$class = get_class($document);
4848

4949
if (!isset($counters[$class])) {

‎examples/streaming-sample.php

+19
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
<?php
2+
3+
use ApiClients\Client\Twitter\Client;
4+
use function ApiClients\Foundation\resource_pretty_print;
5+
6+
require dirname(__DIR__) . DIRECTORY_SEPARATOR . 'vendor/autoload.php';
7+
$config = require 'resolve_config.php';
8+
9+
(new Client(
10+
$config['consumer']['key'],
11+
$config['consumer']['secret']
12+
))->withAccessToken(
13+
$config['access_token']['token'],
14+
$config['access_token']['secret']
15+
)->stream()->sample(function ($document) {
16+
resource_pretty_print($document);
17+
}, function ($e) {
18+
echo (string)$e;
19+
});

‎examples/user-async.php

+2
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,8 @@
3030
}
3131
}
3232

33+
$users = array_unique($users);
34+
3335
foreach ($users as $user) {
3436
$client->user($user)->done(function (UserInterface $user) {
3537
resource_pretty_print($user);

‎examples/user.php

+2
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@
2525
}
2626
}
2727

28+
$users = array_unique($users);
29+
2830
foreach ($users as $user) {
2931
resource_pretty_print($client->user($user));
3032
}

‎resources.yml

+3-3
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
11
yaml_location: yaml
2-
api_settings: ApiClients\Twitter\ApiSettings
2+
api_settings: ApiClients\Client\Twitter\ApiSettings
33
src:
44
path: src/Resource
5-
namespace: ApiClients\Twitter\Resource
5+
namespace: ApiClients\Client\Twitter\Resource
66
tests:
77
path: tests/Resource
8-
namespace: ApiClients\Tests\Twitter\Resource
8+
namespace: ApiClients\Tests\Client\Twitter\Resource

‎src/ApiSettings.php

+1-1
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@
1111
use ApiClients\Foundation\Oauth1\Options as Oauth1Options;
1212
use ApiClients\Tools\Psr7\Oauth1\Definition;
1313

14-
class ApiSettings
14+
final class ApiSettings
1515
{
1616
const NAMESPACE = 'ApiClients\\Client\\Twitter\\Resource';
1717

‎src/AsyncClient.php

+16-40
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
use ApiClients\Foundation\Transport\CommandBus\Command\RequestCommand;
1212
use ApiClients\Foundation\Transport\CommandBus\Command\StreamingRequestCommand;
1313
use ApiClients\Foundation\Transport\Options as TransportOptions;
14+
use ApiClients\Tools\CommandBus\CommandBus;
1415
use ApiClients\Tools\Psr7\Oauth1\Definition;
1516
use GuzzleHttp\Psr7\Request;
1617
use Psr\Http\Message\RequestInterface;
@@ -24,8 +25,6 @@
2425

2526
final class AsyncClient
2627
{
27-
const STREAM_DELIMITER = "\r\n";
28-
2928
/**
3029
* @var string
3130
*/
@@ -46,6 +45,11 @@ final class AsyncClient
4645
*/
4746
protected $client;
4847

48+
/**
49+
* @var AsyncStreamingClient
50+
*/
51+
protected $streamingClient;
52+
4953
public function __construct(
5054
string $consumerKey,
5155
string $consumerSecret,
@@ -107,6 +111,11 @@ public function withOutAccessToken(): AsyncClient
107111
);
108112
}
109113

114+
public function getCommandBus(): CommandBus
115+
{
116+
return $this->client->getFromContainer(CommandBus::class);
117+
}
118+
110119
public function user(string $user): PromiseInterface
111120
{
112121
return $this->client->handle(new RequestCommand(
@@ -116,45 +125,12 @@ public function user(string $user): PromiseInterface
116125
});
117126
}
118127

119-
public function sampleStream(): Observable
120-
{
121-
return $this->stream(
122-
new Request('GET', 'https://stream.twitter.com/1.1/statuses/sample.json')
123-
);
124-
}
125-
126-
public function filteredStream(array $filter = []): Observable
128+
public function stream(): AsyncStreamingClient
127129
{
128-
$postData = http_build_query($filter);
129-
return $this->stream(
130-
new Request(
131-
'POST',
132-
'https://stream.twitter.com/1.1/statuses/filter.json',
133-
[
134-
'Content-Type' => 'application/x-www-form-urlencoded',
135-
'Content-Length' => strlen($postData),
136-
],
137-
$postData
138-
)
139-
);
140-
}
130+
if (!($this->streamingClient instanceof AsyncStreamingClient)) {
131+
$this->streamingClient = new AsyncStreamingClient($this->client);
132+
}
141133

142-
protected function stream(RequestInterface $request): Observable
143-
{
144-
return Promise::toObservable($this->client->handle(new StreamingRequestCommand(
145-
$request
146-
)))->switchLatest()->lift(function () {
147-
return new CutOperator(self::STREAM_DELIMITER);
148-
})->filter(function (string $json) {
149-
return trim($json) !== ''; // To keep the stream alive Twitter sends an empty line at times
150-
})->jsonDecode()->flatMap(function (array $document) {
151-
if (isset($document['delete'])) {
152-
return Promise::toObservable($this->client->handle(
153-
new HydrateCommand('DeletedTweet', $document['delete'])
154-
));
155-
}
156-
157-
return Promise::toObservable($this->client->handle(new HydrateCommand('Tweet', $document)));
158-
});
134+
return $this->streamingClient;
159135
}
160136
}

0 commit comments

Comments
 (0)