Skip to content

Commit cc6421c

Browse files
davidwdanmbonneau
davidwdan
authored and
mbonneau
committed
Added singleInstance operator (#185)
1 parent c915843 commit cc6421c

File tree

4 files changed

+197
-1
lines changed

4 files changed

+197
-1
lines changed

‎demo/share/singleInstance.php

+24
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
<?php
2+
3+
require_once __DIR__ . '/../bootstrap.php';
4+
5+
$interval = Rx\Observable::interval(1000);
6+
7+
$source = $interval
8+
->take(2)
9+
->do(function () {
10+
echo 'Side effect', PHP_EOL;
11+
});
12+
13+
$single = $source->singleInstance();
14+
15+
// two simultaneous subscriptions, lasting 2 seconds
16+
$single->subscribe($createStdoutObserver('SourceA '));
17+
$single->subscribe($createStdoutObserver('SourceB '));
18+
19+
\Rx\Observable::timer(5000)->subscribe(function () use ($single, &$createStdoutObserver) {
20+
// resubscribe two times again, more than 5 seconds later,
21+
// long after the original two subscriptions have ended
22+
$single->subscribe($createStdoutObserver('SourceC '));
23+
$single->subscribe($createStdoutObserver('SourceD '));
24+
});

‎demo/share/singleInstance.php.expect

+16
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
Side effect
2+
SourceA Next value: 0
3+
SourceB Next value: 0
4+
Side effect
5+
SourceA Next value: 1
6+
SourceB Next value: 1
7+
SourceA Complete!
8+
SourceB Complete!
9+
Side effect
10+
SourceC Next value: 0
11+
SourceD Next value: 0
12+
Side effect
13+
SourceC Next value: 1
14+
SourceD Next value: 1
15+
SourceC Complete!
16+
SourceD Complete!

‎src/Observable.php

+37-1
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@
1111
use Rx\Observable\ConnectableObservable;
1212
use Rx\Observable\EmptyObservable;
1313
use Rx\Observable\ErrorObservable;
14-
use Rx\Observable\FromPromiseObservable;
1514
use Rx\Observable\ForkJoinObservable;
1615
use Rx\Observable\IntervalObservable;
1716
use Rx\Observable\IteratorObservable;
@@ -1312,6 +1311,43 @@ public function share(): RefCountObservable
13121311
return $this->publish()->refCount();
13131312
}
13141313

1314+
/**
1315+
* Returns an observable sequence that shares a single subscription to the underlying sequence. This observable sequence
1316+
* can be resubscribed to, even if all prior subscriptions have ended.
1317+
*
1318+
* This operator behaves like share() in RxJS 5
1319+
*
1320+
* @return \Rx\Observable An observable sequence that contains the elements of a sequence
1321+
* produced by multicasting the source sequence.
1322+
*
1323+
* @demo share/singleInstance.php
1324+
* @operator
1325+
* @reactivex refcount
1326+
*/
1327+
public function singleInstance(): Observable
1328+
{
1329+
$hasObservable = false;
1330+
$observable = null;
1331+
$source = $this;
1332+
1333+
$getObservable = function () use (&$hasObservable, &$observable, $source): Observable {
1334+
if (!$hasObservable) {
1335+
$hasObservable = true;
1336+
$observable = $source
1337+
->finally(function () use (&$hasObservable) {
1338+
$hasObservable = false;
1339+
})
1340+
->publish()
1341+
->refCount();
1342+
}
1343+
return $observable;
1344+
};
1345+
1346+
return new Observable\AnonymousObservable(function (ObserverInterface $o) use ($getObservable) {
1347+
return $getObservable()->subscribe($o);
1348+
});
1349+
}
1350+
13151351
/**
13161352
* Returns an observable sequence that shares a single subscription to the underlying sequence and starts with an
13171353
* initialValue.

‎test/Rx/Functional/Operator/SingleInstanceTest.php

+120
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Rx\Functional\Operator;
6+
7+
use Rx\Disposable\CompositeDisposable;
8+
use Rx\Disposable\SerialDisposable;
9+
use Rx\Functional\FunctionalTestCase;
10+
11+
class SingleInstanceTest extends FunctionalTestCase
12+
{
13+
/**
14+
* @test
15+
*/
16+
public function singleInstance_basic()
17+
{
18+
$xs = $this->createColdObservable([
19+
onNext(100, 1),
20+
onNext(150, 2),
21+
onNext(200, 3),
22+
onCompleted(250)
23+
]);
24+
25+
$ys = null;
26+
$results1 = $this->scheduler->createObserver();
27+
$results2 = $this->scheduler->createObserver();
28+
$disposable = null;
29+
30+
$this->scheduler->scheduleAbsolute($this->scheduler::CREATED, function () use (&$ys, $xs) {
31+
$ys = $xs->singleInstance();
32+
});
33+
34+
$this->scheduler->scheduleAbsolute($this->scheduler::SUBSCRIBED, function () use (&$ys, &$disposable, $results1, $results2) {
35+
$disposable = new CompositeDisposable([
36+
$ys->subscribe($results1),
37+
$ys->subscribe($results2)
38+
]);
39+
});
40+
41+
$this->scheduler->scheduleAbsolute($this->scheduler::DISPOSED, function () use (&$disposable) {
42+
$disposable->dispose();
43+
});
44+
45+
$this->scheduler->start();
46+
47+
$this->assertMessages([
48+
onNext(300, 1),
49+
onNext(350, 2),
50+
onNext(400, 3),
51+
onCompleted(450)
52+
], $results1->getMessages());
53+
54+
$this->assertMessages([
55+
onNext(300, 1),
56+
onNext(350, 2),
57+
onNext(400, 3),
58+
onCompleted(450)
59+
], $results2->getMessages());
60+
61+
$this->assertSubscriptions([
62+
subscribe(200, 450)
63+
], $xs->getSubscriptions());
64+
}
65+
66+
/**
67+
* @test
68+
*/
69+
public function singleInstance_subscribe_after_stopped()
70+
{
71+
$xs = $this->createColdObservable([
72+
onNext(100, 1),
73+
onNext(150, 2),
74+
onNext(200, 3),
75+
onCompleted(250)
76+
]);
77+
78+
$ys = null;
79+
$results1 = $this->scheduler->createObserver();
80+
$results2 = $this->scheduler->createObserver();
81+
$disposable = new SerialDisposable();
82+
83+
$this->scheduler->scheduleAbsolute(100, function () use (&$ys, $xs) {
84+
$ys = $xs->singleInstance();
85+
});
86+
87+
$this->scheduler->scheduleAbsolute(200, function () use (&$ys, $disposable, $results1) {
88+
$disposable->setDisposable($ys->subscribe($results1));
89+
});
90+
91+
$this->scheduler->scheduleAbsolute(600, function () use (&$ys, $disposable, $results2) {
92+
$disposable->setDisposable($ys->subscribe($results2));
93+
});
94+
95+
$this->scheduler->scheduleAbsolute(900, function () use (&$disposable) {
96+
$disposable->dispose();
97+
});
98+
99+
$this->scheduler->start();
100+
101+
$this->assertMessages([
102+
onNext(300, 1),
103+
onNext(350, 2),
104+
onNext(400, 3),
105+
onCompleted(450)
106+
], $results1->getMessages());
107+
108+
$this->assertMessages([
109+
onNext(700, 1),
110+
onNext(750, 2),
111+
onNext(800, 3),
112+
onCompleted(850)
113+
], $results2->getMessages());
114+
115+
$this->assertSubscriptions([
116+
subscribe(200, 450),
117+
subscribe(600, 850)
118+
], $xs->getSubscriptions());
119+
}
120+
}

0 commit comments

Comments
 (0)