var nile = {};

nile.pipeline = function() {
  var k = arguments[arguments.length - 1];
  for (var i = arguments.length - 2; i >= 0; i--) {
    k = arguments[i](k);
  }
  return k;
};

nile.Echo = function(downstream) {
  return function(input) {
    downstream(input);
  };
};

nile.Interleave2 = function(k1, k2) {
  return function(downstream) {
    return function(input) {
      var out1;
      var out2;
      k1(function(input) { out1 = input })(input.slice(0));
      k2(function(input) { out2 = input })(input);
      var output = [];
      for (var i = 0; i < out1.length; i++)
        output.push(out1[i], out2[i]);
      downstream(output);
    };
  };
};

nile.GroupBy = function(index, size) {
  return function(downstream) {
    return function(input) {
      var buckets = {};
      for (var i = 0; i < input.length; i += size) {
        var v = input.slice(i, size);
        var key = v[index];
        var bucket = buckets[key];
        if (!bucket)
          bucket = buckets[key] = [];
        bucket.push(v);
      }
      for (bucket in buckets) {
        if (buckets.hasOwnProperty(bucket))
          downstream(buckets[bucket]);
      }
    };
  };
};

nile.SortBy = function(index, size) {
  return function(downstream) {
    return function(input) {
      var vectors = [];
      for (var i = 0; i < input.length; i += size)
        vectors.push(input.slice(i, size));
      vectors.sort(function(a, b) {
        if (a[index] < b[index])
          return -1;
        else if (a[index] > b[index])
          return 1;
        else
          return 0;
      });
      downstream(input);
    };
  };
};

@@ -855,17 +855,19 @@
 var 
-v
+key
  = input
 .sli
@@ -866,45 +866,13 @@
 nput
-.slice(i, size);%0A        var key = v%5B
+%5Bi + 
 inde
@@ -981,21 +981,73 @@
     
-bucket.push(v
+for (var j = 0; j %3C size; j++)%0A          bucket.push(input%5Bi + j%5D
 );%0A 
@@ -1601,16 +1601,163 @@
     %7D);%0A
+      var output = %5B%5D;%0A      for (var i = 0; i %3C vectors.length; i++)%0A        for (var j = 0; j %3C size; j++)%0A          output.push(vectors%5Bi%5D%5Bj%5D);%0A
       do
@@ -1757,34 +1757,35 @@
      downstream(
-in
+out
 put);%0A    %7D;%0A  %7D

Fix bug with SortBy and GroupBy where we forget to flatten the output stream.