1		% (c) 2024-2026 Lehrstuhl fuer Softwaretechnik und Programmiersprachen,
2		% Heinrich Heine Universitaet Duesseldorf
3		% This software is licenced under EPL 1.0 (http://www.eclipse.org/org/documents/epl-v10.html
4
5		:- module(zmq_rpc, [zmq_rpc_init/2,
6		stream_socket_rpc_init/2,
7		zmq_rpc_destroy/1,
8		zmq_rpc_destroy_all/0,
9		zmq_rpc_send/4,
10
11		stream_socket_rpc_accept_request/3,
12		stream_socket_rpc_send_answer/3,
13		close_stream_server_socket/1
14		]).
15
16		:- use_module(probsrc(module_information)).
17		:- module_info(group, zmq).
18		:- module_info(description, 'RPC via ZMQ implementation (using JSON-RPC 2.0)').
19
20		% Using https://www.jsonrpc.org/specification
21		% We do not support notifications
22		% example JSON-RPC request sent for ZMQ_RPC_SEND(x,"hello",{"a"|->JsonNull}):
23		% {"jsonrpc":"2.0","method":"hello","params":{"a":null},"id":1}
24		% JsonValue = JsonNull, JsonBoolean(BOOL),
25		% JsonNumber(FLOAT), JsonString(STRING),
26		% JsonArray(seq(JsonValue)), JsonObject(STRING +-> JsonValue);
27		% RpcResult = RpcSuccess(JsonValue), RpcError(STRING)
28
29		:- use_module(probsrc(eventhandling), [register_event_listener/3]).
30		:- register_event_listener(reset_specification, zmq_rpc_destroy_all, 'ZMQ cleanup').
31
32		:- use_module(probsrc(pathes_lib), [safe_load_foreign_resource/2]).
33		:- use_module(probsrc(custom_explicit_sets), [expand_custom_set_to_list/2,convert_to_avl/2]).
34		:- use_module(probsrc(tools_strings), [ajoin/2]).
35		:- use_module(extrasrc(json_parser), [json_parse/2, json_write/2]).
36		:- use_module(extrasrc(json_freetype), [json_parser_to_json_freetype/2, json_freetype_to_json_parser/2]).
37		:- use_module(probsrc(error_manager),[add_error/3, add_warning/3, add_internal_error/2, add_message/3]).
38		:- use_module(probsrc(debug),[debug_println/2, debug_format/3]).
39		:- use_module(library(lists)).
40		:- use_module(library(codesio), [with_output_to_codes/2]).
41		:- use_module(library(sockets)).
42		:- use_module(probsrc(gensym),[gennum/1]).
43
44		foreign(zmq_rpc_init_impl, zmq_rpc_init_impl(+string, [-address])).
45		foreign(zmq_rpc_destroy_impl, zmq_rpc_destroy_impl(+address)).
46		foreign(zmq_rpc_send_str_impl, zmq_rpc_send_str_impl(+address, +codes)).
47		foreign(zmq_rpc_receive_str_impl, zmq_rpc_receive_str_impl(+address, [-codes])).
48		foreign_resource(zmq_rpc, [zmq_rpc_init_impl, zmq_rpc_destroy_impl, zmq_rpc_send_str_impl, zmq_rpc_receive_str_impl]).
49
50		:- dynamic loaded/0.
51
52		init_zmq_rpc :-
53		(loaded -> true
54		; assertz(loaded),
55		safe_load_foreign_resource(zmq_rpc, zmq_rpc)
56		).
57
58		:- dynamic open_socket/2.
59		% open_socket(InternalSockentNrName,SocketTerm)
60		% SocketTerm is either integer for ZMQ or sicstus_stream_socket(Stream)
61
62		% API to decide how to send/receive messages depending on socket type
63		rpc_send_codes_request(sicstus_stream_socket(Stream),Codes) :- !,
64		% send request as a single line; supposes Codes does not contain newline characters
65		%format(user_output,'>>> ~s~n',[Codes]),
66		format(Stream,'~s~n',[Codes]), flush_output(Stream).
67		rpc_send_codes_request(Socket, ReqCodes) :- % send request via ZMQ API
68		catch_zmq_error(zmq_rpc_send_str_impl(Socket,ReqCodes),'Cannot send ZMQ request:').
69
70		rpc_receive_codes_result(sicstus_stream_socket(Stream),Codes) :- !, read_line(Stream,Codes).
71		rpc_receive_codes_result(Socket, ReqCodes) :-
72		% after a failed receive, the ZMQ (REQ) socket cannot send another request and has to be destroyed
73		catch_zmq_error(zmq_rpc_receive_str_impl(Socket,ReqCodes),'Cannot receive ZMQ result:').
74
75		rpc_destroy_socket(sicstus_stream_socket(Stream)) :- !, close(Stream).
76		rpc_destroy_socket(Socket) :- catch_zmq_error(zmq_rpc_destroy_impl(Socket),'Cannot destroy ZMQ socket:').
77
78		% the C functions raise error(system_error,system_error(zmq_error(Function,Message))) if a ZMQ call fails
79		catch_zmq_error(Call,Msg) :-
80		catch(Call,error(system_error,system_error(ZmqError)),
81		(add_error(zmq_rpc,Msg,ZmqError),fail)).
82
83		rpc_init(sicstus_stream_socket(_)) :- !. % no init required for these kinds of sockets
84		rpc_init(_) :- init_zmq_rpc.
85
86		% ------------------------
87
88		% set-up a regular Stream socket using single-line ndjson messages
89		stream_socket_rpc_init(PortTerm,Out) :-
90		% TODO: also allow to pass nodename, with empty ('') a connection is made to local machine
91		catch(socket_client_open('':PortTerm, Stream, [type(text),encoding(utf8)]),Exc,
92		(add_error(zmq_rpc,'Cannot open socket as client:',Exc),fail)),
93		gennum(GN),
94		SocketNr is -1-GN, % we assume negative numbers not used by ZMQ,
95		% makes API easier to use/type for end-users than returning e.g. string(Name)
96		(open_socket(SocketNr,_) -> add_internal_error('Socket number clash:',SocketNr) ; true),
97		assertz(open_socket(SocketNr,sicstus_stream_socket(Stream))),
98		Out = int(SocketNr).
99
100
101		% ------------------------
102
103		% set-up a ZMQ socket
104		zmq_rpc_init(Endpoint, Out) :-
105		atom(Endpoint),
106		init_zmq_rpc,
107		catch_zmq_error(zmq_rpc_init_impl(Endpoint, Res),'Cannot open ZMQ socket:'),
108		debug_println(9,zmq_rpc_init_impl(Endpoint, Res)),
109		integer(Res),
110		assertz(open_socket(Res,Res)),
111		Out = int(Res).
112
113		% ------------------------
114
115		% destroy a socket with internal number SocketNr, works with all kinds of sockets
116		zmq_rpc_destroy(Socket) :- \+ open_socket(Socket,_),!,
117		add_error(zmq_rpc_destroy,'Socket not open for ZMQ_RPC_DESTROY (use ZMQ_RPC_INIT or SOCKET_RPC_INIT first):',Socket),
118		fail.
119		zmq_rpc_destroy(Socket) :-
120		open_socket(Socket,SocketTerm),
121		retractall(open_socket(Socket,_)), % also if destroying fails, so that zmq_rpc_destroy_all terminates
122		rpc_init(SocketTerm),
123		rpc_destroy_socket(SocketTerm).
124
125		% also closes the server sockets of stream_socket_rpc_accept_request
126		zmq_rpc_destroy_all :-
127		(open_socket(Socket,_) -> zmq_rpc_destroy(Socket), zmq_rpc_destroy_all
128		; open_server_socket(Port,_) -> close_stream_server_socket(Port), zmq_rpc_destroy_all
129		; true).
130
131		% ------------------------
132
133		% send a request on a socket; works with all sockets
134		zmq_rpc_send(Socket, _Name, _Args, _ResAst) :- \+ open_socket(Socket,_),!,
135		add_error(zmq_rpc_send,'Socket not open for ZMQ_RPC_SEND (use ZMQ_RPC_INIT first):',Socket),
136		fail.
137		zmq_rpc_send(SocketNr, Name, Args, ResAst) :-
138		open_socket(SocketNr,Socket),
139		atom(Name),
140		!,
141		next_rpc_request_id(Id),
142		write_json_rpc_request(Name, Args, Id, JsonRpcReq),
143		safe_json_write(JsonRpcReq, ReqCodes),
144		rpc_init(Socket),
145		debug_format(19,'Sending JSON-RPC on socket ~w: ~s~n',[SocketNr,ReqCodes]),
146		rpc_send_codes_request(Socket, ReqCodes),
147		rpc_receive_codes_result(Socket, ResCodes),
148		debug_format(19,'Received JSON-RPC result on socket ~w: ~s~n',[SocketNr,ResCodes]),
149		(ResCodes=end_of_file -> add_error(zmq_rpc,'JSON-RPC server not responding:',ResCodes),fail
150		; json_parse(ResCodes, ResJson) -> true
151		; add_error(zmq_rpc,'Cannot parse JSON-RPC result:',ResCodes),fail),
152		extract_json_rpc_result(ResJson, ReceivedId, ResAst),
153		(ReceivedId=Id -> true
154		; ajoin(['Id ',ReceivedId,' of JSON-RPC result object does not match expected id:'],Msg),
155		add_error(zmq_rpc,Msg,Id),
156		fail
157		).
158		zmq_rpc_send(Socket, Name, ArgsAst, ResAst) :-
159		add_internal_error('Illegal call to ZMQ_RPC_SEND:',zmq_rpc_send(Socket, Name, ArgsAst, ResAst)),
160		fail.
161
162		write_json_rpc_request(Method, Params, Id, Json) :-
163		json_freetype_to_json_parser(freeval('JsonValue', 'JsonObject', Params), JsonParams),
164		(number(Id) -> JsonId = number(Id) ; JsonId = string(Id)),
165		Json = json([jsonrpc=string('2.0'), method=string(Method), params=JsonParams, id=JsonId]).
166
167		write_json_rpc_answer(Id,freeval('RpcResult', 'RpcSuccess', ResultTerm),Json) :- !,
168		write_json_rpc_success_answer(Id, ResultTerm, Json).
169		write_json_rpc_answer(Id,freeval('RpcResult', 'RpcError', string(ErrorMsg)),Json) :- !,
170		write_json_rpc_error_answer(Id, ErrorMsg, Json).
171		write_json_rpc_answer(Id,Other,Json) :-
172		add_error(zmq_rpc,'Not a valid RpcResult freevalue:',Other),
173		write_json_rpc_error_answer(Id, 'Internal server error: illegal result object', Json).
174
175		write_json_rpc_success_answer(Id, ResultTerm, Json) :-
176		json_freetype_to_json_parser(ResultTerm, JsonResult),
177		(number(Id) -> JsonId = number(Id) ; JsonId = string(Id)),
178		Json = json([jsonrpc=string('2.0'), result=JsonResult, id=JsonId]).
179		write_json_rpc_error_answer(Id, String, Json) :-
180		(number(Id) -> JsonId = number(Id) ; JsonId = string(Id)),
181		Json = json([jsonrpc=string('2.0'), error=string(String), id=JsonId]).
182
183
184		:- dynamic last_rpc_request_id/1.
185		next_rpc_request_id(Id) :-
186		(last_rpc_request_id(Last) -> Id is Last+1 ; Id=1),
187		retractall(last_rpc_request_id(_)),
188		assertz(last_rpc_request_id(Id)).
189
190		extract_json_rpc_result(json(JsonObject), Id, ResAst) :-
191		member(jsonrpc=string(Version), JsonObject), !,
192		(Version='2.0' -> true ; add_warning(zmq_rpc,'Expected version 2.0 JSON-RPC object: ',Version)),
193		(memberchk(id=JsonId, JsonObject)
194		-> (JsonId = number(Id) -> true ; JsonId = string(Id) -> true ;
195		add_error(zmq_rpc,'Cannot extract valid id from result: ',JsonId),fail
196		)
197		; add_error(zmq_rpc,'JSON-RPC result object has no id: ',JsonObject),
198		fail
199		),
200		% "error":null (as sent by JSON-RPC 1.0 servers together with the result) is not an error
201		(member(error=E, JsonObject), E \= '@'(null)
202		-> extract_error_result(E, WrappedRes), ResAst = freeval('RpcResult', 'RpcError', WrappedRes)
203		; extract_success_result(JsonObject, WrappedRes), ResAst = freeval('RpcResult', 'RpcSuccess', WrappedRes)
204		).
205		extract_json_rpc_result(Obj, _Id, _ResAst) :-
206		add_error(zmq_rpc,'Response object does not conform to JSON-RPC:',Obj),
207		fail.
208
209		extract_error_result(E, string(Res)) :-
210		(E=json(ErrorObject) -> true ; ErrorObject=[]),
211		(member(code=number(Code), ErrorObject) -> true ; Code = -1),
212		(member(message=string(Message), ErrorObject) -> true ; Message = 'N/A'),
213		ajoin(['error ', Code, ': ', Message], Res).
214
215		extract_success_result(JsonObject, ResAst) :-
216		member(result=Res, JsonObject), !,
217		json_parser_to_json_freetype(Res, ResAst).
218		extract_success_result(Obj, _ResAst) :-
219		add_error(zmq_rpc,'Response object has no result attribute:',Obj),
220		fail.
221
222
223		% ------------------------
224		% a simple server for test-purposes
225
226		:- dynamic open_server_socket/2, open_server_client_stream/3.
227
228		stream_socket_init(Port,Socket) :-
229		open_server_socket(Port,S),!,
230		Socket=S.
231		stream_socket_init(Port,Socket) :-
232		Loopback=true, % only accept connections from local machine
233		socket_server_open(Port,Socket,[loopback(Loopback)]),
234		assert(open_server_socket(Port,Socket)),
235		format('Opened RPC-JSON socket (for ndjson, UTF8 streams) on port ~w~n',[Port]).
236
237		stream_socket_rpc_server_stream_init(Port,Stream) :-
238		open_server_client_stream(Port,_C,S),!,
239		Stream=S.
240		stream_socket_rpc_server_stream_init(Port,Stream) :-
241		stream_socket_init(Port,Socket),
242		socket_server_accept(Socket, Client, Stream, [type(text),encoding(utf8)]),
243		assert(open_server_client_stream(Port,Client,Stream)),
244		format('Client connected: ~w~n',[Client]).
245
246		close_stream_server_socket(Port) :-
247		open_server_socket(Port,Socket),!,
248		(retract(open_server_client_stream(Port,_,Stream)) -> close(Stream) ; true),
249		format('Closing socket on port ~w~n',[Port]),
250		retract(open_server_socket(Port,Socket)),
251		retractall(open_server_request(Port,_)),
252		socket_server_close(Socket).
253		close_stream_server_socket(Port) :-
254		add_message(zmq_rpc,'No socket for port open: ',Port).
255
256		:- dynamic open_server_request/2.
257		% predicate to accept a JSON-RPC request on a given port
258		% it opens socket and waits for client if necessary
259		stream_socket_rpc_accept_request(Port,_,_) :-
260		open_server_request(Port,Id),
261		add_warning(zmq_rpc,'Previous JSON-RPC request has not been answered: ',Port:Id),
262		fail.
263		stream_socket_rpc_accept_request(Port,Id,FreeVal) :-
264		stream_socket_rpc_server_stream_init(Port,Stream),
265		read_line(Stream,ResCodes),!,
266		debug_format(19,'Received JSON-RPC request line on port ~w: ~s~n',[Port,ResCodes]),
267		(ResCodes=end_of_file
268		-> close_stream_server_socket(Port), % or only close stream?
269		json_parser_to_json_freetype('@'(null),FreeVal)
270		; (json_parse(ResCodes, ResJson) -> true
271		; add_error(zmq_rpc,'Cannot parse JSON-RPC request:',ResCodes),fail),
272		(ResJson = json(ResObject) -> true
273		; add_error(zmq_rpc,'JSON-RPC request is not an object:',ResJson),fail),
274		% JSON-RPC allows numbers and strings as ids
275		(member(id=JsonId,ResObject), (JsonId = number(Id) ; JsonId = string(Id)) -> true
276		; add_error(zmq_rpc,'JSON-RPC request has no valid id:',ResObject),fail),
277		assert(open_server_request(Port,Id)),
278		json_parser_to_json_freetype(ResJson,FreeVal)
279		).
280
281		stream_socket_rpc_send_answer(Port,Id,AnswerTerm) :-
282		open_server_request(Port,Id),
283		open_server_client_stream(Port,_,Stream),!,
284		write_json_rpc_answer(Id, AnswerTerm, JsonRpcReq),
285		safe_json_write(JsonRpcReq, ReqCodes),
286		debug_format(19,'Sending JSON-RPC answer for request ~w: ~s~n',[Id,ReqCodes]),
287		format(Stream,'~s~n',[ReqCodes]),
288		flush_output(Stream),
289		retract(open_server_request(Port,Id)).
290		stream_socket_rpc_send_answer(Port,Id,_) :-
291		add_error(zmq_rpc,'No open server request on port:id ',Port:Id),
292		fail.
293
294		safe_json_write(Json,Codes) :-
295		(json_write(Json,Codes) -> true ; add_error(zmq_rpc,'Cannot convert JSON:',Json),fail).
296
297		% ----------------------
298		:- public test_server/1.
299		test_server(Port) :-
300		call_cleanup(test_server_loop(Port),
301		(open_server_socket(Port,_) -> close_stream_server_socket(Port) ; true)).
302		test_server_loop(Port) :-
303		stream_socket_rpc_accept_request(Port,Id,FreeVal),
304		json_parser_to_json_freetype('@'(null),JsonNull),
305		(FreeVal=JsonNull -> true % the client closed the connection
306		; write(' B: '), translate:print_bvalue(FreeVal),nl,
307		% TODO: provide predicate to send response
308		stream_socket_rpc_send_answer(Port,Id,freeval('RpcResult','RpcError',string('not implemented'))),
309		% open_server_client_stream(Port,_,Stream),
310		% format(user_output,'sending response >>> {"jsonrpc":"2.0","id":~w,"error":"not_implemented"}~n',[Id]),
311		% format(Stream,'{"jsonrpc":"2.0","id":~w,"error":"not_implemented"}~n',[Id]),
312		% %format(Stream,'{"jsonrpc":"2.0","id":~w,"result":true}~n',[Id]), % --> RpcSuccess(JsonBoolean(TRUE)) for client
313		% flush_output(Stream),
314		% retract(open_server_request(Port,Id)),
315		test_server_loop(Port)
316		).
317