lire 2:2.1.1-2.1 / usr / share / perl5 / Lire / DlfStream.pm

package Lire::DlfStream;

use strict;

use Lire::DlfSchema;
use Lire::DlfQuery;
use Lire::I18N qw/ensure_utf8 /;
use Lire::Utils qw/ check_param check_object_param /;

use Carp;

=pod

=head1 NAME

Lire::DlfStream - Interface to DLF data stream

=head1 SYNOPSIS

  use Lire::DlfStore;

  my $store = Lire::DlfStore->open( "mystore" );
  my $dlf_stream = $store->open_dlf_stream( "www", "r" );

  print "Data begins on ", scalar localtime $dlf_stream->start_time(), "\n";
  print "Data ends on   ", scalar localtime $dlf_stream->end_time(), "\n";
  while ( my $dlf = $dlf_stream->read_dlf() ) {
    ...
  }

=head1 DESCRIPTION

This object encapsulates DLF stream.

=cut

#------------------------------------------------------------------------
# Constructor new( $store, $name, $mode )
#
# Used by Lire::DlfStore
sub new {
    my ( $pkg, $store, $name, $mode, $sort_spec ) = @_;

    check_object_param( $store, 'store', 'Lire::DlfStore' );
    check_param( $mode, 'mode', sub { $mode eq 'r' ||
                                      $mode eq 'w' },
                 "'mode' should be one of 'r' or 'w'" );

    croak "only 'r' mode stream can use a sort_spec"
      if $mode ne 'r' && $sort_spec;

    my $self = bless { '_store'     => $store,
                       '_name'      => $name,
                       '_mode'      => $mode,
                       '_sort_spec' => $sort_spec,
                       '_schema'    => Lire::DlfSchema::load_schema( $name ),
                     }, $pkg;

    if ( ! $self->{'_store'}->has_dlf_stream( $self->{'_name'} ) ) {
        $self->{'_schema'}->create_sql_schema( $self->{'_store'} );
    } elsif ( $self->{'_schema'}->needs_sql_schema_migration($self->{'_store'}))
    {
        $self->{'_schema'}->migrate_sql_schema( $self->{'_store'} );
    }

    if ( $mode eq "w" ) {
        my $sql = $self->{'_schema'}->sql_insert_query();
        $self->{'_sth'} = $self->{'_store'}->_dbh()->prepare_cached( $sql );
        if ( $self->{'_schema'}->isa( 'Lire::DerivedSchema' ) ) {
            $sql = $self->{'_schema'}->sql_link_insert_query();
            $self->{'_link_sth'} = $self->{'_store'}->_dbh()->prepare_cached( $sql );
        }
    } else {
        $self->{'_dlf_reader'} =
          $self->{'_schema'}->dlf_query( $sort_spec )->execute( $store );
    }

    return $self;
}

=pod

=head2 name

Returns the schema's name of the DlfStream.

=cut

sub name {
    $_[0]{'_name'};
}

=pod

=head2 mode()

Returns the mode in which the DlfStream was opened.

=cut

sub mode {
    $_[0]{'_mode'};
}

=pod

=head2 sort_spec()

Returns the sort specification that is set on the stream.

=cut

sub sort_spec {
    $_[0]{'_sort_spec'};
}

=pod

=head2 close()

This method should be called when the Lire::DlfStream isn't needed
anymore, otherwise the same DlfStream cannot be opened until then.

=cut

sub close {
    my $self = $_[0];

    if ( $self->{'_sth'} ) {
        $self->{'_sth'}->finish();
        delete $self->{'_sth'};
    } elsif ( $self->{'_dlf_reader'} ) {
        $self->{'_dlf_reader'}->dlf_query()->release();
        delete $self->{'_dlf_reader'};
    }
}

sub DESTROY {
    $_[0]->close();
}

=pod

=head2 nrecords()

Returs the number of DLF records in the stream.

=cut

sub nrecords {
    my $self = $_[0];
    return $self->_get_stat( "nrecords", "count()");
}

=pod

=head2 start_time()

Returns the timestamp of the oldest DLF record in the stream in
seconds since the epoch.

=cut

sub start_time {
    my $self = $_[0];

    my $field = $self->{'_schema'}->timestamp_field()->name();
    return $self->_get_stat( 'start_time',
                             sprintf( "min(%s)", $field ) );
}

=pod

=head2 end_time()

Returns the timestamp of the newest DLF record in the stream in
seconds since the epoch.

=cut

sub end_time {
    my $self = $_[0];

    my $field = $self->{'_schema'}->timestamp_field()->name();
    return $self->_get_stat( 'end_time', sprintf( "max(%s)", $field ) );
}

sub _get_stat {
    my ( $self, $name, $expr ) = @_;

    my $query = new Lire::DlfQuery( $self->{'_name'} );
    $query->add_aggr_field( $name, $expr );
    my $result = $query->execute( $self->{'_store'} );
    my $stat = $result->next_row()->{$name};
    $query->release();

    return $stat;
}

=pod

=head2 read_dlf()

Returns an hash reference containing the next DLF record in the
stream. It returns undef once the end of the stream is reached.

This method will throw an exception if the DlfStream isn't open in 'r'
mode or if there is an error reading the DLF record.

=cut
sub read_dlf {
    my $self = $_[0];

    croak "read_dlf() can't be called on a DlfStream open in 'w' mode"
      if $self->{'_mode'} ne 'r';

    return $self->{'_dlf_reader'}->next_row();
}

=pod

=head2 read_dlf_aref()

Returns the next DLF record in the stream as an array reference. The
fields are in the order specified by the schema.

This method will throw an exception if the DlfStream isn't open in 'r'
mode or if there is an error reading the DLF record.

=cut
sub read_dlf_aref {
    my $self = $_[0];

    croak "read_dlf_aref() can't be called on a DlfStream open in 'w' mode"
      if $self->{'_mode'} ne 'r';

    return $self->{'_dlf_reader'}->next_row_aref();
}

=pod

=head2 write_dlf( $dlf, [ $link_ids ] )

Writes the fields contained in the hash ref $dlf to the DLF stream.

This method will throw an exception if there is an error writing the
DLF record or if the stream isn't opened in 'w' mode.

The $link_ids parameter is used when the stream's schema is a
Lire::DerivedSchema. It should be an array reference containing the
DLF ids of the records which are linked to this record.

=cut

sub write_dlf {
    my ( $self, $dlf, $link_ids ) = @_;

    croak "write_dlf() can't be called on a DlfStream open in 'r' mode"
      if $self->{'_mode'} ne 'w';

    croak "missing DLF hash reference"
      unless defined $dlf;

    croak "DLF record isn't an hash reference: $dlf"
      unless ref $dlf eq 'HASH';

    croak "'dlf_id' field must be set when using write_dlf() on an extended schema"
      if ( $self->{'_schema'}->isa( 'Lire::ExtendedSchema' )
           && ! defined $dlf->{'dlf_id'} );

    croak "'link_ids' parameter should be an ARRAY reference, not '$link_ids'"
      if ( $self->{'_schema'}->isa( 'Lire::DerivedSchema' )
           && defined $link_ids && ref( $link_ids ) ne 'ARRAY' );

    my @values = ();
    foreach my $f ( map { $_->name() } $self->{'_schema'}->_sql_fields() ) {
        my $v = $dlf->{$f};
        #   replace control chars with ?
        $v =~ tr/\000-\037/?/
          if defined $v;
        push @values, ensure_utf8( $v );
    }
    $self->{'_sth'}->execute( @values);

    $self->_write_derived_links( $link_ids )
      if $self->{'_schema'}->isa( 'Lire::DerivedSchema' );

    return;
}

sub _write_derived_links {
    my ( $self, $link_ids ) = @_;

    $link_ids ||= [];
    return unless @$link_ids;

    my $src_id = $self->{'_store'}->_dbh()->selectrow_arrayref( 'SELECT last_insert_rowid()' );
    foreach my $link_id ( @$link_ids ) {
        $self->{'_link_sth'}->execute( $src_id->[0], $link_id);
    }
}

=pod

=head2 clean( [ $time ] )

This method will remove all DLF records older than $time. It $time is
omitted, all Dlf records will be removed.

=cut

sub clean {
    my ( $self, $time ) = @_;

    croak "clean() can't be called on a DlfStream open in 'r' mode"
      if $self->{'_mode'} ne 'w';

    my $sql = $self->{'_schema'}->sql_clean_query( defined $time );
    if ( defined $time ) {
        $self->{'_store'}->_dbh()->do( $sql, {}, $time );
    } else {
        $self->{'_store'}->_dbh()->do( $sql );
    }

    return;
}

1;