Skip to content
Permalink
Browse files

[Messenger] set amqp content_type based on serialization format

  • Loading branch information...
Tobion committed Jun 2, 2019
1 parent d8224b8 commit d3965e4de8cf2a4abbf8ebc221446fc396a71ac1
@@ -69,6 +69,39 @@ public function testItSendsTheEncodedMessageWithoutHeaders()
$sender->send($envelope);
}
public function testContentTypeHeaderIsMovedToAttribute()
{
$envelope = new Envelope(new DummyMessage('Oy'));
$encoded = ['body' => '...', 'headers' => ['type' => DummyMessage::class, 'Content-Type' => 'application/json']];
$serializer = $this->getMockBuilder(SerializerInterface::class)->getMock();
$serializer->method('encode')->with($envelope)->willReturnOnConsecutiveCalls($encoded);
$connection = $this->getMockBuilder(Connection::class)->disableOriginalConstructor()->getMock();
unset($encoded['headers']['Content-Type']);
$stamp = new AmqpStamp(null, AMQP_NOPARAM, ['content_type' => 'application/json']);
$connection->expects($this->once())->method('publish')->with($encoded['body'], $encoded['headers'], 0, $stamp);
$sender = new AmqpSender($connection, $serializer);
$sender->send($envelope);
}
public function testContentTypeHeaderDoesNotOverwriteAttribute()
{
$envelope = (new Envelope(new DummyMessage('Oy')))->with($stamp = new AmqpStamp('rk', AMQP_NOPARAM, ['content_type' => 'custom']));
$encoded = ['body' => '...', 'headers' => ['type' => DummyMessage::class, 'Content-Type' => 'application/json']];
$serializer = $this->getMockBuilder(SerializerInterface::class)->getMock();
$serializer->method('encode')->with($envelope)->willReturnOnConsecutiveCalls($encoded);
$connection = $this->getMockBuilder(Connection::class)->disableOriginalConstructor()->getMock();
unset($encoded['headers']['Content-Type']);
$connection->expects($this->once())->method('publish')->with($encoded['body'], $encoded['headers'], 0, $stamp);
$sender = new AmqpSender($connection, $serializer);
$sender->send($envelope);
}
/**
* @expectedException \Symfony\Component\Messenger\Exception\TransportException
*/
@@ -55,7 +55,8 @@ public function testEncodedIsHavingTheBodyAndTypeHeader()
$this->assertArrayHasKey('body', $encoded);
$this->assertArrayHasKey('headers', $encoded);
$this->assertArrayHasKey('type', $encoded['headers']);
$this->assertEquals(DummyMessage::class, $encoded['headers']['type']);
$this->assertSame(DummyMessage::class, $encoded['headers']['type']);
$this->assertSame('application/json', $encoded['headers']['Content-Type']);
}
public function testUsesTheCustomFormatAndContext()
@@ -50,12 +50,26 @@ public function send(Envelope $envelope): Envelope
$delay = $delayStamp->getDelay();
}
$amqpStamp = $envelope->last(AmqpStamp::class);
if (isset($encodedMessage['headers']['Content-Type'])) {
$contentType = $encodedMessage['headers']['Content-Type'];
unset($encodedMessage['headers']['Content-Type']);
$attributes = $amqpStamp ? $amqpStamp->getAttributes() : [];
if (!isset($attributes['content_type'])) {
$attributes['content_type'] = $contentType;
$amqpStamp = new AmqpStamp($amqpStamp ? $amqpStamp->getRoutingKey() : null, $amqpStamp ? $amqpStamp->getFlags() : AMQP_NOPARAM, $attributes);
}
}
try {
$this->connection->publish(
$encodedMessage['body'],
$encodedMessage['headers'] ?? [],
$delay,
$envelope->last(AmqpStamp::class)
$amqpStamp
);
} catch (\AMQPException $e) {
throw new TransportException($e->getMessage(), 0, $e);
@@ -101,7 +101,7 @@ public function encode(Envelope $envelope): array
$envelope = $envelope->withoutStampsOfType(NonSendableStampInterface::class);
$headers = ['type' => \get_class($envelope->getMessage())] + $this->encodeStamps($envelope);
$headers = ['type' => \get_class($envelope->getMessage())] + $this->encodeStamps($envelope) + $this->getContentTypeHeader();
return [
'body' => $this->serializer->serialize($envelope->getMessage(), $this->format, $context),
@@ -157,4 +157,29 @@ private function findFirstSerializerStamp(array $stamps): ?SerializerStamp
return null;
}
private function getContentTypeHeader(): array
{
$mimeType = $this->getMimeTypeForFormat();
return null === $mimeType ? [] : ['Content-Type' => $mimeType];
}
private function getMimeTypeForFormat(): ?string
{
switch ($this->format) {
case 'json':
return 'application/json';
case 'xml':
return 'application/xml';
case 'yml':
case 'yaml':
return 'application/x-yaml';
case 'csv':
return 'text/csv';
}
return null;
}
}

0 comments on commit d3965e4

Please sign in to comment.
You can’t perform that action at this time.