Split off Supply.rotor from Supply.buffering

src/core/Supply.pm

@@ -89,8 +89,11 @@ my role Supply {
     method map(&mapper)        { SupplyOperations.map(self, &mapper) }
     method uniq(:&as,:&with)   { SupplyOperations.uniq(self, :&as, :&with) }
     method squish(:&as,:&with) { SupplyOperations.squish(self, :&as, :&with) }
-    method buffering( :$elems, :$overlap, :$seconds ) {
-        SupplyOperations.buffering( self, :$elems, :$overlap, :$seconds)
+    method rotor( $elems?, $overlap? ) {
+        SupplyOperations.rotor(self, $elems, $overlap)
+    }
+    method buffering( :$elems, :$seconds ) {
+        SupplyOperations.buffering( self, :$elems, :$seconds)
     }
     method merge(*@s)          { SupplyOperations.merge(self, @s) }
     method zip(*@s,:&with)     { SupplyOperations.zip(self, @s, :&with) }

src/core/SupplyOperations.pm

@@ -240,20 +240,54 @@ my class SupplyOperations is repr('Uninstantiable') {
         }
         MapSupply.new(:source($a), :&mapper)
     }
-
-    method buffering(Supply $s, :$elems, :$overlap; :$seconds ) {
 
-        return $s if !$elems and !$overlap and !$seconds;  # nothing to do
-        die "cannot have :overlap ($overlap) and :seconds ($seconds)"
-          if $overlap and $seconds;
+    method rotor(Supply $s, $elems is copy, $overlap is copy ) {
 
-        my class BufferingSupply does Supply does PrivatePublishing {
+        $elems   //= 2;
+        $overlap //= 1;
+        return $s if $elems == 1 and $overlap == 0;  # nothing to do
+
+        my class RotorSupply does Supply does PrivatePublishing {
             has $!source;
             has $.elems;
             has $.overlap;
+
+            submethod BUILD(:$!source, :$!elems, :$!overlap) { }
+
+            method tap(|c) {
+                my $tap = self.Supply::tap(|c);
+
+                my @buffered;
+                sub flush {
+                    self!more([@buffered]);
+                    @buffered.splice( 0, +@buffered - $!overlap );
+                }
+
+                $!source.tap( -> \val {
+                      @buffered.push: val;
+                      flush if @buffered.elems == $!elems;
+                  },
+                  done => {
+                      flush if @buffered;
+                      self!done();
+                  },
+                  quit => -> $ex { self!quit($ex) });
+                $tap
+            }
+        }
+        RotorSupply.new(:source($s), :$elems, :$overlap)
+    }
+
+    method buffering(Supply $s, :$elems, :$seconds ) {
+
+        return $s if (!$elems or $elems == 1) and !$seconds;  # nothing to do
+
+        my class BufferingSupply does Supply does PrivatePublishing {
+            has $!source;
+            has $.elems;
             has $.seconds;
 
-            submethod BUILD(:$!source, :$!elems, :$!overlap, :$!seconds) { }
+            submethod BUILD(:$!source, :$!elems, :$!seconds) { }
 
             method tap(|c) {
                 my $tap = self.Supply::tap(|c);
@@ -262,12 +296,7 @@ my class SupplyOperations is repr('Uninstantiable') {
                 my $last_time;
                 sub flush {
                     self!more([@buffered]);
-                    if $!overlap {
-                        @buffered.splice( 0, +@buffered - $!overlap );
-                    }
-                    else {
-                        @buffered = ();
-                    }
+                    @buffered = ();
                 }
 
                 my &more = do {
@@ -314,7 +343,7 @@ my class SupplyOperations is repr('Uninstantiable') {
                 $tap
             }
         }
-        BufferingSupply.new(:source($s), :$elems, :$overlap, :$seconds)
+        BufferingSupply.new(:source($s), :$elems, :$seconds)
     }
 
     method merge(*@s) {