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