Implement :overlap in Supply.buffering

src/core/Supply.pm

@@ -89,8 +89,8 @@ 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,:$seconds) {
-        SupplyOperations.buffering( self, :$elems, :$seconds)
+    method buffering( :$elems, :$overlap, :$seconds ) {
+        SupplyOperations.buffering( self, :$elems, :$overlap, :$seconds)
     }
     method merge(*@s)          { SupplyOperations.merge(self, @s) }
     method zip(*@s,:&with)     { SupplyOperations.zip(self, @s, :&with) }

src/core/SupplyOperations.pm

@@ -241,16 +241,19 @@ my class SupplyOperations is repr('Uninstantiable') {
         MapSupply.new(:source($a), :&mapper)
     }
 
-    method buffering(Supply $s, :$elems, :$seconds ) {
+    method buffering(Supply $s, :$elems, :$overlap; :$seconds ) {
 
-        return $s if !$elems and !$seconds;  # nothing to do
+        return $s if !$elems and !$overlap and !$seconds;  # nothing to do
+        die "cannot have :overlap ($overlap) and :seconds ($seconds)"
+          if $overlap and $seconds;
 
         my class BufferingSupply does Supply does PrivatePublishing {
             has $!source;
             has $.elems;
+            has $.overlap;
             has $.seconds;
 
-            submethod BUILD(:$!source, :$!elems, :$!seconds) { }
+            submethod BUILD(:$!source, :$!elems, :$!overlap, :$!seconds) { }
 
             method tap(|c) {
                 my $tap = self.Supply::tap(|c);
@@ -259,7 +262,12 @@ my class SupplyOperations is repr('Uninstantiable') {
                 my $last_time;
                 sub flush {
                     self!more([@buffered]);
-                    @buffered = ();
+                    if $!overlap {
+                        @buffered.splice( 0, +@buffered - $!overlap );
+                    }
+                    else {
+                        @buffered = ();
+                    }
                 }
 
                 my &more = do {
@@ -306,7 +314,7 @@ my class SupplyOperations is repr('Uninstantiable') {
                 $tap
             }
         }
-        BufferingSupply.new(:source($s), :$elems, :$seconds)
+        BufferingSupply.new(:source($s), :$elems, :$overlap, :$seconds)
     }
 
     method merge(*@s) {