view release on metacpan or  search on metacpan

lib/DBIO/MySQL/Async/Storage.pm  view on Meta::CPAN

sub _conninfo_hash {
  my ($self, $ci) = @_;
  # Use already-normalized stored conninfo; only do fallback resolution
  # when called in contexts where _conninfo is not yet set
  my $conninfo = defined($ci) ? $ci : $self->{_conninfo};
  return unless defined $conninfo;
  return $conninfo if ref $conninfo ne 'HASH';
  # Already normalized by _normalize_async_connect_info; return as-is
  return $conninfo;
}

# --- SQL Generation ---


sub sql_maker {
  my $self = shift;
  $self->{sql_maker} ||= do {
    my $class = $self->{sql_maker_class};
    $class->new(
      quote_char     => '`',
      name_sep       => '.',
    );
  };
}

sub _generate_sql {
  my ($self, $op, @args) = @_;
  my $sm = $self->sql_maker;
  my $method = {
    select        => 'select',
    select_single => 'select',
    insert        => 'insert',
    update        => 'update',
    delete        => 'delete',
  }->{$op} or croak "Unknown operation: $op";

  return $sm->$method(@args);
}

# --- Async Query Execution ---


sub select_async {
  my $self = shift;
  return $self->_run_crud('select', undef, @_);
}


sub select_single_async {
  my $self = shift;
  return $self->_run_crud('select_single', undef, @_);
}


sub insert_async {
  my $self = shift;
  return $self->_run_crud('insert', undef, @_);
}


sub update_async {
  my $self = shift;
  return $self->_run_crud('update', undef, @_);
}


sub delete_async {
  my $self = shift;
  return $self->_run_crud('delete', undef, @_);
}

# Build SQL for a CRUD op and run it on a connection. When $mdb is
# undef, a pooled connection is acquired and released around the query;
# when $mdb is given, the query runs pinned on that connection (no
# release) — used by the transaction context so all CRUD inside a
# txn_do_async hits the BEGIN/COMMIT connection.
sub _run_crud {
  my ($self, $op, $mdb, @args) = @_;

  if ($op eq 'select_single') {
    my ($sql, @bind) = $self->sql_maker->select(@args);
    return $self->_run_on_conn($mdb, sub {
      my $conn = shift;
      $self->_query_on($conn, $sql, \@bind);
    })->then(sub {
      my @rows = @_;
      return @rows ? $rows[0] : undef;
    });
  }

  if ($op eq 'insert') {
    my ($sql, @bind) = $self->sql_maker->insert(@args);
    return $self->_run_on_conn($mdb, sub {
      my $conn = shift;
      $self->_query_on($conn, $sql, \@bind)->then(sub {
        # LAST_INSERT_ID() must run on the SAME connection as the INSERT.
        my $lii_f = $self->_query_on($conn, 'SELECT LAST_INSERT_ID()', []);
        $lii_f->on_done(sub {
          my @rows = @_;
          $self->{_last_insert_id} = $rows[0]->[0] if @rows && defined $rows[0]->[0];
        });
        return $lii_f;
      });
    });
  }

  my $method = { select => 'select', update => 'update', delete => 'delete' }->{$op}
    or croak "Unknown CRUD operation: $op";
  my ($sql, @bind) = $self->sql_maker->$method(@args);
  return $self->_run_on_conn($mdb, sub {
    my $conn = shift;
    $self->_query_on($conn, $sql, \@bind);
  });
}

# Resolve a connection and run $cb->($conn) on it. When $mdb is given the
# connection is pinned: $cb runs on it and it is NOT released (the txn
# owner manages its lifecycle). When $mdb is undef a pooled connection is
# acquired via acquire->then — honoring the pool's waiter queue when the
# pool is full — and released once $cb's Future is ready, on success or
# failure, so connections never leak.

lib/DBIO/MySQL/Async/Storage.pm  view on Meta::CPAN

    return $f;
  });
}

# --- Pipeline Mode ---


sub pipeline {
  my ($self, $coderef) = @_;

  return $self->pool->acquire->then(sub {
    my $mdb = shift;
    $mdb->query('SET autocommit=0');  # MariaDB pipelining needs explicit tx mode

    my $result = eval { $coderef->($self) };
    my $err = $@;

    if ($err) {
      $mdb->query('ROLLBACK', sub {});  # we don't care about the result here
      $self->pool->release($mdb);
      return Future->fail($err);
    }

    # Sync the pipeline — callback fires when all results are in
    my $f = Future->new;
    $mdb->query('COMMIT', sub {
      $mdb->query('SET autocommit=1', sub {});  # reset
      $self->pool->release($mdb);
      if (ref $result && $result->can('then')) {
        $result->on_done(sub { $f->done(@_) })
               ->on_fail(sub { $f->fail(@_) });
      } else {
        $f->done($result);
      }
    });

    return $f;
  });
}

# --- Sync Fallbacks ---
# These allow sync methods (->all, ->first etc.) to work
# by blocking the event loop. Useful for scripts/migrations.

sub select {
  my $self = shift;
  return $self->select_async(@_)->get;
}

sub select_single {
  my $self = shift;
  return $self->select_single_async(@_)->get;
}

sub insert {
  my $self = shift;
  return $self->insert_async(@_)->get;
}

sub update {
  my $self = shift;
  return $self->update_async(@_)->get;
}

sub delete {
  my $self = shift;
  return $self->delete_async(@_)->get;
}

sub txn_do {
  my $self = shift;
  return $self->txn_do_async(@_)->get;
}

# --- Schema Integration ---

sub schema { $_[0]->{schema} }
sub debug  { $_[0]->{debug} }

sub connected { defined $_[0]->{pool} && $_[0]->pool->available > 0 }


sub in_txn { 0 }


sub last_insert_id {
  my ($self, $source, @cols) = @_;
  return $self->{_last_insert_id};
}

sub disconnect {
  my $self = shift;
  if ($self->{pool}) {
    $self->{pool}->shutdown;
    $self->{pool} = undef;
  }
}

sub DESTROY {
  my $self = shift;
  $self->disconnect if $self->{pool};
}

1;

__END__

=pod

=encoding UTF-8

=head1 NAME

DBIO::MySQL::Async::Storage - Async MySQL/MariaDB storage driver using EV::MariaDB

=head1 VERSION

version 0.900000

=head1 DESCRIPTION