3 package Cog; # persistant Gearman Worker - see the pun?
7 extends 'Gearman::Driver::Worker::Base';
10 with 'APKPM::Gearman';
11 with 'APKPM::Gearman::Client';
18 use Data::Dump qw(dump);
20 use Regexp::Common qw(net);
27 my $variant = $ARGV[0] || 'ZTEDSLAM';
28 warn "variant: $variant\n";
30 my $driver = Gearman::Driver->new(
31 server => 'localhost:4730',
34 logfile => "log/$variant.log",
38 my $worker = $w1->new();
41 my $crm = $worker->do( 'CRM_search' => "TIP_UREDJAJA:$variant" );
46 foreach my $user ( @$crm ) {
47 next unless $user->{IP_UREDAJA} =~ /$RE{net}{IPv4}/;
48 $poll->{ $user->{IP_UREDAJA} }->{ $user->{SHELF_SLOT_PORT} }++;
51 warn "# poll = ", dump($poll);
53 foreach my $ip ( keys %$poll ) {
55 my $method = $variant . '_' . $ip;
57 my $module = 'H1::' . $variant;
58 my $zte = $module->new( ip => $ip );
61 max_processes => 1, # FIXME increase?
74 my ( $self, $job, $crm ) = @_;
76 my $redis = Redis->new;
78 if ( exists $crm->{logout} ) {
80 $redis->sadd("$variant.$ip.logout", $self->datetime_now);
81 return { logout => 'ok' };
84 my $port = $crm->{SHELF_SLOT_PORT} || die "no SHELF_SLOT_PORT";
85 my $username = $crm->{USERNAME} || die "no USERNAME";
88 $redis->sadd("$variant.invalid.SHELF_SLOT_PORT" => $username );
89 return "invalid port $port";
93 eval { $hash = $zte->hash( $port ) };
96 $redis->sadd("$variant.$ip.error" => $@);
99 $redis->sadd("$variant.$ip.empty" => $port);
102 $redis->sadd("$variant.$ip.ok" => $port);
105 $self->do_background_json( 'Store_insert', {
107 ip => $crm->{IP_MANAGEMENT}, # FIXME IP_UREDAJA ?
108 username => $username,
109 timestamp => $self->datetime_now,
110 h => $self->to_hstore($hash),
113 return "ok $username $ip $port"; # body end
121 warn "$method added\n";
129 name => "poll_$variant",
131 name => "poll_$variant",
134 my ( $self, $job, $workload ) = @_;
136 my $crm = $self->do( 'CRM_search' => "TIP_UREDJAJA:$variant" );
140 my $gc = Gearman::Client->new;
141 $gc->job_servers( $self->server || 'localhost:4730' );
142 my $taskset = $gc->new_task_set;
145 my $redis = Redis->new;
146 $redis->del( $_ ) foreach $redis->keys("$variant.*");
148 $redis->set( "$variant.time.start" => $self->datetime_now );
150 foreach my $user ( @$crm ) {
151 my $ip = $user->{IP_UREDAJA};
152 my $port = $user->{SHELF_SLOT_PORT};
154 if ( $ip !~ /$RE{net}{IPv4}/ ) {
155 $redis->sadd("$variant.invalid.IP_UREDAJA" => $ip);
157 } elsif ( $port !~ m{\d+(/\d+)+} ) {
158 $redis->sadd("$variant.$ip.invalid.PORT" => $port);
160 } elsif ( $seen->{ $ip }->{ $port }++ ) {
161 $redis->sadd("$variant.$ip.duplicate" => $port);
164 $redis->incr("$variant.queued");
165 $redis->sadd("$variant.$ip.queued" => $port);
167 my $name = $variant . '_' . $ip;
168 $taskset->add_task( $name, $self->e_json( $user ), {
169 on_complete => sub { $redis->sadd("$variant.$ip.complete" => $port) },
170 on_fail => sub { $redis->sadd("$variant.$ip.failed" => $port) },
174 warn "# queue logouts";
175 foreach my $ip ( keys %$seen ) {
176 $taskset->add_task( $variant . '_' . $ip, $self->e_json( { logout => 1 } ) );
182 $redis->set( "$variant.time.finish" => $self->datetime_now );
183 warn "# seen ", dump($seen);
185 my $ips = scalar keys %$seen;
186 return "polled $ips IPs"; # body end