Http.serve.stream.talk: def of bend-kit-http@0.38.2.0
Import: import bend-kit-http@0.38.2.0/http.bend as Http
Lines 6648-6727 of bend-kit-http@0.38.2.0/http.bend: https://jend.leeeet.dev/read.txt?path=bend-kit-http@0.38.2.0/http.bend&start_line=6648&end_line=6727

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

def serve.stream.talk(
  ~S: Type, ~start: Req -> IO(S), ~piece: S -> Bytes.Bytes -> IO(S & Bool),
  ~finish: S -> IO(Reply), ~abort: S -> StreamAbort -> IO(Unit), ~observe: U32 -> Completion.Outcome -> IO(Unit), +settings: Limits, +inbox: Chan(Admission.Command), step: StreamServing<S>
) -> IO(Unit):
  Limits{max, cap, hm, bm, im, wm} = settings
  match step:
    case StreamDone{}:
      IO.pure(Unit, Unit{})
    case StreamCheck{held, connection, state}:
      Serving{(socket, turn), phase} = connection
      serve.stream.talk(~S, ~start, ~piece, ~finish, ~abort, ~observe, settings, inbox,
        serve.stream.ready(~S, held, socket, phase, state, admission.ready(turn)))
    case StreamReady{+held, +ready, connection, state}:
      do IO<Unit>:
        admitted : Bool <- admission.stream.take(Bool.not(held) && ready, inbox)
        serve.stream.talk(~S, ~start, ~piece, ~finish, ~abort, ~observe, settings, inbox,
          StreamChecked{held || ready, admitted, connection, state})
    case StreamChecked{+held, admitted, connection, m}:
      match admitted:
        case False{}:
          Serving{(socket, turn), phase} = connection
          do IO<Unit>:
            serve.stream.abort(~S, ~abort, StreamFailed{503}, m)
            admission.reject(Completion.Observed{status => outcome => observe(status, outcome)}, socket, admission.turn.head(turn), wm)
        case True{}:
          Serving{(s, t), phase} = connection
          match t:
            case TurnClose{}:
              do IO<Unit>:
                Socket.close(s)
                serve.stream.abort(~S, ~abort, StreamClosed{}, m)
                admission.give.held(held, inbox)
            case TurnInputFailed{operation, code, message}:
              do IO<Unit>:
                Socket.close(s)
                serve.stream.abort(~S, ~abort, StreamInputFailed{operation, code, message}, m)
                admission.give.held(held, inbox)
            case TurnFail{+status}:
              do IO<Unit>:
                serve.stream.abort(~S, ~abort, StreamFailed{status}, m)
                reply_fail(Completion.Observed{status => outcome => observe(status, outcome)}, s, status, wm)
                admission.give.held(held, inbox)
            case TurnReq{req, rest, v11}:
              Req{+method, path, +headers, body} = req
              do IO<Unit>:
                next : S & Bool <- serve.stream.feed(~S, ~start, ~piece,
                  Req{method, path, headers, Bytes.new(0)}, m, [body])
                after : Maybe<&1, Socket & Turn> <- serve.stream.respond(~S, ~finish, ~observe, settings, s, rest, headers, v11, method, next)
                Admission.give(inbox, 1)
                serve.stream.talk(~S, ~start, ~piece, ~finish, ~abort, ~observe, settings, inbox,
                  serve.stream.after(~S, after))
            case TurnMore{sv, cont}:
              match sv:
                case SvHead{buf}:
                  do IO<Unit>:
                    received : (Socket & Turn) & Phase <- talk.recv(inbox, held, settings, s, SvHead{buf}, cont, phase)
                    serve.stream.talk(~S, ~start, ~piece, ~finish, ~abort, ~observe, settings, inbox,
                      StreamCheck{held, talk.state(received), m})
                case SvLen{req, +v11, +left, parts}:
                  Req{+method, +path, +headers, body} = req
                  do IO<Unit>:
                    now : Time.Duration <- Time.mono()
                    reading : Phase = phase.body(phase, settings, now)
                    next : S & Bool <- serve.stream.feed(~S, ~start, ~piece,
                      Req{method, path, headers, body}, m, List.reverse(&1, Bytes.Bytes, parts))
                    received : (Socket & Turn) & Phase <- talk.recv(inbox, held, settings, s,
                      SvLen{Req{method, path, headers, Bytes.new(0)}, v11, left, Nil{}}, cont, Some{reading})
                    serve.stream.talk(~S, ~start, ~piece, ~finish, ~abort, ~observe, settings, inbox,
                      StreamCheck{held, talk.state(received), Some{next}})
                case SvChunk{req, +v11, +n, st, parts}:
                  Req{+method, +path, +headers, body} = req
                  do IO<Unit>:
                    now : Time.Duration <- Time.mono()
                    reading : Phase = phase.body(phase, settings, now)
                    next : S & Bool <- serve.stream.feed(~S, ~start, ~piece,
                      Req{method, path, headers, body}, m, List.reverse(&1, Bytes.Bytes, parts))
                    received : (Socket & Turn) & Phase <- talk.recv(inbox, held, settings, s,
                      SvChunk{Req{method, path, headers, Bytes.new(0)}, v11, n, st, Nil{}}, cont, Some{reading})
                    serve.stream.talk(~S, ~start, ~piece, ~finish, ~abort, ~observe, settings, inbox,
                      StreamCheck{held, talk.state(received), Some{next}})

Uses (48):
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=Bool
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=Bool.and
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=Bool.not
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=Bool.or
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=Chan
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=IO
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=IO.bind
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=IO.pure
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=List
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=List.reverse
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=Maybe
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=Pair
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=Sigma
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=Socket
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=Socket.close
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=U32
  https://jend.leeeet.dev/definition.txt?path=Base/base.bend&name=Unit
  https://jend.leeeet.dev/definition.txt?path=bend-kit-bytes@0.3.3.0/bytes.bend&name=Bytes
  https://jend.leeeet.dev/definition.txt?path=bend-kit-bytes@0.3.3.0/bytes.bend&name=new
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/admission.bend&name=Command
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/admission.bend&name=give
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/completion.bend&name=Outcome
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/completion.bend&name=Receipt
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=Limits
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=Phase
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=Reply
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=Req
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=Serving
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=StreamAbort
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=StreamServing
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=Sv
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=Turn
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=admission.give.held
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=admission.ready
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=admission.reject
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=admission.stream.take
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=admission.turn.head
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=phase.body
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=reply_fail
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=serve.stream.abort
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=serve.stream.after
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=serve.stream.feed
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=serve.stream.ready
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=serve.stream.respond
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=talk.recv
  https://jend.leeeet.dev/definition.txt?path=bend-kit-http@0.38.2.0/http.bend&name=talk.state
  https://jend.leeeet.dev/definition.txt?path=bend-kit-time@0.3.0.0/time.bend&name=Duration
  https://jend.leeeet.dev/definition.txt?path=bend-kit-time@0.3.0.0/time.bend&name=mono