view release on metacpan or  search on metacpan

lib/DBIx/PgLink/RemoteAction.pm  view on Meta::CPAN

    my $sub_name = $_->{conv_to_remote};
    {
      name      => $_->{column_name},
      data_type => $t->{standard_type}, # standard type code
      conv_to_remote_coderef => $self->resolve_converter_method($sub_name),
    }
  } @{$q->params};

  # prepare result conversion (by field name)
  my %result_conv = map {
    $_->{column_name} => $self->resolve_converter_method( $_->{conv_to_local} )
  } grep { $_->{conv_to_local} } $a->columns->metadata;

  return {
    sth         => $self->adapter->prepare_cached($query_text),
    params      => \@params,
    result_conv => \%result_conv,
    returns_set => ($a->can('returns_set') ? $a->returns_set : 1), # return set by default
  };
};


sub remote_statement_bind_and_execute {
  my ($self, $st, $param_values) = @_;

  # typed binding ~2x times slower than untyped
  my $typed = $self->adapter->require_parameter_type;

  my $i = 0;
  for my $v (@{$param_values}) {
    my $m = $st->{params}->[$i++];

    my $sub = $m->{conv_to_remote_coderef};
    $sub->($self->adapter, $v) if $sub; # conversion

    if ($typed) {
      my $type_code = $m->{data_type}; # SQL Standard type code
      if (defined $type_code) {
        $st->{sth}->bind_param($i, $v, {TYPE=>$type_code});
        trace_msg('INFO', "Bind[$i/$type_code]: " . (defined $v ? $v : 'NULL')) if trace_level >= 3;
      } else {
        $st->{sth}->bind_param($i, $v);
        trace_msg('INFO', "Bind[$i]: " . (defined $v ? $v : 'NULL')) if trace_level >= 3;
      }
    }
  }

  if ($typed) {
    $st->{sth}->execute;
  } else {
    $st->{sth}->execute(@{$param_values});
  }
}


# cache to the session end
memoize 'remote_statement_prepare_cached';


# store session-level filter for SELECT remote data, by object_id
has 'query_session_filter' => (is=>'ro', isa=>'HashRef', default=>sub{{}} );


method remote_accessor_query => named (
  object_id    => { isa=>'Int', required=>1 },
  where        => { isa=>'Str', required=>0, default=>'' },
  param_values => { isa=>'PostgreSQLArray', required=>0, coerce=>1, default=>[] },
  param_types  => { isa=>'PostgreSQLArray', required=>0, coerce=>1, default=>[] },
) => sub {
  my ($self, $p) = @_;

  # use session filter, if there is no user-supplied params and WHERE-clause
  unless ($p->{where} || @{$p->{param_values}}) {
    my $f = $self->query_session_filter->{$p->{object_id}};
    $p = $f if $f;
  }

  my $a = $self->load_accessor_cached($p->{object_id});

  my $st = $self->remote_statement_prepare_cached(
    accessor => $a,
    action   => 'S', 
    where    => $p->{where},
  );

  trace_msg('INFO', "QUERY to " . $a->remote_object_quoted) if trace_level >= 2;
  # -------------------------------------------------------
  if (@{$p->{param_types}}) {
    $self->_bind_user_params( $st->{sth}, $p->{param_values}, $p->{param_types} );
    $st->{sth}->execute;
  } else {
    $self->remote_statement_bind_and_execute( $st, $p->{param_values} );
  }
  # -------------------------------------------------------

  if ($st->{returns_set}) {
    # return setof record
    while (my $row = $st->{sth}->fetchrow_hashref) {
      # convert resultset values by column name
      while (my ($col, $sub) = each %{$st->{result_conv}}) {
        $sub->( $self->adapter, $row->{$col} );
      }
      # pipe row to PostgreSQL
      main::return_next(\%{$row});
    }
    $st->{sth}->finish;
    return undef;
  } else {
    # return scalar
    my $result = $st->{sth}->fetchrow_array;
    my (undef, $sub) = each %{$st->{result_conv}}; # only one
    $sub->($result) if $sub;
    $st->{sth}->finish;
    return $result;
  }
};


method set_query_session_filter => named (
  object_id    => { isa=>'Int', required=>1 },
  where        => { isa=>'Str', required=>0, default=>'' },