Abstrair AWS SQS no Ruby

Como prometido na postagem anterior, vou mostrar como abstrair AWS SQS no Ruby. Este post cobre a criação total de uma classe chamada Queue Manager, incluindo testes automatizados com o RSpec.

Vantagens ao abstrair AWS SQS?

Toda a complexibilidade do tratamento das filas e suas mensagens ficam em um só ponto. Isso é muito bom, pois facilita a manutenção do código referente à filas, isso inclui correção de bugs e até mesmo a alteração do serviço de fila atual para um outro.

Imagine se você quiser trocar o AWS SQS por RabbitMQ. Não seria muito mais fácil alterar apenas uma classe do seu projeto e ter todo ele atualizado?

Uma outra vantagem é que você vai evitar a repetição de código em várias partes do seu projeto.

Lembrando que estamos utilizando o Ruby 2.2.7 e aws-sdk 2.9.11.

Preparando o ambiente

Crie o diretório do novo projeto e após o acesse:

mkdir queue_manager_aws_sdk_v2_sqs_ruby
cd queue_manager_aws_sdk_v2_sqs_ruby/

Crie o arquivo Gemfile:

touch Gemfile

Após abra o arquivo que acabou de criar, com seu editor de texto preferido, cole as seguintes linhas e salve:

source "https://rubygems.org"

gem 'aws-sdk', '2.9.11'

group :test, :development do
  gem 'rspec'
  gem 'byebug'
end

Instale as Gems que acabou de adicionar no Gemfile:

bundle install

Crie a base dos testes automatizados:

rspec --init

Após altere o arquivo spec_helper.rb, que se encontra dentro do diretório lib para utilizar a configuração de stub do AWS SDK. Confira como fazer: Habilitando Stub Aws SQS

Crie o diretório onde vai ficar a classe que irá abstrair AWS SQS:

mkdir lib

Dentro do diretório lib, crie o arquivo que conterá o código da nossa nova classe:

touch ./lib/queue_manager.rb

Crie também o arquivo que conterá os testes:

touch ./spec/queue_manager_spec.rb

Crie a base dos testes no arquivo queue_manager_spec.rb:

require 'spec_helper.rb'

describe "QueueManager" do
  
end

Agora faça o require do arquivo da nossa nova classe no arquivo spec_helper.rb para que o RSpec a carregue ao inicializar os testes. Para isso adicione o seguinte código, abaixo do require ‘aws-sdk’:

require 'queue_manager.rb'

Pronto! Agora podemos prosseguir com a criação da classe Queue Manager.

Criação da classe que vai abstrair AWS SQS

Vamos pensar um pouco… Qual é o mínimo que a nossa classe deve prover? Não seria o envio, recebimento e remoção de mensagens em uma determinada fila? Abra o arquivo lib/queue_manager.rb no seu editor preferido e mãos à obra!

Iremos começar pela base da classe que irá abstrair AWS SQS:

class QueueManager
  def initialize(queue_name)
    @queue_name = queue_name
  end

  private
  def client
    @client ||= Aws::SQS::Client.new
  end

  def queue_name
    @queue_name
  end

  def queue_url
    @queue_url ||= get_queue_url
  end

  def get_queue_url
    client.get_queue_url({queue_name: queue_name}).queue_url
  end
end

Basicamente recebemos o nome de uma fila ao instânciar a classe e guardamos esta informação em uma variável global dentro da classe:

  def initialize(queue_name)
    @queue_name = queue_name
  end

Criamos também um método privado que retorna o nome da fila. Poderíamos ficar utilizando a variável global sempre, mas é mais elegante utilizarmos um método para isso.

  def queue_name
    @queue_name
  end

O método privado client retorna uma instância da classe Aws::SQS::Client e guarda esta instância em uma variável global @client. Este método irá instânciar o objeto apenas se a variável @client for nula, ou seja, teoricamente apenas na primeira vez que o método for chamado.

  def client
    @client ||= Aws::SQS::Client.new
  end

O método queue_url é responsável por retornar a URL da fila. Em praticamente todos os métodos do Aws::SQS::Client, a URL da fila é um parâmetro obrigatório. Veja aqui mais detalhes sobre a utilização do Aws::SQS.

  def queue_url
    @queue_url ||= get_queue_url
  end

Podemos ver que este método chama um outro método (get_queue_url) caso a variável global @queue_url tenha seu valor nulo.

Por sua vez, o método get_queue_url, tenta recuperar a URL da fila, através da instância Aws::SQS::Client utilizando método get_queue_url, passando o nome da fila (queue_name) por parâmetro:

  def get_queue_url
    client.get_queue_url({queue_name: queue_name}).queue_url
  end

Legal! Mas o que acontece se a fila não existir? Estoura o seguinte erro: Aws::SQS::Errors::NonExistentQueue.

Para ficar mais simples a utilização do serviço de filas, que tal a gente criar a fila e retornar sua URL, caso retorne o erro dito acima? Vamos lá!

Fallback em caso de erro: Aws::SQS::Errors::NonExistentQueue

Vamos criar o método privado que será responsável por criar uma fila e retornar sua URL:

  def create_queue
    client.create_queue({queue_name: queue_name},{:http_open_timeout => 1.hours, :http_read_timeout => 1.hours, :logger => nil, :visibility_timeout => 1.minutes}).queue_url
  end

Neste método, utilizamos o método create_queue da classe Aws::SQS::Client. Note que passei o nome da fila (queue_name) e algum parâmetros de configuração. Confira mais detalhes deste método na documentação oficianal do SDK AWS v2.

Agora vamos fazer com que o método get_queue_url chame o método create_queue caso estoure o erro de fila inexistente:

  def get_queue_url
    begin
      client.get_queue_url({queue_name: queue_name}).queue_url
    rescue Aws::SQS::Errors::NonExistentQueue
      create_queue
    end
  end

Pronto! Agora podemos seguir com os métodos responsáveis por prover o mínimo de funcionalidades da Queue Manager.

Abstrair Aws SQS Client send_message

  def send_message(message_data)
    client.send_message({queue_url: queue_url, message_body: data })
  end

Simplemente recebemos o conteúdo da mensagem e enviamos para a fila. Lembra do método queue_url? Olha ele alí sendo passando por parâmetro.

Abstrair Aws SQS Client delete_message

  def delete_message(message_handle)
    client.delete_message({queue_url: queue_url, receipt_handle: message_handle})
  end

Recebemos o parâmetro message_handle e enviamos como o parâmetro receipt_handle para a fila. Toda mensagem que é recuperada, apresenta o campo receipt_handle nela.

Abstrair Aws SQS Client receive_message

Para abstrair o método receive_message da classe Aws::SQS::Client, devemos realizar uma alteração o método initialize da nossa classe Queue Manager:

  def initialize(queue_name, visibility_timeout = 1.minutes)
    @queue_name = queue_name
    @visibility_timeout = visibility_timeout
  end

A partir de agora recebemos o parâmetro visibility_timeout e guardamos na variável global @visibility_timeout. Esta variável será enviada todas as vezes que a gente utilizar o método de recebimento de mensagens da nossa classe. Como o nome da variável já diz, ela guarda o tempo que uma mensagem deverá ficar in flight após ser recuperada. Este tempo é em segundos. Por padrão definimos 60 segundos (1.minutes).

Vamos criar o seguinte método privado para que possamos utilizar a variável global @visibility_timeout de forma elegante:

  def visibility_timeout
    @visibility_timeout
  end

Agora vamos à criação do método receive_message:

  def receive_message
    client.receive_message({queue_url: queue_url, visibility_timeout: visibility_timeout})
  end

Enviamos a URL da fila e o tempo que a mensagem ficará in flight. Veja que utilizamos o método visibility_timeout para enviar o parâmetro de mesmo nome, para a fila.

Agora que já fizemos as funcionalidades básicas, que tal permitir o envio, recebimento e remoção de mensagens em lote? Em alguns casos vale a pena realizar operações em batch, para melhorar a performance.

Abstrair AWS SQS Client send_message_batch

  def send_message_batch(batch_data = [])
    client.send_message_batch({queue_url: queue_url, entries: batch_data})
  end

O método é bem simples, porém devemos tomar cuidado com a estrutura que o Aws::SQS:Client.send_message_batch espera.

Cada mensagem do parâmetro entries deve conter a seguinte estrutura mínima:

{id: "String", message_body: "String"}

Segue um exemplo:

[
  {id: "something", message_body: "String"},
  {id: "anything", message_body: "anything"},
  {id: "nothing", message_body: "nothing"}
]

As IDs serão retornadas na mensagem, para a conferência de cada mensagem. Ou seja, saberemos se uma determinada mensagem do batch foi processada com sucesso ou não.

Poderíamos criar a id de cada mensagem de acordo com o message_body de cada uma delas, e verificar o resultado para retornar algum tipo de mensagem para o usuário da aplicação. Este tipo de transformação e validação soa como uma regra de negócio, logo não deve ser tratada na classe Queue Manager e sim na classe que for estendê-la ou for utilizá-la.

Abstrair AWS SQS Client receive_message_batch

  def receive_message_batch(batch_size = 10)
    client.receive_message({queue_url: queue_url, max_number_of_messages: batch_size, visibility_timeout: visibility_timeout})
  end

Utilizamos o método receive_message da classe Aws::SQS::Client, passando como parâmetro max_number_of_messages que é responsável por dizer a fila quantas mensagens no máximo ela poderá retornar. Por padrão setamos batch_size com 10 (valor máximo aceito pelo AWS SQS), porém este valor pode ser menor que 10.

Também passamos o parâmetro visibility_timeout, que irá receber o valor do método visibility_timeout.

Abstrair AWS SQS Client delete_message_batch

  def delete_message_batch(batch_data = [])
    client.delete_message_batch({queue_url: queue_url, entries: batch_data})
  end

O parâmetro entries é um array que deve conter itens no seguinte formato:

{id: "something", receipt_handle: "something"}

Estes dados vêm em cada mensagem recebida através do receive_message.

Agora que já conseguimos manipular as mensagens de todas as formas possíveis, que tal a gente criar mais dois métodos para manipular a fila? Vamos nessa! :D

Abstrair AWS SQS Client purge_queue

  def purge_queue
    client.purge_queue({queue_url: queue_url})
  end

Com este método, removemos todas as mensagens da fila, sem excluir a fila.

Abstrair AWS SQS Client delete_queue

  def delete_queue
    client.delete_queue({queue_url: queue_url})
    @queue_url = nil
  end

Com este método, a  gente exclui a fila, independente dela estar vazia ou não. Após a gente seta a variável @queue_url para nulo, assim evitando erros caso logo depois, na mesma instância a gente tente manipular a fila. Pois com a variável @queue_url nula, novamente a gente tenta pegar a URL da fila e cao não exista a gente cria uma automaticamente. Lembra do método queue_url? :)

Que tal a gente saber o tamanho da fila? Seria uma boa, não?

Abstrair AWS SQS Client get_queue_attributes

O método get_queue_attributes, da classe Aws::SQS::Client serve para pegar alguns atributos da fila. Aqui vamos utilizá-lo para saber quantas mensagens temos na fila:

  def queue_size
    response = client.get_queue_attributes({queue_url: queue_url, attribute_names: ["ApproximateNumberOfMessages", "ApproximateNumberOfMessagesNotVisible", "ApproximateNumberOfMessagesDelayed"]})
    response.attributes["ApproximateNumberOfMessages"].to_i + response.attributes["ApproximateNumberOfMessagesNotVisible"].to_i + response.attributes["ApproximateNumberOfMessagesDelayed"].to_i
  end

Este método espera que a gente envie os nomes dos atributos a serem retornados. Você pode ver mais detalhes sobre os atributos da fila na documentação oficial da AWS.

Vamos recuperar os seguintes atributos:

• ApproximateNumberOfMessages - Número aproximado de mensagens disponíveis na fila;
• ApproximateNumberOfMessagesNotVisible - Número aproximado de mensagens in flight;
• ApproximateNumberOfMessagesDelayed - Número aproximado de mensagens que estão aguardando para entrar na sua fila SQS. Sim! É uma fila da sua fila! :D

O método queue_size pega todos estes atributos e converte seus valores para inteiro e retorna a soma deles.

Talvez seja interessante recuperar estes valores separados, não? Então vamos lá!

  def queue_available_size
    response = client.get_queue_attributes({queue_url: queue_url, attribute_names: ["ApproximateNumberOfMessages"]})
    response.attributes["ApproximateNumberOfMessages"].to_i
  end

  def queue_unavailable_size
    response = client.get_queue_attributes({queue_url: queue_url, attribute_names: ["ApproximateNumberOfMessagesNotVisible"]})
    response.attributes["ApproximateNumberOfMessagesNotVisible"].to_i
  end

  def queue_waiting_size
    response = client.get_queue_attributes({queue_url: queue_url, attribute_names: ["ApproximateNumberOfMessagesDelayed"]})
    response.attributes["ApproximateNumberOfMessagesDelayed"].to_i
  end

Pronto! Temos cada um deles separado!

Abstrair AWS SQS QueuePoller

A abstração aqui será bem simples! Iremos retornar um objeto Aws::SQS::QueuePoller instanciado. O restante das tratativas serão feitas pela aplicação que for utilizar este poller.

  def poller
    @queue_poller ||= Aws::SQS::QueuePoller.new(queue_url, {max_number_of_messages:10, skip_delete: true, wait_time_seconds: nil, visibility_timeout: visibility_timeout})
  end

Para mais detalhes dos parâmetros confira a minha postagem anterior.

Vamos parar por aqui! Lembre-se de por o require ‘aws-sdk’ caso for usar esta classe em seu projeto.

Confira: Abstrair AWS SQS no Ruby Parte 2: Criando testes automatizados