[ViewVC] Diff of: cvs/AnyEvent-MP/MP.pm

Comparing AnyEvent-MP/MP.pm (file contents):
Revision 1.2 by root, Fri Jul 31 20:55:46 2009 UTC vs.
Revision 1.6 by root, Sat Aug 1 10:02:33 2009 UTC

…		…
27	This module (-family) implements a simple message passing framework.	27	This module (-family) implements a simple message passing framework.
28		28
29	Despite its simplicity, you can securely message other processes running	29	Despite its simplicity, you can securely message other processes running
30	on the same or other hosts.	30	on the same or other hosts.
31		31
		32	At the moment, this module family is severly brokena nd underdocumented,
		33	so do not use. This was uploaded mainly to resreve the CPAN namespace -
		34	stay tuned!
		35
32	=head1 CONCEPTS	36	=head1 CONCEPTS
33		37
34	=over 4	38	=over 4
35		39
36	=item port	40	=item port
37		41
38	A port is something you can send messages to with the C<snd> function, and	42	A port is something you can send messages to with the C<snd> function, and
39	you can register C<rcv> handlers with. All C<rcv> handlers will receive	43	you can register C<rcv> handlers with. All C<rcv> handlers will receive
40	messages they match, messages will not be queued.	44	messages they match, messages will not be queued.
41		45
42	=item port id - C<pid@host#portname>	46	=item port id - C<noderef#portname>
43		47
44	A port id is always the node id, a hash-mark (C<#>) as separator, followed	48	A port id is always the noderef, a hash-mark (C<#>) as separator, followed
45	by a port name.	49	by a port name (a printable string of unspecified format).
46
47	A port name can be a well known port (basically an identifier/bareword),
48	or a generated name, consisting of node id, a dot (C<.>), and an
49	identifier.
50		50
51	=item node	51	=item node
52		52
53	A node is a single process containing at least one port - the node	53	A node is a single process containing at least one port - the node
54	port. You can send messages to node ports to let them create new ports,	54	port. You can send messages to node ports to let them create new ports,
55	among other things.	55	among other things.
56		56
57	Initially, nodes are either private (single-process only) or hidden	57	Initially, nodes are either private (single-process only) or hidden
58	(connected to a father node only). Only when they epxlicitly "go public"	58	(connected to a master node only). Only when they epxlicitly "become
59	can you send them messages form unrelated other nodes.	59	public" can you send them messages from unrelated other nodes.
60		60
61	Public nodes automatically connect to all other public nodes in a network	61	=item noderef - C<host:port,host:port...>, C<id@noderef>, C<id>
62	when they connect, creating a full mesh.
63		62
64	=item node id - C<host:port>, C<id@host>, C<id>
65
66	A node ID is a string that either uniquely identifies a given node (For	63	A noderef is a string that either uniquely identifies a given node (for
67	private and hidden nodes), or contains a recipe on how to reach a given	64	private and hidden nodes), or contains a recipe on how to reach a given
68	node (for public nodes).	65	node (for public nodes).
69		66
70	=back	67	=back
71		68
72	=head1 FUNCTIONS	69	=head1 VARIABLES/FUNCTIONS
73		70
74	=over 4	71	=over 4
75		72
76	=cut	73	=cut
77		74
…		…
88		85
89	use AE ();	86	use AE ();
90		87
91	use base "Exporter";	88	use base "Exporter";
92		89
93	our $VERSION = '0.0';	90	our $VERSION = '0.01';
94	our @EXPORT = qw(NODE $NODE $PORT snd rcv _any_);	91	our @EXPORT = qw(NODE $NODE $PORT snd rcv _any_);
95		92
96	our $DEFAULT_SECRET;	93	our $DEFAULT_SECRET;
97	our $DEFAULT_PORT = "4040";	94	our $DEFAULT_PORT = "4040";
98		95
…		…
109	}	106	}
110		107
111	$DEFAULT_SECRET	108	$DEFAULT_SECRET
112	}	109	}
113		110
		111	=item NODE / $NODE
		112
		113	The C<NODE ()> function and the C<$NODE> variable contain the noderef of
		114	the local node. The value is initialised by a call to C<become_public> or
		115	C<become_slave>, after which all local port identifiers become invalid.
		116
		117	=cut
		118
114	our $UNIQ = sprintf "%x.%x", $$, time; # per-process/node unique cookie	119	our $UNIQ = sprintf "%x.%x", $$, time; # per-process/node unique cookie
		120	our $ID = "a0";
115	our $PUBLIC = 0;	121	our $PUBLIC = 0;
116	our $NODE;	122	our $NODE;
117	our $PORT;	123	our $PORT;
118		124
119	our %NODE; # node id to transport mapping, or "undef", for local node	125	our %NODE; # node id to transport mapping, or "undef", for local node
…		…
149	for $noderef, split /,/, $noderef;	155	for $noderef, split /,/, $noderef;
150		156
151	$node	157	$node
152	}	158	}
153		159
		160	=item snd $portid, type => @data
		161
		162	=item snd $portid, @msg
		163
		164	Send the given message to the given port ID, which can identify either a
		165	local or a remote port.
		166
		167	While the message can be about anything, it is highly recommended to use
		168	a constant string as first element.
		169
		170	The message data effectively becomes read-only after a call to this
		171	function: modifying any argument is not allowed and can cause many
		172	problems.
		173
		174	The type of data you can transfer depends on the transport protocol: when
		175	JSON is used, then only strings, numbers and arrays and hashes consisting
		176	of those are allowed (no objects). When Storable is used, then anything
		177	that Storable can serialise and deserialise is allowed, and for the local
		178	node, anything can be passed.
		179
		180	=cut
		181
154	sub snd($@) {	182	sub snd(@) {
155	my ($noderef, $port) = split /#/, shift, 2;	183	my ($noderef, $port) = split /#/, shift, 2;
156		184
157	add_node $noderef	185	add_node $noderef
158	unless exists $NODE{$noderef};	186	unless exists $NODE{$noderef};
159		187
160	$NODE{$noderef}->send ([$port, [@_]]);	188	$NODE{$noderef}->send (["$port", [@_]]);
		189	}
		190
		191	=item rcv $portid, type => $callback->(@msg)
		192
		193	=item rcv $portid, $smartmatch => $callback->(@msg)
		194
		195	=item rcv $portid, [$smartmatch...] => $callback->(@msg)
		196
		197	Register a callback on the port identified by C<$portid>, which I<must> be
		198	a local port.
		199
		200	The callback has to return a true value when its work is done, after
		201	which is will be removed, or a false value in which case it will stay
		202	registered.
		203
		204	If the match is an array reference, then it will be matched against the
		205	first elements of the message, otherwise only the first element is being
		206	matched.
		207
		208	Any element in the match that is specified as C<_any_> (a function
		209	exported by this module) matches any single element of the message.
		210
		211	While not required, it is highly recommended that the first matching
		212	element is a string identifying the message. The one-string-only match is
		213	also the most efficient match (by far).
		214
		215	=cut
		216
		217	sub rcv($@) {
		218	my ($port, $match, $cb) = @_;
		219
		220	my $port = $PORT{$port}
		221	or do {
		222	my ($noderef, $lport) = split /#/, $port;
		223	"AnyEvent::MP::Node::Self" eq ref $NODE{$noderef}
		224	or Carp::croak "$port: can only rcv on local ports";
		225
		226	$PORT{$lport}
		227	or Carp::croak "$port: port does not exist";
		228
		229	$PORT{$port} = $PORT{$lport} # also return
		230	};
		231
		232	if (!ref $match) {
		233	push @{ $port->{rc0}{$match} }, [$cb];
		234	} elsif (("ARRAY" eq ref $match && !ref $match->[0])) {
		235	my ($type, @match) = @$match;
		236	@match
		237	? push @{ $port->{rcv}{$match->[0]} }, [$cb, \@match]
		238	: push @{ $port->{rc0}{$match->[0]} }, [$cb];
		239	} else {
		240	push @{ $port->{any} }, [$cb, $match];
		241	}
161	}	242	}
162		243
163	sub _inject {	244	sub _inject {
164	my ($port, $msg) = @{+shift};	245	my ($port, $msg) = @{+shift};
165		246
166	$port = $PORT{$port}	247	$port = $PORT{$port}
167	or return;	248	or return;
168		249
169	use Data::Dumper;	250	@_ = @$msg;
170	warn Dumper $msg;	251
		252	for (@{ $port->{rc0}{$msg->[0]} }) {
		253	$_ && &{$_->[0]}
		254	&& undef $_;
		255	}
		256
		257	for (@{ $port->{rcv}{$msg->[0]} }) {
		258	$_ && [@_[1..$#{$_->[1]}]] ~~ $_->[1]
		259	&& &{$_->[0]}
		260	&& undef $_;
		261	}
		262
		263	for (@{ $port->{any} }) {
		264	$_ && [@_[0..$#{$_->[1]}]] ~~ $_->[1]
		265	&& &{$_->[0]}
		266	&& undef $_;
		267	}
171	}	268	}
172		269
173	sub normalise_noderef($) {	270	sub normalise_noderef($) {
174	my ($noderef) = @_;	271	my ($noderef) = @_;
175		272
…		…
244	my $noderef = join ",", ref $_[0] ? @{+shift} : shift;	341	my $noderef = join ",", ref $_[0] ? @{+shift} : shift;
245	my @args = @_;	342	my @args = @_;
246		343
247	$NODE = (normalise_noderef $noderef)->recv;	344	$NODE = (normalise_noderef $noderef)->recv;
248		345
249	my $self = new AnyEvent::MP::Node::Self noderef => $NODE;
250
251	$NODE{""} = $self; # empty string == local node
252
253	for my $t (split /,/, $NODE) {	346	for my $t (split /,/, $NODE) {
254	$NODE{$t} = $self;	347	$NODE{$t} = $NODE{""};
255		348
256	my ($host, $port) = AnyEvent::Socket::parse_hostport $t;	349	my ($host, $port) = AnyEvent::Socket::parse_hostport $t;
257		350
258	$LISTENER{$t} = AnyEvent::MP::Transport::mp_server $host, $port,	351	$LISTENER{$t} = AnyEvent::MP::Transport::mp_server $host, $port,
259	@args,	352	@args,
…		…
276	$PUBLIC = 1;	369	$PUBLIC = 1;
277	}	370	}
278		371
279	=back	372	=back
280		373
		374	=head1 NODE MESSAGES
		375
		376	Nodes understand the following messages sent to them. Many of them take
		377	arguments called C<@reply>, which will simply be used to compose a reply
		378	message - C<$reply[0]> is the port to reply to, C<$reply[1]> the type and
		379	the remaining arguments are simply the message data.
		380
		381	=over 4
		382
		383	=cut
		384
		385	#############################################################################
		386	# self node code
		387
		388	sub _new_port($) {
		389	my ($name) = @_;
		390
		391	my ($noderef, $portname) = split /#/, $name;
		392
		393	$PORT{$name} =
		394	$PORT{$portname} = {
		395	names => [$name, $portname],
		396	};
		397	}
		398
		399	$NODE{""} = new AnyEvent::MP::Node::Self noderef => $NODE;
		400	_new_port "";
		401
		402	=item relay => $port, @msg
		403
		404	Simply forwards the message to the given port.
		405
		406	=cut
		407
		408	rcv "", relay => \&snd;
		409
		410	=item eval => $string[ @reply]
		411
		412	Evaluates the given string. If C<@reply> is given, then a message of the
		413	form C<@reply, $@, @evalres> is sent.
		414
		415	Example: crash another node.
		416
		417	snd $othernode, eval => "exit";
		418
		419	=cut
		420
		421	rcv "", eval => sub {
		422	my (undef, $string, @reply) = @_;
		423	my @res = eval $string;
		424	snd @reply, "$@", @res if @reply;
		425	};
		426
		427	=item time => @reply
		428
		429	Replies the the current node time to C<@reply>.
		430
		431	Example: tell the current node to send the current time to C<$myport> in a
		432	C<timereply> message.
		433
		434	snd $NODE, time => $myport, timereply => 1, 2;
		435	# => snd $myport, timereply => 1, 2, <time>
		436
		437	=cut
		438
		439	rcv "", time => sub { shift; snd @_, AE::time };
		440
		441	=back
		442
281	=head1 SEE ALSO	443	=head1 SEE ALSO
282		444
283	L<AnyEvent>.	445	L<AnyEvent>.
284		446
285	=head1 AUTHOR	447	=head1 AUTHOR

Diff Legend

-–
+Removed lines
-+
+Added lines
-<
+Changed lines
->
+Changed lines

Comparing AnyEvent-MP/MP.pm (file contents): Revision 1.2 by root, Fri Jul 31 20:55:46 2009 UTC vs. Revision 1.6 by root, Sat Aug 1 10:02:33 2009 UTC

Diff Legend

Comparing AnyEvent-MP/MP.pm (file contents):
Revision 1.2 by root, Fri Jul 31 20:55:46 2009 UTC vs.
Revision 1.6 by root, Sat Aug 1 10:02:33 2009 UTC