Estabelecer Conexão com um Fluxo de Dados e Consumir Dados

Use as seguintes etapas para estabelecer conexão com um Fluxo de Dados e consumir dados:

Etapa 1: Criar o URL WebSocket

Leia os seguintes valores do AsyncAPI YAML gerado:

servers.<name>.protocol: ws ou wss.
servers.<name>.host: nome do host e, quando presente, porta.
channels.<name>.address: caminho WebSocket completo para o Fluxo de Dados.

wss://<server-host><channel-address>?begin=<percent-encoded-begin-value>

Não codifique o caminho: O OCI GoldenGate, uma implantação on-premises e um proxy reverso podem expor diferentes prefixos de roteamento. Copie o endereço do host e do canal do YAML gerado ou dos links canônicos retornados pela API REST.

Etapa 2: selecione o valor inicial

Especifique o valor BEGIN com base nas diretrizes fornecidas na seguinte tabela:

Valor	Uso	Diretrizes
`now`	Começar com nova atividade	Convertido para o timestamp atual e resolvido em relação à trilha disponível. As alterações antes do ponto resolvido não são entregues.
`earliest`	Backfill inicial da trilha disponível	Convertido para timestamp zero. A entrega começa no registro mais antigo ainda disponível na trilha configurada, não necessariamente no início do histórico de origem.
Timestamp ISO 8601	Comece perto de um limite de tempo de negócios	O serviço executa uma pesquisa baseada em timestamp. Use um fuso horário explícito, por exemplo, `2026-07-13T04:00:00Z`.
Salvo `pos`	Recuperação após uma execução anterior bem-sucedida	O serviço executa uma pesquisa baseada em posição. O resultado quando a posição está indisponível depende da qualidade do serviço.

Porcentagem de codificação do valor inicial

Um token pos pode conter caracteres como +, / e = que têm significado especial em um URL. Codifique o token completo em porcentagem antes de adicioná-lo à string de consulta. Não modifique ou decodifique o token na lógica do aplicativo.

Etapa 3: Autenticação e TLS

Selecione o método de autenticação. Use a seguinte tabela para determinar o método mais adequado:

Modo	Quando usar	Resposta do Cliente
Autenticação básica	Comum com um serviço exposto diretamente ou proxy reverso	Envie um cabeçalho `Authorization: Basic ...`. Mantenha as credenciais em um armazenamento secreto ou em uma injeção de ambiente, não no URL ou código-fonte.
Autenticação do digest	Quando exigido pelo proxy reverso	Use um cliente ou modelo gerado que implemente o desafio de compilação HTTP durante o handshake WebSocket.
Certificado do cliente	Autenticação de serviço baseada em certificado ou TLS mútuo	Carregue o certificado do cliente e a chave privada no contexto TLS e valide o servidor em relação à CA configurada.
Proxy reverso seguro	Gateway encerra ou encaminha TLS	Siga o modo de autenticação de proxy. Um proxy seguro pode exigir credenciais básicas/digest e um certificado de cliente.

Etapa 4: Gerar um cliente do AsyncAPI

O sistema Oracle fornece um modelo de cliente WebSocket para o Gerador AsyncAPI. Ele pode gerar JavaScript, Python, Java, Go ou todos os clientes suportados. O gerador requer Node.js 18.12 ou posterior e npm 8.19 ou posterior.

npm install -g @asyncapi/generator

git clone https://github.com/oracle-samples/websocket-client-template.git
cd websocket-client-template

ag /path/to/sales-orders.yaml .
  -o output
  -p server=production
  -p language=python
  -p authorization=basic

cd output
python -m pip install -r requirements.txt
export ASYNCAPI_WS_CLIENT_USERNAME="$OGG_USER"
export ASYNCAPI_WS_CLIENT_PASSWORD="$OGG_PASSWORD"
python client.py

O modelo também suporta variáveis de certificado ASYNCAPI_WS_CLIENT_CERT, ASYNCAPI_WS_CLIENT_KEY e ASYNCAPI_WS_CA_CERT. Revise as versões de código e dependência geradas no processo de desenvolvimento de software seguro usual da sua organização.

Etapa 5: Compreender mensagens WebSocket

O payload de mensagem usual é um array JSON. Um quadro pode conter um ou mais registros de esquema, metadados, DML ou DDL. O cliente deve iterar o array e classificar cada objeto de forma independente.

Backpressure: O serviço envia registros sem uma confirmação do cliente por registro. Mantenha o loop de recepção leve. Coloque o processamento de negócios lento em uma fila interna delimitada ou camada de preparação durável, monitore a profundidade da fila e interrompa ou escale o consumidor antes que a memória esteja esgotada.

Etapa 6: Loop de processamento do consumidor

Receba uma mensagem WebSocket completa.
Faça parse do array JSON. Quando CloudEvents estiver ativado, desencapsule o membro data.
Classifique registros de esquema, metadados de fluxo, metadados de objeto, DML e DDL.
Valide registros em relação ao esquema em cache mais recente quando a validação for necessária.
Aplique a operação de negócios ou publique-a em um sistema downstream durável.
Após a operação ser bem-sucedida, salve o registro pos usando o design de checkpoint atômico ou idempotente selecionado.
Na perda de conexão, reconecte-se com o último pos durável e o backoff exponencial.