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

Esercitazioni su node.js: esercizio 10

Testo dell'esercizio
Ancora un esercizio sull'uso degli stream associati a richieste e risposte http.
Creare un server http che riceva, in streaming sulla richiesta, un file da salvare nella directory corrente di lavoro, rappresentata dalla variabile globale __dirname.
Creare un client http che effettui una richiesta POST al server, mandando in streaming sulla richiesta il contenuto di un file di testo.

Suggerimenti
Trovate una bozza di soluzione a questo indirizzo: si tratta di implementare due programmi, un client http ed un server http. 
Il client crea un oggetto richiesta, passando opportuni opzioni di configurazione (indirizzo server, porta) e manderà in streaming sulla richiesta un file di testo tramite una pipe, chiudendo la richiesta quando la pipe ha finito di travasare i dati dal file alla richiesta stessa.
Il server, dualmente, legge in streaming la richiesta e scriverà i dati ricevuti su uno stream di scrittura associato ad un file, sempre tramite una pipe: appena finito chiuderà la risposta.

Buon Lavoro!

Una possibile soluzione
Vediamo prima il codice del programma client

 var http = require('http'),   
     fs = require('fs');  
   
 var options = {  
    host: 'localhost',   
    port: 8000, path: '/',   
    method: 'POST'  
 };  
   
 var req = http.request(options, function(res) {  
    console.log(res.statusCode);  
 });  
   
 var readStream = fs.ReadStream(__dirname + '/in.txt');  
        
      readStream.pipe(req);  
   
      readStream.on('close', function () {  
            req.end();  
            console.log('Io ho terminato di mandare il file al server...');  
      });
      console.log('Resto in attesa di chiusura connessione da parte del server ...');

Il codice è molto pulito e intuitivo, a patto di ricordarsi dell'asincronicità di alcune funzioni.
Per fare una richiesta http ad un server, viene creato un oggetto options con i dettagli di conettività: serve a configurare una richiest http di tipo post al server in esecuzione su localhost e porta 8000.

Viene utilizzato il modulo http per fare la richiesta. Il metodo request ritona subito uno stream di scrittura: tutto cio' che mandiamo su questo stream arriverà al server: la funzione di callback passata al metodo request viene eseguita solo quando il server finisce di rispondere.

Vine creato uno stream di lettura per il file in.txt presente nella cartella di lavoro, rappresentata dalla variabile globale __dirname: tramite una pipe viene scritto il suo contenuto sullo stream di scrittura req.

Quando finisce il flusso di lettura, viene invocato il metodo end() dello stream per permettere al server di elaborare la richiesta.

Vediamo ora la parte server

 var http = require('http'),   
     fs = require('fs'),  
     writeStream;  
   
 var server = http.createServer(function (req, res) {  
    writeStream = fs.createWriteStream(__dirname +   
                         'out_' +   
                         new Date().getTime() +  
                         '.txt');  
    req.pipe(writeStream);  
    req.on('end', function() {  
       writeStream.end();  
       res.statusCode = 200;  
       res.end('OK');  
    });  
 });  
   
 server.listen(8000, function() {  
      console.log('server http in ascolto sulla porta 8000');  
 });  

In maniera duale, per ogni richiesta che arriva al server, viene creato uno stream di scrittura associato ad un file presente in __dirname e con nome out_msda1/1/1970.txt su cui viene mandata in pipe il flusso di lettura rappresentato dalla richiesta.

Quando il flusso della richiesta è terminato, viene chiuso il flusso di scrittura del file e terminata la richiesta, con un codice standard di 200 (così come definito dal protocollo http) che causerà la chiusura della connessione.

Per testare il programma, mandate prima in esecuzione il server e poi il client da due shell diverse, magari provate a mandare in esecuzione, sempre su shell diverse, più programmi client,

Notate come, nonostante la funzione di callback sul client venga eseguita abbastanza in fretta, il server ci mette un po' a chiudere la connessione e quindi a far terminare il client: credo che sia una anomalia che dovrebbe essere messa a posto al più presto.

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

Ancora su modulo http in node.js

In questo post, traduco alcuni passaggi di un bell'articolo raggiungibile a questo indirizzo.

Scalabilità, big data, real time: queste le scommesse che node.js vuole vincere rispetto alla concorrenza grazie al suo modello di input/output non bloccante.

A patto di saper usare correttamente gli stream ...

Implementiamo un semplice server web che fornisca, ad ogni richiesta, il contenuto di un file statico presente sul server

 var http = require('http'),   
    fs = require('fs');  
    
 var server = http.createServer(function (req, res) {  
       fs.readFile(__dirname + '/data.txt', function (err, data) {  
       res.end(data);  
    });  
 });  
    
 server.listen(8000);  

La soluzione presentata funzione alla perfezione, collegatevi con un browser all'indirizzo localhost:8080, dopo aver mandato in esecuzione il programma, e riceverete il contenuto del file data.txt presente nella directory di lavoro, rappresentata dalla variabile globale __dirname.

Il programma legge il file in modalità asincrona e poi esegue la funzione di callback che ne scrive il contenuto sullo stream associato alla risposta: peccato che tutto il file venga prima caricato in memoria; se arrivano n richieste contemporanee  (con n speriamo grande :)), il nostro server si troverà seduto in men che non si dica.

Per rendere il nostro server scalabile, dobbiamo utilizzare un altro approccio, basato sul corretto utilizzo di stream e pipe

 var http = require('http'),   
     fs = require('fs');  
    
 var server = http.createServer(function (req, res) {  
        var stream = fs.createReadStream(__dirname + '/data.txt');  
        stream.pipe(res);  
     });  
 server.listen(8000);  

Creaiamo uno stream in lettura associato al file come sorgente dati e lo mandiamo sullo stream della risposta tramite la pipe.

Con questa soluzione, i dati verranno mandati al server, in blocchi, non appena disponibili, senza necessità di bufferizzazione sul server.

Con i prossimi post, vedremo come implementare un server web per creare la nostra prima applicazione web dinamica.

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

Esercitazioni su node.js: esercizio 9

Testo dell'esercizio
Ancora più semplice.
Scrivere un programma che scriva sulla console ciò che digitate dalla tastiera.
Vincoli: utilizzare solo stream e pipe.

Suggerimento
process.stdin è un readable stream ...dunque? Per interrompere il programma ctrl + c ;)

Buon Lavoro!

Una possibile soluzione

 process.stdin.pipe(process.stdout);  

Mandiamo su standard output, tramite pipe, ciò che arriva da standard input.

Alla prox.
Ivan

Esercitazioni su node.js: esercizio 8

Testo dell'esercizio
Bene, prima di continuare, vi propongo alcuni esercizi sugli stream e pipe.
Scrivere un programma che scriva sulla console il contenuto di un file di testo passato come argomento sulla linea di comando.
Vincoli: utilizzare solo stream e pipe.

Buon Lavoro!

Suggerimento
process.stdout è un writable stream su cui collegare, tramite pipe, un readable stream associato al file e process.stdout è lo standard output associato all'esecuzione del programma...

Una possibile soluzione

 var fs = require('fs'),  
     file = process.argv[2];  
   
 fs.createReadStream(file).pipe(process.stdout);  

Semplice e conciso: creo uno stream di lettura sul file e lo pippo sullo standard output :))

Alla prox.
Ivan