| … | |
… | |
| 197 | unreg_groups $node; |
197 | unreg_groups $node; |
| 198 | delete $port{$node}; |
198 | delete $port{$node}; |
| 199 | }; |
199 | }; |
| 200 | |
200 | |
| 201 | snd $port, addr => $AnyEvent::MP::Kernel::LISTENER; |
201 | snd $port, addr => $AnyEvent::MP::Kernel::LISTENER; |
| 202 | snd $port, connect_nodes => \%addr if %addr; |
202 | snd $port, connect => \%addr if %addr; |
| 203 | snd $port, set => \%lreg if %lreg; |
203 | snd $port, set => \%lreg if %lreg; |
| 204 | } |
204 | } |
| 205 | |
205 | |
| 206 | # other nodes connect via this |
206 | # other nodes connect via this |
| 207 | sub connect { |
207 | sub connect { |
| … | |
… | |
| 221 | if (@$addresses) { |
221 | if (@$addresses) { |
| 222 | for my $other (values %AnyEvent::MP::NODE) { |
222 | for my $other (values %AnyEvent::MP::NODE) { |
| 223 | if ($other->{transport}) { |
223 | if ($other->{transport}) { |
| 224 | if ($addr{$other->{id}} && !@{ $addr{$other->{id}} }) { |
224 | if ($addr{$other->{id}} && !@{ $addr{$other->{id}} }) { |
| 225 | $AnyEvent::MP::Kernel::WARN->(9, "helping $other->{id} to find $node."); |
225 | $AnyEvent::MP::Kernel::WARN->(9, "helping $other->{id} to find $node."); |
| 226 | snd $port{$other->{id}}, connect_nodes => { $node => $addresses }; |
226 | snd $port{$other->{id}}, connect => { $node => $addresses }; |
| 227 | } |
227 | } |
| 228 | } |
228 | } |
| 229 | } |
229 | } |
| 230 | } |
230 | } |
| 231 | }, |
231 | }, |
| 232 | connect_nodes => sub { |
232 | connect => sub { |
| 233 | my ($kv) = @_; |
233 | my ($kv) = @_; |
| 234 | |
234 | |
| 235 | use JSON::XS;#d# |
235 | use JSON::XS;#d# |
| 236 | my $kv_txt = JSON::XS->new->encode ($kv);#d# |
236 | my $kv_txt = JSON::XS->new->encode ($kv);#d# |
| 237 | $AnyEvent::MP::Kernel::WARN->(9, "$node told us it knows about $kv_txt.");#d# |
237 | $AnyEvent::MP::Kernel::WARN->(9, "$node told us it knows about $kv_txt.");#d# |
| … | |
… | |
| 240 | my $node = AnyEvent::MP::Kernel::add_node $id; |
240 | my $node = AnyEvent::MP::Kernel::add_node $id; |
| 241 | $node->connect (@$addresses); |
241 | $node->connect (@$addresses); |
| 242 | start_node $id; |
242 | start_node $id; |
| 243 | } |
243 | } |
| 244 | }, |
244 | }, |
| 245 | find_node => sub { |
245 | find => sub { |
| 246 | my ($othernode) = @_; |
246 | my ($othernode) = @_; |
| 247 | |
247 | |
| 248 | $AnyEvent::MP::Kernel::WARN->(9, "$node asked us to find $othernode."); |
248 | $AnyEvent::MP::Kernel::WARN->(9, "$node asked us to find $othernode."); |
| 249 | snd $port{$node}, connect_nodes => { $othernode => $addr{$othernode} } |
249 | snd $port{$node}, connect => { $othernode => $addr{$othernode} } |
| 250 | if $addr{$othernode}; |
250 | if $addr{$othernode}; |
| 251 | }, |
251 | }, |
| 252 | set => sub { |
252 | set => sub { |
| 253 | set_groups $node, shift; |
253 | set_groups $node, shift; |
| 254 | }, |
254 | }, |
| … | |
… | |
| 269 | unreg_groups $node; |
269 | unreg_groups $node; |
| 270 | |
270 | |
| 271 | # forget about the node |
271 | # forget about the node |
| 272 | delete $addr{$node}; |
272 | delete $addr{$node}; |
| 273 | # ask other nodes if they know the node |
273 | # ask other nodes if they know the node |
| 274 | snd $_, find_node => $node |
274 | snd $_, find => $node |
| 275 | for values %port; |
275 | for values %port; |
| 276 | } |
276 | } |
| 277 | #warn "node<$node,$is_up>\n";#d# |
277 | #warn "node<$node,$is_up>\n";#d# |
| 278 | } |
278 | } |
| 279 | |
279 | |
