Connettersi a un flusso di dati e utilizzare i dati

Per connettersi a un flusso di dati e utilizzare i dati, effettuare le operazioni riportate di seguito.

Passo 1: creare l'URL WebSocket

Leggere i seguenti valori dal file YAML AsyncAPI generato:

servers.<name>.protocol: ws o wss.
servers.<name>.host: nome host e, se presente, porta.
channels.<name>.address: percorso WebSocket completo per il flusso di dati.

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

Non codificare il percorso: OCI GoldenGate, una distribuzione on-premise e un proxy inverso possono esporre prefissi di instradamento diversi. Copiare l'indirizzo host e canale dall'YAML generato o dai collegamenti canonici restituiti dall'API REST.

Passo 2: selezionare il valore iniziale

Specificare il valore BEGIN in base alle linee guida fornite nella tabella seguente:

Valore	Usa	Linee guida
`now`	Inizia con una nuova attività	Convertito nell'indicatore orario corrente e risolto rispetto al trail disponibile. Le modifiche prima del punto risolto non vengono consegnate.
`earliest`	Backill iniziale dal trail disponibile	Convertito in indicatore orario zero. La consegna inizia dal primo record ancora disponibile nel trail configurato, non necessariamente dall'inizio della cronologia di origine.
Indicatore orario ISO 8601	Inizia vicino a un limite orario di lavoro	Il servizio esegue una ricerca basata sull'indicatore orario. Utilizzare un fuso orario esplicito, ad esempio `2026-07-13T04:00:00Z`.
Salvato `pos`	Recupero dopo un'esecuzione precedente riuscita	Il servizio esegue una ricerca basata sulla posizione. Il risultato quando la posizione non è disponibile dipende dalla qualità del servizio.

Codifica percentuale del valore iniziale

Un token pos può contenere caratteri quali +, / e = che hanno un significato speciale in un URL. Codifica in percentuale il token completo prima di aggiungerlo alla stringa di query. Non modificare o decodificare il token nella logica applicazione.

Passo 3: autenticazione e TLS

Selezionare il metodo di autenticazione. Utilizzare la tabella seguente per determinare il metodo più adatto.

Modalità	Quando usarla	Risposta del cliente
Autenticazione Basic	Comune con un servizio esposto direttamente o proxy inverso	Invia un'intestazione `Authorization: Basic ...`. Conservare le credenziali in un archivio segreto o in un ambiente, non nell'URL o nel codice sorgente.
Autenticazione digest	Quando richiesto dal proxy inverso	Utilizzare un client o un modello generato che implementa la richiesta di verifica digest HTTP durante l'handshake WebSocket.
Certificato client	Autenticazione del servizio basata su certificato o TLS reciproca	Caricare il certificato client e la chiave privata nel contesto TLS e convalidare il server rispetto alla CA configurata.
Proxy sicuro	Gateway termina o inoltra TLS	Seguire la modalità di autenticazione proxy. Un proxy sicuro può richiedere sia le credenziali di base/digest che un certificato client.

Passo 4: Generare un client da AsyncAPI

Oracle fornisce un modello client WebSocket per il generatore AsyncAPI. Può generare JavaScript, Python, Java, Go o tutti i client supportati. Il generatore richiede Node.js 18.12 o versione successiva e npm 8.19 o versione successiva.

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

Il modello supporta anche le variabili di certificato ASYNCAPI_WS_CLIENT_CERT, ASYNCAPI_WS_CLIENT_KEY e ASYNCAPI_WS_CA_CERT. Rivedere le versioni di codice e dipendenza generate nel solito processo di sviluppo software sicuro dell'organizzazione.

Passo 5: Comprendere i messaggi WebSocket

Il solito payload del messaggio è un array JSON. Un frame può contenere uno o più record di schema, metadati, DML o DDL. Il client deve iterare l'array e classificare ogni oggetto in modo indipendente.

Backpressure: il servizio esegue il push dei record senza una conferma del client per record. Tenere il ciclo di ricezione leggero. Metti un'elaborazione aziendale lenta dietro una coda interna delimitata o un livello di staging permanente, monitora la profondità della coda e arresta o ridimensiona il consumatore prima che la memoria sia esaurita.

Passo 6: ciclo di elaborazione del consumatore

Ricevere un messaggio WebSocket completo.
Analizza l'array JSON. Quando CloudEvents è abilitato, annullare il wrapping del membro data.
Classifica i record di schema, metadati del flusso, metadati dell'oggetto, DML e DDL.
Convalidare i record rispetto allo schema inserito nella cache più recente quando è richiesta la convalida.
Applicare l'operazione aziendale o pubblicarla in un sistema a valle durevole.
Una volta completata l'operazione, salvare il record pos utilizzando la struttura di checkpoint atomico o idempotente selezionata.
In caso di perdita della connessione, riconnettersi all'ultimo pos durevole e al backoff esponenziale.