File Coverage

blib/lib/Bif/Sync/Client.pm

Criterion	Covered	Total	%
statement	16	18	88.8
branch			n/a
condition			n/a
subroutine	6	6	100.0
pod			n/a
total	22	24	91.6

package Bif::Sync::Client;
use strict;
use warnings;
use feature 'state';
use AnyEvent;
use Bif::Mo;
use Coro::Handle;
use Log::Any '$log';
use Sys::Cmd qw/spawn/;

our $VERSION = '0.1.5_7';
extends 'Bif::Sync';

has name => ( required => 1, );

has location => ( is => 'ro', );

has hub_id => (
    is      => 'rw',
    default => sub {
        my $self   = shift;
        my $hub_id = $self->db->xval(
            select => 'h.id',
            from   => 'hubs h',
            where  => { 'h.location' => $self->location },
        );
        return $hub_id;
    },
);

has child => ( is => 'rw' );

has child_watcher => ( is => 'rw' );

has stderr_watcher => ( is => 'rw' );

has debug_bifsync => ( is => 'ro' );

has rh => ( is => 'rw' );

has wh => ( is => 'rw' );

sub BUILD {
    my $self = shift;
    $self->trigger_on_update('connecting...');

    if ( $self->location =~ m!^ssh://(.+)! ) {
        $self->child(
            spawn( 'ssh', $1, 'bifsync', $self->debug_bifsync ? '--debug' : (),
            )
        );
    }
    else {
        my @bifsync;
        if ($main::BIF_BIFSYNC) {   # set by t/lib/Test/Bif.pm (for Win32 tests)
            @bifsync = ( $^X, $main::BIF_BIFSYNC );
        }
        elsif ( $^O eq 'MSWin32' ) {    # Assume bifsync next to bif
            @bifsync = ( $^X, Path::Tiny::path($0)->parent->child('bifsync') );
        }
        else {
            @bifsync = ('bifsync');     # rest of world can use PATH
        }

        $self->child(
            spawn(
                @bifsync, $self->debug_bifsync
                ? '--debug'
                : (),
                $self->location
            )
        );
    }

    $self->child_watcher(
        AE::child $self->child->pid,
        sub {
            $self->on_error->('child process ended unexpectedly');
        }
    );

    my $stderr = $self->child->stderr;
    my $name   = $self->name;
    my $debug  = $self->debug;

    $self->stderr_watcher(
        AE::io $stderr,
        0,
        sub {
            # This doesn't need to be every time, but I can't find
            # where Coro is setting default fh back to 1
            state $once = select $App::bif::pager->fh if $App::bif::pager;

            my $line = $stderr->getline;
            if ( !defined $line ) {
                $self->stderr_watcher(undef);
                return;
            }

            # This needs to be the same in the cleanup_errors code if
            # you don't want errors going to fd2...
            if ($debug) {
                $line =~ s!\n$!!;
                $log->debug("$name: $line");
            }
            else {
                warn "$name: $line";
            }
        }
    );

    $self->rh(
        Coro::Handle->new_from_fh( $self->child->stdout, timeout => 30 ) );
    $self->wh(
        Coro::Handle->new_from_fh( $self->child->stdin, timeout => 30 ) );

    $self->new_temp_table;
    return;
}

sub trigger_on_update {
    my $self = shift;

    if ( my $msg = shift ) {
        return $self->on_update->( $self, $msg );
    }

    if ( $self->changes_tosend ) {
        if ( $self->changes_torecv ) {
            $self->on_update->(
                $self,
                'sent: '
                  . ( $self->changes_sent // '' ) . '/'
                  . $self->changes_tosend
                  . ' received: '
                  . ( $self->changes_recv // '' ) . '/'
                  . $self->changes_torecv
                  . ' duplicates: '
                  . $self->changes_dup
            );
        }
        else {
            $self->on_update->(
                $self,
                'sent: '
                  . ( $self->changes_sent // '' ) . '/'
                  . $self->changes_tosend
            );
        }
    }
    elsif ( $self->changes_torecv ) {
        $self->on_update->(
            $self,
            'received: '
              . ( $self->changes_recv // '' ) . '/'
              . $self->changes_torecv
              . ' duplicates: '
              . $self->changes_dup
        );
    }
    else {
        $self->on_update->( $self, 'no changes' );
    }
}

sub bootstrap_identity {
    my $self = shift;

    $self->write( 'IMPORT', 'self' );

    my ( $action, $type, $uuid ) = $self->read;
    return $action
      unless ( $action eq 'EXPORT' and $type eq 'identity', and $uuid );

    require Bif::Sync::Plugin::Identity;

    my $status = $self->real_import_identity;
    return $status unless $status eq 'IdentityImported';

    my $dbw = $self->db;
    my ( $iid, $uid ) = $dbw->xlist(
        select     => [ 'n.id', 'n.first_change_id' ],
        from       => 'nodes n',
        inner_join => 'identities i',
        on         => 'i.id = n.id',
        where => { 'n.uuid' => $uuid, },
    );

    return 'IdentityNotImported' unless $iid;

    $dbw->xdo(
        insert_into => 'bifkv',
        values      => { key => 'self', identity_id => $iid },
    );

    return $status;
}

sub pull_hub {
    my $self = shift;
    my $name = shift;

    $self->write( 'IMPORT', 'hub' );

    my ( $action, $type, $uuid ) = $self->read;
    if ( $action eq 'EXPORT' and $type eq 'hub' ) {
        return 'NoUUID' unless $uuid;
        require Bif::Sync::Plugin::Repo;
        return $self->real_import_hub($uuid);
    }
    return $action;
}

sub sync_hub {
    my $self = shift;
    my $id = shift || die 'sync_hub($id)';

    my $hub = $self->db->xhashref(
        select     => [ 'n.uuid', 'h.hash' ],
        from       => 'hubs h',
        inner_join => 'nodes n',
        on         => 'n.id = h.id',
        where => { 'h.id' => $id },
    );

    $self->write( 'SYNC', 'hub', $hub->{uuid}, $hub->{hash} );
    my ( $action, $type, $uuid, $hash ) = $self->read;

    return 'ProtocolError'
      unless defined $action && $action eq 'SYNC'
      and defined $type      && $type eq 'hub'
      and defined $uuid      && $uuid eq $hub->{uuid}
      and defined $hash;

    if ( $hash eq $hub->{hash} ) {
        $self->trigger_on_update('no changes');
        return 'HubMatch';
    }

    require Bif::Sync::Plugin::Repo;
    return $self->real_sync( 'hub', $id );
}

sub transfer_hub_changes {
    my $self = shift;
    require Bif::Sync::Plugin::Repo;

    $self->write( 'TRANSFER', 'hub_changes' );
    return $self->real_transfer_hub_changes;
}

sub sync_project {
    my $self = shift;
    my $id = shift || die 'sync_project($id)';

    my $pinfo = $self->db->xhashref(
        select     => [ 'n.uuid', 'p.hash', 'n.id' ],
        from       => 'projects p',
        inner_join => 'nodes n',
        on         => 'n.id = p.id',
        where      => {
            'p.id' => $id,
        },
    );

    $self->write( 'SYNC', 'project', $pinfo->{uuid}, $pinfo->{hash} );
    my ( $action, $type, $uuid, $hash ) = $self->read;

    return 'ProtocolError'
      unless defined $action && $action eq 'SYNC'
      and defined $type      && $type eq 'project'
      and defined $uuid      && $uuid eq $pinfo->{uuid}
      and defined $hash;

    if ( $hash eq $pinfo->{hash} ) {
        $self->trigger_on_update('no changes');
        return 'ProjectMatch';
    }

    require Bif::Sync::Plugin::Project;
    return $self->real_sync( 'project', $id );
}

sub transfer_project_related_changes {
    my $self = shift;
    require Bif::Sync::Plugin::Project;

    $self->write( 'TRANSFER', 'project_related_changes' );
    return $self->real_transfer_project_related_changes;
}

sub export_project {
    my $self  = shift;
    my $pinfo = shift;

    $self->write( 'EXPORT', 'project', $pinfo->{uuid}, $pinfo->{path} );

    my ( $action, $type ) = $self->read;
    if ( $action eq 'IMPORT' and $type eq 'project' ) {
        require Bif::Sync::Plugin::Project;
        return $self->real_export_project( $pinfo->{id} );
    }
    return $action;
}

sub cleanup_errors {
    my $self = shift;
    return unless $self->stderr_watcher;
    $self->stderr_watcher(undef);
    return if $^O eq 'MSWin32';

    my $name   = $self->name;
    my $child  = $self->child or return;
    my $stderr = $child->stderr or return;

    $stderr->blocking(0);

    my $debug = $self->debug;
    while ( my $line = $stderr->getline ) {
        if ($debug) {
            $line =~ s!\n$!!;
            $log->debug("$name: $line");
        }
        else {
            warn "$name: $line";
        }
    }

    return;
}

sub disconnect {
    my $self = shift;
    $self->cleanup_errors;

    $self->write('QUIT') if $self->child_watcher;
    $self->child_watcher(undef);

    return unless my $child = $self->child;
    $child->close;
    $child->wait_child;
    $self->child(undef);

    return;
}

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

1;

__END__