Skip to content
This repository was archived by the owner on Jul 19, 2024. It is now read-only.

Commit d15d1f8

Browse files
geyer-zageyer-za
authored
Merge pull request #13 from Superbalist/rest-subscriber-issue
pull return immediately, with sleep
2 parents 47b8be5 + 70f32ba commit d15d1f8

File tree

3 files changed

+162
-2
lines changed

3 files changed

+162
-2
lines changed

‎changelog.md

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
11
# Changelog
22

3+
## 5.2.0 - 2019-03-19
4+
5+
* Add ability to use returnImmediately flag when pulling messages
6+
37
## 5.1.0 - 2019-02-28
48

59
* Bump up google/cloud requirement to ^0.95.0

‎src/GoogleCloudPubSubAdapter.php

Lines changed: 55 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,16 @@ class GoogleCloudPubSubAdapter implements PubSubAdapterInterface
4040
*/
4141
protected $maxMessages;
4242

43+
/**
44+
* @var bool
45+
*/
46+
protected $returnImmediately;
47+
48+
/**
49+
* @var int
50+
*/
51+
protected $returnImmediatelyPause;
52+
4353
/**
4454
* @param PubSubClient $client
4555
* @param string $clientIdentifier
@@ -54,14 +64,18 @@ public function __construct(
5464
$autoCreateTopics = true,
5565
$autoCreateSubscriptions = true,
5666
$backgroundBatching = false,
57-
$maxMessages = 1000
67+
$maxMessages = 1000,
68+
$returnImmediately = false,
69+
$returnImmediatelyPause = 500000
5870
) {
5971
$this->client = $client;
6072
$this->clientIdentifier = $clientIdentifier;
6173
$this->autoCreateTopics = $autoCreateTopics;
6274
$this->autoCreateSubscriptions = $autoCreateSubscriptions;
6375
$this->backgroundBatching = $backgroundBatching;
6476
$this->maxMessages = $maxMessages;
77+
$this->returnImmediately = $returnImmediately;
78+
$this->returnImmediatelyPause = (int) $returnImmediatelyPause;
6579
}
6680

6781
/**
@@ -140,6 +154,40 @@ public function areSubscriptionsAutoCreated()
140154
return $this->autoCreateSubscriptions;
141155
}
142156

157+
/**
158+
* Set if a pull should return immediately if there are no messages
159+
* @param bool $returnImmediately
160+
*/
161+
public function setReturnImmediately($returnImmediately) {
162+
$this->returnImmediately = $returnImmediately;
163+
}
164+
165+
/**
166+
* Return the return immediately configuration
167+
* @return bool
168+
*/
169+
public function getReturnImmediately() {
170+
return $this->returnImmediately;
171+
}
172+
173+
/**
174+
* Set the amount of time to pause between attempts to pull messages if return immediately is enabled.
175+
* Value is in microseconds
176+
*
177+
* @param int $returnImmediatelyPause
178+
*/
179+
public function setReturnImmediatelyPause($returnImmediatelyPause) {
180+
$this->returnImmediatelyPause = (int) $returnImmediatelyPause;
181+
}
182+
183+
/**
184+
* Return the return immediately pause configuration
185+
* @return int
186+
*/
187+
public function getReturnImmediatelyPause() {
188+
return $this->returnImmediatelyPause;
189+
}
190+
143191
/**
144192
* Set whether or not background batching is enabled.
145193
*
@@ -191,15 +239,20 @@ public function subscribe($channel, callable $handler)
191239
$subscription = $this->getSubscriptionForChannel($channel);
192240

193241
$isSubscriptionLoopActive = true;
242+
$isPauseEnabled = $this->returnImmediately && ($this->returnImmediatelyPause > 0);
194243

195244
while ($isSubscriptionLoopActive) {
196245
$messages = $subscription->pull([
197246
'grpcOptions' => [
198247
'timeoutMillis' => null,
199248
],
200249
'maxMessages' => $this->maxMessages,
250+
'returnImmediately' => $this->returnImmediately,
201251
]);
202-
252+
if ($isPauseEnabled && empty($messages)) {
253+
usleep($this->returnImmediatelyPause);
254+
continue;
255+
}
203256
foreach ($messages as $message) {
204257
/** @var Message $message */
205258
$payload = Utils::unserializeMessagePayload($message->data());

‎tests/GoogleCloudPubSubAdapterTest.php

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,26 @@ public function testGetSetBackgroundBatching()
6363
$this->assertTrue($adapter->isBackgroundBatchingEnabled());
6464
}
6565

66+
public function testGetSetReturnImmediately()
67+
{
68+
$client = Mockery::mock(PubSubClient::class);
69+
$adapter = new GoogleCloudPubSubAdapter($client);
70+
$this->assertFalse($adapter->getReturnImmediately());
71+
72+
$adapter->setReturnImmediately(true);
73+
$this->assertTrue($adapter->getReturnImmediately());
74+
}
75+
76+
public function testGetSetReturnImmediatelyPause()
77+
{
78+
$client = Mockery::mock(PubSubClient::class);
79+
$adapter = new GoogleCloudPubSubAdapter($client);
80+
$this->assertEquals(500000, $adapter->getReturnImmediatelyPause());
81+
82+
$adapter->setReturnImmediatelyPause(1000000);
83+
$this->assertEquals(1000000, $adapter->getReturnImmediatelyPause());
84+
}
85+
6686
public function testPublishWhenTopicMustBeCreated()
6787
{
6888
$topic = Mockery::mock(Topic::class);
@@ -320,6 +340,7 @@ public function testSubscribeWhenSubscriptionMustBeCreated()
320340
'timeoutMillis' => null,
321341
],
322342
'maxMessages' => 1000,
343+
'returnImmediately' => false
323344
])
324345
->once()
325346
->andReturn($messageBatch1);
@@ -336,6 +357,7 @@ public function testSubscribeWhenSubscriptionMustBeCreated()
336357
'timeoutMillis' => null,
337358
],
338359
'maxMessages' => 1000,
360+
'returnImmediately' => false
339361
])
340362
->once()
341363
->andReturn($messageBatch2);
@@ -397,6 +419,7 @@ public function testSubscribeWhenSubscriptionExists()
397419
'timeoutMillis' => null,
398420
],
399421
'maxMessages' => 1000,
422+
'returnImmediately' => false
400423
])
401424
->once()
402425
->andReturn($messageBatch1);
@@ -412,6 +435,7 @@ public function testSubscribeWhenSubscriptionExists()
412435
'timeoutMillis' => null,
413436
],
414437
'maxMessages' => 1000,
438+
'returnImmediately' => false
415439
])
416440
->once()
417441
->andReturn($messageBatch2);
@@ -471,6 +495,7 @@ public function testSubscribeWhenAutoTopicCreationIsDisabled()
471495
'timeoutMillis' => null,
472496
],
473497
'maxMessages' => 1000,
498+
'returnImmediately' => false
474499
])
475500
->once()
476501
->andReturn($messageBatch1);
@@ -486,6 +511,7 @@ public function testSubscribeWhenAutoTopicCreationIsDisabled()
486511
'timeoutMillis' => null,
487512
],
488513
'maxMessages' => 1000,
514+
'returnImmediately' => false
489515
])
490516
->once()
491517
->andReturn($messageBatch2);
@@ -521,4 +547,81 @@ public function testSubscribeWhenAutoTopicCreationIsDisabled()
521547

522548
$adapter->subscribe('channel_name', [$handler1, 'handle']);
523549
}
550+
551+
public function testSubscribeWhenReturnImmediatelyIsEnabled()
552+
{
553+
$message1 = new Message(['data' => '{"hello":"world"}'], ['ackId' => 1]);
554+
$message2 = new Message(['data' => '"this is a string"'], ['ackId' => 2]);
555+
$message3 = new Message(['data' => '"unsubscribe"'], ['ackId' => 3]);
556+
557+
$messageBatch1 = [
558+
$message1,
559+
$message2,
560+
];
561+
562+
$messageBatch2 = [
563+
$message3,
564+
];
565+
566+
$subscription = Mockery::mock(Subscription::class);
567+
$subscription->shouldReceive('exists')
568+
->once()
569+
->andReturn(true);
570+
$subscription->shouldNotHaveReceived('create');
571+
572+
$expectedPullOptions = [
573+
'grpcOptions' => [
574+
'timeoutMillis' => null,
575+
],
576+
'maxMessages' => 1000,
577+
'returnImmediately' => true
578+
];
579+
580+
$subscription->shouldReceive('pull')
581+
->with($expectedPullOptions)
582+
->once()
583+
->andReturn($messageBatch1);
584+
$subscription->shouldReceive('acknowledge')
585+
->with($message1)
586+
->once();
587+
$subscription->shouldReceive('acknowledge')
588+
->with($message2)
589+
->once();
590+
591+
$subscription->shouldReceive('pull')
592+
->with($expectedPullOptions)
593+
->once()
594+
->andReturn($messageBatch2);
595+
$subscription->shouldReceive('acknowledge')
596+
->with($message3)
597+
->once();
598+
599+
$topic = Mockery::mock(Topic::class);
600+
$topic->shouldReceive('exists')
601+
->once()
602+
->andReturn(true);
603+
$topic->shouldNotHaveReceived('create');
604+
$topic->shouldReceive('subscription')
605+
->with('default.channel_name')
606+
->once()
607+
->andReturn($subscription);
608+
609+
$client = Mockery::mock(PubSubClient::class);
610+
$client->shouldReceive('topic')
611+
->with('channel_name')
612+
->once()
613+
->andReturn($topic);
614+
615+
$handler1 = Mockery::mock(\stdClass::class);
616+
$handler1->shouldReceive('handle')
617+
->with(['hello' => 'world'])
618+
->once();
619+
$handler1->shouldReceive('handle')
620+
->with('this is a string')
621+
->once();
622+
623+
$adapter = new GoogleCloudPubSubAdapter($client);
624+
$adapter->setReturnImmediately(true);
625+
$adapter->subscribe('channel_name', [$handler1, 'handle']);
626+
}
524627
}

0 commit comments

Comments
 (0)