make changes_wait_once really acting as longpoll connection.

src/couchbeam.erl

@@ -729,11 +729,10 @@ changes(Db) ->
 %%                  {heartbeat, string()|boolean()}
 changes(#db{server=Server, options=IbrowseOpts}=Db, Options) ->
     Url = make_url(Server, [db_url(Db), "/_changes"], Options),
-    case db_request(get, Url, ["200"], IbrowseOpts) of
-        {ok, _, _, Body} ->
-            {ok, couchbeam_util:json_decode(Body)};
-        Error ->
-            Error
+    case request_stream({self(), once}, get, Url, IbrowseOpts) of
+        {ok, ReqId} ->
+            couchbeam_changes:wait_for_change(ReqId);
+        {error, Error} -> {error, Error}
     end.
 
 %% @doc wait for longpoll changes 

src/couchbeam_changes.erl

@@ -6,14 +6,42 @@
 -module(couchbeam_changes).
 
 -include("couchbeam.hrl").
--export([continuous_acceptor/2]).
+-export([wait_for_change/1, continuous_acceptor/2]).
 
 -export([decode_row/1]).
 -record(state, {
     partial_chunk = <<"">>
 }).
 
--define(BUFFER_SIZE, 1000).
+
+wait_for_change(Reqid) ->
+    wait_for_change(Reqid, []).
+
+wait_for_change(Reqid, Acc) ->
+    receive
+        {ibrowse_async_response_end, Reqid} ->
+            Change = iolist_to_binary(lists:reverse(Acc)),
+            try
+                {ok, couchbeam_util:json_decode(Change)}
+            catch
+            throw:{invalid_json, Error} ->
+                {error, Error}
+            end;
+        {ibrowse_async_response, Reqid, {error,Error}} ->
+            {error, Error};
+        {ibrowse_async_response, Reqid, Chunk} ->
+            ibrowse:stream_next(Reqid),
+            wait_for_change(Reqid, [Chunk|Acc]);
+        {ibrowse_async_headers, Reqid, Status, Headers} ->
+            if Status =/= "200" ->
+                    {error, {Status, Headers}};
+                true ->
+                    ibrowse:stream_next(Reqid), 
+                    wait_for_change(Reqid, Acc) 
+            end
+    end.
+
+
 
 %% @doc initiate continuous loop 
 continuous_acceptor(Pid, PidRef) ->