Stream e pipe (con una p ...)

Con questo post, vorrei approfondirei con voi i concetti di stream e pipe presenti nella piattaforma node.js

Uno stream non è altro che un flusso di dati dinamico: può essere una sorgente di dati, una destinazione di dati o entrambi.

Una pipe è un condotto che unisce una sorgente dati ad una destinazione dati, preoccupandosi di adattare le eventuali differenti velocità di elaborazione dati delle sorgenti e delle destinazioni.

Il concetto di Stream è un'astrazione: i dati veri e propri possono risiedere in un file così come provenire da un server, l'utilizzo degli stream e pipe è identico indipendentemente dal tipo di sorgente/destinzaione dati.

In pratica uno stream è un emettitore di eventi, che implementa particolati metodi: a seconda dei particolari metodi implementati si parla di Readables Stream, Writable Streams e Duplex

Readable Streams
Uno stream di lettura ci permette di leggere dati da una sorgente. La sorgente puo' essere qualsiasi cosa: un file sul file system, un buffer in memoria o ancora un altro stream.
Per leggere da uno stream basta mettersi in ascolto dell'evento data emesso dai stream di tipo readable.

 var fs = require('fs');  
 var streamLettura = fs.createReadStream('file.txt');  
 var data = '';  
 
 streamLettura.setEncoding('utf-8');
 
 streamLettura.on('data', function(chunk) {  
   data+=chunk;  
 });  
    
 streamLettura.on('end', function() {  
   console.log(data);  
 });  

In questo caso, la sorgente dati è costituito da un file.

Inizialmente lo stream è in uno stato statico ma non appena attacchiamo una funzione di callback all'evento data i dati cominciano a fluire fino a che terminano e lo stream emette l'evento end che possiamo intercettare con un'altra funzione di callback.

Se questo codice vi ricorda la soluzione a questo esercizio, siete sulla buona strada per capire la flessibilità di questo meccanismo: indipendentemente dal tipo di sorgente dati, la gestione dei dati viene sempre fatta allo stesso modo; l'oggetto response, in questo caso, è uno stream readable la cui sorgente dati è un socket di rete.

Di default, i dati letti da uno stream sono un oggetto di tipo Buffer, impostando il corretto encoding (ad es. utf-8), i dati vengono passati alla funzione di callback in formato stringa.

Writable Streams
Uno stream di scrittura ci permette si scrivere dati su una destinazione. Supponiamo di voler copiare un contenuto di un file in un altro file

 var fs = require('fs');  
 var readableStream = fs.createReadStream('file.txt');  
 var writableStream = fs.createWriteStream('file2.txt');  
    
 readableStream.setEncoding('utf8');  
    
 readableStream.on('data', function(chunk) {  
   writableStream.write(chunk);  
 });  

Semplice e intutitivo.

Il metodo write ritorna true o false a seconda dell'effettiva elaborazione dei dati in scrittura, se ritorna false, significa che dati aggiuntivi vengono memorizzati in memoria fino a che lo stream di scrittura non diventa di nuovo pronto: uno stream di scrittura emette l'evento drain per segnalare che è di nuovo pronto ad accettare dati senza bufferizzarli.

Se abbiamo dunque uno stream di lettura veloce ed uno stream di scrittura lento, la soluzione vista sopra potrebbe non essere efficiente perchè i dati emessi dalla sorgente verrebbero parcheggiati in un buffer in memoria, rallentando l'elaborazione dei dati; per rendere efficiente una elaborazione di questo tipo si utilizzano le pipe

Pipe
Vedetelo come un condotto che unisce sorgente e destinazione, che si preoccupa di far arrivare i dati a destinazione quando è effetivamente pronta a riceverli: anche in questo caso l'utilizzo è semplice e intuitivo

 var fs = require('fs');  
 var readableStream = fs.createReadStream('file1.txt');  
 var writableStream = fs.createWriteStream('file2.txt');  
    
 readableStream.pipe(writableStream);  

L'esempio copia i dati di file1.txt in file file2.txt, creandolo se non esiste.

Duplex
Sono stream che si comportano sia some sorgente che destinazione di dati: pensate ad esempio ad un socket di connessione TCP.
Immaginate un servizio in ascolto su una porta che gestisce un socket che è fullduplex: può ricevere dati e mandare dati sul socket. Sembra comlesso vero? Vediamo come implementare un server TCP di questo tipo con node.js

 var net = require('net');  
 var server = net.createServer(function(socket) {  
    console.log('server connesso');  
    socket.on('end', function() {  
       console.log('server disconnected');  
    });  
    socket.write('Benvenuto dal server eco ;))\r\n');  
    socket.on('data', function(data) {  
       console.log('ricevuti ' + data);
    });  
    socket.pipe(socket);  
 });  

 server.listen(8000, function() {  
    console.log('server in esecuzione su porta 8000');  
 });  
   

Per implementare un server TCP abbiamo bisogno del modulo net.

Creiamo un server con net.createServer() fornendo la funzione di callback che verrà eseguita ogni volta che un client si connette al nostro server.

Con il metodo listen mettiamo in ascolto il servizio sulla porta specificata come argomento.

La funzione di callback riceve un oggetto di tipo stream duplex su cui il server puo' leggere e/o scrivere i dati, secondo il protocollo che si vuole mettere in piedi.

In questo caso, per ogni connessione effettuata da un client, il server saluta e aspetta dati dal client che vengono immediatamente ritornati al client stesso tramite l'operazione di pipe effettuata da sorgente (socket) e destinazione (sempre socket!).

Per testare l'applicazione, eseguite il programma con node programma.js, il server si metterà in ascolto sulla porta 8000, per terminare il server digitare ctrl + c nella console.

Aprite un'altra console e digitate telnet localhost 8000, sulla console apparirà il saluto del server, se provate a digitare un qualsiasi carattere, vi verrà immediatamente rispedito indietro dal server che traccerà sulla sua console i dati elaborati, grazie alla gestione dell'evento data sul socket sorgente.

Nel prossimo post vedremo come implementare un server http, in pochi passi saremo in grado di implementare complesse applicazioni web con javascript ;)

Ho aggiornato su github il progetto, effettuate un git pull per aggiornare i sorgenti che avete in locale, come mostrato in questo post

Come al solito, per dubbi e/o domande: commentate!

Alla prox.
Ivan