Fixed some end of stream issue.

demo/streams-test.js

@@ -3,43 +3,49 @@ var sys=require("sys");
 var posix=require("fs");
 var Buffer = require('buffer').Buffer;
 
-sys.puts('Compressing...');
-var gzdata = '';
-var gzip=new compress.GzipStream();
-gzip.setInputEncoding('utf8');
-gzip.setEncoding('binary');
-gzip.addListener('data', function(data) {
-  gzdata += data;
-}).addListener('error', function(err) {
-  throw err;
-}).addListener('end', function() {
-  sys.puts('Compressed length: ' + gzdata.length);
-});
+function doCompress(continuation) {
+  sys.puts('Making compression requests...');
+  var gzdata = '';
+  var gzip=new compress.GzipStream();
+  gzip.setInputEncoding('utf8');
+  gzip.setEncoding('binary');
+  gzip.addListener('data', function(data) {
+    gzdata += data;
+  }).addListener('error', function(err) {
+    throw err;
+  }).addListener('end', function() {
+    sys.puts('Compressed length: ' + gzdata.length);
+    continuation(gzdata);
+  });
 
-gzip.write("My data that needs ");
-gzip.write("to be compressed. 01234567890.");
-gzip.close();
-sys.puts('Compression finished.');
+  gzip.write("My data that needs ");
+  gzip.write("to be compressed. 01234567890.");
+  gzip.close();
+  sys.puts('Requests done.');
+}
 
-var d1 = gzdata.substr(0, 25);
-var d2 = gzdata.substr(25);
+function doDecompress(gzdata) {
+  var d1 = gzdata.substr(0, 25);
+  var d2 = gzdata.substr(25);
 
+  sys.puts('Making decompression requests...');
+  var rawdata = '';
+  var gunzip = new compress.GunzipStream();
+  gunzip.setInputEncoding('binary');
+  gunzip.setEncoding('utf8');
+  gunzip.addListener('data', function(data) {
+    rawdata += data;
+  }).addListener('error', function(err) {
+    throw err;
+  }).addListener('end', function() {
+    sys.puts('Decompressed length: ' + rawdata.length);
+    sys.puts('Raw data: ' + rawdata);
+  });
+  gunzip.write(d1);
+  gunzip.write(d2);
+  gunzip.close();
+  sys.puts('Requests done.');
+}
 
-sys.puts('Decompressing...');
-var rawdata = '';
-var gunzip = new compress.GunzipStream();
-gunzip.setInputEncoding('binary');
-gunzip.setEncoding('utf8');
-gunzip.addListener('data', function(data) {
-  rawdata += data;
-}).addListener('error', function(err) {
-  throw err;
-}).addListener('end', function() {
-  sys.puts('Decompressed length: ' + rawdata.length);
-  sys.puts('Raw data: ' + rawdata);
-});
-gunzip.write(d1);
-gunzip.write(d2);
-gunzip.close();
-sys.puts('Decompression finished.');
+doCompress(doDecompress);
 

lib/compress.js

@@ -108,11 +108,18 @@ CommonStream.prototype.emitData_ = function(data) {
     this.dataQueue_.push(data);
   } else {
     for (var i = 0; i < this.dataQueue_.length; ++i) {
-      this.emit('data', this.dataQueue_[i]);
+      var data = this.dataQueue_[i];
+      if (data !== null) {
+        this.emit('data', data);
+      } else {
+        this.emit('end');
+      }
     }
     this.dataQueue_.length = 0;
-    if (data != null) {
+    if (data !== null) {
       this.emit('data', data);
+    } else {
+      this.emit('end');
     }
   }
 };
@@ -138,7 +145,7 @@ CommonStream.prototype.emitEvent_ = function(err, data, fin) {
   this.emitData_(data);
 
   if (fin) {
-    this.emit('end');
+    this.emitData_(null);
   }
 };