# Mrc.pm: Class Used for Managing Multi-run Collections

package RDA::Object::Mrc;

# $Id: Mrc.pm,v 1.18 2015/05/08 18:18:13 RDA Exp $
# ARCS: $Header: /home/cvs/cvs/RDA_8/src/scripting/lib/RDA/Object/Mrc.pm,v 1.18 2015/05/08 18:18:13 RDA Exp $
#
# Change History
# 20150508  MSC  Improve the documentation.

=head1 NAME

RDA::Object::Mrc - Class Used for Managing Multi-run Collections

=head1 SYNOPSIS

 require RDA::Object::Mrc;

=head1 DESCRIPTION

The objects of the C<RDA::Object::Mrc> class are used to manage collections
performed in multiple runs. It is a subclass of L<RDA::Object|RDA::Object>.

The following methods are available:

=cut

use strict;

BEGIN
{ use Exporter;
  use IO::File;
  use RDA::Text qw(get_string);
  use RDA::Error;
  use RDA::Object;
  use RDA::Object::Content qw($RE_DC);
  use RDA::Object::Item;
  use RDA::Object::Rda;
  use RDA::Object::Toc;
  use RDA::SDCL::Block qw($CONT $RET_DIE $SPC_REF $SPC_VAL);
  use RDA::Value::Scalar qw(:value);
}

# Define the global public variables
use vars qw($STRINGS $VERSION @ISA %SDCL);
$VERSION = sprintf('%d.%02d', q$Revision: 1.18 $ =~ /(\d+)\.(\d+)/);
@ISA     = qw(RDA::Object Exporter);
%SDCL    = (
  als => {
    ## no critic (Interpolation)
    'beginCollect'      => ['$[MRC]', 'begin'],
    'endCollect'        => ['$[MRC]', 'end'],
    'getMrcCollections' => ['$[MRC]', 'get_collections'],
    'getMrcMembers'     => ['$[MRC]', 'get_members'],
    'getMrcModules'     => ['$[MRC]', 'get_modules'],
    'getMrcSets'        => ['$[MRC]', 'get_sets'],
    'isMrcActive'       => ['$[MRC]', 'is_active'],
    'logMrcSections'    => ['$[MRC]', 'log_sections'],
    'logMrcStatistics'  => ['$[MRC]', 'log_stats'],
    'validate'          => ['$[MRC]', 'validate'],
    },
  beg => \&_begin_mrc,
  cmd => {
    'collect'  => [\&_exe_collect, \&_get_collect, 0, 0],
    },
  dep => [qw(RDA::Object::Output)],
  end => \&_end_mrc,
  inc => [qw(RDA::Object)],
  met => {
    'begin'           => {ret => 0},
    'end'             => {ret => 0},
    'get_collections' => {ret => 1},
    'get_modules'     => {ret => 1},
    'get_info'        => {ret => 0},
    'get_members'     => {ret => 1},
    'get_sets'        => {ret => 1},
    'is_active'       => {ret => 0},
    'log_sections'    => {ret => 0},
    'log_stats'       => {ret => 0},
    'set_info'        => {ret => 0},
    'validate'        => {ret => 0},
    },
  top => 'MRC',
  );

# Define the global private constants
my $SET = qr/^(([A-Z][A-Z\d]+)\.)?(\w+)$/i;

my $RPT_EOT = "\n.N1\n";
my $RPT_LST = "    \001* ";
my $RPT_NXT = ".N1\n";
my $RPT_SUB = "    \001  ";
my $RPT_TXT = q{    };

# Define the global private variables

# Report the package version
sub Version
{ return $VERSION;
}

=head2 S<$h = RDA::Object::Mrc-E<gt>new($col,$pkg)>

The multi-run collection control object constructor. This method takes the
collector and package object references as arguments.

It is represented by a blessed hash reference. The following special keys are
used:

=over 12

=item S<    B<'agt' > > Reference to the agent object (G)

=item S<    B<'cfg' > > Reference to the RDA software configuration (G,L)

=item S<    B<'cnt' > > Reference to the RDA content control object (G,L)

=item S<    B<'col' > > Reference to the collector object (L)

=item S<    B<'nam' > > Module name (L)

=item S<    B<'oid' > > Object identifier (C,G,L)

=item S<    B<'out' > > Reference to the global reporting control object (L)

=item S<    B<'par' > > Reference to the control object (C,L)

=item S<    B<'pkg' > > Package reference (L)

=item S<    B<'rpt' > > Reference to the local reporting control object (L)

=item S<    B<'typ' > > Object type: C (Collection), G (Global), or L (Local)

=item S<    B<'_acc'> > Section acceptance indicator (L)

=item S<    B<'_bkp'> > Backup (L)

=item S<    B<'_chg'> > Setting changes hash (C)

=item S<    B<'_col'> > Collection definitions (L)

=item S<    B<'_cur'> > Current table of content record (L)

=item S<    B<'_def'> > Group definitions (G)

=item S<    B<'_dft'> > Default acceptance status (L)

=item S<    B<'_dir'> > Collect directory structure (G)

=item S<    B<'_exe'> > Section execution log indicator (L)

=item S<    B<'_lvl'> > Collection level (L)

=item S<    B<'_man'> > Collection manual hash (C)

=item S<    B<'_mod'> > Module list (C)

=item S<    B<'_prv'> > Previous run table of content (L)

=item S<    B<'_ref'> > Table of content record used as reference (L)

=item S<    B<'_sct'> > Name of the current active section (L)

=item S<    B<'_set'> > Set definitions (L)

=item S<    B<'_sta'> > Statistics log indicator (L)

=item S<    B<'_toc'> > Current run table of content (L)

=item S<    B<'_typ'> > User type (L)

=back

Internal keys are prefixed by an underscore.

=cut

sub new
{ my ($cls, $obj, $pkg) = @_;
  my ($def, $nam, $ref, $slf);

  # Create the object
  if ($ref = ref($cls))
  { $def = ref($pkg)
      ? [oid  => $pkg->get_oid,
         pkg  => $pkg,
         rpt  => $pkg->get_output]
      : [oid  => $cls->get_oid];
    $slf = bless {
      cfg  => $obj->get_config,
      cnt  => $obj->get_content,
      col  => $obj,
      out  => $obj->get_output,
      par  => $cls,
      typ  => 'L',
      _exe => 0,
      _lvl => 0,
      _sta => 0,
      _typ => 1,
      @{$def},
      }, $ref;
  }
  else
  { $ref = $obj->get_config;
    $slf = bless {
      agt  => $obj->get_agent,
      cfg  => $ref,
      cnt  => $obj->get_content,
      oid  => $obj->get_oid,
      typ  => 'G',
      _def => {},
      _dir => $ref->get_group('D_RDA_COL'),
      _lvl => $obj->get_level,
      }, $cls;
  }

  # Return the object reference
  return $slf;
}

=head2 S<$h-E<gt>delete_object>

This method deletes the object.

=cut

sub delete_object
{ if ($_[0]->{'typ'} eq 'C')
  { # Delete the definitions
    if (exists($_[0]->{'_def'}))
    { foreach my $def (values(%{$_[0]->{'_def'}}))
      { next unless ref($def);
        foreach my $col (values(%{$def->{'_col'}}))
        { $col->delete_object;
        }
        undef %{$def->{'_col'}};
        undef %{$def->{'_set'}};
        undef %{$def};
      }
    }
  }
  elsif ($_[0]->{'typ'} eq 'L')
  { # Restore any original context
    $_[0]->end;
  }

  # Delete the object itself
  $_[0]->SUPER::delete_object;
  return;
}

=head1 LOCAL CONTROL OBJECT METHODS

=head2 S<$h-E<gt>begin($oid[,$flag])>

This method ends any active collection context, saves the original context, and
starts a new collection context for the specified module. When the flag is set,
it assumes that the collections are performed as an user without super user
privileges.

It does not start a collection unless the base collection is already done and
returns -1.

It returns 0 on successful completion.

=cut

sub begin
{ my ($slf, $oid, $typ) = @_;
  my ($abr, $bkp, $col, $out, $pkg, $toc, $top);

  die get_string('NOT_LOCAL') unless $slf->{'typ'} eq 'L';
  die get_string('BAD_BEGIN') if $slf->{'_col'};

  # Restore any existing backup
  $slf->end;

  # Abort unless the base collection is done
  $col = $slf->{'col'};
  return -1 unless $col->is_done($oid);
  $slf->{'nam'} = $oid;

  # Define the user type
  $slf->{'_typ'} = $typ;

  # Load the table of content
  $abr = $col->get_first("STATUS.$oid.W_ABR");
  $slf->{'_prv'} = $slf->{'_ref'} =
  $slf->{'_toc'} = $slf->{'_cur'} = $slf->{'out'}->load_run($abr);

  # Define the new reporting context
  $slf->{'_bkp'} = $bkp = {};
  $pkg = $slf->{'pkg'};
  $top = $slf->get_top;

  $bkp->{'abr'} = $pkg->set_info('abr', $abr);
  $bkp->{'nam'} = $pkg->set_info('nam', $oid);
  $bkp->{'def'} = $pkg->set_info('def', $col->find('SETUP.'.$oid, 1));

  $slf->{'rpt'} = $out = $col->get_output->new($pkg);
  $bkp->{'OUT'} = $pkg->set_top('OUT', $out);
  unless ($typ)
  { $out->set_info('mrc', 1);

    $out->enable_index(1);
    $out->load_index($slf->{'_prv'}, 0);
    $out->purge('M', q{.}, -1);
  }

  $toc = RDA::Object::Toc->new($out);
  $out->set_info('toc', $toc);
  $bkp->{'TOC'} = $pkg->set_top('TOC', $out);

  # Initialize the usage counters
  $col->init_usage($oid);

  # Indicate the successful completion
  return 0;
}

=head2 S<$h-E<gt>end>

This method ends all operations in the current collection context and restores
the original context.

=cut

sub end
{ my ($slf) = @_;
  my ($bkp, $col, $oid, $pkg);

  die get_string('NOT_LOCAL') unless $slf->{'typ'} eq 'L';
  if (defined($bkp = delete($slf->{'_bkp'})))
  { # Generate the table of content
    $slf->{'rpt'}->save_toc($slf->{'_toc'});

    # Delete the current report control object
    $slf->{'rpt'}->delete_object;

    # Restore the original context
    $pkg = $slf->{'pkg'};
    $pkg->set_top('TOC', $bkp->{'TOC'});
    $pkg->set_top('OUT', $slf->{'rpt'} = $bkp->{'OUT'});
    $pkg->set_info('abr', $bkp->{'abr'});
    $pkg->set_info('def', $bkp->{'def'});
    $pkg->set_info('nam', $bkp->{'nam'});

    # Update status and usage
    $oid = delete($slf->{'nam'});
    $col = $slf->{'col'};
    $col->set_value("STATUS.$oid.G_MRC", time);
    $col->update_usage($oid, 1, 1);
    $col->save if $col->should_save;
  }
  return 0;
}

=head2 S<$h-E<gt>get_members($set[,$list[,$flag]]])>

This method returns the list of collection modules corresponding to the
specified collections. It uses C<default> as default group. An asterisk (C<*>)
represents all collections. When the flag is set, it keeps the prefix
associations. It applies the corresponding collection settings.

=cut

sub get_members  ## no critic (Complex)
{ my ($slf, $set, $lst, $flg) = @_;
  my ($cnt, $col, $def, $itm, $loc, $nam, $obj, $sel, $seq, $tbl,
      @lst, @grp, %mod);

  # Validate the set
  die get_string('NOT_LOCAL')     unless $slf->{'typ'} eq 'L';
  die get_string('NO_SET')        unless defined($set);
  die get_string('BAD_SET', $set) unless $set =~ m/^((\w+)\.)?(\w+)$/;
  $nam = lc($3);
  if ($1)
  { $loc = $2;
    $sel = $slf->{'par'}->select([$loc]);
    die get_string('UNKNOWN_SET', $set)
      unless exists($sel->{$loc}) && exists($sel->{$loc}->{'_set'}->{$nam});
    $def = $sel->{$loc}->{'_set'}->{$nam};
  }
  else
  { $def = _get_set($slf->{'par'}->select, $nam);
  }

  # Determine the collection list
  if (!defined($lst))
  { @lst = ('default') if exists($def->{'default'});
  }
  elsif ($lst eq q{*})
  { @lst = keys(%{$def});
  }
  elsif (ref($lst) eq 'ARRAY')
  { @lst = grep {exists($def->{$_})} @{$lst};
  }
  else
  { @lst = grep {exists($def->{$_})} split(/\|/, lc($lst));
  }

  # Analyze the collections
  $cnt = $slf->{'cnt'};
  $col = $slf->{'col'};
  $seq = $col->tie_value('STATUS.N_MRC', {});
  $itm = $col->get_info('set');
  foreach my $itm (@lst)
  { $obj = $def->{$itm};
    if (exists($obj->{'_mod'}))
    { foreach my $mod (@{$obj->{'_mod'}})
      { $mod{$flg ? $mod : $1} = exists($seq->{$nam})
          ? $seq->{$1}
          : _get_seq($cnt, $seq, $1)
          if $mod =~ m/^([^\-\|]+)/;
      }
    }
    if (exists($obj->{'_chg'}))
    { foreach my $key (keys(%{$tbl = $obj->{'_chg'}}))
      { if ($key =~ m/^(\w+)\/(.*)$/)
        { next if $1 eq 'PRF';
          $col->get_item($1)->set_temp($2, $tbl->{$key});
        }
        else
        { $itm->set_temp($key, $tbl->{$key});
        }
      }
    }
  }
  return (sort {$mod{$a} <=> $mod{$b} || $a cmp $b} keys(%mod));
}

# Get the collection sequence
sub _get_seq
{ my ($cnt, $seq, $nam) = @_;
  my ($val);

  ($val) = $cnt->get_sequence('MC', $nam);
  return $seq->{$nam} = $val;
}

# Retrieve the set description
sub _get_set
{ my ($sel, $nam) = @_;
  my ($tbl);

  foreach my $grp (q{}, '<ENV>', 'RDA')
  { return $sel->{$grp}->{'_set'}->{$nam}
      if exists($sel->{$grp}) && exists($sel->{$grp}->{'_set'}->{$nam});
  }
  foreach my $grp (keys(%{$sel}))
  { next unless exists($sel->{$grp}->{'_set'}->{$nam});
    die get_string('MORE_SETS', $nam) if ref($tbl);
    $tbl = $sel->{$grp}->{'_set'}->{$nam};
  }
  return $tbl;
}

=head2 S<$h-E<gt>get_modules>

This method returns the list of the modules where multi-run collections are
enabled.

=cut

sub get_modules
{ my ($slf) = @_;

  die get_string('NOT_LOCAL') unless $slf->{'typ'} eq 'L';
  return (sort $slf->{'col'}->get_info('sta')->grep('^B_MRC$', 'or'));
}

=head2 S<$h-E<gt>is_active>

This method indicates whether a completion collection is running.

=cut

sub is_active
{ my ($slf) = @_;

  die get_string('NOT_LOCAL') unless $slf->{'typ'} eq 'L';
  return $slf->{'_typ'} ? 0 : 1;
}

=head2 S<$h-E<gt>log_sections($flag[,$section...])>

This method specifies whether collect requests should report section execution
in the calling module. You can specify a list of optional sections, which are
not reported. It returns the previous status.

=cut

sub log_sections
{ my ($slf, $flg, @sct) = @_;

  die get_string('NOT_LOCAL') unless $slf->{'typ'} eq 'L';
  ($flg, $slf->{'_exe'}) =
    ($slf->{'_exe'} ? 1 : 0, $flg ? {map {$_ => 0} ('begin', @sct)} : 0);
  return $flg;
}

=head2 S<$h-E<gt>log_stats($flag)>

This method specifies whether collect requests should log statistics in the
event log. It returns the previous status.

=cut

sub log_stats
{ my ($slf, $flg) = @_;

  die get_string('NOT_LOCAL') unless $slf->{'typ'} eq 'L';
  ($flg, $slf->{'_sta'}) = ($slf->{'_sta'}, $flg);
  return $flg;
}

=head2 S<$h-E<gt>validate([$flag])>

This method accepts or rejects the current section. By default, it accepts the
section.

=cut

sub validate
{ my ($slf, $flg) = @_;

  die get_string('ERR_VALIDATE')
    unless exists($slf->{'_sct'});
  return $slf->{'_acc'} = defined($flg) ? $flg : 1;
}

=head1 GLOBAL CONTROL OBJECT METHODS

=head2 S<$h-E<gt>find_set($group,$set[,$default])>

This method verifies the existence pf the specified set and returns a
normalized set name. Unless you specify a default value as an extra argument,
it raises an error when it can not find the set.

=cut

sub find_set
{ my ($slf, $grp, $set, $dft) = @_;
  my ($nam, $sel);

  die get_string('NO_SET')        unless defined($set);
  die get_string('BAD_SET', $set) unless $set =~ $SET;

  $slf = $slf->get_top;
  if ($1)
  { $grp = uc($2);
    $nam = lc($3);
    $sel = $slf->select([$grp]);
    return "$grp.$nam"
      if exists($sel->{$grp}) && exists($sel->{$grp}->{'_set'}->{$nam});
  }
  else
  { $nam = lc($set);
    $sel = $slf->select($grp);
    foreach my $sub (sort keys(%{$sel}))
    { return "$sub.$nam" if exists($sel->{$sub}->{'_set'}->{$nam});
    }
  }
  die get_string('UNKNOWN_SET', $set) unless defined($dft);
  return $dft;
}

=head2 S<$h-E<gt>get_collection($name[,$group])>

This method returns the definition of the specified multi-run collection. In a
scalar context, it returns a reference to its definition. In an array context,
it returns a list containing a reference to its definition, the collection
name, and the collection name.

=cut

sub get_collection
{ my @grp = _get_collection(@_);

  return @grp if wantarray;
  return $grp[0];
}

sub _get_collection
{ my ($slf, $col, $grp) = @_;
  my ($def, $loc, $nam, $rec, $sel, @grp);

  $slf = $slf->get_top;
  $def = $slf->{'_def'};

  # Validate the collection name
  die get_string('BAD_COLLECTION', $col)
    unless $col =~ m/^((\w+)\.)?(\w+:\w+)$/;
  $nam = $3;

  # Treat group-qualified collection name
  if ($1)
  { $loc = $2;
    $sel = $slf->select([$loc]);
    return ($sel->{$loc}->{'_col'}->{$nam}, $nam, $loc)
      if exists($sel->{$loc}) && exists($sel->{$loc}->{'_col'}->{$nam});
    return ();
  }

  # Treat other collection name
  $sel = $slf->select($grp);

  foreach my $grp ('<ENV>', 'RDA')
  { return ($sel->{$grp}->{'_col'}->{$nam}, $nam, $grp)
      if exists($sel->{$grp}) && exists($sel->{$grp}->{'_col'}->{$nam});
  }
  foreach my $grp (keys(%{$sel}))
  { next unless exists($sel->{$grp}) && exists($sel->{$grp}->{'_col'}->{$nam});
    return () if @grp;
    @grp = ($sel->{$grp}->{'_col'}->{$nam}, $nam, $grp);
  }
  return @grp;
}

=head2 S<$h-E<gt>get_collections([$set[$skip...]])>

This method returns the list of all defined multi-run collections.

=cut

sub get_collections
{ my ($slf, $set, @skp) = @_;
  my ($def, $grp, $sel, $tbl, %skp);

  $slf = $slf->get_top;

  # Select the relevant definitions
  $def = {};
  %skp = map {$_ => 1} @skp;
  if (defined($set))
  { die get_string('BAD_SET', $set) unless $set =~ $SET;
    if ($1)
    { $grp = uc($2);
      $set = lc($3);
      $sel = $slf->select([$grp]);
      if (exists($sel->{$grp}) && exists($sel->{$grp}->{'_set'}->{$set}))
      { foreach my $key (keys(%{$tbl = $sel->{$grp}->{'_set'}->{$set}}))
        { $def->{$key} = $tbl->{$key} unless $skp{$key};
        }
      }
    }
    else
    { $set = lc($set);
      $sel = $slf->select;
      foreach my $hsh (values(%{$sel}))
      { next unless exists($hsh->{'_set'}->{$set});
        foreach my $key (keys(%{$tbl = $hsh->{'_set'}->{$set}}))
        { $def->{$key} = $tbl->{$key} unless $skp{$key};
        }
      }
    }
  }
  else
  { $sel = $slf->select;
    foreach my $hsh (values(%{$sel}))
    { foreach my $key (keys(%{$tbl = $hsh->{'_col'}}))
      { $def->{$key} = $tbl->{$key} unless $skp{$key};
      }
    }
  }

  # Return the collection list
  return (sort keys(%{$def})) if wantarray;
  return $def;
}

=head2 S<$h-E<gt>get_sets([$group])>

This method returns the list of all defined collection sets.

=cut

sub get_sets
{ my ($slf, $grp) = @_;
  my ($sel, %set);

  # Select the relevant definitions
  $slf = $slf->get_top;
  $sel = $slf->select($grp);

  # Determine the relevant sets
  foreach my $def (values(%{$sel}))
  { foreach my $nam (keys(%{$def->{'_set'}}))
    { $set{$nam} = 1
    }
  }

  # Return the collection set list
  return (sort keys(%set));
}

=head2 S<$h-E<gt>load($path[,$group])>

This method loads the collection definitions from the specified file and
returns a definition hash.

=cut

sub load
{ my ($slf, $fil, $grp) = @_;
  my ($ifh, $pth);

  $ifh = IO::File->new;
  $pth = RDA::Object::Rda->is_absolute($fil)
    ? $fil
    : RDA::Object::Rda->cat_file($slf->{'_dir'}, $fil);
  $ifh->open("<$pth") or die get_string('ERR_OPEN', $pth, $!);
  return parse($slf, $ifh, $grp, 'ERR_PARSE', $fil);  # Text:ERR_PARSE
}

=head2 S<$h-E<gt>parse($ifh,$group,$error...)>

This method loads the collection definitions from the specified input handle
and returns a definition hash.

=cut

sub parse  ## no critic (Complex)
{ my ($slf, $ifh, $grp, @err) = @_;
  my ($cls, $cur, $def, $err, $key, $lin, $msg, $pos, $str, $val, @tbl);

  # Load the collection definition
  $slf = $slf->get_top;
  $cls = ref($slf);
  $def = {_col => {}, _set => {}};
  $err = RDA::Error->new;
  $pos = 0;
  $lin = q{};
  while (<$ifh>)
  { # Trim leading spaces
    s/^\s+//;
    s/[\r\n]+$//;
    $lin .= $_;

    # Join continuation line
    $pos++;
    next if $lin =~ s/\\$//;
    $lin =~ s/\s+$//;

    # Parse the line
    eval {
      if ($cur && $lin =~ s/^((\w+\/)?(\w+\.)*(\w+))\s*=\s*//)
      { $key = uc($1);
        $str = uc($3);
        $val = RDA::Object::Item::decode_value($slf, \$lin, 1);
        die $msg if ($msg = RDA::Object::Item->validate($str, $val));
        die get_string('BAD_VALUE') unless $lin =~ m/^\s*(#.*)?$/;
        $cur->{'_chg'}->{$key} = $val;
      }
      elsif ($cur && $lin =~ s/^\?(\w+:\w+)?\s*=\s*//)
      { $key = defined($1) ? '_dsc_'.lc($1) : '_dsc';
        $val = _decode(\$lin, get_string('BAD_DESC'));
        $cur->{$key} = $val;
      }
      elsif ($cur && $lin =~ s/^!(\w+:\w+)?(!\w+)?\s*=\s*//)
      { $key = defined($1) ? lc($1) : q{};
        $key .= $2 if $2;
        $val = _decode(\$lin, get_string('BAD_MAN'));
        $cur->{'_man'}->{$key} = $val;
      }
      elsif ($lin =~ s/^\[(\w+:\w+(\|\w+:\w+)*)\]$//)
      { @tbl = split(/\|/, $1);
        $cur = bless {oid => $tbl[0], par => $slf, typ => 'C'}, $cls;
        foreach my $set (@tbl)
        { $def->{'_col'}->{$key = lc($set)} = $cur;
          $def->{'_set'}->{$1}->{$2} = $cur if $key =~ m/^(.*):(.*)$/;
        }
      }
      elsif ($cur && $lin =~
        s/^\*\s*=\s*((\w+\:)?\w+([\-\|]\w+)*(,(\w+\:)?\w+([\-\|]\w+)*)*)//)
      { $cur->{'_mod'} = [map {_norm_module($grp, $_)} split(/\s*,\s*/, $1)];
        die get_string('BAD_LIST') unless $lin =~ m/^\s*(#.*)?$/;
      }
      elsif ($lin !~ m/^(?:#.*)?$/)
      { die get_string('BAD_SPEC');
      }
    };

    # Report an error
    if ($@)
    { my ($rec, $txt);

      if (defined($rec = $err->parse_error($@)))
      { $txt = shift(@{$rec});
        last if $txt =~ m/^last/;
        unshift(@{$rec}, get_string('Error', $txt, $pos));
        $err->add_errors($rec);
      }
    }

    # Prepare the next line
    $lin = q{};
  }
  $ifh->close;

  # Terminate if errors are encountered
  $slf->{'agt'}->abort($err->purge_errors, get_string(@err))
    if $err->has_errors;

  # Return the definifions
  return $def;
}

sub _decode
{ my ($lin, $err) = @_;
  my ($val);

  if ($$lin =~ s/"(.*?)"//)
  { $val = $1;
    die $err unless $$lin =~ m/^\s*(#.*)?$/;
  }
  else
  { $val = $$lin;
  }
  return RDA::Object::decode($val);
}

sub _norm_module
{ my ($grp, $mod) = @_;

  return $mod unless index($mod, q{:}) < 0;  ## no critic (Unless)
  die get_string('NO_GROUP') unless defined($grp);
  return $grp.q{:}.$mod;
}

sub find
{ die get_string('NO_OBJECT');
}

=head2 S<$h-E<gt>select([$group])>

This method selects the relevant collection definitions.

=cut

sub select  ## no critic (Builtin)
{ my ($slf, $grp) = @_;
  my ($agt, $def, $ifh, $lvl, $pth, $sel, $tbl);

  # Initialization
  $slf = $slf->get_top;
  $agt = $slf->{'agt'};
  $def = $slf->{'_def'};
  $ifh = IO::File->new;
  $lvl = $slf->{'_lvl'};
  $sel = {};

  # Load the environment specific definitions on the first use
  unless (exists($def->{'<ENV>'}))
  { $def->{'<ENV>'} = undef;
    if (($pth = $agt->get_env('RDA_MRC')) && -r $pth)
    { eval {
        $ifh->open("<$pth") or die get_string('ERR_OPEN', $pth, $!);
        unless ($lvl < 10)  ## no critic (Unless)
        { $agt->trace(get_string('Env', $pth));
          $agt->trace(get_string('Loading', $pth));
        }
        $def->{'<ENV>'} = parse($slf, $ifh, undef, 'ERR_PARSE', $pth);
        };
      $agt->add_error($@) if $@;
    }
  }
  $sel->{'<ENV>'} = $def->{'<ENV>'} if defined($def->{'<ENV>'});

  # Treat the relevant collection definition files
  $def = $slf->{'_def'};
  $tbl = $slf->{'cnt'}->get_list('MR', $grp);
  foreach my $key (keys(%{$tbl}))
  { # Load the definition file on first use
    unless (exists($def->{$key}))
    { $def->{$key} = undef;
      eval {
        $pth = $tbl->{$key};
        $ifh->open("<$pth") or die get_string('ERR_OPEN', $pth, $!);
        $agt->trace(get_string('Loading', $pth))
        unless $lvl < 10;  ## no critic (Unless)
        $def->{$key} = parse($slf, $ifh, $key, 'ERR_PARSE', $pth);
        };
      $agt->add_error($@) if $@;
    }

    # Select it
    $sel->{$key} = $def->{$key} if defined($def->{$key});
  }

  # Return the selected definitions
  return $sel;
}

=head2 S<$h-E<gt>xref($select[,$flag])>

This method produces a cross-reference of the multi-run collection definitions
and the related modules. When the flag is set, it includes the groups without
title.

=cut

sub xref
{ my ($slf, $sel, $flg, $nam) = @_;
  my ($buf, $mod, $obj, $tbl, $uid, %bad, %col, %mod, %set);

  # Get the collection modules
  $slf = $slf->get_top;
  foreach my $mod ($slf->{'cnt'}->get_modules('MC'))
  { $mod{$mod} = [];
  }

  # Analyze the definitions
  foreach my $grp (sort keys(%{$sel}))
  { # Analyze the collection sets
    foreach my $set (sort keys(%{$tbl = $sel->{$grp}->{'_set'}}))
    { $set{$grp.q{.}.$set} = [map {"$grp.$set:$_"} sort keys(%{$tbl->{$set}})];
    }

    # Analyze the groups
    foreach my $nam (sort keys(%{$tbl = $sel->{$grp}->{'_col'}}))
    { $obj = $tbl->{$nam};
      next unless $flg || $obj->get_title($nam);
      $col{$uid = $grp.q{.}.$nam} = [];
      if (exists($obj->{'_mod'}))
      { foreach my $col (@{$obj->{'_mod'}})
        { ($mod) = split(/[\-\|]/, $col);
          if (exists($mod{$mod}))
          { push(@{$mod{$mod}}, $uid);
          }
          else
          { push(@{$bad{$mod}}, $uid);
          }
          push(@{$col{$uid}}, $col);
        }
      }
    }
  }

  # Produce the cross-reference
  $buf = _dsp_name($nam || get_string('Xref')).$RPT_NXT;
  $buf .= _xref_dsp(\%set, 'XrefSets',        'set', q{-});
  $buf .= _xref_dsp(\%col, 'XrefCollections', 'mrc', q{-});
  $buf .= _xref_dsp(\%mod, 'XrefReferenced',  'collect');
  $buf .= _xref_dsp(\%bad, 'XrefUnknown');
  return $buf;
}

# Display a result set
sub _xref_dsp
{ my ($tbl, $ttl, $typ, $dft) = @_;
  my ($buf, $str, @tbl);

  return q{} unless ref($tbl) eq 'HASH';
  @tbl = keys(%{$tbl});
  @tbl = grep {@{$tbl->{$_}}} @tbl unless defined($dft);
  return q{} unless @tbl;

  $buf = _dsp_table(get_string($ttl));
  foreach my $nam (sort @tbl)
  { $str = @{$tbl->{$nam}}
      ? q{``}.join(q{``, ``}, @{$tbl->{$nam}}).q{``}
      : $dft;
    $str =~ s/\|/\\174/g;
    $buf .= _dsp_row($typ ? qq{!!$typ:$nam!$nam!!} : qq{``$nam``}, $str);
  }
  return $buf.$RPT_EOT;
}

=head1 GROUP OBJECT METHODS

=head2 S<$h-E<gt>display($name[,$flag])>

This method displays the manual page of the specified collection. When the flag
is set, it includes the collection settings.

=cut

sub display
{ my ($slf, $nam, $det, $flg) = @_;
  my ($buf, $cnt, $dsc, $lnk, $mod, $pat, $tbl, @pre);

  # Initialization
  die get_string('NOT_COLLECTION') unless $slf->{'typ'} eq 'C';
  $cnt = $slf->{'par'}->{'cnt'};
  $nam = $slf->{'_alt'}->{$nam} if exists($slf->{'_alt'}->{$nam});

  # Display the collection name and title
  $buf = _dsp_title(get_string('TtlName'))._dsp_text($RPT_TXT,
    get_string('DspName', $nam, $slf->get_title($nam, q{})), 1);

  # Display the text elements
  if (exists($slf->{'_man'}))
  { $buf .= _dsp_title(get_string('TtlDesc'));
    foreach my $key (sort keys(%{$tbl = $slf->{'_man'}}))
    { $buf .= _dsp_block($RPT_TXT, $tbl->{$key}, 1)
        unless $key =~ m/^\w*(\!(\w+))?$/  ## no critic (Unless)
            && defined($1) && $2 ne $nam;
    }
  }

  # Display the modules and their descriptions
  if (exists($slf->{'_mod'}))
  { $buf .= _dsp_title(get_string('TtlModules'))
      ._dsp_text($RPT_TXT, get_string('DspModules', $nam));
    foreach my $itm (@{$slf->{'_mod'}})
    { if ($itm =~ m/\-/)
      { ($mod, @pre) = split(/\-/, $itm);
        $lnk = q{!!collect:}.$mod.q{!}.$cnt->get_abbr('MC', $mod, $mod).q{!!};
        $lnk .= qq{ ($dsc)} if ($dsc = $cnt->get_desc('MC', $mod, q{}));
        $buf .= _dsp_text($RPT_LST,
          get_string('DspSections', join(q{``, ``}, @pre), $lnk));
      }
      else
      { ($mod, $pat) = split(/\|/, $itm, 2);
        $lnk = q{!!collect:}.$mod.q{!}.$cnt->get_abbr('MC', $mod, $mod).q{!!};
        $lnk .= qq{ ($dsc)} if ($dsc = $cnt->get_desc('MC', $mod, q{}));
        $buf .= _dsp_text($RPT_LST, defined($pat)
          ? get_string('DspPattern', $pat, $lnk)
          : get_string('DspAll', $lnk));
      }
    }
    $buf .= $RPT_NXT;
  }

  # Display the collection settings
  if ($det && exists($slf->{'_chg'}))
  { $buf .= _dsp_title(get_string('TtlSettings'))
      ._dsp_text($RPT_TXT, get_string('DspSettings', $nam));
    foreach my $key (sort keys(%{$tbl = $slf->{'_chg'}}))
    { $buf .= _dsp_text($RPT_SUB, q{``}.$key.q{=}
        .RDA::Object::Item::encode_value($tbl->{$key}).q{``});
    }
    $buf .= $RPT_NXT;
  }

  # Display the copyright and trademark notices
  $buf .= _dsp_title(get_string('TtlCopyright'))
    ._dsp_text($RPT_TXT, get_string('Copyright'), 1)
    ._dsp_title(get_string('TtlTrademark'))
    ._dsp_text($RPT_TXT, get_string('Trademark'))
    unless $flg;

  # Return the result
  return $buf;
}

=head2 S<$h-E<gt>get_mapping>

This method returns the associate modules

=cut

sub get_mapping
{ my ($slf) = @_;
  my ($tbl);

  die get_string('NOT_COLLECTION') unless $slf->{'typ'} eq 'C';
  $tbl = {};
  if (exists($slf->{'_mod'}))
  { foreach my $mod (@{$slf->{'_mod'}})
    { $tbl->{$1} = "$2/$3" if $mod =~ m/^((\w+):(\w+))/;
    }
  }
  return $tbl;
}

=head2 S<$h-E<gt>get_title($name[,$default])>

This method returns the description of the specified collection or the default
value when not found.

=cut

sub get_title
{ my ($slf, $nam, $ttl) = @_;

  die get_string('NOT_COLLECTION') unless $slf->{'typ'} eq 'C';
  return $ttl unless $nam;
  $nam = $1 if $nam =~ m/^[A-Z][A-Z\d]+\.(.*)$/i;
  return
    exists($slf->{"_dsc_$nam"}) ? $slf->{"_dsc_$nam"} :
    exists($slf->{'_dsc'})      ? $slf->{'_dsc'} :
                                  $ttl;
}

# --- Internal reporting routines ---------------------------------------------

sub _dsp_block
{ my ($pre, $txt, $nxt) = @_;
  my $buf = q{};

  foreach my $str (split(/\n|\\n/, $txt))
  { if ($str =~ m/^(\s*[o\*\-]\s+)(.*)$/)
    { $buf .= qq{.I '$pre\001$1'\n$2\n\n};
    }
    else
    { $buf .= qq{.I '$pre'\n$str\n\n};
    }
  }
  $buf .= qq{.N $nxt\n} if $nxt;
  return $buf;
}

sub _dsp_name
{ my ($ttl) = @_;

  return qq{.R '$ttl'\n};
}

sub _dsp_row
{ return join(q{|}, @_).qq{\n};
}

sub _dsp_table
{ my ($pre, $txt) = @_;

  return defined($txt) ? qq{.M 2 '$pre|$txt'\n} : qq{.M 2 '$pre'\n}
}

sub _dsp_text
{ my ($pre, $txt, $nxt) = @_;

  $txt =~ s/\n{2,}/\n\\040\n/g;
  $txt =~ s/(\n|\\n)/\n\n.I '$pre'\n/g;
  return qq{.I '$pre'\n$txt\n\n}.($nxt ? qq{.N $nxt\n} : q{});
}

sub _dsp_title
{ my ($ttl) = @_;

  return qq{.T '$ttl'\n};
}

# --- Section management ------------------------------------------------------

# Start a new collection
sub _begin_collect  ## no critic (Complex)
{ my ($slf, $ctl, $skp, $ctx, $col) = @_;
  my ($mod, $par, $rec, $tbl);

  # Manage the table of content contribution
  if (exists($slf->{'_cur'}))
  { $par = $slf->{'_ref'};
    push(@{$par->{'lin'}}, $par->{'col'}->{$col} = {
        col => {},
        nam => $col,
        lin => [],
        par => $par,
        sct => {},
        typ => 'C',
        }) unless exists($par->{'col'}->{$col});
    $slf->{'_ref'} = $par->{'col'}->{$col};
    $par = $slf->{'_cur'};
    push(@{$par->{'lin'}}, $par->{'col'}->{$col} = {
        col => {},
        nam => $col,
        lin => [],
        par => $par,
        sct => {},
        typ => 'C',
        }) unless exists($par->{'col'}->{$col});
    $slf->{'_cur'} = $par->{'col'}->{$col};
  }
  elsif ($slf->{'_typ'})
  { $slf->{'_prv'} = $rec = $slf->{'out'}->get_info('prv');
    $slf->{'_ref'} = exists($rec->{'col'}->{$col})
      ? $rec->{'col'}->{$col}
      :
    $rec->{'col'}->{$col} = {
      col => {},
      nam => $col,
      lin => [],
      par => $rec,
      sct => {},
      typ => 'C',
      } unless exists($rec->{'col'}->{$col});
    $slf->{'_ref'} = $rec->{'col'}->{$col};
    $slf->{'_toc'} = $slf->{'_cur'} = {
      col => {},
      nam => $col,
      lin => [],
      sct => {},
      typ => 'C',
      };
    $slf->{'rpt'}->load_index($slf->{'_prv'}, 1);
  }
  else
  { die get_string('BAD_MODULE');
  }

  # Update the multi-run collection flag
  if (exists($slf->{'nam'}))
  { $slf->{'col'}->set_value("STATUS.$slf->{'nam'}.B_MRC",
      $slf->{'_typ'} ? 0 : 1);
  }
  elsif ($slf->{'oid'} =~ $RE_DC)
  { $slf->{'col'}->set_value("STATUS.$2.$3.B_MRC", $slf->{'_typ'} ? 0 : 1);
  }

  # Determine the control section
  ## no critic (Interpolation)
  $ctl->{'begin'} = $ctl->{'end'} = -1;
  if ($ctx->check_variable('@CONTROL_SECTIONS'))
  { foreach my $nam ($ctx->get_value('@CONTROL_SECTIONS')->eval_as_array)
    { $ctl->{$nam} = 1;
    }
  }

  # Determine the sections to execute
  if (!$slf->{'_typ'})
  { # Load the sections already collected
    foreach my $nam (keys(%{$tbl = $slf->{'_cur'}->{'sct'}}))
    { $skp->{$nam} = 1 if $tbl->{$nam}->{'typ'} eq 'R';
    }
  }
  elsif ($> && $ctx->check_variable('@ROOT_SECTIONS'))
  { foreach my $nam ($ctx->get_value('@ROOT_SECTIONS')->eval_as_array)
    { $skp->{$nam} = 1;
    }
  }

  # Do not skip common sections
  if ($ctx->check_variable('@COMMON_SECTIONS'))
  { foreach my $nam ($ctx->get_value('@COMMON_SECTIONS')->eval_as_array)
    { delete($skp->{$nam});
    }
  }

  # Determine the validation default
  $slf->{'_dft'} = $ctx->check_variable('$VALIDATE')
    ? $ctx->get_value('$VALIDATE')->eval_as_scalar
    : 0;

  # Indicate the start of the collection
  return $slf->{'rpt'}->begin_capture;
}

# Start a new section
sub _begin_section
{ my ($slf, $ctl, $skp, $nam) = @_;
  my ($typ);

  # Always execute a control section
  if (exists($ctl->{$nam}))
  { $slf->{'rpt'}->begin_capture;
    return 1;
  }

  # Skip a section when requested
  if (exists($skp->{$nam}))
  { my ($cur, $rec);

    $cur = $slf->{'_cur'};
    $rec = {
      lin => [],
      nam => $nam,
      rpt => [],
      typ => 'E',
      };
    if ($slf->{'_typ'})
    { push(@{$cur->{'lin'}}, exists($slf->{'_ref'}->{'sct'}->{$nam})
        ? $slf->{'_ref'}->{'sct'}->{$nam}
        : $rec);
    }
    elsif (!exists($cur->{'sct'}->{$nam}))
    { push(@{$cur->{'lin'}}, $cur->{'sct'}->{$nam} = $rec);
    }
    elsif (($cur = $cur->{'sct'}->{$nam})->{'typ'} ne 'R')
    { foreach my $key (keys(%{$rec}))
      { $cur->{$key} = $rec->{$key};
      }
    }
    return 0;
  }

  # Determine whether a common section requires cleanup
  $typ = $slf->{'_typ'} ? 'S' : 'R';
  $slf->{'_cln'} = (exists($slf->{'_ref'}->{'sct'}->{$nam})
    && $slf->{'_ref'}->{'sct'}->{$nam}->{'typ'} eq $typ)
    ? $slf->{'_ref'}->{'sct'}->{$nam}->{'rpt'}
    : undef;

  # Treat a section
  $slf->{'_acc'} = $slf->{'_dft'};
  $slf->{'_sct'} = $nam;
  $slf->{'rpt'}->begin_section($nam, $slf->{'_typ'});
  return 1;
}

# End the collection
sub _end_collect
{ my ($slf) = @_;

  if (exists($slf->{'_cur'}->{'par'}))
  { $slf->{'_cur'} = $slf->{'_cur'}->{'par'};
  }
  else
  { $slf->{'rpt'}->save_toc(delete($slf->{'_cur'}));
  }
  $slf->{'_ref'} = $slf->{'_ref'}->{'par'}
    if $slf->{'_typ'} && exists($slf->{'_ref'}->{'par'});
  return --$slf->{'_lvl'};
}

# End the section
sub _end_section
{ my ($slf) = @_;
  my ($cur, $nam, $prv, $rec, $sct);

  $cur = $slf->{'_cur'};
  if (defined($nam = delete($slf->{'_sct'})))
  { $rec = $slf->{'rpt'}->end_section($nam, $slf->{'_typ'}, $slf->{'_acc'},
      $slf->{'_cln'});
    $sct = $cur->{'sct'};
    if (!exists($sct->{$nam}))
    { $sct->{$nam} = $rec;
      push(@{$cur->{'lin'}}, $rec);
    }
    elsif ($slf->{'_acc'} || $sct->{$nam}->{'typ'} ne 'S' || !$slf->{'_typ'})
    { $prv = $sct->{$nam};
      foreach my $key (keys(%{$rec}))
      { $prv->{$key} = $rec->{$key};
      }
    }
  }
  elsif ($slf->{'_typ'})
  { push(@{$cur->{'lin'}}, @{$slf->{'rpt'}->get_section});
  }
  return;
}

# --- SDCL extensions ---------------------------------------------------------

# Define the global variable
sub _begin_mrc
{ my ($pkg) = @_;
  my ($col);

  $col = $pkg->get_collector;
  $pkg->set_top('MRC', $col->get_mrc->new($col, $pkg));
  return;
}

# Close all active reports
sub _end_mrc
{ shift->get_top('MRC')->end;
  return;
}

# Get a collection definition
sub _get_collect
{ my ($slf, $spc, $str) = @_;

  if ($$str =~ s/^\&\{\s*//)
  { $spc->[$SPC_REF] = $slf->parse_value($str);
    die get_string('BAD_NAME') unless $$str =~ s/^\}\s*//;
  }
  elsif ($$str =~ s/^(([A-Z][A-Z\d]*:)?\w+(\|\w+)*)\s*//)
  { $spc->[$SPC_REF] = $1;
  }
  else
  { die get_string('BAD_NAME');
  }
  $spc->[$SPC_VAL] = $slf->parse_sub_list($str);
  return;
}

# Execute a collection
sub _exe_collect  ## no critic (Complex)
{ my ($slf, $spc) = @_;
  my ($blk, $col, $dft, $die, $err, $lng, $mrc, $nam, $pkg, $pre, $sct, $tim,
      $top, $val, $vol);

  $pkg = $slf->{'_pkg'};
  $lng = $pkg->{'lng'};
  $top = $pkg->get_top;
  $col = $top->{'COL'};
  $mrc = $top->{'MRC'};
  die get_string('BAD_NESTING') if exists($mrc->{'_sct'});

  # Load the block
  $val = $val->eval_as_string if ref($val = $spc->[$SPC_REF]);
  ($nam, $pre) = split(/\|/, $val, 2);
  return $CONT unless defined($nam) && length($nam);
  unless ($blk = $lng->find_package($nam))
  { $blk = $lng->load_file($nam, $pkg->{'dir'}) ||
           $lng->search_package($pkg->{'grp'}, $nam);
    die get_string('NO_MODULE', $nam) unless defined($blk);
    $blk->{'glb'} = {%{$top->{'glb'}}};
    $blk->{'_lib'} = $top->get_lib->new;
  }

  # Execute the associated code block
  $dft = $top->{'_dft'};
  if ($mrc->{'_sta'})
  { $vol = $top->{'OUT'}->get_info('spc') if exists($top->{'OUT'});
    $tim = time;
  }
  if (exists($blk->{'_sct'}) && exists(($sct = $blk->{'_sct'})->{q{-}}))
  { my ($arg, $ctx, $cur, $dst, $exe, $flg, $ret, $src, @cls, %ctl, %skp);

    $blk->{'_par'} = $slf;

    # Transfer new existing classes
    $src = $blk->{'use'};
    $dst = $top->{'use'};
    $flg = $blk->{'_use'};
    @cls = grep {!exists($dst->{$_})} keys(%{$src});
    foreach my $cls (sort {$src->{$a}->{'rnk'} <=> $src->{$b}->{'rnk'}
      || $a cmp $b} @cls)
    { $dst->{$cls} = $src->{$cls};
      &{$src->{$cls}->{'beg'}}($top, $flg->{$cls})
        if exists($src->{$cls}->{'beg'});
    }

    # Evaluate the argument list
    $arg = $spc->[$SPC_VAL]->eval_value;

    # Create the execution context and manage recursive calls
    ++$mrc->{'_lvl'};
    $ctx = $blk->{'ctx'}->push_context($slf, $slf->{'ctx'}, 1);

    # Declare the arguments
    $ctx->set_value('@arg', $arg);  ## no critic (Interpolation)

    # Execute the code before any section
    $err = 0;
    unless ($blk->{'dft'})
    { $ctx->{'val'} = $VAL_UNDEF;
      eval {$ret = $sct->{q{-}}->exec_block(q{section '-'})};
      $ret = $blk->check_die($@) if $@;
      ++$err if $ret < 0;
      $blk->{'dft'} = 1;
    }

    # Select the sections to execute
    if (defined($pre))
    { _begin_collect($mrc, \%ctl, \%skp, $ctx, $blk->get_oid.q{|}.$pre);
      $blk->{'nxt'} = [grep {m/^$pre\_/} @{$blk->{'_exe'}}];
    }
    else
    { _begin_collect($mrc, \%ctl, \%skp, $ctx, $blk->get_oid);
      $blk->{'nxt'} = [@{$blk->{'_exe'}}];
    }
    if ($exe = $mrc->{'_exe'})
    { foreach my $nxt (@{$blk->{'nxt'}})
      { $pkg->{'sct'}->{$nxt} = 0 unless exists($exe->{$nxt});
      }
    }
    unshift(@{$blk->{'nxt'}}, 'begin') if exists($sct->{'begin'});

    # Execute the selected sections
    $ctx->{'val'} = $VAL_UNDEF;
    while (defined($cur = shift(@{$blk->{'nxt'}})))
    { next unless _begin_section($mrc, \%ctl, \%skp, $cur);
      last if $blk->check_quotas;
      $blk->{'sct'}->{$cur} = 1;
      eval {$ret = $sct->{$cur}->exec_block(qq{section '$cur'})};
      if ($die = $@)
      { eval {$ret = $blk->check_die($die)};
        $ret = $RET_DIE if ($die = $@);
      }
      elsif ($exe)
      { $pkg->{'sct'}->{$cur} = 1 unless exists($exe->{$cur});
      }
      _end_section($mrc);
      if ($ret)
      { ++$err unless $ret > 0;  ## no critic (Unless)
        last;
      }
    }
    unless ($die)
    { $blk->{'val'} = $ctx->get_internal('val')->eval_as_scalar
        unless $ret < 0;  ## no critic (Unless)
      if (exists($sct->{'end'}))
      { $blk->{'sct'}->{'end'} = 1;
        eval {$ret = $sct->{'end'}->exec_block(q{section 'end'})};
        $ret = $blk->check_die($@) if $@;
        ++$err if $ret < 0;
      }
    }

    # Restore the previous context
    $ctx->pop_context($blk, $slf);
    _end_collect($mrc);
  }
  else
  { $err = -1;
  }
  $top->{'_dft'} = $dft;

  # Log statistics
  $col->log('m', $blk->get_oid, $blk->get_version,
    defined($vol) ? $top->{'OUT'}->get_info('spc') - $vol : 0, time - $tim)
    if $mrc->{'_sta'};

  # Transfer the execution error
  $top->add_error(get_string('ERR_COLLECT'), $blk->purge_errors) if $err;

  # Keep or free the block memory
  if ($blk->{'ctx'}->check_variable('$KEEP_BLOCK')) ## no critic (Interpolation)
  { $lng->keep_package($blk);
  }
  else
  { # Resynchronize the calling block
    $top->{'OUT'}->deprefix($blk) if exists($top->{'OUT'});

    # Delete the block
    $lng->remove_package($blk);
  }

  # Propagate any error
  die $die if $die;
  $top->{'agt'}->abort if $err;

  # Indicate the successful completion
  return $CONT;
}

1;

__END__

=head1 SEE ALSO

L<RDA::Agent|RDA::Agent>,
L<RDA::Error|RDA::Error>,
L<RDA::Object|RDA::Object>,
L<RDA::Object::Collect|RDA::Object::Collect>,
L<RDA::Object::Content|RDA::Object::Content>,
L<RDA::Object::Item|RDA::Object::Item>,
L<RDA::Object::Output|RDA::Object::Output>,
L<RDA::Object::Rda|RDA::Object::Rda>,
L<RDA::Object::Report|RDA::Object::Report>,
L<RDA::Object::Toc|RDA::Object::Toc>,
L<RDA::SDCL::Block|RDA::SDCL::Block>,
L<RDA::Text|RDA::Text>,
L<RDA::Value::Scalar|RDA::Value::Scalar>

=head1 COPYRIGHT NOTICE

Copyright (c) 2002, 2016, Oracle and/or its affiliates. All rights reserved.

=head1 TRADEMARK NOTICE

Oracle and Java are registered trademarks of Oracle and/or its
affiliates. Other names may be trademarks of their respective owners.

=cut