Skip to content

Commit 7830b80

Browse files
ASKozienkoASKozienko
committed
wamp
1 parent d86e9cd commit 7830b80

14 files changed

+931
-1
lines changed

‎composer.json

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,9 @@
2929
"stomp-php/stomp-php": "^4",
3030
"php-http/guzzle6-adapter": "^1.1",
3131
"php-http/client-common": "^1.7@dev",
32-
"richardfullmer/rabbitmq-management-api": "^2.0"
32+
"richardfullmer/rabbitmq-management-api": "^2.0",
33+
"thruway/pawl-transport": "^0.5.0",
34+
"voryx/thruway": "^0.5.3"
3335
},
3436
"require-dev": {
3537
"phpunit/phpunit": "^5.5",
@@ -75,6 +77,7 @@
7577
"Enqueue\\Stomp\\": "pkg/stomp/",
7678
"Enqueue\\Test\\": "pkg/test/",
7779
"Enqueue\\Dsn\\": "pkg/dsn/",
80+
"Enqueue\\Wamp\\": "pkg/wamp/",
7881
"Enqueue\\": "pkg/enqueue/"
7982
},
8083
"exclude-from-classmap": [

‎pkg/wamp/.gitignore

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
*~
2+
/composer.lock
3+
/composer.phar
4+
/phpunit.xml
5+
/vendor/
6+
/.idea/

‎pkg/wamp/.travis.yml

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
sudo: false
2+
3+
git:
4+
depth: 10
5+
6+
language: php
7+
8+
php:
9+
- '7.1'
10+
- '7.2'
11+
12+
cache:
13+
directories:
14+
- $HOME/.composer/cache
15+
16+
install:
17+
- composer self-update
18+
- composer install
19+
20+
script:
21+
- vendor/bin/phpunit --exclude-group=functional

‎pkg/wamp/LICENSE

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
The MIT License (MIT)
2+
Copyright (c) 2017 Kotliar Maksym
3+
4+
Permission is hereby granted, free of charge, to any person obtaining a copy
5+
of this software and associated documentation files (the "Software"), to deal
6+
in the Software without restriction, including without limitation the rights
7+
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
8+
copies of the Software, and to permit persons to whom the Software is furnished
9+
to do so, subject to the following conditions:
10+
11+
The above copyright notice and this permission notice shall be included in all
12+
copies or substantial portions of the Software.
13+
14+
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
15+
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
16+
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
17+
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
18+
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
19+
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
20+
THE SOFTWARE.

‎pkg/wamp/README.md

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
# Web Application Messaging Protocol Transport
2+
3+
[![Gitter](https://badges.gitter.im/php-enqueue/Lobby.svg)](https://gitter.im/php-enqueue/Lobby)
4+
[![Build Status](https://travis-ci.org/php-enqueue/wamp.png?branch=master)](https://travis-ci.org/php-enqueue/wamp)
5+
[![Total Downloads](https://poser.pugx.org/enqueue/wamp/d/total.png)](https://packagist.org/packages/enqueue/wamp)
6+
[![Latest Stable Version](https://poser.pugx.org/enqueue/wamp/version.png)](https://packagist.org/packages/enqueue/wamp)
7+
8+
This is an implementation of [queue interop](https://github.com/queue-interop/queue-interop). It uses [Thruway](https://github.com/voryx/Thruway) internally.
9+
10+
## Resources
11+
12+
* [Site](https://enqueue.forma-pro.com/)
13+
* [Documentation](https://github.com/php-enqueue/enqueue-dev/blob/master/docs/index.md)
14+
* [Questions](https://gitter.im/php-enqueue/Lobby)
15+
* [Issue Tracker](https://github.com/php-enqueue/enqueue-dev/issues)
16+
17+
## Developed by Forma-Pro
18+
19+
Forma-Pro is a full stack development company which interests also spread to open source development.
20+
Being a team of strong professionals we have an aim an ability to help community by developing cutting edge solutions in the areas of e-commerce, docker & microservice oriented architecture where we have accumulated a huge many-years experience.
21+
Our main specialization is Symfony framework based solution, but we are always looking to the technologies that allow us to do our job the best way. We are committed to creating solutions that revolutionize the way how things are developed in aspects of architecture & scalability.
22+
23+
If you have any questions and inquires about our open source development, this product particularly or any other matter feel free to contact at opensource@forma-pro.com
24+
25+
## License
26+
27+
It is released under the [MIT License](LICENSE).

‎pkg/wamp/WampConnectionFactory.php

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,112 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Enqueue\Wamp;
6+
7+
use Enqueue\Dsn\Dsn;
8+
use Interop\Queue\ConnectionFactory;
9+
use Interop\Queue\Context;
10+
use Thruway\Peer\Client;
11+
use Thruway\Transport\PawlTransportProvider;
12+
13+
class WampConnectionFactory implements ConnectionFactory
14+
{
15+
/**
16+
* @var array
17+
*/
18+
private $config;
19+
20+
/**
21+
* The config could be an array, string DSN or null. In case of null it will attempt to connect to Ratchet localhost.
22+
*
23+
* $config = [
24+
* 'lazy' => true,
25+
* 'dsn' => 'wamp://127.0.0.1:9090',
26+
* 'host' => '127.0.0.1',
27+
* 'port' => '9090',
28+
* 'max_retries' => 15,
29+
* 'initial_retry_delay' => 1.5,
30+
* 'max_retry_delay' => 300,
31+
* 'retry_delay_growth' => 1.5,
32+
* ]
33+
*
34+
* or
35+
*
36+
* wamp://127.0.0.1:9090?max_retries=10
37+
*
38+
* @param array|string|null $config
39+
*/
40+
public function __construct($config = 'wamp:')
41+
{
42+
if (empty($config)) {
43+
$config = $this->parseDsn('wamp:');
44+
} elseif (is_string($config)) {
45+
$config = $this->parseDsn($config);
46+
} elseif (is_array($config)) {
47+
$config = empty($config['dsn']) ? $config : $this->parseDsn($config['dsn']);
48+
} else {
49+
throw new \LogicException('The config must be either an array of options, a DSN string or null');
50+
}
51+
52+
$config = array_replace([
53+
'lazy' => true,
54+
'host' => '127.0.0.1',
55+
'port' => '9090',
56+
'max_retries' => 15,
57+
'initial_retry_delay' => 1.5,
58+
'max_retry_delay' => 300,
59+
'retry_delay_growth' => 1.5,
60+
], $config);
61+
62+
$this->config = $config;
63+
}
64+
65+
public function createContext(): Context
66+
{
67+
if ($this->config['lazy']) {
68+
return new WampContext(function () {
69+
return $this->establishConnection();
70+
});
71+
}
72+
73+
return new WampContext($this->establishConnection());
74+
}
75+
76+
private function establishConnection(): Client
77+
{
78+
$uri = sprintf('ws://%s:%s', $this->config['host'], $this->config['port']);
79+
80+
$client = new Client('realm1');
81+
$client->addTransportProvider(new PawlTransportProvider($uri));
82+
$client->setReconnectOptions([
83+
'max_retries' => $this->config['max_retries'],
84+
'initial_retry_delay' => $this->config['initial_retry_delay'],
85+
'max_retry_delay' => $this->config['max_retry_delay'],
86+
'retry_delay_growth' => $this->config['retry_delay_growth'],
87+
]);
88+
89+
return $client;
90+
}
91+
92+
private function parseDsn(string $dsn): array
93+
{
94+
$dsn = new Dsn($dsn);
95+
96+
if ('wamp' !== $dsn->getSchemeProtocol()) {
97+
throw new \LogicException(sprintf(
98+
'The given scheme protocol "%s" is not supported. It must be "wamp"',
99+
$dsn->getSchemeProtocol()
100+
));
101+
}
102+
103+
return array_filter(array_replace($dsn->getQuery(), [
104+
'host' => $dsn->getHost(),
105+
'port' => $dsn->getPort(),
106+
'max_retries' => $dsn->getInt('max_retries'),
107+
'initial_retry_delay' => $dsn->getFloat('initial_retry_delay'),
108+
'max_retry_delay' => $dsn->getInt('max_retry_delay'),
109+
'retry_delay_growth' => $dsn->getFloat('retry_delay_growth'),
110+
]), function ($value) { return null !== $value; });
111+
}
112+
}

‎pkg/wamp/WampConsumer.php

Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,127 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Enqueue\Wamp;
6+
7+
use Interop\Queue\Consumer;
8+
use Interop\Queue\Exception\InvalidMessageException;
9+
use Interop\Queue\Message;
10+
use Interop\Queue\Queue;
11+
use React\EventLoop\Timer\Timer;
12+
use Thruway\ClientSession;
13+
use Thruway\Peer\Client;
14+
15+
class WampConsumer implements Consumer
16+
{
17+
/**
18+
* @var WampContext
19+
*/
20+
private $context;
21+
22+
/**
23+
* @var WampDestination
24+
*/
25+
private $queue;
26+
27+
/**
28+
* @var Client
29+
*/
30+
private $client;
31+
32+
/**
33+
* @var WampMessage
34+
*/
35+
private $message;
36+
37+
/**
38+
* @var Timer
39+
*/
40+
private $timer;
41+
42+
public function __construct(WampContext $context, WampDestination $destination)
43+
{
44+
$this->context = $context;
45+
$this->queue = $destination;
46+
}
47+
48+
public function getQueue(): Queue
49+
{
50+
return $this->queue;
51+
}
52+
53+
public function receive(int $timeout = 0): ?Message
54+
{
55+
$init = false;
56+
$this->timer = null;
57+
$this->message = null;
58+
59+
if (null === $this->client) {
60+
$init = true;
61+
62+
$this->client = $this->context->getClient();
63+
$this->client->setAttemptRetry(true);
64+
$this->client->on('open', function (ClientSession $session) {
65+
66+
$session->subscribe($this->queue->getQueueName(), function ($args) {
67+
$this->message = WampMessage::jsonUnserialize($args[0]);
68+
69+
$this->client->emit('do-stop');
70+
});
71+
});
72+
73+
$this->client->on('do-stop', function () {
74+
if ($this->timer) {
75+
$this->client->getLoop()->cancelTimer($this->timer);
76+
}
77+
78+
$this->client->getLoop()->stop();
79+
});
80+
}
81+
82+
if ($timeout > 0) {
83+
$this->timer = $this->client->getLoop()->addTimer($timeout / 1000, function () {
84+
$this->client->emit('do-stop');
85+
});
86+
}
87+
88+
if ($init) {
89+
$this->client->start(false);
90+
}
91+
92+
$this->client->getLoop()->run();
93+
94+
return $this->message ?: null;
95+
}
96+
97+
public function receiveNoWait(): ?Message
98+
{
99+
return $this->receive(100);
100+
}
101+
102+
/**
103+
* {@inheritdoc}
104+
*
105+
* @param WampMessage $message
106+
*/
107+
public function acknowledge(Message $message): void
108+
{
109+
// do nothing. redis transport always works in auto ack mode
110+
}
111+
112+
/**
113+
* {@inheritdoc}
114+
*
115+
* @param WampMessage $message
116+
*/
117+
public function reject(Message $message, bool $requeue = false): void
118+
{
119+
InvalidMessageException::assertMessageInstanceOf($message, WampMessage::class);
120+
121+
// do nothing on reject. redis transport always works in auto ack mode
122+
123+
if ($requeue) {
124+
$this->context->createProducer()->send($this->queue, $message);
125+
}
126+
}
127+
}

0 commit comments

Comments
 (0)