% Since version 1.8
% Demonstrate the utility predicates "init_join" and "join" that allow parallel streams of processing to be joined.
% These predicates reside in utils.prova and should be consulted from there.

:- eval(consult("utils.prova")).

:- eval(process_join()).

process_join() :-
	println(["==========process_join=========="]),
	iam(Me),
	% A more or less useless message to signal the start of the conversation. It may do something useful in a real application. 
	sendMsg(XID,self,Me,start,[]),
	% The result of the above is that we now have the conversation-id XID initialised.
	% Create a join predicate join_1 with the list of required input patterns in the last argument.
	init_join(XID,join_1,[c(_),b(_)]),
	% This will create two parallel processing streams.
	% Note that Prova does not run them in separate threads but they are independent and can communicate
	%    with other agents asynchronously effectively allowing for parallel processing.
	fork_a_b(Me,XID).

fork_a_b(Me,XID) :-
	sendMsg(XID,self,Me,reply,a(1),corr_a),
	rcvMsg(XID,self,Me,reply,a(1),corr_a),
	fork_c_d(Me,XID).
fork_a_b(Me,XID) :-
	sendMsg(XID,self,Me,reply,b(1),corr_b),
	rcvMsg(XID,self,Me,reply,b(1),corr_b),
	% Tell the join join_1 that a new pattern is ready
	join(Me,XID,join_1,b(1)).

fork_c_d(Me,XID) :-
	sendMsg(XID,self,Me,reply,c(1),corr_c),
	rcvMsg(XID,self,Me,reply,c(1),corr_c),
	% Tell the join join_1 that a new pattern is ready
	join(Me,XID,join_1,c(1)).
fork_c_d(Me) :-
	sendMsg(XID,self,Me,reply,d(1),corr_d),
	rcvMsg(XID,self,Me,reply,d(1),corr_d).

% The following rule will be invoked by join once all the inputs are assembled.
join_1(Me,XID,Inputs) :-
	println(["Joined for XID=",XID," with inputs: ",Inputs]).

% A testing harness for printing all incoming 'reply' messages.
rcvMsg(XID,Protocol,From,reply,[X|Xs]|Extra) :-
	println(["reply for conversation-id ",XID,": ",[X|Xs],","|Extra]).
