Progress

hspec/Hasql/PipelineSpec.hs

@@ -9,7 +9,7 @@ spec :: Spec
 spec = do
   describe "Single-statement" do
     describe "Unprepared" do
-      fit "Collects results and sends params" do
+      it "Collects results and sends params" do
         result <-
           Dsl.runPipelineOnLocalDb
             $ GenerateSeries.pipeline False GenerateSeries.Params {start = 0, end = 2}
@@ -23,14 +23,6 @@ spec = do
         shouldBe result (Right [0 .. 2])
 
   describe "Normally" do
-    describe "On prepared statements" do
-      it "Collects results and sends params" do
-        result <-
-          Dsl.runPipelineOnLocalDb
-            $ replicateM 2
-            $ GenerateSeries.pipeline True GenerateSeries.Params {start = 0, end = 2}
-        shouldBe result (Right [[0 .. 2], [0 .. 2]])
-
     describe "On unprepared statements" do
       it "Collects results and sends params" do
         result <-
@@ -39,6 +31,14 @@ spec = do
             $ GenerateSeries.pipeline False GenerateSeries.Params {start = 0, end = 2}
         shouldBe result (Right [[0 .. 2], [0 .. 2]])
 
+    describe "On prepared statements" do
+      it "Collects results and sends params" do
+        result <-
+          Dsl.runPipelineOnLocalDb
+            $ replicateM 2
+            $ GenerateSeries.pipeline True GenerateSeries.Params {start = 0, end = 2}
+        shouldBe result (Right [[0 .. 2], [0 .. 2]])
+
   describe "When some part fails" do
     it "Works" do
       pending

library/Hasql/Decoders/Result.hs

@@ -19,6 +19,11 @@ run :: Result a -> Bool -> LibPQ.Result -> IO (Either ResultError a)
 run (Result reader) idt result =
   runExceptT (runReaderT reader (idt, result))
 
+{-# INLINE pipelineSync #-}
+pipelineSync :: Result ()
+pipelineSync =
+  checkExecStatus [LibPQ.PipelineSync]
+
 {-# INLINE noResult #-}
 noResult :: Result ()
 noResult =

library/Hasql/Pipeline/Core.hs

@@ -32,8 +32,8 @@ run (Pipeline send) (Connection.Connection pqConnectionRef integerDatetimes regi
         runCommandFailing pqConnection $ Pq.pipelineSync pqConnection
         putStrLn "recv"
         recvResult <- recv
-        putStrLn "dropRemainders"
-        handleEither =<< Decoders.Results.run Decoders.Results.dropRemainders pqConnection integerDatetimes
+        putStrLn "pipelineSync"
+        handleEither =<< Decoders.Results.run (Decoders.Results.single Decoders.Result.pipelineSync) pqConnection integerDatetimes
         putStrLn "exitPipelineMode"
         runCommandFailing pqConnection $ Pq.exitPipelineMode pqConnection
         putStrLn "return"
@@ -100,14 +100,11 @@ statement params (Statement.Statement sql (Encoders.Params encoder) (Decoders.Re
                       then pure (True, Right (key, recv))
                       else (False,) . Left . commandToQueryError . ClientError <$> Pq.errorMessage connection
                   where
-                    recv :: IO (Either QueryError ())
-                    recv = do
-                      Pq.getResult connection >>= \case
-                        Nothing ->
-                          Left . commandToQueryError . ClientError <$> Pq.errorMessage connection
-                        Just result ->
-                          mapLeft (commandToQueryError . ResultError)
-                            <$> Decoders.Result.run Decoders.Result.noResult integerDatetimes result
+                    recv =
+                      fmap (mapLeft commandToQueryError)
+                        $ (<*)
+                        <$> Decoders.Results.run (Decoders.Results.single Decoders.Result.noResult) connection integerDatetimes
+                        <*> Decoders.Results.run Decoders.Results.dropRemainders connection integerDatetimes
                 onOldRemoteKey key =
                   pure (Right (key, pure (Right ())))