Skip to content

Commit a512b53

Browse files
ASKozienkoASKozienko
committed
wamp
1 parent 99778d7 commit a512b53

9 files changed

+174
-34
lines changed

‎pkg/wamp/JsonSerializer.php

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Enqueue\Wamp;
6+
7+
class JsonSerializer implements Serializer
8+
{
9+
public function toString(WampMessage $message): string
10+
{
11+
$json = json_encode([
12+
'body' => $message->getBody(),
13+
'properties' => $message->getProperties(),
14+
'headers' => $message->getHeaders(),
15+
]);
16+
17+
if (JSON_ERROR_NONE !== json_last_error()) {
18+
throw new \InvalidArgumentException(sprintf(
19+
'The malformed json given. Error %s and message %s',
20+
json_last_error(),
21+
json_last_error_msg()
22+
));
23+
}
24+
25+
return $json;
26+
}
27+
28+
public function toMessage(string $string): WampMessage
29+
{
30+
$data = json_decode($string, true);
31+
if (JSON_ERROR_NONE !== json_last_error()) {
32+
throw new \InvalidArgumentException(sprintf(
33+
'The malformed json given. Error %s and message %s',
34+
json_last_error(),
35+
json_last_error_msg()
36+
));
37+
}
38+
39+
return new WampMessage($data['body'], $data['properties'], $data['headers']);
40+
}
41+
}

‎pkg/wamp/Serializer.php

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Enqueue\Wamp;
6+
7+
interface Serializer
8+
{
9+
public function toString(WampMessage $message): string;
10+
11+
public function toMessage(string $string): WampMessage;
12+
}

‎pkg/wamp/SerializerAwareTrait.php

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Enqueue\Wamp;
6+
7+
trait SerializerAwareTrait
8+
{
9+
/**
10+
* @var Serializer
11+
*/
12+
private $serializer;
13+
14+
/**
15+
* @param Serializer $serializer
16+
*/
17+
public function setSerializer(Serializer $serializer)
18+
{
19+
$this->serializer = $serializer;
20+
}
21+
22+
/**
23+
* @return Serializer
24+
*/
25+
public function getSerializer()
26+
{
27+
return $this->serializer;
28+
}
29+
}

‎pkg/wamp/Tests/Spec/JsonSerializerTest.php

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
<?php
2+
3+
namespace Enqueue\Wamp\Tests;
4+
5+
use Enqueue\Wamp\JsonSerializer;
6+
use Enqueue\Wamp\Serializer;
7+
use Enqueue\Test\ClassExtensionTrait;
8+
use Enqueue\Wamp\WampMessage;
9+
use PHPUnit\Framework\TestCase;
10+
11+
/**
12+
* @group Wamp
13+
*/
14+
class JsonSerializerTest extends TestCase
15+
{
16+
use ClassExtensionTrait;
17+
18+
public function testShouldImplementSerializerInterface()
19+
{
20+
$this->assertClassImplements(Serializer::class, JsonSerializer::class);
21+
}
22+
23+
public function testCouldBeConstructedWithoutAnyArguments()
24+
{
25+
new JsonSerializer();
26+
}
27+
28+
public function testShouldConvertMessageToJsonString()
29+
{
30+
$serializer = new JsonSerializer();
31+
32+
$message = new WampMessage('theBody', ['aProp' => 'aPropVal'], ['aHeader' => 'aHeaderVal']);
33+
34+
$json = $serializer->toString($message);
35+
36+
$this->assertSame('{"body":"theBody","properties":{"aProp":"aPropVal"},"headers":{"aHeader":"aHeaderVal"}}', $json);
37+
}
38+
39+
public function testThrowIfFailedToEncodeMessageToJson()
40+
{
41+
$serializer = new JsonSerializer();
42+
43+
$resource = fopen(__FILE__, 'r');
44+
45+
//guard
46+
$this->assertInternalType('resource', $resource);
47+
48+
$message = new WampMessage('theBody', ['aProp' => $resource]);
49+
50+
$this->expectException(\LogicException::class);
51+
$this->expectExceptionMessage('The malformed json given.');
52+
$serializer->toString($message);
53+
}
54+
55+
public function testShouldConvertJsonStringToMessage()
56+
{
57+
$serializer = new JsonSerializer();
58+
59+
$message = $serializer->toMessage('{"body":"theBody","properties":{"aProp":"aPropVal"},"headers":{"aHeader":"aHeaderVal"}}');
60+
61+
$this->assertInstanceOf(WampMessage::class, $message);
62+
63+
$this->assertSame('theBody', $message->getBody());
64+
$this->assertSame(['aProp' => 'aPropVal'], $message->getProperties());
65+
$this->assertSame(['aHeader' => 'aHeaderVal'], $message->getHeaders());
66+
}
67+
68+
public function testThrowIfFailedToDecodeJsonToMessage()
69+
{
70+
$serializer = new JsonSerializer();
71+
72+
$this->expectException(\LogicException::class);
73+
$this->expectExceptionMessage('The malformed json given.');
74+
$serializer->toMessage('{]');
75+
}
76+
}

‎pkg/wamp/WampConsumer.php

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
use Interop\Queue\Exception\InvalidMessageException;
99
use Interop\Queue\Message;
1010
use Interop\Queue\Queue;
11-
use React\EventLoop\Timer\Timer;
11+
use React\EventLoop\TimerInterface;
1212
use Thruway\ClientSession;
1313
use Thruway\Peer\Client;
1414

@@ -35,7 +35,7 @@ class WampConsumer implements Consumer
3535
private $message;
3636

3737
/**
38-
* @var Timer
38+
* @var TimerInterface
3939
*/
4040
private $timer;
4141

@@ -69,7 +69,7 @@ public function receive(int $timeout = 0): ?Message
6969
$this->client->on('open', function (ClientSession $session) {
7070

7171
$session->subscribe($this->queue->getQueueName(), function ($args) {
72-
$this->message = WampMessage::jsonUnserialize($args[0]);
72+
$this->message = $this->context->getSerializer()->toMessage($args[0]);
7373

7474
$this->client->emit('do-stop');
7575
});

‎pkg/wamp/WampContext.php

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
use Interop\Queue\Context;
99
use Interop\Queue\Destination;
1010
use Interop\Queue\Exception\InvalidDestinationException;
11+
use Interop\Queue\Exception\PurgeQueueNotSupportedException;
1112
use Interop\Queue\Exception\TemporaryQueueNotSupportedException;
1213
use Interop\Queue\Message;
1314
use Interop\Queue\Producer;
@@ -18,6 +19,8 @@
1819

1920
class WampContext implements Context
2021
{
22+
use SerializerAwareTrait;
23+
2124
/**
2225
* @var Client[]
2326
*/
@@ -41,6 +44,8 @@ public function __construct($client)
4144
Client::class
4245
));
4346
}
47+
48+
$this->setSerializer(new JsonSerializer());
4449
}
4550

4651
public function createMessage(string $body = '', array $properties = [], array $headers = []): Message
@@ -82,6 +87,7 @@ public function createSubscriptionConsumer(): SubscriptionConsumer
8287

8388
public function purgeQueue(Queue $queue): void
8489
{
90+
throw PurgeQueueNotSupportedException::providerDoestNotSupportIt();
8591
}
8692

8793
public function close(): void

‎pkg/wamp/WampMessage.php

Lines changed: 1 addition & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
use Interop\Queue\Impl\MessageTrait;
88
use Interop\Queue\Message;
99

10-
class WampMessage implements Message, \JsonSerializable
10+
class WampMessage implements Message
1111
{
1212
use MessageTrait;
1313

@@ -18,27 +18,4 @@ public function __construct(string $body = '', array $properties = [], array $he
1818
$this->headers = $headers;
1919
$this->redelivered = false;
2020
}
21-
22-
public function jsonSerialize(): array
23-
{
24-
return [
25-
'body' => $this->getBody(),
26-
'properties' => $this->getProperties(),
27-
'headers' => $this->getHeaders(),
28-
];
29-
}
30-
31-
public static function jsonUnserialize(string $json): self
32-
{
33-
$data = json_decode($json, true);
34-
if (JSON_ERROR_NONE !== json_last_error()) {
35-
throw new \InvalidArgumentException(sprintf(
36-
'The malformed json given. Error %s and message %s',
37-
json_last_error(),
38-
json_last_error_msg()
39-
));
40-
}
41-
42-
return new self($data['body'], $data['properties'], $data['headers']);
43-
}
4421
}

‎pkg/wamp/WampProducer.php

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -91,7 +91,9 @@ public function send(Destination $destination, Message $message): void
9191
$this->client->emit('do-stop');
9292
};
9393

94-
$this->session->publish($destination->getTopicName(), [json_encode($message->jsonSerialize())], [], ['acknowledge' => true])
94+
$payload = $this->context->getSerializer()->toString($message);
95+
96+
$this->session->publish($destination->getTopicName(), [$payload], [], ['acknowledge' => true])
9597
->then($onFinish, $onFinish);
9698
});
9799

‎pkg/wamp/WampSubscriptionConsumer.php

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66

77
use Interop\Queue\Consumer;
88
use Interop\Queue\SubscriptionConsumer;
9-
use React\EventLoop\Timer\Timer;
9+
use React\EventLoop\TimerInterface;
1010
use Thruway\ClientSession;
1111
use Thruway\Peer\Client;
1212

@@ -30,13 +30,10 @@ class WampSubscriptionConsumer implements SubscriptionConsumer
3030
private $client;
3131

3232
/**
33-
* @var Timer
33+
* @var TimerInterface
3434
*/
3535
private $timer;
3636

37-
/**
38-
* @param WampContext $context
39-
*/
4037
public function __construct(WampContext $context)
4138
{
4239
$this->context = $context;
@@ -65,7 +62,7 @@ public function consume(int $timeout = 0): void
6562

6663
foreach ($this->subscribers as $queue => $subscriber) {
6764
$session->subscribe($queue, function ($args) use ($subscriber, $session) {
68-
$message = WampMessage::jsonUnserialize($args[0]);
65+
$message = $this->context->getSerializer()->toMessage($args[0]);
6966

7067
/**
7168
* @var WampConsumer $consumer

0 commit comments

Comments
 (0)