Add flow control

Flow control is disabled by default. The initial flow value
must be set to enable it (either for the entire connection
or on a per-request basis). Flow applies to all HTTP streams
as well as Websocket. HTTP/2 pushed streams receive the same
value as their originating stream.

1 files changed, 11 insertions, 7 deletions

diff --git a/src/gun_sse_h.erl b/src/gun_sse_h.erl
index 11dc443..9d8836e 100644
--- a/src/gun_sse_h.erl
+++ b/src/gun_sse_h.erl
@@ -38,18 +38,22 @@ init(ReplyTo, StreamRef, _, Headers, _) ->
 			disable
 	end.
 
--spec handle(_, binary(), State) -> {done, State} when State::#state{}.
-handle(IsFin, Data, State=#state{reply_to=ReplyTo, stream_ref=StreamRef, sse_state=SSE0}) ->
+-spec handle(_, binary(), State) -> {done, non_neg_integer(), State} when State::#state{}.
+handle(IsFin, Data, State) ->
+	handle(IsFin, Data, State, 0).
+
+handle(IsFin, Data, State=#state{reply_to=ReplyTo, stream_ref=StreamRef, sse_state=SSE0}, Flow) ->
 	case cow_sse:parse(Data, SSE0) of
 		{event, Event, SSE} ->
 			ReplyTo ! {gun_sse, self(), StreamRef, Event},
-			handle(IsFin, <<>>, State#state{sse_state=SSE});
+			handle(IsFin, <<>>, State#state{sse_state=SSE}, Flow + 1);
 		{more, SSE} ->
-			_ = case IsFin of
+			Inc = case IsFin of
 				fin ->
-					ReplyTo ! {gun_sse, self(), StreamRef, fin};
+					ReplyTo ! {gun_sse, self(), StreamRef, fin},
+					1;
 				_ ->
-					ok
+					0
 			end,
-			{done, State#state{sse_state=SSE}}
+			{done, Flow + Inc, State#state{sse_state=SSE}}
 	end.