def serve.stream.talk
bend-kit-http@0.38.2.0 · http.bend:6648
import bend-kit-http@0.38.2.0/http.bend as Http
import Base
import bend-kit-http@0.38.2.0/admission.bend as Admission
import bend-kit-http@0.38.2.0/completion.bend as Completion
import bend-kit-time@0.3.0.0/time.bend as Time
import bend-kit-bytes@0.3.3.0/bytes.bend as Bytes
6648def serve.stream.talk(
6649 ~S: Type, ~start: Req -> IO(S), ~piece: S -> Bytes.Bytes -> IO(S & Bool),
6650 ~finish: S -> IO(Reply), ~abort: S -> StreamAbort -> IO(Unit), ~observe: U32 -> Completion.Outcome -> IO(Unit), +settings: Limits, +inbox: Chan(Admission.Command), step: StreamServing<S>
6651) -> IO(Unit):
6652 Limits{max, cap, hm, bm, im, wm} = settings
6653 match step:
6654 case StreamDone{}:
6655 IO.pure(Unit, Unit{})
6656 case StreamCheck{held, connection, state}:
6657 Serving{(socket, turn), phase} = connection
6658 serve.stream.talk(~S, ~start, ~piece, ~finish, ~abort, ~observe, settings, inbox,
6659 serve.stream.ready(~S, held, socket, phase, state, admission.ready(turn)))
6660 case StreamReady{+held, +ready, connection, state}:
6661 do IO<Unit>:
6662 admitted : Bool <- admission.stream.take(Bool.not(held) && ready, inbox)
6663 serve.stream.talk(~S, ~start, ~piece, ~finish, ~abort, ~observe, settings, inbox,
6664 StreamChecked{held || ready, admitted, connection, state})
6665 case StreamChecked{+held, admitted, connection, m}:
6666 match admitted:
6667 case False{}:
6668 Serving{(socket, turn), phase} = connection
6669 do IO<Unit>:
6670 serve.stream.abort(~S, ~abort, StreamFailed{503}, m)
6671 admission.reject(Completion.Observed{status => outcome => observe(status, outcome)}, socket, admission.turn.head(turn), wm)
6672 case True{}:
6673 Serving{(s, t), phase} = connection
6674 match t:
6675 case TurnClose{}:
6676 do IO<Unit>:
6677 Socket.close(s)
6678 serve.stream.abort(~S, ~abort, StreamClosed{}, m)
6679 admission.give.held(held, inbox)
6680 case TurnInputFailed{operation, code, message}:
6681 do IO<Unit>:
6682 Socket.close(s)
6683 serve.stream.abort(~S, ~abort, StreamInputFailed{operation, code, message}, m)
6684 admission.give.held(held, inbox)
6685 case TurnFail{+status}:
6686 do IO<Unit>:
6687 serve.stream.abort(~S, ~abort, StreamFailed{status}, m)
6688 reply_fail(Completion.Observed{status => outcome => observe(status, outcome)}, s, status, wm)
6689 admission.give.held(held, inbox)
6690 case TurnReq{req, rest, v11}:
6691 Req{+method, path, +headers, body} = req
6692 do IO<Unit>:
6693 next : S & Bool <- serve.stream.feed(~S, ~start, ~piece,
6694 Req{method, path, headers, Bytes.new(0)}, m, [body])
6695 after : Maybe<&1, Socket & Turn> <- serve.stream.respond(~S, ~finish, ~observe, settings, s, rest, headers, v11, method, next)
6696 Admission.give(inbox, 1)
6697 serve.stream.talk(~S, ~start, ~piece, ~finish, ~abort, ~observe, settings, inbox,
6698 serve.stream.after(~S, after))
6699 case TurnMore{sv, cont}:
6700 match sv:
6701 case SvHead{buf}:
6702 do IO<Unit>:
6703 received : (Socket & Turn) & Phase <- talk.recv(inbox, held, settings, s, SvHead{buf}, cont, phase)
6704 serve.stream.talk(~S, ~start, ~piece, ~finish, ~abort, ~observe, settings, inbox,
6705 StreamCheck{held, talk.state(received), m})
6706 case SvLen{req, +v11, +left, parts}:
6707 Req{+method, +path, +headers, body} = req
6708 do IO<Unit>:
6709 now : Time.Duration <- Time.mono()
6710 reading : Phase = phase.body(phase, settings, now)
6711 next : S & Bool <- serve.stream.feed(~S, ~start, ~piece,
6712 Req{method, path, headers, body}, m, List.reverse(&1, Bytes.Bytes, parts))
6713 received : (Socket & Turn) & Phase <- talk.recv(inbox, held, settings, s,
6714 SvLen{Req{method, path, headers, Bytes.new(0)}, v11, left, Nil{}}, cont, Some{reading})
6715 serve.stream.talk(~S, ~start, ~piece, ~finish, ~abort, ~observe, settings, inbox,
6716 StreamCheck{held, talk.state(received), Some{next}})
6717 case SvChunk{req, +v11, +n, st, parts}:
6718 Req{+method, +path, +headers, body} = req
6719 do IO<Unit>:
6720 now : Time.Duration <- Time.mono()
6721 reading : Phase = phase.body(phase, settings, now)
6722 next : S & Bool <- serve.stream.feed(~S, ~start, ~piece,
6723 Req{method, path, headers, body}, m, List.reverse(&1, Bytes.Bytes, parts))
6724 received : (Socket & Turn) & Phase <- talk.recv(inbox, held, settings, s,
6725 SvChunk{Req{method, path, headers, Bytes.new(0)}, v11, n, st, Nil{}}, cont, Some{reading})
6726 serve.stream.talk(~S, ~start, ~piece, ~finish, ~abort, ~observe, settings, inbox,
6727 StreamCheck{held, talk.state(received), Some{next}})