view release on metacpan or  search on metacpan

lib/DBIx/Class/Async/Schema.pm  view on Meta::CPAN

sub schema_version {
    my $self  = shift;
    my $class = $self->{_async_db}->{_schema_class};

    unless ($class) {
        croak("schema_class is not defined in " . ref($self));
    }

    return $class->schema_version if $class->can('schema_version');

    return undef;
}

sub sync_metadata {
    my ($self) = @_;

    my $async_db = $self->{_async_db};
    my @futures;

    # Ping every worker in the pool
    for (1 .. $async_db->{_workers_config}->{_count}) {
        push @futures, DBIx::Class::Async::_call_worker($async_db, 'ping', {});
    }

    return Future->wait_all(@futures);
}

sub schema_class { shift->{_async_db}->{_schema_class}; }

sub source_ {
    my ($self, $source_name) = @_;

    unless (exists $self->{_sources_cache}{$source_name}) {
        my $source = eval { $self->{_native_schema}->source($source_name) };

        croak("No such source '$source_name'") if $@ || !$source;

        $self->{_sources_cache}{$source_name} = $source;
    }

    return $self->{_sources_cache}{$source_name};
}

sub sources_ { shift->{_native_schema}->sources; }

sub storage  { shift->{_storage}; }

sub source {
    my ($self, $source_name) = @_;

    # 1. Retrieve the cached entry
    my $cached = $self->{_sources_cache}{$source_name};

    # 2. Check if we need to (re)fetch:
    #    Either we have no entry, or it's a raw HASH (autovivification artifact)
    if (!$cached || !blessed($cached)) {

        # Clean up any "ghost" hash before re-fetching
        delete $self->{_sources_cache}{$source_name};

        # 3. Use the persistent provider to keep ResultSource objects alive
        $self->{_metadata_provider} ||= do {
            my $class = $self->{_async_db}->{_schema_class};
            eval "require $class" or die "Could not load schema class $class: $@";
            $class->connect(@{$self->{_async_db}->{_connect_info}});
        };

        # 4. Fetch the source and validate its blessing
        my $source_obj = eval { $self->{_metadata_provider}->source($source_name) };

        if (blessed($source_obj)) {
            $self->{_sources_cache}{$source_name} = $source_obj;
        } else {
            return undef;
        }
    }

    return $self->{_sources_cache}{$source_name};
}

sub sources {
    my $self = shift;

    my $schema_class = $self->{_async_db}->{_schema_class};
    my $connect_info = $self->{_async_db}->{_connect_info};
    my $temp_schema = $schema_class->connect(@{$connect_info});
    my @sources = $temp_schema->sources;

    $temp_schema->storage->disconnect;

    return @sources;
}

sub txn_begin {
    my $self = shift;

    # We return the future so the caller can wait for the 'BEGIN' to finish
    return DBIx::Class::Async::_call_worker(
        $self->{_async_db},
        'txn_begin',
        {}
    );
}

sub txn_commit {
    my $self = shift;

    return DBIx::Class::Async::_call_worker(
        $self->{_async_db},
        'txn_commit',
        {}
    );
}

sub txn_rollback {
    my $self = shift;

    return DBIx::Class::Async::_call_worker(
        $self->{_async_db},
        'txn_rollback',
        {}