From 611f9a9b78cab4005892e13dffb7a2c8e44580ee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lo=C3=AFc=20Hoguin?= Date: Fri, 2 Aug 2019 14:30:08 +0200 Subject: 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. --- src/gun_sse_h.erl | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) (limited to 'src/gun_sse_h.erl') 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. -- cgit v1.2.3