%% -------------------------------------------------------------------
%%
%% riakc_pb_socket: protocol buffer client
%%
%% Copyright (c) 2007-2013 Basho Technologies, Inc. All Rights Reserved.
%%
%% This file is provided to you under the Apache License,
%% Version 2.0 (the "License"); you may not use this file
%% except in compliance with the License. You may obtain
%% a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing,
%% software distributed under the License is distributed on an
%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
%% KIND, either express or implied. See the License for the
%% specific language governing permissions and limitations
%% under the License.
%%
%% -------------------------------------------------------------------
%% @doc Manages a connection to Riak via the Protocol Buffers
%% transport and executes the commands that can be performed over that
%% connection.
%% @end
-module(riakc_pb_socket).
-include_lib("kernel/include/inet.hrl").
-include_lib("riak_pb/include/riak_pb.hrl").
-include_lib("riak_pb/include/riak_kv_pb.hrl").
-include_lib("riak_pb/include/riak_pb_kv_codec.hrl").
-include_lib("riak_pb/include/riak_search_pb.hrl").
-include_lib("riak_pb/include/riak_yokozuna_pb.hrl").
-include_lib("riak_pb/include/riak_dt_pb.hrl").
-include("riakc.hrl").
-behaviour(gen_server).
-export([start_link/2, start_link/3,
start/2, start/3,
stop/1,
set_options/2, set_options/3,
is_connected/1, is_connected/2,
ping/1, ping/2,
get_client_id/1, get_client_id/2,
set_client_id/2, set_client_id/3,
get_server_info/1, get_server_info/2,
get/3, get/4, get/5,
put/2, put/3, put/4,
delete/3, delete/4, delete/5,
delete_vclock/4, delete_vclock/5, delete_vclock/6,
delete_obj/2, delete_obj/3, delete_obj/4,
list_buckets/1, list_buckets/2, list_buckets/3,
stream_list_buckets/1, stream_list_buckets/2, stream_list_buckets/3,
legacy_list_buckets/2,
list_keys/2, list_keys/3,
stream_list_keys/2, stream_list_keys/3,
get_bucket/2, get_bucket/3, get_bucket/4,
get_bucket_type/2, get_bucket_type/3,
set_bucket/3, set_bucket/4, set_bucket/5,
set_bucket_type/3, set_bucket_type/4,
reset_bucket/2, reset_bucket/3, reset_bucket/4,
mapred/3, mapred/4, mapred/5,
mapred_stream/4, mapred_stream/5, mapred_stream/6,
mapred_bucket/3, mapred_bucket/4, mapred_bucket/5,
mapred_bucket_stream/5, mapred_bucket_stream/6,
search/3, search/4, search/5, search/6,
get_index/4, get_index/5, get_index/6, get_index/7, %% @deprecated
get_index_eq/4, get_index_range/5, get_index_eq/5, get_index_range/6,
cs_bucket_fold/3,
default_timeout/1,
tunnel/4]).
%% Counter API
-export([counter_incr/4, counter_val/3]).
%% with options
-export([counter_incr/5, counter_val/4]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]).
%% Yokozuna admin commands
-export([list_search_indexes/1, list_search_indexes/2,
create_search_index/2, create_search_index/4,
get_search_index/2, get_search_index/3,
delete_search_index/2, delete_search_index/3,
set_search_index/3,
get_search_schema/2, get_search_schema/3,
create_search_schema/3, create_search_schema/4]).
%% Datatypes API
-export([fetch_type/3, fetch_type/4,
update_type/4, update_type/5,
modify_type/5]).
-deprecated({get_index,'_', eventually}).
-type ctx() :: any().
-type rpb_req() :: {tunneled, msg_id(), binary()} | atom() | tuple().
-type rpb_resp() :: atom() | tuple().
-type msg_id() :: non_neg_integer(). %% Request identifier for tunneled message types
-type search_admin_opt() :: {timeout, timeout()} |
{call_timeout, timeout()}.
-type search_admin_opts() :: [search_admin_opt()].
-type index_opt() :: {timeout, timeout()} |
{call_timeout, timeout()} |
{stream, boolean()} |
{continuation, binary()} |
{pagination_sort, boolean()} |
{max_results, non_neg_integer() | all}.
-type index_opts() :: [index_opt()].
-type range_index_opt() :: {return_terms, boolean()} |
{term_regex, binary()}.
-type range_index_opts() :: [index_opt() | range_index_opt()].
-type cs_opt() :: {timeout, timeout()} |
{continuation, binary()} |
{max_results, non_neg_integer() | all} |
{start_key, binary()} |
{start_incl, boolean()} |
{end_key, binary()} |
{end_incl, boolean()}.
-type cs_opts() :: [cs_opt()].
%% Which client operation the default timeout is being requested
%% for. `timeout' is the global default timeout. Any of these defaults
%% can be overridden by setting the application environment variable
%% of the same name on the `riakc' application, for example:
%% `application:set_env(riakc, ping_timeout, 5000).'
-record(request, {ref :: reference(), msg :: rpb_req(), from, ctx :: ctx(), timeout :: timeout(),
tref :: reference() | undefined }).
-type portnum() :: non_neg_integer(). %% The TCP port number of the Riak node's Protocol Buffers interface
-type address() :: string() | atom() | inet:ip_address(). %% The TCP/IP host name or address of the Riak node
-record(state, {address :: address(), % address to connect to
port :: portnum(), % port to connect to
auto_reconnect = false :: boolean(), % if true, automatically reconnects to server
% if false, exits on connection failure/request timeout
queue_if_disconnected = false :: boolean(), % if true, add requests to queue if disconnected
sock :: port() | ssl:sslsocket(), % gen_tcp socket
keepalive = false :: boolean(), % if true, enabled TCP keepalive for the socket
transport = gen_tcp :: 'gen_tcp' | 'ssl',
active :: #request{} | undefined, % active request
queue :: queue() | undefined, % queue of pending requests
connects=0 :: non_neg_integer(), % number of successful connects
failed=[] :: [connection_failure()], % breakdown of failed connects
connect_timeout=infinity :: timeout(), % timeout of TCP connection
credentials :: undefined | {string(), string()}, % username/password
cacertfile, % Path to CA certificate file
certfile, % Path to client certificate file, when using
% certificate authentication
keyfile, % Path to certificate keyfile, when using
% certificate authentication
ssl_opts = [], % Arbitrary SSL options, see the erlang SSL
% documentation.
reconnect_interval=?FIRST_RECONNECT_INTERVAL :: non_neg_integer()}).
%% @private Like `gen_server:call/3', but with the timeout hardcoded
%% to `infinity'.
call_infinity(Pid, Msg) ->
gen_server:call(Pid, Msg, infinity).
%% @doc Create a linked process to talk with the riak server on Address:Port
%% Client id will be assigned by the server.
-spec start_link(address(), portnum()) -> {ok, pid()} | {error, term()}.
start_link(Address, Port) ->
start_link(Address, Port, []).
%% @doc Create a linked process to talk with the riak server on Address:Port with Options.
%% Client id will be assigned by the server.
-spec start_link(address(), portnum(), client_options()) -> {ok, pid()} | {error, term()}.
start_link(Address, Port, Options) when is_list(Options) ->
gen_server:start_link(?MODULE, [Address, Port, Options], []).
%% @doc Create a process to talk with the riak server on Address:Port.
%% Client id will be assigned by the server.
-spec start(address(), portnum()) -> {ok, pid()} | {error, term()}.
start(Address, Port) ->
start(Address, Port, []).
%% @doc Create a process to talk with the riak server on Address:Port with Options.
-spec start(address(), portnum(), client_options()) -> {ok, pid()} | {error, term()}.
start(Address, Port, Options) when is_list(Options) ->
gen_server:start(?MODULE, [Address, Port, Options], []).
%% @doc Disconnect the socket and stop the process.
-spec stop(pid()) -> ok.
stop(Pid) ->
call_infinity(Pid, stop).
%% @doc Change the options for this socket. Allows you to connect with one
%% set of options then run with another (e.g. connect with no options to
%% make sure the server is there, then enable queue_if_disconnected).
%% @equiv set_options(Pid, Options, infinity)
%% @see start_link/3
-spec set_options(pid(), client_options()) -> ok.
set_options(Pid, Options) ->
call_infinity(Pid, {set_options, Options}).
%% @doc Like set_options/2, but with a gen_server timeout.
%% @see start_link/3
%% @deprecated
-spec set_options(pid(), client_options(), timeout()) -> ok.
set_options(Pid, Options, Timeout) ->
gen_server:call(Pid, {set_options, Options}, Timeout).
%% @doc Determines whether the client is connected. Returns true if
%% connected, or false and a list of connection failures and frequencies if
%% disconnected.
%% @equiv is_connected(Pid, infinity)
-spec is_connected(pid()) -> true | {false, [connection_failure()]}.
is_connected(Pid) ->
call_infinity(Pid, is_connected).
%% @doc Determines whether the client is connected, with the specified
%% timeout to the client process. Returns true if connected, or false
%% and a list of connection failures and frequencies if disconnected.
%% @see is_connected/1
%% @deprecated
-spec is_connected(pid(), timeout()) -> true | {false, [connection_failure()]}.
is_connected(Pid, Timeout) ->
gen_server:call(Pid, is_connected, Timeout).
%% @doc Ping the server
%% @equiv ping(Pid, default_timeout(ping_timeout))
-spec ping(pid()) -> pong.
ping(Pid) ->
call_infinity(Pid, {req, rpbpingreq, default_timeout(ping_timeout)}).
%% @doc Ping the server specifying timeout
-spec ping(pid(), timeout()) -> pong.
ping(Pid, Timeout) ->
call_infinity(Pid, {req, rpbpingreq, Timeout}).
%% @doc Get the client id for this connection
%% @equiv get_client_id(Pid, default_timeout(get_client_id_timeout))
-spec get_client_id(pid()) -> {ok, client_id()} | {error, term()}.
get_client_id(Pid) ->
get_client_id(Pid, default_timeout(get_client_id_timeout)).
%% @doc Get the client id for this connection specifying timeout
-spec get_client_id(pid(), timeout()) -> {ok, client_id()} | {error, term()}.
get_client_id(Pid, Timeout) ->
call_infinity(Pid, {req, rpbgetclientidreq, Timeout}).
%% @doc Set the client id for this connection
%% @equiv set_client_id(Pid, ClientId, default_timeout(set_client_id_timeout))
-spec set_client_id(pid(), client_id()) -> {ok, client_id()} | {error, term()}.
set_client_id(Pid, ClientId) ->
set_client_id(Pid, ClientId, default_timeout(set_client_id_timeout)).
%% @doc Set the client id for this connection specifying timeout
-spec set_client_id(pid(), client_id(), timeout()) -> {ok, client_id()} | {error, term()}.
set_client_id(Pid, ClientId, Timeout) ->
call_infinity(Pid,
{req, #rpbsetclientidreq{client_id = ClientId},
Timeout}).
%% @doc Get the server information for this connection
%% @equiv get_server_info(Pid, default_timeout(get_server_info_timeout))
-spec get_server_info(pid()) -> {ok, server_info()} | {error, term()}.
get_server_info(Pid) ->
get_server_info(Pid, default_timeout(get_server_info_timeout)).
%% @doc Get the server information for this connection specifying timeout
-spec get_server_info(pid(), timeout()) -> {ok, server_info()} | {error, term()}.
get_server_info(Pid, Timeout) ->
call_infinity(Pid, {req, rpbgetserverinforeq, Timeout}).
%% @doc Get bucket/key from the server.
%% Will return {error, notfound} if the key is not on the server.
%% @equiv get(Pid, Bucket, Key, [], default_timeout(get_timeout))
-spec get(pid(), bucket(), key()) -> {ok, riakc_obj()} | {error, term()}.
get(Pid, Bucket, Key) ->
get(Pid, Bucket, Key, [], default_timeout(get_timeout)).
%% @doc Get bucket/key from the server specifying timeout.
%% Will return {error, notfound} if the key is not on the server.
%% @equiv get(Pid, Bucket, Key, Options, Timeout)
-spec get(pid(), bucket(), key(), TimeoutOrOptions::timeout() | get_options()) ->
{ok, riakc_obj()} | {error, term()} | unchanged.
get(Pid, Bucket, Key, Timeout) when is_integer(Timeout); Timeout =:= infinity ->
get(Pid, Bucket, Key, [], Timeout);
get(Pid, Bucket, Key, Options) ->
get(Pid, Bucket, Key, Options, default_timeout(get_timeout)).
%% @doc Get bucket/key from the server supplying options and timeout.
%% unchanged will be returned when the
%% {if_modified, Vclock} option is specified and the
%% object is unchanged.
-spec get(pid(), bucket(), key(), get_options(), timeout()) ->
{ok, riakc_obj()} | {error, term()} | unchanged.
get(Pid, Bucket, Key, Options, Timeout) ->
{T, B} = maybe_bucket_type(Bucket),
Req = get_options(Options, #rpbgetreq{type =T, bucket = B, key = Key}),
call_infinity(Pid, {req, Req, Timeout}).
%% @doc Put the metadata/value in the object under bucket/key
%% @equiv put(Pid, Obj, [])
%% @see put/4
-spec put(pid(), riakc_obj()) ->
ok | {ok, riakc_obj()} | {ok, key()} | {error, term()}.
put(Pid, Obj) ->
put(Pid, Obj, []).
%% @doc Put the metadata/value in the object under bucket/key with options or timeout.
%% @equiv put(Pid, Obj, Options, Timeout)
%% @see put/4
-spec put(pid(), riakc_obj(), TimeoutOrOptions::timeout() | put_options()) ->
ok | {ok, riakc_obj()} | riakc_obj() | {ok, key()} | {error, term()}.
put(Pid, Obj, Timeout) when is_integer(Timeout); Timeout =:= infinity ->
put(Pid, Obj, [], Timeout);
put(Pid, Obj, Options) ->
put(Pid, Obj, Options, default_timeout(put_timeout)).
%% @doc Put the metadata/value in the object under bucket/key with
%% options and timeout. Put throws `siblings' if the
%% riakc_obj contains siblings that have not been resolved by
%% calling {@link riakc_obj:select_sibling/2.} or {@link
%% riakc_obj:update_value/2} and {@link
%% riakc_obj:update_metadata/2}. If the object has no key and
%% the Riak node supports it, `{ok, Key::key()}' will be returned
%% when the object is created, or `{ok, Obj::riakc_obj()}' if
%% `return_body' was specified.
%% @throws siblings
%% @end
-spec put(pid(), riakc_obj(), put_options(), timeout()) ->
ok | {ok, riakc_obj()} | riakc_obj() | {ok, key()} | {error, term()}.
put(Pid, Obj, Options, Timeout) ->
Content = riak_pb_kv_codec:encode_content({riakc_obj:get_update_metadata(Obj),
riakc_obj:get_update_value(Obj)}),
Req = put_options(Options,
#rpbputreq{bucket = riakc_obj:only_bucket(Obj),
type = riakc_obj:bucket_type(Obj),
key = riakc_obj:key(Obj),
vclock = riakc_obj:vclock(Obj),
content = Content}),
call_infinity(Pid, {req, Req, Timeout}).
%% @doc Delete the key/value
%% @equiv delete(Pid, Bucket, Key, [])
-spec delete(pid(), bucket(), key()) -> ok | {error, term()}.
delete(Pid, Bucket, Key) ->
delete(Pid, Bucket, Key, []).
%% @doc Delete the key/value specifying timeout or options. Note that the rw quorum is deprecated, use r and w.
%% @equiv delete(Pid, Bucket, Key, Options, Timeout)
-spec delete(pid(), bucket(), key(), TimeoutOrOptions::timeout() | delete_options()) ->
ok | {error, term()}.
delete(Pid, Bucket, Key, Timeout) when is_integer(Timeout); Timeout =:= infinity ->
delete(Pid, Bucket, Key, [], Timeout);
delete(Pid, Bucket, Key, Options) ->
delete(Pid, Bucket, Key, Options, default_timeout(delete_timeout)).
%% @doc Delete the key/value with options and timeout. Note that the rw quorum is deprecated, use r and w.
-spec delete(pid(), bucket(), key(), delete_options(), timeout()) -> ok | {error, term()}.
delete(Pid, Bucket, Key, Options, Timeout) ->
{T, B} = maybe_bucket_type(Bucket),
Req = delete_options(Options, #rpbdelreq{type = T, bucket = B, key = Key}),
call_infinity(Pid, {req, Req, Timeout}).
%% @doc Delete the object at Bucket/Key, giving the vector clock.
%% @equiv delete_vclock(Pid, Bucket, Key, VClock, [])
-spec delete_vclock(pid(), bucket(), key(), riakc_obj:vclock()) -> ok | {error, term()}.
delete_vclock(Pid, Bucket, Key, VClock) ->
delete_vclock(Pid, Bucket, Key, VClock, []).
%% @doc Delete the object at Bucket/Key, specifying timeout or options and giving the vector clock.
%% @equiv delete_vclock(Pid, Bucket, Key, VClock, Options, Timeout)
-spec delete_vclock(pid(), bucket(), key(), riakc_obj:vclock(), TimeoutOrOptions::timeout() | delete_options()) ->
ok | {error, term()}.
delete_vclock(Pid, Bucket, Key, VClock, Timeout) when is_integer(Timeout); Timeout =:= infinity ->
delete_vclock(Pid, Bucket, Key, VClock, [], Timeout);
delete_vclock(Pid, Bucket, Key, VClock, Options) ->
delete_vclock(Pid, Bucket, Key, VClock, Options, default_timeout(delete_timeout)).
%% @doc Delete the key/value with options and timeout and giving the
%% vector clock. This form of delete ensures that subsequent get and
%% put requests will be correctly ordered with the delete.
%% @see delete_obj/4
-spec delete_vclock(pid(), bucket(), key(), riakc_obj:vclock(), delete_options(), timeout()) ->
ok | {error, term()}.
delete_vclock(Pid, Bucket, Key, VClock, Options, Timeout) ->
{T, B} = maybe_bucket_type(Bucket),
Req = delete_options(Options, #rpbdelreq{type = T, bucket = B, key = Key,
vclock=VClock}),
call_infinity(Pid, {req, Req, Timeout}).
%% @doc Delete the riak object.
%% @equiv delete_vclock(Pid, riakc_obj:bucket(Obj), riakc_obj:key(Obj), riakc_obj:vclock(Obj))
%% @see delete_vclock/6
-spec delete_obj(pid(), riakc_obj()) -> ok | {error, term()}.
delete_obj(Pid, Obj) ->
delete_vclock(Pid, riakc_obj:bucket(Obj), riakc_obj:key(Obj),
riakc_obj:vclock(Obj), [], default_timeout(delete_timeout)).
%% @doc Delete the riak object with options.
%% @equiv delete_vclock(Pid, riakc_obj:bucket(Obj), riakc_obj:key(Obj), riakc_obj:vclock(Obj), Options)
%% @see delete_vclock/6
-spec delete_obj(pid(), riakc_obj(), delete_options()) -> ok | {error, term()}.
delete_obj(Pid, Obj, Options) ->
delete_vclock(Pid, riakc_obj:bucket(Obj), riakc_obj:key(Obj),
riakc_obj:vclock(Obj), Options, default_timeout(delete_timeout)).
%% @doc Delete the riak object with options and timeout.
%% @equiv delete_vclock(Pid, riakc_obj:bucket(Obj), riakc_obj:key(Obj), riakc_obj:vclock(Obj), Options, Timeout)
%% @see delete_vclock/6
-spec delete_obj(pid(), riakc_obj(), delete_options(), timeout()) -> ok | {error, term()}.
delete_obj(Pid, Obj, Options, Timeout) ->
delete_vclock(Pid, riakc_obj:bucket(Obj), riakc_obj:key(Obj),
riakc_obj:vclock(Obj), Options, Timeout).
%% @doc List all buckets on the server.
%% This is a potentially expensive operation and should not be used in production.
%% @equiv list_buckets(Pid, default_timeout(list_buckets_timeout))
-spec list_buckets(pid()) -> {ok, [bucket()]} | {error, term()}.
list_buckets(Pid) ->
list_buckets(Pid, <<"default">>, []).
%% @doc List all buckets on the server specifying server-side timeout.
%% This is a potentially expensive operation and should not be used in production.
-spec list_buckets(pid(), timeout()|list()|binary()) -> {ok, [bucket()]} |
{error, term()}.
list_buckets(Pid, Type) when is_binary(Type) ->
list_buckets(Pid, Type, []);
list_buckets(Pid, Timeout) when is_integer(Timeout) ->
list_buckets(Pid, <<"default">>, [{timeout, Timeout}]);
list_buckets(Pid, Options) ->
list_buckets(Pid, <<"default">>, Options).
list_buckets(Pid, Type, Options) when is_binary(Type), is_list(Options) ->
case stream_list_buckets(Pid, Type, Options) of
{ok, ReqId} ->
wait_for_list(ReqId);
Error ->
Error
end.
stream_list_buckets(Pid) ->
stream_list_buckets(Pid, <<"default">>, []).
stream_list_buckets(Pid, Type) when is_binary(Type) ->
stream_list_buckets(Pid, Type, []);
stream_list_buckets(Pid, Timeout) when is_integer(Timeout) ->
stream_list_buckets(Pid, <<"default">>,[{timeout, Timeout}]);
stream_list_buckets(Pid, Options) ->
stream_list_buckets(Pid, <<"default">>, Options).
stream_list_buckets(Pid, Type, Options) ->
ServerTimeout =
case proplists:get_value(timeout, Options, none) of
none -> ?DEFAULT_PB_TIMEOUT;
ST -> ST
end,
ReqId = mk_reqid(),
call_infinity(Pid, {req, #rpblistbucketsreq{timeout=ServerTimeout,
type=Type,
stream=true},
ServerTimeout, {ReqId, self()}}).
legacy_list_buckets(Pid, Options) ->
ServerTimeout =
case proplists:get_value(timeout, Options, none) of
none -> ?DEFAULT_PB_TIMEOUT;
ST -> ST
end,
call_infinity(Pid, {req, #rpblistbucketsreq{timeout=ServerTimeout},
ServerTimeout}).
%% @doc List all keys in a bucket
%% This is a potentially expensive operation and should not be used in production.
%% @equiv list_keys(Pid, Bucket, default_timeout(list_keys_timeout))
-spec list_keys(pid(), bucket()) -> {ok, [key()]} | {error, term()}.
list_keys(Pid, Bucket) ->
list_keys(Pid, Bucket, []).
%% @doc List all keys in a bucket specifying timeout. This is
%% implemented using {@link stream_list_keys/3} and then waiting for
%% the results to complete streaming.
%% This is a potentially expensive operation and should not be used in production.
-spec list_keys(pid(), bucket(), list()|timeout()) -> {ok, [key()]} |
{error, term()}.
list_keys(Pid, Bucket, infinity) ->
list_keys(Pid, Bucket, [{timeout, undefined}]);
list_keys(Pid, Bucket, Timeout) when is_integer(Timeout) ->
list_keys(Pid, Bucket, [{timeout, Timeout}]);
list_keys(Pid, Bucket, Options) ->
case stream_list_keys(Pid, Bucket, Options) of
{ok, ReqId} ->
wait_for_list(ReqId);
Error ->
Error
end.
%% @doc Stream list of keys in the bucket to the calling process. The
%% process receives these messages.
%% ``` {ReqId::req_id(), {keys, [key()]}}
%% {ReqId::req_id(), done}'''
%% This is a potentially expensive operation and should not be used in production.
%% @equiv stream_list_keys(Pid, Bucket, default_timeout(stream_list_keys_timeout))
-spec stream_list_keys(pid(), bucket()) -> {ok, req_id()} | {error, term()}.
stream_list_keys(Pid, Bucket) ->
stream_list_keys(Pid, Bucket, []).
%% @doc Stream list of keys in the bucket to the calling process specifying server side
%% timeout.
%% The process receives these messages.
%% ``` {ReqId::req_id(), {keys, [key()]}}
%% {ReqId::req_id(), done}'''
%% This is a potentially expensive operation and should not be used in production.
%% @equiv stream_list_keys(Pid, Bucket, Timeout, default_timeout(stream_list_keys_call_timeout))
-spec stream_list_keys(pid(), bucket(), integer()|list()) ->
{ok, req_id()} |
{error, term()}.
stream_list_keys(Pid, Bucket, infinity) ->
stream_list_keys(Pid, Bucket, [{timeout, undefined}]);
stream_list_keys(Pid, Bucket, Timeout) when is_integer(Timeout) ->
stream_list_keys(Pid, Bucket, [{timeout, Timeout}]);
stream_list_keys(Pid, Bucket, Options) ->
ServerTimeout =
case proplists:get_value(timeout, Options, none) of
none -> ?DEFAULT_PB_TIMEOUT;
ST -> ST
end,
{T, B} = maybe_bucket_type(Bucket),
ReqMsg = #rpblistkeysreq{type = T, bucket = B, timeout = ServerTimeout},
ReqId = mk_reqid(),
call_infinity(Pid, {req, ReqMsg, ServerTimeout, {ReqId, self()}}).
%% @doc Get bucket properties.
%% @equiv get_bucket(Pid, Bucket, default_timeout(get_bucket_timeout))
-spec get_bucket(pid(), bucket()) -> {ok, bucket_props()} | {error, term()}.
get_bucket(Pid, Bucket) ->
get_bucket(Pid, Bucket, default_timeout(get_bucket_timeout)).
%% @doc Get bucket properties specifying a server side timeout.
%% @equiv get_bucket(Pid, Bucket, Timeout, default_timeout(get_bucket_call_timeout))
-spec get_bucket(pid(), bucket(), timeout()) -> {ok, bucket_props()} | {error, term()}.
get_bucket(Pid, Bucket, Timeout) ->
get_bucket(Pid, Bucket, Timeout, default_timeout(get_bucket_call_timeout)).
%% @doc Get bucket properties specifying a server side and local call timeout.
%% @deprecated because `CallTimeout' is ignored
-spec get_bucket(pid(), bucket(), timeout(), timeout()) -> {ok, bucket_props()} |
{error, term()}.
get_bucket(Pid, Bucket, Timeout, _CallTimeout) ->
{T, B} = maybe_bucket_type(Bucket),
Req = #rpbgetbucketreq{type = T, bucket = B},
call_infinity(Pid, {req, Req, Timeout}).
get_bucket_type(Pid, BucketType) ->
get_bucket_type(Pid, BucketType, default_timeout(get_bucket_timeout)).
get_bucket_type(Pid, BucketType, Timeout) ->
Req = #rpbgetbuckettypereq{type = BucketType},
call_infinity(Pid, {req, Req, Timeout}).
%% @doc Set bucket properties.
%% @equiv set_bucket(Pid, Bucket, BucketProps, default_timeout(set_bucket_timeout))
-spec set_bucket(pid(), bucket(), bucket_props()) -> ok | {error, term()}.
set_bucket(Pid, Bucket, BucketProps) ->
set_bucket(Pid, Bucket, BucketProps, default_timeout(set_bucket_timeout)).
%% @doc Set bucket properties specifying a server side timeout.
%% @equiv set_bucket(Pid, Bucket, BucketProps, Timeout, default_timeout(set_bucket_call_timeout))
-spec set_bucket(pid(), bucket(), bucket_props(), timeout()) -> ok | {error, term()}.
set_bucket(Pid, Bucket, BucketProps, Timeout) ->
set_bucket(Pid, Bucket, BucketProps, Timeout,
default_timeout(set_bucket_call_timeout)).
%% @doc Set bucket properties specifying a server side and local call timeout.
%% @deprecated because `CallTimeout' is ignored
-spec set_bucket(pid(), bucket(), bucket_props(), timeout(), timeout()) -> ok | {error, term()}.
set_bucket(Pid, Bucket, BucketProps, Timeout, _CallTimeout) ->
PbProps = riak_pb_codec:encode_bucket_props(BucketProps),
{T, B} = maybe_bucket_type(Bucket),
Req = #rpbsetbucketreq{type = T, bucket = B, props = PbProps},
call_infinity(Pid, {req, Req, Timeout}).
set_bucket_type(Pid, BucketType, BucketProps) ->
set_bucket_type(Pid, BucketType, BucketProps, default_timeout(set_bucket_timeout)).
set_bucket_type(Pid, BucketType, BucketProps, Timeout) ->
PbProps = riak_pb_codec:encode_bucket_props(BucketProps),
Req = #rpbsetbuckettypereq{type = BucketType, props = PbProps},
call_infinity(Pid, {req, Req, Timeout}).
%% @doc Reset bucket properties back to the defaults.
%% @equiv reset_bucket(Pid, Bucket, default_timeout(reset_bucket_timeout), default_timeout(reset_bucket_call_timeout))
-spec reset_bucket(pid(), bucket) -> ok | {error, term()}.
reset_bucket(Pid, Bucket) ->
reset_bucket(Pid, Bucket, default_timeout(reset_bucket_timeout), default_timeout(reset_bucket_call_timeout)).
%% @doc Reset bucket properties back to the defaults.
%% @equiv reset_bucket(Pid, Bucket, Timeout, default_timeout(reset_bucket_call_timeout))
-spec reset_bucket(pid(), bucket, timeout()) -> ok | {error, term()}.
reset_bucket(Pid, Bucket, Timeout) ->
reset_bucket(Pid, Bucket, Timeout, default_timeout(reset_bucket_call_timeout)).
%% @doc Reset bucket properties back to the defaults.
%% @deprecated because `CallTimeout' is ignored
-spec reset_bucket(pid(), bucket, timeout(), timeout()) -> ok | {error, term()}.
reset_bucket(Pid, Bucket, Timeout, _CallTimeout) ->
{T, B} = maybe_bucket_type(Bucket),
Req = #rpbresetbucketreq{type = T, bucket = B},
call_infinity(Pid, {req, Req, Timeout}).
%% @doc Perform a MapReduce job across the cluster.
%% See the MapReduce documentation for explanation of behavior.
%% @equiv mapred(Inputs, Query, default_timeout(mapred))
-spec mapred(pid(), mapred_inputs(), [mapred_queryterm()]) ->
{ok, mapred_result()} |
{error, {badqterm, mapred_queryterm()}} |
{error, timeout} |
{error, term()}.
mapred(Pid, Inputs, Query) ->
mapred(Pid, Inputs, Query, default_timeout(mapred_timeout)).
%% @doc Perform a MapReduce job across the cluster with a job timeout.
%% See the MapReduce documentation for explanation of behavior.
%% @equiv mapred(Pid, Inputs, Query, Timeout, default_timeout(mapred_call_timeout))
-spec mapred(pid(), mapred_inputs(), [mapred_queryterm()], timeout()) ->
{ok, mapred_result()} |
{error, {badqterm, mapred_queryterm()}} |
{error, timeout} |
{error, term()}.
mapred(Pid, Inputs, Query, Timeout) ->
mapred(Pid, Inputs, Query, Timeout, default_timeout(mapred_call_timeout)).
%% @doc Perform a MapReduce job across the cluster with a job and
%% local call timeout. See the MapReduce documentation for
%% explanation of behavior. This is implemented by using
%% mapred_stream/6 and then waiting for all results.
%% @see mapred_stream/6
-spec mapred(pid(), mapred_inputs(), [mapred_queryterm()], timeout(), timeout()) ->
{ok, mapred_result()} |
{error, {badqterm, mapred_queryterm()}} |
{error, timeout} |
{error, term()}.
mapred(Pid, Inputs, Query, Timeout, CallTimeout) ->
case mapred_stream(Pid, Inputs, Query, self(), Timeout, CallTimeout) of
{ok, ReqId} ->
wait_for_mapred(ReqId, Timeout);
Error ->
Error
end.
%% @doc Perform a streaming MapReduce job across the cluster sending results
%% to ClientPid.
%% See the MapReduce documentation for explanation of behavior.
%% The ClientPid will receive messages in this format:
%% ``` {ReqId::req_id(), {mapred, Phase::non_neg_integer(), mapred_result()}}
%% {ReqId::req_id(), done}'''
%% @equiv mapred_stream(ConnectionPid, Inputs, Query, ClientPid, default_timeout(mapred_stream_timeout))
-spec mapred_stream(ConnectionPid::pid(),Inputs::mapred_inputs(),Query::[mapred_queryterm()], ClientPid::pid()) ->
{ok, req_id()} |
{error, {badqterm, mapred_queryterm()}} |
{error, timeout} |
{error, Err :: term()}.
mapred_stream(Pid, Inputs, Query, ClientPid) ->
mapred_stream(Pid, Inputs, Query, ClientPid, default_timeout(mapred_stream_timeout)).
%% @doc Perform a streaming MapReduce job with a timeout across the cluster.
%% sending results to ClientPid.
%% See the MapReduce documentation for explanation of behavior.
%% The ClientPid will receive messages in this format:
%% ``` {ReqId::req_id(), {mapred, Phase::non_neg_integer(), mapred_result()}}
%% {ReqId::req_id(), done}'''
%% @equiv mapred_stream(ConnectionPid, Inputs, Query, ClientPid, Timeout, default_timeout(mapred_stream_call_timeout))
-spec mapred_stream(ConnectionPid::pid(),Inputs::mapred_inputs(),Query::[mapred_queryterm()], ClientPid::pid(), Timeout::timeout()) ->
{ok, req_id()} |
{error, {badqterm, mapred_queryterm()}} |
{error, timeout} |
{error, Err :: term()}.
mapred_stream(Pid, Inputs, Query, ClientPid, Timeout) ->
mapred_stream(Pid, Inputs, Query, ClientPid, Timeout,
default_timeout(mapred_stream_call_timeout)).
%% @doc Perform a streaming MapReduce job with a map/red timeout across the cluster,
%% a local call timeout and sending results to ClientPid.
%% See the MapReduce documentation for explanation of behavior.
%% The ClientPid will receive messages in this format:
%% ``` {ReqId::req_id(), {mapred, Phase::non_neg_integer(), mapred_result()}}
%% {ReqId::req_id(), done}'''
%% @deprecated because `CallTimeout' is ignored
-spec mapred_stream(ConnectionPid::pid(),Inputs::mapred_inputs(),
Query::[mapred_queryterm()], ClientPid::pid(),
Timeout::timeout(), CallTimeout::timeout()) ->
{ok, req_id()} |
{error, {badqterm, mapred_queryterm()}} |
{error, timeout} |
{error, Err :: term()}.
mapred_stream(Pid, {index,Bucket,Name,Key}, Query, ClientPid, Timeout, CallTimeout) when is_tuple(Name) ->
Index = riakc_obj:index_id_to_bin(Name),
mapred_stream(Pid, {index,Bucket,Index,Key}, Query, ClientPid, Timeout, CallTimeout);
mapred_stream(Pid, {index,Bucket,Name,StartKey,EndKey}, Query, ClientPid, Timeout, CallTimeout) when is_tuple(Name) ->
Index = riakc_obj:index_id_to_bin(Name),
mapred_stream(Pid, {index,Bucket,Index,StartKey,EndKey}, Query, ClientPid, Timeout, CallTimeout);
mapred_stream(Pid, {index,Bucket,Name,Key}, Query, ClientPid, Timeout, CallTimeout) when is_binary(Name) andalso is_integer(Key) ->
BinKey = list_to_binary(integer_to_list(Key)),
mapred_stream(Pid, {index,Bucket,Name,BinKey}, Query, ClientPid, Timeout, CallTimeout);
mapred_stream(Pid, {index,Bucket,Name,StartKey,EndKey}, Query, ClientPid, Timeout, CallTimeout) when is_binary(Name) andalso is_integer(StartKey) ->
BinStartKey = list_to_binary(integer_to_list(StartKey)),
mapred_stream(Pid, {index,Bucket,Name,BinStartKey,EndKey}, Query, ClientPid, Timeout, CallTimeout);
mapred_stream(Pid, {index,Bucket,Name,StartKey,EndKey}, Query, ClientPid, Timeout, CallTimeout) when is_binary(Name) andalso is_integer(EndKey) ->
BinEndKey = list_to_binary(integer_to_list(EndKey)),
mapred_stream(Pid, {index,Bucket,Name,StartKey,BinEndKey}, Query, ClientPid, Timeout, CallTimeout);
mapred_stream(Pid, Inputs, Query, ClientPid, Timeout, _CallTimeout) ->
MapRed = [{'inputs', Inputs},
{'query', Query},
{'timeout', Timeout}],
send_mapred_req(Pid, MapRed, ClientPid).
%% @doc Perform a MapReduce job against a bucket across the cluster.
%% See the MapReduce documentation for explanation of behavior.
%% This uses list_keys under the hood and so is potentially an expensive operation that should not be used in production.
%% @equiv mapred_bucket(Pid, Bucket, Query, default_timeout(mapred_bucket_timeout))
-spec mapred_bucket(Pid::pid(), Bucket::bucket(), Query::[mapred_queryterm()]) ->
{ok, mapred_result()} |
{error, {badqterm, mapred_queryterm()}} |
{error, timeout} |
{error, Err :: term()}.
mapred_bucket(Pid, Bucket, Query) ->
mapred_bucket(Pid, Bucket, Query, default_timeout(mapred_bucket_timeout)).
%% @doc Perform a MapReduce job against a bucket with a timeout
%% across the cluster.
%% See the MapReduce documentation for explanation of behavior.
%% This uses list_keys under the hood and so is potentially an expensive operation that should not be used in production.
%% @equiv mapred_bucket(Pid, Bucket, Query, Timeout, default_timeout(mapred_bucket_call_timeout))
-spec mapred_bucket(Pid::pid(), Bucket::bucket(), Query::[mapred_queryterm()], Timeout::timeout()) ->
{ok, mapred_result()} |
{error, {badqterm, mapred_queryterm()}} |
{error, timeout} |
{error, Err :: term()}.
mapred_bucket(Pid, Bucket, Query, Timeout) ->
mapred_bucket(Pid, Bucket, Query, Timeout, default_timeout(mapred_bucket_call_timeout)).
%% @doc Perform a MapReduce job against a bucket with a timeout
%% across the cluster and local call timeout.
%% See the MapReduce documentation for explanation of behavior.
%% This uses list_keys under the hood and so is potentially an expensive operation that should not be used in production.
-spec mapred_bucket(Pid::pid(), Bucket::bucket(), Query::[mapred_queryterm()],
Timeout::timeout(), CallTimeout::timeout()) ->
{ok, mapred_result()} |
{error, {badqterm, mapred_queryterm()}} |
{error, timeout} |
{error, Err :: term()}.
mapred_bucket(Pid, Bucket, Query, Timeout, CallTimeout) ->
case mapred_bucket_stream(Pid, Bucket, Query, self(), Timeout, CallTimeout) of
{ok, ReqId} ->
wait_for_mapred(ReqId, Timeout);
Error ->
Error
end.
%% @doc Perform a streaming MapReduce job against a bucket with a timeout
%% across the cluster.
%% See the MapReduce documentation for explanation of behavior.
%% This uses list_keys under the hood and so is potentially an expensive operation that should not be used in production.
%% The ClientPid will receive messages in this format:
%% ``` {ReqId::req_id(), {mapred, Phase::non_neg_integer(), mapred_result()}}
%% {ReqId::req_id(), done}'''
%% @equiv mapred_bucket_stream(Pid, Bucket, Query, ClientPid, Timeout, default_timeout(mapred_bucket_stream_call_timeout))
-spec mapred_bucket_stream(ConnectionPid::pid(), bucket(), [mapred_queryterm()], ClientPid::pid(), timeout()) ->
{ok, req_id()} |
{error, term()}.
mapred_bucket_stream(Pid, Bucket, Query, ClientPid, Timeout) ->
mapred_bucket_stream(Pid, Bucket, Query, ClientPid, Timeout,
default_timeout(mapred_bucket_stream_call_timeout)).
%% @doc Perform a streaming MapReduce job against a bucket with a server timeout
%% across the cluster and a call timeout.
%% See the MapReduce documentation for explanation of behavior.
%% This uses list_keys under the hood and so is potentially an expensive operation that should not be used in production.
%% The ClientPid will receive messages in this format:
%% ``` {ReqId::req_id(), {mapred, Phase::non_neg_integer(), mapred_result()}}
%% {ReqId::req_id(), done}'''
%% @deprecated because `CallTimeout' is ignored
-spec mapred_bucket_stream(ConnectionPid::pid(), bucket(), [mapred_queryterm()], ClientPid::pid(), timeout(), timeout()) ->
{ok, req_id()} | {error, term()}.
mapred_bucket_stream(Pid, Bucket, Query, ClientPid, Timeout, _CallTimeout) ->
MapRed = [{'inputs', Bucket},
{'query', Query},
{'timeout', Timeout}],
send_mapred_req(Pid, MapRed, ClientPid).
%% @doc Execute a search query. This command will return an error
%% unless executed against a Riak Search cluster.
-spec search(pid(), binary(), binary()) ->
{ok, search_result()} | {error, term()}.
search(Pid, Index, SearchQuery) ->
search(Pid, Index, SearchQuery, []).
%% @doc Execute a search query. This command will return an error
%% unless executed against a Riak Search cluster.
-spec search(pid(), binary(), binary(), search_options()) ->
{ok, search_result()} | {error, term()}.
search(Pid, Index, SearchQuery, Options) ->
Timeout = default_timeout(search_timeout),
search(Pid, Index, SearchQuery, Options, Timeout).
%% @doc Execute a search query. This command will return an error
%% unless executed against a Riak Search cluster.
-spec search(pid(), binary(), binary(), search_options(), timeout()) ->
{ok, search_result()} | {error, term()}.
search(Pid, Index, SearchQuery, Options, Timeout) ->
CallTimeout = default_timeout(search_call_timeout),
search(Pid, Index, SearchQuery, Options, Timeout, CallTimeout).
%% @doc Execute a search query. This command will return an error
%% unless executed against a Riak Search cluster.
%% @deprecated because `CallTimeout' is ignored
-spec search(pid(), binary(), binary(), search_options(), timeout(), timeout()) ->
{ok, search_result()} | {error, term()}.
search(Pid, Index, SearchQuery, Options, Timeout, _CallTimeout) ->
Req = search_options(Options, #rpbsearchqueryreq{q = SearchQuery, index = Index}),
call_infinity(Pid, {req, Req, Timeout}).
-spec get_search_schema(pid(), binary(), search_admin_opts()) ->
{ok, search_schema()} | {error, term()}.
get_search_schema(Pid, SchemaName, Opts) ->
Timeout = proplists:get_value(timeout, Opts, default_timeout(search_timeout)),
Req = #rpbyokozunaschemagetreq{ name = SchemaName },
call_infinity(Pid, {req, Req, Timeout}).
-spec get_search_schema(pid(), binary()) ->
{ok, search_schema()} | {error, term()}.
get_search_schema(Pid, SchemaName) ->
get_search_schema(Pid, SchemaName, []).
-spec get_search_index(pid(), binary(), search_admin_opts()) ->
{ok, search_index()} | {error, term()}.
get_search_index(Pid, Index, Opts) ->
Timeout = proplists:get_value(timeout, Opts, default_timeout(search_timeout)),
Req = #rpbyokozunaindexgetreq{ name = Index },
Results = call_infinity(Pid, {req, Req, Timeout}),
case Results of
{ok, [Result]} ->
{ok, Result};
{ok, []} ->
{error, notfound};
X -> X
end.
-spec get_search_index(pid(), binary()) ->
{ok, search_index()} | {error, term()}.
get_search_index(Pid, Index) ->
get_search_index(Pid, Index, []).
-spec list_search_indexes(pid(), search_admin_opts()) ->
{ok, [search_index()]} | {error, term()}.
list_search_indexes(Pid, Opts) ->
Timeout = proplists:get_value(timeout, Opts, default_timeout(search_timeout)),
call_infinity(Pid, {req, #rpbyokozunaindexgetreq{}, Timeout}).
-spec list_search_indexes(pid()) ->
{ok, [search_index()]} | {error, term()}.
list_search_indexes(Pid) ->
list_search_indexes(Pid, []).
%% @doc Create a schema, which is a required component of an index.
-spec create_search_schema(pid(), binary(), binary()) ->
ok | {error, term()}.
create_search_schema(Pid, SchemaName, Content) ->
create_search_schema(Pid, SchemaName, Content, []).
%% @doc Create a schema, which is a required component of an index.
-spec create_search_schema(pid(), binary(), binary(), search_admin_opts()) ->
ok | {error, term()}.
create_search_schema(Pid, SchemaName, Content, Opts) ->
Timeout = proplists:get_value(timeout, Opts, default_timeout(search_timeout)),
Req = #rpbyokozunaschemaputreq{
schema = #rpbyokozunaschema{name = SchemaName, content = Content}
},
call_infinity(Pid, {req, Req, Timeout}).
%% @doc Create a search index.
-spec create_search_index(pid(), binary()) ->
ok | {error, term()}.
create_search_index(Pid, Index) ->
create_search_index(Pid, Index, <<>>, []).
%% @doc Create a search index.
-spec create_search_index(pid(), binary(), binary(), search_admin_opts()) ->
ok | {error, term()}.
create_search_index(Pid, Index, SchemaName, Opts) ->
Timeout = proplists:get_value(timeout, Opts, default_timeout(search_timeout)),
NVal = proplists:get_value(n_val, Opts),
Req = #rpbyokozunaindexputreq{
index = #rpbyokozunaindex{name = Index,
schema = SchemaName,
n_val = NVal}
},
call_infinity(Pid, {req, Req, Timeout}).
%% @doc Delete a search index.
-spec delete_search_index(pid(), binary()) ->
ok | {error, term()}.
delete_search_index(Pid, Index) ->
delete_search_index(Pid, Index, []).
%% @doc Delete a search index.
-spec delete_search_index(pid(), binary(), search_admin_opts()) ->
ok | {error, term()}.
delete_search_index(Pid, Index, Opts) ->
Timeout = proplists:get_value(timeout, Opts, default_timeout(search_timeout)),
Req = #rpbyokozunaindexdeletereq{name = Index},
call_infinity(Pid, {req, Req, Timeout}).
-spec set_search_index(pid(), bucket(), binary()) ->
ok | {error, term()}.
set_search_index(Pid, Bucket, Index) ->
set_bucket(Pid, Bucket, [{search_index, Index}]).
%% Deprecated, argument explosion functions for indexes
%% @doc Execute a secondary index equality query.
%%
%% @deprecated use {@link get_index_eq/4}
%% @see get_index_eq/4
-spec get_index(pid(), bucket(), binary() | secondary_index_id(), key() | integer()) ->
{ok, index_results()} | {error, term()}.
get_index(Pid, Bucket, Index, Key) ->
get_index_eq(Pid, Bucket, Index, Key).
%% @doc Execute a secondary index equality query with specified
%% timeouts.
%%
%% @deprecated use {@link get_index_eq/5}
%% @see get_index_eq/5
-spec get_index(pid(), bucket(), binary() | secondary_index_id(), key() | integer(), timeout(), timeout()) ->
{ok, index_results()} | {error, term()}.
get_index(Pid, Bucket, Index, Key, Timeout, _CallTimeout) ->
get_index_eq(Pid, Bucket, Index, Key, [{timeout, Timeout}]).
%% @doc Execute a secondary index range query.
%%
%% @deprecated use {@link get_index_range/5}
%% @see get_index_range/5
-spec get_index(pid(), bucket(), binary() | secondary_index_id(), key() | integer(), key() | integer()) ->
{ok, index_results()} | {error, term()}.
get_index(Pid, Bucket, Index, StartKey, EndKey) ->
get_index_range(Pid, Bucket, Index, StartKey, EndKey).
%% @doc Execute a secondary index range query with specified
%% timeouts.
%%
%% @deprecated use {@link get_index_range/6}
%% @see get_index_range/6
-spec get_index(pid(), bucket(), binary() | secondary_index_id(), key() | integer() | list(),
key() | integer() | list(), timeout(), timeout()) ->
{ok, index_results()} | {error, term()}.
get_index(Pid, Bucket, Index, StartKey, EndKey, Timeout, _CallTimeout) ->
get_index_range(Pid, Bucket, Index, StartKey, EndKey, [{timeout, Timeout}]).
%% @doc Execute a secondary index equality query.
%% equivalent to all defaults for the options.
%% @see get_index_eq/5. for options and their effect
-spec get_index_eq(pid(), bucket(), binary() | secondary_index_id(), key() | integer()) ->
{ok, index_results()} | {error, term()}.
get_index_eq(Pid, Bucket, Index, Key) ->
get_index_eq(Pid, Bucket, Index, Key, []).
%% @doc Execute a secondary index equality query with specified options
%%
%% - timeout:
- milliseconds to wait for a response from riak
%% - stream:
- true | false. Stream results to calling process
%% - continuation:
- The opaque, binary continuation returned from a previous query.
%% Requests the next results.
%% - max_results:
- Positive integer, maximum number of results to return.
%% Expect a
continuation in the response if this option is used.
%%
%% @end
-spec get_index_eq(pid(), bucket(), binary() | secondary_index_id(), key() | integer(), index_opts()) ->
{ok, index_results()} | {error, term()}.
get_index_eq(Pid, Bucket, {binary_index, Name}, Key, Opts) when is_binary(Key) ->
Index = list_to_binary(lists:append([Name, "_bin"])),
get_index_eq(Pid, Bucket, Index, Key, Opts);
get_index_eq(Pid, Bucket, {integer_index, Name}, Key, Opts) when is_integer(Key) ->
Index = list_to_binary(lists:append([Name, "_int"])),
BinKey = list_to_binary(integer_to_list(Key)),
get_index_eq(Pid, Bucket, Index, BinKey, Opts);
get_index_eq(Pid, Bucket, Index, Key, Opts) ->
Timeout = proplists:get_value(timeout, Opts),
MaxResults = proplists:get_value(max_results, Opts),
PgSort = proplists:get_value(pagination_sort, Opts),
Stream = proplists:get_value(stream, Opts, false),
Continuation = proplists:get_value(continuation, Opts),
{T, B} = maybe_bucket_type(Bucket),
Req = #rpbindexreq{type=T, bucket=B, index=Index, qtype=eq,
key=encode_2i(Key),
max_results=MaxResults,
pagination_sort=PgSort,
stream=Stream,
continuation=Continuation,
timeout=Timeout},
Call = case Stream of
true ->
ReqId = mk_reqid(),
{req, Req, Timeout, {ReqId, self()}};
false ->
{req, Req, Timeout}
end,
call_infinity(Pid, Call).
%% @doc Execute a secondary index range query.
-spec get_index_range(pid(), bucket(), binary() | secondary_index_id(), key() | integer(), key() | integer()) ->
{ok, index_results()} | {error, term()}.
get_index_range(Pid, Bucket, Index, StartKey, EndKey) ->
get_index_range(Pid, Bucket, Index, StartKey, EndKey, []).
%% @doc Execute a secondary index range query with specified options.
%% As well as the options documented for `get_index_eq/5', there is a further options
%% `{return_terms, boolean{}'. When `true' the indexed values will be returned
%% as well as the primary key. The formt of the returned values is
%% `{results, [{value, primary_key}]}'
%% @end
%% @see get_index_eq/5. for effect of options.
-spec get_index_range(pid(), bucket(), binary() | secondary_index_id(), key() | integer() | list(),
key() | integer() | list(), range_index_opts()) ->
{ok, index_results()} | {error, term()}.
get_index_range(Pid, Bucket, {binary_index, Name}, StartKey, EndKey, Opts) when is_binary(StartKey) andalso is_binary(EndKey) ->
Index = list_to_binary(lists:append([Name, "_bin"])),
get_index_range(Pid, Bucket, Index, StartKey, EndKey, Opts);
get_index_range(Pid, Bucket, {integer_index, Name}, StartKey, EndKey, Opts) when is_integer(StartKey) andalso is_integer(EndKey) ->
Index = list_to_binary(lists:append([Name, "_int"])),
BinStartKey = list_to_binary(integer_to_list(StartKey)),
BinEndKey = list_to_binary(integer_to_list(EndKey)),
get_index_range(Pid, Bucket, Index, BinStartKey, BinEndKey, Opts);
get_index_range(Pid, Bucket, Index, StartKey, EndKey, Opts) ->
Timeout = proplists:get_value(timeout, Opts),
ReturnTerms = proplists:get_value(return_terms, Opts),
TermRegex = proplists:get_value(term_regex, Opts),
MaxResults = proplists:get_value(max_results, Opts),
PgSort = proplists:get_value(pagination_sort, Opts),
Stream = proplists:get_value(stream, Opts, false),
Continuation = proplists:get_value(continuation, Opts),
{T, B} = maybe_bucket_type(Bucket),
Req = #rpbindexreq{type=T, bucket=B, index=Index, qtype=range,
range_min=encode_2i(StartKey),
range_max=encode_2i(EndKey),
return_terms=ReturnTerms,
term_regex=TermRegex,
max_results=MaxResults,
pagination_sort = PgSort,
stream=Stream,
continuation=Continuation,
timeout=Timeout},
Call = case Stream of
true ->
ReqId = mk_reqid(),
{req, Req, Timeout, {ReqId, self()}};
false ->
{req, Req, Timeout}
end,
call_infinity(Pid, Call).
encode_2i(Value) when is_integer(Value) ->
list_to_binary(integer_to_list(Value));
encode_2i(Value) when is_list(Value) ->
list_to_binary(Value);
encode_2i(Value) when is_binary(Value) ->
Value.
%% @doc secret function, do not use, or I come to your house and keeel you.
-spec cs_bucket_fold(pid(), bucket(), cs_opts()) -> {ok, reference()} | {error, term()}.
cs_bucket_fold(Pid, Bucket, Opts) when is_pid(Pid), (is_binary(Bucket) orelse
is_tuple(Bucket)), is_list(Opts) ->
Timeout = proplists:get_value(timeout, Opts),
StartKey = proplists:get_value(start_key, Opts, <<>>),
EndKey = proplists:get_value(end_key, Opts),
MaxResults = proplists:get_value(max_results, Opts),
StartIncl = proplists:get_value(start_incl, Opts, true),
EndIncl = proplists:get_value(end_incl, Opts, false),
Continuation = proplists:get_value(continuation, Opts),
{T, B} = maybe_bucket_type(Bucket),
Req = #rpbcsbucketreq{type=T, bucket=B,
start_key=StartKey,
end_key=EndKey,
start_incl=StartIncl,
end_incl=EndIncl,
max_results=MaxResults,
continuation=Continuation,
timeout=Timeout},
ReqId = mk_reqid(),
Call = {req, Req, Timeout, {ReqId, self()}},
call_infinity(Pid, Call).
%% @doc Return the default timeout for an operation if none is provided.
%% Falls back to the default timeout.
-spec default_timeout(timeout_name()) -> timeout().
default_timeout(OpTimeout) ->
case application:get_env(riakc, OpTimeout) of
{ok, EnvTimeout} ->
EnvTimeout;
undefined ->
case application:get_env(riakc, timeout) of
{ok, Timeout} ->
Timeout;
undefined ->
?DEFAULT_PB_TIMEOUT
end
end.
%% @doc Send a pre-encoded msg over the protocol buffer connection
%% Returns {ok, Response} or {error, Reason}
-spec tunnel(pid(), msg_id(), iolist(), timeout()) -> {ok, binary()} | {error, term()}.
tunnel(Pid, MsgId, Pkt, Timeout) ->
Req = {tunneled, MsgId, Pkt},
call_infinity(Pid, {req, Req, Timeout}).
%% @doc increment the counter at `bucket', `key' by `amount'
-spec counter_incr(pid(), bucket(), key(), integer()) -> ok.
counter_incr(Pid, Bucket, Key, Amount) ->
counter_incr(Pid, Bucket, Key, Amount, []).
%% @doc increment the counter at `Bucket', `Key' by `Amount'.
%% use the provided `write_quorum()' `Options' for the operation.
%% A counter increment is a lot like a riak `put' so the semantics
%% are the same for the given options.
-spec counter_incr(pid(), bucket(), key(), integer(), [write_quorum()]) ->
ok | {error, term()}.
counter_incr(Pid, Bucket, Key, Amount, Options) ->
{_, B} = maybe_bucket_type(Bucket),
Req = counter_incr_options(Options, #rpbcounterupdatereq{bucket=B, key=Key, amount=Amount}),
call_infinity(Pid, {req, Req, default_timeout(put_timeout)}).
%% @doc get the current value of the counter at `Bucket', `Key'.
-spec counter_val(pid(), bucket(), key()) ->
{ok, integer()} | {error, notfound}.
counter_val(Pid, Bucket, Key) ->
counter_val(Pid, Bucket, Key, []).
%% @doc get the current value of the counter at `Bucket', `Key' using
%% the `read_qurom()' `Options' provided.
-spec counter_val(pid(), bucket(), key(), [read_quorum()]) ->
{ok, integer()} | {error, term()}.
counter_val(Pid, Bucket, Key, Options) ->
{_, B} = maybe_bucket_type(Bucket),
Req = counter_val_options(Options, #rpbcountergetreq{bucket=B, key=Key}),
call_infinity(Pid, {req, Req, default_timeout(get_timeout)}).
%% @doc Fetches the representation of a convergent datatype from Riak.
-spec fetch_type(pid(), {BucketType::binary(), Bucket::binary()}, Key::binary()) ->
{ok, riakc_datatype:datatype()} | {error, term()}.
fetch_type(Pid, BucketAndType, Key) ->
fetch_type(Pid, BucketAndType, Key, []).
%% @doc Fetches the representation of a convergent datatype from Riak,
%% using the given request options.
-spec fetch_type(pid(), {BucketType::binary(), Bucket::binary()}, Key::binary(), [proplists:property()]) ->
{ok, riakc_datatype:datatype()} | {error, term()}.
fetch_type(Pid, BucketAndType, Key, Options) ->
Req = riak_pb_dt_codec:encode_fetch_request(BucketAndType, Key, Options),
call_infinity(Pid, {req, Req, default_timeout(get_timeout)}).
%% @doc Updates the convergent datatype in Riak with local
%% modifications stored in the container type.
-spec update_type(pid(), {BucketType::binary(), Bucket::binary()}, Key::binary(), Update::riakc_datatype:update(term())) ->
ok | {ok, Key::binary()} | {ok, riakc_datatype:datatype()} |
{ok, Key::binary(), riakc_datatype:datatype()} | {error, term()}.
update_type(Pid, BucketAndType, Key, Update) ->
update_type(Pid, BucketAndType, Key, Update, []).
%% @doc Updates the convergent datatype in Riak with local
%% modifications stored in the container type, using the given request
%% options.
-spec update_type(pid(), {BucketType::binary(), Bucket::binary()}, Key::binary(),
Update::riakc_datatype:update(term()), [proplists:property()]) ->
ok | {ok, Key::binary()} | {ok, riakc_datatype:datatype()} |
{ok, Key::binary(), riakc_datatype:datatype()} | {error, term()}.
update_type(_Pid, _BucketAndType, _Key, undefined, _Options) ->
{error, unmodified};
update_type(Pid, BucketAndType, Key, {Type, Op, Context}, Options) ->
Req = riak_pb_dt_codec:encode_update_request(BucketAndType, Key, {Type, Op, Context}, Options),
call_infinity(Pid, {req, Req, default_timeout(put_timeout)}).
%% @doc Fetches, applies the given function to the value, and then
%% updates the datatype in Riak. If an existing value is not found,
%% but you want the updates to apply anyway, use the 'create' option.
-spec modify_type(pid(), fun((riakc_datatype:datatype()) -> riakc_datatype:datatype()),
{BucketType::binary(), Bucket::binary()}, Key::binary(), [proplists:property()]) ->
ok | {ok, riakc_datatype:datatype()} | {error, term()}.
modify_type(Pid, Fun, BucketAndType, Key, Options) ->
Create = proplists:get_value(create, Options, true),
case fetch_type(Pid, BucketAndType, Key, Options) of
{ok, Data} ->
NewData = Fun(Data),
Mod = riakc_datatype:module_for_term(NewData),
update_type(Pid, BucketAndType, Key, Mod:to_op(NewData), Options);
{error, {notfound, Type}} when Create ->
%% Not found, but ok to create it
Mod = riakc_datatype:module_for_type(Type),
NewData = Fun(Mod:new()),
update_type(Pid, BucketAndType, Key, Mod:to_op(NewData), Options);
{error, Reason} ->
{error, Reason}
end.
%% ====================================================================
%% gen_server callbacks
%% ====================================================================
%% @private
init([Address, Port, Options]) ->
%% Schedule a reconnect as the first action. If the server is up then
%% the handle_info(reconnect) will run before any requests can be sent.
State = parse_options(Options, #state{address = Address,
port = Port,
queue = queue:new()}),
case connect(State) of
{error, Reason} when State#state.auto_reconnect /= true ->
{stop, {tcp, Reason}};
{error, _Reason} ->
{ok, State};
Ok ->
Ok
end.
%% @private
handle_call({req, Msg, Timeout}, From, State) when State#state.sock =:= undefined ->
case State#state.queue_if_disconnected of
true ->
{noreply, queue_request(new_request(Msg, From, Timeout), State)};
false ->
{reply, {error, disconnected}, State}
end;
handle_call({req, Msg, Timeout, Ctx}, From, State) when State#state.sock =:= undefined ->
case State#state.queue_if_disconnected of
true ->
{noreply, queue_request(new_request(Msg, From, Timeout, Ctx), State)};
false ->
{reply, {error, disconnected}, State}
end;
handle_call({req, Msg, Timeout}, From, State) when State#state.active =/= undefined ->
{noreply, queue_request(new_request(Msg, From, Timeout), State)};
handle_call({req, Msg, Timeout, Ctx}, From, State) when State#state.active =/= undefined ->
{noreply, queue_request(new_request(Msg, From, Timeout, Ctx), State)};
handle_call({req, Msg, Timeout}, From, State) ->
{noreply, send_request(new_request(Msg, From, Timeout), State)};
handle_call({req, Msg, Timeout, Ctx}, From, State) ->
{noreply, send_request(new_request(Msg, From, Timeout, Ctx), State)};
handle_call(is_connected, _From, State) ->
case State#state.sock of
undefined ->
{reply, {false, State#state.failed}, State};
_ ->
{reply, true, State}
end;
handle_call({set_options, Options}, _From, State) ->
{reply, ok, parse_options(Options, State)};
handle_call(stop, _From, State) ->
_ = disconnect(State),
{stop, normal, ok, State}.
%% @private
handle_info({tcp_error, _Socket, Reason}, State) ->
error_logger:error_msg("PBC client TCP error for ~p:~p - ~p\n",
[State#state.address, State#state.port, Reason]),
disconnect(State);
handle_info({tcp_closed, _Socket}, State) ->
disconnect(State);
handle_info({ssl_error, _Socket, Reason}, State) ->
error_logger:error_msg("PBC client SSL error for ~p:~p - ~p\n",
[State#state.address, State#state.port, Reason]),
disconnect(State);
handle_info({ssl_closed, _Socket}, State) ->
disconnect(State);
%% Make sure the two Sock's match. If a request timed out, but there was
%% a response queued up behind it we do not want to process it. Instead
%% it should drop through and be ignored.
handle_info({Proto, Sock, Data}, State=#state{sock = Sock, active = Active})
when Proto == tcp; Proto == ssl ->
<> = Data,
Resp = case Active#request.msg of
{tunneled, _MsgID} ->
%% don't decode tunneled replies, we may not recognize the msgid
{MsgCode, MsgData};
_ ->
riak_pb_codec:decode(MsgCode, MsgData)
end,
NewState = case Resp of
#rpberrorresp{} ->
NewState1 = maybe_reply(on_error(Active, Resp, State)),
dequeue_request(NewState1#state{active = undefined});
_ ->
case process_response(Active, Resp, State) of
{reply, Response, NewState0} ->
%% Send reply and get ready for the next request - send the next request
%% if one is queued up
cancel_req_timer(Active#request.tref),
_ = send_caller(Response, NewState0#state.active),
dequeue_request(NewState0#state{active = undefined});
{pending, NewState0} -> %% Request is still pending - do not queue up a new one
NewActive = restart_req_timer(Active),
NewState0#state{active = NewActive}
end
end,
case State#state.transport of
gen_tcp ->
ok = inet:setopts(Sock, [{active, once}]);
ssl ->
ok = ssl:setopts(Sock, [{active, once}])
end,
{noreply, NewState};
handle_info({req_timeout, Ref}, State) ->
case State#state.active of %%
undefined ->
{noreply, remove_queued_request(Ref, State)};
Active ->
case Ref == Active#request.ref of
true -> %% Matches the current operation
NewState = maybe_reply(on_timeout(State#state.active, State)),
disconnect(NewState#state{active = undefined});
false ->
{noreply, remove_queued_request(Ref, State)}
end
end;
handle_info(reconnect, State) ->
case connect(State) of
{ok, NewState} ->
{noreply, dequeue_request(NewState)};
{error, Reason} ->
%% Update the failed count and reschedule a reconnection
NewState = State#state{failed = orddict:update_counter(Reason, 1, State#state.failed)},
disconnect(NewState)
end;
handle_info(_, State) ->
{noreply, State}.
%% @private
handle_cast(_Msg, State) ->
{noreply, State}.
%% @private
terminate(_Reason, _State) -> ok.
%% @private
code_change(_OldVsn, State, _Extra) -> {ok, State}.
%% ====================================================================
%% internal functions
%% ====================================================================
%% @private
%% Parse options
parse_options([], State) ->
%% Once all options are parsed, make sure auto_reconnect is enabled
%% if queue_if_disconnected is enabled.
case State#state.queue_if_disconnected of
true ->
State#state{auto_reconnect = true};
_ ->
State
end;
parse_options([{connect_timeout, T}|Options], State) when is_integer(T) ->
parse_options(Options, State#state{connect_timeout = T});
parse_options([{queue_if_disconnected,Bool}|Options], State) when
Bool =:= true; Bool =:= false ->
parse_options(Options, State#state{queue_if_disconnected = Bool});
parse_options([queue_if_disconnected|Options], State) ->
parse_options([{queue_if_disconnected, true}|Options], State);
parse_options([{auto_reconnect,Bool}|Options], State) when
Bool =:= true; Bool =:= false ->
parse_options(Options, State#state{auto_reconnect = Bool});
parse_options([auto_reconnect|Options], State) ->
parse_options([{auto_reconnect, true}|Options], State);
parse_options([{keepalive,Bool}|Options], State) when is_boolean(Bool) ->
parse_options(Options, State#state{keepalive = Bool});
parse_options([keepalive|Options], State) ->
parse_options([{keepalive, false}|Options], State);
parse_options([{credentials, User, Pass}|Options], State) ->
parse_options(Options, State#state{credentials={User, Pass}});
parse_options([{certfile, File}|Options], State) ->
parse_options(Options, State#state{certfile=File});
parse_options([{cacertfile, File}|Options], State) ->
parse_options(Options, State#state{cacertfile=File});
parse_options([{keyfile, File}|Options], State) ->
parse_options(Options, State#state{keyfile=File});
parse_options([{ssl_opts, Opts}|Options], State) ->
parse_options(Options, State#state{ssl_opts=Opts}).
maybe_reply({reply, Reply, State}) ->
Request = State#state.active,
NewRequest = send_caller(Reply, Request),
State#state{active = NewRequest};
maybe_reply({noreply, State}) ->
State.
%% @private
%% Reply to caller - form clause first in case a ReqId/Client was passed
%% in as the context and gen_server:reply hasn't been called yet.
send_caller(Msg, #request{ctx = {ReqId, Client},
from = undefined}=Request) ->
Client ! {ReqId, Msg},
Request;
send_caller(Msg, #request{from = From}=Request) when From /= undefined ->
gen_server:reply(From, Msg),
Request#request{from = undefined}.
get_options([], Req) ->
Req;
get_options([{basic_quorum, BQ} | Rest], Req) ->
get_options(Rest, Req#rpbgetreq{basic_quorum = BQ});
get_options([{notfound_ok, NFOk} | Rest], Req) ->
get_options(Rest, Req#rpbgetreq{notfound_ok = NFOk});
get_options([{r, R} | Rest], Req) ->
get_options(Rest, Req#rpbgetreq{r = riak_pb_kv_codec:encode_quorum(R)});
get_options([{pr, PR} | Rest], Req) ->
get_options(Rest, Req#rpbgetreq{pr = riak_pb_kv_codec:encode_quorum(PR)});
get_options([{timeout, T} | Rest], Req) when is_integer(T)->
get_options(Rest, Req#rpbgetreq{timeout = T});
get_options([{timeout, _T} | _Rest], _Req) ->
erlang:error(badarg);
get_options([{if_modified, VClock} | Rest], Req) ->
get_options(Rest, Req#rpbgetreq{if_modified = VClock});
get_options([head | Rest], Req) ->
get_options(Rest, Req#rpbgetreq{head = true});
get_options([deletedvclock | Rest], Req) ->
get_options(Rest, Req#rpbgetreq{deletedvclock = true});
get_options([{n_val, N} | Rest], Req)
when is_integer(N), N > 0 ->
get_options(Rest, Req#rpbgetreq{n_val = N});
get_options([{sloppy_quorum, Bool} | Rest], Req)
when Bool == true; Bool == false ->
get_options(Rest, Req#rpbgetreq{sloppy_quorum = Bool});
get_options([{_, _} | _Rest], _Req) ->
erlang:error(badarg).
put_options([], Req) ->
Req;
put_options([{w, W} | Rest], Req) ->
put_options(Rest, Req#rpbputreq{w = riak_pb_kv_codec:encode_quorum(W)});
put_options([{dw, DW} | Rest], Req) ->
put_options(Rest, Req#rpbputreq{dw = riak_pb_kv_codec:encode_quorum(DW)});
put_options([{pw, PW} | Rest], Req) ->
put_options(Rest, Req#rpbputreq{pw = riak_pb_kv_codec:encode_quorum(PW)});
put_options([{timeout, T} | Rest], Req) when is_integer(T) ->
put_options(Rest, Req#rpbputreq{timeout = T});
put_options([{timeout, _T} | _Rest], _Req) ->
erlang:error(badarg);
put_options([return_body | Rest], Req) ->
put_options(Rest, Req#rpbputreq{return_body = 1});
put_options([return_head | Rest], Req) ->
put_options(Rest, Req#rpbputreq{return_head = true});
put_options([if_not_modified | Rest], Req) ->
put_options(Rest, Req#rpbputreq{if_not_modified = true});
put_options([if_none_match | Rest], Req) ->
put_options(Rest, Req#rpbputreq{if_none_match = true});
put_options([asis | Rest], Req) ->
put_options(Rest, Req#rpbputreq{asis = true});
put_options([{asis, Val} | Rest], Req) when is_boolean(Val) ->
put_options(Rest, Req#rpbputreq{asis = Val});
put_options([{n_val, N} | Rest], Req)
when is_integer(N), N > 0 ->
put_options(Rest, Req#rpbputreq{n_val = N});
put_options([{sloppy_quorum, Bool} | Rest], Req)
when Bool == true; Bool == false ->
put_options(Rest, Req#rpbputreq{sloppy_quorum = Bool});
put_options([{_, _} | _Rest], _Req) ->
erlang:error(badarg).
delete_options([], Req) ->
Req;
delete_options([{rw, RW} | Rest], Req) ->
delete_options(Rest, Req#rpbdelreq{rw = riak_pb_kv_codec:encode_quorum(RW)});
delete_options([{r, R} | Rest], Req) ->
delete_options(Rest, Req#rpbdelreq{r = riak_pb_kv_codec:encode_quorum(R)});
delete_options([{w, W} | Rest], Req) ->
delete_options(Rest, Req#rpbdelreq{w = riak_pb_kv_codec:encode_quorum(W)});
delete_options([{pr, PR} | Rest], Req) ->
delete_options(Rest, Req#rpbdelreq{pr = riak_pb_kv_codec:encode_quorum(PR)});
delete_options([{pw, PW} | Rest], Req) ->
delete_options(Rest, Req#rpbdelreq{pw = riak_pb_kv_codec:encode_quorum(PW)});
delete_options([{dw, DW} | Rest], Req) ->
delete_options(Rest, Req#rpbdelreq{dw = riak_pb_kv_codec:encode_quorum(DW)});
delete_options([{timeout, T} | Rest], Req) when is_integer(T) ->
delete_options(Rest, Req#rpbdelreq{timeout = T});
delete_options([{timeout, _T} | _Rest], _Req) ->
erlang:error(badarg);
delete_options([{n_val, N} | Rest], Req)
when is_integer(N), N > 0 ->
delete_options(Rest, Req#rpbdelreq{n_val = N});
delete_options([{sloppy_quorum, Bool} | Rest], Req)
when Bool == true; Bool == false ->
delete_options(Rest, Req#rpbdelreq{sloppy_quorum = Bool});
delete_options([{_, _} | _Rest], _Req) ->
erlang:error(badarg).
search_options([], Req) ->
Req;
search_options([{rows, Rows} | Rest], Req) ->
search_options(Rest, Req#rpbsearchqueryreq{rows=Rows});
search_options([{start, Start} | Rest], Req) ->
search_options(Rest, Req#rpbsearchqueryreq{start=Start});
search_options([{sort, Sort} | Rest], Req) ->
search_options(Rest, Req#rpbsearchqueryreq{sort=Sort});
search_options([{filter, Filter} | Rest], Req) ->
search_options(Rest, Req#rpbsearchqueryreq{filter=Filter});
search_options([{df, DF} | Rest], Req) ->
search_options(Rest, Req#rpbsearchqueryreq{df=DF});
search_options([{op, OP} | Rest], Req) ->
search_options(Rest, Req#rpbsearchqueryreq{op=OP});
search_options([{fl, FL} | Rest], Req) ->
search_options(Rest, Req#rpbsearchqueryreq{fl=FL});
search_options([{presort, Presort} | Rest], Req) ->
search_options(Rest, Req#rpbsearchqueryreq{presort=Presort});
search_options([{_, _} | _Rest], _Req) ->
erlang:error(badarg).
counter_incr_options([], Req) ->
Req;
counter_incr_options([{w, W} | Rest], Req) ->
counter_incr_options(Rest, Req#rpbcounterupdatereq{w=riak_pb_kv_codec:encode_quorum(W)});
counter_incr_options([{dw, DW} | Rest], Req) ->
counter_incr_options(Rest, Req#rpbcounterupdatereq{dw=riak_pb_kv_codec:encode_quorum(DW)});
counter_incr_options([{pw, PW} | Rest], Req) ->
counter_incr_options(Rest, Req#rpbcounterupdatereq{pw=riak_pb_kv_codec:encode_quorum(PW)});
counter_incr_options([returnvalue | Rest], Req) ->
counter_incr_options(Rest, Req#rpbcounterupdatereq{returnvalue=true});
counter_incr_options([_ | _Rest], _Req) ->
erlang:error(badarg).
counter_val_options([], Req) ->
Req;
counter_val_options([{basic_quorum, BQ} | Rest], Req) ->
counter_val_options(Rest, Req#rpbcountergetreq{basic_quorum=BQ});
counter_val_options([{notfound_ok, NFOK} | Rest], Req) ->
counter_val_options(Rest, Req#rpbcountergetreq{notfound_ok=NFOK});
counter_val_options([{r, R} | Rest], Req) ->
counter_val_options(Rest, Req#rpbcountergetreq{r=riak_pb_kv_codec:encode_quorum(R)});
counter_val_options([{pr, PR} | Rest], Req) ->
counter_val_options(Rest, Req#rpbcountergetreq{pr=riak_pb_kv_codec:encode_quorum(PR)});
counter_val_options([_ | _Rest], _Req) ->
erlang:error(badarg).
%% Process response from the server - passes back in the request and
%% context the request was issued with.
%% Return noreply if the request is completed, but no reply needed
%% reply if the request is completed with a reply to the caller
%% pending if the request has not completed yet (streaming op)
%% @private
-spec process_response(#request{}, rpb_resp(), #state{}) ->
{reply, term(), #state{}} |
{pending, #state{}}.
process_response(#request{msg = rpbpingreq}, rpbpingresp, State) ->
{reply, pong, State};
process_response(#request{msg = rpbgetclientidreq},
#rpbgetclientidresp{client_id = ClientId}, State) ->
{reply, {ok, ClientId}, State};
process_response(#request{msg = #rpbsetclientidreq{}},
rpbsetclientidresp, State) ->
{reply, ok, State};
process_response(#request{msg = rpbgetserverinforeq},
#rpbgetserverinforesp{node = Node, server_version = ServerVersion}, State) ->
NodeInfo = case Node of
undefined ->
[];
Node ->
[{node, Node}]
end,
VersionInfo = case ServerVersion of
undefined ->
[];
ServerVersion ->
[{server_version, ServerVersion}]
end,
{reply, {ok, NodeInfo++VersionInfo}, State};
process_response(#request{msg = #rpbgetreq{}}, rpbgetresp, State) ->
%% server just returned the rpbgetresp code - no message was encoded
{reply, {error, notfound}, State};
process_response(#request{msg = #rpbgetreq{deletedvclock=true}},
#rpbgetresp{vclock=VC, content=[]}, State) ->
%% server returned a notfound with a vector clock, meaning a tombstone
{reply, {error, notfound, VC}, State};
process_response(#request{msg = #rpbgetreq{}}, #rpbgetresp{unchanged=true}, State) ->
%% object was unchanged
{reply, unchanged, State};
process_response(#request{msg = #rpbgetreq{type = Type, bucket = Bucket, key = Key}},
#rpbgetresp{content = RpbContents, vclock = Vclock}, State) ->
Contents = riak_pb_kv_codec:decode_contents(RpbContents),
B = maybe_make_bucket_type(Type, Bucket),
{reply, {ok, riakc_obj:new_obj(B, Key, Vclock, Contents)}, State};
process_response(#request{msg = #rpbputreq{}},
rpbputresp, State) ->
%% server just returned the rpbputresp code - no message was encoded
{reply, ok, State};
process_response(#request{ msg = #rpbputreq{}},
#rpbputresp{key = Key, content=undefined, vclock=undefined},
State) when is_binary(Key) ->
%% server generated a key and the client didn't request return_body, but
%% the created key is returned
{reply, {ok, Key}, State};
process_response(#request{msg = #rpbputreq{type = Type, bucket = Bucket, key = Key}},
#rpbputresp{content = RpbContents, vclock = Vclock,
key = NewKey}, State) ->
Contents = riak_pb_kv_codec:decode_contents(RpbContents),
ReturnKey = case NewKey of
undefined -> Key;
_ -> NewKey
end,
B = maybe_make_bucket_type(Type, Bucket),
{reply, {ok, riakc_obj:new_obj(B, ReturnKey, Vclock, Contents)}, State};
process_response(#request{msg = #rpbdelreq{}},
rpbdelresp, State) ->
%% server just returned the rpbdelresp code - no message was encoded
{reply, ok, State};
process_response(#request{msg = #rpblistbucketsreq{}}=Request,
#rpblistbucketsresp{buckets = Buckets, done = undefined},
State) ->
_ = send_caller({buckets, Buckets}, Request),
{pending, State};
process_response(#request{msg = #rpblistbucketsreq{}},
#rpblistbucketsresp{done = true},
State) ->
{reply, done, State};
process_response(#request{msg = #rpblistkeysreq{}}=Request,
#rpblistkeysresp{done = Done, keys = Keys}, State) ->
_ = case Keys of
undefined ->
ok;
_ ->
%% Have to directly use send_caller as may want to reply with done below.
send_caller({keys, Keys}, Request)
end,
case Done of
true ->
{reply, done, State};
_ ->
{pending, State}
end;
process_response(#request{msg = #rpbgetbucketreq{}},
#rpbgetbucketresp{props = PbProps}, State) ->
Props = riak_pb_codec:decode_bucket_props(PbProps),
{reply, {ok, Props}, State};
process_response(#request{msg = #rpbgetbuckettypereq{}},
#rpbgetbucketresp{props = PbProps}, State) ->
Props = riak_pb_codec:decode_bucket_props(PbProps),
{reply, {ok, Props}, State};
process_response(#request{msg = #rpbsetbucketreq{}},
rpbsetbucketresp, State) ->
{reply, ok, State};
process_response(#request{msg = #rpbsetbuckettypereq{}},
rpbsetbucketresp, State) ->
{reply, ok, State};
process_response(#request{msg = #rpbmapredreq{content_type = ContentType}}=Request,
#rpbmapredresp{done = Done, phase=PhaseId, response=Data}, State) ->
_ = case Data of
undefined ->
ok;
_ ->
Response = decode_mapred_resp(Data, ContentType),
send_caller({mapred, PhaseId, Response}, Request)
end,
case Done of
true ->
{reply, done, State};
_ ->
{pending, State}
end;
process_response(#request{msg = #rpbindexreq{}}, rpbindexresp, State) ->
Results = ?INDEX_RESULTS{keys=[], continuation=undefined},
{reply, {ok, Results}, State};
process_response(#request{msg = #rpbindexreq{stream=true, return_terms=Terms}}=Request,
#rpbindexresp{results=Results, keys=Keys, done=Done, continuation=Cont}, State) ->
ToSend = process_index_response(Terms, Keys, Results),
_ = send_caller(ToSend, Request),
DoneResponse = {reply, {done, Cont}, State},
case Done of
true -> DoneResponse;
_ -> {pending, State}
end;
process_response(#request{msg = #rpbindexreq{return_terms=Terms}}, #rpbindexresp{results=Results, keys=Keys, continuation=Cont}, State) ->
StreamResponse = process_index_response(Terms, Keys, Results),
RegularResponse = index_stream_result_to_index_result(StreamResponse),
RegularResponseWithContinuation = RegularResponse?INDEX_RESULTS{continuation=Cont},
{reply, {ok, RegularResponseWithContinuation}, State};
process_response(#request{msg = #rpbcsbucketreq{}}, rpbcsbucketresp, State) ->
{pending, State};
process_response(#request{msg = #rpbcsbucketreq{bucket=Bucket}}=Request, #rpbcsbucketresp{objects=Objects, done=Done, continuation=Cont}, State) ->
%% TEMP - cs specific message for fold_objects
ToSend = case Objects of
undefined -> {ok, []};
_ ->
%% make client objects
CObjects = lists:foldr(fun(#rpbindexobject{key=Key,
object=#rpbgetresp{content=Contents, vclock=VClock}}, Acc) ->
DContents = riak_pb_kv_codec:decode_contents(Contents),
[riakc_obj:new_obj(Bucket, Key, VClock, DContents) | Acc] end,
[],
Objects),
{ok, CObjects}
end,
_ = send_caller(ToSend, Request),
DoneResponse = {reply, {done, Cont}, State},
case Done of
true -> DoneResponse;
_ -> {pending, State}
end;
process_response(#request{msg = #rpbsearchqueryreq{}}, prbsearchqueryresp, State) ->
{reply, {error, notfound}, State};
process_response(#request{msg = #rpbsearchqueryreq{index=Index}},
#rpbsearchqueryresp{docs=PBDocs,max_score=MaxScore,
num_found=NumFound}, State) ->
Values = [ {Index, [ riak_pb_codec:decode_pair(Field) || Field <- Doc#rpbsearchdoc.fields] }
|| Doc <- PBDocs ],
Result = #search_results{docs=Values, max_score=MaxScore, num_found=NumFound},
{reply, {ok, Result}, State};
process_response(#request{msg=#rpbresetbucketreq{}}, rpbresetbucketresp, State) ->
{reply, ok, State};
process_response(#request{msg = #rpbcounterupdatereq{returnvalue=true}},
#rpbcounterupdateresp{value=Value}, State) ->
{reply, {ok, Value}, State};
process_response(#request{msg = #rpbcounterupdatereq{}},
rpbcounterupdateresp, State) ->
%% server just returned the rpbcounterupdateresp code - no message was encoded
{reply, ok, State};
process_response(#request{msg = #rpbcountergetreq{}},
rpbcountergetresp, State) ->
{reply, {error, notfound}, State};
process_response(#request{msg = #rpbcountergetreq{}},
#rpbcountergetresp{value=Value}, State) ->
{reply, {ok, Value}, State};
process_response(#request{msg = #dtfetchreq{}}, #dtfetchresp{}=Resp,
State) ->
Reply = case riak_pb_dt_codec:decode_fetch_response(Resp) of
{Type, Value, Context} ->
Mod = riakc_datatype:module_for_type(Type),
{ok, Mod:new(Value, Context)};
{notfound, _Type}=NF ->
{error, NF}
end,
{reply, Reply, State};
process_response(#request{msg = #dtupdatereq{}},
dtupdateresp,
State) ->
{reply, ok, State};
process_response(#request{msg = #dtupdatereq{op=Op, return_body=RB}},
#dtupdateresp{}=Resp,
State) ->
OpType = riak_pb_dt_codec:operation_type(Op),
Reply = case riak_pb_dt_codec:decode_update_response(Resp, OpType, RB) of
ok -> ok;
{ok, Key} -> {ok, Key};
{OpType, Value, Context} ->
Mod = riakc_datatype:module_for_type(OpType),
{ok, Mod:new(Value, Context)};
{Key, {OpType, Value, Context}} when is_binary(Key) ->
Mod = riakc_datatype:module_for_type(OpType),
{ok, Key, Mod:new(Value, Context)}
end,
{reply, Reply, State};
process_response(#request{msg={tunneled,_MsgId}}, Reply, State) ->
%% Tunneled msg response
{reply, {ok, Reply}, State};
process_response(#request{msg = #rpbyokozunaschemaputreq{}},
rpbputresp, State) ->
{reply, ok, State};
process_response(#request{msg = #rpbyokozunaindexputreq{}},
rpbputresp, State) ->
{reply, ok, State};
process_response(#request{msg = #rpbyokozunaindexdeletereq{}},
rpbdelresp, State) ->
{reply, ok, State};
process_response(#request{msg = #rpbyokozunaindexgetreq{}},
rpbyokozunaindexgetresp, State) ->
{reply, {ok, []}, State};
process_response(#request{msg = #rpbyokozunaindexgetreq{}},
#rpbyokozunaindexgetresp{index=Indexes}, State) ->
Results = [[{index,Index#rpbyokozunaindex.name},
{schema,Index#rpbyokozunaindex.schema},
{n_val,Index#rpbyokozunaindex.n_val}]
|| Index <- Indexes ],
{reply, {ok, Results}, State};
process_response(#request{msg = #rpbyokozunaschemagetreq{}},
#rpbyokozunaschemagetresp{schema=Schema}, State) ->
Result = [{name,Schema#rpbyokozunaschema.name}, {content,Schema#rpbyokozunaschema.content}],
{reply, {ok, Result}, State};
process_response(Request, Reply, State) ->
%% Unknown request/response combo
{reply, {error, {unknown_response, Request, Reply}}, State}.
%% Helper for index responses
-spec process_index_response(undefined | boolean(), list(), list()) ->
index_stream_result().
process_index_response(undefined, Keys, _) ->
?INDEX_STREAM_RESULT{keys=Keys};
process_index_response(false, Keys, _) ->
?INDEX_STREAM_RESULT{keys=Keys};
process_index_response(true, [], Results) ->
%% rpbpair is abused to send Value,Key pairs as Key, Value pairs
%% in a 2i query the 'key' is the index value and the 'value'
%% the indexed objects primary key
Res = [{V, K} || #rpbpair{key=V, value=K} <- Results],
?INDEX_STREAM_RESULT{terms=Res};
process_index_response(true, Keys, []) ->
?INDEX_STREAM_RESULT{keys=Keys}.
-spec index_stream_result_to_index_result(index_stream_result()) ->
index_results().
index_stream_result_to_index_result(?INDEX_STREAM_RESULT{keys=Keys,
terms=Terms}) ->
?INDEX_RESULTS{keys=Keys,
terms=Terms}.
%% Called after sending a message - supports returning a
%% request id for streaming calls
%% @private
after_send(#request{msg = #rpblistbucketsreq{}, ctx = {ReqId, _Client}},
State) ->
{reply, {ok, ReqId}, State};
after_send(#request{msg = #rpblistkeysreq{}, ctx = {ReqId, _Client}}, State) ->
{reply, {ok, ReqId}, State};
after_send(#request{msg = #rpbmapredreq{}, ctx = {ReqId, _Client}}, State) ->
{reply, {ok, ReqId}, State};
after_send(#request{msg = #rpbindexreq{stream=true}, ctx = {ReqId, _Client}}, State) ->
{reply, {ok, ReqId}, State};
after_send(#request{msg = #rpbcsbucketreq{}, ctx = {ReqId, _Client}}, State) ->
{reply, {ok, ReqId}, State};
after_send(_Request, State) ->
{noreply, State}.
%% Called on timeout for an operation
%% @private
on_timeout(_Request, State) ->
{reply, {error, timeout}, State}.
%%
%% Called after receiving an error message - supports reruning
%% an error for streaming calls
%% @private
on_error(_Request, ErrMsg, State) ->
{reply, fmt_err_msg(ErrMsg), State}.
%% Format the PB encoded error message
fmt_err_msg(ErrMsg) ->
case ErrMsg#rpberrorresp.errcode of
Code when Code =:= 0; Code =:= 1; Code =:= undefined ->
{error, ErrMsg#rpberrorresp.errmsg};
Code ->
{error, {Code, ErrMsg#rpberrorresp.errmsg}}
end.
%% deliberately crash if the handling an error response after
%% the client has been replied to
%% Common code for sending a single bucket or multiple inputs map/request
%% @private
send_mapred_req(Pid, MapRed, ClientPid) ->
ReqMsg = #rpbmapredreq{request = encode_mapred_req(MapRed),
content_type = <<"application/x-erlang-binary">>},
ReqId = mk_reqid(),
Timeout = proplists:get_value(timeout, MapRed, default_timeout(mapred_timeout)),
Timeout1 = if
is_integer(Timeout) ->
%% Add an extra 100ms to the mapred timeout and use that
%% for the socket timeout. This should give the
%% map/reduce a chance to fail and let us know.
Timeout + 100;
true ->
Timeout
end,
call_infinity(Pid, {req, ReqMsg, Timeout1, {ReqId, ClientPid}}).
%% @private
%% Make a new request that can be sent or queued
new_request(Msg, From, Timeout) ->
Ref = make_ref(),
#request{ref = Ref, msg = Msg, from = From, timeout = Timeout,
tref = create_req_timer(Timeout, Ref)}.
new_request(Msg, From, Timeout, Context) ->
Ref = make_ref(),
#request{ref =Ref, msg = Msg, from = From, ctx = Context, timeout = Timeout,
tref = create_req_timer(Timeout, Ref)}.
%% @private
%% Create a request timer if desired, otherwise return undefined.
create_req_timer(infinity, _Ref) ->
undefined;
create_req_timer(undefined, _Ref) ->
undefined;
create_req_timer(Msecs, Ref) ->
erlang:send_after(Msecs, self(), {req_timeout, Ref}).
%% @private
%% Cancel a request timer made by create_timer/2
cancel_req_timer(undefined) ->
ok;
cancel_req_timer(Tref) ->
_ = erlang:cancel_timer(Tref),
ok.
%% @private
%% Restart a request timer
-spec restart_req_timer(#request{}) -> #request{}.
restart_req_timer(Request) ->
case Request#request.tref of
undefined ->
Request;
Tref ->
cancel_req_timer(Tref),
NewTref = create_req_timer(Request#request.timeout,
Request#request.ref),
Request#request{tref = NewTref}
end.
%% @private
%% Connect the socket if disconnected
connect(State) when State#state.sock =:= undefined ->
#state{address = Address, port = Port, connects = Connects} = State,
case gen_tcp:connect(Address, Port,
[binary, {active, once}, {packet, 4},
{keepalive, State#state.keepalive}],
State#state.connect_timeout) of
{ok, Sock} ->
State1 = State#state{sock = Sock, connects = Connects+1,
reconnect_interval = ?FIRST_RECONNECT_INTERVAL},
case State#state.credentials of
undefined ->
{ok, State1};
_ ->
start_tls(State1)
end;
Error ->
Error
end.
-spec start_tls(#state{}) -> {ok, #state{}} | {error, term()}.
start_tls(State=#state{sock=Sock}) ->
%% Send STARTTLS
StartTLSCode = riak_pb_codec:msg_code(rpbstarttls),
ok = gen_tcp:send(Sock, <>),
receive
{tcp_error, Sock, Reason} ->
{error, Reason};
{tcp_closed, Sock} ->
{error, closed};
{tcp, Sock, Data} ->
<> = Data,
case riak_pb_codec:decode(MsgCode, MsgData) of
rpbstarttls ->
Options = [{verify, verify_peer},
{cacertfile, State#state.cacertfile}] ++
[{K, V} || {K, V} <- [{certfile,
State#state.certfile},
{keyfile,
State#state.keyfile}],
V /= undefined] ++
State#state.ssl_opts,
case ssl:connect(Sock, Options, 1000) of
{ok, SSLSock} ->
ok = ssl:setopts(SSLSock, [{active, once}]),
start_auth(State#state{sock=SSLSock, transport=ssl});
{error, Reason2} ->
{error, Reason2}
end;
#rpberrorresp{} ->
%% Server doesn't know about STARTTLS or security is
%% disabled. We can't fall back to the regular old
%% protocol here because then SSL could be stripped by a
%% man-in-the-middle proxy that presents insecure
%% communication to the client, but does secure
%% communication to the server.
{error, no_security}
end
end.
start_auth(State=#state{credentials={User,Pass}, sock=Sock}) ->
ok = ssl:send(Sock, riak_pb_codec:encode(#rpbauthreq{user=User,
password=Pass})),
receive
{ssl_error, Sock, Reason} ->
{error, Reason};
{ssl_closed, Sock} ->
{error, closed};
{ssl, Sock, Data} ->
<> = Data,
case riak_pb_codec:decode(MsgCode, MsgData) of
rpbauthresp ->
ok = ssl:setopts(Sock, [{active, once}]),
{ok, State};
#rpberrorresp{} = Err ->
fmt_err_msg(Err)
end
end.
%% @private
%% Disconnect socket if connected
disconnect(State) ->
%% Tell any pending requests we've disconnected
_ = case State#state.active of
undefined ->
ok;
Request ->
send_caller({error, disconnected}, Request)
end,
%% Make sure the connection is really closed
case State#state.sock of
undefined ->
ok;
Sock ->
Transport = State#state.transport,
Transport:close(Sock)
end,
%% Decide whether to reconnect or exit
NewState = State#state{sock = undefined, active = undefined},
case State#state.auto_reconnect of
true ->
%% Schedule the reconnect message and return state
erlang:send_after(State#state.reconnect_interval, self(), reconnect),
{noreply, increase_reconnect_interval(NewState)};
false ->
{stop, disconnected, NewState}
end.
%% Double the reconnect interval up to the maximum
increase_reconnect_interval(State) ->
case State#state.reconnect_interval of
Interval when Interval < ?MAX_RECONNECT_INTERVAL ->
NewInterval = min(Interval+Interval, ?MAX_RECONNECT_INTERVAL),
State#state{reconnect_interval = NewInterval};
_ ->
State
end.
%% Send a request to the server and prepare the state for the response
%% @private
send_request(Request0, State) when State#state.active =:= undefined ->
{Request, Pkt} = encode_request_message(Request0),
Transport = State#state.transport,
case Transport:send(State#state.sock, Pkt) of
ok ->
maybe_reply(after_send(Request, State#state{active = Request}));
{error, Reason} ->
error_logger:warning_msg("Socket error while sending riakc request: ~p.", [Reason]),
Transport:close(State#state.sock),
maybe_enqueue_and_reconnect(Request, State#state{sock=undefined})
end.
%% Already encoded (for tunneled messages), but must provide Message Id
%% for responding to the second form of send_request.
encode_request_message(#request{msg={tunneled,MsgId,Pkt}}=Req) ->
{Req#request{msg={tunneled,MsgId}},[MsgId|Pkt]};
%% Unencoded Request (the normal PB client path)
encode_request_message(#request{msg=Msg}=Req) ->
{Req, riak_pb_codec:encode(Msg)}.
%% If the socket was closed, see if we can enqueue the request and
%% trigger a reconnect. Otherwise, return an error to the requestor.
maybe_enqueue_and_reconnect(Request, State) ->
maybe_reconnect(State),
enqueue_or_reply_error(Request, State).
%% Trigger an immediate reconnect if automatic reconnection is
%% enabled.
maybe_reconnect(#state{auto_reconnect=true}) -> self() ! reconnect;
maybe_reconnect(_) -> ok.
%% If we can queue while disconnected, do so, otherwise tell the
%% caller that the socket was disconnected.
enqueue_or_reply_error(Request, #state{queue_if_disconnected=true}=State) ->
queue_request(Request, State);
enqueue_or_reply_error(Request, State) ->
_ = send_caller({error, disconnected}, Request),
State.
%% Queue up a request if one is pending
%% @private
queue_request(Request, State) ->
State#state{queue = queue:in(Request, State#state.queue)}.
%% Try and dequeue request and send onto the server if one is waiting
%% @private
dequeue_request(State) ->
case queue:out(State#state.queue) of
{empty, _} ->
State;
{{value, Request}, Q2} ->
send_request(Request, State#state{queue = Q2})
end.
%% Remove a queued request by reference - returns same queue if ref not present
%% @private
remove_queued_request(Ref, State) ->
L = queue:to_list(State#state.queue),
case lists:keytake(Ref, #request.ref, L) of
false -> % Ref not queued up
State;
{value, Req, L2} ->
{reply, Reply, NewState} = on_timeout(Req, State),
_ = send_caller(Reply, Req),
NewState#state{queue = queue:from_list(L2)}
end.
%% @private
mk_reqid() -> erlang:phash2(erlang:now()). % only has to be unique per-pid
%% @private
wait_for_list(ReqId) ->
wait_for_list(ReqId, []).
%% @private
wait_for_list(ReqId, Acc) ->
receive
{ReqId, done} -> {ok, lists:flatten(Acc)};
{ReqId, {error, Reason}} -> {error, Reason};
{ReqId, {_, Res}} -> wait_for_list(ReqId, [Res|Acc])
end.
%% @private
wait_for_mapred(ReqId, Timeout) ->
wait_for_mapred_first(ReqId, Timeout).
%% Wait for the first mapred result, so we know at least one phase
%% that will be delivering results.
wait_for_mapred_first(ReqId, Timeout) ->
case receive_mapred(ReqId, Timeout) of
done ->
{ok, []};
{mapred, Phase, Res} ->
wait_for_mapred_one(ReqId, Timeout, Phase,
acc_mapred_one(Res, []));
{error, _}=Error ->
Error;
timeout ->
{error, {timeout, []}}
end.
%% So far we have only received results from one phase. This method
%% of accumulating a single phases's outputs will be more efficient
%% than the repeated orddict:append_list/3 used when accumulating
%% outputs from multiple phases.
wait_for_mapred_one(ReqId, Timeout, Phase, Acc) ->
case receive_mapred(ReqId, Timeout) of
done ->
{ok, finish_mapred_one(Phase, Acc)};
{mapred, Phase, Res} ->
%% still receiving for just one phase
wait_for_mapred_one(ReqId, Timeout, Phase,
acc_mapred_one(Res, Acc));
{mapred, NewPhase, Res} ->
%% results from a new phase have arrived - track them all
Dict = [{NewPhase, Res},{Phase, Acc}],
wait_for_mapred_many(ReqId, Timeout, Dict);
{error, _}=Error ->
Error;
timeout ->
{error, {timeout, finish_mapred_one(Phase, Acc)}}
end.
%% Single-phase outputs are kept as a reverse list of results.
acc_mapred_one([R|Rest], Acc) ->
acc_mapred_one(Rest, [R|Acc]);
acc_mapred_one([], Acc) ->
Acc.
finish_mapred_one(Phase, Acc) ->
[{Phase, lists:reverse(Acc)}].
%% Tracking outputs from multiple phases.
wait_for_mapred_many(ReqId, Timeout, Acc) ->
case receive_mapred(ReqId, Timeout) of
done ->
{ok, finish_mapred_many(Acc)};
{mapred, Phase, Res} ->
wait_for_mapred_many(
ReqId, Timeout, acc_mapred_many(Phase, Res, Acc));
{error, _}=Error ->
Error;
timeout ->
{error, {timeout, finish_mapred_many(Acc)}}
end.
%% Many-phase outputs are kepts as a proplist of reversed lists of
%% results.
acc_mapred_many(Phase, Res, Acc) ->
case lists:keytake(Phase, 1, Acc) of
{value, {Phase, PAcc}, RAcc} ->
[{Phase,acc_mapred_one(Res,PAcc)}|RAcc];
false ->
[{Phase,acc_mapred_one(Res,[])}|Acc]
end.
finish_mapred_many(Acc) ->
[ {P, lists:reverse(A)} || {P, A} <- lists:keysort(1, Acc) ].
%% Receive one mapred message.
-spec receive_mapred(req_id(), timeout()) ->
done | {mapred, integer(), [term()]} | {error, term()} | timeout.
receive_mapred(ReqId, Timeout) ->
receive {ReqId, Msg} ->
%% Msg should be `done', `{mapred, Phase, Results}', or
%% `{error, Reason}'
Msg
after Timeout ->
timeout
end.
%% Encode the MapReduce request using term to binary
%% @private
-spec encode_mapred_req(term()) -> binary().
encode_mapred_req(Req) ->
term_to_binary(Req).
%% Decode a partial phase response
%% @private
-spec decode_mapred_resp(binary(), binary()) -> term().
decode_mapred_resp(Data, <<"application/x-erlang-binary">>) ->
try
binary_to_term(Data)
catch
_:Error -> % On error, merge in with the other results
[{error, Error}]
end.
maybe_bucket_type({Type, Bucket}) ->
{Type, Bucket};
maybe_bucket_type(Bucket) ->
{undefined, Bucket}.
maybe_make_bucket_type(undefined, Bucket) ->
Bucket;
maybe_make_bucket_type(Type, Bucket) ->
{Type, Bucket}.
%% ====================================================================
%% unit tests
%% ====================================================================
%% Tests disabled until they can be prevented from running when included
%% as a dependency.
%%
-ifdef(TEST).
-compile(export_all).
-include_lib("eunit/include/eunit.hrl").
%% Get the test host - check env RIAK_TEST_PB_HOST then env 'RIAK_TEST_HOST_1'
%% falling back to 127.0.0.1
test_ip() ->
case os:getenv("RIAK_TEST_PB_HOST") of
false ->
case os:getenv("RIAK_TEST_HOST_1") of
false ->
"127.0.0.1";
Host ->
Host
end;
Host ->
Host
end.
%% Test port - check env RIAK_TEST_PBC_1
test_port() ->
case os:getenv("RIAK_TEST_PBC_1") of
false ->
8087;
PortStr ->
list_to_integer(PortStr)
end.
%% Riak node under test - used to setup/configure/tweak it for tests
test_riak_node() ->
case os:getenv("RIAK_TEST_NODE_1") of
false ->
'riak@127.0.0.1';
NodeStr ->
list_to_atom(NodeStr)
end.
%% Node for the eunit node for distributed erlang
test_eunit_node() ->
case os:getenv("RIAK_EUNIT_NODE") of
false ->
'eunit@127.0.0.1';
EunitNodeStr ->
list_to_atom(EunitNodeStr)
end.
%% Cookie for distributed erlang
test_cookie() ->
case os:getenv("RIAK_TEST_COOKIE") of
false ->
'riak';
CookieStr ->
list_to_atom(CookieStr)
end.
%% Get the riak version from the init boot script, turn it into a list
%% of integers.
riak_version() ->
StrVersion = element(2, rpc:call(test_riak_node(), init, script_id, [])),
{match, [Major, Minor, Patch|_]} = re:run(StrVersion, "\\d+", [global, {capture, first, list}]),
[ list_to_integer(V) || [V] <- [Major, Minor, Patch]].
%% Compare the first three part version array with the second.
%% returns `greater', `less', or `equal'.
compare_versions([M1,N1,P1], [M2,N2,P2]) ->
V1 = (M1*1000000)+(N1*1000)+(P1),
V2 = (M2*1000000)+(N2*1000)+(P2),
case {V1 > V2, V1 == V2} of
{true,_} ->
greater;
{false,false} ->
less;
{false,true} ->
equal
end.
%% Resets the riak node
reset_riak() ->
%% sleep because otherwise we're going to kill the vnodes too fast
%% for the supervisor's maximum restart frequency, which will bring
%% down the entire node
?assertEqual(ok, maybe_start_network()),
case compare_versions(riak_version(), [1,2,0]) of
less ->
reset_riak_legacy();
_ ->
reset_riak_12()
end.
reset_solr(Pid) ->
%% clear indexes
{ok, Indexes} = ?MODULE:list_search_indexes(Pid),
[ ?MODULE:delete_search_index(Pid, proplists:get_value(index,Index)) || Index <- Indexes ],
wait_until( fun() ->
{ok, []} == ?MODULE:list_search_indexes(Pid)
end, 20, 1000),
ok.
%% Resets a Riak 1.2+ node, which can run the memory backend in 'test'
%% mode.
reset_riak_12() ->
set_test_backend(),
ok = rpc:call(test_riak_node(), riak_kv_memory_backend, reset, []),
reset_ring().
%% Sets up the memory/test backend, leaving it alone if already set properly.
set_test_backend() ->
Env = rpc:call(test_riak_node(), application, get_all_env, [riak_kv]),
Backend = proplists:get_value(storage_backend, Env),
Test = proplists:get_value(test, Env),
case {Backend, Test} of
{riak_kv_memory_backend, true} ->
ok;
_ ->
ok = rpc:call(test_riak_node(), application, set_env, [riak_kv, storage_backend, riak_kv_memory_backend]),
ok = rpc:call(test_riak_node(), application, set_env, [riak_kv, test, true]),
Vnodes = rpc:call(test_riak_node(), riak_core_vnode_manager, all_vnodes, [riak_kv_vnode]),
[ ok = rpc:call(test_riak_node(), supervisor, terminate_child, [riak_core_vnode_sup, Pid]) ||
{_, _, Pid} <- Vnodes ]
end.
%% Resets a Riak 1.1 and earlier node.
reset_riak_legacy() ->
timer:sleep(500),
%% Until there is a good way to empty the vnodes, require the
%% test to run with ETS and kill the vnode master/sup to empty all the ETS tables
%% and the ring manager to remove any bucket properties
ok = rpc:call(test_riak_node(), application, set_env, [riak_kv, storage_backend, riak_kv_memory_backend]),
%% Restart the vnodes so they come up with ETS
ok = supervisor:terminate_child({riak_kv_sup, test_riak_node()}, riak_kv_vnode_master),
ok = supervisor:terminate_child({riak_core_sup, test_riak_node()}, riak_core_vnode_sup),
{ok, _} = supervisor:restart_child({riak_core_sup, test_riak_node()}, riak_core_vnode_sup),
{ok, _} = supervisor:restart_child({riak_kv_sup, test_riak_node()}, riak_kv_vnode_master),
%% Clear the MapReduce cache
ok = rpc:call(test_riak_node(), riak_kv_mapred_cache, clear, []),
%% Now reset the ring so bucket properties are default
reset_ring().
%% Resets the ring to a fresh one, effectively deleting any bucket properties.
reset_ring() ->
Ring = rpc:call(test_riak_node(), riak_core_ring, fresh, []),
ok = rpc:call(test_riak_node(), riak_core_ring_manager, set_my_ring, [Ring]).
%% Finds the pid of the PB listener process
riak_pb_listener_pid() ->
{Children, Proc} = case compare_versions(riak_version(), [1,2,0]) of
less ->
{supervisor:which_children({riak_kv_sup, test_riak_node()}),
riak_kv_pb_listener};
_ ->
{supervisor:which_children({riak_api_sup, test_riak_node()}),
riak_api_pb_listener}
end,
hd([Pid || {_,Pid,_,[Mod]} <- Children, Mod == Proc]).
pause_riak_pb_listener() ->
Pid = riak_pb_listener_pid(),
rpc:call(test_riak_node(), sys, suspend, [Pid]).
resume_riak_pb_listener() ->
Pid = riak_pb_listener_pid(),
rpc:call(test_riak_node(), sys, resume, [Pid]).
kill_riak_pb_sockets() ->
Children = case compare_versions(riak_version(), [1,2,0]) of
less ->
supervisor:which_children({riak_kv_pb_socket_sup, test_riak_node()});
_ ->
supervisor:which_children({riak_api_pb_sup, test_riak_node()})
end,
case Children of
[] ->
ok;
[_|_] ->
Pids = [Pid || {_,Pid,_,_} <- Children],
[rpc:call(test_riak_node(), erlang, exit, [Pid, kill]) || Pid <- Pids],
erlang:yield(),
kill_riak_pb_sockets()
end.
maybe_start_network() ->
%% Try to spin up net_kernel
os:cmd("epmd -daemon"),
case net_kernel:start([test_eunit_node(), longnames]) of
{ok, _} ->
erlang:set_cookie(test_riak_node(), test_cookie()),
ok;
{error, {already_started, _}} ->
ok;
X ->
X
end.
bad_connect_test() ->
%% Start with an unlikely port number
?assertEqual({error, {tcp, econnrefused}}, start({127,0,0,1}, 65535)).
queue_disconnected_test() ->
%% Start with an unlikely port number
{ok, Pid} = start({127,0,0,1}, 65535, [queue_if_disconnected]),
?assertEqual({error, timeout}, ping(Pid, 10)),
?assertEqual({error, timeout}, list_keys(Pid, <<"b">>, 10)),
stop(Pid).
auto_reconnect_bad_connect_test() ->
%% Start with an unlikely port number
{ok, Pid} = start({127,0,0,1}, 65535, [auto_reconnect]),
?assertEqual({false, []}, is_connected(Pid)),
?assertEqual({error, disconnected}, ping(Pid)),
?assertEqual({error, disconnected}, list_keys(Pid, <<"b">>)),
stop(Pid).
server_closes_socket_test() ->
%% Silence SASL junk when socket closes.
error_logger:tty(false),
%% Set up a dummy socket to send requests on
{ok, Listen} = gen_tcp:listen(0, [binary, {packet, 4}, {active, false}]),
{ok, Port} = inet:port(Listen),
{ok, Pid} = start("127.0.0.1", Port),
{ok, Sock} = gen_tcp:accept(Listen),
?assertMatch(true, is_connected(Pid)),
%% Send a ping request in another process so the test doesn't block
Self = self(),
spawn(fun() -> Self ! ping(Pid, infinity) end),
%% Make sure request received then close the socket
{ok, _ReqMsg} = gen_tcp:recv(Sock, 0),
ok = gen_tcp:close(Sock),
ok = gen_tcp:close(Listen),
receive
Msg1 -> % result of ping from spawned process above
?assertEqual({error, disconnected}, Msg1)
end,
%% Wait for spawned process to exit
Mref = erlang:monitor(process, Pid),
receive
Msg2 ->
?assertMatch({'DOWN', Mref, process, _, _}, Msg2)
end.
auto_reconnect_server_closes_socket_test() ->
%% Set up a dummy socket to send requests on
{ok, Listen} = gen_tcp:listen(0, [binary, {packet, 4}, {active, false}]),
{ok, Port} = inet:port(Listen),
{ok, Pid} = start_link("127.0.0.1", Port, [auto_reconnect]),
{ok, Sock} = gen_tcp:accept(Listen),
?assertMatch(true, is_connected(Pid)),
%% Send a ping request in another process so the test doesn't block
Self = self(),
spawn(fun() -> Self ! ping(Pid, infinity) end),
%% Make sure request received then close the socket
{ok, _ReqMsg} = gen_tcp:recv(Sock, 0),
ok = gen_tcp:close(Sock),
ok = gen_tcp:close(Listen),
receive
Msg ->
?assertEqual({error, disconnected}, Msg)
end,
%% Server will not have had a chance to reconnect yet, reason counters empty.
?assertMatch({false, []}, is_connected(Pid)),
stop(Pid).
dead_socket_pid_returns_to_caller_test() ->
%% Set up a dummy socket to send requests on
{ok, Listen} = gen_tcp:listen(0, [binary, {packet, 4}, {active, false}]),
{ok, Port} = inet:port(Listen),
{ok, Pid} = start("127.0.0.1", Port),
{ok, Sock} = gen_tcp:accept(Listen),
?assertMatch(true, is_connected(Pid)),
%% Send a ping request in another process so the test doesn't block
Self = self(),
spawn(fun() -> Self ! (catch ping(Pid, infinity)) end),
%% Make sure request received then kill the process
{ok, _ReqMsg} = gen_tcp:recv(Sock, 0),
exit(Pid, kill),
receive
Msg ->
?assertMatch({'EXIT', {killed, _}}, Msg)
end,
%% Cleanup
ok = gen_tcp:close(Sock),
ok = gen_tcp:close(Listen).
pb_socket_test_() ->
{setup,
fun() ->
%% Grab the riakclient_pb.proto file
code:add_pathz("../ebin"),
ok = maybe_start_network()
end,
fun(_) ->
net_kernel:stop()
end,
{generator,
fun() ->
case catch net_adm:ping(test_riak_node()) of
pong ->
live_node_tests();
_ ->
[] %% {skipped, need_live_server};
end
end}}.
%% Check the reconnect interval increases up to the max and sticks there
increase_reconnect_interval_test() ->
increase_reconnect_interval_test(#state{}).
increase_reconnect_interval_test(State) ->
CurrInterval = State#state.reconnect_interval,
NextState = increase_reconnect_interval(State),
case NextState#state.reconnect_interval of
?MAX_RECONNECT_INTERVAL ->
FinalState = increase_reconnect_interval(NextState),
?assertEqual(?MAX_RECONNECT_INTERVAL, FinalState#state.reconnect_interval);
NextInterval->
?assert(NextInterval > CurrInterval),
increase_reconnect_interval_test(NextState)
end.
%% Retry `Fun' until it returns `Retry' times, waiting `Delay'
%% milliseconds between retries. This is our eventual consistency bread
%% and butter
wait_until(Fun) when is_function(Fun) ->
wait_until(Fun, 20, 500).
wait_until(_, 0, _) ->
fail;
wait_until(Fun, Retry, Delay) when Retry > 0 ->
Pass = Fun(),
case Pass of
true ->
ok;
_ ->
timer:sleep(Delay),
wait_until(Fun, Retry-1, Delay)
end.
%%
%% Tests to run against a live node - NB the node gets reconfigured and generally messed with
%%
live_node_tests() ->
[{"ping",
?_test( begin
{ok, Pid} = start_link(test_ip(), test_port()),
?assertEqual(pong, ?MODULE:ping(Pid)),
?assertEqual(true, is_connected(Pid)),
stop(Pid)
end)},
{"reconnect test",
?_test( begin
%% Make sure originally there
{ok, Pid} = start_link(test_ip(), test_port()),
%% Change the options to allow reconnection/queueing
set_options(Pid, [queue_if_disconnected]),
%% Kill the socket
kill_riak_pb_sockets(),
?assertEqual(pong, ?MODULE:ping(Pid)),
stop(Pid)
end)},
{"set client id",
?_test(
begin
{ok, Pid} = start_link(test_ip(), test_port()),
{ok, <>} = ?MODULE:get_client_id(Pid),
NewId = <<(OrigId+1):32>>,
ok = ?MODULE:set_client_id(Pid, NewId),
{ok, NewId} = ?MODULE:get_client_id(Pid)
end)},
{"version",
?_test(
begin
{ok, Pid} = start_link(test_ip(), test_port()),
{ok, ServerInfo} = ?MODULE:get_server_info(Pid),
[{node, _}, {server_version, _}] = lists:sort(ServerInfo)
end)},
{"get_should_read_put_test()",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
O0 = riakc_obj:new(<<"b">>, <<"k">>),
O = riakc_obj:update_value(O0, <<"v">>),
{ok, PO} = ?MODULE:put(Pid, O, [return_body]),
{ok, GO} = ?MODULE:get(Pid, <<"b">>, <<"k">>),
?assertEqual(riakc_obj:get_contents(PO), riakc_obj:get_contents(GO))
end)},
{"get should read put with timeout",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
O0 = riakc_obj:new(<<"b">>, <<"k">>),
O = riakc_obj:update_value(O0, <<"v">>),
{ok, PO} = ?MODULE:put(Pid, O, [{w, 1}, {dw, 1}, return_body]),
{ok, GO} = ?MODULE:get(Pid, <<"b">>, <<"k">>, 500),
?assertEqual(riakc_obj:get_contents(PO), riakc_obj:get_contents(GO))
end)},
{"get should read put with options",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
O0 = riakc_obj:new(<<"b">>, <<"k">>),
O = riakc_obj:update_value(O0, <<"v">>),
{ok, PO} = ?MODULE:put(Pid, O, [{w, 1}, {dw, 1}, return_body]),
{ok, GO} = ?MODULE:get(Pid, <<"b">>, <<"k">>, [{r, 1}]),
?assertEqual(riakc_obj:get_contents(PO), riakc_obj:get_contents(GO))
end)},
{"get should read put with non integer options",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
O0 = riakc_obj:new(<<"b">>, <<"k">>),
O = riakc_obj:update_value(O0, <<"v">>),
{ok, PO} = ?MODULE:put(Pid, O, [{w, all}, {dw, quorum}, return_body]),
{ok, GO} = ?MODULE:get(Pid, <<"b">>, <<"k">>, [{r, one}]),
?assertEqual(riakc_obj:get_contents(PO), riakc_obj:get_contents(GO))
end)},
{"put and delete with timeout",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
PO = riakc_obj:new(<<"b">>, <<"puttimeouttest">>, <<"value">>),
ok = ?MODULE:put(Pid, PO, 500),
{ok, GO} = ?MODULE:get(Pid, <<"b">>, <<"puttimeouttest">>, 500),
?assertEqual(<<"value">>, riakc_obj:get_value(GO)),
ok = ?MODULE:delete(Pid, <<"b">>, <<"puttimeouttest">>, 500),
{error, notfound} = ?MODULE:get(Pid, <<"b">>, <<"puttimeouttest">>)
end)},
{"update_should_change_value_test()",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
O0 = riakc_obj:new(<<"b">>, <<"k">>),
O = riakc_obj:update_value(O0, <<"v">>),
{ok, PO} = ?MODULE:put(Pid, O, [return_body]),
PO2 = riakc_obj:update_value(PO, <<"v2">>),
ok = ?MODULE:put(Pid, PO2),
{ok, GO} = ?MODULE:get(Pid, <<"b">>, <<"k">>),
?assertEqual(<<"v2">>, riakc_obj:get_value(GO))
end)},
{"key_should_be_missing_after_delete_test()",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
%% Put key/value
O0 = riakc_obj:new(<<"b">>, <<"k">>),
O = riakc_obj:update_value(O0, <<"v">>),
{ok, _PO} = ?MODULE:put(Pid, O, [return_body]),
%% Prove it really got stored
{ok, GO1} = ?MODULE:get(Pid, <<"b">>, <<"k">>),
?assertEqual(<<"v">>, riakc_obj:get_value(GO1)),
%% Delete and check no longer found
ok = ?MODULE:delete(Pid, <<"b">>, <<"k">>),
{error, notfound} = ?MODULE:get(Pid, <<"b">>, <<"k">>)
end)},
{"delete missing key test",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
%% Delete and check no longer found
ok = ?MODULE:delete(Pid, <<"notabucket">>, <<"k">>, [{rw, 1}]),
{error, notfound} = ?MODULE:get(Pid, <<"notabucket">>, <<"k">>)
end)},
{"empty_list_buckets_test()",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
?assertEqual({ok, []}, ?MODULE:list_buckets(Pid))
end)},
{"list_buckets_test()",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
Bs = lists:sort([list_to_binary(["b"] ++ integer_to_list(N)) || N <- lists:seq(1, 10)]),
F = fun(B) ->
O=riakc_obj:new(B, <<"key">>),
?MODULE:put(Pid, riakc_obj:update_value(O, <<"val">>))
end,
[F(B) || B <- Bs],
{ok, LBs} = ?MODULE:list_buckets(Pid),
?assertEqual(Bs, lists:sort(LBs))
end)},
{"list_keys_test()",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
Bucket = <<"listkeys">>,
Ks = lists:sort([list_to_binary(integer_to_list(N)) || N <- lists:seq(1, 10)]),
F = fun(K) ->
O=riakc_obj:new(Bucket, K),
?MODULE:put(Pid, riakc_obj:update_value(O, <<"val">>))
end,
[F(K) || K <- Ks],
{ok, LKs} = ?MODULE:list_keys(Pid, Bucket),
?assertEqual(Ks, lists:sort(LKs)),
%% Make sure it works with an infinite timeout (will reset the timeout
%% timer after each packet)
{ok, LKs2} = ?MODULE:list_keys(Pid, Bucket, infinity),
?assertEqual(Ks, lists:sort(LKs2))
end)},
{"get bucket properties test",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
{ok, Props} = get_bucket(Pid, <<"b">>),
?assertEqual(3, proplists:get_value(n_val, Props)),
?assertEqual(false, proplists:get_value(allow_mult, Props))
end)},
{"set bucket properties test",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
ok = set_bucket(Pid, <<"b">>, [{n_val, 2}, {allow_mult, false}]),
{ok, Props} = get_bucket(Pid, <<"b">>),
?assertEqual(2, proplists:get_value(n_val, Props)),
?assertEqual(false, proplists:get_value(allow_mult, Props))
end)},
{"allow_mult should allow dupes",
?_test(begin
reset_riak(),
{ok, Pid1} = start_link(test_ip(), test_port()),
{ok, Pid2} = start_link(test_ip(), test_port()),
ok = set_bucket(Pid1, <<"multibucket">>, [{allow_mult, true}]),
?MODULE:delete(Pid1, <<"multibucket">>, <<"foo">>),
{error, notfound} = ?MODULE:get(Pid1, <<"multibucket">>, <<"foo">>),
O = riakc_obj:new(<<"multibucket">>, <<"foo">>),
O1 = riakc_obj:update_value(O, <<"pid1">>),
O2 = riakc_obj:update_value(O, <<"pid2">>),
ok = ?MODULE:put(Pid1, O1),
ok = ?MODULE:put(Pid2, O2),
{ok, O3} = ?MODULE:get(Pid1, <<"multibucket">>, <<"foo">>),
?assertEqual([<<"pid1">>, <<"pid2">>], lists:sort(riakc_obj:get_values(O3))),
O4 = riakc_obj:update_value(riakc_obj:select_sibling(1, O3), <<"resolved">>),
ok = ?MODULE:put(Pid1, O4),
{ok, GO} = ?MODULE:get(Pid1, <<"multibucket">>, <<"foo">>),
?assertEqual([<<"resolved">>], lists:sort(riakc_obj:get_values(GO))),
?MODULE:delete(Pid1, <<"multibucket">>, <<"foo">>)
end)},
{"update object test",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
O0 = riakc_obj:new(<<"b">>, <<"k">>, <<"d">>),
io:format("O0: ~p\n", [O0]),
{ok, O1} = riakc_pb_socket:put(Pid, O0, [return_body]),
io:format("O1: ~p\n", [O1]),
M1 = riakc_obj:get_metadata(O1),
M2 = dict:store(?MD_LINKS, [{{<<"b">>, <<"k1">>}, <<"t1">>}], M1),
O2 = riakc_obj:update_metadata(O1, M2),
riakc_pb_socket:put(Pid, O2)
end)},
{"queue test",
?_test(begin
%% Would really like this in a nested {setup, blah} structure
%% but eunit does not allow
{ok, Pid} = start_link(test_ip(), test_port()),
pause_riak_pb_listener(),
Me = self(),
%% this request will block as
spawn(fun() -> Me ! {1, ping(Pid)} end),
%% this request should be queued as socket will not be created
spawn(fun() -> Me ! {2, ping(Pid)} end),
resume_riak_pb_listener(),
receive {1,Ping1} -> ?assertEqual(Ping1, pong) end,
receive {2,Ping2} -> ?assertEqual(Ping2, pong) end
end)},
{"timeout queue test",
?_test(begin
%% Would really like this in a nested {setup, blah} structure
%% but eunit does not allow
pause_riak_pb_listener(),
{ok, Pid} = start_link(test_ip(), test_port(), [queue_if_disconnected]),
Me = self(),
%% this request will block as
spawn(fun() -> Me ! {1, ping(Pid, 0)} end),
%% this request should be queued as socket will not be created
spawn(fun() -> Me ! {2, ping(Pid, 0)}, Me ! running end),
receive running -> ok end,
resume_riak_pb_listener(),
receive {1,Ping1} -> ?assertEqual({error, timeout}, Ping1) end,
receive {2,Ping2} -> ?assertEqual({error, timeout}, Ping2) end
end)},
{"ignore stale tref test",
?_test(begin
%% Would really like this in a nested {setup, blah} structure
%% but eunit does not allow
{ok, Pid} = start_link(test_ip(), test_port()),
Pid ! {req_timeout, make_ref()},
?assertEqual(pong, ping(Pid))
end)},
{"infinite timeout ping test",
?_test(begin
%% Would really like this in a nested {setup, blah} structure
%% but eunit does not allow
{ok, Pid} = start_link(test_ip(), test_port()),
?assertEqual(pong, ping(Pid, infinity)),
?assertEqual(pong, ping(Pid, undefined))
end)},
{"javascript_source_map_test()",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
B = <<"bucket">>,
K = <<"foo">>,
O=riakc_obj:new(B, K),
?MODULE:put(Pid, riakc_obj:update_value(O, <<"2">>, "application/json")),
?assertEqual({ok, [{0, [2]}]},
?MODULE:mapred(Pid,
[{B, K}],
[{map, {jsanon, <<"function (v) { return [JSON.parse(v.values[0].data)]; }">>},
undefined, true}]))
end)},
{"javascript_named_map_test()",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
B = <<"bucket">>,
K = <<"foo">>,
O=riakc_obj:new(B, K),
?MODULE:put(Pid, riakc_obj:update_value(O, <<"99">>, "application/json")),
?assertEqual({ok, [{0, [99]}]},
?MODULE:mapred(Pid,
[{B, K}],
[{map, {jsfun, <<"Riak.mapValuesJson">>},
undefined, true}]))
end)},
{"javascript_source_map_reduce_test()",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
Store = fun({K,V}) ->
O=riakc_obj:new(<<"bucket">>, K),
?MODULE:put(Pid,riakc_obj:update_value(O, V, "application/json"))
end,
[Store(KV) || KV <- [{<<"foo">>, <<"2">>},
{<<"bar">>, <<"3">>},
{<<"baz">>, <<"4">>}]],
?assertEqual({ok, [{1, [3]}]},
?MODULE:mapred(Pid,
[{<<"bucket">>, <<"foo">>},
{<<"bucket">>, <<"bar">>},
{<<"bucket">>, <<"baz">>}],
[{map, {jsanon, <<"function (v) { return [1]; }">>},
undefined, false},
{reduce, {jsanon,
<<"function(v) {
total = v.reduce(
function(prev,curr,idx,array) {
return prev+curr;
}, 0);
return [total];
}">>},
undefined, true}]))
end)},
{"javascript_named_map_reduce_test()",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
Store = fun({K,V}) ->
O=riakc_obj:new(<<"bucket">>, K),
?MODULE:put(Pid,riakc_obj:update_value(O, V, "application/json"))
end,
[Store(KV) || KV <- [{<<"foo">>, <<"2">>},
{<<"bar">>, <<"3">>},
{<<"baz">>, <<"4">>}]],
?assertEqual({ok, [{1, [9]}]},
?MODULE:mapred(Pid,
[{<<"bucket">>, <<"foo">>},
{<<"bucket">>, <<"bar">>},
{<<"bucket">>, <<"baz">>}],
[{map, {jsfun, <<"Riak.mapValuesJson">>}, undefined, false},
{reduce, {jsfun, <<"Riak.reduceSum">>}, undefined, true}]))
end)},
{"javascript_bucket_map_reduce_test()",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
Store = fun({K,V}) ->
O=riakc_obj:new(<<"bucket">>, K),
?MODULE:put(Pid,riakc_obj:update_value(O, V, "application/json"))
end,
[Store(KV) || KV <- [{<<"foo">>, <<"2">>},
{<<"bar">>, <<"3">>},
{<<"baz">>, <<"4">>}]],
?assertEqual({ok, [{1, [9]}]},
?MODULE:mapred_bucket(Pid, <<"bucket">>,
[{map, {jsfun, <<"Riak.mapValuesJson">>}, undefined, false},
{reduce, {jsfun, <<"Riak.reduceSum">>}, undefined, true}]))
end)},
{"javascript_arg_map_reduce_test()",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
O=riakc_obj:new(<<"bucket">>, <<"foo">>),
?MODULE:put(Pid, riakc_obj:update_value(O, <<"2">>, "application/json")),
?assertEqual({ok, [{1, [10]}]},
?MODULE:mapred(Pid,
[{{<<"bucket">>, <<"foo">>}, 5},
{{<<"bucket">>, <<"foo">>}, 10},
{{<<"bucket">>, <<"foo">>}, 15},
{{<<"bucket">>, <<"foo">>}, -15},
{{<<"bucket">>, <<"foo">>}, -5}],
[{map, {jsanon, <<"function(v, arg) { return [arg]; }">>},
undefined, false},
{reduce, {jsfun, <<"Riak.reduceSum">>}, undefined, true}]))
end)},
{"erlang_map_reduce_test()",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
Store = fun({K,V}) ->
O=riakc_obj:new(<<"bucket">>, K),
?MODULE:put(Pid,riakc_obj:update_value(O, V, "application/json"))
end,
[Store(KV) || KV <- [{<<"foo">>, <<"2">>},
{<<"bar">>, <<"3">>},
{<<"baz">>, <<"4">>}]],
{ok, [{1, Results}]} = ?MODULE:mapred(Pid,
[{<<"bucket">>, <<"foo">>},
{<<"bucket">>, <<"bar">>},
{<<"bucket">>, <<"baz">>}],
[{map, {modfun, riak_kv_mapreduce,
map_object_value},
undefined, false},
{reduce, {modfun, riak_kv_mapreduce,
reduce_set_union},
undefined, true}]),
?assertEqual([<<"2">>, <<"3">>, <<"4">>], lists:sort(Results))
end)},
{"erlang_map_reduce_binary_2i_test()",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
Store = fun({K,V,I}) ->
O=riakc_obj:new(<<"bucket">>, K),
MD=riakc_obj:add_secondary_index(dict:new(), I),
O2=riakc_obj:update_metadata(O,MD),
?MODULE:put(Pid,riakc_obj:update_value(O2, V, "application/json"))
end,
[Store(KV) || KV <- [{<<"foo">>, <<"2">>, {{binary_index, "idx"}, [<<"a">>]}},
{<<"bar">>, <<"3">>, {{binary_index, "idx"}, [<<"b">>]}},
{<<"baz">>, <<"4">>, {{binary_index, "idx"}, [<<"a">>]}}]],
{ok, [{1, Results}]} = ?MODULE:mapred(Pid,
{index,<<"bucket">>,{binary_index, "idx"}, <<"a">>},
[{map, {modfun, riak_kv_mapreduce,
map_object_value},
undefined, false},
{reduce, {modfun, riak_kv_mapreduce,
reduce_set_union},
undefined, true}]),
?assertEqual([<<"2">>, <<"4">>], lists:sort(Results))
end)},
{"erlang_map_reduce_integer_2i_test()",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
Store = fun({K,V,I}) ->
O=riakc_obj:new(<<"bucket">>, K),
MD=riakc_obj:add_secondary_index(dict:new(), I),
O2=riakc_obj:update_metadata(O,MD),
?MODULE:put(Pid,riakc_obj:update_value(O2, V, "application/json"))
end,
[Store(KV) || KV <- [{<<"foo">>, <<"2">>, {{integer_index, "idx"}, [4]}},
{<<"bar">>, <<"3">>, {{integer_index, "idx"}, [7]}},
{<<"baz">>, <<"4">>, {{integer_index, "idx"}, [4]}}]],
{ok, [{1, Results}]} = ?MODULE:mapred(Pid,
{index,<<"bucket">>,{integer_index, "idx"},3,5},
[{map, {modfun, riak_kv_mapreduce,
map_object_value},
undefined, false},
{reduce, {modfun, riak_kv_mapreduce,
reduce_set_union},
undefined, true}]),
?assertEqual([<<"2">>, <<"4">>], lists:sort(Results))
end)},
{"missing_key_erlang_map_reduce_test()",
?_test(begin
reset_riak(),
{ok, Pid} = {ok, Pid} = start_link(test_ip(), test_port()),
{ok, Results} = ?MODULE:mapred(Pid, [{<<"bucket">>, <<"foo">>},
{<<"bucket">>, <<"bar">>},
{<<"bucket">>, <<"baz">>}],
[{map, {modfun, riak_kv_mapreduce,
map_object_value},
<<"include_notfound">>, false},
{reduce, {modfun, riak_kv_mapreduce,
reduce_set_union},
undefined, true}]),
[{1, [{error, notfound}|_]}] = Results end)},
{"missing_key_javascript_map_reduce_test()",
?_test(begin
reset_riak(),
{ok, Pid} = {ok, Pid} = start_link(test_ip(), test_port()),
{ok, Results} = ?MODULE:mapred(Pid, [{<<"bucket">>, <<"foo">>},
{<<"bucket">>, <<"bar">>},
{<<"bucket">>, <<"baz">>}],
[{map, {jsfun, <<"Riak.mapValuesJson">>},
undefined, false},
{reduce, {jsfun, <<"Riak.reduceSort">>},
undefined, true}]),
[{1, [{not_found, {_, _},<<"undefined">>}|_]}] = Results end)},
{"map reduce bad inputs",
?_test(begin
{ok, Pid} = start_link(test_ip(), test_port()),
Res = ?MODULE:mapred(Pid, undefined,
[{map, {jsfun, <<"Riak.mapValuesJson">>},
undefined, false},
{reduce, {jsfun, <<"Riak.reduceSum">>},
undefined, true}]),
?assertEqual({error, <<"{inputs,{\"Inputs must be a binary bucket, a tuple of bucket and key-filters, a list of target tuples, or a search, index, or modfun tuple:\",\n undefined}}">>},
Res )
end)},
{"map reduce bad input keys",
?_test(begin
{ok, Pid} = start_link(test_ip(), test_port()),
Res = ?MODULE:mapred(Pid, [<<"b">>], % no {B,K} tuple
[{map, {jsfun, <<"Riak.mapValuesJson">>},
undefined, false},
{reduce, {jsfun, <<"Riak.reduceSum">>},
undefined, true}]),
?assertEqual({error,<<"{inputs,{\"Inputs target tuples must be {B,K} or {{B,K},KeyData}:\",[<<\"b\">>]}}">>},
Res)
end)},
{"map reduce bad query",
?_test(begin
{ok, Pid} = start_link(test_ip(), test_port()),
Res = ?MODULE:mapred(Pid, [{<<"b">>,<<"k">>}], % no {B,K} tuple
undefined),
?assertEqual({error,<<"{query,{\"Query takes a list of step tuples\",undefined}}">>},
Res)
end)},
{"get should convert erlang terms",
?_test(begin
reset_riak(),
TestNode = test_riak_node(),
MyBin = <<"some binary">>,
MyTerm = [<<"b">>,<<"a_term">>,{some_term, ['full', "of", 123, 654.321]}],
BinObj = rpc:call(TestNode, riak_object, new,
[<<"b">>, <<"a_bin">>, MyBin]),
TermObj = rpc:call(TestNode, riak_object, new,
[<<"b">>, <<"a_term">>, MyTerm]),
{ok, C} = rpc:call(TestNode, riak, local_client, []),
%% parameterized module trickery - stick it as the last argument
ok = rpc:call(TestNode, riak_client, put, [BinObj, 1, C]),
ok = rpc:call(TestNode, riak_client, put, [TermObj, 1, C]),
{ok, Pid} = start_link(test_ip(), test_port()),
{ok, GotBinObj} = ?MODULE:get(Pid, <<"b">>, <<"a_bin">>),
{ok, GotTermObj} = ?MODULE:get(Pid, <<"b">>, <<"a_term">>),
?assertEqual(riakc_obj:get_value(GotBinObj), MyBin),
?assertEqual(riakc_obj:get_content_type(GotTermObj),
"application/x-erlang-binary"),
?assertEqual(binary_to_term(riakc_obj:get_value(GotTermObj)), MyTerm)
end)},
{"putting without a key should generate one",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
PO = riakc_obj:new(<<"b">>, undefined, <<"value">>),
Res1 = ?MODULE:put(Pid, PO),
Res2 = ?MODULE:put(Pid, PO),
?assertMatch({ok, _Key}, Res1),
% Make sure the same key isn't generated twice
?assert(Res1 =/= Res2)
end)},
{"putting without a key should generate one with return_body",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
PO = riakc_obj:new(<<"b">>, undefined, <<"value">>),
{ok, Obj1} = ?MODULE:put(Pid, PO, [return_body]),
{ok, Obj2} = ?MODULE:put(Pid, PO, [return_body]),
%% Make sure the same key isn't generated twice
?assertEqual(element(1, Obj1), riakc_obj),
?assertEqual(element(1, Obj2), riakc_obj),
?assert(riakc_obj:key(Obj1) /= riakc_obj:key(Obj2))
end)},
{"conditional gets should return unchanged if the vclock matches",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
PO = riakc_obj:new(<<"b">>, <<"key">>, <<"value">>),
?MODULE:put(Pid, PO),
{ok, Obj} = ?MODULE:get(Pid, <<"b">>, <<"key">>),
VClock = riakc_obj:vclock(Obj),
%% object hasn't changed
?assertEqual(unchanged, ?MODULE:get(Pid, <<"b">>, <<"key">>,
[{if_modified, VClock}])),
%% change the object and make sure unchanged isn't returned
P1 = riakc_obj:update_value(Obj, <<"newvalue">>),
?MODULE:put(Pid, P1),
?assertMatch({ok, _}, ?MODULE:get(Pid, <<"b">>, <<"key">>,
[{if_modified, VClock}]))
end)},
{"the head get option should return the object metadata without the value",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
PO = riakc_obj:new(<<"b">>, <<"key">>, <<"value">>),
?MODULE:put(Pid, PO),
{ok, Obj} = ?MODULE:get(Pid, <<"b">>, <<"key">>, [head]),
?assertEqual(<<>>, riakc_obj:get_value(Obj)),
{ok, Obj2} = ?MODULE:get(Pid, <<"b">>, <<"key">>, []),
?assertEqual(<<"value">>, riakc_obj:get_value(Obj2))
end)},
{"conditional put should allow you to avoid overwriting a value if it already exists",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
PO = riakc_obj:new(<<"b">>, <<"key">>, <<"value">>),
?assertEqual(ok, ?MODULE:put(Pid, PO, [if_none_match])),
?assertEqual({error, <<"match_found">>}, ?MODULE:put(Pid, PO, [if_none_match]))
end)},
{"conditional put should allow you to avoid overwriting a value if its been updated",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
PO = riakc_obj:new(<<"b">>, <<"key">>, <<"value">>),
{ok, Obj} = ?MODULE:put(Pid, PO, [return_body]),
Obj2 = riakc_obj:update_value(Obj, <<"newvalue">>),
?assertEqual(ok, ?MODULE:put(Pid, Obj2, [if_not_modified])),
?assertEqual({error, <<"modified">>}, ?MODULE:put(Pid, Obj2, [if_not_modified]))
end)},
{"if_not_modified should fail if the object is not found",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
PO = riakc_obj:new(<<"b">>, <<"key">>, <<"value">>),
?assertEqual({error, <<"notfound">>}, ?MODULE:put(Pid, PO, [if_not_modified]))
end)},
{"return_head should empty out the value in the riak object",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
PO = riakc_obj:new(<<"b">>, <<"key">>, <<"value">>),
{ok, Obj} = ?MODULE:put(Pid, PO, [return_head]),
?assertEqual(<<>>, riakc_obj:get_value(Obj))
end)},
{"return_head should empty out all values when there's siblings",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
ok = set_bucket(Pid, <<"b">>, [{allow_mult, true}]),
PO = riakc_obj:new(<<"b">>, <<"key">>, <<"value">>),
{ok, Obj} = ?MODULE:put(Pid, PO, [return_head]),
?assertEqual(<<>>, riakc_obj:get_value(Obj)),
{ok, Obj2} = ?MODULE:put(Pid, PO, [return_head]),
?assertEqual([<<>>, <<>>], riakc_obj:get_values(Obj2))
end)},
{"user metadata manipulation",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
O0 = riakc_obj:new(<<"b">>, <<"key0">>, <<"value0">>),
MD0 = riakc_obj:get_update_metadata(O0),
MD1 = riakc_obj:set_user_metadata_entry(MD0, {<<"Key1">>,<<"Val1">>}),
O1 = riakc_obj:update_metadata(O0, MD1),
?assertEqual(ok, ?MODULE:put(Pid, O1)),
{ok, O2} = ?MODULE:get(Pid, <<"b">>, <<"key0">>),
MD2 = riakc_obj:get_update_metadata(O2),
?assertEqual([{<<"Key1">>,<<"Val1">>}], riakc_obj:get_user_metadata_entries(MD2)),
MD3 = riakc_obj:set_user_metadata_entry(MD2, {<<"Key2">>,<<"Val2">>}),
O3 = riakc_obj:update_metadata(O2, MD3),
?assertEqual(ok, ?MODULE:put(Pid, O3)),
{ok, O4} = ?MODULE:get(Pid, <<"b">>, <<"key0">>),
?assertEqual(2, length(riakc_obj:get_user_metadata_entries(riakc_obj:get_update_metadata(O4))))
end)},
{"binary secondary index manipulation",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
O0 = riakc_obj:new(<<"b">>, <<"key1">>, <<"value1">>),
MD0 = riakc_obj:get_update_metadata(O0),
MD1 = riakc_obj:set_secondary_index(MD0, [{{binary_index, "idx"},[<<"aaa">>]}]),
O1 = riakc_obj:update_metadata(O0, MD1),
?assertEqual(ok, ?MODULE:put(Pid, O1)),
{ok, O2} = ?MODULE:get(Pid, <<"b">>, <<"key1">>),
MD2 = riakc_obj:get_update_metadata(O2),
?assertEqual([<<"aaa">>], lists:sort(riakc_obj:get_secondary_index(MD2,{binary_index,"idx"}))),
MD3 = riakc_obj:add_secondary_index(MD2, [{{binary_index, "idx"},[<<"bbb">>,<<"aaa">>,<<"ccc">>]}]),
O3 = riakc_obj:update_metadata(O2, MD3),
?assertEqual(ok, ?MODULE:put(Pid, O3)),
?assertEqual({ok,?INDEX_RESULTS{keys=[<<"key1">>]}},
?MODULE:get_index(Pid, <<"b">>, {binary_index, "idx"}, <<"bbb">>)),
{ok, O4} = ?MODULE:get(Pid, <<"b">>, <<"key1">>),
MD4 = riakc_obj:get_update_metadata(O4),
?assertEqual([<<"aaa">>,<<"bbb">>,<<"ccc">>], lists:sort(riakc_obj:get_secondary_index(MD4, {binary_index, "idx"}))),
MD5 = riakc_obj:delete_secondary_index(MD4,{binary_index,"idx"}),
O5 = riakc_obj:update_metadata(O4, MD5),
?assertEqual(ok, ?MODULE:put(Pid, O5))
end)},
{"integer secondary index manipulation",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
O0 = riakc_obj:new(<<"b">>, <<"key2">>, <<"value2">>),
MD0 = riakc_obj:get_update_metadata(O0),
MD1 = riakc_obj:set_secondary_index(MD0, [{{integer_index, "idx"},[67]}]),
O1 = riakc_obj:update_metadata(O0, MD1),
?assertEqual(ok, ?MODULE:put(Pid, O1)),
{ok, O2} = ?MODULE:get(Pid, <<"b">>, <<"key2">>),
MD2 = riakc_obj:get_update_metadata(O2),
?assertEqual([67], lists:sort(riakc_obj:get_secondary_index(MD2,{integer_index,"idx"}))),
MD3 = riakc_obj:add_secondary_index(MD2, [{{integer_index, "idx"},[56,10000,100]}]),
O3 = riakc_obj:update_metadata(O2, MD3),
?assertEqual(ok, ?MODULE:put(Pid, O3)),
?assertEqual({ok,?INDEX_RESULTS{keys=[<<"key2">>]}},
?MODULE:get_index(Pid, <<"b">>, {integer_index, "idx"}, 50, 60)),
{ok, O4} = ?MODULE:get(Pid, <<"b">>, <<"key2">>),
MD4 = riakc_obj:get_update_metadata(O4),
?assertEqual([56,67,100,10000], lists:sort(riakc_obj:get_secondary_index(MD4, {integer_index, "idx"}))),
MD5 = riakc_obj:delete_secondary_index(MD4,{integer_index,"idx"}),
O5 = riakc_obj:update_metadata(O4, MD5),
?assertEqual(ok, ?MODULE:put(Pid, O5))
end)},
{"counter increment / decrement / get value",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
unlink(Pid),
Bucket = <<"counter_test_bucket">>,
Key = <<"test_counter">>,
%% counters require allow_mult to be true
ok = set_bucket(Pid, Bucket, [{allow_mult, true}]),
ok = ?MODULE:counter_incr(Pid, Bucket, Key, 10),
?assertEqual({ok, 10}, ?MODULE:counter_val(Pid, Bucket, Key)),
ok = ?MODULE:counter_incr(Pid, Bucket, Key, -5, [{w, quorum}, {pw, one}, {dw, all}]),
?assertEqual({ok, 5}, ?MODULE:counter_val(Pid, Bucket, Key, [{pr, one}]))
end)},
{"create a search index / get / list / delete",
{timeout, 30, ?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
reset_solr(Pid),
Index = <<"indextest">>,
SchemaName = <<"_yz_default">>,
?assertEqual(ok,
?MODULE:create_search_index(Pid,
Index,
SchemaName,
[{n_val,2}])),
wait_until( fun() ->
case ?MODULE:get_search_index(Pid, Index) of
{ok, IndexData} ->
proplists:get_value(index, IndexData) == Index andalso
proplists:get_value(schema, IndexData) == SchemaName andalso
proplists:get_value(n_val, IndexData) == 2;
{error, <<"notfound">>} ->
false
end
end, 20, 1000 ),
?assertEqual({ok, [[{index,Index},
{schema,SchemaName},
{n_val,2}]]},
?MODULE:list_search_indexes(Pid)),
?assertEqual(ok, ?MODULE:delete_search_index(Pid, Index))
end)}},
{"create a search schema / get",
{timeout, 30, ?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
reset_solr(Pid),
Schema = <<"
_yz_id
">>,
Index = <<"schemaindex">>,
SchemaName = <<"myschema">>,
?assertEqual(ok, ?MODULE:create_search_schema(Pid, SchemaName, Schema)),
?assertEqual(ok, ?MODULE:create_search_index(Pid, Index, SchemaName, [])),
wait_until( fun() ->
case ?MODULE:list_search_indexes(Pid) of
{ok, []} ->
false;
{ok, [IndexData|_]} ->
proplists:get_value(index, IndexData) == Index andalso
proplists:get_value(schema, IndexData) == SchemaName andalso
proplists:get_value(n_val, IndexData) == 3
end
end, 20, 1000 ),
wait_until( fun() ->
case ?MODULE:get_search_schema(Pid, SchemaName) of
{ok, SchemaData} ->
proplists:get_value(name, SchemaData) == SchemaName andalso
proplists:get_value(content, SchemaData) == Schema;
{error, <<"notefound">>} ->
false
end
end, 20, 1000 )
end)}},
{"create a search index and tie to a bucket",
{timeout, 30, ?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
Index = <<"myindex">>,
Bucket = <<"mybucket">>,
?assertEqual(ok, ?MODULE:create_search_index(Pid, Index)),
wait_until( fun() ->
case ?MODULE:get_search_index(Pid, Index) of
{ok, IndexData} ->
proplists:get_value(index, IndexData) == Index;
{error, <<"notfound">>} ->
false
end
end, 20, 1000 ),
ok = ?MODULE:set_search_index(Pid, Bucket, Index),
PO = riakc_obj:new(Bucket, <<"fred">>, <<"{\"name_s\":\"Freddy\"}">>, "application/json"),
{ok, _Obj} = ?MODULE:put(Pid, PO, [return_head]),
wait_until( fun() ->
{ok, Result} = search(Pid, Index, <<"*:*">>),
1 == Result#search_results.num_found
end, 20, 1000 )
end)}},
{"search utf8",
{timeout, 30, ?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
reset_solr(Pid),
Index = <<"myindex">>,
Bucket = <<"mybucket">>,
?assertEqual(ok, ?MODULE:create_search_index(Pid, Index)),
wait_until( fun() ->
case ?MODULE:get_search_index(Pid, Index) of
{ok, IndexData} ->
proplists:get_value(index, IndexData) == Index;
{error, <<"notfound">>} ->
false
end
end, 20, 1000 ),
ok = ?MODULE:set_search_index(Pid, Bucket, Index),
PO = riakc_obj:new(Bucket, <<"fred">>, <<"{\"name_s\":\"בָּרָא\"}">>, "application/json"),
{ok, _Obj} = ?MODULE:put(Pid, PO, [return_head]),
wait_until( fun() ->
{ok, Result} = search(Pid, Index, <<"name_s:בָּרָא">>),
1 == Result#search_results.num_found
end )
end)}},
{"trivial set delete",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
ok = update_type(Pid,
{<<"set_bucket">>, <<"bucket">>}, <<"key">>,
riakc_set:to_op(riakc_set:add_element(<<"X">>, riakc_set:new()))),
{ok, S0} = fetch_type(Pid, {<<"set_bucket">>, <<"bucket">>}, <<"key">>),
?assert(riakc_set:is_element(<<"X">>, S0)),
?assertEqual(riakc_set:size(S0), 1),
ok = update_type(Pid,
{<<"set_bucket">>, <<"bucket">>}, <<"key">>,
riakc_set:to_op(riakc_set:del_element(<<"X">>, S0))),
{ok, S1} = fetch_type(Pid, {<<"set_bucket">>, <<"bucket">>}, <<"key">>),
?assertNot(riakc_set:is_element(<<"X">>, S1)),
?assertEqual(riakc_set:size(S1), 0)
end)},
{"add and remove items in nested set in map",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
ok = riakc_pb_socket:update_type(Pid,
{<<"map_bucket">>, <<"bucket">>}, <<"key">>,
riakc_map:to_op(riakc_map:update({<<"set">>, set},
fun(S) ->
riakc_set:add_element(<<"X">>,
riakc_set:add_element(<<"Y">>, S))
end, riakc_map:new()))),
{ok, M0} = riakc_pb_socket:fetch_type(Pid, {<<"map_bucket">>, <<"bucket">>}, <<"key">>),
L0 = riakc_map:fetch({<<"set">>, set}, M0),
?assert(lists:member(<<"X">>, L0)),
?assert(lists:member(<<"Y">>, L0)),
?assertEqual(length(L0), 2),
M1 = riakc_map:update({<<"set">>, set},
fun(S) -> riakc_set:del_element(<<"X">>,
riakc_set:add_element(<<"Z">>, S)) end,
M0),
ok = riakc_pb_socket:update_type(Pid,
{<<"map_bucket">>, <<"bucket">>}, <<"key">>,
riakc_map:to_op(M1)),
{ok, M2} = riakc_pb_socket:fetch_type(Pid, {<<"map_bucket">>, <<"bucket">>}, <<"key">>),
L1 = riakc_map:fetch({<<"set">>, set}, M2),
?assert(lists:member(<<"Y">>, L1)),
?assert(lists:member(<<"Z">>, L1)),
?assertEqual(length(L1), 2)
end)},
{"increment nested counter",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
ok = riakc_pb_socket:update_type(Pid,
{<<"map_bucket">>, <<"bucket">>}, <<"key">>,
riakc_map:to_op(riakc_map:update({<<"counter">>, counter},
fun(C) ->
riakc_counter:increment(5, C)
end, riakc_map:new()))),
{ok, M0} = fetch_type(Pid, {<<"map_bucket">>, <<"bucket">>}, <<"key">>),
C0 = riakc_map:fetch({<<"counter">>, counter}, M0),
?assertEqual(C0, 5),
M1 = riakc_map:update({<<"counter">>, counter},
fun(C) -> riakc_counter:increment(200, C) end,
M0),
M2 = riakc_map:update({<<"counter">>, counter},
fun(C) -> riakc_counter:decrement(117, C) end,
M1),
M3 = riakc_map:update({<<"counter">>, counter},
fun(C) -> riakc_counter:increment(256, C) end,
M2),
ok = riakc_pb_socket:update_type(Pid,
{<<"map_bucket">>, <<"bucket">>}, <<"key">>,
riakc_map:to_op(M3)),
{ok, M4} = fetch_type(Pid, {<<"map_bucket">>, <<"bucket">>}, <<"key">>),
C1 = riakc_map:fetch({<<"counter">>, counter}, M4),
?assertEqual(C1, 344)
end)},
{"updated nested lww register",
?_test(begin
reset_riak(),
%% The word "stone" translated into Russian and Thai
StoneInRussian = [1051,1102,1082,32,1082,1072,1084,1085,1077,1091,1083,1086,
1074,1080,1090,1077,1083,1103],
StoneInThai = [3627,3636,3609],
{ok, Pid} = start_link(test_ip(), test_port()),
ok = riakc_pb_socket:update_type(Pid,
{<<"map_bucket">>, <<"bucket">>},
<<"key">>,
riakc_map:to_op(
riakc_map:update(
{<<"register">>, register},
fun(R) ->
riakc_register:set(
term_to_binary({"barney", "rubble", StoneInRussian}),
R)
end, riakc_map:new()))),
{ok, M0} = fetch_type(Pid, {<<"map_bucket">>, <<"bucket">>}, <<"key">>),
R0 = riakc_map:fetch({<<"register">>, register}, M0),
?assertEqual(binary_to_term(R0), {"barney", "rubble", StoneInRussian}),
ok = riakc_pb_socket:update_type(Pid,
{<<"map_bucket">>, <<"bucket">>},
<<"key">>,
riakc_map:to_op(
riakc_map:update(
{<<"register">>, register},
fun(R) ->
riakc_register:set(
term_to_binary({"barney", "rubble", StoneInThai}),
R)
end, M0))),
{ok, M1} = fetch_type(Pid, {<<"map_bucket">>, <<"bucket">>}, <<"key">>),
R1 = riakc_map:fetch({<<"register">>, register}, M1),
?assertEqual(binary_to_term(R1), {"barney", "rubble", StoneInThai})
end)},
{"throw exception for undefined context for delete",
?_test(begin
reset_riak(),
?assertThrow(context_required, riakc_set:del_element(<<"X">>,
riakc_set:add_element(<<"X">>,
riakc_set:new()))),
?assertThrow(context_required, riakc_map:erase({<<"counter">>, counter}, riakc_map:new())),
?assertThrow(context_required, riakc_map:erase({<<"set">>, set}, riakc_map:new())),
?assertThrow(context_required, riakc_map:erase({<<"map">>, map}, riakc_map:new())),
?assertThrow(context_required, riakc_map:update({<<"set">>, set}, fun(S) -> riakc_set:del_element(<<"Y">>, S) end, riakc_map:new())),
?assertThrow(context_required, riakc_flag:disable(riakc_flag:new()))
end)},
{"delete bogus item from set",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
ok = update_type(Pid,
{<<"set_bucket">>, <<"bucket">>}, <<"key">>,
riakc_set:to_op(riakc_set:add_element(<<"X">>, riakc_set:new()))),
{ok, S0} = fetch_type(Pid, {<<"set_bucket">>, <<"bucket">>}, <<"key">>),
?assert(riakc_set:is_element(<<"X">>, S0)),
?assertEqual(riakc_set:size(S0), 1),
ok = update_type(Pid,
{<<"set_bucket">>, <<"bucket">>}, <<"key">>,
riakc_set:to_op(riakc_set:del_element(<<"Y">>, S0))),
{ok, S1} = fetch_type(Pid, {<<"set_bucket">>, <<"bucket">>}, <<"key">>),
?assert(riakc_set:is_element(<<"X">>, S1)),
?assertEqual(riakc_set:size(S1), 1)
end)},
{"add redundant item to set",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
ok = update_type(Pid,
{<<"set_bucket">>, <<"bucket">>}, <<"key">>,
riakc_set:to_op(riakc_set:add_element(<<"X">>, riakc_set:new()))),
{ok, S0} = fetch_type(Pid, {<<"set_bucket">>, <<"bucket">>}, <<"key">>),
?assert(riakc_set:is_element(<<"X">>, S0)),
?assertEqual(riakc_set:size(S0), 1),
ok = update_type(Pid,
{<<"set_bucket">>, <<"bucket">>}, <<"key">>,
riakc_set:to_op(riakc_set:add_element(<<"X">>, S0))),
{ok, S1} = fetch_type(Pid, {<<"set_bucket">>, <<"bucket">>}, <<"key">>),
?assert(riakc_set:is_element(<<"X">>, S1)),
?assertEqual(riakc_set:size(S1), 1)
end)},
{"add and remove redundant item to/from set",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
ok = update_type(Pid,
{<<"set_bucket">>, <<"bucket">>}, <<"key">>,
riakc_set:to_op(riakc_set:add_element(<<"X">>,
riakc_set:add_element(<<"Y">>, riakc_set:new())))),
{ok, S0} = fetch_type(Pid, {<<"set_bucket">>, <<"bucket">>}, <<"key">>),
?assert(riakc_set:is_element(<<"X">>, S0)),
?assert(riakc_set:is_element(<<"Y">>, S0)),
?assertEqual(riakc_set:size(S0), 2),
ok = update_type(Pid,
{<<"set_bucket">>, <<"bucket">>}, <<"key">>,
riakc_set:to_op(riakc_set:del_element(<<"X">>, riakc_set:add_element(<<"X">>, S0)))),
{ok, S1} = fetch_type(Pid, {<<"set_bucket">>, <<"bucket">>}, <<"key">>),
?assert(riakc_set:is_element(<<"X">>, S1)),
?assert(riakc_set:is_element(<<"Y">>, S1)),
?assertEqual(riakc_set:size(S1), 2)
end)},
{"remove then add redundant item from/to set",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
ok = update_type(Pid,
{<<"set_bucket">>, <<"bucket">>}, <<"key">>,
riakc_set:to_op(riakc_set:add_element(<<"X">>,
riakc_set:add_element(<<"Y">>, riakc_set:new())))),
{ok, S0} = fetch_type(Pid, {<<"set_bucket">>, <<"bucket">>}, <<"key">>),
?assert(riakc_set:is_element(<<"X">>, S0)),
?assert(riakc_set:is_element(<<"Y">>, S0)),
?assertEqual(riakc_set:size(S0), 2),
ok = update_type(Pid,
{<<"set_bucket">>, <<"bucket">>}, <<"key">>,
riakc_set:to_op(riakc_set:add_element(<<"X">>, riakc_set:del_element(<<"X">>, S0)))),
{ok, S1} = fetch_type(Pid, {<<"set_bucket">>, <<"bucket">>}, <<"key">>),
?assert(riakc_set:is_element(<<"X">>, S1)),
?assert(riakc_set:is_element(<<"Y">>, S1)),
?assertEqual(riakc_set:size(S1), 2)
end)},
{"remove item from set with outdated context",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
ok = update_type(Pid,
{<<"set_bucket">>, <<"bucket">>}, <<"key">>,
riakc_set:to_op(riakc_set:add_element(<<"X">>,
riakc_set:add_element(<<"Y">>, riakc_set:new())))),
{ok, S0} = fetch_type(Pid, {<<"set_bucket">>, <<"bucket">>}, <<"key">>),
?assert(riakc_set:is_element(<<"X">>, S0)),
?assert(riakc_set:is_element(<<"Y">>, S0)),
?assertEqual(riakc_set:size(S0), 2),
ok = update_type(Pid,
{<<"set_bucket">>, <<"bucket">>}, <<"key">>,
riakc_set:to_op(riakc_set:add_element(<<"Z">>, riakc_set:new()))),
ok = update_type(Pid,
{<<"set_bucket">>, <<"bucket">>}, <<"key">>,
riakc_set:to_op(riakc_set:del_element(<<"Z">>, S0))),
{ok, S1} = fetch_type(Pid, {<<"set_bucket">>, <<"bucket">>}, <<"key">>),
?assert(riakc_set:is_element(<<"X">>, S1)),
?assert(riakc_set:is_element(<<"Y">>, S1)),
?assert(riakc_set:is_element(<<"Z">>, S1)),
?assertEqual(riakc_set:size(S1), 3)
end)},
{"add item to nested set in map while also removing set",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
ok = riakc_pb_socket:update_type(Pid,
{<<"map_bucket">>, <<"bucket">>}, <<"key">>,
riakc_map:to_op(riakc_map:update({<<"set">>, set},
fun(S) ->
riakc_set:add_element(<<"X">>,
riakc_set:add_element(<<"Y">>, S))
end, riakc_map:new()))),
{ok, M0} = fetch_type(Pid, {<<"map_bucket">>, <<"bucket">>}, <<"key">>),
L0 = riakc_map:fetch({<<"set">>, set}, M0),
?assert(lists:member(<<"X">>, L0)),
?assert(lists:member(<<"Y">>, L0)),
?assertEqual(length(L0), 2),
M1 = riakc_map:update({<<"set">>, set},
fun(S) -> riakc_set:add_element(<<"Z">>, S) end,
M0),
M2 = riakc_map:erase({<<"set">>, set}, M1),
ok = update_type(Pid,
{<<"map_bucket">>, <<"bucket">>}, <<"key">>,
riakc_map:to_op(M2)),
{ok, M3} = fetch_type(Pid, {<<"map_bucket">>, <<"bucket">>}, <<"key">>),
L1 = riakc_map:fetch({<<"set">>, set}, M3),
?assert(lists:member(<<"Z">>, L1)),
?assertEqual(length(L1), 1)
end)},
{"increment nested counter in map while also removing counter",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
ok = riakc_pb_socket:update_type(Pid,
{<<"map_bucket">>, <<"bucket">>}, <<"key">>,
riakc_map:to_op(riakc_map:update({<<"counter">>, counter},
fun(C) ->
riakc_counter:increment(5, C)
end, riakc_map:new()))),
{ok, M0} = fetch_type(Pid, {<<"map_bucket">>, <<"bucket">>}, <<"key">>),
C0 = riakc_map:fetch({<<"counter">>, counter}, M0),
?assertEqual(C0, 5),
M1 = riakc_map:update({<<"counter">>, counter},
fun(C) -> riakc_counter:increment(2, C) end,
M0),
M2 = riakc_map:erase({<<"counter">>, counter}, M1),
ok = update_type(Pid,
{<<"map_bucket">>, <<"bucket">>}, <<"key">>,
riakc_map:to_op(M2)),
{ok, M3} = fetch_type(Pid, {<<"map_bucket">>, <<"bucket">>}, <<"key">>),
C1 = riakc_map:fetch({<<"counter">>, counter}, M3),
%% Expected result depends on combination of vnodes involved, so accept either answer
?assert(C1 =:= 2 orelse C1 =:= 7)
end)},
{"add item to nested set in nested map in map while also removing nested map",
?_test(begin
reset_riak(),
{ok, Pid} = start_link(test_ip(), test_port()),
M0 = riakc_map:update({<<"map">>, map},
fun(M) ->
riakc_map:update({<<"set">>, set},
fun(S) ->
riakc_set:add_element(<<"X">>,
riakc_set:add_element(<<"Y">>, S))
end,
M)
end,
riakc_map:new()),
ok = riakc_pb_socket:update_type(Pid,
{<<"map_bucket">>, <<"bucket">>}, <<"key">>,
riakc_map:to_op(M0)),
{ok, M1} = fetch_type(Pid, {<<"map_bucket">>, <<"bucket">>}, <<"key">>),
L0 = orddict:fetch({<<"set">>, set}, riakc_map:fetch({<<"map">>, map}, M1)),
?assert(lists:member(<<"X">>, L0)),
?assert(lists:member(<<"Y">>, L0)),
?assertEqual(length(L0), 2),
M2 = riakc_map:update({<<"map">>, map},
fun(M) -> riakc_map:update({<<"set">>, set},
fun(S) -> riakc_set:add_element(<<"Z">>, S) end,
M)
end,
M1),
M3 = riakc_map:erase({<<"map">>, map}, M2),
ok = update_type(Pid,
{<<"map_bucket">>, <<"bucket">>}, <<"key">>,
riakc_map:to_op(M3)),
{ok, M4} = fetch_type(Pid, {<<"map_bucket">>, <<"bucket">>}, <<"key">>),
L1 = orddict:fetch({<<"set">>, set}, riakc_map:fetch({<<"map">>, map}, M4)),
?assert(lists:member(<<"Z">>, L1)),
?assertEqual(length(L1), 1)
end)}
].
-endif.