proxy the response from the other node

3 files changed, 58 insertions, 7 deletions

diff --git a/src/chttpd/src/chttpd.erl b/src/chttpd/src/chttpd.erl
index 87fb34158..adde0730f 100644
--- a/src/chttpd/src/chttpd.erl
+++ b/src/chttpd/src/chttpd.erl
@@ -742,6 +742,8 @@ start_chunked_response(#httpd{mochi_req=MochiReq}=Req, Code, Headers0) ->
     end,
     {ok, Resp}.
 
+send_chunk({remote, _Pid, _Ref} = Resp, Data) ->
+    couch_httpd:send_chunk(Resp, Data);
 send_chunk(Resp, Data) ->
     Resp:write_chunk(Data),
     {ok, Resp}.
diff --git a/src/chttpd/src/chttpd_node.erl b/src/chttpd/src/chttpd_node.erl
index f8736f13f..202070279 100644
--- a/src/chttpd/src/chttpd_node.erl
+++ b/src/chttpd/src/chttpd_node.erl
@@ -124,17 +124,41 @@ handle_node_req(#httpd{path_parts=[_, Node | PathParts],
     {_, Query, Fragment} = mochiweb_util:urlsplit_path(RawUri),
     NewPath0 = "/" ++ lists:join("/", [?b2l(P) || P <- PathParts]),
     NewRawPath = mochiweb_util:urlunsplit_path({NewPath0, Query, Fragment}),
-    MochiReq = mochiweb_request:new(self(),
+    MaxSize =  config:get_integer("httpd", "max_http_request_size", 4294967296),
+    NewOpts = [{body, MochiReq0:recv_body(MaxSize)} | MochiReq0:get(opts)],
+    Ref = erlang:make_ref(),
+    MochiReq = mochiweb_request:new({remote, self(), Ref},
+                               NewOpts,
                                MochiReq0:get(method),
                                NewRawPath,
                                MochiReq0:get(version),
                                MochiReq0:get(headers)),
-    call_node(Node, couch_httpd, handle_request, [MochiReq]);
+    call_node(Node, couch_httpd, handle_request, [MochiReq]),
+    recv_loop(Ref, MochiReq0);
 handle_node_req(#httpd{path_parts=[_]}=Req) ->
     chttpd:send_error(Req, {bad_request, <<"Incomplete path to _node request">>});
 handle_node_req(Req) ->
     chttpd:send_error(Req, not_found).
 
+recv_loop(Ref, ReqResp) ->
+    receive
+        {Ref, Code, Headers, _Args, start_response} ->
+            recv_loop(Ref, ReqResp:start({Code, Headers}));
+        {Ref, Code, Headers, chunked, respond} ->
+            Resp = ReqResp:respond({Code, Headers, chunked}),
+            recv_loop(Ref, Resp);
+        {Ref, Code, Headers, Args, respond} ->
+            Resp = ReqResp:respond({Code, Headers, Args}),
+            {ok, Resp};
+        {Ref, chunk, <<>>} ->
+            ReqResp:write_chunk(<<>>),
+            {ok, ReqResp};
+        {Ref, chunk, Data} ->
+            ReqResp:write_chunk(Data),
+            recv_loop(Ref, ReqResp);
+        _Else ->
+            recv_loop(Ref, ReqResp)
+    end.
 
 call_node(Node0, Mod, Fun, Args) when is_binary(Node0) ->
     Node1 = try
diff --git a/src/couch/src/couch_httpd.erl b/src/couch/src/couch_httpd.erl
index ba35d6f1b..73d3c1704 100644
--- a/src/couch/src/couch_httpd.erl
+++ b/src/couch/src/couch_httpd.erl
@@ -221,6 +221,8 @@ make_fun_spec_strs(SpecStr) ->
     re:split(SpecStr, "(?<=})\\s*,\\s*(?={)", [{return, list}]).
 
 handle_request(MochiReq) ->
+    Body = proplists:get_value(body, MochiReq:get(opts)),
+    erlang:put(mochiweb_request_body, Body),
     apply(?MODULE, handle_request, [MochiReq | get_httpd_handlers()]).
 
 handle_request(MochiReq, DefaultFun, UrlHandlers, DbUrlHandlers,
@@ -262,7 +264,7 @@ handle_request_int(MochiReq, DefaultFun,
         MochiReq:get(method),
         RawUri,
         MochiReq:get(version),
-        MochiReq:get(peer),
+        peer(MochiReq),
         mochiweb_headers:to_list(MochiReq:get(headers))
     ]),
 
@@ -305,7 +307,7 @@ handle_request_int(MochiReq, DefaultFun,
 
     HttpReq = #httpd{
         mochi_req = MochiReq,
-        peer = MochiReq:get(peer),
+        peer = peer(MochiReq),
         method = Method,
         requested_path_parts =
             [?l2b(unquote(Part)) || Part <- string:tokens(RequestedPath, "/")],
@@ -752,6 +754,9 @@ start_chunked_response(#httpd{mochi_req=MochiReq}=Req, Code, Headers0) ->
     end,
     {ok, Resp}.
 
+send_chunk({remote, Pid, Ref} = Resp, Data) ->
+    Pid ! {Ref, chunk, Data},
+    {ok, Resp};
 send_chunk(Resp, Data) ->
     case iolist_size(Data) of
     0 -> ok; % do nothing
@@ -759,6 +764,9 @@ send_chunk(Resp, Data) ->
     end,
     {ok, Resp}.
 
+last_chunk({remote, Pid, Ref} = Resp) ->
+    Pid ! {Ref, chunk, <<>>},
+    {ok, Resp};
 last_chunk(Resp) ->
     Resp:write_chunk([]),
     {ok, Resp}.
@@ -1181,9 +1189,18 @@ before_response(Req0, Code0, Headers0, {json, JsonObj}) ->
 before_response(Req0, Code0, Headers0, Args0) ->
     chttpd_plugin:before_response(Req0, Code0, Headers0, Args0).
 
-respond_(#httpd{mochi_req = MochiReq}, Code, Headers, _Args, start_response) ->
+respond_(#httpd{mochi_req = MochiReq} = Req, Code, Headers, Args, Type) ->
+    case MochiReq:get(socket) of
+        {remote, Pid, Ref} ->
+            Pid ! {Ref, Code, Headers, Args, Type},
+            {remote, Pid, Ref};
+        _Else ->
+            http_respond_(Req, Code, Headers, Args, Type)
+    end.
+
+http_respond_(#httpd{mochi_req = MochiReq}, Code, Headers, _Args, start_response) ->
     MochiReq:start_response({Code, Headers});
-respond_(#httpd{mochi_req = MochiReq}, 413, Headers, Args, Type) ->
+http_respond_(#httpd{mochi_req = MochiReq}, 413, Headers, Args, Type) ->
     % Special handling for the 413 response. Make sure the socket is closed as
     % we don't know how much data was read before the error was thrown. Also
     % drain all the data in the receive buffer to avoid connction being reset
@@ -1195,9 +1212,17 @@ respond_(#httpd{mochi_req = MochiReq}, 413, Headers, Args, Type) ->
     Socket = MochiReq:get(socket),
     mochiweb_socket:recv(Socket, ?MAX_DRAIN_BYTES, ?MAX_DRAIN_TIME_MSEC),
     Result;
-respond_(#httpd{mochi_req = MochiReq}, Code, Headers, Args, Type) ->
+http_respond_(#httpd{mochi_req = MochiReq}, Code, Headers, Args, Type) ->
     MochiReq:Type({Code, Headers, Args}).
 
+peer(MochiReq) ->
+    case MochiReq:get(socket) of
+        {remote, Pid, _} ->
+            node(Pid);
+        _ ->
+            MochiReq:get(peer)
+    end.
+
 %%%%%%%% module tests below %%%%%%%%
 
 -ifdef(TEST).