brightecapital/sqs

AWS SQS Transport (fork and improve from enqueue/sqs)

1.0.3 2019-10-10 05:39 UTC

Requires

• php: ^7.1.3
• aws/aws-sdk-php: ~3.26
• enqueue/dsn: ^0.9
• queue-interop/queue-interop: ^0.7|^0.8

Requires (Dev)

• enqueue/test: 0.9.x-dev
• phpunit/phpunit: ~5.4.0
• queue-interop/queue-spec: ^0.6

MIT

• Ken Ngo

• queue
• messaging
• aws
• sqs

README

Description

This project was forked from enqueue/sqs and made the following improvements:

• Move all classes to src
• Rename Tests to tests
• Move examples to tests
• Change namespace to Brighte\Sqs

SqsProducer->send():

public function send(Destination $destination, Message $message): void
    {
        InvalidDestinationException::assertDestinationInstanceOf($destination, SqsDestination::class);
        InvalidMessageException::assertMessageInstanceOf($message, SqsMessage::class);

        $body = $message->getBody();
        if (empty($body)) {
            throw new InvalidMessageException('The message body must be a non-empty string.');
        }

        $arguments = [
            '@region' => $destination->getRegion(),
            'MessageBody' => $body,
            'QueueUrl' => $this->context->getQueueUrl($destination),
        ];

        if (null !== $this->deliveryDelay) {
            $arguments['DelaySeconds'] = (int) $this->deliveryDelay / 1000;
        }

        if ($message->getDelaySeconds()) {
            $arguments['DelaySeconds'] = $message->getDelaySeconds();
        }

        if ($message->getMessageDeduplicationId()) {
            $arguments['MessageDeduplicationId'] = $message->getMessageDeduplicationId();
        }

        if ($message->getMessageGroupId()) {
            $arguments['MessageGroupId'] = $message->getMessageGroupId();
        }

        if ($message->getHeaders()) {
            $arguments['MessageAttributes']['Headers'] = [
                'DataType' => 'String',
                'StringValue' => json_encode([$message->getHeaders()]),
            ];
        }
        
        if ($message->getProperties()) {
            foreach ($message->getProperties() as $name => $value) {
                $arguments['MessageAttributes'][$name] = ['DataType' => 'String', 'StringValue' => $value];
            }
        }

        $result = $this->context->getSqsClient()->sendMessage($arguments);

        if (false == $result->hasKey('MessageId')) {
            throw new \RuntimeException('Message was not sent');
        }

        $message->setMessageId($result['MessageId']);
    }

SqsConsumer->covertMessage():

protected function convertMessage(array $sqsMessage): SqsMessage
    {
        $message = $this->context->createMessage();

        $message->setBody($sqsMessage['Body']);
        $message->setReceiptHandle($sqsMessage['ReceiptHandle']);

        if (isset($sqsMessage['Attributes'])) {
            $message->setAttributes($sqsMessage['Attributes']);

            if (isset($sqsMessage['Attributes']['MessageDeduplicationId'])) {
                $message->setMessageDeduplicationId($sqsMessage['Attributes']['MessageDeduplicationId']);
            }

            if (isset($sqsMessage['Attributes']['MessageGroupId'])) {
                $message->setMessageGroupId($sqsMessage['Attributes']['MessageGroupId']);
            }
        }

        if (isset($sqsMessage['Attributes']['ApproximateReceiveCount'])) {
            $message->setRedelivered(((int) $sqsMessage['Attributes']['ApproximateReceiveCount']) > 1);
        }

        if (isset($sqsMessage['MessageAttributes'])) {
            foreach ($sqsMessage['MessageAttributes'] as $name => $attribute) {
                if ($name == 'Headers') {
                    $headers = json_decode($attribute['StringValue'], true);
                    $message->setHeaders($headers);
                } else {
                    $message->setProperty($name, $attribute['StringValue']);
                }
            }
        }

        if (isset($sqsMessage['MessageId'])) {
            $message->setMessageId($sqsMessage['MessageId']);
        }

        return $message;
    }