Ajouter des flux de données

Avec Oracle GoldenGate Data Streams, l'accès direct aux données au format indiqué par l'utilisateur est activé via un canal de socket Web dédié qui suit un protocole de transmission en continu simple.

Le protocole Data Streams utilise le mode Push pour envoyer des données au client. Le client crée d'abord une ressource de transmission en continu sur le serveur via une demande HTTP RESTful. Une fois la ressource de transmission en continu créée, le client établit une connexion WebSocket via l'adresse de ressource de transmission en continu. Une fois le canal WebSocket établi, Data Streams commence à transmettre les données immédiatement et en continu sans attendre la réponse ou l'accusé de réception du client.

L'exemple de client python suivant illustre l'interaction entre le client et le service de transmission en continu de données :

import asyncio
import requests
import websockets
import json

async def client():
    ### create the streaming resource
    payload = {"source":{"trail":"a1"}}
    response = requests.post(
  'http://name:pswd@localhost:9002/services/v2/stream/s1', json=payload)

### establish websocket connection and receive data continuously
uri = "ws://name:pswd@localhost:9002/services/v2/stream/s1?begin=earliest"
async with websockets.connect(uri) as websocket:
        while True:
            resp = await websocket.recv()
            records = json.loads(resp)
            for rec in records:
                print(rec)
asyncio.get_event_loop().run_until_complete(client())

Dans le programme client donné, une charge utile de flux de données simple indiquant le nom de la trace de données source est fournie lors de la création de l'adresse de ressource de flux de données s1. Dans une application du monde réel, des charges utiles de flux de données très complexes peuvent être utilisées pendant la phase d'établissement de liaison du protocole de transmission en continu pour configurer le comportement de transmission en continu des données.

Par exemple, la charge utile de demande de flux de données suivante indique les règles de filtrage, le format d'encodage et bufferSize, ainsi que le nom de trace de source de données requis.

{   
    "$schema"    : "ogg:dataStream",
    "source"     : {"trail":"a1"},
    "rules"      : [{
        "action" : "exclude",
        "filter" : {
            "objectNames" : [ "PDB2.AR.*", "PDB1.U1.*" ]
        }
    }],
    "encoding"   : “json",
    "bufferSize" : 2097152
}

Lors de l'établissement de la connexion websocket, côté client indique la position begin (en tant que paramètre de requête dans l'URL de connexion websocket) pour démarrer la transmission en continu des données. La position de début peut avoir l'une des valeurs suivantes :

Chaque enregistrement LCR sans métadonnées contient une position opaque (inclut CSN, XID, no d'enregistrement dans la transaction). Le côté client est responsable de la maintenance de la position du dernier enregistrement LCR traité. Le service de flux de données est chargé de localiser le point de démarrage/redémarrage correct en fonction de la position de début donnée.

S'il s'agit de la première connexion d'un client au service de flux de données, le client doit fournir un horodatage indiquant où démarrer la transmission en continu des données. Le mot-clé now sera converti en horodatage actuel et le mot-clé earliest sera converti en horodatage 0.

Vous pouvez également utiliser une chaîne d'horodatage au format ISO 8601 pour la position begin. Dans tous les cas, le service de flux de données effectue une recherche basée sur l'horodatage sur la trace source pour déterminer la position de début.

S'il s'agit du cas de récupération/redémarrage, le client doit fournir la dernière position enregistrée traitée au service de flux de données lors de l'établissement de liaison. Le service de flux de données effectue une recherche basée sur le profil sur la trace source pour déterminer la position de début. Le comportement de la récupération de transmission en continu de données dépend également du niveau QoS indiqué dans le flux de données.

Vous pouvez utiliser les API Rest suivantes pour gérer les flux de données GoldenGate.

CloudEvents est une spécification pour décrire les données d'événement dans des formats communs afin de fournir une interopérabilité entre les services, les plates-formes et les systèmes. Les flux de données Oracle GoldenGate ne prenant actuellement en charge que l'encodage de données JSON, la prise en charge du format CloudEvents est limitée au format d'événement JSON. La spécification complète pour le format d'événement JSON pour CloudEvents est disponible à l'adresse suivante :

https://github.com/cloudevents/spec/blob/main/cloudevents/spec.md

https://github.com/cloudevents/spec/blob/main/cloudevents/formats/json-format.md

https://github.com/cloudevents/spec/blob/main/cloudevents/formats/cloudevents.json

{
    "specversion" : "1.0",
    "type" : "com.example.someevent",
    "source" : "/mycontext",
    "id" : "A234-1234-1234",
    "datacontenttype" : "application/json",
    "data: {…}
}