Strumienie (Streams) umożliwiają sekwencyjne przetwarzanie danych w kawałkach, oszczędzając pamięć RAM przy dużych plikach.
Przegląd Typów Strumieni
| Typ | Kierunek | Opis |
|---|---|---|
Readable |
Tylko odczyt | Źródło, z którego można odczytywać dane |
Writable |
Tylko zapis | Miejsce docelowe, do którego można zapisywać dane |
Transform |
Odczyt i zapis | Modyfikuje lub przekształca dane w trakcie przepływu |
Duplex |
Odczyt i zapis | Zarówno do odczytu, jak i zapisu z niezależnymi kanałami (np. gniazdo TCP) |
Potoki i nowoczesne stream/promises
Nowoczesny Node.js oferuje funkcję pipeline do bezpiecznego łączenia strumieni z automatyczną obsługą błędów:
const { pipeline } = require('stream/promises');
const fs = require('fs');
const zlib = require('zlib');
async function compressFile() {
await pipeline(
fs.createReadStream('source.tar'),
zlib.createGzip(),
fs.createWriteStream('source.tar.gz')
);
console.log('Kompresja zakończona');
}
Tworzenie Własnych Strumieni
1. Własny Strumień Readable
const { Readable } = require('stream');
function createTimestampStream() {
let count = 0;
return new Readable({
objectMode: true,
read() {
if (count >= 5) {
this.push(null); // Koniec strumienia
} else {
this.push({ id: ++count, time: new Date() });
}
}
});
}
2. Własny Strumień Transform
const { Transform } = require('stream');
const uppercaseTransformer = new Transform({
transform(chunk, encoding, callback) {
const transformed = chunk.toString().toUpperCase();
callback(null, transformed);
}
});
3. Własny Strumień Writable
const { Writable } = require('stream');
const consoleWriter = new Writable({
objectMode: true,
write(chunk, encoding, callback) {
console.log('Zapis fragmentu:', chunk);
callback();
}
});
Tryb Flowing a Tryb Paused
Strumienie odczytu działają w trybie wstrzymanym (paused) lub płynnym (flowing):
const stream = fs.createReadStream('data.csv');
// Dołączenie nasłuchiwania 'data' automatycznie aktywuje tryb flowing
stream.on('data', (chunk) => {
console.log(`Odebrano ${chunk.length} bajtów`);
});
// Jawna kontrola przepływu
stream.pause(); // Wstrzymanie zdarzeń
stream.resume(); // Wznowienie zdarzeń strumieniowych
Zdarzenia Strumieni
| Zdarzenie | Dotyczy | Warunek Wyzwolenia |
|---|---|---|
data |
Readable |
Emitowane, gdy fragment danych jest gotowy do odczytu |
end |
Readable |
Emitowane, gdy brak kolejnych danych do odczytania |
finish |
Writable |
Emitowane, gdy wszystkie zapisane dane zostały przekazane |
error |
All | Emitowane w przypadku nieobsłużonego błędu strumienia |
close |
All | Emitowane po zamknięciu powiązanego zasobu bazowego |