{-# LANGUAGE ExistentialQuantification #-}
{-# LANGUAGE UnboxedSums #-}
{-# LANGUAGE UnboxedTuples #-}
{-# LANGUAGE PatternSynonyms #-}
{-# LANGUAGE MagicHash #-}
{-# LANGUAGE ViewPatterns #-}
-- | Reading the bytes of a TDS response, when those bytes arrive in pieces.
--
-- = The problem
--
-- A server's response is one long sequence of bytes, but it reaches the client as
-- a series of packets, and a packet can end anywhere: between two columns, in the
-- middle of an integer, or thousands of bytes into a long value.  A reader that
-- reaches the end of a packet has to stop, wait for the next one, and carry on
-- exactly where it was.  Most of the code that understands TDS shouldn't have to
-- know that any of this happens.
--
-- = Recipes
--
-- A /recipe/ describes what to read and in what order, written as ordinary monadic
-- code.  This one reads a length byte and then that many bytes:
--
-- > shortValue :: Parser B.ByteString
-- > shortValue = word8 >>= \n -> boundedBytes (fromIntegral n)
--
-- A recipe is only a value, so defining one reads nothing.  It says nothing about
-- packets either, so it gives the same answer wherever the packets happen to end.
-- 'parseChunks' runs a recipe over payloads already in memory, and returns the
-- result and the input left over:
--
-- > parseChunks shortValue ["\3abcdef"]         -- ("abc", ["def"])
-- > parseChunks shortValue ["\3a", "b", "cdef"] -- ("abc", ["def"])
--
-- = Suspending
--
-- When a recipe runs out of input partway through, it doesn't fail and it doesn't
-- block.  It returns a suspended computation, which holds everything it still has
-- to do, including what it has already read.  Whoever is running the recipe, called
-- its /driver/, fetches the next payload and gives it to that computation:
--
-- > case runParser shortValue "\3a" of
-- >   Right (Got value rest) -> ...            -- complete within this payload
-- >   Left suspended -> inspect suspended >>= \case
-- >     Right (NeedMore continue) -> ...       -- call continue (More nextPayload)
-- >     ...
--
-- This is what keeps the recipes simple.  Where a read has got to is simply where
-- the code is, so a recipe holds no record of its column or its position.  It is
-- also what keeps the module independent of the transport, because TLS, SMP
-- sessions and the packet framing all sit behind the driver.  Most callers should
-- use 'runFrom' or 'parseChunks' rather than write a driver, because a driver has a
-- rule to follow, which 'runFrom' describes.
--
-- = Writing your own
--
-- A recipe for something hmssql doesn't read yet needs only the reads below and
-- @do@ notation.  'fail' stops with a 'ProtocolError'.  For example, a token with a
-- 16-bit length around a list of strings:
--
-- > names :: Parser [Text]
-- > names = do
-- >   size <- word16le
-- >   isolate (fromIntegral size) $ do
-- >     n <- word8
-- >     count (fromIntegral n) bVarChar
--
-- 'chunk' is the most general read, because it gives whatever part of the requested
-- bytes is available, so any format can be built from it.
--
-- A recipe can fail in two ways, which mean different things.  Throwing, as 'fail'
-- does, ends the recipe, so nothing more can be read and the connection is broken.
-- Use it for bytes that make no sense.  A value that was read correctly but can't be
-- used is a 'failValue' instead, once its bytes have been read.  It skips the rest of
-- the recipe up to the nearest 'recover' or 'attempt', which goes on reading from where
-- the failure was.
--
-- This module only reads.  To send something hmssql has no type for, such as a
-- table-valued parameter, build a 'Hmssql.Query.Types.Request' whose body uses
-- "Hmssql.Tds.PacketWriter" and the builders of "Hmssql.Tds.Client".
module Hmssql.Tds.Parser
  ( -- * Recipes and their driver
    Parser, parse, runParser, runParserAt, parseChunks, runFrom
  , Got (..), At (..), Await (..), Feed (..), feedBytes, Runner (..), feedRunner
  , effect, consume, consumeWhole, dynamic, oneShotRecipe
    -- * Failed values
    -- | A value that was read but can't be used fails its recipe with 'failValue',
    -- and 'recover' or 'attempt' catches the failure further out, where the reading
    -- can go on.  A row decoder uses this so that a failed cell stops the row without
    -- every cell's result being an 'Either'.
  , failValue, recover, attempt
    -- * Read-only environments
    -- | A recipe can read a value that isn't in the input.  A row decoder uses this
    -- for the row's null bitmap, so that one recipe, prepared once, serves every row:
    --
    -- > plusBase :: Recipe Word8 Word8
    -- > plusBase = (+) <$> ask <*> liftParser word8
    -- >
    -- > parseChunks (withEnv 10 plusBase) ["\5"]    -- (15, [])
  , Recipe, ask, withEnv, liftParser
    -- * Primitive reads
  , peekToken, peekWord8
  , word8, word16le, word32le, word64le, word32be
  , bytes, ownedBytes, boundedBytes, ownedBoundedBytes
  , chunk, chunkOrDiscard, skip, count
  , bVarChar, usVarChar, isolate, limited
    -- * Value ownership and assembly
  , ValueBytes (..), valueBytes, keepBytes, bytesValue, convertBytes, prefixedValue, prefixedRowValue
  , Accumulator, accumulator, append, finish
    -- * Read failures
  , ReadFailure (..), DecodeError (..), ProtocolError (..), ResultAbandoned (..)
  ) where

import Control.Exception (Exception, throwIO, mask_, finally)
import Control.Monad (unless)
import Data.IORef
import qualified Streaming as S
import qualified Streaming.Prelude as S
import Streaming (Of (..))
import qualified Data.ByteString as B
import qualified Data.ByteString.Internal as BI
import qualified Data.ByteString.Unsafe as BU
import Data.Bits (shiftL, testBit, (.|.))
import Data.Text (Text)
import Hmssql.Internal.Derived
import qualified Hmssql.Tds.Utf16 as Utf16
import Data.Word (Word8, Word16, Word32, Word64)
import Foreign.ForeignPtr (ForeignPtr)
import Foreign.Marshal.Utils (copyBytes)
import Foreign.Ptr (plusPtr)
import Data.Primitive.SmallArray (SmallArray (..), smallArrayFromListN)
import GHC.Exts (Int (..), Int#, indexSmallArray#, int2Word#, isTrue#, ltWord#, oneShot, (+#), (>=#), (==#))
import Streaming (Stream, inspect)
import Streaming.Internal (Stream (..))
import qualified Hmssql.Tds.Bytes as Read

-- The shape of this module.
--
-- Where a read has got to is where the code is.  The continuation is ordinary
-- Haskell, so a value suspended at a packet boundary needs no phase to resume, an
-- interrupted drain needs no position, and a row cut by a packet is neither rewound
-- nor decoded twice.  A reader that is pushed chunks instead has to store all of
-- that by hand: a column, a phase, and a carry for a delimiter cut by a packet's
-- end.
--
-- Costs and alternatives, measured against the saved binaries in
-- docs/dev/value-streaming-research.md:
--
-- * A bind must be able to reach its continuation from the suspended branch, so
--   wherever GHC inlines a whole row, that continuation is used twice and becomes a
--   heap closure per bind. This is why row decoders are closures compiled per column
--   and not one inlined function: staging them as a selection and a read, so that a
--   row inlined whole, cost about 1.2 times as much in Decode and Mock.query.
-- * Continuation-passing style has the same capture on every read, including reads
--   that never suspend. hpgsql measured about 4.7% from leaving it, and a CPS parser
--   was measured and rejected here too.
-- * Two readings of one recipe, one giving up at the buffer's end with nothing
--   captured, made Recipe a record GHC would not fuse: Decode typed about 2 times as
--   slow, on top of the staging.
-- * Passing the input as the three unboxed fields of a 'B.ByteString', in 'Result'
--   and in 'Recipe''s argument, doubled a typed Mock.query's allocation (3.5 to
--   7.5 MB) and made it 1.7 times as slow.  The RTS has generic-apply entry points
--   only for arguments that are all pointers (stg_ap_p to stg_ap_pppppp, and the
--   pv, ppv and pppv forms) and for one lone unboxed word, so an unknown call whose
--   arguments mix pointers and unboxed words is split into chunks, and each chunk but
--   the last builds a PAP in the RTS, which ticky attributes to no closure.  An index
--   into a 'ByteArray#' with unboxed offsets would meet the same rule.  The payload
--   and a boxed position, below at 'Result', is what works.

-- | How a streaming consumer gets the next piece of its value.  'consume' gives the
-- consumer a stream, and each pull from that stream calls the 'Runner'.
--
-- The reference holds the suspended recipe that reads the value.  'runAwait' runs
-- that recipe until it produces its next result, fetching payloads whenever it needs
-- them, and it saves its progress back in the reference as it goes, so that an
-- interrupted pull doesn't repeat anything.
--
-- The driver supplies the 'Runner' because only the driver knows where the payloads
-- come from and what a pull must hold while it reads them.  The connection's own
-- 'Runner' masks each pull and reads from that response's packets.  'feedRunner' builds
-- one from a function that returns the next payload.
newtype Runner = Runner
  { Runner -> forall a. IORef (Stream Await IO (Got a)) -> IO (Got a)
runAwait :: forall a. IORef (Stream Await IO (Got a)) -> IO (Got a) }

-- | What a driver gives a suspended recipe: the next payload, with the transport and
-- the TDS packet headers already removed, or the end of the message.
data Feed
  = More !B.ByteString
    -- ^ The next payload, to read as usual.
  | Discard !B.ByteString
    -- ^ The next payload of a response being thrown away.  Nobody will use a value
    -- that is still being assembled, so its recipe may release its buffer and skip the
    -- rest of the value.  See 'chunkOrDiscard'.
  | EndOfMessage
    -- ^ There are no more payloads.  This is the normal way to finish at a token
    -- boundary ('peekToken'), and a 'ProtocolError' anywhere else.

-- | The payload, for a read that treats 'More' and 'Discard' alike, or 'Nothing' at
-- the end of the message.
feedBytes :: Feed -> Maybe B.ByteString
feedBytes :: Feed -> Maybe ByteString
feedBytes (More ByteString
bytes) = ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
bytes
feedBytes (Discard ByteString
bytes) = ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
bytes
feedBytes Feed
EndOfMessage = Maybe ByteString
forall a. Maybe a
Nothing

-- | A recipe's result, and the part of the current payload it didn't read.  The next
-- recipe starts with those bytes.
data Got a = Got !a {-# UNPACK #-} !B.ByteString

-- | Where a recipe that finished within its payload got to ('runParserAt'): its
-- value, or the failed value ('failValue') that nothing recovered, and in either
-- case the position after it in the same payload.
data At a = At !a {-# UNPACK #-} !Int | FailedAt !ReadFailure {-# UNPACK #-} !Int

-- | What a suspended recipe asks its driver for.  A suspended recipe is a 'Stream' of
-- these requests, so a driver sees three things as it runs one.  'Step' is one of these
-- requests, 'Effect' is a short piece of 'IO' that the driver just runs, such as growing
-- a value's buffer, and 'Return' is the final 'Got'.  Using 'inspect' from the
-- @streaming@ package runs the effects and gives the next request or the result.
data Await next
  = NeedMore (Feed -> next)
    -- ^ The recipe needs more input.  The driver fetches a payload and calls this
    -- function, which gives the rest of the recipe.
  | forall x. RunConsumer !Bool (Runner -> IO x) (Maybe x -> next)
    -- ^ The recipe has reached a value that a caller's code wants to read as it
    -- arrives.  The driver runs that code with a 'Runner' for its pulls, outside any
    -- mask the driver holds for reading, and gives its result to the function, which
    -- gives the rest of the recipe.  A driver that is throwing the response away
    -- gives 'Nothing' instead of running the code.  An exception from the code isn't
    -- given to the function: it ends the driver, as an exception of the caller's
    -- does.  The 'Bool' says whether the code pulls the value from the response
    -- ('consume'), so that it may wait for the server, or has the value in hand
    -- ('consumeWhole') and never pulls.

instance Functor Await where
  fmap :: forall a b. (a -> b) -> Await a -> Await b
fmap a -> b
f (NeedMore Feed -> a
k) = (Feed -> b) -> Await b
forall next. (Feed -> next) -> Await next
NeedMore (a -> b
f (a -> b) -> (Feed -> a) -> Feed -> b
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Feed -> a
k)
  fmap a -> b
f (RunConsumer Bool
live Runner -> IO x
io Maybe x -> a
k) = Bool -> (Runner -> IO x) -> (Maybe x -> b) -> Await b
forall next x.
Bool -> (Runner -> IO x) -> (Maybe x -> next) -> Await next
RunConsumer Bool
live Runner -> IO x
io (a -> b
f (a -> b) -> (Maybe x -> a) -> Maybe x -> b
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Maybe x -> a
k)

-- What running a recipe over a buffer gives: the value and the unread bytes, or a
-- suspended computation. A parse that completes within its buffer builds no Stream.
--
-- An unboxed sum returns completed reads without allocating a control value,
-- even when a dynamically selected recipe prevents inlining. Pattern synonyms
-- keep that representation private and preserve strict values and buffers.
-- Only suspended computations use Stream; its effects and binds handle resumption.
--
-- A failed value is a third alternative, which binds pass on without running their
-- continuations, so a recipe that succeeds builds no 'Either' per value.  A suspended
-- computation can fail too, so its result is an 'Outcome', and 'parse' and
-- 'runParser' turn it into the public 'Got' for drivers.  A failure that reaches them
-- is a bug in the recipe, which should have caught it with 'recover', so it is a
-- 'ProtocolError'.
--
-- 'Result', 'Recipe''s constructor and what is built on them ('run', 'parser', 'await',
-- 'resume', 'stream', 'primitive') stay private on purpose; design.md, "Reading and
-- writing bytes: which layer", gives the reasons.
--
-- A recipe runs on a payload and a position in it, and a completed read returns
-- where it got to, as a position in the same payload.  The payload is one pointer
-- shared by every recipe called within it, and the position is a boxed 'Int', so
-- that a recipe called through an unknown call takes three pointers (stg_ap_ppp)
-- and returns its position unboxed: 16 bytes for the 'Int' box per such call
-- instead of 32 for a 'B.ByteString' of the unread bytes, and nothing at all for a
-- call GHC can see, because the worker it makes takes an 'Int#'.  With the unread
-- bytes as the input, every column of every row returned a fresh 'B.ByteString'.
-- A payload changes only across a suspension, which runs through 'Outcome' and
-- starts the rest of the recipe at position 0 of the next payload; a driver gets
-- the unread bytes back as a 'B.ByteString' built once per 'parse' or 'runParser',
-- or the position itself from 'runParserAt'.
--
-- Measured on 2026-11-04 against the input as the unread bytes, per row by
-- tasty-bench over many iterations: a typed (Int32, Int64) row in Mock.query from
-- 366 to 286 bytes, the nullable NBCROW row from 1,123 to 804, a 'SqlRow' of the
-- same two columns from 544 to 528 (its cells are still unknown calls, and a
-- row's unread bytes cost one 'B.ByteString' where the last cell's used to serve),
-- and a typed row of 50 columns in Decode from 15,560 to 12,220.  The object code
-- of a two-field record's FromSqlRow instance went from 56.5 to 51.0 KB, and of
-- one for each field type by 9 to 12%.  Of the per-row saving, half is the smaller
-- box and half is 'liftA2' below, which turns a column's call into a known one.
type Result a = (# (# a, Int# #) | Stream Await IO (Outcome a) | (# ReadFailure, Int# #) #)

pattern Parsed :: a -> Int -> Result a
pattern $bParsed :: forall a. a -> Int -> Result a
$mParsed :: forall {r} {a}. Result a -> (a -> Int -> r) -> ((# #) -> r) -> r
Parsed a pos <- (# (# a, (I# -> pos) #) | | #)
  where Parsed !a
a (I# Int#
pos) = (# (# a
a, Int#
pos #) | | #)

pattern Suspended :: Stream Await IO (Outcome a) -> Result a
pattern $bSuspended :: forall a. Stream Await IO (Outcome a) -> Result a
$mSuspended :: forall {r} {a}.
Result a -> (Stream Await IO (Outcome a) -> r) -> ((# #) -> r) -> r
Suspended suspended = (# | suspended | #)

pattern Failed :: ReadFailure -> Int -> Result a
pattern $bFailed :: forall a. ReadFailure -> Int -> Result a
$mFailed :: forall {r} {a}.
Result a -> (ReadFailure -> Int -> r) -> ((# #) -> r) -> r
Failed failure pos <- (# | | (# failure, (I# -> pos) #) #)
  where Failed !ReadFailure
failure (I# Int#
pos) = (# | | (# ReadFailure
failure, Int#
pos #) #)
{-# COMPLETE Parsed, Suspended, Failed #-}

-- The bytes of the payload from a position on, which is what a driver gets back.
unread :: B.ByteString -> Int -> B.ByteString
unread :: ByteString -> Int -> ByteString
unread ByteString
bs Int
pos = Int -> ByteString -> ByteString
BU.unsafeDrop Int
pos ByteString
bs
{-# INLINE unread #-}

-- A slice of the payload at a position, whose bounds the caller has checked.
slice :: B.ByteString -> Int -> Int -> B.ByteString
slice :: ByteString -> Int -> Int -> ByteString
slice ByteString
bs Int
pos Int
n = Int -> ByteString -> ByteString
BU.unsafeTake Int
n (Int -> ByteString -> ByteString
BU.unsafeDrop Int
pos ByteString
bs)
{-# INLINE slice #-}

-- The bytes of the payload still unread from a position.
remaining :: B.ByteString -> Int -> Int
remaining :: ByteString -> Int -> Int
remaining ByteString
bs Int
pos = ByteString -> Int
B.length ByteString
bs Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
pos
{-# INLINE remaining #-}

-- How a suspended computation ends: with its value or a failed value, and the
-- unread part of the last payload.
data Outcome a = Done !a {-# UNPACK #-} !B.ByteString | Stopped !ReadFailure {-# UNPACK #-} !B.ByteString

-- | A recipe that can also read a value of type @env@ with 'ask', which 'withEnv'
-- supplies when the recipe runs.  It reads input and suspends just as a 'Parser' does.
newtype Recipe env a = Recipe { forall env a. Recipe env a -> env -> ByteString -> Int -> Result a
runRecipe :: env -> B.ByteString -> Int -> Result a }

-- | A recipe with no environment, which is the usual kind.
type Parser = Recipe ()

run :: Parser a -> B.ByteString -> Int -> Result a
run :: forall a. Parser a -> ByteString -> Int -> Result a
run Parser a
p = Parser a -> () -> ByteString -> Int -> Result a
forall env a. Recipe env a -> env -> ByteString -> Int -> Result a
runRecipe Parser a
p ()
{-# INLINE run #-}

parser :: (B.ByteString -> Int -> Result a) -> Parser a
parser :: forall a. (ByteString -> Int -> Result a) -> Parser a
parser ByteString -> Int -> Result a
p = (() -> ByteString -> Int -> Result a) -> Recipe () a
forall env a.
(env -> ByteString -> Int -> Result a) -> Recipe env a
Recipe (\() -> ByteString -> Int -> Result a
p)
{-# INLINE parser #-}

-- | The value given to 'withEnv'.  This reads no input.
ask :: Recipe env env
ask :: forall env. Recipe env env
ask = (env -> ByteString -> Int -> Result env) -> Recipe env env
forall env a.
(env -> ByteString -> Int -> Result a) -> Recipe env a
Recipe ((env -> ByteString -> Int -> Result env) -> Recipe env env)
-> (env -> ByteString -> Int -> Result env) -> Recipe env env
forall a b. (a -> b) -> a -> b
$ \env
env ByteString
_ Int
pos -> env -> Int -> Result env
forall a. a -> Int -> Result a
Parsed env
env Int
pos
{-# INLINE ask #-}

-- | Use a 'Parser', such as one of the reads below, inside a 'Recipe'.
liftParser :: Parser a -> Recipe env a
liftParser :: forall a env. Parser a -> Recipe env a
liftParser Parser a
p = (env -> ByteString -> Int -> Result a) -> Recipe env a
forall env a.
(env -> ByteString -> Int -> Result a) -> Recipe env a
Recipe ((env -> ByteString -> Int -> Result a) -> Recipe env a)
-> (env -> ByteString -> Int -> Result a) -> Recipe env a
forall a b. (a -> b) -> a -> b
$ \env
_ -> Parser a -> ByteString -> Int -> Result a
forall a. Parser a -> ByteString -> Int -> Result a
run Parser a
p
{-# INLINE liftParser #-}

-- | Supply the environment, which makes the recipe a 'Parser'.  The same recipe can be
-- used again with another value.
withEnv :: env -> Recipe env a -> Parser a
withEnv :: forall env a. env -> Recipe env a -> Parser a
withEnv env
env Recipe env a
p = (ByteString -> Int -> Result a) -> Parser a
forall a. (ByteString -> Int -> Result a) -> Parser a
parser (Recipe env a -> env -> ByteString -> Int -> Result a
forall env a. Recipe env a -> env -> ByteString -> Int -> Result a
runRecipe Recipe env a
p env
env)
{-# INLINE withEnv #-}

-- | A recipe chosen at run time, such as a column's prepared recipe, which GHC
-- therefore calls without seeing into it.  Wrapping it here costs nothing in the
-- recipe and saves the 16 bytes that the call would otherwise allocate for its
-- position.  Leave it off a recipe GHC can inline, because there the wrapper only
-- adds a table lookup.
dynamic :: Recipe env a -> Recipe env a
dynamic :: forall env a. Recipe env a -> Recipe env a
dynamic (Recipe env -> ByteString -> Int -> Result a
p) = (env -> ByteString -> Int -> Result a) -> Recipe env a
forall env a.
(env -> ByteString -> Int -> Result a) -> Recipe env a
Recipe ((env -> ByteString -> Int -> Result a) -> Recipe env a)
-> (env -> ByteString -> Int -> Result a) -> Recipe env a
forall a b. (a -> b) -> a -> b
$ \env
env ByteString
bs (I# Int#
pos) -> let !box :: Int
box = Int# -> Int
boxPosition Int#
pos in env -> ByteString -> Int -> Result a
p env
env ByteString
bs Int
box
{-# INLINE dynamic #-}

-- | The same recipe, for one that is built to be run once, such as a step of a
-- loop that builds the rest of a row's recipe as it reads.  GHC then takes the
-- recipe's input along with the function's own arguments, rather than build the
-- recipe as closures and thunks and run it after.  On a recipe that runs more than
-- once, such as a prepared one, the work before its input would be repeated for
-- every run.
oneShotRecipe :: Recipe env a -> Recipe env a
oneShotRecipe :: forall env a. Recipe env a -> Recipe env a
oneShotRecipe (Recipe env -> ByteString -> Int -> Result a
p) = (env -> ByteString -> Int -> Result a) -> Recipe env a
forall env a.
(env -> ByteString -> Int -> Result a) -> Recipe env a
Recipe ((env -> ByteString -> Int -> Result a)
-> env -> ByteString -> Int -> Result a
forall a b. (a -> b) -> a -> b
oneShot (\env
env -> (ByteString -> Int -> Result a) -> ByteString -> Int -> Result a
forall a b. (a -> b) -> a -> b
oneShot (\ByteString
bs -> (Int -> Result a) -> Int -> Result a
forall a b. (a -> b) -> a -> b
oneShot (\Int
pos -> env -> ByteString -> Int -> Result a
p env
env ByteString
bs Int
pos))))
{-# INLINE oneShotRecipe #-}

-- A position as a boxed 'Int' for a call GHC can't see into, from a table of boxes
-- built once.  The callers force the box before the call: an unknown call's argument
-- is lazy, so left to itself GHC passed a thunk of the lookup, 24 bytes in place of
-- the 16 saved.  A recipe takes its position boxed (see 'Result'), so each column of a
-- row called through an unknown call allocated a fresh box, 16 bytes, which for a
-- typed row of two columns was a tenth of the row's allocation and for a 'SqlRow'
-- cell a quarter of the cell's.  The table covers a payload of the default packet
-- size (8000 bytes); a position beyond it, in a connection with larger packets, is
-- boxed as before.  It is a CAF of about 197 KB on the heap (8,192 boxes and the
-- array) once anything reads a row.  Measured on 2026-11-04 by tasty-bench over
-- many iterations, per row: a typed row of 50 columns in Decode, whose driver is
-- unchanged, from 12,222 to 11,406 bytes, which is 51 boxes; a typed (Int32, Int64)
-- row in Mock.query from 366 to 284 together with 'runParserAt' and the driver
-- keeping its position, of which the two boxes are 32, the unread bytes 32 and the
-- row's 'Right' 16 by ticky.
boxPosition :: Int# -> Int
boxPosition :: Int# -> Int
boxPosition Int#
pos
  | Int# -> Bool
isTrue# (Int# -> Word#
int2Word# Int#
pos Word# -> Word# -> Int#
`ltWord#` Word#
8192##) = case SmallArray Int
positionBoxes of
      SmallArray SmallArray# Int
boxes -> case SmallArray# Int -> Int# -> (# Int #)
forall a. SmallArray# a -> Int# -> (# a #)
indexSmallArray# SmallArray# Int
boxes Int#
pos of (# Int
box #) -> Int
box
  | Bool
otherwise = Int# -> Int
I# Int#
pos
{-# INLINE boxPosition #-}

positionBoxes :: SmallArray Int
positionBoxes :: SmallArray Int
positionBoxes = Int -> [Int] -> SmallArray Int
forall a. Int -> [a] -> SmallArray a
smallArrayFromListN Int
8192 [Int
0 .. Int
8191]
{-# NOINLINE positionBoxes #-}

instance Functor (Recipe env) where
  fmap :: forall a b. (a -> b) -> Recipe env a -> Recipe env b
fmap a -> b
f (Recipe env -> ByteString -> Int -> Result a
p) = (env -> ByteString -> Int -> Result b) -> Recipe env b
forall env a.
(env -> ByteString -> Int -> Result a) -> Recipe env a
Recipe ((env -> ByteString -> Int -> Result b) -> Recipe env b)
-> (env -> ByteString -> Int -> Result b) -> Recipe env b
forall a b. (a -> b) -> a -> b
$ \env
env ByteString
bs Int
pos -> case env -> ByteString -> Int -> Result a
p env
env ByteString
bs Int
pos of
    Parsed a
a Int
end -> b -> Int -> Result b
forall a. a -> Int -> Result a
Parsed (a -> b
f a
a) Int
end
    Suspended Stream Await IO (Outcome a)
suspended -> Stream Await IO (Outcome a)
-> (a -> ByteString -> Result b) -> Result b
forall a b.
Stream Await IO (Outcome a)
-> (a -> ByteString -> Result b) -> Result b
resume Stream Await IO (Outcome a)
suspended (\a
a ByteString
_ -> b -> Int -> Result b
forall a. a -> Int -> Result a
Parsed (a -> b
f a
a) Int
0)
    Failed ReadFailure
failure Int
end -> ReadFailure -> Int -> Result b
forall a. ReadFailure -> Int -> Result a
Failed ReadFailure
failure Int
end
  {-# INLINE fmap #-}

instance Applicative (Recipe env) where
  pure :: forall a. a -> Recipe env a
pure a
a = (env -> ByteString -> Int -> Result a) -> Recipe env a
forall env a.
(env -> ByteString -> Int -> Result a) -> Recipe env a
Recipe ((env -> ByteString -> Int -> Result a) -> Recipe env a)
-> (env -> ByteString -> Int -> Result a) -> Recipe env a
forall a b. (a -> b) -> a -> b
$ \env
_ ByteString
_ Int
pos -> a -> Int -> Result a
forall a. a -> Int -> Result a
Parsed a
a Int
pos
  {-# INLINE pure #-}
  Recipe env -> ByteString -> Int -> Result (a -> b)
pf <*> :: forall a b. Recipe env (a -> b) -> Recipe env a -> Recipe env b
<*> Recipe env -> ByteString -> Int -> Result a
pa = (env -> ByteString -> Int -> Result b) -> Recipe env b
forall env a.
(env -> ByteString -> Int -> Result a) -> Recipe env a
Recipe ((env -> ByteString -> Int -> Result b) -> Recipe env b)
-> (env -> ByteString -> Int -> Result b) -> Recipe env b
forall a b. (a -> b) -> a -> b
$ \env
env ByteString
bs Int
pos -> case env -> ByteString -> Int -> Result (a -> b)
pf env
env ByteString
bs Int
pos of
    Parsed a -> b
f Int
next -> case env -> ByteString -> Int -> Result a
pa env
env ByteString
bs Int
next of
      Parsed a
a Int
end -> b -> Int -> Result b
forall a. a -> Int -> Result a
Parsed (a -> b
f a
a) Int
end
      Suspended Stream Await IO (Outcome a)
suspended -> Stream Await IO (Outcome a)
-> (a -> ByteString -> Result b) -> Result b
forall a b.
Stream Await IO (Outcome a)
-> (a -> ByteString -> Result b) -> Result b
resume Stream Await IO (Outcome a)
suspended (\a
a ByteString
_ -> b -> Int -> Result b
forall a. a -> Int -> Result a
Parsed (a -> b
f a
a) Int
0)
      Failed ReadFailure
failure Int
end -> ReadFailure -> Int -> Result b
forall a. ReadFailure -> Int -> Result a
Failed ReadFailure
failure Int
end
    Suspended Stream Await IO (Outcome (a -> b))
suspended -> Stream Await IO (Outcome (a -> b))
-> ((a -> b) -> ByteString -> Result b) -> Result b
forall a b.
Stream Await IO (Outcome a)
-> (a -> ByteString -> Result b) -> Result b
resume Stream Await IO (Outcome (a -> b))
suspended (\a -> b
f ByteString
rest -> Recipe env b -> env -> ByteString -> Int -> Result b
forall env a. Recipe env a -> env -> ByteString -> Int -> Result a
runRecipe ((a -> b) -> Recipe env a -> Recipe env b
forall a b. (a -> b) -> Recipe env a -> Recipe env b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap a -> b
f ((env -> ByteString -> Int -> Result a) -> Recipe env a
forall env a.
(env -> ByteString -> Int -> Result a) -> Recipe env a
Recipe env -> ByteString -> Int -> Result a
pa)) env
env ByteString
rest Int
0)
    Failed ReadFailure
failure Int
next -> ReadFailure -> Int -> Result b
forall a. ReadFailure -> Int -> Result a
Failed ReadFailure
failure Int
next
  {-# INLINE (<*>) #-}
  -- Written out, rather than the default '<*>' after 'fmap', for two reasons.  The
  -- default was a function of the library's that GHC didn't inline, so each
  -- column's recipe was called from it through an unknown call.  And where the
  -- function is unknown, as in a row's shared generic composition, it is applied
  -- to both values at once instead of building a partial application per value.
  liftA2 :: forall a b c.
(a -> b -> c) -> Recipe env a -> Recipe env b -> Recipe env c
liftA2 a -> b -> c
f (Recipe env -> ByteString -> Int -> Result a
pa) (Recipe env -> ByteString -> Int -> Result b
pb) = (env -> ByteString -> Int -> Result c) -> Recipe env c
forall env a.
(env -> ByteString -> Int -> Result a) -> Recipe env a
Recipe ((env -> ByteString -> Int -> Result c) -> Recipe env c)
-> (env -> ByteString -> Int -> Result c) -> Recipe env c
forall a b. (a -> b) -> a -> b
$ \env
env ByteString
bs Int
pos -> case env -> ByteString -> Int -> Result a
pa env
env ByteString
bs Int
pos of
    Parsed a
a Int
next -> case env -> ByteString -> Int -> Result b
pb env
env ByteString
bs Int
next of
      Parsed b
b Int
end -> c -> Int -> Result c
forall a. a -> Int -> Result a
Parsed (a -> b -> c
f a
a b
b) Int
end
      Suspended Stream Await IO (Outcome b)
suspended -> Stream Await IO (Outcome b)
-> (b -> ByteString -> Result c) -> Result c
forall a b.
Stream Await IO (Outcome a)
-> (a -> ByteString -> Result b) -> Result b
resume Stream Await IO (Outcome b)
suspended (\b
b ByteString
_ -> c -> Int -> Result c
forall a. a -> Int -> Result a
Parsed (a -> b -> c
f a
a b
b) Int
0)
      Failed ReadFailure
failure Int
end -> ReadFailure -> Int -> Result c
forall a. ReadFailure -> Int -> Result a
Failed ReadFailure
failure Int
end
    Suspended Stream Await IO (Outcome a)
suspended -> Stream Await IO (Outcome a)
-> (a -> ByteString -> Result c) -> Result c
forall a b.
Stream Await IO (Outcome a)
-> (a -> ByteString -> Result b) -> Result b
resume Stream Await IO (Outcome a)
suspended (\a
a ByteString
rest -> Recipe env c -> env -> ByteString -> Int -> Result c
forall env a. Recipe env a -> env -> ByteString -> Int -> Result a
runRecipe ((b -> c) -> Recipe env b -> Recipe env c
forall a b. (a -> b) -> Recipe env a -> Recipe env b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap (a -> b -> c
f a
a) ((env -> ByteString -> Int -> Result b) -> Recipe env b
forall env a.
(env -> ByteString -> Int -> Result a) -> Recipe env a
Recipe env -> ByteString -> Int -> Result b
pb)) env
env ByteString
rest Int
0)
    Failed ReadFailure
failure Int
next -> ReadFailure -> Int -> Result c
forall a. ReadFailure -> Int -> Result a
Failed ReadFailure
failure Int
next
  {-# INLINE liftA2 #-}

instance Monad (Recipe env) where
  Recipe env -> ByteString -> Int -> Result a
p >>= :: forall a b. Recipe env a -> (a -> Recipe env b) -> Recipe env b
>>= a -> Recipe env b
k = (env -> ByteString -> Int -> Result b) -> Recipe env b
forall env a.
(env -> ByteString -> Int -> Result a) -> Recipe env a
Recipe ((env -> ByteString -> Int -> Result b) -> Recipe env b)
-> (env -> ByteString -> Int -> Result b) -> Recipe env b
forall a b. (a -> b) -> a -> b
$ \env
env ByteString
bs Int
pos -> case env -> ByteString -> Int -> Result a
p env
env ByteString
bs Int
pos of
    Parsed a
a Int
next -> Recipe env b -> env -> ByteString -> Int -> Result b
forall env a. Recipe env a -> env -> ByteString -> Int -> Result a
runRecipe (a -> Recipe env b
k a
a) env
env ByteString
bs Int
next
    Suspended Stream Await IO (Outcome a)
suspended -> Stream Await IO (Outcome a)
-> (a -> ByteString -> Result b) -> Result b
forall a b.
Stream Await IO (Outcome a)
-> (a -> ByteString -> Result b) -> Result b
resume Stream Await IO (Outcome a)
suspended (\a
a ByteString
rest -> Recipe env b -> env -> ByteString -> Int -> Result b
forall env a. Recipe env a -> env -> ByteString -> Int -> Result a
runRecipe (a -> Recipe env b
k a
a) env
env ByteString
rest Int
0)
    Failed ReadFailure
failure Int
next -> ReadFailure -> Int -> Result b
forall a. ReadFailure -> Int -> Result a
Failed ReadFailure
failure Int
next
  {-# INLINE (>>=) #-}

-- Keep the recursive Stream traversal off the ordinary, completed path.
-- A suspension traverses the enclosing bind depth, unlike an immediate result:
-- Stream's bind rebuilds each layer through every enclosing bind. That is O(depth)
-- per packet boundary, and the depth is about a row's columns, since long values loop
-- inside 'chunk' and 'skip' rather than through a bind per chunk. NOINLINE because
-- inlining this into each bind would copy the traversal to every read.  The
-- continuation gets the unread part of the last payload, and its result is a
-- position in it.
resume :: Stream Await IO (Outcome a) -> (a -> B.ByteString -> Result b) -> Result b
resume :: forall a b.
Stream Await IO (Outcome a)
-> (a -> ByteString -> Result b) -> Result b
resume Stream Await IO (Outcome a)
suspended a -> ByteString -> Result b
k = Stream Await IO (Outcome b) -> Result b
forall a. Stream Await IO (Outcome a) -> Result a
Suspended (Stream Await IO (Outcome b) -> Result b)
-> Stream Await IO (Outcome b) -> Result b
forall a b. (a -> b) -> a -> b
$ Stream Await IO (Outcome a)
suspended Stream Await IO (Outcome a)
-> (Outcome a -> Stream Await IO (Outcome b))
-> Stream Await IO (Outcome b)
forall a b.
Stream Await IO a -> (a -> Stream Await IO b) -> Stream Await IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
  Done a
a ByteString
rest -> ByteString -> Result b -> Stream Await IO (Outcome b)
forall a. ByteString -> Result a -> Stream Await IO (Outcome a)
stream ByteString
rest (a -> ByteString -> Result b
k a
a ByteString
rest)
  Stopped ReadFailure
failure ByteString
rest -> Outcome b -> Stream Await IO (Outcome b)
forall (f :: * -> *) (m :: * -> *) r. r -> Stream f m r
Return (ReadFailure -> ByteString -> Outcome b
forall a. ReadFailure -> ByteString -> Outcome a
Stopped ReadFailure
failure ByteString
rest)
{-# NOINLINE resume #-}

-- | Fail with a value that was read but can't be used.  The bytes after it are left
-- for whatever 'recover' runs next.
failValue :: ReadFailure -> Recipe env a
failValue :: forall env a. ReadFailure -> Recipe env a
failValue ReadFailure
failure = (env -> ByteString -> Int -> Result a) -> Recipe env a
forall env a.
(env -> ByteString -> Int -> Result a) -> Recipe env a
Recipe ((env -> ByteString -> Int -> Result a) -> Recipe env a)
-> (env -> ByteString -> Int -> Result a) -> Recipe env a
forall a b. (a -> b) -> a -> b
$ \env
_ ByteString
_ Int
pos -> ReadFailure -> Int -> Result a
forall a. ReadFailure -> Int -> Result a
Failed ReadFailure
failure Int
pos
{-# INLINE failValue #-}

-- | Run a recipe, and if a value in it fails ('failValue'), run the handler from
-- where the failure left the input.
recover :: Recipe env a -> (ReadFailure -> Recipe env a) -> Recipe env a
recover :: forall env a.
Recipe env a -> (ReadFailure -> Recipe env a) -> Recipe env a
recover (Recipe env -> ByteString -> Int -> Result a
p) ReadFailure -> Recipe env a
handler = (env -> ByteString -> Int -> Result a) -> Recipe env a
forall env a.
(env -> ByteString -> Int -> Result a) -> Recipe env a
Recipe ((env -> ByteString -> Int -> Result a) -> Recipe env a)
-> (env -> ByteString -> Int -> Result a) -> Recipe env a
forall a b. (a -> b) -> a -> b
$ \env
env ByteString
bs Int
pos -> case env -> ByteString -> Int -> Result a
p env
env ByteString
bs Int
pos of
  Parsed a
a Int
end -> a -> Int -> Result a
forall a. a -> Int -> Result a
Parsed a
a Int
end
  Suspended Stream Await IO (Outcome a)
suspended -> Stream Await IO (Outcome a)
-> (ReadFailure -> ByteString -> Result a) -> Result a
forall a.
Stream Await IO (Outcome a)
-> (ReadFailure -> ByteString -> Result a) -> Result a
recoverSuspended Stream Await IO (Outcome a)
suspended (\ReadFailure
failure ByteString
rest -> Recipe env a -> env -> ByteString -> Int -> Result a
forall env a. Recipe env a -> env -> ByteString -> Int -> Result a
runRecipe (ReadFailure -> Recipe env a
handler ReadFailure
failure) env
env ByteString
rest Int
0)
  Failed ReadFailure
failure Int
end -> Recipe env a -> env -> ByteString -> Int -> Result a
forall env a. Recipe env a -> env -> ByteString -> Int -> Result a
runRecipe (ReadFailure -> Recipe env a
handler ReadFailure
failure) env
env ByteString
bs Int
end
{-# INLINE recover #-}

-- Off the completed path, as 'resume' is.
recoverSuspended :: Stream Await IO (Outcome a) -> (ReadFailure -> B.ByteString -> Result a) -> Result a
recoverSuspended :: forall a.
Stream Await IO (Outcome a)
-> (ReadFailure -> ByteString -> Result a) -> Result a
recoverSuspended Stream Await IO (Outcome a)
suspended ReadFailure -> ByteString -> Result a
handler = Stream Await IO (Outcome a) -> Result a
forall a. Stream Await IO (Outcome a) -> Result a
Suspended (Stream Await IO (Outcome a) -> Result a)
-> Stream Await IO (Outcome a) -> Result a
forall a b. (a -> b) -> a -> b
$ Stream Await IO (Outcome a)
suspended Stream Await IO (Outcome a)
-> (Outcome a -> Stream Await IO (Outcome a))
-> Stream Await IO (Outcome a)
forall a b.
Stream Await IO a -> (a -> Stream Await IO b) -> Stream Await IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
  done :: Outcome a
done@(Done a
_ ByteString
_) -> Outcome a -> Stream Await IO (Outcome a)
forall (f :: * -> *) (m :: * -> *) r. r -> Stream f m r
Return Outcome a
done
  Stopped ReadFailure
failure ByteString
rest -> ByteString -> Result a -> Stream Await IO (Outcome a)
forall a. ByteString -> Result a -> Stream Await IO (Outcome a)
stream ByteString
rest (ReadFailure -> ByteString -> Result a
handler ReadFailure
failure ByteString
rest)
{-# NOINLINE recoverSuspended #-}

-- | Run a recipe, and give its failed value, if it has one, as a 'Left'.
attempt :: Recipe env a -> Recipe env (Either ReadFailure a)
attempt :: forall env a. Recipe env a -> Recipe env (Either ReadFailure a)
attempt Recipe env a
p = Recipe env (Either ReadFailure a)
-> (ReadFailure -> Recipe env (Either ReadFailure a))
-> Recipe env (Either ReadFailure a)
forall env a.
Recipe env a -> (ReadFailure -> Recipe env a) -> Recipe env a
recover (a -> Either ReadFailure a
forall a b. b -> Either a b
Right (a -> Either ReadFailure a)
-> Recipe env a -> Recipe env (Either ReadFailure a)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Recipe env a
p) (Either ReadFailure a -> Recipe env (Either ReadFailure a)
forall a. a -> Recipe env a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either ReadFailure a -> Recipe env (Either ReadFailure a))
-> (ReadFailure -> Either ReadFailure a)
-> ReadFailure
-> Recipe env (Either ReadFailure a)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ReadFailure -> Either ReadFailure a
forall a b. a -> Either a b
Left)
{-# INLINE attempt #-}

instance MonadFail (Recipe env) where
  fail :: forall a. String -> Recipe env a
fail String
msg = Parser a -> Recipe env a
forall a env. Parser a -> Recipe env a
liftParser (IO a -> Parser a
forall a. IO a -> Parser a
effect (ProtocolError -> IO a
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO (String -> ProtocolError
ProtocolError String
msg)))

-- | Run a short piece of 'IO' as part of a recipe, and evaluate its result.  It runs
-- once, under whatever mask the driver holds, so it must not block.  Code that can
-- block, such as a caller's consumer, belongs in 'consume' instead.
effect :: IO a -> Parser a
effect :: forall a. IO a -> Parser a
effect IO a
io = (ByteString -> Int -> Result a) -> Parser a
forall a. (ByteString -> Int -> Result a) -> Parser a
parser ((ByteString -> Int -> Result a) -> Parser a)
-> (ByteString -> Int -> Result a) -> Parser a
forall a b. (a -> b) -> a -> b
$ \ByteString
bs Int
pos -> Stream Await IO (Outcome a) -> Result a
forall a. Stream Await IO (Outcome a) -> Result a
Suspended (Stream Await IO (Outcome a) -> Result a)
-> Stream Await IO (Outcome a) -> Result a
forall a b. (a -> b) -> a -> b
$ IO (Stream Await IO (Outcome a)) -> Stream Await IO (Outcome a)
forall (f :: * -> *) (m :: * -> *) r.
m (Stream f m r) -> Stream f m r
Effect (IO (Stream Await IO (Outcome a)) -> Stream Await IO (Outcome a))
-> IO (Stream Await IO (Outcome a)) -> Stream Await IO (Outcome a)
forall a b. (a -> b) -> a -> b
$ do
  !a <- IO a
io
  pure (Return (Done a (unread bs pos)))

-- Suspend until more bytes arrive, then run the recipe from the new buffer. An
-- empty payload (a packet with only a header) waits again. Reads reached here are
-- inside a token, so a message that ends now is malformed; 'peekToken' is the one read
-- for which the end is a result.
await :: Parser a -> Result a
await :: forall a. Parser a -> Result a
await Parser a
p = Stream Await IO (Outcome a) -> Result a
forall a. Stream Await IO (Outcome a) -> Result a
Suspended (Stream Await IO (Outcome a) -> Result a)
-> Stream Await IO (Outcome a) -> Result a
forall a b. (a -> b) -> a -> b
$ Await (Stream Await IO (Outcome a)) -> Stream Await IO (Outcome a)
forall (f :: * -> *) (m :: * -> *) r.
f (Stream f m r) -> Stream f m r
Step (Await (Stream Await IO (Outcome a))
 -> Stream Await IO (Outcome a))
-> Await (Stream Await IO (Outcome a))
-> Stream Await IO (Outcome a)
forall a b. (a -> b) -> a -> b
$ (Feed -> Stream Await IO (Outcome a))
-> Await (Stream Await IO (Outcome a))
forall next. (Feed -> next) -> Await next
NeedMore ((Feed -> Stream Await IO (Outcome a))
 -> Await (Stream Await IO (Outcome a)))
-> (Feed -> Stream Await IO (Outcome a))
-> Await (Stream Await IO (Outcome a))
forall a b. (a -> b) -> a -> b
$ \Feed
feed -> case Feed -> Maybe ByteString
feedBytes Feed
feed of
  Just ByteString
bs | ByteString -> Bool
B.null ByteString
bs -> ByteString -> Result a -> Stream Await IO (Outcome a)
forall a. ByteString -> Result a -> Stream Await IO (Outcome a)
stream ByteString
bs (Parser a -> Result a
forall a. Parser a -> Result a
await Parser a
p)
          | Bool
otherwise -> ByteString -> Result a -> Stream Await IO (Outcome a)
forall a. ByteString -> Result a -> Stream Await IO (Outcome a)
stream ByteString
bs (Parser a -> ByteString -> Int -> Result a
forall a. Parser a -> ByteString -> Int -> Result a
run Parser a
p ByteString
bs Int
0)
  Maybe ByteString
Nothing -> ByteString -> Result a -> Stream Await IO (Outcome a)
forall a. ByteString -> Result a -> Stream Await IO (Outcome a)
stream ByteString
B.empty (Parser a -> ByteString -> Int -> Result a
forall a. Parser a -> ByteString -> Int -> Result a
run (String -> Parser a
forall a. String -> Recipe () a
forall (m :: * -> *) a. MonadFail m => String -> m a
fail String
"message ended inside a token") ByteString
B.empty Int
0)

------------------------------------------------------------------------------
-- Primitive reads

-- | The next byte, which is left unread, or 'Nothing' if the message has ended.  This
-- is how to look for the next token, because it is the one read for which the end of
-- the message isn't an error.
peekToken :: Parser (Maybe Word8)
peekToken :: Parser (Maybe Word8)
peekToken = (ByteString -> Int -> Result (Maybe Word8)) -> Parser (Maybe Word8)
forall a. (ByteString -> Int -> Result a) -> Parser a
parser ((ByteString -> Int -> Result (Maybe Word8))
 -> Parser (Maybe Word8))
-> (ByteString -> Int -> Result (Maybe Word8))
-> Parser (Maybe Word8)
forall a b. (a -> b) -> a -> b
$ \ByteString
bs Int
pos ->
  if Int
pos Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< ByteString -> Int
B.length ByteString
bs then Maybe Word8 -> Int -> Result (Maybe Word8)
forall a. a -> Int -> Result a
Parsed (Word8 -> Maybe Word8
forall a. a -> Maybe a
Just (ByteString -> Int -> Word8
Read.word8At ByteString
bs Int
pos)) Int
pos
  else Stream Await IO (Outcome (Maybe Word8)) -> Result (Maybe Word8)
forall a. Stream Await IO (Outcome a) -> Result a
Suspended (Stream Await IO (Outcome (Maybe Word8)) -> Result (Maybe Word8))
-> Stream Await IO (Outcome (Maybe Word8)) -> Result (Maybe Word8)
forall a b. (a -> b) -> a -> b
$ Await (Stream Await IO (Outcome (Maybe Word8)))
-> Stream Await IO (Outcome (Maybe Word8))
forall (f :: * -> *) (m :: * -> *) r.
f (Stream f m r) -> Stream f m r
Step (Await (Stream Await IO (Outcome (Maybe Word8)))
 -> Stream Await IO (Outcome (Maybe Word8)))
-> Await (Stream Await IO (Outcome (Maybe Word8)))
-> Stream Await IO (Outcome (Maybe Word8))
forall a b. (a -> b) -> a -> b
$ (Feed -> Stream Await IO (Outcome (Maybe Word8)))
-> Await (Stream Await IO (Outcome (Maybe Word8)))
forall next. (Feed -> next) -> Await next
NeedMore ((Feed -> Stream Await IO (Outcome (Maybe Word8)))
 -> Await (Stream Await IO (Outcome (Maybe Word8))))
-> (Feed -> Stream Await IO (Outcome (Maybe Word8)))
-> Await (Stream Await IO (Outcome (Maybe Word8)))
forall a b. (a -> b) -> a -> b
$ \Feed
feed -> case Feed -> Maybe ByteString
feedBytes Feed
feed of
    Maybe ByteString
Nothing -> Outcome (Maybe Word8) -> Stream Await IO (Outcome (Maybe Word8))
forall a. a -> Stream Await IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Maybe Word8 -> ByteString -> Outcome (Maybe Word8)
forall a. a -> ByteString -> Outcome a
Done Maybe Word8
forall a. Maybe a
Nothing ByteString
B.empty)
    Just ByteString
next -> ByteString
-> Result (Maybe Word8) -> Stream Await IO (Outcome (Maybe Word8))
forall a. ByteString -> Result a -> Stream Await IO (Outcome a)
stream ByteString
next (Parser (Maybe Word8) -> ByteString -> Int -> Result (Maybe Word8)
forall a. Parser a -> ByteString -> Int -> Result a
run Parser (Maybe Word8)
peekToken ByteString
next Int
0)

-- | The next byte, which is left unread.  The end of the message is a 'ProtocolError'.
peekWord8 :: Parser Word8
peekWord8 :: Parser Word8
peekWord8 = (ByteString -> Int -> Result Word8) -> Parser Word8
forall a. (ByteString -> Int -> Result a) -> Parser a
parser ((ByteString -> Int -> Result Word8) -> Parser Word8)
-> (ByteString -> Int -> Result Word8) -> Parser Word8
forall a b. (a -> b) -> a -> b
$ \ByteString
bs Int
pos -> if Int
pos Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>= ByteString -> Int
B.length ByteString
bs then Parser Word8 -> Result Word8
forall a. Parser a -> Result a
await Parser Word8
peekWord8 else Word8 -> Int -> Result Word8
forall a. a -> Int -> Result a
Parsed (ByteString -> Int -> Word8
Read.word8At ByteString
bs Int
pos) Int
pos

-- A fixed-width read. Only the bounds check and the load are inlined at each use;
-- a word cut by a packet boundary is joined by the out-of-line slow path, which
-- would otherwise be copied into every field of every decoder.
primitive :: Int -> (B.ByteString -> Int -> a) -> Parser a
primitive :: forall a. Int -> (ByteString -> Int -> a) -> Parser a
primitive Int
n ByteString -> Int -> a
read = (ByteString -> Int -> Result a) -> Parser a
forall a. (ByteString -> Int -> Result a) -> Parser a
parser ((ByteString -> Int -> Result a) -> Parser a)
-> (ByteString -> Int -> Result a) -> Parser a
forall a b. (a -> b) -> a -> b
$ \ByteString
bs Int
pos ->
  if ByteString -> Int -> Int
remaining ByteString
bs Int
pos Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>= Int
n
    then a -> Int -> Result a
forall a. a -> Int -> Result a
Parsed (ByteString -> Int -> a
read ByteString
bs Int
pos) (Int
pos Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
n)
    else Int -> (ByteString -> Int -> a) -> ByteString -> Int -> Result a
forall a.
Int -> (ByteString -> Int -> a) -> ByteString -> Int -> Result a
primitiveSlow Int
n ByteString -> Int -> a
read ByteString
bs Int
pos
{-# INLINE primitive #-}

primitiveSlow :: Int -> (B.ByteString -> Int -> a) -> B.ByteString -> Int -> Result a
primitiveSlow :: forall a.
Int -> (ByteString -> Int -> a) -> ByteString -> Int -> Result a
primitiveSlow Int
n ByteString -> Int -> a
read = Parser a -> ByteString -> Int -> Result a
forall a. Parser a -> ByteString -> Int -> Result a
run ((ByteString -> a) -> Recipe () ByteString -> Parser a
forall a b. (a -> b) -> Recipe () a -> Recipe () b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap (\ByteString
bytes -> ByteString -> Int -> a
read ByteString
bytes Int
0) (Int -> Recipe () ByteString
boundedBytes Int
n))
{-# NOINLINE primitiveSlow #-}

-- | One byte.
word8 :: Parser Word8
word8 :: Parser Word8
word8 = Int -> (ByteString -> Int -> Word8) -> Parser Word8
forall a. Int -> (ByteString -> Int -> a) -> Parser a
primitive Int
1 ByteString -> Int -> Word8
Read.word8At
{-# INLINE word8 #-}
-- | A little-endian 16-bit integer, as most integers in TDS are.
word16le :: Parser Word16
word16le :: Parser Word16
word16le = Int -> (ByteString -> Int -> Word16) -> Parser Word16
forall a. Int -> (ByteString -> Int -> a) -> Parser a
primitive Int
2 ByteString -> Int -> Word16
Read.word16LeAt
{-# INLINE word16le #-}
-- | A little-endian 32-bit integer.
word32le :: Parser Word32
word32le :: Parser Word32
word32le = Int -> (ByteString -> Int -> Word32) -> Parser Word32
forall a. Int -> (ByteString -> Int -> a) -> Parser a
primitive Int
4 ByteString -> Int -> Word32
Read.word32LeAt
{-# INLINE word32le #-}
-- | A little-endian 64-bit integer.
word64le :: Parser Word64
word64le :: Parser Word64
word64le = Int -> (ByteString -> Int -> Word64) -> Parser Word64
forall a. Int -> (ByteString -> Int -> a) -> Parser a
primitive Int
8 ByteString -> Int -> Word64
Read.word64LeAt
{-# INLINE word64le #-}

-- | A big-endian 32-bit integer.
word32be :: Parser Word32
word32be :: Parser Word32
word32be = (Word32 -> Word8 -> Word32) -> Word32 -> ByteString -> Word32
forall a. (a -> Word8 -> a) -> a -> ByteString -> a
B.foldl' (\Word32
a Word8
b -> (Word32
a Word32 -> Int -> Word32
forall a. Bits a => a -> Int -> a
`shiftL` Int
8) Word32 -> Word32 -> Word32
forall a. Bits a => a -> a -> a
.|. Word8 -> Word32
forall a b. (Integral a, Num b) => a -> b
fromIntegral Word8
b) Word32
0 (ByteString -> Word32) -> Recipe () ByteString -> Parser Word32
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Int -> Recipe () ByteString
boundedBytes Int
4

-- | Exactly this many bytes (the second argument), in one 'B.ByteString'.  If they lie
-- within the current payload, the result is a slice of it and nothing is copied.
-- Otherwise they are copied into a buffer of their own as they arrive.
--
-- The first argument is how much may be allocated before the bytes have arrived, as
-- 'append' describes.  It protects against a server that declares a huge length.
--
-- A slice keeps its whole packet alive for as long as it is held, so use 'ownedBytes'
-- for anything kept after the current token.
bytes :: Int -> Int -> Parser B.ByteString
bytes :: Int -> Int -> Recipe () ByteString
bytes Int
ahead Int
n = (ByteString -> ByteString) -> Int -> Int -> Recipe () ByteString
contiguous ByteString -> ByteString
forall a. a -> a
id Int
ahead Int
n
{-# INLINE bytes #-}

-- | Like 'bytes', but the result is always cheap to keep, because a slice is copied.
ownedBytes :: Int -> Int -> Parser B.ByteString
ownedBytes :: Int -> Int -> Recipe () ByteString
ownedBytes Int
ahead Int
n = (ByteString -> ByteString) -> Int -> Int -> Recipe () ByteString
contiguous ByteString -> ByteString
B.copy Int
ahead Int
n
{-# INLINE ownedBytes #-}

-- | 'bytes' that may allocate the whole length ahead.  This is for short metadata whose
-- length field is small enough to bound what a server can ask for, such as the length
-- byte of a B_VARCHAR.
boundedBytes :: Int -> Parser B.ByteString
boundedBytes :: Int -> Recipe () ByteString
boundedBytes Int
n = Int -> Int -> Recipe () ByteString
bytes Int
n Int
n

-- | 'ownedBytes' that may allocate the whole length ahead, as 'boundedBytes' may.
ownedBoundedBytes :: Int -> Parser B.ByteString
ownedBoundedBytes :: Int -> Recipe () ByteString
ownedBoundedBytes Int
n = Int -> Int -> Recipe () ByteString
ownedBytes Int
n Int
n

-- The slow path gathers through 'chunk', never 'chunkOrDiscard': metadata has to be
-- parsed even in a drain, so it can't be given up on as a value can.
contiguous :: (B.ByteString -> B.ByteString) -> Int -> Int -> Parser B.ByteString
contiguous :: (ByteString -> ByteString) -> Int -> Int -> Recipe () ByteString
contiguous ByteString -> ByteString
borrowed Int
ahead Int
n = (ByteString -> Int -> Result ByteString) -> Recipe () ByteString
forall a. (ByteString -> Int -> Result a) -> Parser a
parser ((ByteString -> Int -> Result ByteString) -> Recipe () ByteString)
-> (ByteString -> Int -> Result ByteString) -> Recipe () ByteString
forall a b. (a -> b) -> a -> b
$ \ByteString
bs Int
pos ->
  if Int
n Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
0 then Recipe () ByteString -> ByteString -> Int -> Result ByteString
forall a. Parser a -> ByteString -> Int -> Result a
run (String -> Recipe () ByteString
forall a. String -> Recipe () a
forall (m :: * -> *) a. MonadFail m => String -> m a
fail String
"negative byte count") ByteString
bs Int
pos
  else if Int
n Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= ByteString -> Int -> Int
remaining ByteString
bs Int
pos
    then ByteString -> Int -> Result ByteString
forall a. a -> Int -> Result a
Parsed (ByteString -> ByteString
borrowed (ByteString -> Int -> Int -> ByteString
slice ByteString
bs Int
pos Int
n)) (Int
pos Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
n)
    else Int -> Int -> ByteString -> Int -> Result ByteString
contiguousSlow Int
ahead Int
n ByteString
bs Int
pos
{-# INLINE contiguous #-}

contiguousSlow :: Int -> Int -> B.ByteString -> Int -> Result B.ByteString
contiguousSlow :: Int -> Int -> ByteString -> Int -> Result ByteString
contiguousSlow Int
ahead Int
n = Recipe () ByteString -> ByteString -> Int -> Result ByteString
forall a. Parser a -> ByteString -> Int -> Result a
run (Int -> Accumulator -> Recipe () ByteString
gather Int
n (Maybe Word64 -> Accumulator
accumulator (Word64 -> Maybe Word64
forall a. a -> Maybe a
Just (Int -> Word64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
n))))
  where
    gather :: Int -> Accumulator -> Recipe () ByteString
gather Int
0 Accumulator
a = IO ByteString -> Recipe () ByteString
forall a. IO a -> Parser a
effect (Accumulator -> IO ByteString
finish Accumulator
a)
    gather Int
remaining Accumulator
a = do
      bytes <- Int -> Recipe () ByteString
chunk Int
remaining
      next <- effect (append ahead a bytes)
      gather (remaining - B.length bytes) next
{-# NOINLINE contiguousSlow #-}

-- | Between 1 and this many bytes, as a slice of the current payload, waiting for a
-- payload if there are none left.  Asking for zero gives an empty result without
-- needing input.  Loop to read more than one payload holds, and copy what you keep.
chunk :: Int -> Parser B.ByteString
chunk :: Int -> Recipe () ByteString
chunk Int
n = (ByteString -> Int -> Result ByteString) -> Recipe () ByteString
forall a. (ByteString -> Int -> Result a) -> Parser a
parser ((ByteString -> Int -> Result ByteString) -> Recipe () ByteString)
-> (ByteString -> Int -> Result ByteString) -> Recipe () ByteString
forall a b. (a -> b) -> a -> b
$ \ByteString
bs Int
pos ->
  if Int
n Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
0 then Recipe () ByteString -> ByteString -> Int -> Result ByteString
forall a. Parser a -> ByteString -> Int -> Result a
run (String -> Recipe () ByteString
forall a. String -> Recipe () a
forall (m :: * -> *) a. MonadFail m => String -> m a
fail String
"negative byte count") ByteString
bs Int
pos
  else if Int
n Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0 then ByteString -> Int -> Result ByteString
forall a. a -> Int -> Result a
Parsed ByteString
B.empty Int
pos
  else if Int
pos Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>= ByteString -> Int
B.length ByteString
bs then Recipe () ByteString -> Result ByteString
forall a. Parser a -> Result a
await (Int -> Recipe () ByteString
chunk Int
n)
  else let k :: Int
k = Int -> Int -> Int
forall a. Ord a => a -> a -> a
min Int
n (ByteString -> Int -> Int
remaining ByteString
bs Int
pos)
       in ByteString -> Int -> Result ByteString
forall a. a -> Int -> Result a
Parsed (ByteString -> Int -> Int -> ByteString
slice ByteString
bs Int
pos Int
k) (Int
pos Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
k)

-- | 'chunk' for a recipe that collects a value.  It gives 'Nothing' if the response
-- started being thrown away ('Discard') while this read was waiting for input.  The
-- recipe should then release what it has collected and 'skip' the rest of the value,
-- because 'Nothing' reads no bytes.
chunkOrDiscard :: Int -> Parser (Maybe B.ByteString)
chunkOrDiscard :: Int -> Parser (Maybe ByteString)
chunkOrDiscard Int
n = (ByteString -> Int -> Result (Maybe ByteString))
-> Parser (Maybe ByteString)
forall a. (ByteString -> Int -> Result a) -> Parser a
parser ((ByteString -> Int -> Result (Maybe ByteString))
 -> Parser (Maybe ByteString))
-> (ByteString -> Int -> Result (Maybe ByteString))
-> Parser (Maybe ByteString)
forall a b. (a -> b) -> a -> b
$ \ByteString
bs Int
pos ->
  if Int
n Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= Int
0 Bool -> Bool -> Bool
|| Int
pos Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< ByteString -> Int
B.length ByteString
bs then Parser (Maybe ByteString)
-> ByteString -> Int -> Result (Maybe ByteString)
forall a. Parser a -> ByteString -> Int -> Result a
run (ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just (ByteString -> Maybe ByteString)
-> Recipe () ByteString -> Parser (Maybe ByteString)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Int -> Recipe () ByteString
chunk Int
n) ByteString
bs Int
pos
  else Stream Await IO (Outcome (Maybe ByteString))
-> Result (Maybe ByteString)
forall a. Stream Await IO (Outcome a) -> Result a
Suspended (Stream Await IO (Outcome (Maybe ByteString))
 -> Result (Maybe ByteString))
-> Stream Await IO (Outcome (Maybe ByteString))
-> Result (Maybe ByteString)
forall a b. (a -> b) -> a -> b
$ Await (Stream Await IO (Outcome (Maybe ByteString)))
-> Stream Await IO (Outcome (Maybe ByteString))
forall (f :: * -> *) (m :: * -> *) r.
f (Stream f m r) -> Stream f m r
Step (Await (Stream Await IO (Outcome (Maybe ByteString)))
 -> Stream Await IO (Outcome (Maybe ByteString)))
-> Await (Stream Await IO (Outcome (Maybe ByteString)))
-> Stream Await IO (Outcome (Maybe ByteString))
forall a b. (a -> b) -> a -> b
$ (Feed -> Stream Await IO (Outcome (Maybe ByteString)))
-> Await (Stream Await IO (Outcome (Maybe ByteString)))
forall next. (Feed -> next) -> Await next
NeedMore ((Feed -> Stream Await IO (Outcome (Maybe ByteString)))
 -> Await (Stream Await IO (Outcome (Maybe ByteString))))
-> (Feed -> Stream Await IO (Outcome (Maybe ByteString)))
-> Await (Stream Await IO (Outcome (Maybe ByteString)))
forall a b. (a -> b) -> a -> b
$ \Feed
feed -> case Feed
feed of
    Discard ByteString
next -> Outcome (Maybe ByteString)
-> Stream Await IO (Outcome (Maybe ByteString))
forall a. a -> Stream Await IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Maybe ByteString -> ByteString -> Outcome (Maybe ByteString)
forall a. a -> ByteString -> Outcome a
Done Maybe ByteString
forall a. Maybe a
Nothing ByteString
next)
    More ByteString
next -> ByteString
-> Result (Maybe ByteString)
-> Stream Await IO (Outcome (Maybe ByteString))
forall a. ByteString -> Result a -> Stream Await IO (Outcome a)
stream ByteString
next (Parser (Maybe ByteString)
-> ByteString -> Int -> Result (Maybe ByteString)
forall a. Parser a -> ByteString -> Int -> Result a
run (Int -> Parser (Maybe ByteString)
chunkOrDiscard Int
n) ByteString
next Int
0)
    Feed
EndOfMessage -> ByteString
-> Result (Maybe ByteString)
-> Stream Await IO (Outcome (Maybe ByteString))
forall a. ByteString -> Result a -> Stream Await IO (Outcome a)
stream ByteString
B.empty (Parser (Maybe ByteString)
-> ByteString -> Int -> Result (Maybe ByteString)
forall a. Parser a -> ByteString -> Int -> Result a
run (String -> Parser (Maybe ByteString)
forall a. String -> Recipe () a
forall (m :: * -> *) a. MonadFail m => String -> m a
fail String
"message ended inside a value") ByteString
B.empty Int
0)

-- | Pass over this many bytes, across any number of payloads, keeping none of them.
skip :: Int -> Parser ()
skip :: Int -> Parser ()
skip Int
n = (ByteString -> Int -> Result ()) -> Parser ()
forall a. (ByteString -> Int -> Result a) -> Parser a
parser ((ByteString -> Int -> Result ()) -> Parser ())
-> (ByteString -> Int -> Result ()) -> Parser ()
forall a b. (a -> b) -> a -> b
$ \ByteString
bs Int
pos ->
  if Int
n Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
0 then Parser () -> ByteString -> Int -> Result ()
forall a. Parser a -> ByteString -> Int -> Result a
run (String -> Parser ()
forall a. String -> Recipe () a
forall (m :: * -> *) a. MonadFail m => String -> m a
fail String
"negative byte count") ByteString
bs Int
pos
  else if Int
n Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= ByteString -> Int -> Int
remaining ByteString
bs Int
pos then () -> Int -> Result ()
forall a. a -> Int -> Result a
Parsed () (Int
pos Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
n)
  else Parser () -> Result ()
forall a. Parser a -> Result a
await (Int -> Parser ()
skip (Int
n Int -> Int -> Int
forall a. Num a => a -> a -> a
- ByteString -> Int -> Int
remaining ByteString
bs Int
pos))

-- | A B_VARCHAR: a string of UTF-16LE with a one-byte length in characters.
bVarChar :: Parser Text
bVarChar :: Parser Text
bVarChar = Parser Word8
word8 Parser Word8 -> (Word8 -> Parser Text) -> Parser Text
forall a b. Recipe () a -> (a -> Recipe () b) -> Recipe () b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \Word8
n -> ByteString -> Text
Utf16.decodeUtf16LE (ByteString -> Text) -> Recipe () ByteString -> Parser Text
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Int -> Recipe () ByteString
boundedBytes (Int
2 Int -> Int -> Int
forall a. Num a => a -> a -> a
* Word8 -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral Word8
n)

-- | A US_VARCHAR: a string of UTF-16LE with a two-byte length in characters.
usVarChar :: Parser Text
usVarChar :: Parser Text
usVarChar = Parser Word16
word16le Parser Word16 -> (Word16 -> Parser Text) -> Parser Text
forall a b. Recipe () a -> (a -> Recipe () b) -> Recipe () b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \Word16
n -> ByteString -> Text
Utf16.decodeUtf16LE (ByteString -> Text) -> Recipe () ByteString -> Parser Text
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Int -> Recipe () ByteString
boundedBytes (Int
2 Int -> Int -> Int
forall a. Num a => a -> a -> a
* Word16 -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral Word16
n)

-- | Run a recipe this many times and collect the results in order.  A negative count
-- is a 'ProtocolError'.
count :: Int -> Parser a -> Parser [a]
count :: forall a. Int -> Parser a -> Parser [a]
count Int
n Parser a
recipe
  | Int
n Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
0 = String -> Recipe () [a]
forall a. String -> Recipe () a
forall (m :: * -> *) a. MonadFail m => String -> m a
fail String
"negative item count"
  | Bool
otherwise = Int -> [a] -> Recipe () [a]
forall {t}. (Eq t, Num t) => t -> [a] -> Recipe () [a]
go Int
n []
  where
    go :: t -> [a] -> Recipe () [a]
go t
0 [a]
acc = [a] -> Recipe () [a]
forall a. a -> Recipe () a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([a] -> [a]
forall a. [a] -> [a]
reverse [a]
acc)
    go t
remaining [a]
acc = Parser a
recipe Parser a -> (a -> Recipe () [a]) -> Recipe () [a]
forall a b. Recipe () a -> (a -> Recipe () b) -> Recipe () b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \a
a -> t -> [a] -> Recipe () [a]
go (t
remaining t -> t -> t
forall a. Num a => a -> a -> a
- t
1) (a
a a -> [a] -> [a]
forall a. a -> [a] -> [a]
: [a]
acc)

-- | Run a recipe on exactly the next @n@ bytes, for a token that declares its own
-- length.  The recipe can't read past them, and it is a 'ProtocolError' if it leaves
-- any unread.  The bytes are collected first, so this suits short tokens rather than
-- anything large or anything repeated per row.  Compare 'limited'.
isolate :: Int -> Parser a -> Parser a
isolate :: forall a. Int -> Parser a -> Parser a
isolate Int
n Parser a
p = do
  bs <- Int -> Recipe () ByteString
boundedBytes Int
n
  effect $ do
    (a, rest) <- parseChunks p [bs]
    if all B.null rest then pure a else throwIO (ProtocolError "unconsumed bytes in token envelope")

-- | Run a recipe, with a 'ProtocolError' if it receives more than this many bytes
-- before it finishes.  This is for a token with no overall length of its own, so that
-- the lengths inside it can't make the reader wait or collect without end.  Unlike
-- 'isolate', it collects nothing first, and the recipe may finish early.
limited :: Int -> Parser a -> Parser a
limited :: forall a. Int -> Parser a -> Parser a
limited Int
limit Parser a
p = (ByteString -> Int -> Result a) -> Parser a
forall a. (ByteString -> Int -> Result a) -> Parser a
parser ((ByteString -> Int -> Result a) -> Parser a)
-> (ByteString -> Int -> Result a) -> Parser a
forall a b. (a -> b) -> a -> b
$ \ByteString
bs Int
pos -> case Parser a -> ByteString -> Int -> Result a
forall a. Parser a -> ByteString -> Int -> Result a
run Parser a
p ByteString
bs Int
pos of
  Parsed a
a Int
end
    | Int
end Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
pos Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= Int
limit -> a -> Int -> Result a
forall a. a -> Int -> Result a
Parsed a
a Int
end
    | Bool
otherwise -> Parser a -> ByteString -> Int -> Result a
forall a. Parser a -> ByteString -> Int -> Result a
run (String -> Parser a
forall a. String -> Recipe () a
forall (m :: * -> *) a. MonadFail m => String -> m a
fail String
"metadata token exceeds its size limit") ByteString
bs Int
pos
  Suspended Stream Await IO (Outcome a)
s -> Stream Await IO (Outcome a) -> Result a
forall a. Stream Await IO (Outcome a) -> Result a
Suspended (Int -> Stream Await IO (Outcome a) -> Stream Await IO (Outcome a)
forall {a}.
Int -> Stream Await IO (Outcome a) -> Stream Await IO (Outcome a)
bound (ByteString -> Int -> Int
remaining ByteString
bs Int
pos) Stream Await IO (Outcome a)
s)
  Failed ReadFailure
failure Int
end -> ReadFailure -> Int -> Result a
forall a. ReadFailure -> Int -> Result a
Failed ReadFailure
failure Int
end
  where
    bound :: Int -> Stream Await IO (Outcome a) -> Stream Await IO (Outcome a)
bound Int
received Stream Await IO (Outcome a)
s = IO (Stream Await IO (Outcome a)) -> Stream Await IO (Outcome a)
forall (f :: * -> *) (m :: * -> *) r.
m (Stream f m r) -> Stream f m r
Effect (IO (Stream Await IO (Outcome a)) -> Stream Await IO (Outcome a))
-> IO (Stream Await IO (Outcome a)) -> Stream Await IO (Outcome a)
forall a b. (a -> b) -> a -> b
$ Stream Await IO (Outcome a)
-> IO (Either (Outcome a) (Await (Stream Await IO (Outcome a))))
forall (m :: * -> *) (f :: * -> *) r.
Monad m =>
Stream f m r -> m (Either r (f (Stream f m r)))
inspect Stream Await IO (Outcome a)
s IO (Either (Outcome a) (Await (Stream Await IO (Outcome a))))
-> (Either (Outcome a) (Await (Stream Await IO (Outcome a)))
    -> IO (Stream Await IO (Outcome a)))
-> IO (Stream Await IO (Outcome a))
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
      Left Outcome a
outcome
        | Int
received Int -> Int -> Int
forall a. Num a => a -> a -> a
- ByteString -> Int
B.length (Outcome a -> ByteString
forall a. Outcome a -> ByteString
outcomeRest Outcome a
outcome) Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= Int
limit -> Stream Await IO (Outcome a) -> IO (Stream Await IO (Outcome a))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Outcome a -> Stream Await IO (Outcome a)
forall (f :: * -> *) (m :: * -> *) r. r -> Stream f m r
Return Outcome a
outcome)
        | Bool
otherwise -> ProtocolError -> IO (Stream Await IO (Outcome a))
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO (String -> ProtocolError
ProtocolError (String
"token still unfinished after " String -> String -> String
forall a. [a] -> [a] -> [a]
++ Int -> String
forall a. Show a => a -> String
show Int
received String -> String -> String
forall a. [a] -> [a] -> [a]
++ String
" bytes"))
      Right (NeedMore Feed -> Stream Await IO (Outcome a)
k)
        | Int
received Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
limit -> ProtocolError -> IO (Stream Await IO (Outcome a))
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO (String -> ProtocolError
ProtocolError (String
"token still unfinished after " String -> String -> String
forall a. [a] -> [a] -> [a]
++ Int -> String
forall a. Show a => a -> String
show Int
received String -> String -> String
forall a. [a] -> [a] -> [a]
++ String
" bytes"))
        | Bool
otherwise -> Stream Await IO (Outcome a) -> IO (Stream Await IO (Outcome a))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Stream Await IO (Outcome a) -> IO (Stream Await IO (Outcome a)))
-> Stream Await IO (Outcome a) -> IO (Stream Await IO (Outcome a))
forall a b. (a -> b) -> a -> b
$ Await (Stream Await IO (Outcome a)) -> Stream Await IO (Outcome a)
forall (f :: * -> *) (m :: * -> *) r.
f (Stream f m r) -> Stream f m r
Step (Await (Stream Await IO (Outcome a))
 -> Stream Await IO (Outcome a))
-> Await (Stream Await IO (Outcome a))
-> Stream Await IO (Outcome a)
forall a b. (a -> b) -> a -> b
$ (Feed -> Stream Await IO (Outcome a))
-> Await (Stream Await IO (Outcome a))
forall next. (Feed -> next) -> Await next
NeedMore ((Feed -> Stream Await IO (Outcome a))
 -> Await (Stream Await IO (Outcome a)))
-> (Feed -> Stream Await IO (Outcome a))
-> Await (Stream Await IO (Outcome a))
forall a b. (a -> b) -> a -> b
$ \Feed
feed ->
            Int -> Stream Await IO (Outcome a) -> Stream Await IO (Outcome a)
bound (Int
received Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int -> (ByteString -> Int) -> Maybe ByteString -> Int
forall b a. b -> (a -> b) -> Maybe a -> b
maybe Int
0 ByteString -> Int
B.length (Feed -> Maybe ByteString
feedBytes Feed
feed)) (Feed -> Stream Await IO (Outcome a)
k Feed
feed)
      Right (RunConsumer Bool
live Runner -> IO x
io Maybe x -> Stream Await IO (Outcome a)
k) -> Stream Await IO (Outcome a) -> IO (Stream Await IO (Outcome a))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Stream Await IO (Outcome a) -> IO (Stream Await IO (Outcome a)))
-> Stream Await IO (Outcome a) -> IO (Stream Await IO (Outcome a))
forall a b. (a -> b) -> a -> b
$ Await (Stream Await IO (Outcome a)) -> Stream Await IO (Outcome a)
forall (f :: * -> *) (m :: * -> *) r.
f (Stream f m r) -> Stream f m r
Step (Await (Stream Await IO (Outcome a))
 -> Stream Await IO (Outcome a))
-> Await (Stream Await IO (Outcome a))
-> Stream Await IO (Outcome a)
forall a b. (a -> b) -> a -> b
$ Bool
-> (Runner -> IO x)
-> (Maybe x -> Stream Await IO (Outcome a))
-> Await (Stream Await IO (Outcome a))
forall next x.
Bool -> (Runner -> IO x) -> (Maybe x -> next) -> Await next
RunConsumer Bool
live Runner -> IO x
io (Int -> Stream Await IO (Outcome a) -> Stream Await IO (Outcome a)
bound Int
received (Stream Await IO (Outcome a) -> Stream Await IO (Outcome a))
-> (Maybe x -> Stream Await IO (Outcome a))
-> Maybe x
-> Stream Await IO (Outcome a)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Maybe x -> Stream Await IO (Outcome a)
k)

------------------------------------------------------------------------------
-- Value ownership and assembly

-- | A value's bytes, and whether they are cheap to keep.  'Borrowed' bytes are a slice
-- of a receive buffer.  Nothing overwrites them, but holding them keeps the whole buffer
-- alive.  'Owned' bytes are in a buffer of their own.  The difference lets a decoder
-- that keeps bytes copy only when it has to, and lets one that only looks at them, such
-- as an integer's, never copy.
data ValueBytes = Borrowed !B.ByteString | Owned !B.ByteString

-- | The bytes, to look at now.  Keeping them may keep a receive buffer alive.
valueBytes :: ValueBytes -> B.ByteString
valueBytes :: ValueBytes -> ByteString
valueBytes (Borrowed ByteString
bs) = ByteString
bs
valueBytes (Owned ByteString
bs) = ByteString
bs
{-# INLINE valueBytes #-}

-- | The bytes, cheap to keep, because 'Borrowed' bytes are copied.
keepBytes :: ValueBytes -> B.ByteString
keepBytes :: ValueBytes -> ByteString
keepBytes (Borrowed ByteString
bs) = ByteString -> ByteString
B.copy ByteString
bs
keepBytes (Owned ByteString
bs) = ByteString
bs
{-# INLINE keepBytes #-}

-- | A value of a known length, given the allocation bound of 'bytes' and the length.
-- 'Nothing' means that the response started being thrown away while the value was
-- being assembled.  The partial value has been released and its remaining bytes
-- skipped, so the recipe can carry on with what follows.
bytesValue :: Int -> Int -> Parser (Maybe ValueBytes)
bytesValue :: Int -> Int -> Parser (Maybe ValueBytes)
bytesValue Int
ahead Int
n = (ByteString -> Int -> Result (Maybe ValueBytes))
-> Parser (Maybe ValueBytes)
forall a. (ByteString -> Int -> Result a) -> Parser a
parser ((ByteString -> Int -> Result (Maybe ValueBytes))
 -> Parser (Maybe ValueBytes))
-> (ByteString -> Int -> Result (Maybe ValueBytes))
-> Parser (Maybe ValueBytes)
forall a b. (a -> b) -> a -> b
$ \ByteString
bs Int
pos ->
  if Int
n Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
0 then Parser (Maybe ValueBytes)
-> ByteString -> Int -> Result (Maybe ValueBytes)
forall a. Parser a -> ByteString -> Int -> Result a
run (String -> Parser (Maybe ValueBytes)
forall a. String -> Recipe () a
forall (m :: * -> *) a. MonadFail m => String -> m a
fail String
"negative byte count") ByteString
bs Int
pos
  else if Int
n Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= ByteString -> Int -> Int
remaining ByteString
bs Int
pos then Maybe ValueBytes -> Int -> Result (Maybe ValueBytes)
forall a. a -> Int -> Result a
Parsed (ValueBytes -> Maybe ValueBytes
forall a. a -> Maybe a
Just (ByteString -> ValueBytes
Borrowed (ByteString -> Int -> Int -> ByteString
slice ByteString
bs Int
pos Int
n))) (Int
pos Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
n)
  else Int -> Int -> ByteString -> Int -> Result (Maybe ValueBytes)
bytesValueSlow Int
ahead Int
n ByteString
bs Int
pos
{-# INLINE bytesValue #-}

-- | Read a value of a known length, as 'bytesValue' does, and convert it.  A value
-- that lies within the current payload goes to the first conversion as a slice of it,
-- and one that crossed payloads goes to the second, assembled in a buffer of its own.
-- A 'Left' from either is a failed value ('failValue') after the bytes, and a value
-- that a drain threw away is 'ReadDiscarded'.
convertBytes :: Int -> Int -> (B.ByteString -> Either ReadFailure a)
  -> (B.ByteString -> Either ReadFailure a) -> Parser a
-- Nearly every cell is a value lying in the buffer.  Given to the conversion here, it
-- needs no 'Maybe', 'ValueBytes' or boxed slice: bound through 'bytesValue', those
-- reached the conversion because its continuation was shared with the suspended
-- branch, and a 'SqlRow' cell built 80 bytes of them, by ticky.
convertBytes :: forall a.
Int
-> Int
-> (ByteString -> Either ReadFailure a)
-> (ByteString -> Either ReadFailure a)
-> Parser a
convertBytes Int
ahead Int
n ByteString -> Either ReadFailure a
borrowed ByteString -> Either ReadFailure a
owned = (ByteString -> Int -> Result a) -> Parser a
forall a. (ByteString -> Int -> Result a) -> Parser a
parser ((ByteString -> Int -> Result a) -> Parser a)
-> (ByteString -> Int -> Result a) -> Parser a
forall a b. (a -> b) -> a -> b
$ \ByteString
bs Int
pos ->
  if Int
n Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>= Int
0 Bool -> Bool -> Bool
&& Int
n Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= ByteString -> Int -> Int
remaining ByteString
bs Int
pos
    then case ByteString -> Either ReadFailure a
borrowed (ByteString -> Int -> Int -> ByteString
slice ByteString
bs Int
pos Int
n) of
           Right a
a -> a -> Int -> Result a
forall a. a -> Int -> Result a
Parsed a
a (Int
pos Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
n)
           Left ReadFailure
failure -> ReadFailure -> Int -> Result a
forall a. ReadFailure -> Int -> Result a
Failed ReadFailure
failure (Int
pos Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
n)
    else Int
-> Int
-> (ByteString -> Either ReadFailure a)
-> (ByteString -> Either ReadFailure a)
-> ByteString
-> Int
-> Result a
forall a.
Int
-> Int
-> (ByteString -> Either ReadFailure a)
-> (ByteString -> Either ReadFailure a)
-> ByteString
-> Int
-> Result a
convertBytesSlow Int
ahead Int
n ByteString -> Either ReadFailure a
borrowed ByteString -> Either ReadFailure a
owned ByteString
bs Int
pos
{-# INLINE convertBytes #-}

-- | A value that lies in the payload, converted as 'convertBytes' converts one, or
-- else another recipe. Its delimiter is a length prefix of one or two bytes,
-- little-endian, with a value that means NULL, or no prefix and a fixed length: the
-- delimiter arguments are the prefix's width and the NULL value, or 0 and the length.
-- A NULL runs the first recipe after the prefix. A value that doesn't lie wholly in
-- the payload, or any other width, runs the second from the delimiter, which it must
-- read itself.  The environment is passed to both recipes unread.
prefixedValue :: Int -> Int -> (B.ByteString -> Either ReadFailure a) -> Recipe env a -> Recipe env a -> Recipe env a
prefixedValue :: forall a env.
Int
-> Int
-> (ByteString -> Either ReadFailure a)
-> Recipe env a
-> Recipe env a
-> Recipe env a
prefixedValue Int
width Int
marker ByteString -> Either ReadFailure a
convert (Recipe env -> ByteString -> Int -> Result a
null) (Recipe env -> ByteString -> Int -> Result a
other) = (env -> ByteString -> Int -> Result a) -> Recipe env a
forall env a.
(env -> ByteString -> Int -> Result a) -> Recipe env a
Recipe ((env -> ByteString -> Int -> Result a) -> Recipe env a)
-> (env -> ByteString -> Int -> Result a) -> Recipe env a
forall a b. (a -> b) -> a -> b
$ \env
env ByteString
bs Int
pos ->
  Int
-> ByteString
-> Int
-> Int
-> (ByteString -> Either ReadFailure a)
-> (env -> ByteString -> Int -> Result a)
-> (env -> ByteString -> Int -> Result a)
-> env
-> ByteString
-> Int
-> Result a
forall a env.
Int
-> ByteString
-> Int
-> Int
-> (ByteString -> Either ReadFailure a)
-> (env -> ByteString -> Int -> Result a)
-> (env -> ByteString -> Int -> Result a)
-> env
-> ByteString
-> Int
-> Result a
prefixedAt Int
0 ByteString
B.empty Int
width Int
marker ByteString -> Either ReadFailure a
convert env -> ByteString -> Int -> Result a
null env -> ByteString -> Int -> Result a
other env
env ByteString
bs Int
pos
{-# INLINE prefixedValue #-}

-- | 'prefixedValue' for a column of a row, whose environment is the NBCROW bitmap
-- (empty for a ROW token) and whose first argument is the column's index from
-- zero.  A column the bitmap marks NULL has no delimiter, and runs the NULL recipe
-- at once.
prefixedRowValue :: Int -> Int -> Int -> (B.ByteString -> Either ReadFailure a) -> Recipe B.ByteString a -> Recipe B.ByteString a -> Recipe B.ByteString a
prefixedRowValue :: forall a.
Int
-> Int
-> Int
-> (ByteString -> Either ReadFailure a)
-> Recipe ByteString a
-> Recipe ByteString a
-> Recipe ByteString a
prefixedRowValue Int
column Int
width Int
marker ByteString -> Either ReadFailure a
convert (Recipe ByteString -> ByteString -> Int -> Result a
null) (Recipe ByteString -> ByteString -> Int -> Result a
other) = (ByteString -> ByteString -> Int -> Result a)
-> Recipe ByteString a
forall env a.
(env -> ByteString -> Int -> Result a) -> Recipe env a
Recipe ((ByteString -> ByteString -> Int -> Result a)
 -> Recipe ByteString a)
-> (ByteString -> ByteString -> Int -> Result a)
-> Recipe ByteString a
forall a b. (a -> b) -> a -> b
$ \ByteString
bitmap ByteString
bs Int
pos ->
  Int
-> ByteString
-> Int
-> Int
-> (ByteString -> Either ReadFailure a)
-> (ByteString -> ByteString -> Int -> Result a)
-> (ByteString -> ByteString -> Int -> Result a)
-> ByteString
-> ByteString
-> Int
-> Result a
forall a env.
Int
-> ByteString
-> Int
-> Int
-> (ByteString -> Either ReadFailure a)
-> (env -> ByteString -> Int -> Result a)
-> (env -> ByteString -> Int -> Result a)
-> env
-> ByteString
-> Int
-> Result a
prefixedAt Int
column ByteString
bitmap Int
width Int
marker ByteString -> Either ReadFailure a
convert ByteString -> ByteString -> Int -> Result a
null ByteString -> ByteString -> Int -> Result a
other ByteString
bitmap ByteString
bs Int
pos
{-# INLINE prefixedRowValue #-}

-- The dispatch of both entry points, on the unwrapped recipe functions, so that
-- neither builds a 'Recipe' to run per value.  Inlined into each, a built-in
-- conversion still has one call site, given the slice's fields unboxed.
prefixedAt :: Int -> B.ByteString -> Int -> Int
  -> (B.ByteString -> Either ReadFailure a)
  -> (env -> B.ByteString -> Int -> Result a)
  -> (env -> B.ByteString -> Int -> Result a)
  -> env -> B.ByteString -> Int -> Result a
prefixedAt :: forall a env.
Int
-> ByteString
-> Int
-> Int
-> (ByteString -> Either ReadFailure a)
-> (env -> ByteString -> Int -> Result a)
-> (env -> ByteString -> Int -> Result a)
-> env
-> ByteString
-> Int
-> Result a
prefixedAt Int
column ByteString
bitmap Int
width Int
marker ByteString -> Either ReadFailure a
convert env -> ByteString -> Int -> Result a
null env -> ByteString -> Int -> Result a
other env
env ByteString
bs Int
pos =
  case Int
-> ByteString
-> Int
-> Int
-> ByteString
-> Int
-> (# Int#, Int# #)
valueSpan Int
column ByteString
bitmap Int
width Int
marker ByteString
bs Int
pos of
    (# Int#
start, Int#
n #)
      | Int# -> Bool
isTrue# (Int#
n Int# -> Int# -> Int#
>=# Int#
0#) -> case ByteString -> Either ReadFailure a
convert (ByteString -> Int -> Int -> ByteString
slice ByteString
bs (Int# -> Int
I# Int#
start) (Int# -> Int
I# Int#
n)) of
          Right a
a -> a -> Int -> Result a
forall a. a -> Int -> Result a
Parsed a
a (Int# -> Int
I# (Int#
start Int# -> Int# -> Int#
+# Int#
n))
          Left ReadFailure
failure -> ReadFailure -> Int -> Result a
forall a. ReadFailure -> Int -> Result a
Failed ReadFailure
failure (Int# -> Int
I# (Int#
start Int# -> Int# -> Int#
+# Int#
n))
      | Int# -> Bool
isTrue# (Int#
n Int# -> Int# -> Int#
==# Int#
-1#) -> env -> ByteString -> Int -> Result a
null env
env ByteString
bs (Int# -> Int
I# Int#
start)
      | Bool
otherwise -> env -> ByteString -> Int -> Result a
other env
env ByteString
bs Int
pos
{-# INLINE prefixedAt #-}

-- Where a prefixed value starts and its length, or -1 for a NULL (and where its
-- prefix ends), or -2 for anything else.  The bitmap test is here, out of line,
-- rather than in each built-in decoder: inlined there it cost 22,688 bytes of
-- code and 56 bytes of recipe setup per column, for the same allocation per row
-- (GHC 9.12.4, 2026-11-04).
valueSpan :: Int -> B.ByteString -> Int -> Int -> B.ByteString -> Int -> (# Int#, Int# #)
valueSpan :: Int
-> ByteString
-> Int
-> Int
-> ByteString
-> Int
-> (# Int#, Int# #)
valueSpan Int
column ByteString
bitmap Int
width Int
marker ByteString
bs Int
pos
  | Bool -> Bool
not (ByteString -> Bool
B.null ByteString
bitmap) Bool -> Bool -> Bool
&& Word8 -> Int -> Bool
forall a. Bits a => a -> Int -> Bool
testBit (HasCallStack => ByteString -> Int -> Word8
ByteString -> Int -> Word8
B.index ByteString
bitmap (Int
column Int -> Int -> Int
forall a. Integral a => a -> a -> a
`div` Int
8)) (Int
column Int -> Int -> Int
forall a. Integral a => a -> a -> a
`mod` Int
8) = Int -> Int -> (# Int#, Int# #)
found Int
pos (-Int
1)
  | Bool
otherwise = case Int
width of
      Int
0 | Int
marker Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= ByteString -> Int -> Int
remaining ByteString
bs Int
pos -> Int -> Int -> (# Int#, Int# #)
found Int
pos Int
marker
      Int
1 | ByteString -> Int -> Int
remaining ByteString
bs Int
pos Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>= Int
1 -> Int -> Int -> (# Int#, Int# #)
prefixed Int
1 (Word8 -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral (ByteString -> Int -> Word8
Read.word8At ByteString
bs Int
pos))
      Int
2 | ByteString -> Int -> Int
remaining ByteString
bs Int
pos Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>= Int
2 -> Int -> Int -> (# Int#, Int# #)
prefixed Int
2 (Word16 -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral (ByteString -> Int -> Word16
Read.word16LeAt ByteString
bs Int
pos))
      Int
_ -> (# Int#, Int# #)
other
  where
    prefixed :: Int -> Int -> (# Int#, Int# #)
prefixed Int
w Int
n
      | Int
n Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
marker = Int -> Int -> (# Int#, Int# #)
found (Int
pos Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
w) (-Int
1)
      | Int
n Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= ByteString -> Int -> Int
remaining ByteString
bs (Int
pos Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
w) = Int -> Int -> (# Int#, Int# #)
found (Int
pos Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
w) Int
n
      | Bool
otherwise = (# Int#, Int# #)
other
    found :: Int -> Int -> (# Int#, Int# #)
found (I# Int#
start) (I# Int#
n) = (# Int#
start, Int#
n #)
    other :: (# Int#, Int# #)
other = (# Int#
0#, Int#
-2# #)
{-# NOINLINE valueSpan #-}

convertBytesSlow :: Int -> Int -> (B.ByteString -> Either ReadFailure a)
  -> (B.ByteString -> Either ReadFailure a) -> B.ByteString -> Int -> Result a
convertBytesSlow :: forall a.
Int
-> Int
-> (ByteString -> Either ReadFailure a)
-> (ByteString -> Either ReadFailure a)
-> ByteString
-> Int
-> Result a
convertBytesSlow Int
ahead Int
n ByteString -> Either ReadFailure a
borrowed ByteString -> Either ReadFailure a
owned = Parser a -> ByteString -> Int -> Result a
forall a. Parser a -> ByteString -> Int -> Result a
run (Parser a -> ByteString -> Int -> Result a)
-> Parser a -> ByteString -> Int -> Result a
forall a b. (a -> b) -> a -> b
$ Int -> Int -> Parser (Maybe ValueBytes)
bytesValue Int
ahead Int
n Parser (Maybe ValueBytes)
-> (Maybe ValueBytes -> Parser a) -> Parser a
forall a b. Recipe () a -> (a -> Recipe () b) -> Recipe () b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
  Just (Borrowed ByteString
bytes) -> (ReadFailure -> Parser a)
-> (a -> Parser a) -> Either ReadFailure a -> Parser a
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either ReadFailure -> Parser a
forall env a. ReadFailure -> Recipe env a
failValue a -> Parser a
forall a. a -> Recipe () a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ByteString -> Either ReadFailure a
borrowed ByteString
bytes)
  Just (Owned ByteString
bytes) -> (ReadFailure -> Parser a)
-> (a -> Parser a) -> Either ReadFailure a -> Parser a
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either ReadFailure -> Parser a
forall env a. ReadFailure -> Recipe env a
failValue a -> Parser a
forall a. a -> Recipe () a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ByteString -> Either ReadFailure a
owned ByteString
bytes)
  Maybe ValueBytes
Nothing -> ReadFailure -> Parser a
forall env a. ReadFailure -> Recipe env a
failValue ReadFailure
ReadDiscarded
{-# NOINLINE convertBytesSlow #-}

bytesValueSlow :: Int -> Int -> B.ByteString -> Int -> Result (Maybe ValueBytes)
bytesValueSlow :: Int -> Int -> ByteString -> Int -> Result (Maybe ValueBytes)
bytesValueSlow Int
ahead Int
n = Parser (Maybe ValueBytes)
-> ByteString -> Int -> Result (Maybe ValueBytes)
forall a. Parser a -> ByteString -> Int -> Result a
run (Int -> Accumulator -> Parser (Maybe ValueBytes)
gather Int
n (Maybe Word64 -> Accumulator
accumulator (Word64 -> Maybe Word64
forall a. a -> Maybe a
Just (Int -> Word64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
n))))
  where
    gather :: Int -> Accumulator -> Parser (Maybe ValueBytes)
gather Int
0 Accumulator
a = ValueBytes -> Maybe ValueBytes
forall a. a -> Maybe a
Just (ValueBytes -> Maybe ValueBytes)
-> (ByteString -> ValueBytes) -> ByteString -> Maybe ValueBytes
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ByteString -> ValueBytes
Owned (ByteString -> Maybe ValueBytes)
-> Recipe () ByteString -> Parser (Maybe ValueBytes)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IO ByteString -> Recipe () ByteString
forall a. IO a -> Parser a
effect (Accumulator -> IO ByteString
finish Accumulator
a)
    gather Int
remaining Accumulator
a = Int -> Parser (Maybe ByteString)
chunkOrDiscard Int
remaining Parser (Maybe ByteString)
-> (Maybe ByteString -> Parser (Maybe ValueBytes))
-> Parser (Maybe ValueBytes)
forall a b. Recipe () a -> (a -> Recipe () b) -> Recipe () b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
      Maybe ByteString
Nothing -> Int -> Parser ()
skip Int
remaining Parser () -> Parser (Maybe ValueBytes) -> Parser (Maybe ValueBytes)
forall a b. Recipe () a -> Recipe () b -> Recipe () b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Maybe ValueBytes -> Parser (Maybe ValueBytes)
forall a. a -> Recipe () a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe ValueBytes
forall a. Maybe a
Nothing
      Just ByteString
bytes -> do
        next <- IO Accumulator -> Parser Accumulator
forall a. IO a -> Parser a
effect (Int -> Accumulator -> ByteString -> IO Accumulator
append Int
ahead Accumulator
a ByteString
bytes)
        gather (remaining - B.length bytes) next
{-# NOINLINE bytesValueSlow #-}

-- A buffer under construction has no immutable view until it is retired.
data Buffer = Buffer !(ForeignPtr Word8) !Int !Int
--
-- The one mutable thing in the module, and its rule is run-once. Copying each slice
-- in as its packet arrives is what keeps no packet alive and makes one copy. But
-- 'append' writes in place into a buffer that 'finish' can hand out as a ByteString
-- without copying, while a Stream value can in principle be run twice. A suspended
-- value's continuation holds an Accumulator, so running one twice would write into
-- bytes that may already be frozen. Every driver therefore saves the step it got from
-- 'inspect', never the Stream it inspected, before it can be interrupted, so no effect
-- is repeated. This rule reaches outside the module: the drivers are 'feedRunner',
-- 'runFrom', and stepCore and readPiece in Hmssql.Connection.Internal. Nothing tests
-- a violation. Consuming packets as a list and joining at the end would be pure, but
-- peaks at twice the value and keeps whole receive buffers alive.

-- | A long value being assembled from pieces as they arrive.  Use each one only once,
-- because 'append' may write into the same memory and its result replaces its argument.
-- A recipe holding one must not be run twice, as 'runFrom' explains.
data Accumulator
  = Known !Int !Buffer
  | Unknown !Int ![B.ByteString] !Buffer

-- | Start a value with this declared length, or with an unknown one.
accumulator :: Maybe Word64 -> Accumulator
accumulator :: Maybe Word64 -> Accumulator
accumulator Maybe Word64
total = case Maybe Word64
total of
  Just Word64
n -> Int -> Buffer -> Accumulator
Known (Word64 -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Word64 -> Word64 -> Word64
forall a. Ord a => a -> a -> a
min Word64
n (Int -> Word64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int
forall a. Bounded a => a
maxBound :: Int)))) Buffer
empty
  Maybe Word64
Nothing -> Int -> [ByteString] -> Buffer -> Accumulator
Unknown Int
0 [] Buffer
empty
  where empty :: Buffer
empty = ForeignPtr Word8 -> Int -> Int -> Buffer
Buffer ForeignPtr Word8
BI.nullForeignPtr Int
0 Int
0

-- | Add a piece to the value, which is copied, so the piece's packet can be released.
-- Use the result in place of the argument.
--
-- The first argument bounds what a declared length alone can allocate.  A server that
-- declares a huge value and doesn't send it costs that much memory, and after that at
-- most three times what it has actually sent.  A value that turns out longer than it
-- declared carries on as a value of unknown length.
append :: Int -> Accumulator -> B.ByteString -> IO Accumulator
-- Quarters because doubling reached twice the value just over a power of two; pieces
-- because doubling an unknown value's buffer peaks near three times it.
append :: Int -> Accumulator -> ByteString -> IO Accumulator
append Int
ahead Accumulator
a ByteString
src = case Accumulator
a of
  Known Int
total (Buffer ForeignPtr Word8
fp Int
cap Int
len)
    | Int
k Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
forall a. Bounded a => a
maxBound Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
len -> ProtocolError -> IO Accumulator
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO (String -> ProtocolError
ProtocolError String
"value length exceeds addressable memory")
    | Int
needed Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= Int
cap -> do
        ForeignPtr Word8 -> Int -> Int -> ByteString -> IO ()
copy ForeignPtr Word8
fp Int
cap Int
len ByteString
src
        Accumulator -> IO Accumulator
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Int -> Buffer -> Accumulator
Known Int
total (ForeignPtr Word8 -> Int -> Int -> Buffer
Buffer ForeignPtr Word8
fp Int
cap Int
needed))
    | Int
cap Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0 Bool -> Bool -> Bool
|| Int
needed Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= Int
total -> do
        let size :: Int
size | Int
cap Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0 = Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
k ((Int -> Bool) -> (Int -> Int) -> Int -> Int
forall a. (a -> Bool) -> (a -> a) -> a -> a
until (Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
0 Int
ahead) (Int -> Int -> Int
forall a. Integral a => a -> a -> a
`quot` Int
4) Int
total)
                 | Bool
otherwise = Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
needed (if Int
cap Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
total Int -> Int -> Int
forall a. Integral a => a -> a -> a
`quot` Int
4 then Int
total else Int
4 Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
cap)
        fp' <- Int -> IO (ForeignPtr Word8)
forall a. Int -> IO (ForeignPtr a)
BI.mallocByteString Int
size
        copy fp' size 0 (BI.BS fp len)
        copy fp' size len src
        pure (Known total (Buffer fp' size needed))
    | Bool
otherwise -> Int -> Accumulator -> ByteString -> IO Accumulator
append Int
ahead (Int -> [ByteString] -> Buffer -> Accumulator
Unknown Int
0 [] (ForeignPtr Word8 -> Int -> Int -> Buffer
Buffer ForeignPtr Word8
fp Int
cap Int
len)) ByteString
src
    where needed :: Int
needed = Int
len Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
k
  Unknown Int
before [ByteString]
full (Buffer ForeignPtr Word8
fp Int
cap Int
len)
    | Int
k Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
forall a. Bounded a => a
maxBound Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
len Bool -> Bool -> Bool
|| Int
cap Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
forall a. Bounded a => a
maxBound Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
before -> ProtocolError -> IO Accumulator
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO (String -> ProtocolError
ProtocolError String
"value length exceeds addressable memory")
    | Int
len Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
k Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= Int
cap -> do
        ForeignPtr Word8 -> Int -> Int -> ByteString -> IO ()
copy ForeignPtr Word8
fp Int
cap Int
len ByteString
src
        Accumulator -> IO Accumulator
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Int -> [ByteString] -> Buffer -> Accumulator
Unknown Int
before [ByteString]
full (ForeignPtr Word8 -> Int -> Int -> Buffer
Buffer ForeignPtr Word8
fp Int
cap (Int
len Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
k)))
    | Bool
otherwise -> do
        let m :: Int
m = Int
cap Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
len
            size :: Int
size = Int -> Int -> Int
forall a. Ord a => a -> a -> a
max (Int
k Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
m) (if Int
cap Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0 then Int -> Int -> Int
forall a. Ord a => a -> a -> a
min Int
65536 (Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
0 Int
ahead) else Int -> Int -> Int
forall a. Ord a => a -> a -> a
min Int
1048576 (Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
65536 (Int
before Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
cap)))
        ForeignPtr Word8 -> Int -> Int -> ByteString -> IO ()
copy ForeignPtr Word8
fp Int
cap Int
len (Int -> ByteString -> ByteString
B.take Int
m ByteString
src)
        fp' <- Int -> IO (ForeignPtr Word8)
forall a. Int -> IO (ForeignPtr a)
BI.mallocByteString Int
size
        copy fp' size 0 (B.drop m src)
        let full' = if Int
cap Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0 then [ByteString]
full else ForeignPtr Word8 -> Int -> ByteString
BI.BS ForeignPtr Word8
fp Int
cap ByteString -> [ByteString] -> [ByteString]
forall a. a -> [a] -> [a]
: [ByteString]
full
        pure (Unknown (before + cap) full' (Buffer fp' size (k - m)))
  where k :: Int
k = ByteString -> Int
B.length ByteString
src

-- A bounded memcpy between buffers known to be alive, hence unsafeWithForeignPtr.
-- The check can't fail by the arithmetic in 'append'. It is here because a mistake in
-- that arithmetic would otherwise write outside the allocation.
copy :: ForeignPtr Word8 -> Int -> Int -> B.ByteString -> IO ()
copy :: ForeignPtr Word8 -> Int -> Int -> ByteString -> IO ()
copy ForeignPtr Word8
dst Int
capacity Int
at (BI.BS ForeignPtr Word8
src Int
n)
  | Int
at Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
< Int
0 Bool -> Bool -> Bool
|| Int
at Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
capacity Bool -> Bool -> Bool
|| Int
n Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
capacity Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
at = ProtocolError -> IO ()
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO (String -> ProtocolError
ProtocolError String
"value copy outside allocated buffer")
  | Int
n Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0 = () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
  | Bool
otherwise = ForeignPtr Word8 -> (Ptr Word8 -> IO ()) -> IO ()
forall a b. ForeignPtr a -> (Ptr a -> IO b) -> IO b
BI.unsafeWithForeignPtr ForeignPtr Word8
dst ((Ptr Word8 -> IO ()) -> IO ()) -> (Ptr Word8 -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Ptr Word8
d -> ForeignPtr Word8 -> (Ptr Word8 -> IO ()) -> IO ()
forall a b. ForeignPtr a -> (Ptr a -> IO b) -> IO b
BI.unsafeWithForeignPtr ForeignPtr Word8
src ((Ptr Word8 -> IO ()) -> IO ()) -> (Ptr Word8 -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Ptr Word8
s ->
      Ptr Word8 -> Ptr Word8 -> Int -> IO ()
forall a. Ptr a -> Ptr a -> Int -> IO ()
copyBytes (Ptr Word8
d Ptr Word8 -> Int -> Ptr Word8
forall a b. Ptr a -> Int -> Ptr b
`plusPtr` Int
at) Ptr Word8
s Int
n

-- | The assembled value, which is safe to keep.  Don't use the 'Accumulator' again.
finish :: Accumulator -> IO B.ByteString
finish :: Accumulator -> IO ByteString
finish = \case
  Known Int
_ Buffer
b -> ByteString -> IO ByteString
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ByteString -> IO ByteString) -> ByteString -> IO ByteString
forall a b. (a -> b) -> a -> b
$! Buffer -> ByteString
finishBuffer Buffer
b
  Unknown Int
_ [] Buffer
b -> ByteString -> IO ByteString
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ByteString -> IO ByteString) -> ByteString -> IO ByteString
forall a b. (a -> b) -> a -> b
$! Buffer -> ByteString
finishBuffer Buffer
b
  Unknown Int
_ [ByteString]
full (Buffer ForeignPtr Word8
fp Int
_ Int
len) -> ByteString -> IO ByteString
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ByteString -> IO ByteString) -> ByteString -> IO ByteString
forall a b. (a -> b) -> a -> b
$! [ByteString] -> ByteString
B.concat ([ByteString] -> [ByteString]
forall a. [a] -> [a]
reverse (ForeignPtr Word8 -> Int -> ByteString
BI.BS ForeignPtr Word8
fp Int
len ByteString -> [ByteString] -> [ByteString]
forall a. a -> [a] -> [a]
: [ByteString]
full))
  where
    finishBuffer :: Buffer -> ByteString
finishBuffer (Buffer ForeignPtr Word8
fp Int
cap Int
len)
      | Int
len Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0 = ByteString
B.empty
      | Int
cap Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
len Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
16 (Int
len Int -> Int -> Int
forall a. Integral a => a -> a -> a
`div` Int
16) = ForeignPtr Word8 -> Int -> ByteString
BI.BS ForeignPtr Word8
fp Int
len
      | Bool
otherwise = ByteString -> ByteString
B.copy (ForeignPtr Word8 -> Int -> ByteString
BI.BS ForeignPtr Word8
fp Int
len)

------------------------------------------------------------------------------
-- Drivers

-- | Start a recipe on a payload, as a suspended computation for a driver to run with
-- 'inspect'.  Pass an empty payload to start with no input.  A failed value that the
-- recipe didn't 'recover' from is a 'ProtocolError'.
parse :: Parser a -> B.ByteString -> Stream Await IO (Got a)
parse :: forall a. Parser a -> ByteString -> Stream Await IO (Got a)
parse Parser a
p ByteString
bs = case Parser a -> ByteString -> Int -> Result a
forall a. Parser a -> ByteString -> Int -> Result a
run Parser a
p ByteString
bs Int
0 of
  Parsed a
a Int
pos -> Got a -> Stream Await IO (Got a)
forall (f :: * -> *) (m :: * -> *) r. r -> Stream f m r
Return (a -> ByteString -> Got a
forall a. a -> ByteString -> Got a
Got a
a (ByteString -> Int -> ByteString
unread ByteString
bs Int
pos))
  Suspended Stream Await IO (Outcome a)
s -> Stream Await IO (Outcome a) -> Stream Await IO (Got a)
forall a. Stream Await IO (Outcome a) -> Stream Await IO (Got a)
got Stream Await IO (Outcome a)
s
  Failed ReadFailure
failure Int
_ -> ReadFailure -> Stream Await IO (Got a)
forall r. ReadFailure -> Stream Await IO r
uncaught ReadFailure
failure

-- | Start a recipe on a payload.  The result is 'Right' if it finished within the
-- payload, and otherwise 'Left' with the suspended computation.  This lets a loop
-- handle the common case without a 'Stream'.
runParser :: Parser a -> B.ByteString -> Either (Stream Await IO (Got a)) (Got a)
runParser :: forall a.
Parser a -> ByteString -> Either (Stream Await IO (Got a)) (Got a)
runParser Parser a
p ByteString
bytes = case Parser a -> ByteString -> Int -> Result a
forall a. Parser a -> ByteString -> Int -> Result a
run Parser a
p ByteString
bytes Int
0 of
  Parsed a
value Int
pos -> Got a -> Either (Stream Await IO (Got a)) (Got a)
forall a b. b -> Either a b
Right (a -> ByteString -> Got a
forall a. a -> ByteString -> Got a
Got a
value (ByteString -> Int -> ByteString
unread ByteString
bytes Int
pos))
  Suspended Stream Await IO (Outcome a)
continuation -> Stream Await IO (Got a) -> Either (Stream Await IO (Got a)) (Got a)
forall a b. a -> Either a b
Left (Stream Await IO (Outcome a) -> Stream Await IO (Got a)
forall a. Stream Await IO (Outcome a) -> Stream Await IO (Got a)
got Stream Await IO (Outcome a)
continuation)
  Failed ReadFailure
failure Int
_ -> Stream Await IO (Got a) -> Either (Stream Await IO (Got a)) (Got a)
forall a b. a -> Either a b
Left (ReadFailure -> Stream Await IO (Got a)
forall r. ReadFailure -> Stream Await IO r
uncaught ReadFailure
failure)
{-# INLINE runParser #-}

-- | 'runParser' from a position in the payload, for a driver that reads one
-- recipe after another from the same payload, as the connection reads rows, and
-- handles a failed value itself.  A recipe that finishes within the payload gives
-- where it got to as a position in it, so the driver keeps the payload and a
-- position rather than building the unread bytes after each recipe.  A failed
-- value that nothing recovered is 'FailedAt' rather than a 'ProtocolError'; a row's
-- recipe ('Hmssql.Decoding.CompiledRow') leaves its failure for the driver this
-- way.  A recipe that suspends gives its failure as a 'Left' at the end.
runParserAt :: Parser a -> B.ByteString -> Int -> Either (Stream Await IO (Got (Either ReadFailure a))) (At a)
runParserAt :: forall a.
Parser a
-> ByteString
-> Int
-> Either (Stream Await IO (Got (Either ReadFailure a))) (At a)
runParserAt Parser a
p ByteString
bytes (I# Int#
pos) = let !box :: Int
box = Int# -> Int
boxPosition Int#
pos in case Parser a -> ByteString -> Int -> Result a
forall a. Parser a -> ByteString -> Int -> Result a
run Parser a
p ByteString
bytes Int
box of
  Parsed a
value Int
end -> At a
-> Either (Stream Await IO (Got (Either ReadFailure a))) (At a)
forall a b. b -> Either a b
Right (a -> Int -> At a
forall a. a -> Int -> At a
At a
value Int
end)
  Suspended Stream Await IO (Outcome a)
continuation -> Stream Await IO (Got (Either ReadFailure a))
-> Either (Stream Await IO (Got (Either ReadFailure a))) (At a)
forall a b. a -> Either a b
Left (Stream Await IO (Outcome a)
-> Stream Await IO (Got (Either ReadFailure a))
forall a.
Stream Await IO (Outcome a)
-> Stream Await IO (Got (Either ReadFailure a))
gotEither Stream Await IO (Outcome a)
continuation)
  Failed ReadFailure
failure Int
end -> At a
-> Either (Stream Await IO (Got (Either ReadFailure a))) (At a)
forall a b. b -> Either a b
Right (ReadFailure -> Int -> At a
forall a. ReadFailure -> Int -> At a
FailedAt ReadFailure
failure Int
end)
{-# INLINE runParserAt #-}

-- A suspended computation as drivers see it.
got :: Stream Await IO (Outcome a) -> Stream Await IO (Got a)
got :: forall a. Stream Await IO (Outcome a) -> Stream Await IO (Got a)
got Stream Await IO (Outcome a)
s = Stream Await IO (Outcome a)
s Stream Await IO (Outcome a)
-> (Outcome a -> Stream Await IO (Got a))
-> Stream Await IO (Got a)
forall a b.
Stream Await IO a -> (a -> Stream Await IO b) -> Stream Await IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
  Done a
a ByteString
rest -> Got a -> Stream Await IO (Got a)
forall (f :: * -> *) (m :: * -> *) r. r -> Stream f m r
Return (a -> ByteString -> Got a
forall a. a -> ByteString -> Got a
Got a
a ByteString
rest)
  Stopped ReadFailure
failure ByteString
_ -> ReadFailure -> Stream Await IO (Got a)
forall r. ReadFailure -> Stream Await IO r
uncaught ReadFailure
failure

gotEither :: Stream Await IO (Outcome a) -> Stream Await IO (Got (Either ReadFailure a))
gotEither :: forall a.
Stream Await IO (Outcome a)
-> Stream Await IO (Got (Either ReadFailure a))
gotEither Stream Await IO (Outcome a)
s = Stream Await IO (Outcome a)
s Stream Await IO (Outcome a)
-> (Outcome a -> Stream Await IO (Got (Either ReadFailure a)))
-> Stream Await IO (Got (Either ReadFailure a))
forall a b.
Stream Await IO a -> (a -> Stream Await IO b) -> Stream Await IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
  Done a
a ByteString
rest -> Got (Either ReadFailure a)
-> Stream Await IO (Got (Either ReadFailure a))
forall (f :: * -> *) (m :: * -> *) r. r -> Stream f m r
Return (Either ReadFailure a -> ByteString -> Got (Either ReadFailure a)
forall a. a -> ByteString -> Got a
Got (a -> Either ReadFailure a
forall a b. b -> Either a b
Right a
a) ByteString
rest)
  Stopped ReadFailure
failure ByteString
rest -> Got (Either ReadFailure a)
-> Stream Await IO (Got (Either ReadFailure a))
forall (f :: * -> *) (m :: * -> *) r. r -> Stream f m r
Return (Either ReadFailure a -> ByteString -> Got (Either ReadFailure a)
forall a. a -> ByteString -> Got a
Got (ReadFailure -> Either ReadFailure a
forall a b. a -> Either a b
Left ReadFailure
failure) ByteString
rest)

uncaught :: ReadFailure -> Stream Await IO r
uncaught :: forall r. ReadFailure -> Stream Await IO r
uncaught ReadFailure
failure = IO (Stream Await IO r) -> Stream Await IO r
forall (f :: * -> *) (m :: * -> *) r.
m (Stream f m r) -> Stream f m r
Effect (ProtocolError -> IO (Stream Await IO r)
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO (String -> ProtocolError
ProtocolError (String
"a failed value reached the driver: " String -> String -> String
forall a. [a] -> [a] -> [a]
++ ReadFailure -> String
forall a. Show a => a -> String
show ReadFailure
failure)))

-- A result as a suspended computation, given the payload its position is in.
stream :: B.ByteString -> Result a -> Stream Await IO (Outcome a)
stream :: forall a. ByteString -> Result a -> Stream Await IO (Outcome a)
stream ByteString
bs (Parsed a
a Int
pos) = Outcome a -> Stream Await IO (Outcome a)
forall (f :: * -> *) (m :: * -> *) r. r -> Stream f m r
Return (a -> ByteString -> Outcome a
forall a. a -> ByteString -> Outcome a
Done a
a (ByteString -> Int -> ByteString
unread ByteString
bs Int
pos))
stream ByteString
_ (Suspended Stream Await IO (Outcome a)
s) = Stream Await IO (Outcome a)
s
stream ByteString
bs (Failed ReadFailure
failure Int
pos) = Outcome a -> Stream Await IO (Outcome a)
forall (f :: * -> *) (m :: * -> *) r. r -> Stream f m r
Return (ReadFailure -> ByteString -> Outcome a
forall a. ReadFailure -> ByteString -> Outcome a
Stopped ReadFailure
failure (ByteString -> Int -> ByteString
unread ByteString
bs Int
pos))
{-# INLINE stream #-}

outcomeRest :: Outcome a -> B.ByteString
outcomeRest :: forall a. Outcome a -> ByteString
outcomeRest (Done a
_ ByteString
rest) = ByteString
rest
outcomeRest (Stopped ReadFailure
_ ByteString
rest) = ByteString
rest

-- | Run a recipe over payloads already in memory, followed by the end of the message.
-- The result is the recipe's value and whatever input it didn't read.
parseChunks :: Parser a -> [B.ByteString] -> IO (a, [B.ByteString])
parseChunks :: forall a. Parser a -> [ByteString] -> IO (a, [ByteString])
parseChunks Parser a
p [ByteString]
chunks = do
  input <- [ByteString] -> IO (IORef [ByteString])
forall a. a -> IO (IORef a)
newIORef [ByteString]
chunks
  let feed = IORef [ByteString]
-> ([ByteString] -> ([ByteString], Feed)) -> IO Feed
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef [ByteString]
input (([ByteString] -> ([ByteString], Feed)) -> IO Feed)
-> ([ByteString] -> ([ByteString], Feed)) -> IO Feed
forall a b. (a -> b) -> a -> b
$ \case
        [] -> ([], Feed
EndOfMessage)
        ByteString
c : [ByteString]
cs -> ([ByteString]
cs, ByteString -> Feed
More ByteString
c)
  Got a rest <- runFrom feed p B.empty
  later <- readIORef input
  pure (a, if B.null rest then later else rest : later)

-- | A 'Runner' that gets payloads from this action.  Its caller decides what to mask
-- and what to do about exceptions.  It saves its progress before each step, so that
-- retrying after a failed fetch, or looking again at a finished value, repeats nothing.
-- A value's recipe that asks to run a consumer of its own is a 'ProtocolError', and it
-- stays failed in the reference, so that later cleanup can't run the consumer.
feedRunner :: IO Feed -> Runner
feedRunner :: IO Feed -> Runner
feedRunner IO Feed
feed = (forall a. IORef (Stream Await IO (Got a)) -> IO (Got a)) -> Runner
Runner ((forall a. IORef (Stream Await IO (Got a)) -> IO (Got a))
 -> Runner)
-> (forall a. IORef (Stream Await IO (Got a)) -> IO (Got a))
-> Runner
forall a b. (a -> b) -> a -> b
$ \IORef (Stream Await IO (Got a))
current ->
  let loop :: IO (Got a)
loop = IORef (Stream Await IO (Got a)) -> IO (Stream Await IO (Got a))
forall a. IORef a -> IO a
readIORef IORef (Stream Await IO (Got a))
current IO (Stream Await IO (Got a))
-> (Stream Await IO (Got a)
    -> IO (Either (Got a) (Await (Stream Await IO (Got a)))))
-> IO (Either (Got a) (Await (Stream Await IO (Got a))))
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Stream Await IO (Got a)
-> IO (Either (Got a) (Await (Stream Await IO (Got a))))
forall (m :: * -> *) (f :: * -> *) r.
Monad m =>
Stream f m r -> m (Either r (f (Stream f m r)))
inspect IO (Either (Got a) (Await (Stream Await IO (Got a))))
-> (Either (Got a) (Await (Stream Await IO (Got a))) -> IO (Got a))
-> IO (Got a)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
        Left Got a
got -> IORef (Stream Await IO (Got a)) -> Stream Await IO (Got a) -> IO ()
forall a. IORef a -> a -> IO ()
writeIORef IORef (Stream Await IO (Got a))
current (Got a -> Stream Await IO (Got a)
forall a. a -> Stream Await IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Got a
got) IO () -> IO (Got a) -> IO (Got a)
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Got a -> IO (Got a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Got a
got
        Right await :: Await (Stream Await IO (Got a))
await@(NeedMore Feed -> Stream Await IO (Got a)
k) -> do
          IORef (Stream Await IO (Got a)) -> Stream Await IO (Got a) -> IO ()
forall a. IORef a -> a -> IO ()
writeIORef IORef (Stream Await IO (Got a))
current (Await (Stream Await IO (Got a)) -> Stream Await IO (Got a)
forall (m :: * -> *) (f :: * -> *) r.
(Monad m, Functor f) =>
f (Stream f m r) -> Stream f m r
S.wrap Await (Stream Await IO (Got a))
await)
          next <- IO Feed
feed
          writeIORef current (k next)
          loop
        Right RunConsumer{} -> do
          let err :: ProtocolError
err = String -> ProtocolError
ProtocolError String
"a value recipe cannot invoke a nested consumer"
          -- The outer consumer may catch this exception. Its subsequent drain
          -- must encounter the failure, not hand the rejected action to a driver.
          IORef (Stream Await IO (Got a)) -> Stream Await IO (Got a) -> IO ()
forall a. IORef a -> a -> IO ()
writeIORef IORef (Stream Await IO (Got a))
current (IO (Stream Await IO (Got a)) -> Stream Await IO (Got a)
forall (f :: * -> *) (m :: * -> *) r.
m (Stream f m r) -> Stream f m r
Effect (ProtocolError -> IO (Stream Await IO (Got a))
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO ProtocolError
err))
          ProtocolError -> IO (Got a)
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO ProtocolError
err
   in IO (Got a)
loop

-- | Run a recipe to its end, starting with this payload and fetching the rest from the
-- action.  It also runs any consumers the recipe holds, with 'feedRunner' for their
-- pulls.  The result includes the unread part of the last payload.
--
-- This is the driver to use unless you need something it doesn't do.  A driver of your
-- own must follow one rule: it must run each step of a suspended recipe only once.
-- After 'inspect', save what 'inspect' returned, and not the computation it was given,
-- before anything can interrupt the driver.  Running a step twice repeats its effects,
-- and an 'Accumulator' it holds would write into memory it may already have given away.
runFrom :: IO Feed -> Parser a -> B.ByteString -> IO (Got a)
runFrom :: forall a. IO Feed -> Parser a -> ByteString -> IO (Got a)
runFrom IO Feed
feed Parser a
p ByteString
initial = do
  current <- Stream Await IO (Got a) -> IO (IORef (Stream Await IO (Got a)))
forall a. a -> IO (IORef a)
newIORef (Parser a -> ByteString -> Stream Await IO (Got a)
forall a. Parser a -> ByteString -> Stream Await IO (Got a)
parse Parser a
p ByteString
initial)
  let runner = IO Feed -> Runner
feedRunner IO Feed
feed
      loop = IORef (Stream Await IO (Got a)) -> IO (Stream Await IO (Got a))
forall a. IORef a -> IO a
readIORef IORef (Stream Await IO (Got a))
current IO (Stream Await IO (Got a))
-> (Stream Await IO (Got a)
    -> IO (Either (Got a) (Await (Stream Await IO (Got a)))))
-> IO (Either (Got a) (Await (Stream Await IO (Got a))))
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Stream Await IO (Got a)
-> IO (Either (Got a) (Await (Stream Await IO (Got a))))
forall (m :: * -> *) (f :: * -> *) r.
Monad m =>
Stream f m r -> m (Either r (f (Stream f m r)))
inspect IO (Either (Got a) (Await (Stream Await IO (Got a))))
-> (Either (Got a) (Await (Stream Await IO (Got a))) -> IO (Got a))
-> IO (Got a)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
        Left Got a
got -> IORef (Stream Await IO (Got a)) -> Stream Await IO (Got a) -> IO ()
forall a. IORef a -> a -> IO ()
writeIORef IORef (Stream Await IO (Got a))
current (Got a -> Stream Await IO (Got a)
forall a. a -> Stream Await IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Got a
got) IO () -> IO (Got a) -> IO (Got a)
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Got a -> IO (Got a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Got a
got
        Right await :: Await (Stream Await IO (Got a))
await@(NeedMore Feed -> Stream Await IO (Got a)
k) -> do
          IORef (Stream Await IO (Got a)) -> Stream Await IO (Got a) -> IO ()
forall a. IORef a -> a -> IO ()
writeIORef IORef (Stream Await IO (Got a))
current (Await (Stream Await IO (Got a)) -> Stream Await IO (Got a)
forall (m :: * -> *) (f :: * -> *) r.
(Monad m, Functor f) =>
f (Stream f m r) -> Stream f m r
S.wrap Await (Stream Await IO (Got a))
await)
          next <- IO Feed
feed
          writeIORef current (k next)
          loop
        Right await :: Await (Stream Await IO (Got a))
await@(RunConsumer Bool
_ Runner -> IO x
io Maybe x -> Stream Await IO (Got a)
k) -> do
          IORef (Stream Await IO (Got a)) -> Stream Await IO (Got a) -> IO ()
forall a. IORef a -> a -> IO ()
writeIORef IORef (Stream Await IO (Got a))
current (Await (Stream Await IO (Got a)) -> Stream Await IO (Got a)
forall (m :: * -> *) (f :: * -> *) r.
(Monad m, Functor f) =>
f (Stream f m r) -> Stream f m r
S.wrap Await (Stream Await IO (Got a))
await)
          result <- Runner -> IO x
io Runner
runner
          writeIORef current (k (Just result))
          loop
  loop

-- | Give a stream of bytes to a function that reads it as the bytes arrive, then carry
-- on with the recipe.  This is how a caller's code reads a large value without it being
-- in memory ('Hmssql.Decoding.streamField' is built on it).  The result is what the
-- function returned, or 'Nothing' when the driver is throwing the response away and
-- didn't run it.  Whatever part of the stream the function didn't read is read past
-- afterwards, so the recipe continues in the right place either way.  An exception
-- the function throws ends the driver instead ('RunConsumer'); the saved step then
-- still holds the value, and a drain that resumes it reads past the rest.
--
-- The driver runs the function ('RunConsumer'), and each pull goes through the driver's
-- 'Runner'.  Each chunk is a copy, so it can be kept.  Pulling the same part of the
-- stream twice, or pulling after the function has returned, throws 'ResultAbandoned'.
consume :: Stream (Of B.ByteString) Parser ()
  -> (Stream (Of B.ByteString) IO () -> IO a) -> Parser (Maybe a)
consume :: forall a.
Stream (Of ByteString) (Recipe ()) ()
-> (Stream (Of ByteString) IO () -> IO a) -> Parser (Maybe a)
consume Stream (Of ByteString) (Recipe ()) ()
source Stream (Of ByteString) IO () -> IO a
use = (ByteString -> Int -> Result (Maybe a)) -> Parser (Maybe a)
forall a. (ByteString -> Int -> Result a) -> Parser a
parser ((ByteString -> Int -> Result (Maybe a)) -> Parser (Maybe a))
-> (ByteString -> Int -> Result (Maybe a)) -> Parser (Maybe a)
forall a b. (a -> b) -> a -> b
$ \ByteString
bs Int
pos -> Stream Await IO (Outcome (Maybe a)) -> Result (Maybe a)
forall a. Stream Await IO (Outcome a) -> Result a
Suspended (Stream Await IO (Outcome (Maybe a)) -> Result (Maybe a))
-> Stream Await IO (Outcome (Maybe a)) -> Result (Maybe a)
forall a b. (a -> b) -> a -> b
$ IO (Stream Await IO (Outcome (Maybe a)))
-> Stream Await IO (Outcome (Maybe a))
forall (f :: * -> *) (m :: * -> *) r.
m (Stream f m r) -> Stream f m r
Effect (IO (Stream Await IO (Outcome (Maybe a)))
 -> Stream Await IO (Outcome (Maybe a)))
-> IO (Stream Await IO (Outcome (Maybe a)))
-> Stream Await IO (Outcome (Maybe a))
forall a b. (a -> b) -> a -> b
$ do
  current <- Stream
  Await
  IO
  (Got
     (Either
        () (Of ByteString (Stream (Of ByteString) (Recipe ()) ()))))
-> IO
     (IORef
        (Stream
           Await
           IO
           (Got
              (Either
                 () (Of ByteString (Stream (Of ByteString) (Recipe ()) ()))))))
forall a. a -> IO (IORef a)
newIORef (Parser
  (Either () (Of ByteString (Stream (Of ByteString) (Recipe ()) ())))
-> ByteString
-> Stream
     Await
     IO
     (Got
        (Either
           () (Of ByteString (Stream (Of ByteString) (Recipe ()) ()))))
forall a. Parser a -> ByteString -> Stream Await IO (Got a)
parse (Stream (Of ByteString) (Recipe ()) ()
-> Parser
     (Either () (Of ByteString (Stream (Of ByteString) (Recipe ()) ())))
forall (m :: * -> *) (f :: * -> *) r.
Monad m =>
Stream f m r -> m (Either r (f (Stream f m r)))
S.inspect Stream (Of ByteString) (Recipe ()) ()
source) (ByteString -> Int -> ByteString
unread ByteString
bs Int
pos))
  active <- newIORef (Just (0 :: Int))
  let pull Runner
runner Int
generation = IO (Stream (Of ByteString) IO ()) -> Stream (Of ByteString) IO ()
forall (m :: * -> *) (f :: * -> *) r.
(Monad m, Functor f) =>
m (Stream f m r) -> Stream f m r
S.effect (IO (Stream (Of ByteString) IO ()) -> Stream (Of ByteString) IO ())
-> IO (Stream (Of ByteString) IO ())
-> Stream (Of ByteString) IO ()
forall a b. (a -> b) -> a -> b
$ IO (Stream (Of ByteString) IO ())
-> IO (Stream (Of ByteString) IO ())
forall a. IO a -> IO a
mask_ (IO (Stream (Of ByteString) IO ())
 -> IO (Stream (Of ByteString) IO ()))
-> IO (Stream (Of ByteString) IO ())
-> IO (Stream (Of ByteString) IO ())
forall a b. (a -> b) -> a -> b
$ do
        valid <- IORef (Maybe Int) -> (Maybe Int -> (Maybe Int, Bool)) -> IO Bool
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef (Maybe Int)
active ((Maybe Int -> (Maybe Int, Bool)) -> IO Bool)
-> (Maybe Int -> (Maybe Int, Bool)) -> IO Bool
forall a b. (a -> b) -> a -> b
$ \Maybe Int
state -> case Maybe Int
state of
          Just Int
n | Int
n Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
generation -> (Int -> Maybe Int
forall a. a -> Maybe a
Just (Int
n Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1), Bool
True)
          Maybe Int
_ -> (Maybe Int
state, Bool
False)
        unless valid (throwIO ResultAbandoned)
        Got step rest <- runAwait runner current
        case step of
          Left () -> Stream (Of ByteString) IO () -> IO (Stream (Of ByteString) IO ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Stream (Of ByteString) IO ()
forall a. a -> Stream (Of ByteString) IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
          Right (ByteString
chunk :> Stream (Of ByteString) (Recipe ()) ()
next) -> do
            IORef
  (Stream
     Await
     IO
     (Got
        (Either
           () (Of ByteString (Stream (Of ByteString) (Recipe ()) ())))))
-> Stream
     Await
     IO
     (Got
        (Either
           () (Of ByteString (Stream (Of ByteString) (Recipe ()) ()))))
-> IO ()
forall a. IORef a -> a -> IO ()
writeIORef IORef
  (Stream
     Await
     IO
     (Got
        (Either
           () (Of ByteString (Stream (Of ByteString) (Recipe ()) ())))))
current (Parser
  (Either () (Of ByteString (Stream (Of ByteString) (Recipe ()) ())))
-> ByteString
-> Stream
     Await
     IO
     (Got
        (Either
           () (Of ByteString (Stream (Of ByteString) (Recipe ()) ()))))
forall a. Parser a -> ByteString -> Stream Await IO (Got a)
parse (Stream (Of ByteString) (Recipe ()) ()
-> Parser
     (Either () (Of ByteString (Stream (Of ByteString) (Recipe ()) ())))
forall (m :: * -> *) (f :: * -> *) r.
Monad m =>
Stream f m r -> m (Either r (f (Stream f m r)))
S.inspect Stream (Of ByteString) (Recipe ()) ()
next) ByteString
rest)
            Stream (Of ByteString) IO () -> IO (Stream (Of ByteString) IO ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ByteString -> Stream (Of ByteString) IO ()
forall (m :: * -> *) a. Monad m => a -> Stream (Of a) m ()
S.yield (ByteString -> ByteString
B.copy ByteString
chunk) Stream (Of ByteString) IO ()
-> Stream (Of ByteString) IO () -> Stream (Of ByteString) IO ()
forall a b.
Stream (Of ByteString) IO a
-> Stream (Of ByteString) IO b -> Stream (Of ByteString) IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Runner -> Int -> Stream (Of ByteString) IO ()
pull Runner
runner (Int
generation Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1))
      finishConsumer t
result = IO (Stream Await IO (Outcome t)) -> Stream Await IO (Outcome t)
forall (f :: * -> *) (m :: * -> *) r.
m (Stream f m r) -> Stream f m r
Effect (IO (Stream Await IO (Outcome t)) -> Stream Await IO (Outcome t))
-> IO (Stream Await IO (Outcome t)) -> Stream Await IO (Outcome t)
forall a b. (a -> b) -> a -> b
$ do
        remaining <- IORef
  (Stream
     Await
     IO
     (Got
        (Either
           () (Of ByteString (Stream (Of ByteString) (Recipe ()) ())))))
-> IO
     (Stream
        Await
        IO
        (Got
           (Either
              () (Of ByteString (Stream (Of ByteString) (Recipe ()) ())))))
forall a. IORef a -> IO a
readIORef IORef
  (Stream
     Await
     IO
     (Got
        (Either
           () (Of ByteString (Stream (Of ByteString) (Recipe ()) ())))))
current
        pure (drain remaining result)
      drain Stream
  Await IO (Got (Either () (Of a (Stream (Of a) (Recipe ()) ()))))
remaining t
result = Stream
  Await IO (Got (Either () (Of a (Stream (Of a) (Recipe ()) ()))))
remaining Stream
  Await IO (Got (Either () (Of a (Stream (Of a) (Recipe ()) ()))))
-> (Got (Either () (Of a (Stream (Of a) (Recipe ()) ())))
    -> Stream Await IO (Outcome t))
-> Stream Await IO (Outcome t)
forall a b.
Stream Await IO a -> (a -> Stream Await IO b) -> Stream Await IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \(Got Either () (Of a (Stream (Of a) (Recipe ()) ()))
step ByteString
rest) -> case Either () (Of a (Stream (Of a) (Recipe ()) ()))
step of
        Left () -> IO (Stream Await IO (Outcome t)) -> Stream Await IO (Outcome t)
forall (f :: * -> *) (m :: * -> *) r.
m (Stream f m r) -> Stream f m r
Effect (IO (Stream Await IO (Outcome t)) -> Stream Await IO (Outcome t))
-> IO (Stream Await IO (Outcome t)) -> Stream Await IO (Outcome t)
forall a b. (a -> b) -> a -> b
$ do
          IORef
  (Stream
     Await
     IO
     (Got
        (Either
           () (Of ByteString (Stream (Of ByteString) (Recipe ()) ())))))
-> Stream
     Await
     IO
     (Got
        (Either
           () (Of ByteString (Stream (Of ByteString) (Recipe ()) ()))))
-> IO ()
forall a. IORef a -> a -> IO ()
writeIORef IORef
  (Stream
     Await
     IO
     (Got
        (Either
           () (Of ByteString (Stream (Of ByteString) (Recipe ()) ())))))
current (Got
  (Either () (Of ByteString (Stream (Of ByteString) (Recipe ()) ())))
-> Stream
     Await
     IO
     (Got
        (Either
           () (Of ByteString (Stream (Of ByteString) (Recipe ()) ()))))
forall a. a -> Stream Await IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either () (Of ByteString (Stream (Of ByteString) (Recipe ()) ()))
-> ByteString
-> Got
     (Either () (Of ByteString (Stream (Of ByteString) (Recipe ()) ())))
forall a. a -> ByteString -> Got a
Got (()
-> Either
     () (Of ByteString (Stream (Of ByteString) (Recipe ()) ()))
forall a b. a -> Either a b
Left ()) ByteString
B.empty))
          Stream Await IO (Outcome t) -> IO (Stream Await IO (Outcome t))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Outcome t -> Stream Await IO (Outcome t)
forall a. a -> Stream Await IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (t -> ByteString -> Outcome t
forall a. a -> ByteString -> Outcome a
Done t
result ByteString
rest))
        Right (a
_ :> Stream (Of a) (Recipe ()) ()
next) -> Stream
  Await IO (Got (Either () (Of a (Stream (Of a) (Recipe ()) ()))))
-> t -> Stream Await IO (Outcome t)
drain (Parser (Either () (Of a (Stream (Of a) (Recipe ()) ())))
-> ByteString
-> Stream
     Await IO (Got (Either () (Of a (Stream (Of a) (Recipe ()) ()))))
forall a. Parser a -> ByteString -> Stream Await IO (Got a)
parse (Stream (Of a) (Recipe ()) ()
-> Parser (Either () (Of a (Stream (Of a) (Recipe ()) ())))
forall (m :: * -> *) (f :: * -> *) r.
Monad m =>
Stream f m r -> m (Either r (f (Stream f m r)))
S.inspect Stream (Of a) (Recipe ()) ()
next) ByteString
rest) t
result
  pure $ Step $ RunConsumer True (\Runner
runner -> Stream (Of ByteString) IO () -> IO a
use (Runner -> Int -> Stream (Of ByteString) IO ()
pull Runner
runner Int
0) IO a -> IO () -> IO a
forall a b. IO a -> IO b -> IO a
`finally` IORef (Maybe Int) -> Maybe Int -> IO ()
forall a. IORef a -> a -> IO ()
writeIORef IORef (Maybe Int)
active Maybe Int
forall a. Maybe a
Nothing) finishConsumer

-- | 'consume' for a value already read whole: the function gets a stream of one
-- chunk, these bytes (none, for an empty value), which it may keep, and the recipe goes on after it with no
-- reading of its own.  The driver runs the function as it runs 'consume''s, and the
-- stream has the same rules, but the function never pulls from the response, so a
-- driver can run it among the rows it has already read.  This is what
-- 'Hmssql.Decoding.streamField' does with a value small enough to read whole first,
-- where 'consume''s references, masked pulls and drain per value cost several
-- times what the value does.
consumeWhole :: B.ByteString -> (Stream (Of B.ByteString) IO () -> IO a) -> Parser (Maybe a)
consumeWhole :: forall a.
ByteString
-> (Stream (Of ByteString) IO () -> IO a) -> Parser (Maybe a)
consumeWhole ByteString
bytes Stream (Of ByteString) IO () -> IO a
use = (ByteString -> Int -> Result (Maybe a)) -> Parser (Maybe a)
forall a. (ByteString -> Int -> Result a) -> Parser a
parser ((ByteString -> Int -> Result (Maybe a)) -> Parser (Maybe a))
-> (ByteString -> Int -> Result (Maybe a)) -> Parser (Maybe a)
forall a b. (a -> b) -> a -> b
$ \ByteString
bs Int
pos ->
  Stream Await IO (Outcome (Maybe a)) -> Result (Maybe a)
forall a. Stream Await IO (Outcome a) -> Result a
Suspended (Stream Await IO (Outcome (Maybe a)) -> Result (Maybe a))
-> Stream Await IO (Outcome (Maybe a)) -> Result (Maybe a)
forall a b. (a -> b) -> a -> b
$ Await (Stream Await IO (Outcome (Maybe a)))
-> Stream Await IO (Outcome (Maybe a))
forall (f :: * -> *) (m :: * -> *) r.
f (Stream f m r) -> Stream f m r
Step (Await (Stream Await IO (Outcome (Maybe a)))
 -> Stream Await IO (Outcome (Maybe a)))
-> Await (Stream Await IO (Outcome (Maybe a)))
-> Stream Await IO (Outcome (Maybe a))
forall a b. (a -> b) -> a -> b
$ Bool
-> (Runner -> IO a)
-> (Maybe a -> Stream Await IO (Outcome (Maybe a)))
-> Await (Stream Await IO (Outcome (Maybe a)))
forall next x.
Bool -> (Runner -> IO x) -> (Maybe x -> next) -> Await next
RunConsumer Bool
False Runner -> IO a
forall {p}. p -> IO a
start (\Maybe a
result -> Outcome (Maybe a) -> Stream Await IO (Outcome (Maybe a))
forall a. a -> Stream Await IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Maybe a -> ByteString -> Outcome (Maybe a)
forall a. a -> ByteString -> Outcome a
Done Maybe a
result (ByteString -> Int -> ByteString
unread ByteString
bs Int
pos)))
  where
    start :: p -> IO a
start p
_ = do
      active <- Bool -> IO (IORef Bool)
forall a. a -> IO (IORef a)
newIORef Bool
True
      let once = IO (Stream (Of ByteString) IO ()) -> Stream (Of ByteString) IO ()
forall (m :: * -> *) (f :: * -> *) r.
(Monad m, Functor f) =>
m (Stream f m r) -> Stream f m r
S.effect (IO (Stream (Of ByteString) IO ()) -> Stream (Of ByteString) IO ())
-> IO (Stream (Of ByteString) IO ())
-> Stream (Of ByteString) IO ()
forall a b. (a -> b) -> a -> b
$ do
            valid <- IORef Bool -> (Bool -> (Bool, Bool)) -> IO Bool
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef Bool
active (\Bool
v -> (Bool
False, Bool
v))
            unless valid (throwIO ResultAbandoned)
            pure (if B.null bytes then pure () else S.yield bytes)
      use once `finally` writeIORef active False

------------------------------------------------------------------------------
-- Read failures
--
-- These live here because the recipes throw them or fail with them; "Hmssql.Errors"
-- re-exports the exceptions, and "Hmssql.Decoding" re-exports 'ReadFailure'.

-- | Why a value or a row was read past without a result.
data ReadFailure
  = DecodeFailed !DecodeError
    -- ^ The value was read but can't be decoded, or the row doesn't have the
    -- columns its decoder expects.
  | ReadDiscarded
    -- ^ The response is being thrown away, so the value wasn't assembled and its
    -- consumer didn't run.  Nothing reads this result.

instance Show ReadFailure where
  showsPrec :: Int -> ReadFailure -> String -> String
showsPrec Int
p = \case
    DecodeFailed DecodeError
a -> Int -> Addr# -> DecodeError -> String -> String
forall a. Show a => Int -> Addr# -> a -> String -> String
showCon1 Int
p Addr#
"DecodeFailed"# DecodeError
a
    ReadFailure
ReadDiscarded -> String -> String -> String
showString String
"ReadDiscarded"
  {-# OPAQUE showsPrec #-}

-- | A row or value from the server did not fit the requested Haskell type:
-- a column count or SQL type the row type doesn't accept (found when the
-- result set's columns arrive, before any row), or a value that doesn't
-- decode (a NULL for a non-'Maybe' field, an out-of-range integer, a
-- 'Hmssql.Decoding.refineField' that said no).  The rest of the response
-- is read before it is thrown, so the connection stays usable.
data DecodeError = DecodeError
  { DecodeError -> Maybe Text
decodeErrorColumn :: !(Maybe Text)
    -- ^ The column at fault, when the error is one column's.
  , DecodeError -> String
decodeErrorMessage :: !String
  }

instance Show DecodeError where
  showsPrec :: Int -> DecodeError -> String -> String
showsPrec Int
p (DecodeError Maybe Text
a String
b) =
    Int -> Addr# -> [Field] -> String -> String
showRecord Int
p Addr#
"DecodeError"# [Addr#
"decodeErrorColumn"# Addr# -> Maybe Text -> Field
forall a. Show a => Addr# -> a -> Field
:= Maybe Text
a, Addr#
"decodeErrorMessage"# Addr# -> String -> Field
forall a. Show a => Addr# -> a -> Field
:= String
b]
  {-# OPAQUE showsPrec #-}

instance Eq DecodeError where
  DecodeError Maybe Text
a String
b == :: DecodeError -> DecodeError -> Bool
== DecodeError Maybe Text
a' String
b' = Maybe Text -> Maybe Text -> Bool
forall a. Eq a => a -> a -> Bool
eq Maybe Text
a Maybe Text
a' Bool -> Bool -> Bool
&& String -> String -> Bool
forall a. Eq a => a -> a -> Bool
eq String
b String
b'
  {-# OPAQUE (==) #-}

instance Exception DecodeError

-- | The byte stream from the server could not be interpreted.  The connection
-- is unusable afterwards.
newtype ProtocolError = ProtocolError String

instance Show ProtocolError where
  showsPrec :: Int -> ProtocolError -> String -> String
showsPrec Int
p (ProtocolError String
a) = Int -> Addr# -> String -> String -> String
forall a. Show a => Int -> Addr# -> a -> String -> String
showCon1 Int
p Addr#
"ProtocolError"# String
a
  {-# OPAQUE showsPrec #-}

instance Exception ProtocolError

-- | A result stream was consumed after a newer request was started on the
-- same connection, which discards any unread results, or after a read of
-- it was interrupted by an asynchronous exception, which cancels the
-- statement (see 'Hmssql.Query.queryS').  Also thrown when pulling again
-- from a stream that was already read to its end.
--
-- On a MARS connection ('Hmssql.Connection.connectMars') a newer request
-- leaves the stream open, so this is thrown only by a stream that was
-- canceled: by 'Hmssql.Query.cancelResponse' or the end of its
-- 'Hmssql.Query.withQueryS', by 'Hmssql.Connection.cancelActiveStatement',
-- a 'Hmssql.Connection.reset', the rollback of a transaction it was open
-- in, or an interrupted read.
data ResultAbandoned = ResultAbandoned
  deriving stock (Int -> ResultAbandoned -> String -> String
[ResultAbandoned] -> String -> String
ResultAbandoned -> String
(Int -> ResultAbandoned -> String -> String)
-> (ResultAbandoned -> String)
-> ([ResultAbandoned] -> String -> String)
-> Show ResultAbandoned
forall a.
(Int -> a -> String -> String)
-> (a -> String) -> ([a] -> String -> String) -> Show a
$cshowsPrec :: Int -> ResultAbandoned -> String -> String
showsPrec :: Int -> ResultAbandoned -> String -> String
$cshow :: ResultAbandoned -> String
show :: ResultAbandoned -> String
$cshowList :: [ResultAbandoned] -> String -> String
showList :: [ResultAbandoned] -> String -> String
Show)

instance Exception ResultAbandoned