Forem: Mayderson Mello

Race condition em sistemas distribuídos

Mayderson Mello — Sat, 03 Feb 2024 23:06:48 +0000

Cenário:

Temos um serviço de pagamento que consome dados de duas filas:

clients-processor: fila com dados de clientes para cadastrar
payment-processor: fila com dados de um pagamento para ser processado

Esse serviço processa os pagamentos, porém ele só processa pagamentos de clientes que estão cadastrados.

Problema:

Como tratar o problema de race condition quando um evento de pagamento é recebido primeiro que o evento de cadastro de clientes (sem utilizar bancos de dados). Para simplificar mais temos a premissa que o evento de cadastro de cliente sempre será recebido.

A solução:

clients-processor: Uma fila para evento de cadastro de clientes
payment-processor: Uma fila para evento de pagamentos
payment-reprocessor: Uma fila para reprocessar eventos de pagamentos utilizando o plugin delayed_message_exchange do RabbitMQ

Quando um evento de pagamento é recebido e o cliente ainda não está cadastrado, aumentamos o valor do delay da mensagem em 1 segundo. A mensagem é então encaminhada para a fila de reprocessamento. Em algum momento, quando um cliente for efetivamente cadastrado, o pagamento correspondente será processado.

Para exemplificar melhor, o fluxo grama abaixo mostra a abordagem utilizada:

Exemplo:

Crie uma aplicação com Golang + RabbitMQ para implementar toda abordagem comentada acima, você pode encontrá-la no meu GitHub, e lá terá as instruções para rodar.

Para o exemplo, vamos enviar duas mensagens, uma com o pagamento e outra com o novo cliente.

Ao receber um pagamento de um cliente com o ID 1, o pagamento não vai ser processado, a mensagem é enviada para uma fila e o delay vai aumentando, até que a mensagem de cadastro do cliente é recebida, no proximo recebimento do pagamento ele será processado e a operação finaliza.

Vídeo da aplicação:

Bom é isso, espero que tenha gostado da estratégia, sobre a aplicação, a mesma foi escrita em Golang, mas mesmo que você não desenvolva com Go, é uma aplicação muito simples e acredito que não terá problema em entender o código.

Server-Sent Events (SSE) e como implementar

Mayderson Mello — Sat, 08 Apr 2023 00:54:23 +0000

Server-Sent Events (SSE) é uma tecnologia que permite a comunicação unidirecional entre o servidor e o cliente em tempo real. SSE é uma alternativa aos WebSockets que oferece várias vantagens, como menor sobrecarga de rede e menor uso de recursos. Neste tutorial, iremos aprender como implementar SSE em Go e compará-lo com WebSockets.

O que é SSE?

SSE é um protocolo de comunicação baseado em HTTP que permite que o servidor envie eventos para o cliente. O cliente se inscreve em um fluxo SSE, que é uma conexão persistente entre o cliente e o servidor. O servidor envia eventos em formato de texto para o cliente, que é capaz de processá-los em tempo real. Os eventos podem ser nomeados e o payload pode ser em formato de texto ou JSON.

Como implementar?

O exemplo a seguir mostra como implementar SSE utilizando Golang, mas não se apegue a linguagem pois pode ser qualquer outra, basta definir os Headers corretos que você será capaz reproduzir.

Back-end (Golang)



package main

import (
    "fmt"
    "net/http"
    "time"
)

func main() {
    http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
        http.ServeFile(w, r, "index.html")
    })

    http.HandleFunc("/sse", func(w http.ResponseWriter, _ *http.Request) {
        w.Header().Set("Content-Type", "text/event-stream")
        w.Header().Set("Cache-Control", "no-cache")
        w.Header().Set("Connection", "keep-alive")

        messageCounter := 0

        for {
            someMessage := fmt.Sprintf("Message: %v", messageCounter)

            eventName := "event: myCustomEvent\n"
            payloadMessage := fmt.Sprintf("data: %s\n\n", someMessage)

            w.Write([]byte(eventName))
            w.Write([]byte(payloadMessage))

            time.Sleep(1 * time.Second)

            w.(http.Flusher).Flush()

            messageCounter++
        }
    })

    http.ListenAndServe(":3333", nil)
}

A rota /sse define os cabeçalhos necessários para enviar eventos SSE. O cabeçalho Content-Type é definido como text/event-stream, Cache-Control é definido como no-cache para evitar o armazenamento em cache e Connection é definido como keep-alive para manter a conexão aberta. Em seguida, o servidor inicia um loop infinito para enviar eventos SSE. O evento é opcional e pode ser usado para especificar o tipo de evento que está sendo enviado. O payload contém os dados que estão sendo enviados.

O uso do \n é obrigatório, você pode observar que uso ele ao definir o nome do evento e payload da mensagem. O SSE vai conseguir interpretar quando essas informações foram definidas após essa quebra de linha, por fim, o \n adicional informa que acabou aquele evento.

Exemplo de como vai ficar no final após a interpretação:



event: myCustomEvent
data: Message: 1

event: myCustomEvent
data: Message: 2

Front-end (HTML + JavaScript)



<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <meta http-equiv="X-UA-Compatible" content="IE=edge">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
  <title>SSE</title>
</head>
<body>
  <h1>SSE - Real Time Messages</h1>

  <ul id="messages"></ul>

  <script>
    const eventSource = new EventSource('/sse');

    eventSource.addEventListener('myCustomEvent', event => {
      const messagesElement = document.getElementById('messages');
      const li = document.createElement('li');

      li.innerText = event.data;

      messagesElement.appendChild(li);

      console.log(event);
    });
  </script>
</body>
</html>

O código acima é muito simples, basicamente vamos estar recebendo as mensagens que o nosso Back-end vai mandar, e vamos adicionar elas em tela, somente isso.

A instância do EventSource, recebe a URL do endpoint SSE que o servidor está ouvindo (/sse neste caso). Isso estabelece uma conexão persistente entre o cliente e o servidor, permitindo que o servidor envie atualizações em tempo real para o cliente.

Em seguida, é adicionado um ouvinte de eventos para o evento personalizado myCustomEvent, que é enviado pelo servidor. Quando esse evento é recebido, uma nova linha de lista li é criada, contendo o texto da mensagem que foi enviada pelo servidor. A linha de lista é então adicionada ao elemento ul com o id messages, que é onde as mensagens são exibidas na página.

Vantagens e Desvantagens do SSE

SSE tem várias vantagens em relação aos WebSockets. A primeira é que ele é baseado em HTTP e usa a porta HTTP padrão (80), o que significa que ele pode passar por firewalls e proxies sem problemas. Além disso, o SSE requer menos sobrecarga de rede e recursos do servidor do que o WebSocket, porque apenas o servidor pode enviar dados. Isso significa que o servidor não precisa lidar com tantas conexões de saída e, portanto, pode lidar com mais conexões de entrada. Finalmente, o SSE é fácil de implementar, pois é baseado em HTTP e usa o mesmo modelo de programação.

No entanto, o SSE também tem algumas desvantagens. A primeira é que ele só suporta comunicação unidirecional (do servidor para o cliente), o que significa que o cliente não pode enviar dados para o servidor. Além disso, o SSE não é suportado por todos os navegadores e plataformas, o que significa que você precisará testar se o SSE funciona em todos os navegadores e plataformas que você deseja suportar.

Comparação SSE vs WebSockets

Embora SSE e WebSockets possam parecer semelhantes, eles têm algumas diferenças importantes. A principal diferença é que o WebSocket é bidirecional, o que significa que tanto o cliente quanto o servidor podem enviar dados. Isso torna o WebSocket mais adequado para aplicativos em tempo real que exigem interação bidirecional entre o cliente e o servidor, como jogos online ou aplicativos de bate-papo. O SSE, por outro lado, é mais adequado para aplicativos unidirecionais, como feeds de notícias ou atualizações de preços.

Além disso, o WebSocket usa um protocolo de comunicação personalizado que é implementado em um nível mais baixo do que o HTTP. Isso significa que o WebSocket é capaz de suportar recursos avançados, como multiplexação de canais, compressão de dados e autenticação do lado do servidor. O SSE, por outro lado, é implementado como um recurso do HTTP e, portanto, não pode usar esses recursos avançados.

Enviando Push Notification com Node.js para o Browser

Mayderson Mello — Sat, 28 Jan 2023 23:15:49 +0000

Introdução

Neste tutorial vamos aprender a enviar Push Notification para o Browser usando Node.js, irei utilizar também o React com Vite, mas poderia ser qualquer outra coisa, pois vamos utilizar de API's nativas da WEB para isso.

Será abordado o uso da Notification API para solicitarmos a permissão para o usuário, PushManager API para receber notificações de servidores de terceiros, web-push para fazer o envio da notificação no lado servidor, e por fim o Service Worker, para lidar com a notificação em Background.

Criando Back-end

Antes de tudo, precisamos criar nosso servidor em Node.js para fazer o envio dessas notificações, vou utilizar o Express para isso, mas poderia ser com qualquer outro, como o Fastify etc.

Instalando pacotes necessários e configurando script

$ mkdir server
$ cd server
$ npm init -y
$ npm i express cors web-push
$ npm i typescript tsx @types/express @types/cors @types/node @types/web-push -D
$ npx tsc --init

Adicione no seu package.json o script de dev:

"scripts": {
  "dev": "tsx watch src/server.ts"
}

Criando servidor

Crie um arquivo dentro da pasta src chamado server.ts e adicione o código abaixo:

import express from 'express';
import cors from 'cors';
import WebPush from 'web-push';

const app = express();

app.use(express.json());
app.use(cors());

// WebPush

// console.log(WebPush.generateVAPIDKeys());
const publicKey = 'DASdasdjk1hdhajshdashdgadaskdhaj...';
const privateKey = 'dsaduiw18eudasd1d9sdd3wddasdasa...';

WebPush.setVapidDetails('http://localhost:3333', publicKey, privateKey);

// Routes

interface ISubscription {
  subscription: {
    endpoint: string;
    keys: {
      p256dh: string;
      auth: string;
    };
  };
}

app.get('/notification/push/public_key', (request, response) => {
  return response.json({ publicKey });
});

app.post('/notification/push/register', (request, response) => {
  console.log(request.body);

  return response.sendStatus(201);
});

app.post('/notification/push/send', async (request, response) => {
  const { subscription } = request.body as ISubscription;

  WebPush.sendNotification(
    subscription,
    JSON.stringify({
      icon: 'your-icon-link.png',
      title: "'Your title',"
      body: 'Content of your message',
      imageUrl: 'your-image-link.png'
    }),
  );

  return response.sendStatus(201);
});

app.listen(3333, () => console.log('SERVER IS RUNNING AT :3333'));

Bom de Back-end é só isso, para você gerar sua publicKey e privateKey é necessário rodar o código console.log(WebPush.generateVAPIDKeys()); somente uma vez, com isso ele vai te retornar suas chaves, ai basta você substituir essas fictícias que coloquei pelas as suas.

Sobre as rotas do servidor:

/notification/push/public_key: Essa rota é responsável por enviar nossa publicKey para o Front-end.
/notification/push/register: Essa rota serviria para você salvar no Banco de Dados por exemplo a subscription que o Front-end iria enviar, e com essa informação você poderia enviar a notificação depois, mas para não estender muito não irei abordar essa parte de salvar em algum Banco de Dados.
/notification/push/send: Essa rota serve para fazer o envio da notificação, nela estou recebendo a subscription no body, pois é necessário para o envio da notificação, mas caso você tenha persistido essas subscriptions no Banco de Dados, você poderia substituir o recebimento dela nessa rota, e realizar a consulta dessas subscriptions existentes.

Preparando nosso Front-end

Criando o Service Worker

Vamos criar um arquivo chamado service-worker.ts dentro da pasta public pois esse arquivo precisa estar disponível publicamente.

Estou utilizando .ts pois TypeScript é vida ❤️, mas se você quiser pode usar .js e remover as tipagens que irei adicionar, de qualquer forma se você usar com TypeScript, vai precisar converter para JavaScript no final, mas fique tranquilo irei mostrar como posteriormente.

/// <reference no-default-lib="true"/>
/// <reference lib="esnext" />
/// <reference lib="webworker" />

interface INotificationPayload {
  icon: string;
  title: string;
  body: string;
  imageUrl?: string;
}

const sw = self as unknown as ServiceWorkerGlobalScope & typeof globalThis;

sw.addEventListener('push', (event) => {
  const notificationPayload = event.data?.json() as INotificationPayload;

  event.waitUntil(
    sw.registration.showNotification(notificationPayload.title, {
      icon: notificationPayload.icon,
      body: notificationPayload.body,
      image: notificationPayload.imageUrl
    }),
  );
});

O código acima é bem simples, basicamente adicionamos um listener chamado push, com isso, quando for enviado alguma notificação vamos ficar sabendo, e teremos acesso ao evento que foi disparado, com esse evento conseguimos obter o payload que é os dados que foram enviados, e com esses dados conseguimos exibir de fato a notificação para o usuário.

Você pode explorar mais atributos, que podem ser passados dentro desse objeto de configuração da função showNotification, por meio da documentação oficial.

Solicitando permissão

Agora precisamos solicitar as permissões para o usuário, ele precisa de fato autorizar o recebimento dessas notificações para estar chegando para ele.

Você pode implementar o código abaixo no arquivo principal do seu projeto, para que quando o mesmo for iniciado chame a função abaixo.

function notifyMe() {
  if (!('Notification' in window)) {
    alert('This browser does not support desktop notification');
  } else if (Notification.permission === 'granted') {
    handleServiceWorker();
  } else if (Notification.permission !== 'denied') {
    Notification.requestPermission().then((permission) => {
      if (permission === 'granted') {
        handleServiceWorker();
      }
    });
  }
}

notifyMe();

Basicamente na função acima, realizamos uma verificação para checar se temos acesso a Notification API, em seguida verificamos se já temos acesso garantido a notificação, se já tivermos só executamos a função handleServiceWorker que criaremos depois, por fim se ainda não tivermos concedido permissão para a notificação, solicitamos ao usuário para que garanta o acesso.

Criando função handleServiceWorker

Essa função vai realizar o registro do nosso Service Worker, e após ser realizado o registro com sucesso, chamamos a função getSubscription do pushManager, pois ela recupera uma assinatura push existente e retorna seus detalhes, caso não encontrar retorna null.

Quando não temos uma assinatura existente, chamamos nosso endpoint /notification/push/public_key para retornar nossa publicKey, e com o valor dessa chave realizamos o subscribe, assim da próxima vez que tentarmos acessar a aplicação já vamos ter nossa inscrição feita.

Fiz também para fim de exemplo a chamada para o endpoint de registro da subscription e de envio da notificação, mas cabe a você definir quando e onde realizar o envio dessas notificações da melhor maneira.

async function handleServiceWorker() {
  navigator.serviceWorker
    .register('service-worker.js')
    .then(async (serviceWorker) => {
      serviceWorker.update();

      let subscription = await serviceWorker.pushManager.getSubscription();

      if (!subscription) {
        const publicKeyResponse = await api.get<{ publicKey: string }>(
          '/notification/push/public_key',
        );

        await serviceWorker.pushManager.subscribe({
          userVisibleOnly: true,
          applicationServerKey: publicKeyResponse.data.publicKey,
        });

        subscription = await serviceWorker.pushManager.getSubscription();
      }

      await axios.post('http://localhost:3333/notification/push/register', {
        subscription,
      });

      await axios.post('http://localhost:3333/notification/push/send', {
        subscription,
      });
    });
}

obs: usei o axios como lib para realizar a chamada HTTP, mas poderia ser o fetch normalmente.

Convertendo Service Worker para JavaScript

Essa etapa não é necessária se você criou o arquivo service-worker.js, mas caso tenha criado ele em TypeScript execute os passos abaixo:

Instale a dependência esbuild como dependência de desenvolvimento:

$ npm i esbuild -D

Adicione o script abaixo no seu arquivo package.json:

"scripts": {
  "build-sw": "esbuild public/service-worker.ts --outdir=public --bundle --sourcemap --minify --format=esm --legal-comments=external"
}

Agora é só executar o script build-sw:

$ npm run build-sw

Pronto, agora o service-worker.ts vai ser convertido para JavaScript.

Criando react hook personalizado para fazer requisições

Mayderson Mello — Fri, 27 Jan 2023 00:08:23 +0000

Introdução

Este tutorial mostrará como criar um hook customizado chamado useFetcher usando o useReducer, ele vai permitir que façamos requisições HTTP de maneira fácil além de gerenciarmos o estado dessas requisições.

Imports

Vamos começar criando um arquivo chamado useFetcher.ts, gosto de colocar esses custom hooks dentro de src/hooks, agora vamos importar as dependências necessárias:

import { useCallback, useEffect, useReducer } from 'react';

Criar hook useFetcher

Em seguida, vamos criar uma função chamada useFetcher que possui um parâmetro para a URL que desejamos fazer a requisição, essa função terá também um generic que vai auxiliar na tipagem desse hook, e o tipo StateProps representa o que será retornado desse hook:

type StateProps<T> = {
  data?: T;
  loading: boolean;
  error?: Error;
};

function useFetcher<T = unknown>(url: string): StateProps<T> {
  ...
}

Agora vamos criar nosso estado inicial:

function useFetcher<T = unknown>(url: string): StateProps<T> {
  const initialState: StateProps<T> = {
    data: undefined,
    loading: true,
    error: undefined,
  };
}

Certo agora precisamos criar nossa função de reducer, ela será responsável por conter nosso state e action, no state vamos ter acesso ao data, loading e error e no action teremos acesso ao type e payload:

type ActionType<T> =
  | { type: 'loading' }
  | { type: 'fetched'; payload: T }
  | { type: 'error'; payload: Error };

function useFetcher<T = unknown>(url: string): StateProps<T> {
  ...

  function reducer(state: StateProps<T>, action: ActionType<T>) {
    switch (action.type) {
      case 'loading':
        return { ...state, loading: true } as StateProps<T>;
      case 'fetched':
        return {
          ...state,
          data: action.payload,
          loading: false,
        } as StateProps<T>;
      case 'error':
        return {
          ...state,
          error: action.payload,
          loading: false,
        } as StateProps<T>;
      default:
        return state;
    }
  }
}

Basicamente, na função acima usamos o switch para verificar se o type disparado é loading, fetched ou error, e com isso retornamos um objeto, com a cópia do estado e as modificações necessárias.

Agora vamos criar nosso handleFetch, ele será responsável por realizar de fato a requisição, além disso ele realiza dispatch do type necessário em cada momento da chamada a API:

1º - Ao executar a função ele dispara o type: loading.
2º - Ao finalizar com sucesso a requisição ele dispara o type: fetched + payload, esse payload é os dados retornados da API.
3º - Em caso de erro na requisição ele dispara o type: error + payload, esse payload será o erro que foi retornado.

function useFetcher<T = unknown>(url: string): StateProps<T> {
  ...

  const [state, dispatch] = useReducer(reducer, initialState);

  const handleFetch = useCallback(async () => {
    dispatch({ type: 'loading' });

    try {
      const response = await axios.get<T>(url);

      dispatch({ type: 'fetched', payload: response.data });
    } catch (error) {
      if (axios.isAxiosError(error) || error instanceof Error) {
        dispatch({ type: 'error', payload: error });
      }
    }
  }, [url]);

  useEffect(() => {
    handleFetch();
  }, [handleFetch]);

  return {
    data: state.data,
    loading: state.loading,
    error: state.error,
  };
}

obs: utilizei como exemplo o axios para realizar a requisição, mas poderia ser trocado pelo fetch normalmente.

Testando useFetcher

Por fim, só falta testar o nosso custom hook:

import { useFetcher } from '../hooks/useFetcher';

interface ITodo {
  userId: number;
  id: number;
  title: string;
  completed: boolean;
}

const Home = () => {
  const {
    loading: loadingTodos,
    data: todosData,
    error: todosError,
  } = useFetcher<ITodo>('https://jsonplaceholder.typicode.com/todos/1');

  return (
    <div>
      {loadingTodos && <h1>Loading...</h1>}

      {todosData && <pre>{JSON.stringify(todosData, null, 2)}</pre>}

      {todosError && <h1>{todosError?.message}</h1>}
    </div>
  );
};