%%% Copyright (C) 2013 - Aleksandr Mescheryakov. All rights reserved. %% @private -module(odi_sock). -behavior(gen_server). -export([start_link/0, close/1, get_parameter/2]). -export([handle_call/3, handle_cast/2, handle_info/2]). -export([init/1, code_change/3, terminate/2]). -export([on_response/3, command/2]). -include("../include/odi.hrl"). -record(state, {mod, %socket module: gen_tcp or ssl(unsupported) sock, %opened socket session_id = -1, %OrientDB session Id token = <<>>, open_mode = not_connected, %connection opened with: connect() | db_open() data = <<>>, %received data from socket queue = queue:new(), %commands queue timeout = 10000, %network timeout global_properties = #{}, callbacks = #{}, version = null }). %% -- client interface -- start_link() -> gen_server:start_link(?MODULE, [], []). close(C) when is_pid(C) -> catch gen_server:cast(C, stop), ok. get_parameter(C, Name) -> gen_server:call(C, {get_parameter, to_binary(Name)}, infinity). % --- support functions --- to_binary(B) when is_binary(B) -> B; to_binary(L) when is_list(L) -> list_to_binary(L). %% -- gen_server implementation -- init([]) -> {ok, #state{}}. terminate(_Reason, State) -> handle_cast(stop, State), ok. handle_call(Command, From, #state{queue=Q, timeout=Timeout} = State) -> Req = {{call, From}, Command}, case command(Command, State#state{queue = queue:in(Req, Q)}) of {noreply, State2} -> {noreply, State2, Timeout}; Error -> {reply, Error, State} end. handle_cast({{Method, From, Ref}, Command} = Req, State) when (Method == cast), is_pid(From), is_reference(Ref) -> #state{queue = Q} = State, command(Command, State#state{queue = queue:in(Req, Q)}); handle_cast(stop, #state{sock = Sock} = State) -> case is_port(Sock) of true -> gen_tcp:close(Sock); _ -> false end, {stop, normal, flush_queue(State, {stop, closed})}. flush_queue(#state{queue = Q} = State, Error) -> case queue:is_empty(Q) of false -> flush_queue(finish(State, Error), Error); true -> State end. handle_info(timeout, State) -> {stop, timeout, flush_queue(State, {error, timeout})}; % Receive messages from socket: % on socket close handle_info({Closed, Sock}, #state{sock = Sock} = State) when Closed == tcp_closed; Closed == ssl_closed -> lager:debug("Socket closed by the remote side", []), {stop, sock_closed, flush_queue(State, {error, sock_closed})}; % on socket error handle_info({Error, Sock, Reason}, #state{sock = Sock} = State) when Error == tcp_error; Error == ssl_error -> lager:debug("Socket error: ~p", [Reason]), Why = {sock_error, Reason}, {stop, Why, flush_queue(State, {error, Why})}; % socket ok handle_info({inet_reply, _, ok}, State) -> {noreply, State}; % socket is not ok handle_info({inet_reply, Reason, Status}, State) -> lager:warning("Socket not OK: ~p", [Reason]), {stop, Status, flush_queue(State, {error, Status})}; % receive data from socket handle_info({tcp, Sock, Data2}, #state{data = Data, sock = Sock} = State) -> loop(State#state{data = <>}). %Handle code change code_change(_OldVsn, State, _Extra) -> {ok, State}. % -- Commands processing -- % ? % command(Command, State = #state{sync_required = true}) % when Command /= sync -> % {noreply, finish(State, {error, sync_required})}; %This is the first operation requested by the client when it needs to work with the server instance without openning a database. %It returns the session id of the client. % Request: (driver-name:string)(driver-version:string)(protocol-version:short)(client-id:string) % (serialization-impl:string)(token-session:boolean)(support-push)(collect-stats) % (user-name:string)(user-password:string) % Response: (session-id:int)(token:bytes) command({connect, Host, Username, Password, Opts}, State) -> case pre_connect(State, Host, Opts) of {error, _} = E -> {noreply, finish(State, E)}; {ok, State2} -> sendRequest(State2, ?O_CONNECT, [string, string, short, string, string, bool, bool, bool, string, string], [?O_DRV_NAME, ?O_DRV_VER, State2#state.version, null, ?O_RECORD_SERIALIZER_BINARY, false, true, false, Username, Password]), {noreply, State2} end; %This is the first operation the client should call. It opens a database on the remote OrientDB Server. %Returns the Session-Id to being reused for all the next calls and the list of configured clusters. % Request: (driver-name:string)(driver-version:string)(protocol-version:short)(client-id:string) % (serialization-impl:string)(token-session:boolean)(support-push:boolean)(collect-stats:boolean) % (database-name:string)(user-name:string)(user-password:string) % Response: (session-id:int)(token:bytes)(num-of-clusters:short)[(cluster-name:string)(cluster-id:short)] % (cluster-config:bytes)(orientdb-release:string) %dbType = document | graph. command({db_open, Host, DBName, Username, Password, Opts}, State) -> case pre_connect(State, Host, Opts) of {error, _} = E -> {noreply, finish(State, E)}; {ok, State2} -> TokenSession = true, sendRequest(State2, ?O_DB_OPEN, [string, string, short, string, string, bool, bool, bool, string, string, string], [?O_DRV_NAME, ?O_DRV_VER, State2#state.version, null, ?O_RECORD_SERIALIZER_BINARY, TokenSession, true, false, DBName, Username, Password]), {noreply, State2} end; %Creates a database in the remote OrientDB server instance % Request: (database-name:string)(database-type:string)(storage-type:string)(backup-path) % Response: empty command({db_create, DatabaseName, DatabaseType, StorageType, BackupPath}, State) -> sendRequest(State, ?O_DB_CREATE, [string, string, string, string], [DatabaseName, DatabaseType, StorageType, BackupPath]), {noreply, State}; %Closes the database and the network connection to the OrientDB Server instance. No return is expected. The socket is also closed. % Request: empty % Response: no response, the socket is just closed at server side command({db_close}, State) -> sendRequest(State, ?O_DB_CLOSE, [], []), handle_cast(stop, State), {stop, normal, State}; %Asks if a database exists in the OrientDB Server instance. It returns true (non-zero) or false (zero). % Request: (database-name:string) <-- before 1.0rc1 this was empty (server-storage-type:string - since 1.5-snapshot) % Response: (result:bool) command({db_exist, DatabaseName, StorageType}, State) -> sendRequest(State, ?O_DB_EXIST, [string, string], [DatabaseName, StorageType]), {noreply, State}; %Reloads database information. Available since 1.0rc4. % Request: empty % Response: (num-of-clusters:short)[(cluster-name:string)(cluster-id:short)] command({db_reload}, State) -> sendRequest(State, ?O_DB_RELOAD, [], []), {noreply, State}; %Removes a database from the OrientDB Server instance. %It returns nothing if the database has been deleted or throws a OStorageException if the database doesn't exists. % Request: (database-name:string)(server-storage-type:string - since 1.5-snapshot) % Response: empty command({db_delete, DatabaseName, ServerStorageType}, State) -> sendRequest(State, ?O_DB_DELETE, [string, string], [DatabaseName, ServerStorageType]), {noreply, State}; %Asks for the size of a database in the OrientDB Server instance. % Request: empty % Response: (size:long) command({db_size}, State) -> sendRequest(State, ?O_DB_SIZE, [], []), {noreply, State}; %Asks for the number of records in a database in the OrientDB Server instance. % Request: empty % Response: (count:long) command({db_countrecords}, State) -> sendRequest(State, ?O_DB_COUNTRECORDS, [], []), {noreply, State}; %Add a new data cluster. % Request: (type:string)(name:string)(location:string)(datasegment-name:string) % Response: (new-cluster:short) command({datacluster_add, Name, ClusterId}, State) -> sendRequest(State, ?O_DATACLUSTER_ADD, [string, short], [Name, ClusterId]), {noreply, State}; %Remove a cluster. % Request: (cluster-number:short) % Response: (delete-on-clientside:byte) command({datacluster_remove, ClusterId}, State) -> sendRequest(State, ?O_DATACLUSTER_REMOVE, [short], [ClusterId]), {noreply, State}; %Create a new record. Returns the position in the cluster of the new record. New records can have version > 0 (since v1.0) in case the RID has been recycled. % Request: (cluster-id:short)(record-content:bytes)(record-type:byte)(mode:byte) % Response: (cluster-id:short)(cluster-position:long)(record-version:int)(count-of-collection-changes)[(uuid-most-sig-bits:long)(uuid-least-sig-bits:long)(updated-file-id:long)(updated-page-index:long)(updated-page-offset:int)]* command({record_create, ClusterId, RecordContent, RecordType, Mode}, State) -> sendRequest(State, ?O_RECORD_CREATE, [short, bytes, byte, byte], [ClusterId, RecordContent, odi_bin:encode_record_type(RecordType), odi_bin:mode_to_byte(Mode)]), {noreply, State}; %Load a record by RecordID, according to a fetch plan % Request: (cluster-id:short)(cluster-position:long)(fetch-plan:string)(ignore-cache:boolean)(load-tombstones:boolean) % Response: [(payload-status:byte)[(record-type:byte)(record-version:int)(record-content:bytes)]*]+ command({record_load, ClusterId, ClusterPosition, FetchPlan, IgnoreCache}, State) -> sendRequest(State, ?O_RECORD_LOAD, [short, long, string, bool, bool], [ClusterId, ClusterPosition, FetchPlan, IgnoreCache, false]), {noreply, State}; %Update a record. Returns the new record's version. % Request: (cluster-id:short)(cluster-position:long)(update-content:boolean)(record-content:bytes)(record-version:int)(record-type:byte)(mode:byte) % Response: (record-version:int)(count-of-collection-changes)[(uuid-most-sig-bits:long)(uuid-least-sig-bits:long)(updated-file-id:long)(updated-page-index:long)(updated-page-offset:int)]* command({record_update, ClusterId, ClusterPosition, UpdateContent, RecordContent, RecordVersion, RecordType, Mode}, State) -> sendRequest(State, ?O_RECORD_UPDATE, [short, long, bool, bytes, integer, byte, byte], [ClusterId, ClusterPosition, UpdateContent, RecordContent, RecordVersion, odi_bin:encode_record_type(RecordType), odi_bin:mode_to_byte(Mode)]), {noreply, State}; %Delete a record by its RecordID. During the optimistic transaction the record will be deleted only if the versions match. %Returns true if has been deleted otherwise false. % Request: (cluster-id:short)(cluster-position:long)(record-version:int)(mode:byte) % Response: (has-been-deleted:boolean) command({record_delete, ClusterId, ClusterPosition, RecordVersion, Mode}, State) -> sendRequest(State, ?O_RECORD_DELETE, [short, long, integer, byte], [ClusterId, ClusterPosition, RecordVersion, odi_bin:mode_to_byte(Mode)]), {noreply, State}; %Executes remote commands. % Request: (mode:byte)(command-payload-length:int)(class-name:string)(command-payload) % Response: % - synchronous commands: [(synch-result-type:byte)[(synch-result-content:?)]]+ % - asynchronous commands: [(asynch-result-type:byte)[(asynch-result-content:?)]*](pre-fetched-record-size)[(pre-fetched-record)]*+ command({command, Query, Mode}, State) -> {Mode, CommandPayload} = case Query of {select, QueryText, Limit, FetchPlan, Params} -> %% (class-name:string)(text:string)(non-text-limit:int)[(fetch-plan:string)](serialized-params:bytes[]) {sync, odi_bin:encode( [string, string, integer, string, bytes], ["q", QueryText, Limit, FetchPlan, encode_params(Params, "params")])}; {live, QueryText, Limit, FetchPlan, Params, _CallBack} -> %% (class-name:string)(text:string)(non-text-limit:int)[(fetch-plan:string)](serialized-params:bytes[]) {live, odi_bin:encode( [string, string, integer, string, bytes], ["com.orientechnologies.orient.core.sql.query.OLiveQuery", QueryText, Limit, FetchPlan, encode_params(Params, "params")])}; {command, Text, SimpleParams, ComplexParams} -> %% (class-name:string)(text:string)(has-simple-parameters:boolean)(simple-paremeters:bytes[])(has-complex-parameters:boolean)(complex-parameters:bytes[]) {sync, odi_bin:encode([string, string, rawbytes, rawbytes], ["c", Text, encode_bool_params(SimpleParams, "parameters"), encode_bool_params(ComplexParams, "compositeKeyParams")])}; {script, Language, Text, SimpleParams, ComplexParams} -> %% (class-name:string)(language:string)(text:string)(has-simple-parameters:boolean)(simple-paremeters:bytes[])(has-complex-parameters:boolean)(complex-parameters:bytes[]) {sync, odi_bin:encode([string, string, string, rawbytes, rawbytes], ["s", Language, Text, encode_bool_params(SimpleParams, "parameters"), encode_bool_params(ComplexParams, "compositeKeyParams")])} end, sendRequest(State, ?O_COMMAND, [byte, bytes], [odi_bin:mode_to_char(Mode), CommandPayload]), {noreply, State}; %generic_query:$s,CommandPayload("com.orientechnologies.orient.core.sql.OCommandSQL",QueryText) %Commits a transaction. This operation flushes all the pending changes to the server side. % Request: (transaction-id:int)(using-tx-log:boolean)(tx-entry)*(0-byte indicating end-of-records) % tx-entry: (1:byte)(operation-type:byte)(cluster-id:short)(cluster-position:long)(record-type:byte)(entry-content) % Response: (created-record-count:int)[(client-specified-cluster-id:short)(client-specified-cluster-position:long)(created-cluster-id:short)(created-cluster-position:long)]*(updated-record-count:int)[(updated-cluster-id:short)(updated-cluster-position:long)(new-record-version:int)]*(count-of-collection-changes:int)[(uuid-most-sig-bits:long)(uuid-least-sig-bits:long)(updated-file-id:long)(updated-page-index:long)(updated-page-offset:int)]* % Operations: [[OperationType, ClusterId, ClusterPosition, RecordType]] command({tx_commit, TxId, UsingTxLog, Operations}, State) -> UnknownStuff = <<0:24>>, sendRequest(State, ?O_TX_COMMIT, [integer, bool, {zero_end, rawbytes}, bytes], [TxId, UsingTxLog, lists:map(fun encode_tx_operation/1, Operations), UnknownStuff]), {noreply, State}; command(Command, State) -> lager:error("Unknown command: ~p", [Command]), {reply, {error, "Unknown command"}, State}. encode_bool_params(null, _Name) -> odi_bin:encode(bool, false); encode_bool_params(Params, Name) -> odi_bin:encode([bool, bytes], [true, encode_params(Params, Name)]). encode_params(null, _Name) -> <<>>; encode_params(Params, Name) -> {ParamsBin, _ParamsOffset} = odi_record_binary:encode_record("", #{Name => {embedded_map, Params}}, 0), ParamsBin. % support functions --- pre_connect(State, Host, Opts) -> Timeout = proplists:get_value(timeout, Opts, 10000), Port = proplists:get_value(port, Opts, 2424), SockOpts = [{active, false}, {packet, raw}, binary, {nodelay, true}], case gen_tcp:connect(Host, Port, SockOpts, Timeout) of {ok, Sock} -> {ok, <>} = gen_tcp:recv(Sock, 2, Timeout), lager:debug("Got version ~p", [Version]), true = Version >= ?O_PROTO_VER_MIN, true = Version =< ?O_PROTO_VER_MAX, ok = inet:setopts(Sock, [{active, true}]), {ok, State#state{mod = gen_tcp, sock = Sock, timeout = Timeout, version = Version}}; {error, _} = E -> E end. sendRequest(#state{mod = Mod, sock = Sock, session_id = SessionId, token = <<>>}, CommandType, Types, Values) -> Data = <>, do_send(Mod, Sock, Data); sendRequest(#state{mod = Mod, sock = Sock, session_id = SessionId, token=Token}, CommandType, Types, Values) -> Data = <>, do_send(Mod, Sock, Data). % port_command() more efficient then gen_tcp:send() do_send(gen_tcp, Sock, Bin) -> lager:debug("Sending: 0x~s", [hex2bin:bin_to_hexstr(Bin)]), try erlang:port_command(Sock, Bin) of true -> ok catch error:_Error -> {error,einval} end; do_send(ssl, _Sock, _Bin) -> {error, ssl_unsupported}. finish(State, Result) -> finish(State, Result, Result). finish(State = #state{queue = Q}, _Notice, Result) -> case queue:get(Q) of % {{cast, From, Ref}, _} -> % From ! {self(), Ref, Result}; % {{incremental, From, Ref}, _} -> % From ! {self(), Ref, Notice}; {{call, From}, _} -> gen_server:reply(From, Result) end, State#state{queue = queue:drop(Q)}. current_command(#state{queue = Q}) -> case queue:len(Q) == 0 of true -> none; false -> {_, Req} = queue:get(Q), Req end. command_tag(State) -> case current_command(State) of none -> none; Req when is_tuple(Req) -> element(1, Req); Req when is_atom(Req) -> Req end. %% -- backend message handling -- %main loop loop(#state{data = <<3:?o_byte, _SessionId:?o_int, Command/binary>>, timeout = Timeout} = State) -> %% A push lager:debug("Received push: 0x~s", [hex2bin:bin_to_hexstr(Command)]), case on_push(Command, State) of {fetch_more, State2} -> {noreply, State2, Timeout}; #state{data = <<>>} = State2 -> {noreply, State2}; State2 -> loop(State2) end; loop(#state{data = Data, timeout = Timeout} = State) -> Cmd = command_tag(State), %erlang:display({recv, Cmd, binary_to_list(Data)}), %erlang:display({recv, Cmd, size(Data)}), case Cmd of none -> {noreply, State#state{data = <<>>}}; _ -> case byte_size(Data) > 0 of true -> lager:debug("Received: 0x~s", [hex2bin:bin_to_hexstr(Data)]), case on_response(Cmd, Data, State) of {fetch_more, State2} -> {noreply, State2, Timeout}; {noreply, #state{data = <<>>} = State2} -> {noreply, State2}; {noreply, State2} -> loop(State2); R = {stop, _Reason2, _State2} -> R end; false -> {noreply, State} end end. on_push(<>, State) -> State2 = handle_push(Command, Bin, State), State2#state{data = Rest}; on_push(_Data, State) -> {fetch_more, State}. handle_push(?O_PUSH_LIVE_QUERY, <>, #state{global_properties = GlobalProperties, callbacks = CallBacks} = State) -> case MessageType of $r -> {{Operation, QueryToken, RecordType, RecordVersion, ClusterId, RecordPosition, RecordBin}, <<>>} = odi_bin:decode([byte, integer, byte, integer, short, long, bytes], Bin), {Class, Data, <<>>} = odi_record_binary:decode_record(RecordType, RecordBin, RecordBin, GlobalProperties), OperationAtom = operation_to_atom(Operation), lager:debug("Got a live record ~p: ~s ~p", [OperationAtom, Class, Data]), #{QueryToken := CallBack} = CallBacks, CallBack(live, {OperationAtom, {{ClusterId, RecordPosition}, document, RecordVersion, Class, Data}}), State; $u -> {QueryToken, <<>>} = odi_bin:decode(integer, Bin), lager:debug("Got a live unsubscription", []), #{QueryToken := CallBack} = CallBacks, CallBack(live_unsubscribe, {}), State#state{callbacks=maps:remove(QueryToken, CallBacks)} end; handle_push(Command, Bin, State) -> lager:warning("Unkwnown push command ~p: ~p", [Command, Bin]), State. %Process empty response message on_empty_response(Bin, State) -> {Status, Message, State2} = response_header(State, Bin), case Status of 1 -> {ErrorInfo,Rest} = odi_bin:decode_error(Message), State3 = finish(State2#state{data = Rest}, {error, ErrorInfo}); 0 -> State3 = finish(State2#state{data = Message}, ok) end, {noreply, State3}. %Process response message without changing State (excl. Data) on_simple_response(Bin, State, Format) -> try {Status, Message, State2} = response_header(State, Bin), case Status of 1 -> {ErrorInfo,Rest} = odi_bin:decode_error(Message), lager:warning("Got an error: ~p", [ErrorInfo]), State3 = finish(State2#state{data = Rest}, {error, ErrorInfo}); 0 -> {Result, Rest} = odi_bin:decode(Format, Message), State3 = finish(State2#state{data = Rest}, Result) end, {noreply, State3} catch throw:not_enough_data -> {fetch_more, State}; Class:Reason -> lager:warning("Error while parsing simple response: ~s", [lager:pr_stacktrace(erlang:get_stacktrace(), {Class, Reason})]), {fetch_more, State} end. response_header(#state{token = <<>>} = State, <>) -> {Status, Message, State}; response_header(#state{token = _Token} = State, <>) -> case NewToken of <<>> -> {Status, Message, State}; _ -> {Status, Message, State#state{token = NewToken}} end. % Response: (session-id:int)(token:bytes) on_response(connect, Bin, #state{sock = Sock} = State) -> <> = Bin, case Status of 1 -> {ErrorInfo,Rest} = odi_bin:decode_error(Message), State2 = State#state{data = Rest}, gen_tcp:close(Sock), State3 = finish(State2, {error, ErrorInfo}); 0 -> <> = Message, State2 = State#state{session_id = SessionId, open_mode = connect, data = Rest}, State3 = finish(State2, ok); _ -> State2 = State#state{data = <<>>}, gen_tcp:close(Sock), State3 = finish(State2, {error, error_server_response}) end, {noreply, State3}; % Response: (session-id:int)(token:bytes)(num-of-clusters:short)[(cluster-name:string)(cluster-id:short)](cluster-config:bytes)(orientdb-release:string) on_response(db_open, Bin, #state{sock = Sock} = State) -> <> = Bin, try case Status of 1 -> {ErrorInfo,Rest} = odi_bin:decode_error(Message), gen_tcp:close(Sock), {noreply, finish(State#state{data = Rest}, {error, ErrorInfo})}; 0 -> {{SessionId, Token, ClusterParams, ClusterConfig, _OrientdbRelease}, Rest} = odi_bin:decode([integer, bytes, {short, [string, short]}, bytes, string], Message), {noreply, finish(State#state{session_id = SessionId, open_mode = db_open, data = Rest, token = Token}, {ClusterParams, ClusterConfig})}; _ -> gen_tcp:close(Sock), {noreply, finish(State#state{data = <<>>}, {error, error_server_response, Bin})} end catch _:_ -> {fetch_more, State} end; % Response: empty on_response(db_create, Bin, State) -> on_empty_response(Bin, State); % Response: none, socket closed. on_response(db_close, _Bin, State) -> {stop, normal, finish(State, db_closed)}; % Response: empty on_response(db_exist, Bin, State) -> on_simple_response(Bin, State, [bool]); % Response: (num-of-clusters:short)[(cluster-name:string)(cluster-id:short)] on_response(db_reload, Bin, State) -> on_simple_response(Bin, State, [{short, [string, short]}]); on_response(db_delete, Bin, State) -> on_empty_response(Bin, State); % Response: (size:long) on_response(db_size, Bin, State) -> on_simple_response(Bin, State, [long]); % Response: (count:long) on_response(db_countrecords, Bin, State) -> on_simple_response(Bin, State, [long]); % Response: (new-cluster:short) on_response(datacluster_add, Bin, State) -> on_simple_response(Bin, State, [short]); % Response: (delete-on-clientside:byte) on_response(datacluster_remove, Bin, State) -> on_simple_response(Bin, State, [bool]); % Response: (cluster-id:short)(cluster-position:long)(record-version:int)(count-of-collection-changes)[(uuid-most-sig-bits:long)(uuid-least-sig-bits:long)(updated-file-id:long)(updated-page-index:long)(updated-page-offset:int)]* on_response(record_create, Bin, State) -> on_simple_response(Bin, State, [short, long, integer, {integer, [longlong, long, long, integer]}]); % Response: [(payload-status:byte)[(record-content:bytes)(record-version:int)(record-type:byte)]*]+ on_response(record_load, Bin, State) -> {Status, Message, State2} = response_header(State, Bin), try case Status of 1 -> {ErrorInfo,Rest} = odi_bin:decode_error(Message), {noreply, finish(State2#state{data = Rest}, {error, ErrorInfo})}; 0 -> {Records, Rest} = decode_records_iterable(Message, State2#state.global_properties, []), State3 = case current_command(State2) of {record_load, 0, 1, "*:-1 index:0", true} -> [{true, document, _Version, _Class, RawSchemas}] = Records, GlobalProperties = odi_typed:index_global_properties(odi_typed:untypify_record(RawSchemas)), lager:debug("Capturing the GlobalProperties: ~p", [GlobalProperties]), State2#state{global_properties=GlobalProperties}; _ -> State2 end, {noreply, finish(State3#state{data = Rest}, Records)} end catch throw:not_enough_data -> {fetch_more, State2}; Class:Reason -> lager:warning("Error while parsing record_load response: ~s", [lager:pr_stacktrace(erlang:get_stacktrace(), {Class, Reason})]), {fetch_more, State2} end; % Response: (record-version:int)(count-of-collection-changes)[(uuid-most-sig-bits:long)(uuid-least-sig-bits:long)(updated-file-id:long)(updated-page-index:long)(updated-page-offset:int)]* on_response(record_update, Bin, State) -> on_simple_response(Bin, State, [integer, {integer, [longlong, long, long, integer]}]); % Response: (has-been-deleted:boolean) on_response(record_delete, Bin, State) -> on_simple_response(Bin, State, [bool]); % Response: % - synchronous commands: [(synch-result-type:byte)[(synch-result-content:?)]]+ on_response(command, Bin, State) -> {Status, Message, State2} = response_header(State, Bin), try case Status of 1 -> {ErrorInfo,Rest} = odi_bin:decode_error(Message), {noreply, finish(State2#state{data = Rest}, {error, ErrorInfo})}; 0 -> {Results, Rest} = decode_command_answer(Message, State2), State3 = case current_command(State2) of {command, {live, _SQL, _Limit, _FetchPlan, _Params, CallBack}, live} -> %% that was a subscription to a live query {[{{-1, -1}, document, 0, "", #{"token" := {integer, Token}}}],[]} = Results, PrevCallBacks = State2#state.callbacks, State2#state{callbacks = PrevCallBacks#{Token => CallBack}}; _ -> State2 end, {noreply, finish(State3#state{data = Rest}, Results)} end catch throw:not_enough_data -> {fetch_more, State2}; Class:Reason -> lager:warning("Error while parsing command response: ~s", [lager:pr_stacktrace(erlang:get_stacktrace(), {Class, Reason})]), {fetch_more, State2} end; % Response: Response: (created-record-count:int)[ % (client-specified-cluster-id:short)(client-specified-cluster-position:long) % (created-cluster-id:short)(created-cluster-position:long) % ]* % (updated-record-count:int)[ % (updated-cluster-id:short)(updated-cluster-position:long) % (new-record-version:int) % ]* % (count-of-collection-changes:int)[ % (uuid-most-sig-bits:long)(uuid-least-sig-bits:long) % (updated-file-id:long)(updated-page-index:long)(updated-page-offset:int) % ]* on_response(tx_commit, Bin, State) -> on_simple_response(Bin, State, [{integer, [[short, long], [short, long]]}, {integer, [[short, long], integer]}, {integer, [longlong, long, long, integer]}]); on_response(_Command, _Bin, State) -> {error, State}. encode_tx_operation({update, Rid, RecordType, Version, UpdateContent, RecordContent}) -> Base = encode_base_tx_operation(1, Rid, RecordType), %% (version:int)(update-content:boolean)(record-content:bytes) (wrong order...) odi_bin:encode([rawbytes, integer, bytes, bool], [Base, Version, RecordContent, UpdateContent]); encode_tx_operation({delete, Rid, RecordType, Version}) -> Base = encode_base_tx_operation(2, Rid, RecordType), %% (version:int) odi_bin:encode([rawbytes, integer], [Base, Version]); encode_tx_operation({create, Rid, RecordType, RecordContent}) -> Base = encode_base_tx_operation(3, Rid, RecordType), %% (record-content:bytes) odi_bin:encode([rawbytes, bytes], [Base, RecordContent]). encode_base_tx_operation(OperationType, {ClusterId, ClusterPosition}, RecordType) -> odi_bin:encode( [byte, byte, short, long, byte], [1, OperationType, ClusterId, ClusterPosition, odi_bin:encode_record_type(RecordType)]). decode_records_iterable(<<>>, _GlobalProperties, Acc) -> {lists:reverse(Acc), <<>>}; decode_records_iterable(<<0:?o_byte, Msg/binary>>, _GlobalProperties, Acc) -> {lists:reverse(Acc), Msg}; decode_records_iterable(<<1:?o_byte, Msg/binary>>, GlobalProperties, Acc) -> {{RecordType, RecordVersion, RecordBin}, NextRecord} = odi_bin:decode([byte, integer, bytes], Msg), {Class, Data, <<>>} = odi_record_binary:decode_record(RecordType, RecordBin, RecordBin, GlobalProperties), decode_records_iterable(NextRecord, GlobalProperties, [{true, odi_bin:decode_record_type(RecordType), RecordVersion, Class, Data} | Acc]); decode_records_iterable(<<2:?o_byte, Msg/binary>>, GlobalProperties, Records) -> {Record, NextRecord} = decode_record(Msg, GlobalProperties), decode_records_iterable(NextRecord, GlobalProperties, [Record | Records]). decode_record(<<0:?o_short, Bin/binary>>, GlobalProperties) -> {{RecordType, ClusterId, RecordPosition, RecordVersion, RecordBin}, Rest} = odi_bin:decode([byte, short, long, integer, bytes], Bin), {Class, Data, <<>>} = odi_record_binary:decode_record(RecordType, RecordBin, RecordBin, GlobalProperties), {{{ClusterId, RecordPosition}, odi_bin:decode_record_type(RecordType), RecordVersion, Class, Data}, Rest}; decode_record(<<-2:?o_short, Rest/binary>>, _GlobalProperties) -> {null, Rest}; decode_record(<<-3:?o_short, Bin/binary>>, _GlobalProperties) -> {{ClusterId, RecordPosition}, Rest} = odi_bin:decode([short, long], Bin), {{{ClusterId, RecordPosition}, null, null, null, null}, Rest}. decode_command_answer(Bin, #state{global_properties = GlobalProperties}) -> {Results, CachedBin} = decode_command_answer_primary(Bin, GlobalProperties), {Cached, Rest} = decode_records_iterable(CachedBin, GlobalProperties, []), {{Results, Cached}, Rest}. decode_command_answer_primary(<<$n:?o_byte, Rest/binary>>, _GlobalProperties) -> {[], Rest}; decode_command_answer_primary(<<$l:?o_byte, Num:?o_int, Rest/binary>>, GlobalProperties) -> decode_record_list(Num, Rest, GlobalProperties, []); decode_command_answer_primary(<<$s:?o_byte, Num:?o_int, Rest/binary>>, GlobalProperties) -> decode_record_list(Num, Rest, GlobalProperties, []); decode_command_answer_primary(<<$i:?o_byte, Rest/binary>>, GlobalProperties) -> decode_records_iterable(Rest, GlobalProperties, []); decode_command_answer_primary(<<$r:?o_byte, Bin/binary>>, GlobalProperties) -> {Record, Rest} = decode_record(Bin, GlobalProperties), {[Record], Rest}; decode_command_answer_primary(<<$w:?o_byte, Bin/binary>>, GlobalProperties) -> {Record, Rest} = decode_record(Bin, GlobalProperties), {[Record], Rest}. decode_record_list(0, Rest, _GlobalProperties, Acc) -> {lists:reverse(Acc), Rest}; decode_record_list(N, Bin, GlobalProperties, Acc) -> {Record, Rest} = decode_record(Bin, GlobalProperties), decode_record_list(N - 1, Rest, GlobalProperties, [Record | Acc]). operation_to_atom(0) -> loaded; operation_to_atom(1) -> updated; operation_to_atom(2) -> deleted; operation_to_atom(3) -> created. -ifdef(TEST). -include_lib("eunit/include/eunit.hrl"). decode_record_load_test() -> %% one record and a linked record Bin = <<1,100,0,0,0,2,0,0,0,29,0,2,86,8,111,117,116,95,0,0,0,14, 22,0,1,0,0,0,1,0,17,0,0,0,0,0,0,0,0, 2,0,0,100,0,17,0,0,0,0,0,0,0,0,0,0,0,1,0,0,0,25,0,2,69,6,111, 117,116,0,0,0,21,13,4,105,110,0,0,0, 23,13,0,20,0,18,0,0>>, {[ {true, document, 2, "V", #{"out_" := {linkbag, [{17, 0}]}}}, {{17, 0}, document, 1, "E", #{"in" := {link, {9, 0}}, "out" := {link, {10, 0}}}} ], <<>>} = decode_records_iterable(Bin, #{}, []). decode_record_iterable_test() -> GlobalProperties = #{ 23 => #{"id" => 23,"name" => "field1","type" => "STRING"}, 24 => #{"id" => 24,"name" => "field2","type" => "LONG"}, 25 => #{"id" => 25,"name" => "field3","type" => "BOOLEAN"}, 26 => #{"id" => 26,"name" => "in","type" => "LINK"}, 27 => #{"id" => 27,"name" => "out","type" => "LINK"} }, Bin = hex2bin:hexstr_to_bin("016400000002000000520008546573742F0000002D3100000033086F75745F0000003416186F75745F54657374456467650000004316000A68656C6C6F540100000001002900000000000000000100000001002A000000000000000002000064002A0000000000000000000000010000001900105465737445646765370000001535000000170032004200020000640029000000000000000000000001000000190010546573744564676535000000153700000017004200320000"), Records = decode_records_iterable(Bin, GlobalProperties, []), Expected = [ {true, document, 2, "Test", #{ "field1" => {string, "hello"}, "field2" => {long, 42}, "out_" => {linkbag, [{41, 0}]}, "out_TestEdge" => {linkbag, [{42, 0}]} }}, {{42, 0}, document, 1, "TestEdge", #{ "in" => {link, {33, 0}}, "out" => {link, {25, 0}} }}, {{41,0}, document, 1, "TestEdge", #{ "in" => {link, {33, 0}}, "out" => {link, {25, 0}} }} ], {Expected, <<>>} = Records. -endif.