Handle option maps in lower level modules

The goal is to keep everything below the _api_wrap module level relatively
intact. To achieve that handle option maps in some places, or translate back to
a proplist or `#httpd{}` records in others.

The `couch_replicator_api:db_from_json/1` function is where endpoint map object
from a `Rep` object are translated into `#httpdb{}` records. Headers are
translated back to lists and ibrowse options into proplist with atom keys.

3 files changed, 127 insertions, 100 deletions

diff --git a/src/couch_replicator/src/couch_replicator_api_wrap.erl b/src/couch_replicator/src/couch_replicator_api_wrap.erl
index a21de4242..da6f28800 100644
--- a/src/couch_replicator/src/couch_replicator_api_wrap.erl
+++ b/src/couch_replicator/src/couch_replicator_api_wrap.erl
@@ -28,7 +28,6 @@
     db_close/1,
     get_db_info/1,
     get_pending_count/2,
-    get_view_info/3,
     update_doc/3,
     update_doc/4,
     update_docs/3,
@@ -39,39 +38,28 @@
     open_doc_revs/6,
     changes_since/5,
     db_uri/1,
-    normalize_db/1
+    db_from_json/1
     ]).
 
--import(couch_replicator_httpc, [
-    send_req/3
-    ]).
-
--import(couch_util, [
-    encode_doc_id/1,
-    get_value/2,
-    get_value/3
-    ]).
 
 -define(MAX_WAIT, 5 * 60 * 1000).
 
 -define(MAX_URL_LEN, 7000).
 -define(MIN_URL_LEN, 200).
 
-db_uri(#httpdb{url = Url}) ->
+db_uri(#{<<"url">> := Url}) ->
     couch_util:url_strip_password(Url);
 
-db_uri(DbName) when is_binary(DbName) ->
-    ?b2l(DbName);
+db_uri(#httpdb{url = Url}) ->
+    couch_util:url_strip_password(Url).
 
-db_uri(Db) ->
-    db_uri(couch_db:name(Db)).
 
+db_open(#{} = Db) ->
+    db_open(Db, false, #{}).
 
-db_open(Db) ->
-    db_open(Db, false, []).
 
-db_open(#httpdb{} = Db1, Create, CreateParams) ->
-    {ok, Db} = couch_replicator_httpc:setup(Db1),
+db_open(#{} = Db0, Create, #{} = CreateParams) when is_boolean(Create) ->
+    {ok, Db} = couch_replicator_httpc:setup(db_from_json(Db0)),
     try
         case Create of
         false ->
@@ -149,14 +137,6 @@ get_pending_count(#httpdb{} = Db, Seq) ->
         {ok, couch_util:get_value(<<"pending">>, Props, null)}
     end).
 
-get_view_info(#httpdb{} = Db, DDocId, ViewName) ->
-    Path = io_lib:format("~s/_view/~s/_info", [DDocId, ViewName]),
-    send_req(Db, [{path, Path}],
-        fun(200, _, {Props}) ->
-            {VInfo} = couch_util:get_value(<<"view_index">>, Props, {[]}),
-            {ok, VInfo}
-        end).
-
 
 ensure_full_commit(#httpdb{} = Db) ->
     send_req(
@@ -434,9 +414,9 @@ changes_since(#httpdb{headers = Headers1, timeout = InactiveTimeout} = HttpDb,
     {undefined, undefined} ->
         QArgs1 = maybe_add_changes_filter_q_args(BaseQArgs, Options),
         {QArgs1, get, [], Headers1};
-    {undefined, _} when is_tuple(Selector) ->
+    {undefined, #{}} ->
         Headers2 = [{"Content-Type", "application/json"} | Headers1],
-        JsonSelector = ?JSON_ENCODE({[{<<"selector">>, Selector}]}),
+        JsonSelector = ?JSON_ENCODE(#{<<"selector">> => Selector}),
         {[{"filter", "_selector"} | BaseQArgs], post, JsonSelector, Headers2};
     {_, undefined} when is_list(DocIds) ->
         Headers2 = [{"Content-Type", "application/json"} | Headers1],
@@ -496,7 +476,8 @@ maybe_add_changes_filter_q_args(BaseQS, Options) ->
         ViewFields0 = [atom_to_list(F) || F <- record_info(fields,  mrargs)],
         ViewFields = ["key" | ViewFields0],
 
-        {Params} = get_value(query_params, Options, {[]}),
+        ParamsMap = #{} = get_value(query_params, Options, #{}),
+        Params = maps:to_list(ParamsMap),
         [{"filter", ?b2l(FilterName)} | lists:foldl(
             fun({K, V}, QSAcc) ->
                 Ks = couch_util:to_list(K),
@@ -546,7 +527,7 @@ options_to_query_args(HttpDb, Path, Options0) ->
             length("GET " ++ FullUrl ++ " HTTP/1.1\r\n") +
             length("&atts_since=") + 6,  % +6 = % encoded [ and ]
             PAs, MaxLen, []),
-        [{"atts_since", ?JSON_ENCODE(RevList)} | QueryArgs1]
+        [{"atts_since", ?b2l(iolist_to_binary(?JSON_ENCODE(RevList)))} | QueryArgs1]
     end.
 
 
@@ -787,7 +768,7 @@ json_to_doc_info({Props}) ->
         RevsInfo0 = lists:map(
             fun({Change}) ->
                 Rev = couch_doc:parse_rev(get_value(<<"rev">>, Change)),
-                Del = couch_replicator_utils:is_deleted(Change),
+                Del = get_value(<<"deleted">>, Change, false),
                 #rev_info{rev=Rev, deleted=Del}
             end, Changes),
 
@@ -895,52 +876,95 @@ header_value(Key, Headers, Default) ->
     end.
 
 
-% Normalize an #httpdb{} or #db{} record such that it can be used for
-% comparisons. This means remove things like pids and also sort options / props.
-normalize_db(#httpdb{} = HttpDb) ->
+maybe_append_create_query_params(Db, Params) when map_size(Params) == 0 ->
+    Db;
+
+maybe_append_create_query_params(Db, #{} = Params) ->
+    ParamList = maps:to_list(Params),
+    NewUrl = Db#httpdb.url ++ "?" ++ mochiweb_util:urlencode(ParamList),
+    Db#httpdb{url = NewUrl}.
+
+
+db_from_json(#{} = DbMap) ->
+    #{
+        <<"url">> := Url,
+        <<"auth_props">> := Auth,
+        <<"headers">> := Headers0,
+        <<"ibrowse_options">> := IBrowseOptions0,
+        <<"timeout">> := Timeout,
+        <<"http_connections">> := HttpConnections,
+        <<"retries">> := Retries,
+        <<"proxy_url">> := ProxyUrl0
+    } = DbMap,
+    Headers = maps:fold(fun(K, V, Acc) ->
+        [{binary_to_list(K), binary_to_list(V)} | Acc]
+    end, [], Headers0),
+    IBrowseOptions = maps:fold(fun
+        (<<"socket_options">>, #{} = SockOpts, Acc) ->
+            SockOptsKVs = maps:fold(fun sock_opts_fold/3, [], SockOpts),
+            [{socket_options, SockOptsKVs} | Acc];
+        (<<"ssl_options">>, #{} = SslOpts, Acc) ->
+            SslOptsKVs = maps:fold(fun ssl_opts_fold/3, [], SslOpts),
+            [{ssl_options, SslOptsKVs} | Acc];
+        (K, V, Acc) when is_binary(V) ->
+            [{binary_to_atom(K, utf8), binary_to_list(V)} | Acc];
+        (K, V, Acc) ->
+            [{binary_to_atom(K, utf8), V} | Acc]
+    end, [], IBrowseOptions0),
+    ProxyUrl = case ProxyUrl0 of
+        null -> undefined;
+        V when is_binary(V) -> binary_to_list(V)
+    end,
     #httpdb{
-        url = HttpDb#httpdb.url,
-        auth_props = lists:sort(HttpDb#httpdb.auth_props),
-        headers = lists:keysort(1, HttpDb#httpdb.headers),
-        timeout = HttpDb#httpdb.timeout,
-        ibrowse_options = lists:keysort(1, HttpDb#httpdb.ibrowse_options),
-        retries = HttpDb#httpdb.retries,
-        http_connections = HttpDb#httpdb.http_connections
-    };
+        url = binary_to_list(Url),
+        auth_props = maps:to_list(Auth),
+        headers = Headers,
+        ibrowse_options = IBrowseOptions,
+        timeout = Timeout,
+        http_connections = HttpConnections,
+        retries = Retries,
+        proxy_url = ProxyUrl
+    }.
 
-normalize_db(<<DbName/binary>>) ->
-    DbName.
 
+send_req(#httpdb{} = HttpDb, Opts, Callback) when is_function(Callback) ->
+    couch_replicator_httpc:send_req(HttpDb, Opts, Callback).
 
-maybe_append_create_query_params(Db, []) ->
-    Db;
 
-maybe_append_create_query_params(Db, CreateParams) ->
-    NewUrl = Db#httpdb.url ++ "?" ++ mochiweb_util:urlencode(CreateParams),
-    Db#httpdb{url = NewUrl}.
+get_value(K, Props) ->
+    couch_util:get_value(K, Props).
+
+
+get_value(K, Props, Default) ->
+    couch_util:get_value(K, Props, Default).
 
 
--ifdef(TEST).
+encode_doc_id(DocId) ->
+    couch_util:encode_doc_id(DocId).
 
--include_lib("eunit/include/eunit.hrl").
 
 
-normalize_http_db_test() ->
-    HttpDb =  #httpdb{
-        url = "http://host/db",
-        auth_props = [{"key", "val"}],
-        headers = [{"k2","v2"}, {"k1","v1"}],
-        timeout = 30000,
-        ibrowse_options = [{k2, v2}, {k1, v1}],
-        retries = 10,
-        http_connections = 20
-    },
-    Expected = HttpDb#httpdb{
-        headers = [{"k1","v1"}, {"k2","v2"}],
-        ibrowse_options = [{k1, v1}, {k2, v2}]
-    },
-    ?assertEqual(Expected, normalize_db(HttpDb)),
-    ?assertEqual(<<"local">>, normalize_db(<<"local">>)).
+% See couch_replicator_docs:ssl_params/1 for ssl parsed options
+% and http://erlang.org/doc/man/ssl.html#type-server_option
+% all latest SSL server options
+%
+ssl_opts_fold(K, V, Acc) when is_boolean(V); is_integer(V) ->
+    [{binary_to_atom(K, utf8), V} | Acc];
+
+ssl_opts_fold(K, null, Acc) ->
+    [{binary_to_atom(K, utf8), undefined} | Acc];
+
+ssl_opts_fold(<<"verify">>, V, Acc) ->
+    [{verify, binary_to_atom(V, utf8)} | Acc];
 
+ssl_opts_fold(K, V, Acc) when is_list(V) ->
+    [{binary_to_atom(K, utf8), binary_to_list(V)} | Acc].
+
+
+% See ?VALID_SOCK_OPTS in couch_replicator_docs for accepted socket options
+%
+sock_opts_fold(K, V, Acc) when is_binary(V) ->
+     [{binary_to_atom(K, utf8), binary_to_atom(V, utf8)} | Acc];
 
--endif.
+sock_opts_fold(K, V, Acc) when is_boolean(V); is_integer(V) ->
+    [{binary_to_atom(K, utf8), V} | Acc].
diff --git a/src/couch_replicator/src/couch_replicator_changes_reader.erl b/src/couch_replicator/src/couch_replicator_changes_reader.erl
index 2e4df5365..6adf1af5e 100644
--- a/src/couch_replicator/src/couch_replicator_changes_reader.erl
+++ b/src/couch_replicator/src/couch_replicator_changes_reader.erl
@@ -22,11 +22,8 @@
 -include_lib("couch_replicator/include/couch_replicator_api_wrap.hrl").
 -include("couch_replicator.hrl").
 
--import(couch_util, [
-    get_value/2
-]).
 
-start_link(StartSeq, #httpdb{} = Db, ChangesQueue, Options) ->
+start_link(StartSeq, #httpdb{} = Db, ChangesQueue, #{} = Options) ->
     Parent = self(),
     {ok, spawn_link(fun() ->
         put(last_seq, StartSeq),
@@ -41,12 +38,12 @@ start_link(StartSeq, Db, ChangesQueue, Options) ->
     end)}.
 
 read_changes(Parent, StartSeq, Db, ChangesQueue, Options) ->
-    Continuous = couch_util:get_value(continuous, Options),
+    Continuous = maps:get(<<"continuous">>, Options, false),
     try
         couch_replicator_api_wrap:changes_since(Db, all_docs, StartSeq,
             fun(Item) ->
                 process_change(Item, {Parent, Db, ChangesQueue, Continuous})
-            end, Options),
+            end, couch_replicator_utils:proplist_options(Options)),
         couch_work_queue:close(ChangesQueue)
     catch
         throw:recurse ->
diff --git a/src/couch_replicator/src/couch_replicator_filters.erl b/src/couch_replicator/src/couch_replicator_filters.erl
index c8980001a..50c37335d 100644
--- a/src/couch_replicator/src/couch_replicator_filters.erl
+++ b/src/couch_replicator/src/couch_replicator_filters.erl
@@ -20,6 +20,7 @@
 ]).
 
 -include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
 
 
 % Parse the filter from replication options proplist.
@@ -27,17 +28,17 @@
 % For `user` filter, i.e. filters specified as user code
 % in source database, this code doesn't fetch the filter
 % code, but only returns the name of the filter.
--spec parse([_]) ->
+-spec parse(#{}) ->
     {ok, nil} |
     {ok, {view, binary(), {[_]}}} |
     {ok, {user, {binary(), binary()}, {[_]}}} |
     {ok, {docids, [_]}} |
     {ok, {mango, {[_]}}} |
     {error, binary()}.
-parse(Options) ->
-    Filter = couch_util:get_value(filter, Options),
-    DocIds = couch_util:get_value(doc_ids, Options),
-    Selector = couch_util:get_value(selector, Options),
+parse(#{} = Options) ->
+    Filter = maps:get(<<"filter">>, Options, undefined),
+    DocIds = maps:get(<<"doc_ids">>, Options, undefined),
+    Selector = maps:get(<<"selector">>, Options, undefined),
     case {Filter, DocIds, Selector} of
         {undefined, undefined, undefined} ->
             {ok, nil};
@@ -53,7 +54,10 @@ parse(Options) ->
         {undefined, _, undefined} ->
             {ok, {docids, DocIds}};
         {undefined, undefined, _} ->
-            {ok, {mango, ejsort(mango_selector:normalize(Selector))}};
+            % Translate it to proplist as normalize doesn't know how
+            % to handle maps
+            Selector1 = ?JSON_DECODE(?JSON_ENCODE(Selector)),
+            {ok, {mango, ejsort(mango_selector:normalize(Selector1))}};
         _ ->
             Err = "`selector`, `filter` and `doc_ids` are mutually exclusive",
             {error, list_to_binary(Err)}
@@ -88,22 +92,24 @@ fetch(DDocName, FilterName, Source) ->
 
 
 % Get replication type and view (if any) from replication document props
--spec view_type([_], [_]) ->
-    {view, {binary(), binary()}} | {db, nil} | {error, binary()}.
-view_type(Props, Options) ->
-    case couch_util:get_value(<<"filter">>, Props) of
-        <<"_view">> ->
-            {QP}  = couch_util:get_value(query_params, Options, {[]}),
-            ViewParam = couch_util:get_value(<<"view">>, QP),
-            case re:split(ViewParam, <<"/">>) of
-                [DName, ViewName] ->
-                    {view, {<< "_design/", DName/binary >>, ViewName}};
-                _ ->
-                    {error, <<"Invalid `view` parameter.">>}
-            end;
+-spec view_type(#{}, #{}) ->
+    {binary(), #{}} | {error, binary()}.
+view_type(#{?FILTER := <<"_view">>}, #{} = Options) ->
+    QP = maps:get(<<"query_params">>, Options, #{}),
+    ViewParam = maps:get(<<"view">>, QP, <<>>),
+    case re:split(ViewParam, <<"/">>) of
+        [DName, ViewName] ->
+            DDocMap = #{
+                <<"ddoc">> => <<"_design/",DName/binary>>,
+                <<"view">> => ViewName
+            },
+            {<<"view">>, DDocMap};
         _ ->
-            {db, nil}
-    end.
+            {error, <<"Invalid `view` parameter.">>}
+    end;
+
+view_type(#{}, #{}) ->
+    {<<"db">>, #{}}.
 
 
 % Private functions
@@ -151,9 +157,9 @@ fetch_internal(DDocName, FilterName, Source) ->
     end.
 
 
--spec query_params([_]) -> {[_]}.
-query_params(Options)->
-    couch_util:get_value(query_params, Options, {[]}).
+-spec query_params(#{}) -> #{}.
+query_params(#{} = Options)->
+    maps:get(<<"query_params">>, Options, #{}).
 
 
 parse_user_filter(Filter) ->