add unique()

lib/Data/Stream/Bulk/Util.pm

@@ -8,10 +8,12 @@ use warnings;
 use Data::Stream::Bulk::Nil;
 use Data::Stream::Bulk::Array;
 
+use Scalar::Util qw(refaddr);
+
 use namespace::clean;
 
 use Sub::Exporter -setup => {
-	exports => [qw(nil bulk cat filter)],
+	exports => [qw(nil bulk cat filter unique)],
 };
 
 sub nil () { Data::Stream::Bulk::Nil->new }
@@ -25,6 +27,11 @@ sub filter (&$) {
 	$stream->filter($filter);
 }
 
+sub unique ($) {
+	my %seen;
+	shift->filter(sub { [ grep { !$seen{ref($_) ? refaddr($_) : $_}++ } @$_ ] }); # FIXME Hash::Util::FieldHash::Compat::id()?
+}
+
 __PACKAGE__
 
 __END__
@@ -78,6 +85,15 @@ Returns C<nil> if no arguments are provided.
 
 Calls C<filter> on $stream with the provided filter.
 
+=item unique $stream
+
+Filter the stream to remove duplicates.
+
+Note that this may potentially scales to O(k) where k is the number of distinct
+items.
+
+In the future this will be optimized to be iterative for sorted streams.
+
 =back
 
 =cut