Streaming de Données

class MisraGries {
    constructor(k) {
        this.k = k; // Nombre maximum d'éléments à suivre
        this.counters = new Map(); // Map pour stocker les éléments et leurs compteurs
    }

    processElement(element) {
        if (this.counters.has(element)) {
            // Si l'élément est déjà dans la Map, incrémenter son compteur
            this.counters.set(element, this.counters.get(element) + 1);
        } else if (this.counters.size < this.k - 1) {
            // Si la Map n'a pas encore atteint sa capacité maximale, ajouter l'élément
            this.counters.set(element, 1);
        } else {
            // Si la Map est pleine, décrémenter tous les compteurs
            for (let [key, value] of this.counters) {
                this.counters.set(key, value - 1);
                if (this.counters.get(key) === 0) {
                    this.counters.delete(key); // Supprimer les éléments dont le compteur est 0
                }
            }
        }
    }

    getFrequentElements() {
        // Retourner les clés de la Map (éléments fréquents)
        return Array.from(this.counters.keys());
    }
}

// Exemple d'utilisation
const stream = [1, 2, 3, 1, 2, 1, 4, 5, 1];
const mg = new MisraGries(3); // Créer une instance de Misra-Gries avec k = 3
stream.forEach(element => mg.processElement(element)); // Traiter chaque élément du flux
console.log(mg.getFrequentElements()); // Output: [1, 2]

class SlidingWindow {
    constructor(size) {
        this.size = size; // Taille de la fenêtre glissante
        this.window = []; // Tableau pour stocker les éléments de la fenêtre
        this.sum = 0; // Somme des éléments dans la fenêtre
    }

    addElement(element) {
        this.window.push(element); // Ajouter un nouvel élément à la fenêtre
        this.sum += element; // Ajouter l'élément à la somme totale
        if (this.window.length > this.size) {
            // Si la fenêtre dépasse la taille spécifiée
            this.sum -= this.window.shift(); // Supprimer l'élément le plus ancien et ajuster la somme
        }
    }

    getAverage() {
        // Retourner la moyenne des éléments dans la fenêtre
        return this.sum / this.window.length;
    }
}

// Exemple d'utilisation
const sw = new SlidingWindow(3); // Créer une fenêtre glissante de taille 3
[1, 2, 3, 4, 5].forEach(num => {
    sw.addElement(num); // Ajouter chaque nombre à la fenêtre
    console.log(`Moyenne après ajout de ${num}: ${sw.getAverage()}`); // Afficher la moyenne actuelle
});
// Output: 1, 1.5, 2, 3, 4

class StreamJoin {
    constructor(windowSize) {
        this.windowSize = windowSize; // Taille de la fenêtre temporelle
        this.streamA = new Map(); // Map pour stocker les éléments du flux A
        this.streamB = new Map(); // Map pour stocker les éléments du flux B
    }

    addToStreamA(key, value, timestamp) {
        this._addToStream(this.streamA, key, value, timestamp); // Ajouter un élément au flux A
    }

    addToStreamB(key, value, timestamp) {
        this._addToStream(this.streamB, key, value, timestamp); // Ajouter un élément au flux B
    }

    _addToStream(stream, key, value, timestamp) {
        if (!stream.has(key)) {
            stream.set(key, []); // Initialiser la liste si la clé n'existe pas
        }
        stream.get(key).push({ value, timestamp }); // Ajouter l'élément à la liste
        this._cleanUpStream(stream, timestamp); // Nettoyer les éléments expirés
    }

    _cleanUpStream(stream, currentTime) {
        for (let [key, values] of stream) {
            // Filtrer les éléments en gardant ceux qui sont dans la fenêtre temporelle
            stream.set(key, values.filter(item => currentTime - item.timestamp <= this.windowSize));
            if (stream.get(key).length === 0) {
                stream.delete(key); // Supprimer la clé si aucune valeur n'est dans la fenêtre
            }
        }
    }

    joinStreams() {
        const results = [];
        for (let [key, valuesA] of this.streamA) {
            if (this.streamB.has(key)) {
                const valuesB = this.streamB.get(key); // Récupérer les valeurs du flux B pour la clé
                for (let a of valuesA) {
                    for (let b of valuesB) {
                        // Ajouter les paires jointes aux résultats
                        results.push({ key, valueA: a.value, valueB: b.value });
                    }
                }
            }
        }
        return results; // Retourner les résultats de la jointure
    }
}

// Exemple d'utilisation
const joiner = new StreamJoin(10); // Créer une instance de jointure de flux avec une fenêtre de 10 unités de temps
joiner.addToStreamA('x', 'A1', 1); // Ajouter un élément au flux A
joiner.addToStreamB('x', 'B1', 2); // Ajouter un élément au flux B
joiner.addToStreamA('y', 'A2', 3); // Ajouter un autre élément au flux A
joiner.addToStreamB('x', 'B2', 5); // Ajouter un autre élément au flux B
joiner.addToStreamB('y', 'B3', 6); // Ajouter un autre élément au flux B
console.log(joiner.joinStreams()); // Output: [ { key: 'x', valueA: 'A1', valueB: 'B1' }, { key: 'x', valueA: 'A1', valueB: 'B2' } ]