local-lib5/lib/perl5/Data/Stream/Bulk.pm

   1 #!/usr/bin/perl
   2
   3 package Data::Stream::Bulk;
   4 use Moose::Role;
   5
   6 use namespace::clean -except => 'meta';
   7
   8 our $VERSION = "0.07";
   9
  10 requires qw(next is_done);
  11
  12 sub items {
  13         my $self = shift;
  14
  15         if ( my $a = $self->next ) {
  16                 return @$a;
  17         } else {
  18                 return ();
  19         }
  20 }
  21
  22 sub all {
  23         my $self = shift;
  24
  25         my @ret;
  26
  27         while ( my $next = $self->next ) {
  28                 push @ret, @$next;
  29         }
  30
  31         return @ret;
  32 }
  33
  34 sub cat {
  35         my ( $self, @streams ) = @_;
  36
  37         return $self unless @streams;
  38
  39         my @cat = $self->list_cat(@streams);
  40
  41         unless ( @cat ) {
  42                 return Data::Stream::Bulk::Nil->new;
  43         } elsif ( @cat == 1 ) {
  44                 return $cat[0];
  45         } else {
  46                 return Data::Stream::Bulk::Cat->new(
  47                         streams => \@cat,
  48                 );
  49         }
  50 }
  51
  52 sub list_cat {
  53         my ( $self, $head, @tail ) = @_;
  54
  55         return $self unless $head;
  56         return ( $self, $head->list_cat(@tail) );
  57 }
  58
  59 sub filter {
  60         my ( $self, $filter ) = @_;
  61
  62         return Data::Stream::Bulk::Filter->new(
  63                 filter => $filter,
  64                 stream => $self,
  65         );
  66 }
  67
  68 sub loaded { 0 }
  69
  70 # load it *after* the entire role is defined
  71 require Data::Stream::Bulk::Cat;
  72 require Data::Stream::Bulk::Nil;
  73 require Data::Stream::Bulk::Filter;
  74
  75 __PACKAGE__
  76
  77 __END__
  78
  79 =pod
  80
  81 =head1 NAME
  82
  83 Data::Stream::Bulk - N at a time iteration API
  84
  85 =head1 SYNOPSIS
  86
  87         # get a bulk stream from somewere
  88         my $s = Data::Stream::Bulk::Foo->new( ... );
  89
  90         # can be used like this:
  91         until ( $s->is_done ) {
  92                 foreach my $item ( $s->items ) {
  93                         process($item);
  94                 }
  95         }
  96
  97         # or like this:
  98         while( my $block = $s->next ) {
  99                 foreach my $item ( @$block ) {
 100                         process($item);
 101                 }
 102         }
 103
 104 =head1 DESCRIPTION
 105
 106 This module tries to find middle ground between one at a time and all at once
 107 processing of data sets.
 108
 109 The purpose of this module is to avoid the overhead of implementing an
 110 iterative api when this isn't necessary, without breaking forward
 111 compatibility in case that becomes necessary later on.
 112
 113 The API optimizes for when a data set typically fits in memory and is returned
 114 as an array, but the consumer cannot assume that the data set is bounded.
 115
 116 The API is destructive in order to minimize the chance that resultsets are
 117 leaked due to improper usage.
 118
 119 =head1 API
 120
 121 =head2 Required Methods
 122
 123 The API requires two methods to be implemented:
 124
 125 =over 4
 126
 127 =item is_done
 128
 129 Should return true if the stream is exhausted.
 130
 131 As long as this method returns a false value (not done) C<next> could
 132 potentially return another block.
 133
 134 =item next
 135
 136 Returns the next block.
 137
 138 Note that C<next> is not guaranteed to return an array reference, even if
 139 C<is_done> returned false prior to calling it.
 140
 141 =back
 142
 143 =head2 Convenience Methods
 144
 145 =over 4
 146
 147 =item items
 148
 149 This method calls C<next> and dereferences the result if there are pending
 150 items.
 151
 152 =item all
 153
 154 Force evaluation of the entire resultset.
 155
 156 Note that for large data sets this might cause swap thrashing of various other
 157 undesired effects. Use with caution.
 158
 159 =item cat @streams
 160
 161 Concatenates this stream with @streams, returning a single stream.
 162
 163 =item list_cat @tail
 164
 165 Returns a possibly cleaned up list of streams.
 166
 167 Used by C<cat>.
 168
 169 Overridden by L<Data::Stream::Bulk::Array>, L<Data::Stream::Bulk::Cat> and
 170 L<Data::Stream::Bulk::Nil> to implement some simple short circuiting.
 171
 172 =item filter $filter
 173
 174 Applies a per-block block filter to the stream.
 175
 176 Returns a possibly new stream with the filtering layered.
 177
 178 C<$filter> is invoked once per block and should return an array reference to
 179 the filtered block.
 180
 181 =item loaded
 182
 183 Should be overridden to return true if all the items are already realized (e.g.
 184 in the case of L<Data::Stream::Bulk::Array>).
 185
 186 Returns false by default.
 187
 188 When true calling C<all> is supposed to be safe (memory usage should be in the
 189 same order of magnitude as stream's own usage).
 190
 191 This is typically useful when tranforming an array is easier than transorming a
 192 stream (e.g. optional duplicate filtering).
 193
 194 =back
 195
 196 =head1 CLASSES
 197
 198 =over 4
 199
 200 =item L<Data::Stream::Bulk::Array>
 201
 202 This class is not a stream at all, but just one block. When the data set easily
 203 fits in memory this class can be used, while retaining forward compatibility
 204 with larger data sets.
 205
 206 =item L<Data::Stream::Bulk::Callback>
 207
 208 Callback driven iteration.
 209
 210 =item L<Data::Stream::Bulk::DBI>
 211
 212 Bulk fetching of data from L<DBI> statement handles.
 213
 214 =item L<Data::Stream::Bulk::DBIC>
 215
 216 L<DBIx::Class::ResultSet> iteration.
 217
 218 =item L<Data::Stream::Bulk::Nil>
 219
 220 An empty result set.
 221
 222 =item L<Data::Stream::Bulk::Cat>
 223
 224 A concatenation of several streams.
 225
 226 =item L<Data::Stream::Bulk::Filter>
 227
 228 A filter wrapping a stream.
 229
 230 =back
 231
 232 =head1 SEE ALSO
 233
 234 L<HOP::Stream>, L<Iterator>, L<Class::Iterator> etc for one by one iteration
 235
 236 L<DBI>, L<DBIx::Class::ResultSet>
 237
 238 L<POE::Filter>
 239
 240 L<Data::Page>
 241
 242 L<Parallel::Iterator>
 243
 244 L<http://en.wikipedia.org/wiki/MapReduce>, LISP, and all that other kool aid
 245
 246 =head1 TODO
 247
 248 =over 4
 249
 250 =item Sorted streams
 251
 252 Add a hint for sorted streams (like C<loaded> but as an attribute in the base
 253 role).
 254
 255 Introduce a C<merge> operation for merging of sorted streams.
 256
 257 Optimize C<unique> to make use of sorting hints for constant space uniquing.
 258
 259 =item More utility functions
 260
 261 To assist in proccessing and creating streams.
 262
 263 =item Coercion tables
 264
 265 L<Moose::Util::TypeConstraints>
 266
 267 =back
 268
 269 =head1 VERSION CONTROL
 270
 271 This module is maintained using git. You can get the latest version from
 272 L<http://github.com/nothingmuch/data-stream-bulk/>.
 273
 274 =head1 AUTHOR
 275
 276 Yuval Kogman E<lt>nothingmuch@woobling.orgE<gt>
 277
 278 =head1 COPYRIGHT
 279
 280         Copyright (c) 2008 Yuval Kogman. All rights reserved
 281         This program is free software; you can redistribute
 282         it and/or modify it under the same terms as Perl itself.
 283
 284 =cut
 285