zmq_rpc

close_stream_server_socket(Port) :-
    open_server_socket(Port,Socket),!,
    (retract(open_server_client_stream(Port,_,Stream)) -> close(Stream) ; true),
    format('Closing socket on port ~w~n',[Port]),
    socket_server_close(Socket),
    retract(open_server_socket(Port,Socket)).

close_stream_server_socket(Port) :-
    add_message(zmq_rpc,'No socket for port open: ',Port).

extract_error_result(E, string(Res)) :-
    (E=json(ErrorObject) -> true ; ErrorObject=[]),
    (member(code=number(Code), ErrorObject) -> true ; Code = -1),
    (member(message=string(Message), ErrorObject) -> true ; Message = 'N/A'),
    ajoin(['error ', Code, ': ', Message], Res).

extract_json_rpc_result(json(JsonObject), Id, ResAst) :-
    member(jsonrpc=string(Version), JsonObject), !,
    (Version='2.0' -> true ; add_warning(zmq_rpc,'Expected version 2.0 JSON-RPC object: ',Version)),
    (memberchk(id=JsonId, JsonObject)
    -> (JsonId = number(Id) -> true ; JsonId = string(Id) -> true ;
    add_error(zmq_rpc,'Cannot extract valid id from result: ',JsonId),fail
    )
    ; add_error(zmq_rpc,'JSON-RPC result object has no id: ',JsonObject),
    fail
    ),
    (member(error=E, JsonObject)
    -> extract_error_result(E, WrappedRes), ResAst = freeval('RpcResult', 'RpcError', WrappedRes)
    ; extract_success_result(JsonObject, WrappedRes), ResAst = freeval('RpcResult', 'RpcSuccess', WrappedRes)
).

extract_json_rpc_result(Obj, _Id, _ResAst) :-
    add_error(zmq_rpc,'Response object does not conform to JSON-RPC:',Obj),
    fail.

extract_success_result(JsonObject, ResAst) :-
    member(result=Res, JsonObject), !,
    json_parser_to_json_freetype(Res, ResAst).

extract_success_result(Obj, _ResAst) :-
    add_error(zmq_rpc,'Response object has no result attribute:',Obj),
    fail.

foreign(zmq_rpc_init_impl, zmq_rpc_init_impl(+string, [-address])).

foreign(zmq_rpc_destroy_impl, zmq_rpc_destroy_impl(+address)).

foreign(zmq_rpc_send_str_impl, zmq_rpc_send_str_impl(+address, +codes)).

foreign(zmq_rpc_receive_str_impl, zmq_rpc_receive_str_impl(+address, [-codes])).

foreign_resource(zmq_rpc, [zmq_rpc_init_impl, zmq_rpc_destroy_impl, zmq_rpc_send_str_impl, zmq_rpc_receive_str_impl]).

init_zmq_rpc :-
    (loaded -> true
    ; assertz(loaded),
    safe_load_foreign_resource(zmq_rpc, zmq_rpc)
).

next_rpc_request_id(Id) :-
    (last_rpc_request_id(Last) -> Id is Last+1 ; Id=1),
    retractall(last_rpc_request_id(_)),
    assertz(last_rpc_request_id(Id)).

rpc_destroy_socket(sicstus_stream_socket(Stream)) :- !, close(Stream).

rpc_destroy_socket(Socket) :- zmq_rpc_destroy_impl(Socket).

rpc_init(sicstus_stream_socket(_)) :- !. % no init required for these kinds of sockets

rpc_init(_) :- init_zmq_rpc.

rpc_receive_codes_result(sicstus_stream_socket(Stream),Codes) :- !, read_line(Stream,Codes).

rpc_receive_codes_result(Socket, ReqCodes) :- zmq_rpc_receive_str_impl(Socket,ReqCodes).

rpc_send_codes_request(sicstus_stream_socket(Stream),Codes) :- !,
    % send request as a single line; supposes Codes does not contain newline characters
    %format(user_output,'>>> ~s~n',[Codes]),
    format(Stream,'~s~n',[Codes]), flush_output(Stream).

rpc_send_codes_request(Socket, ReqCodes) :- % send request via ZMQ API
    zmq_rpc_send_str_impl(Socket,ReqCodes).

safe_json_write(Json,Codes) :-
    (json_write(Json,Codes) -> true ; add_error(zmq_rpc,'Cannot convert JSON:',Json),fail).

stream_socket_init(Port,Socket) :-
    open_server_socket(Port,S),!,
    Socket=S.

stream_socket_init(Port,Socket) :-
    Loopback=true, % only accept connections from local machine
    socket_server_open(Port,Socket,[loopback(Loopback)]),
    assert(open_server_socket(Port,Socket)),
    format('Opened RPC-JSON socket (for ndjson, UTF8 streams) on port ~w~n',[Port]).

stream_socket_rpc_accept_request(Port,_,_) :-
    open_server_request(Port,Id),
    add_warning(zmq_rpc,'Previous JSON-RPC request has not been answered: ',Port:Id),
    fail.

stream_socket_rpc_accept_request(Port,Id,FreeVal) :-
    stream_socket_rpc_server_stream_init(Port,Stream),
    read_line(Stream,ResCodes),!,
    format(user_output,'Received request line: ~s~n',[ResCodes]),
    (ResCodes=end_of_file
    -> close_stream_server_socket(Port), % or only close stream?
    json_parser_to_json_freetype('@'(null),FreeVal)
    ; json_parse(ResCodes, json(ResJson)),
    write(' Prolog: '),write(ResJson),nl,
    (member(id=number(Id),ResJson) -> true
    ; add_error(zmq_rpc,'JSON-RPC request has no id:',ResJson),fail),
    assert(open_server_request(Port,Id)),
    json_parser_to_json_freetype(json(ResJson),FreeVal)
).

stream_socket_rpc_init(PortTerm,Out) :-
    % TODO: also allow to pass nodename, with empty ('') a connection is made to local machine
    catch(socket_client_open('':PortTerm, Stream, [type(text),encoding(utf8)]),Exc,
    (add_error(zmq_rpc,'Cannot open socket as client:',Exc),fail)),
    gennum(GN),
    SocketNr is -1-GN, % we assume negative numbers not used by ZMQ,
    % makes API easier to use/type for end-users than returning e.g. string(Name)
    (open_socket(SocketNr,_) -> add_internal_error('Socket number clash:',SocketNr) ; true),
    assertz(open_socket(SocketNr,sicstus_stream_socket(Stream))),
    Out = int(SocketNr).

stream_socket_rpc_send_answer(Port,Id,AnswerTerm) :-
    open_server_request(Port,Id),
    open_server_client_stream(Port,_,Stream),!,
    write_json_rpc_answer(Id, AnswerTerm, JsonRpcReq),
    safe_json_write(JsonRpcReq, ReqCodes),
    format(user_output,'Response for request ~w >>> ~s~n',[Id,ReqCodes]),
    format(Stream,'~s~n',[ReqCodes]),
    flush_output(Stream),
    retract(open_server_request(Port,Id)).

stream_socket_rpc_send_answer(Port,Id,_) :-
    add_error(zmq_rpc,'No open server request on port:id ',Port:Id),
    fail.

stream_socket_rpc_server_stream_init(Port,Stream) :-
    open_server_client_stream(Port,_C,S),!,
    Stream=S.

stream_socket_rpc_server_stream_init(Port,Stream) :-
    stream_socket_init(Port,Socket),
    socket_server_accept(Socket, Client, Stream, [type(text),encoding(utf8)]),
    assert(open_server_client_stream(Port,Client,Stream)),
    format('Client connected: ~w~n',[Client]).

test_server(Port) :-
    call_cleanup(test_server_loop(Port),
    (close_stream_server_socket(Port))).

test_server_loop(Port) :-
    stream_socket_rpc_accept_request(Port,Id,FreeVal),
    (FreeVal=end_of_file -> true
    ; write(' B: '), translate:print_bvalue(FreeVal),nl,
    % TODO: provide predicate to send response
    stream_socket_rpc_send_answer(Port,Id,freeval('RpcResult','RpcError',string('not implemented'))),
    % open_server_client_stream(Port,_,Stream),
    % format(user_output,'sending response >>> {"jsonrpc":"2.0","id":~w,"error":"not_implemented"}~n',[Id]),
    % format(Stream,'{"jsonrpc":"2.0","id":~w,"error":"not_implemented"}~n',[Id]),
    % %format(Stream,'{"jsonrpc":"2.0","id":~w,"result":true}~n',[Id]), % --> RpcSuccess(JsonBoolean(TRUE)) for client
    % flush_output(Stream),
    % retract(open_server_request(Port,Id)),
    test_server_loop(Port)
).

write_json_rpc_answer(Id,freeval('RpcResult', 'RpcSuccess', ResultTerm),Json) :- !,
    write_json_rpc_success_answer(Id, ResultTerm, Json).

write_json_rpc_answer(Id,freeval('RpcResult', 'RpcError', string(ErrorMsg)),Json) :- !,
    write_json_rpc_error_answer(Id, ErrorMsg, Json).

write_json_rpc_answer(Id,Other,Json) :-
    add_error(zmq_rpc,'Not a valid RpcResult freevalue:',Other),
    write_json_rpc_error_answer(Id, 'Internal server error: illegal result object', Json).

write_json_rpc_error_answer(Id, String, Json) :-
    (number(Id) -> JsonId = number(Id) ; JsonId = string(Id)),
    Json = json([jsonrpc=string('2.0'), error=string(String), id=JsonId]).

write_json_rpc_request(Method, Params, Id, Json) :-
    json_freetype_to_json_parser(freeval('JsonValue', 'JsonObject', Params), JsonParams),
    (number(Id) -> JsonId = number(Id) ; JsonId = string(Id)),
    Json = json([jsonrpc=string('2.0'), method=string(Method), params=JsonParams, id=JsonId]).

write_json_rpc_success_answer(Id, ResultTerm, Json) :-
    json_freetype_to_json_parser(ResultTerm, JsonResult),
    (number(Id) -> JsonId = number(Id) ; JsonId = string(Id)),
    Json = json([jsonrpc=string('2.0'), result=JsonResult, id=JsonId]).

zmq_rpc_destroy(Socket) :- \+ open_socket(Socket,_),!,
    add_error(zmq_rpc_send,'Socket not open for ZMQ_RPC_DESTROY (use ZMQ_RPC_INIT or SMQ_RPC_INIT first):',Socket),
    fail.

zmq_rpc_destroy(Socket) :-
    open_socket(Socket,SocketTerm),
    rpc_init(SocketTerm),
    rpc_destroy_socket(SocketTerm),
    retractall(open_socket(Socket,_)).

zmq_rpc_destroy_all :-
    (open_socket(Socket,_) -> zmq_rpc_destroy(Socket), zmq_rpc_destroy_all ; true).

zmq_rpc_init(Endpoint, Out) :-
    atom(Endpoint),
    init_zmq_rpc,
    zmq_rpc_init_impl(Endpoint, Res),
    debug_println(9,zmq_rpc_init_impl(Endpoint, Res)),
    integer(Res),
    assertz(open_socket(Res,Res)),
    Out = int(Res).

zmq_rpc_send(Socket, _Name, _Args, _ResAst) :- \+ open_socket(Socket,_),!,
    add_error(zmq_rpc_send,'Socket not open for ZMQ_RPC_SEND (use ZMQ_RPC_INIT first):',Socket),
    fail.

zmq_rpc_send(SocketNr, Name, Args, ResAst) :-
    open_socket(SocketNr,Socket),
    atom(Name),
    !,
    next_rpc_request_id(Id),
    write_json_rpc_request(Name, Args, Id, JsonRpcReq),
    safe_json_write(JsonRpcReq, ReqCodes),
    rpc_init(Socket),
    debug_format(19,'Sending JSON-RPC on socket ~w: ~s~n',[SocketNr,ReqCodes]),
    rpc_send_codes_request(Socket, ReqCodes),
    rpc_receive_codes_result(Socket, ResCodes),
    debug_format(19,'Received JSON-RPC result on socket ~w: ~s~n',[SocketNr,ResCodes]),
    (ResCodes=end_of_file -> add_error(zmq_rpc,'JSON-RPC server not responding:',ResCodes),fail
    ; json_parse(ResCodes, ResJson) -> true
    ; add_error(zmq_rpc,'Cannot parse JSON-RPC result:',ResCodes),fail),
    extract_json_rpc_result(ResJson, ReceivedId, ResAst),
    (ReceivedId=Id -> true
    ; ajoin(['Id ',ReceivedId,' of JSON-RPC result object does not match expected id:'],Msg),
    add_error(zmq_rpc,Msg,Id),
    fail
).

zmq_rpc_send(Socket, Name, ArgsAst, ResAst) :-
    add_internal_error('Illegal call to ZMQ_RPC_SEND:',zmq_rpc_send(Socket, Name, ArgsAst, ResAst)),
    fail.