Skip to content

Commit c20cc5a

Browse files
makasimmakasim
authored
Merge pull request #1018 from versh23/stomp-connection
STOMP. add additional configuration
2 parents 34a0f6a + 2b8ade3 commit c20cc5a

4 files changed

Lines changed: 112 additions & 1 deletion

File tree

‎pkg/stomp/StompConnectionFactory.php‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,8 @@
88
use Interop\Queue\ConnectionFactory;
99
use Interop\Queue\Context;
1010
use Stomp\Network\Connection;
11+
use Stomp\Network\Observer\HeartbeatEmitter;
12+
use Stomp\Network\Observer\ServerAliveObserver;
1113

1214
class StompConnectionFactory implements ConnectionFactory
1315
{
@@ -88,11 +90,22 @@ private function establishConnection(): BufferedStompClient
8890
$scheme = (true === $config['ssl_on']) ? 'ssl' : 'tcp';
8991
$uri = $scheme.'://'.$config['host'].':'.$config['port'];
9092
$connection = new Connection($uri, $config['connection_timeout']);
93+
$connection->setWriteTimeout($config['write_timeout']);
94+
$connection->setReadTimeout($config['read_timeout']);
95+
96+
if ($config['send_heartbeat']) {
97+
$connection->getObservers()->addObserver(new HeartbeatEmitter($connection));
98+
}
99+
100+
if ($config['receive_heartbeat']) {
101+
$connection->getObservers()->addObserver(new ServerAliveObserver());
102+
}
91103

92104
$this->stomp = new BufferedStompClient($connection, $config['buffer_size']);
93105
$this->stomp->setLogin($config['login'], $config['password']);
94106
$this->stomp->setVhostname($config['vhost']);
95107
$this->stomp->setSync($config['sync']);
108+
$this->stomp->setHeartbeat($config['send_heartbeat'], $config['receive_heartbeat']);
96109

97110
$this->stomp->connect();
98111
}
@@ -128,6 +141,10 @@ private function parseDsn(string $dsn): array
128141
'sync' => $dsn->getBool('sync'),
129142
'lazy' => $dsn->getBool('lazy'),
130143
'ssl_on' => $dsn->getBool('ssl_on'),
144+
'write_timeout' => $dsn->getDecimal('write_timeout'),
145+
'read_timeout' => $dsn->getDecimal('read_timeout'),
146+
'send_heartbeat' => $dsn->getDecimal('send_heartbeat'),
147+
'receive_heartbeat' => $dsn->getDecimal('receive_heartbeat'),
131148
]), function ($value) { return null !== $value; });
132149
}
133150

@@ -145,6 +162,10 @@ private function defaultConfig(): array
145162
'sync' => false,
146163
'lazy' => true,
147164
'ssl_on' => false,
165+
'write_timeout' => 3,
166+
'read_timeout' => 60,
167+
'send_heartbeat' => 0,
168+
'receive_heartbeat' => 0,
148169
];
149170
}
150171
}

‎pkg/stomp/Tests/Functional/StompConnectionFactoryTest.php‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
<?php
2+
3+
namespace Enqueue\Stomp\Tests\Functional;
4+
5+
use Enqueue\Stomp\StompConnectionFactory;
6+
use Enqueue\Test\RabbitmqStompExtension;
7+
use Stomp\Network\Observer\Exception\HeartbeatException;
8+
use Stomp\Network\Observer\HeartbeatEmitter;
9+
use Stomp\Network\Observer\ServerAliveObserver;
10+
11+
/**
12+
* @group functional
13+
*/
14+
class StompConnectionFactoryTest extends \PHPUnit\Framework\TestCase
15+
{
16+
use RabbitmqStompExtension;
17+
18+
public function testShouldNotCreateConnectionWithSendHeartbeat()
19+
{
20+
$dsn = $this->getDsn().'?send_heartbeat=2000';
21+
$factory = new StompConnectionFactory($dsn);
22+
$this->expectException(HeartbeatException::class);
23+
$factory->createContext()->getStomp();
24+
}
25+
26+
public function testShouldCreateConnectionWithSendHeartbeat()
27+
{
28+
$dsn = $this->getDsn().'?send_heartbeat=2000&read_timeout=1';
29+
$factory = new StompConnectionFactory($dsn);
30+
$context = $factory->createContext();
31+
32+
$observers = $context->getStomp()->getConnection()->getObservers()->getObservers();
33+
$this->assertAttributeEquals([2000, 0], 'heartbeat', $context->getStomp());
34+
$this->assertCount(1, $observers);
35+
$this->assertInstanceOf(HeartbeatEmitter::class, $observers[0]);
36+
}
37+
38+
public function testShouldCreateConnectionWithReceiveHeartbeat()
39+
{
40+
$dsn = $this->getDsn().'?receive_heartbeat=2000';
41+
$factory = new StompConnectionFactory($dsn);
42+
$context = $factory->createContext();
43+
44+
$observers = $context->getStomp()->getConnection()->getObservers()->getObservers();
45+
$this->assertAttributeEquals([0, 2000], 'heartbeat', $context->getStomp());
46+
$this->assertCount(1, $observers);
47+
$this->assertInstanceOf(ServerAliveObserver::class, $observers[0]);
48+
}
49+
}

‎pkg/stomp/Tests/StompConnectionFactoryConfigTest.php‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,10 @@ public static function provideConfigs()
6666
'sync' => false,
6767
'lazy' => true,
6868
'ssl_on' => false,
69+
'write_timeout' => 3,
70+
'read_timeout' => 60,
71+
'send_heartbeat' => 0,
72+
'receive_heartbeat' => 0,
6973
],
7074
];
7175

@@ -83,6 +87,10 @@ public static function provideConfigs()
8387
'sync' => false,
8488
'lazy' => true,
8589
'ssl_on' => false,
90+
'write_timeout' => 3,
91+
'read_timeout' => 60,
92+
'send_heartbeat' => 0,
93+
'receive_heartbeat' => 0,
8694
],
8795
];
8896

@@ -100,6 +108,10 @@ public static function provideConfigs()
100108
'sync' => false,
101109
'lazy' => true,
102110
'ssl_on' => false,
111+
'write_timeout' => 3,
112+
'read_timeout' => 60,
113+
'send_heartbeat' => 0,
114+
'receive_heartbeat' => 0,
103115
],
104116
];
105117

@@ -118,6 +130,10 @@ public static function provideConfigs()
118130
'lazy' => false,
119131
'foo' => 'bar',
120132
'ssl_on' => false,
133+
'write_timeout' => 3,
134+
'read_timeout' => 60,
135+
'send_heartbeat' => 0,
136+
'receive_heartbeat' => 0,
121137
],
122138
];
123139

@@ -136,6 +152,10 @@ public static function provideConfigs()
136152
'lazy' => false,
137153
'foo' => 'bar',
138154
'ssl_on' => false,
155+
'write_timeout' => 3,
156+
'read_timeout' => 60,
157+
'send_heartbeat' => 0,
158+
'receive_heartbeat' => 0,
139159
],
140160
];
141161

@@ -154,6 +174,10 @@ public static function provideConfigs()
154174
'lazy' => false,
155175
'foo' => 'bar',
156176
'ssl_on' => false,
177+
'write_timeout' => 3,
178+
'read_timeout' => 60,
179+
'send_heartbeat' => 0,
180+
'receive_heartbeat' => 0,
157181
],
158182
];
159183

@@ -173,6 +197,10 @@ public static function provideConfigs()
173197
'foo' => 'bar',
174198
'ssl_on' => false,
175199
'baz' => 'bazVal',
200+
'write_timeout' => 3,
201+
'read_timeout' => 60,
202+
'send_heartbeat' => 0,
203+
'receive_heartbeat' => 0,
176204
],
177205
];
178206

@@ -190,6 +218,10 @@ public static function provideConfigs()
190218
'sync' => false,
191219
'lazy' => true,
192220
'ssl_on' => false,
221+
'write_timeout' => 3,
222+
'read_timeout' => 60,
223+
'send_heartbeat' => 0,
224+
'receive_heartbeat' => 0,
193225
],
194226
];
195227

@@ -208,6 +240,10 @@ public static function provideConfigs()
208240
'lazy' => true,
209241
'foo' => 'bar',
210242
'ssl_on' => false,
243+
'write_timeout' => 3,
244+
'read_timeout' => 60,
245+
'send_heartbeat' => 0,
246+
'receive_heartbeat' => 0,
211247
],
212248
];
213249
}

‎pkg/test/RabbitmqStompExtension.php‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,9 +8,14 @@
88

99
trait RabbitmqStompExtension
1010
{
11+
private function getDsn()
12+
{
13+
return getenv('RABITMQ_STOMP_DSN');
14+
}
15+
1116
private function buildStompContext(): StompContext
1217
{
13-
if (false == $dsn = getenv('RABITMQ_STOMP_DSN')) {
18+
if (false == $dsn = $this->getDsn()) {
1419
throw new SkippedTestError('Functional tests are not allowed in this environment');
1520
}
1621

0 commit comments

Comments
 (0)