lib/Sack/View.pm

package Sack::Node;

use warnings;
use strict;

use IO::Socket::INET;
use File::Slurp;
use Carp qw(confess);
use Data::Dump qw(dump);
use Storable;


sub new {
        my $class = shift;
        my $port = shift;
        my $self  = bless { port => $port }, $class;

        my $pid_path = "/tmp/sack.$port.pid";
        if ( -e $pid_path ) {
                my $pid = read_file $pid_path;
                kill 9, $pid && warn "[$port] kill old $pid\n";
        }
        write_file $pid_path, $$;

        my $sock = IO::Socket::INET->new(
                Listen    => SOMAXCONN,
                LocalAddr => '127.0.0.1',
                LocalPort => $port,
                Proto     => 'tcp',
                Reuse     => 1,
        ) or die "[$port] die $!";

        warn "[$port] accept\n";

        my $client = $sock->accept();

        warn "[$port] connect from ", $client->peerhost, $/;

        while ( 1 ) {

                my $data = Storable::fd_retrieve( $client );
                warn "[$port] <<<< ", dump keys %$data;

                my $result;

                if ( $data->{view} ) {
                        $result = { view => $self->view( $data->{view} ) };
                } elsif ( $data->{data} ) {
                        $self->{data} = delete $data->{data};
                        $result = { data => 'loaded' };
                } elsif ( $data->{exit} ) {
                        warn "[$port] exit";
                        exit;
                } else {
                        warn "[$port] UNKNOWN ", dump( $data ), $/;
                        $result = { 'error' => $data };
                }

                warn "[$port] >>>>\n";
                Storable::store_fd( $result => $client );
        }

}

our $rec;
our $out;

sub view {
        my ( $self, $code ) = @_;

        undef $out;

        my $affected = 0;
        $self->report_start;

        my $coderef = eval "sub { $code }";
        if ( $@ ) {
                warn "ABORT code: $@";
                return;
        }


        foreach my $pos ( 0 .. $#{ $self->{data} } ) {
                $rec = $self->{data}->{$pos};
                if ( ! $rec ) {
                        print STDERR "END @ $pos";
                        last;
                }

                eval { $coderef->() };
                if ( $@ ) {
                        warn "ABORT $pos $@\n";
                        last;
                } else {
                        $affected++;
                }

                $pos % 10000 == 0 ? print STDERR $pos :
                $pos % 1000  == 0 ? print STDERR "."  : 0 ;
        };

        $self->report( "$affected affected" );

}

1;
1	dpavlin	92	package Sack::Node;
2
3			use warnings;
4			use strict;
5
6			use IO::Socket::INET;
7			use File::Slurp;
8			use Carp qw(confess);
9			use Data::Dump qw(dump);
10			use Storable;
11
12
13			sub new {
14			my $class = shift;
15			my $port = shift;
16			my $self = bless { port => $port }, $class;
17
18			my $pid_path = "/tmp/sack.$port.pid";
19			if ( -e $pid_path ) {
20			my $pid = read_file $pid_path;
21			kill 9, $pid && warn "[$port] kill old $pid\n";
22			}
23			write_file $pid_path, $$;
24
25			my $sock = IO::Socket::INET->new(
26			Listen => SOMAXCONN,
27			LocalAddr => '127.0.0.1',
28			LocalPort => $port,
29			Proto => 'tcp',
30			Reuse => 1,
31			) or die "[$port] die $!";
32
33			warn "[$port] accept\n";
34
35			my $client = $sock->accept();
36
37			warn "[$port] connect from ", $client->peerhost, $/;
38
39			while ( 1 ) {
40
41			my $data = Storable::fd_retrieve( $client );
42			warn "[$port] <<<< ", dump keys %$data;
43
44			my $result;
45
46			if ( $data->{view} ) {
47			$result = { view => $self->view( $data->{view} ) };
48			} elsif ( $data->{data} ) {
49			$self->{data} = delete $data->{data};
50			$result = { data => 'loaded' };
51			} elsif ( $data->{exit} ) {
52			warn "[$port] exit";
53			exit;
54			} else {
55			warn "[$port] UNKNOWN ", dump( $data ), $/;
56			$result = { 'error' => $data };
57			}
58
59			warn "[$port] >>>>\n";
60			Storable::store_fd( $result => $client );
61			}
62
63			}
64
65			our $rec;
66			our $out;
67
68			sub view {
69			my ( $self, $code ) = @_;
70
71			undef $out;
72
73			my $affected = 0;
74			$self->report_start;
75
76			my $coderef = eval "sub { $code }";
77			if ( $@ ) {
78			warn "ABORT code: $@";
79			return;
80			}
81
82
83			foreach my $pos ( 0 .. $#{ $self->{data} } ) {
84			$rec = $self->{data}->{$pos};
85			if ( ! $rec ) {
86			print STDERR "END @ $pos";
87			last;
88			}
89
90			eval { $coderef->() };
91			if ( $@ ) {
92			warn "ABORT $pos $@\n";
93			last;
94			} else {
95			$affected++;
96			}
97
98			$pos % 10000 == 0 ? print STDERR $pos :
99			$pos % 1000 == 0 ? print STDERR "." : 0 ;
100			};
101
102			$self->report( "$affected affected" );
103
104			}
105
106			1;