AWS SQS/SNS | Parte 2 | Publicando uma mensagem via SNS
Tags: aws, microservices, python, sns
Published: 2020-06-14 01:34:25

Este texto faz parte de uma série, veja também os outros textos dessa série:

- [Implementando Mensageria Pub/Sub com Python, AWS SNS e SQS | Parte 1 ](https://dev.to/mayaramachado/implementando-mensageria-pub-sub-com-python-aws-sns-e-sqs-parte-1-5cpn)

- Publicando uma mensagem utilizando AWS SNS | Parte 2

- [AWS SQS/SNS | Parte 3 | Consumindo filas SQS usando Python](https://dev.to/mchdax/aws-sqs-sns-parte-3-consumindo-filas-sqs-usando-python-4omb)

- [AWS SQS/SNS | Parte 4 | Afinal, o que são as Dead Letter Queue?](https://dev.to/mchdax/aws-sqs-sns-parte-4-afinal-o-que-sao-as-dead-letter-queue-flb)
# Introdução


Dando continuidade na série sobre o uso da AWS SQS e SNS, hoje implementaremos a funcionalidade para publicação de mensagens utilizando SNS, e como utilizar o recurso de filtros dos serviços SNS. De modo mais introdutório, resolvi fazer duas aplicações REST utilizando Django e Django Rest Framework, que serão os serviços de ecommerce e payment_gateway que não possuem comunicação direta e realizam comunicação utilizando tópicos e filas. 

No geral essas aplicações representam operações e processamentos que podem ocorrer em serviços mais completos, os serviços são para simular API's de ecommerce. Quero ressaltar que este artigo não tem o propósito de explicar o funcionamento do Django Rest Framework, por esse motivo o código desenvolvido nesse projeto foi o mais simples possível e apenas funcional o suficiente para a utilização das funcionalidades de fila necessária. 

Pois bem, vamos começar.

# Implementação do projeto

Nesse projeto exemplo, imaginei um cenário onde temos um serviço responsável pelo processamento da API ecommerce, nesse serviço teremos a rota GET de listagem de produtos disponíveis e qual o vendedor daquele produto, também teremos uma rota POST para criação de um pedido de compra. Na outra ponta temos o serviço responsável pelo processamento do pagamento, o que eu chamei de payment gateway, esse serviço irá consumir a quantidade de pedidos processados pelo serviço ecommerce, e irá através de uma rota GET exibir a quantidade arrecadada de pedido para cada vendedor.

O código desenvolvido nesse projeto poderá ser encontrado [aqui](https://github.com/MayaraMachado/sns_and_sqs_project).

# Funcionalidades do serviço

Quando o serviço recebe a solicitação de pedido de compra, a aplicação realiza os processamentos que precisa para a sua execução. Uma vez que o processo foi completado, a aplicação irá disparar uma mensagem para o tópico SNS.

```python
def create(self, purchase_data):
        purchase = Purchase()
        
        products_pk = [product["id"] for product in purchase_data['products']]
        products_list = list(self.product_domain.get_all(query_params={'pk__in': products_pk}))
        purchase.total_price = self.__calculate_total_price(products_list)
        
        # salvar o objeto
        purchase = self.repository.create(purchase)
        
        # salvar os relacionamento ManyToMany
        purchase.product.add(*products_list)
        self.repository.update_m2m(purchase)
        
        # disparar mensagem para tópico SNS
        self.__notificate_transaction_sns(purchase)

        return purchase
```

A mensagem que a aplicação irá enviar vai ser um json contendo o identificador do vendedor, a data da compra, o valor total que aquela compra para aquele lojista e um identificador indicando a qual compra esses dados se referem. 

``` python
def __notificate_transaction_sns(self, purchase, seller_id):
        products = purchase.product.all()
        message = {
                    "seller" : str(seller_id),
                    "purchase_date" :  purchase.created_at.strftime("%Y-%m-%d %H:%M:%S %Z"),
                    "purchase_id" : str(purchase.pk),
                    "total_price" : str(purchase.total_price)
        }

        # Chama a classe de domínio que faz conexão com a AWS SNS
        self.sns_connection.publish_message_to_subscribers(settings.TRANSACTION_TOPIC_SNS, json.dumps(message))

```

# Envio da mensagem via SNS

Publicar uma mensagem em um tópico SNS é muito simples, para uma configuração mínima você precisa passar apenas o código ARN do tópico e a mensagem em questão. A mensagem pode estar em qualquer padrão definido para o projeto, nesse exemplo estou utilizando o envio de um json como mostrado anteriormente. 

Vale lembrar que o código ARN do tópico foi obtido durante a criação do mesmo, o código de criação do tópico e da instanciação do cliente SNS está disponível na Parte 1 dessa série.

```python
def publish_message_to_subscribers(self, topic_arn, message, message_attributes={}):
        publish = self.sns_client.publish(
                                            TopicArn=topic_arn,
                                            Message=message,
                                            MessageAttributes = message_attributes
                                        )
        return publish

```

Esse código é tudo o que você precisa para poder enviar uma mensagem para todos os incritos do tópico, lembrando também que é possível ter contas de email e números de telefone como inscritos de um tópico, desse modo todos eles receberiam igualmente a mensagem disparada sem haver a necessidade de adicionar um parâmetro a mais no envio. Como visto na parte 1 dessa série, apenas adicionei uma fila SQS como um inscrito desse tópico, logo ela foi quem recebeu a mensagem que foi disparada por esse trecho de código e agora nós iremos consumi-la.

# Filtrando o envio de mensagens

Mas lembram que eu falei anteriormente que um tópico pode ter n inscritos e que eles podem ser do tipo SQS, email ou SMS? Imagine que no nosso tópico ao invés de uma única Queue, a gente tivesse duas Queue inscrita e aptas para receberem mensagens, e vamos imaginar também que na verdade gostaríamos nesse cenário de mandar as mensagens para uma Queue em específica. Isso é possível através da utilização dos filtros SNS, mas como funciona?

Toda vez que é disparado o envio de mensagens através da SNS, por padrão o serviço já verifica se existe uma política de filtro existente para algum inscrito, caso tenha ela verifica se o filtro descrito nos atributos da mensagem bate com algum dos filtros encontrados naquele tópico. Nos casos onde não há filtro, a SNS dispara a mensagem para todos os seus inscritos sem ter nenhuma diferenciação. A configuração do filtro é atribuído no momento da inscrição, desse modo, voltando ao código publicado no texto anterior iremos pegar a váriavel  payment_subscription  e configurar essa inscrição.

```python
sns_cliente.set_subscription_attributes(
    SubscriptionArn = payment_subscription['SubscriptionArn'],
    AttributeName ='FilterPolicy', 
    AttributeValue ='{"event_type": ["order_placed"]}'
)
```

No código acima, a gente chama novamente o cliente sns instanciado utilizando o boto3 e chamamos a função para atualizar os atributos da inscrição. Adicionaremos um novo atributo para essa inscrição em específico, e a gente garante isso através do código Arn da inscrição da fila PaymentQueue, que é a fila que vai ser consumida pelo serviço PaymentGateway do meu exemplo. O atributo em questão é uma política de filtro, onde adicionamento um dicionário contendo a chave sinalizando que é um tipo de evento e o seu valor é um array contendo o nome dos filtros aplicados.

Vale ressaltar que esse array pode conter ter 1 ou mais especificações de filtro, e uma mensagem será enviada para esta inscrição caso o atributo da mensagem seja correspondente com ao menos 1 filtro atribuído. A comparação realizada pela AWS é booleana, ou seja, ou é exatamente igual ou não, dessa forma os filtros são case sensitve.

Para enviar uma mensagem com um filtro em específico, nós iremos reescrever a função de notificate_transaction_sns para atribuirmos a especificação do filtro nos atributos da mensagem.

```python
def __notificate_transaction_sns(self, purchase, seller_id):
        products = purchase.product.all()
        message = {
                    "seller" : str(seller_id),
                    "purchase_date" :  purchase.created_at.strftime("%Y-%m-%d %H:%M:%S %Z"),
                    "purchase_id" : str(purchase.pk),
                    "total_price" : str(purchase.total_price)
        }

        message_attributes = {
            'event_type': {
                'DataType': 'String',
                'StringValue': 'order_placed'
            }
        }

        self.sns_connection.publish_message_to_subscribers(settings.TRANSACTION_TOPIC_SNS, json.dumps(message), message_attributes)

```

E voilá, a mensagem será enviada apenas para o inscrição que possua o filtro order_placed atribuída.

# Conclusão e implementações futuras

Na parte 2 dessa série nós vemos como publicar mensagens através do serviço SNS, vimos como pode ser absurdamente prático e vimos também como utilizar os filtros ao nosso favor. Nos próximos textos da série Implementando AWS SQS e SNS iremos ver como consumir as mensagens que enviamos hoje, e vamos descobrir também sobre o conceito das Dead Letter Queue. Vejo você lá!