-- | Coverage for 'Pqi.pipelineSync': marking a synchronization point that
-- batches pipelined commands, and the abort semantics when one of them fails.
module Pqi.Conformance.Operation.PipelineSync
  ( spec,
  )
where

import qualified Pqi
import qualified Pqi as Lq
import Pqi.Conformance.Harness
import qualified Pqi.Conformance.Operation.PipelineSync.Parity as Parity
import Pqi.Conformance.Prelude
import Pqi.Conformance.Scenario (takeCommandResults, takeResult)
import Test.Hspec

spec :: Pqi.Adapter -> SpecWith ByteString
spec :: Adapter -> SpecWith ByteString
spec Adapter
adapter =
  String -> SpecWith ByteString -> SpecWith ByteString
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"pipelineSync" do
    Adapter -> SpecWith ByteString
Parity.spec Adapter
adapter
    String
-> (ByteString -> IO ()) -> SpecWith (Arg (ByteString -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"collects pipelined queries per sync" \ByteString
conninfo ->
      Adapter
-> ByteString
-> (Connection
    -> IO
         (Bool, [Bool], Bool,
          (Maybe ResultObservation, Maybe ResultObservation),
          (Maybe ResultObservation, Maybe ResultObservation),
          (Maybe ResultObservation, Maybe ResultObservation),
          Maybe ResultObservation, Maybe ResultObservation, Bool))
-> IO ()
forall a.
(Eq a, Show a, HasCallStack) =>
Adapter -> ByteString -> (Connection -> IO a) -> IO ()
differential Adapter
adapter ByteString
conninfo \Connection
connection -> do
        entered <- Connection -> IO Bool
Lq.enterPipelineMode Connection
connection
        sent <-
          traverse
            (\ByteString
sql -> Connection
-> ByteString
-> [Maybe (Word32, ByteString, Format)]
-> Format
-> IO Bool
Lq.sendQueryParams Connection
connection ByteString
sql [] Format
Lq.Text)
            ["select 1 :: int4", "select 'two' :: text", "select 3 :: int4, 'three' :: text"]
        synced <- Lq.pipelineSync connection
        first <- takeCommandResults connection
        second <- takeCommandResults connection
        third <- takeCommandResults connection
        syncResult <- takeResult connection
        idle <- takeResult connection
        exited <- Lq.exitPipelineMode connection
        pure (entered, sent, synced, first, second, third, syncResult, idle, exited)

    String
-> (ByteString -> IO ()) -> SpecWith (Arg (ByteString -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"aborts the rest of the pipeline after an error" \ByteString
conninfo ->
      Adapter
-> ByteString
-> (Connection
    -> IO
         (Bool, [Bool], Bool,
          (Maybe ResultObservation, Maybe ResultObservation),
          (Maybe ResultObservation, Maybe ResultObservation),
          (Maybe ResultObservation, Maybe ResultObservation),
          Maybe ResultObservation, Bool))
-> IO ()
forall a.
(Eq a, Show a, HasCallStack) =>
Adapter -> ByteString -> (Connection -> IO a) -> IO ()
differential Adapter
adapter ByteString
conninfo \Connection
connection -> do
        entered <- Connection -> IO Bool
Lq.enterPipelineMode Connection
connection
        sent <-
          traverse
            (\ByteString
sql -> Connection
-> ByteString
-> [Maybe (Word32, ByteString, Format)]
-> Format
-> IO Bool
Lq.sendQueryParams Connection
connection ByteString
sql [] Format
Lq.Text)
            ["select 1", "select 1 / 0", "select 3"]
        synced <- Lq.pipelineSync connection
        first <- takeCommandResults connection
        failed <- takeCommandResults connection
        aborted <- takeCommandResults connection
        syncResult <- takeResult connection
        exited <- Lq.exitPipelineMode connection
        pure (entered, sent, synced, first, failed, aborted, syncResult, exited)

    String
-> (ByteString -> IO ()) -> SpecWith (Arg (ByteString -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"returns a sync result when called without prior commands" \ByteString
conninfo ->
      Adapter
-> ByteString
-> (Connection
    -> IO
         (Bool, Bool, Maybe ResultObservation, Maybe ResultObservation,
          Bool))
-> IO ()
forall a.
(Eq a, Show a, HasCallStack) =>
Adapter -> ByteString -> (Connection -> IO a) -> IO ()
differential Adapter
adapter ByteString
conninfo \Connection
connection -> do
        entered <- Connection -> IO Bool
Lq.enterPipelineMode Connection
connection
        synced <- Lq.pipelineSync connection
        syncResult <- takeResult connection
        trailing <- takeResult connection
        exited <- Lq.exitPipelineMode connection
        pure (entered, synced, syncResult, trailing, exited)