bolty-streamly (empty) → 0.1.0.0
raw patch · 7 files changed
+1428/−0 lines, 7 filesdep +basedep +boltydep +bolty-streamly
Dependencies added: base, bolty, bolty-streamly, data-default, packstream-bolt, sandwich, streamly-core, text, unordered-containers, validation-selective, vector
Files
- LICENSE +52/−0
- bolty-streamly.cabal +164/−0
- changelog.md +16/−0
- integration/Main.hs +449/−0
- readme.md +144/−0
- src/Database/Bolty/Streamly.hs +362/−0
- test/Main.hs +241/−0
+ LICENSE view
@@ -0,0 +1,52 @@+Apache License+Version 2.0, January 2004+http://www.apache.org/licenses/++TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION++1. Definitions.++"License" shall mean the terms and conditions for use, reproduction, and distribution as defined by Sections 1 through 9 of this document.++"Licensor" shall mean the copyright owner or entity authorized by the copyright owner that is granting the License.++"Legal Entity" shall mean the union of the acting entity and all other entities that control, are controlled by, or are under common control with that entity. For the purposes of this definition, "control" means (i) the power, direct or indirect, to cause the direction or management of such entity, whether by contract or otherwise, or (ii) ownership of fifty percent (50%) or more of the outstanding shares, or (iii) beneficial ownership of such entity.++"You" (or "Your") shall mean an individual or Legal Entity exercising permissions granted by this License.++"Source" form shall mean the preferred form for making modifications, including but not limited to software source code, documentation source, and configuration files.++"Object" form shall mean any form resulting from mechanical transformation or translation of a Source form, including but not limited to compiled object code, generated documentation, and conversions to other media types.++"Work" shall mean the work of authorship, whether in Source or Object form, made available under the License, as indicated by a copyright notice that is included in or attached to the work (an example is provided in the Appendix below).++"Derivative Works" shall mean any work, whether in Source or Object form, that is based on (or derived from) the Work and for which the editorial revisions, annotations, elaborations, or other modifications represent, as a whole, an original work of authorship. For the purposes of this License, Derivative Works shall not include works that remain separable from, or merely link (or bind by name) to the interfaces of, the Work and Derivative Works thereof.++"Contribution" shall mean any work of authorship, including the original version of the Work and any modifications or additions to that Work or Derivative Works thereof, that is intentionally submitted to Licensor for inclusion in the Work by the copyright owner or by an individual or Legal Entity authorized to submit on behalf of the copyright owner. For the purposes of this definition, "submitted" means any form of electronic, verbal, or written communication sent to the Licensor or its representatives, including but not limited to communication on electronic mailing lists, source code control systems, and issue tracking systems that are managed by, or on behalf of, the Licensor for the purpose of discussing and improving the Work, but excluding communication that is conspicuously marked or otherwise designated in writing by the copyright owner as "Not a Contribution."++"Contributor" shall mean Licensor and any individual or Legal Entity on behalf of whom a Contribution has been received by Licensor and subsequently incorporated within the Work.++2. Grant of Copyright License. Subject to the terms and conditions of this License, each Contributor hereby grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable copyright license to reproduce, prepare Derivative Works of, publicly display, publicly perform, sublicense, and distribute the Work and such Derivative Works in Source or Object form.++3. Grant of Patent License. Subject to the terms and conditions of this License, each Contributor hereby grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable (except as stated in this section) patent license to make, have made, use, offer to sell, sell, import, and otherwise transfer the Work, where such license applies only to those patent claims licensable by such Contributor that are necessarily infringed by their Contribution(s) alone or by combination of their Contribution(s) with the Work to which such Contribution(s) was submitted. If You institute patent litigation against any entity (including a cross-claim or counterclaim in a lawsuit) alleging that the Work or a Contribution incorporated within the Work constitutes direct or contributory patent infringement, then any patent licenses granted to You under this License for that Work shall terminate as of the date such litigation is filed.++4. Redistribution. You may reproduce and distribute copies of the Work or Derivative Works thereof in any medium, with or without modifications, and in Source or Object form, provided that You meet the following conditions:++ You must give any other recipients of the Work or Derivative Works a copy of this License; and+ You must cause any modified files to carry prominent notices stating that You changed the files; and+ You must retain, in the Source form of any Derivative Works that You distribute, all copyright, patent, trademark, and attribution notices from the Source form of the Work, excluding those notices that do not pertain to any part of the Derivative Works; and+ If the Work includes a "NOTICE" text file as part of its distribution, then any Derivative Works that You distribute must include a readable copy of the attribution notices contained within such NOTICE file, excluding those notices that do not pertain to any part of the Derivative Works, in at least one of the following places: within a NOTICE text file distributed as part of the Derivative Works; within the Source form or documentation, if provided along with the Derivative Works; or, within a display generated by the Derivative Works, if and wherever such third-party notices normally appear. The contents of the NOTICE file are for informational purposes only and do not modify the License. You may add Your own attribution notices within Derivative Works that You distribute, alongside or as an addendum to the NOTICE text from the Work, provided that such additional attribution notices cannot be construed as modifying the License.++ You may add Your own copyright statement to Your modifications and may provide additional or different license terms and conditions for use, reproduction, or distribution of Your modifications, or for any such Derivative Works as a whole, provided Your use, reproduction, and distribution of the Work otherwise complies with the conditions stated in this License.++5. Submission of Contributions. Unless You explicitly state otherwise, any Contribution intentionally submitted for inclusion in the Work by You to the Licensor shall be under the terms and conditions of this License, without any additional terms or conditions. Notwithstanding the above, nothing herein shall supersede or modify the terms of any separate license agreement you may have executed with Licensor regarding such Contributions.++6. Trademarks. This License does not grant permission to use the trade names, trademarks, service marks, or product names of the Licensor, except as required for reasonable and customary use in describing the origin of the Work and reproducing the content of the NOTICE file.++7. Disclaimer of Warranty. Unless required by applicable law or agreed to in writing, Licensor provides the Work (and each Contributor provides its Contributions) on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied, including, without limitation, any warranties or conditions of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A PARTICULAR PURPOSE. You are solely responsible for determining the appropriateness of using or redistributing the Work and assume any risks associated with Your exercise of permissions under this License.++8. Limitation of Liability. In no event and under no legal theory, whether in tort (including negligence), contract, or otherwise, unless required by applicable law (such as deliberate and grossly negligent acts) or agreed to in writing, shall any Contributor be liable to You for damages, including any direct, indirect, special, incidental, or consequential damages of any character arising as a result of this License or out of the use or inability to use the Work (including but not limited to damages for loss of goodwill, work stoppage, computer failure or malfunction, or any and all other commercial damages or losses), even if such Contributor has been advised of the possibility of such damages.++9. Accepting Warranty or Additional Liability. While redistributing the Work or Derivative Works thereof, You may choose to offer, and charge a fee for, acceptance of support, warranty, indemnity, or other liability obligations and/or rights consistent with this License. However, in accepting such obligations, You may act only on Your own behalf and on Your sole responsibility, not on behalf of any other Contributor, and only if You agree to indemnify, defend, and hold each Contributor harmless for any liability incurred by, or claims asserted against, such Contributor by reason of your accepting any such warranty or additional liability.++END OF TERMS AND CONDITIONS
+ bolty-streamly.cabal view
@@ -0,0 +1,164 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.38.1.+--+-- see: https://github.com/sol/hpack++name: bolty-streamly+version: 0.1.0.0+synopsis: Streamly streaming interface for bolty Neo4j driver+description: Streaming queries over Neo4j using streamly streams.+ Wraps bolty's query functions with streamly's Stream type for+ lazy record-by-record consumption.+category: Database+homepage: https://github.com/philippedev101/bolty-streamly#readme+bug-reports: https://github.com/philippedev101/bolty-streamly/issues+author: philippedev101+maintainer: philippedev101@gmail.com+copyright: (c) 2023-2026 philippedev101+license: Apache-2.0+license-file: LICENSE+build-type: Simple+tested-with:+ GHC == 9.6.7 || == 9.8.4 || == 9.10.3 || == 9.12.3+extra-source-files:+ readme.md+ changelog.md++source-repository head+ type: git+ location: https://github.com/philippedev101/bolty-streamly++flag dev+ description: Turn on -Werror and other developer flags+ manual: True+ default: False++library+ exposed-modules:+ Database.Bolty.Streamly+ other-modules:+ Paths_bolty_streamly+ hs-source-dirs:+ src+ default-extensions:+ DataKinds+ DefaultSignatures+ DeriveAnyClass+ DerivingVia+ DuplicateRecordFields+ FunctionalDependencies+ GADTs+ LambdaCase+ MultiWayIf+ OverloadedRecordDot+ OverloadedStrings+ PartialTypeSignatures+ RecordWildCards+ RoleAnnotations+ TypeFamilies+ UnboxedSums+ UnboxedTuples+ UnliftedDatatypes+ UnliftedNewtypes+ ViewPatterns+ ghc-options: -Weverything -Wno-all-missed-specialisations -Wno-implicit-prelude -Wno-missed-specialisations -Wno-missing-import-lists -Wno-missing-safe-haskell-mode -Wno-name-shadowing -Wno-prepositive-qualified-module -Wno-safe -Wno-unsafe+ build-depends:+ base >=4.18 && <5+ , bolty >=0.1.0.2 && <0.2+ , packstream-bolt ==0.1.*+ , streamly-core ==0.2.*+ , text >=2.0 && <2.2+ , unordered-containers ==0.2.*+ , vector ==0.13.*+ default-language: GHC2021+ if flag(dev)+ ghc-options: -Werror++test-suite bolty-streamly-integration+ type: exitcode-stdio-1.0+ main-is: Main.hs+ other-modules:+ Paths_bolty_streamly+ hs-source-dirs:+ integration+ default-extensions:+ DataKinds+ DefaultSignatures+ DeriveAnyClass+ DerivingVia+ DuplicateRecordFields+ FunctionalDependencies+ GADTs+ LambdaCase+ MultiWayIf+ OverloadedRecordDot+ OverloadedStrings+ PartialTypeSignatures+ RecordWildCards+ RoleAnnotations+ TypeFamilies+ UnboxedSums+ UnboxedTuples+ UnliftedDatatypes+ UnliftedNewtypes+ ViewPatterns+ ghc-options: -Weverything -Wno-all-missed-specialisations -Wno-implicit-prelude -Wno-missed-specialisations -Wno-missing-import-lists -Wno-missing-safe-haskell-mode -Wno-name-shadowing -Wno-prepositive-qualified-module -Wno-safe -Wno-unsafe -threaded -rtsopts -with-rtsopts=-N -Wno-missing-export-lists -Wno-missing-local-signatures -Wno-monomorphism-restriction -Wno-unused-imports -Wno-unused-packages+ build-depends:+ base >=4.18 && <5+ , bolty >=0.1.0.2 && <0.2+ , bolty-streamly+ , data-default+ , packstream-bolt ==0.1.*+ , sandwich+ , streamly-core ==0.2.*+ , text >=2.0 && <2.2+ , unordered-containers ==0.2.*+ , validation-selective+ , vector ==0.13.*+ default-language: GHC2021+ if flag(dev)+ ghc-options: -Werror++test-suite bolty-streamly-test+ type: exitcode-stdio-1.0+ main-is: Main.hs+ other-modules:+ Paths_bolty_streamly+ hs-source-dirs:+ test+ default-extensions:+ DataKinds+ DefaultSignatures+ DeriveAnyClass+ DerivingVia+ DuplicateRecordFields+ FunctionalDependencies+ GADTs+ LambdaCase+ MultiWayIf+ OverloadedRecordDot+ OverloadedStrings+ PartialTypeSignatures+ RecordWildCards+ RoleAnnotations+ TypeFamilies+ UnboxedSums+ UnboxedTuples+ UnliftedDatatypes+ UnliftedNewtypes+ ViewPatterns+ ghc-options: -Weverything -Wno-all-missed-specialisations -Wno-implicit-prelude -Wno-missed-specialisations -Wno-missing-import-lists -Wno-missing-safe-haskell-mode -Wno-name-shadowing -Wno-prepositive-qualified-module -Wno-safe -Wno-unsafe -threaded -rtsopts -with-rtsopts=-N -Wno-missing-export-lists -Wno-missing-local-signatures -Wno-monomorphism-restriction -Wno-unused-imports -Wno-unused-packages+ build-depends:+ base >=4.18 && <5+ , bolty >=0.1.0.2 && <0.2+ , bolty-streamly+ , packstream-bolt ==0.1.*+ , sandwich+ , streamly-core ==0.2.*+ , text >=2.0 && <2.2+ , unordered-containers ==0.2.*+ , vector ==0.13.*+ default-language: GHC2021+ if flag(dev)+ ghc-options: -Werror
+ changelog.md view
@@ -0,0 +1,16 @@+# Changelog for `bolty-streamly`++All notable changes to this project will be documented in this file.++The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),+and this project adheres to the+[Haskell Package Versioning Policy](https://pvp.haskell.org/).++## 0.1.0.0++- Initial release+- Streamly Stream interface for bolty queries+- Lazy record-by-record consumption of Neo4j query results+- Connection pool, routing pool, and session streaming with automatic resource management via `bracketIO`+- Streams are composable values — store, pass, and combine them freely+- Typed decoding variants (`As` suffix) with `RowDecoder`
+ integration/Main.hs view
@@ -0,0 +1,449 @@+module Main where++import Test.Sandwich++import Control.Monad.IO.Class (liftIO)+import Data.Default (def)+import qualified Data.HashMap.Lazy as H+import qualified Data.Text as T+import qualified Data.Vector as V+import qualified Validation as Val+import System.Environment (lookupEnv)++import Database.Bolty+import Database.Bolty.Connection (queryIO)+import Data.PackStream.Ps (Ps(..))+import Data.PackStream.Integer (fromPSInteger)+import Data.Int (Int64)+import qualified Control.Exception as E+import Database.Bolty.Decode (DecodeError(..), RowDecoder, column,+ field, int64, text)+import Database.Bolty.Streamly (queryStream, queryStreamP,+ queryStreamAs, queryStreamPAs,+ poolStream, poolStreamP,+ poolStreamAs,+ routingStream, routingStreamP,+ sessionReadStream, sessionReadStreamP,+ sessionWriteStream, sessionWriteStreamP,+ sessionReadStreamAs)+import qualified Streamly.Data.Stream as Stream+import qualified Streamly.Data.Fold as Fold+++-- | Neo4j test config (same as bolty integration tests)+testConfig :: IO Config+testConfig = do+ h <- maybe "127.0.0.1" T.pack <$> lookupEnv "NEO4J_HOST"+ p <- maybe 7687 read <$> lookupEnv "NEO4J_PORT"+ pure def+ { host = h+ , port = p+ , scheme = Basic "neo4j" "testpassword"+ , use_tls = False+ }+++getConfig :: IO ValidatedConfig+getConfig = do+ cfg <- testConfig+ case validateConfig cfg of+ Val.Failure errs -> fail $ "Config invalid: " <> show errs+ Val.Success vc -> pure vc+++getRoutingConfig :: IO ValidatedConfig+getRoutingConfig = do+ cfg <- testConfig+ let routingCfg = cfg { routing = Routing }+ case validateConfig routingCfg of+ Val.Failure errs -> fail $ "Config invalid: " <> show errs+ Val.Success vc -> pure vc+++withConn :: (Connection -> IO a) -> IO a+withConn f = do+ cfg <- getConfig+ conn <- connect cfg+ result <- f conn+ close conn+ pure result+++streamingTests :: TopSpec+streamingTests = describe "Streaming queries" $ do++ it "queryStream returns records one by one" $ do+ result <- liftIO $ withConn $ \p -> do+ s <- queryStream p "UNWIND range(1, 5) AS n RETURN n"+ Stream.fold Fold.toList s+ length result `shouldBe` 5++ it "queryStream returns correct values" $ do+ result <- liftIO $ withConn $ \p -> do+ s <- queryStream p "UNWIND range(1, 3) AS n RETURN n"+ Stream.fold Fold.toList s+ let values = map (\r -> asInt (V.head r)) result+ values `shouldBe` [Just 1, Just 2, Just 3]++ it "queryStream handles empty result set" $ do+ result <- liftIO $ withConn $ \p -> do+ s <- queryStream p "UNWIND [] AS n RETURN n"+ Stream.fold Fold.toList s+ length result `shouldBe` 0++ it "queryStream works with single row" $ do+ result <- liftIO $ withConn $ \p -> do+ s <- queryStream p "RETURN 42 AS answer"+ Stream.fold Fold.toList s+ length result `shouldBe` 1+ case asInt (V.head (head result)) of+ Just n -> n `shouldBe` 42+ Nothing -> expectationFailure "Expected 42"++ it "queryStreamP passes parameters" $ do+ result <- liftIO $ withConn $ \p -> do+ s <- queryStreamP p "RETURN $x AS val" (H.singleton "x" (PsInteger 99))+ Stream.fold Fold.toList s+ case asInt (V.head (head result)) of+ Just n -> n `shouldBe` 99+ Nothing -> expectationFailure "Expected 99"++ it "queryStreamP with list parameter" $ do+ result <- liftIO $ withConn $ \p -> do+ s <- queryStreamP p "UNWIND $xs AS x RETURN x"+ (H.singleton "xs" (PsList $ V.fromList [PsInteger 10, PsInteger 20, PsInteger 30]))+ Stream.fold Fold.toList s+ length result `shouldBe` 3+ let values = map (\r -> asInt (V.head r)) result+ values `shouldBe` [Just 10, Just 20, Just 30]+++streamFoldTests :: TopSpec+streamFoldTests = describe "Stream fold operations" $ do++ it "fold with Fold.length counts records" $ do+ count <- liftIO $ withConn $ \p -> do+ s <- queryStream p "UNWIND range(1, 10) AS n RETURN n"+ Stream.fold Fold.length s+ count `shouldBe` 10++ it "fold with Fold.sum sums values" $ do+ total <- liftIO $ withConn $ \p -> do+ s <- queryStream p "UNWIND range(1, 5) AS n RETURN n"+ Stream.fold (Fold.lmap (\r -> maybe 0 id (asInt (V.head r) >>= fromPSInteger)) Fold.sum) s+ (total :: Int) `shouldBe` 15++ it "Stream.take limits records" $ do+ result <- liftIO $ withConn $ \p -> do+ s <- queryStream p "UNWIND range(1, 100) AS n RETURN n"+ Stream.fold Fold.toList $ Stream.take 3 s+ length result `shouldBe` 3+++streamTransactionTests :: TopSpec+streamTransactionTests = describe "Streaming in transactions" $ do++ it "queryStream inside withTransaction" $ do+ result <- liftIO $ withConn $ \p ->+ withTransaction p $ \conn -> do+ s <- queryStream conn "UNWIND range(1, 5) AS n RETURN n"+ Stream.fold Fold.toList s+ length result `shouldBe` 5++ it "multiple queryStream calls in a transaction" $ do+ (r1, r2) <- liftIO $ withConn $ \p ->+ withTransaction p $ \conn -> do+ s1 <- queryStream conn "UNWIND [1,2,3] AS n RETURN n"+ res1 <- Stream.fold Fold.toList s1+ s2 <- queryStream conn "UNWIND [4,5,6] AS n RETURN n"+ res2 <- Stream.fold Fold.toList s2+ pure (res1, res2)+ length r1 `shouldBe` 3+ length r2 `shouldBe` 3++ it "queryStream with write in transaction" $ do+ liftIO $ withConn $ \p ->+ withTransaction p $ \conn -> do+ _ <- queryIO conn "CREATE (n:TestStreamTx {value: 1})"+ s <- queryStream conn "MATCH (n:TestStreamTx) RETURN n.value AS v"+ result <- Stream.fold Fold.toList s+ length result `shouldBe` 1+ -- Cleanup+ liftIO $ withConn $ \p -> queryIO p "MATCH (n:TestStreamTx) DELETE n" >> pure ()+++largeResultTests :: TopSpec+largeResultTests = describe "Large result streaming" $ do++ it "streams 1000 rows without buffering all" $ do+ count <- liftIO $ withConn $ \p -> do+ s <- queryStream p "UNWIND range(1, 1000) AS n RETURN n"+ Stream.fold Fold.length s+ count `shouldBe` 1000++ it "streams 10000 rows" $ do+ count <- liftIO $ withConn $ \p -> do+ s <- queryStream p "UNWIND range(1, 10000) AS n RETURN n"+ Stream.fold Fold.length s+ count `shouldBe` 10000+++-- ================================================================+-- Pool-based streaming tests+-- ================================================================++poolStreamTests :: TopSpec+poolStreamTests = describe "Pool-based streaming" $ do++ it "poolStream streams records" $ do+ cfg <- liftIO getConfig+ pool <- liftIO $ createPool cfg defaultPoolConfig+ result <- liftIO $ Stream.fold Fold.toList $+ poolStream pool "UNWIND range(1, 5) AS n RETURN n"+ length result `shouldBe` 5+ liftIO $ destroyPool pool++ it "poolStreamP passes parameters" $ do+ cfg <- liftIO getConfig+ pool <- liftIO $ createPool cfg defaultPoolConfig+ result <- liftIO $ Stream.fold Fold.toList $+ poolStreamP pool "RETURN $x AS n" (H.singleton "x" (PsInteger 42))+ case asInt (V.head (head result)) of+ Just n -> n `shouldBe` 42+ Nothing -> expectationFailure "Expected 42"+ liftIO $ destroyPool pool++ it "poolStream works with empty result" $ do+ cfg <- liftIO getConfig+ pool <- liftIO $ createPool cfg defaultPoolConfig+ result <- liftIO $ Stream.fold Fold.toList $+ poolStream pool "UNWIND [] AS n RETURN n"+ length result `shouldBe` 0+ liftIO $ destroyPool pool++ it "poolStream with fold length" $ do+ cfg <- liftIO getConfig+ pool <- liftIO $ createPool cfg defaultPoolConfig+ count <- liftIO $ Stream.fold Fold.length $+ poolStream pool "UNWIND range(1, 100) AS n RETURN n"+ count `shouldBe` 100+ liftIO $ destroyPool pool+++-- ================================================================+-- Routing pool streaming tests+-- ================================================================++routingStreamTests :: TopSpec+routingStreamTests = describe "Routing pool streaming" $ do++ it "routingStream ReadAccess streams records" $ do+ cfg <- liftIO getRoutingConfig+ rp <- liftIO $ createRoutingPool cfg defaultRoutingPoolConfig+ result <- liftIO $ Stream.fold Fold.toList $+ routingStream rp ReadAccess "UNWIND range(1, 5) AS n RETURN n"+ length result `shouldBe` 5+ liftIO $ destroyRoutingPool rp++ it "routingStream WriteAccess creates data" $ do+ cfg <- liftIO getRoutingConfig+ rp <- liftIO $ createRoutingPool cfg defaultRoutingPoolConfig+ _ <- liftIO $ Stream.fold Fold.toList $+ routingStream rp WriteAccess+ "CREATE (n:TestRoutingStream {value: 1}) RETURN n.value AS v"+ -- Verify data was created+ result <- liftIO $ Stream.fold Fold.toList $+ routingStream rp ReadAccess+ "MATCH (n:TestRoutingStream) RETURN n.value AS v"+ V.length (V.fromList result) `shouldBe` 1+ -- Cleanup+ liftIO $ withRoutingConnection rp WriteAccess $ \p ->+ queryIO p "MATCH (n:TestRoutingStream) DELETE n" >> pure ()+ liftIO $ destroyRoutingPool rp++ it "routingStreamP passes parameters" $ do+ cfg <- liftIO getRoutingConfig+ rp <- liftIO $ createRoutingPool cfg defaultRoutingPoolConfig+ result <- liftIO $ Stream.fold Fold.toList $+ routingStreamP rp ReadAccess "RETURN $x AS n"+ (H.singleton "x" (PsInteger 99))+ case asInt (V.head (head result)) of+ Just n -> n `shouldBe` 99+ Nothing -> expectationFailure "Expected 99"+ liftIO $ destroyRoutingPool rp++ it "routingStream handles large results" $ do+ cfg <- liftIO getRoutingConfig+ rp <- liftIO $ createRoutingPool cfg defaultRoutingPoolConfig+ count <- liftIO $ Stream.fold Fold.length $+ routingStream rp ReadAccess "UNWIND range(1, 1000) AS n RETURN n"+ count `shouldBe` 1000+ liftIO $ destroyRoutingPool rp+++-- ================================================================+-- Session streaming tests+-- ================================================================++sessionStreamTests :: TopSpec+sessionStreamTests = describe "Session streaming" $ do++ it "sessionReadStream reads data" $ do+ cfg <- liftIO getConfig+ pool <- liftIO $ createPool cfg defaultPoolConfig+ session <- liftIO $ createSession pool defaultSessionConfig+ result <- liftIO $ Stream.fold Fold.toList $+ sessionReadStream session "UNWIND range(1, 5) AS n RETURN n"+ length result `shouldBe` 5+ liftIO $ destroyPool pool++ it "sessionReadStreamP passes parameters" $ do+ cfg <- liftIO getConfig+ pool <- liftIO $ createPool cfg defaultPoolConfig+ session <- liftIO $ createSession pool defaultSessionConfig+ result <- liftIO $ Stream.fold Fold.toList $+ sessionReadStreamP session "RETURN $x AS n" (H.singleton "x" (PsInteger 42))+ case asInt (V.head (head result)) of+ Just n -> n `shouldBe` 42+ Nothing -> expectationFailure "Expected 42"+ liftIO $ destroyPool pool++ it "sessionWriteStream creates data and produces bookmark" $ do+ cfg <- liftIO getConfig+ pool <- liftIO $ createPool cfg defaultPoolConfig+ session <- liftIO $ createSession pool defaultSessionConfig+ bms0 <- liftIO $ getLastBookmarks session+ bms0 `shouldBe` []+ _ <- liftIO $ Stream.fold Fold.toList $+ sessionWriteStream session+ "CREATE (n:TestSessionStream {value: 1}) RETURN n.value AS v"+ bms1 <- liftIO $ getLastBookmarks session+ length bms1 `shouldBe` 1+ -- Cleanup+ liftIO $ withConnection pool $ \p ->+ queryIO p "MATCH (n:TestSessionStream) DELETE n" >> pure ()+ liftIO $ destroyPool pool++ it "sessionWriteStreamP with parameters" $ do+ cfg <- liftIO getConfig+ pool <- liftIO $ createPool cfg defaultPoolConfig+ session <- liftIO $ createSession pool defaultSessionConfig+ _ <- liftIO $ Stream.fold Fold.toList $+ sessionWriteStreamP session+ "CREATE (n:TestSessionStreamP {value: $v}) RETURN n.value AS v"+ (H.singleton "v" (PsInteger 77))+ -- Verify+ result <- liftIO $ Stream.fold Fold.toList $+ sessionReadStream session+ "MATCH (n:TestSessionStreamP) RETURN n.value AS v"+ case asInt (V.head (head result)) of+ Just n -> n `shouldBe` 77+ Nothing -> expectationFailure "Expected 77"+ -- Cleanup+ liftIO $ withConnection pool $ \p ->+ queryIO p "MATCH (n:TestSessionStreamP) DELETE n" >> pure ()+ liftIO $ destroyPool pool+++-- ================================================================+-- Routing session streaming tests+-- ================================================================++routingSessionStreamTests :: TopSpec+routingSessionStreamTests = describe "Routing session streaming" $ do++ it "routing sessionReadStream works" $ do+ cfg <- liftIO getRoutingConfig+ rp <- liftIO $ createRoutingPool cfg defaultRoutingPoolConfig+ session <- liftIO $ createRoutingSession rp defaultSessionConfig+ result <- liftIO $ Stream.fold Fold.toList $+ sessionReadStream session "UNWIND range(1, 3) AS n RETURN n"+ length result `shouldBe` 3+ liftIO $ destroyRoutingPool rp++ it "routing sessionWriteStream produces bookmark" $ do+ cfg <- liftIO getRoutingConfig+ rp <- liftIO $ createRoutingPool cfg defaultRoutingPoolConfig+ session <- liftIO $ createRoutingSession rp defaultSessionConfig+ _ <- liftIO $ Stream.fold Fold.toList $+ sessionWriteStream session+ "CREATE (n:TestRoutingSessStream {value: 1}) RETURN n.value AS v"+ bms <- liftIO $ getLastBookmarks session+ length bms `shouldBe` 1+ -- Cleanup+ liftIO $ withRoutingConnection rp WriteAccess $ \p ->+ queryIO p "MATCH (n:TestRoutingSessStream) DELETE n" >> pure ()+ liftIO $ destroyRoutingPool rp+++-- ================================================================+-- Streaming decode tests+-- ================================================================++streamingDecodeTests :: TopSpec+streamingDecodeTests = describe "Streaming decode" $ do++ it "queryStreamAs decodes RETURN 1 AS n with column 0 int64" $ do+ result <- liftIO $ withConn $ \conn -> do+ s <- queryStreamAs (column 0 int64) conn "RETURN 1 AS n"+ Stream.fold Fold.toList s+ result `shouldBe` [1 :: Int64]++ it "queryStreamPAs decodes with parameters" $ do+ result <- liftIO $ withConn $ \conn -> do+ s <- queryStreamPAs (column 0 int64) conn "RETURN $x AS n"+ (H.singleton "x" (PsInteger 42))+ Stream.fold Fold.toList s+ result `shouldBe` [42 :: Int64]++ it "queryStreamAs with field name lookup" $ do+ result <- liftIO $ withConn $ \conn -> do+ s <- queryStreamAs (field "name" text) conn "RETURN 'hello' AS name"+ Stream.fold Fold.toList s+ result `shouldBe` [T.pack "hello"]++ it "queryStreamAs with multi-column Applicative decoder" $ do+ let decoder = (,) <$> field "a" text <*> field "b" int64+ result <- liftIO $ withConn $ \conn -> do+ s <- queryStreamAs decoder conn "UNWIND [{a: 'x', b: 1}, {a: 'y', b: 2}] AS row RETURN row.a AS a, row.b AS b"+ Stream.fold Fold.toList s+ result `shouldBe` [(T.pack "x", 1), (T.pack "y", 2)]++ it "queryStreamAs throws DecodeError on type mismatch" $ do+ result <- liftIO $ E.try @DecodeError $ withConn $ \conn -> do+ s <- queryStreamAs (column 0 int64) conn "RETURN 'not_a_number' AS n"+ Stream.fold Fold.toList s+ case result of+ Left (TypeMismatch _ _) -> pure ()+ Left err -> expectationFailure $ "Expected TypeMismatch, got: " <> show err+ Right _ -> expectationFailure "Expected DecodeError"++ it "poolStreamAs decodes through pool" $ do+ cfg <- liftIO getConfig+ pool <- liftIO $ createPool cfg defaultPoolConfig+ result <- liftIO $ Stream.fold Fold.toList $+ poolStreamAs (column 0 int64) pool "RETURN 99 AS n"+ result `shouldBe` [99 :: Int64]+ liftIO $ destroyPool pool++ it "sessionReadStreamAs decodes through session" $ do+ cfg <- liftIO getConfig+ pool <- liftIO $ createPool cfg defaultPoolConfig+ session <- liftIO $ createSession pool defaultSessionConfig+ result <- liftIO $ Stream.fold Fold.toList $+ sessionReadStreamAs (column 0 int64) session "RETURN 7 AS n"+ result `shouldBe` [7 :: Int64]+ liftIO $ destroyPool pool+++main :: IO ()+main = runSandwichWithCommandLineArgs defaultOptions $ do+ streamingTests+ streamFoldTests+ streamTransactionTests+ largeResultTests+ poolStreamTests+ routingStreamTests+ sessionStreamTests+ routingSessionStreamTests+ streamingDecodeTests
+ readme.md view
@@ -0,0 +1,144 @@+# bolty-streamly++Streaming interface for the [bolty](https://github.com/philippedev101/bolty) Neo4j driver, built on [streamly](https://hackage.haskell.org/package/streamly-core).++## Why streaming?++bolty's standard `query` functions buffer the entire result set into a `Vector` before returning. For large result sets (millions of rows, graph traversals, data exports), this can use a lot of memory.++bolty-streamly yields records **one at a time** as they arrive from the server, giving you constant-memory consumption regardless of result set size. Records are pulled from Neo4j in batches via the BOLT protocol's PULL mechanism, but exposed as a single `Stream IO` — you never have to think about batching.++## Quick start++```haskell+import qualified Database.Bolty as Bolt+import qualified Database.Bolty.Streamly as BoltS+import qualified Streamly.Data.Stream as Stream+import qualified Streamly.Data.Fold as Fold+import Data.Default (def)++main :: IO ()+main = do+ let cfg = def{ Bolt.scheme = Bolt.Basic "neo4j" "password", Bolt.use_tls = False }+ case Bolt.validateConfig cfg of+ Failure _ -> error "bad config"+ Success vc -> do+ pool <- Bolt.createPool vc Bolt.defaultPoolConfig+ let s = BoltS.poolStream pool "MATCH (n:Person) RETURN n.name AS name"+ count <- Stream.fold Fold.length s+ putStrLn $ "Processed " <> show count <> " records"+ Bolt.destroyPool pool+```++## API overview++The module exposes four levels of streaming, each with variants for parameters (`P`) and typed decoding (`As`).++Streams from pool, routing, and session functions are ordinary values — you can store them, pass them around, and compose them freely. Connection lifetime is tied to the stream via `bracketIO`: the connection is acquired when consumption begins and released when the stream completes or errors.++### Direct connection++Use when you manage the connection yourself:++```haskell+-- Raw records+queryStream :: Connection -> Text -> IO (Stream IO Record)+queryStreamP :: Connection -> Text -> HashMap Text Ps -> IO (Stream IO Record)++-- Decoded records (throws DecodeError on failure)+queryStreamAs :: RowDecoder a -> Connection -> Text -> IO (Stream IO a)+queryStreamPAs :: RowDecoder a -> Connection -> Text -> HashMap Text Ps -> IO (Stream IO a)+```++### Connection pool++Acquires a pooled connection when the stream is consumed and returns it when the stream finishes:++```haskell+poolStream :: BoltPool -> Text -> Stream IO Record+poolStreamP :: BoltPool -> Text -> HashMap Text Ps -> Stream IO Record+poolStreamAs :: RowDecoder a -> BoltPool -> Text -> Stream IO a+poolStreamPAs :: RowDecoder a -> BoltPool -> Text -> HashMap Text Ps -> Stream IO a+```++Example:++```haskell+pool <- Bolt.createPool vc Bolt.defaultPoolConfig++let people = BoltS.poolStreamAs personDecoder pool "MATCH (p:Person) RETURN p.name, p.age"+Stream.mapM_ print people++Bolt.destroyPool pool+```++### Routing pool (clusters)++Routes queries to the appropriate cluster member based on access mode:++```haskell+routingStream :: RoutingPool -> AccessMode -> Text -> Stream IO Record+routingStreamP :: RoutingPool -> AccessMode -> Text -> HashMap Text Ps -> Stream IO Record+routingStreamAs :: RowDecoder a -> RoutingPool -> AccessMode -> Text -> Stream IO a+routingStreamPAs :: RowDecoder a -> RoutingPool -> AccessMode -> Text -> HashMap Text Ps -> Stream IO a+```++Example:++```haskell+let results = BoltS.routingStreamAs decoder routingPool ReadAccess "MATCH (n) RETURN n"+items <- Stream.fold Fold.toList results+```++### Session (causal consistency)++Runs streaming queries inside managed transactions with automatic bookmark tracking and read/write routing:++```haskell+sessionReadStream :: Session -> Text -> Stream IO Record+sessionWriteStream :: Session -> Text -> Stream IO Record+sessionReadStreamAs :: RowDecoder a -> Session -> Text -> Stream IO a+sessionWriteStreamAs :: RowDecoder a -> Session -> Text -> Stream IO a+-- + P variants for parameters+```++Example:++```haskell+session <- Bolt.createSession pool Bolt.defaultSessionConfig++-- Write some data+Stream.fold Fold.drain $+ BoltS.sessionWriteStream session "CREATE (p:Person {name: 'Alice'})"++-- Read it back (guaranteed to see Alice via bookmarks)+Stream.mapM_ print $+ BoltS.sessionReadStreamAs personDecoder session "MATCH (p:Person) RETURN p.name, p.age"+```++## Low-level: pullStream++If you need to run a query with custom RUN parameters and then stream the PULL phase yourself:++```haskell+pullStream :: Connection -> IO (Stream IO Record)+```++This expects the connection to already be in `Streaming` or `TXstreaming` state (after a RUN has been acknowledged). It handles PULL batching and state transitions automatically.++## Naming convention++| Suffix | Meaning |+|---|---|+| *(none)* | No parameters, raw `Record` stream |+| `P` | With parameters (`HashMap Text Ps`) |+| `As` | Decoded via `RowDecoder a`, no parameters |+| `PAs` | Decoded via `RowDecoder a`, with parameters |++## Supported GHC versions++9.6.7, 9.8.4, 9.10.3, 9.12.3++## License++Apache-2.0
+ src/Database/Bolty/Streamly.hs view
@@ -0,0 +1,362 @@+-- | Streamly streaming interface for bolty Neo4j queries.+--+-- Instead of buffering all result records into a 'V.Vector', this module+-- yields records one-by-one as a @'Stream' IO 'Record'@,+-- allowing constant-memory consumption of large result sets.+--+-- Streams returned by pool, routing, and session functions manage connection+-- lifetime automatically via @bracketIO@ — the connection is acquired when+-- the stream starts and released when it finishes, errors, or is garbage+-- collected. Streams are ordinary values that can be stored, composed,+-- and passed to other functions.+--+-- @+-- import qualified Database.Bolty as Bolt+-- import qualified Database.Bolty.Streamly as BoltS+-- import qualified Streamly.Data.Stream as Stream+-- import qualified Streamly.Data.Fold as Fold+--+-- main :: IO ()+-- main = do+-- pool <- Bolt.createPool cfg Bolt.defaultPoolConfig+-- let s = BoltS.poolStream pool \"MATCH (n) RETURN n\"+-- count <- Stream.fold Fold.length s+-- Bolt.destroyPool pool+-- @+module Database.Bolty.Streamly+ ( -- * Streaming queries (bare connection)+ queryStream+ , queryStreamP+ , queryStreamAs+ , queryStreamPAs+ -- * Low-level streaming+ , pullStream+ -- * Pool-based streaming+ , poolStream+ , poolStreamP+ , poolStreamAs+ , poolStreamPAs+ -- * Routing pool streaming+ , routingStream+ , routingStreamP+ , routingStreamAs+ , routingStreamPAs+ -- * Session streaming+ , sessionReadStream+ , sessionReadStreamP+ , sessionWriteStream+ , sessionWriteStreamP+ , sessionReadStreamAs+ , sessionReadStreamPAs+ , sessionWriteStreamAs+ , sessionWriteStreamPAs+ -- * Re-exports+ , Stream+ ) where++import Control.Exception (throwIO, SomeException, try)+import Data.Kind (Type)+import Data.Text (Text)+import GHC.Stack (HasCallStack)+import qualified Data.HashMap.Lazy as H+import qualified Data.PackStream.Ps as PS+import Data.PackStream.Result (Result(..))+import qualified Data.Vector as V+import Streamly.Data.Stream (Stream)+import qualified Streamly.Data.Stream as Stream++import Database.Bolty.Connection (requestResponseRunIO)+import qualified Database.Bolty.Connection.Pipe as P+import Database.Bolty.Connection.Type+import Database.Bolty.Decode (RowDecoder, decodeRow)+import Database.Bolty.Message.Request (Request(..), defaultPull, Begin(Begin), TelemetryApi(..))+import Database.Bolty.Message.Response (Response(..), Failure(..), successFields)+import Database.Bolty.Pool (BoltPool, CheckedOutConnection(..),+ acquireConnection, releaseConnection,+ releaseConnectionOnError)+import Database.Bolty.Record (Record)+import Database.Bolty.Routing (AccessMode(..), RoutingPool,+ acquireRoutingConnection)+import Database.Bolty.Session (Session(..), SessionPool(..), SessionConfig(..),+ BookmarkManager, getBookmarks, updateBookmark)++import Control.Monad (when)+import Data.IORef (IORef, readIORef, writeIORef)+++-- ---------------------------------------------------------------------------+-- Bare connection streaming (unchanged)+-- ---------------------------------------------------------------------------++-- | Run a Cypher query and return results as a stream of records.+--+-- Records are yielded one at a time as they arrive from the server,+-- without buffering the entire result set in memory.+--+-- Must be called in @Ready@ or @TXready@ state.+queryStream :: HasCallStack => Connection -> Text -> IO (Stream IO Record)+queryStream conn cypher = queryStreamP conn cypher H.empty+++-- | Run a parameterised Cypher query and return results as a stream.+queryStreamP :: HasCallStack => Connection -> Text -> H.HashMap Text PS.Ps -> IO (Stream IO Record)+queryStreamP conn cypher params = do+ _ <- requestResponseRunIO conn cypher params+ pullStream conn+++-- | Run a Cypher query and decode each record using a 'RowDecoder'.+-- Throws 'Database.Bolty.Decode.DecodeError' on decode failure.+queryStreamAs :: HasCallStack => RowDecoder a -> Connection -> Text -> IO (Stream IO a)+queryStreamAs decoder conn cypher = queryStreamPAs decoder conn cypher H.empty+++-- | Run a parameterised Cypher query and decode each record using a 'RowDecoder'.+-- Throws 'Database.Bolty.Decode.DecodeError' on decode failure.+queryStreamPAs :: HasCallStack => RowDecoder a -> Connection -> Text -> H.HashMap Text PS.Ps -> IO (Stream IO a)+queryStreamPAs decoder conn cypher params = do+ runResp <- requestResponseRunIO conn cypher params+ let columns = successFields runResp+ s <- pullStream conn+ pure $ Stream.mapM (decodeOrThrow decoder columns) s+++-- | Decode a single record, throwing 'Database.Bolty.Decode.DecodeError' on failure.+decodeOrThrow :: RowDecoder a -> V.Vector Text -> Record -> IO a+decodeOrThrow decoder columns record =+ case decodeRow decoder columns record of+ Right a -> pure a+ Left err -> throwIO err+++-- | Pull state machine. @NeedPull@ means we need to send a new PULL+-- message to the server. @Done@ means the result set is exhausted.+type PullState :: Type+data PullState = NeedPull | Done+++-- | Stream records from an in-progress PULL.+--+-- Expects the connection to be in @Streaming@ or @TXstreaming@ state+-- (i.e. after a RUN has been sent and acknowledged). Sends PULL messages+-- and yields each 'Record' as it arrives. When the server signals+-- completion, the state transitions back to @Ready@ / @TXready@.+pullStream :: HasCallStack => Connection -> IO (Stream IO Record)+pullStream conn = do+ P.requireStateIO conn [Streaming, TXstreaming] "PULL"+ P.flushIO conn $ RPull defaultPull+ pure $ Stream.unfoldrM (step conn) NeedPull+ where+ step :: HasCallStack => Connection -> PullState -> IO (Maybe (Record, PullState))+ step _ Done = pure Nothing+ step c NeedPull = do+ response <- P.fetchIO c+ case response of+ RRecord record ->+ pure $ Just (record, NeedPull)++ RSuccess meta -> do+ let hasMore = case H.lookup "has_more" meta of+ Just hm -> case PS.fromPs hm of+ Success True -> True+ _ -> False+ Nothing -> False+ if hasMore then do+ P.flushIO c $ RPull defaultPull+ step c NeedPull+ else do+ st <- P.getState c+ P.setState c $ case st of+ TXstreaming -> TXready+ _ -> Ready+ pure Nothing++ RIgnored -> do+ P.reset c+ throwIO ResponseErrorIgnored++ RFailure Failure{code, message} -> do+ P.setState c Failed+ P.reset c+ throwIO $ ResponseErrorFailure code message+++-- ---------------------------------------------------------------------------+-- Pool-based streaming (bracketIO)+-- ---------------------------------------------------------------------------++-- | Run a streaming query using a pooled connection. The connection is+-- checked out when the stream starts and returned when it finishes.+poolStream :: HasCallStack => BoltPool -> Text -> Stream IO Record+poolStream pool cypher = poolStreamP pool cypher H.empty+++-- | Like 'poolStream' but with query parameters.+poolStreamP :: HasCallStack => BoltPool -> Text -> H.HashMap Text PS.Ps -> Stream IO Record+poolStreamP pool cypher params =+ Stream.bracketIO+ (acquireConnection pool)+ releaseConnection+ (\coc -> Stream.concatEffect $ queryStreamP (cocConnection coc) cypher params)+++-- | Like 'poolStream' but decodes each record using a 'RowDecoder'.+poolStreamAs :: HasCallStack => RowDecoder a -> BoltPool -> Text -> Stream IO a+poolStreamAs decoder pool cypher = poolStreamPAs decoder pool cypher H.empty+++-- | Like 'poolStreamP' but decodes each record using a 'RowDecoder'.+poolStreamPAs :: HasCallStack => RowDecoder a -> BoltPool -> Text -> H.HashMap Text PS.Ps -> Stream IO a+poolStreamPAs decoder pool cypher params =+ Stream.bracketIO+ (acquireConnection pool)+ releaseConnection+ (\coc -> Stream.concatEffect $ queryStreamPAs decoder (cocConnection coc) cypher params)+++-- ---------------------------------------------------------------------------+-- Routing pool streaming (bracketIO)+-- ---------------------------------------------------------------------------++-- | Run a streaming query using a routing pool. Automatically selects+-- a server based on the access mode (read replica or leader).+routingStream :: HasCallStack => RoutingPool -> AccessMode -> Text -> Stream IO Record+routingStream rp mode cypher = routingStreamP rp mode cypher H.empty+++-- | Like 'routingStream' but with query parameters.+routingStreamP :: HasCallStack+ => RoutingPool -> AccessMode -> Text -> H.HashMap Text PS.Ps -> Stream IO Record+routingStreamP rp mode cypher params =+ Stream.bracketIO+ (acquireRoutingConnection rp mode)+ releaseConnection+ (\coc -> Stream.concatEffect $ queryStreamP (cocConnection coc) cypher params)+++-- | Like 'routingStream' but decodes each record using a 'RowDecoder'.+routingStreamAs :: HasCallStack => RowDecoder a -> RoutingPool -> AccessMode -> Text -> Stream IO a+routingStreamAs decoder rp mode cypher = routingStreamPAs decoder rp mode cypher H.empty+++-- | Like 'routingStreamP' but decodes each record using a 'RowDecoder'.+routingStreamPAs :: HasCallStack+ => RowDecoder a -> RoutingPool -> AccessMode -> Text -> H.HashMap Text PS.Ps -> Stream IO a+routingStreamPAs decoder rp mode cypher params =+ Stream.bracketIO+ (acquireRoutingConnection rp mode)+ releaseConnection+ (\coc -> Stream.concatEffect $ queryStreamPAs decoder (cocConnection coc) cypher params)+++-- ---------------------------------------------------------------------------+-- Session streaming (bracketIO with BEGIN/COMMIT)+-- ---------------------------------------------------------------------------++-- | Run a streaming read query inside a managed transaction.+-- Handles BEGIN, COMMIT, bookmark propagation, and connection lifecycle.+sessionReadStream :: HasCallStack => Session -> Text -> Stream IO Record+sessionReadStream session cypher = sessionReadStreamP session cypher H.empty+++-- | Like 'sessionReadStream' but with query parameters.+sessionReadStreamP :: HasCallStack => Session -> Text -> H.HashMap Text PS.Ps -> Stream IO Record+sessionReadStreamP session cypher params =+ sessionStreamRaw session ReadAccess cypher params+++-- | Run a streaming write query inside a managed transaction.+sessionWriteStream :: HasCallStack => Session -> Text -> Stream IO Record+sessionWriteStream session cypher = sessionWriteStreamP session cypher H.empty+++-- | Like 'sessionWriteStream' but with query parameters.+sessionWriteStreamP :: HasCallStack => Session -> Text -> H.HashMap Text PS.Ps -> Stream IO Record+sessionWriteStreamP session cypher params =+ sessionStreamRaw session WriteAccess cypher params+++-- | Like 'sessionReadStream' but decodes each record using a 'RowDecoder'.+sessionReadStreamAs :: HasCallStack => RowDecoder a -> Session -> Text -> Stream IO a+sessionReadStreamAs decoder session cypher =+ sessionReadStreamPAs decoder session cypher H.empty+++-- | Like 'sessionReadStreamP' but decodes each record using a 'RowDecoder'.+sessionReadStreamPAs :: HasCallStack+ => RowDecoder a -> Session -> Text -> H.HashMap Text PS.Ps -> Stream IO a+sessionReadStreamPAs decoder session cypher params =+ sessionStreamDecoded decoder session ReadAccess cypher params+++-- | Like 'sessionWriteStream' but decodes each record using a 'RowDecoder'.+sessionWriteStreamAs :: HasCallStack => RowDecoder a -> Session -> Text -> Stream IO a+sessionWriteStreamAs decoder session cypher =+ sessionWriteStreamPAs decoder session cypher H.empty+++-- | Like 'sessionWriteStreamP' but decodes each record using a 'RowDecoder'.+sessionWriteStreamPAs :: HasCallStack+ => RowDecoder a -> Session -> Text -> H.HashMap Text PS.Ps -> Stream IO a+sessionWriteStreamPAs decoder session cypher params =+ sessionStreamDecoded decoder session WriteAccess cypher params+++-- | Internal: session streaming for raw records.+sessionStreamRaw :: HasCallStack+ => Session -> AccessMode -> Text -> H.HashMap Text PS.Ps -> Stream IO Record+sessionStreamRaw Session{sPool, sBookmarks, sConfig, sTelemetrySent} mode cypher params =+ Stream.bracketIO (sessionAcquire sPool sBookmarks sConfig sTelemetrySent mode)+ (sessionCleanup sBookmarks sTelemetrySent)+ (\coc -> Stream.concatEffect $ queryStreamP (cocConnection coc) cypher params)+++-- | Internal: session streaming with record decoding.+sessionStreamDecoded :: HasCallStack+ => RowDecoder a -> Session -> AccessMode -> Text -> H.HashMap Text PS.Ps -> Stream IO a+sessionStreamDecoded decoder Session{sPool, sBookmarks, sConfig, sTelemetrySent} mode cypher params =+ Stream.bracketIO (sessionAcquire sPool sBookmarks sConfig sTelemetrySent mode)+ (sessionCleanup sBookmarks sTelemetrySent)+ (\coc -> Stream.concatEffect $ queryStreamPAs decoder (cocConnection coc) cypher params)+++-- | Acquire a session connection and begin a transaction.+sessionAcquire :: SessionPool -> BookmarkManager -> SessionConfig -> IORef Bool+ -> AccessMode -> IO CheckedOutConnection+sessionAcquire sPool sBookmarks sConfig _telRef mode = do+ let modeChar = case mode of { ReadAccess -> 'r'; WriteAccess -> 'w' }+ let db = sessionDatabase sConfig+ coc <- acquireSessionConnection sPool mode+ let conn = cocConnection coc+ bms <- getBookmarks sBookmarks+ P.beginTx conn $ Begin (V.fromList bms) Nothing H.empty modeChar db Nothing+ pure coc+++-- | Commit the transaction and release the connection.+sessionCleanup :: BookmarkManager -> IORef Bool -> CheckedOutConnection -> IO ()+sessionCleanup sBookmarks sTelemetrySent coc = do+ let conn = cocConnection coc+ result <- try $ do+ mbBookmark <- P.commitTx conn+ case mbBookmark of+ Just bm -> updateBookmark sBookmarks bm+ Nothing -> pure ()+ sent <- readIORef sTelemetrySent+ when (not sent) $ do+ writeIORef sTelemetrySent True+ P.sendTelemetry conn ManagedTransactions+ case result of+ Right () -> releaseConnection coc+ Left (_ :: SomeException) -> releaseConnectionOnError coc+++-- | Acquire a connection based on session pool type.+acquireSessionConnection :: SessionPool -> AccessMode -> IO CheckedOutConnection+acquireSessionConnection (DirectPool pool) _mode = acquireConnection pool+acquireSessionConnection (RoutedPool rp) mode = acquireRoutingConnection rp mode+++-- | Extract the database field from SessionConfig (avoids ambiguous record access).+sessionDatabase :: SessionConfig -> Maybe Text+sessionDatabase SessionConfig{database} = database
+ test/Main.hs view
@@ -0,0 +1,241 @@+module Main where++import Test.Sandwich++import Database.Bolty.Streamly (Stream, queryStream, queryStreamP, pullStream,+ poolStream, poolStreamP,+ routingStream, routingStreamP,+ sessionReadStream, sessionReadStreamP,+ sessionWriteStream, sessionWriteStreamP,+ queryStreamAs, queryStreamPAs,+ poolStreamAs, poolStreamPAs,+ routingStreamAs, routingStreamPAs,+ sessionReadStreamAs, sessionReadStreamPAs,+ sessionWriteStreamAs, sessionWriteStreamPAs)+import Database.Bolty (BoltPool, Connection, Record, Session,+ AccessMode(..))+import Data.Text (Text)+import Database.Bolty.Decode (RowDecoder, column, int64)+import Database.Bolty.Routing (RoutingPool)+import Data.Int (Int64)+import qualified Data.HashMap.Lazy as H+import Data.PackStream.Ps (Ps(..))+++-- Type-level checks: these ensure the module exports compile with+-- the correct types. They are never called at runtime.++_checkQueryStream :: Connection -> IO (Stream IO Record)+_checkQueryStream conn = queryStream conn "RETURN 1"++_checkQueryStreamP :: Connection -> IO (Stream IO Record)+_checkQueryStreamP conn = queryStreamP conn "RETURN $x" (H.singleton "x" (PsInteger 42))++_checkPullStream :: Connection -> IO (Stream IO Record)+_checkPullStream conn = pullStream conn++_checkPoolStream :: BoltPool -> Stream IO Record+_checkPoolStream pool = poolStream pool "RETURN 1"++_checkPoolStreamP :: BoltPool -> Stream IO Record+_checkPoolStreamP pool =+ poolStreamP pool "RETURN $x" (H.singleton "x" (PsInteger 1))++_checkRoutingStream :: RoutingPool -> Stream IO Record+_checkRoutingStream rp = routingStream rp ReadAccess "RETURN 1"++_checkRoutingStreamP :: RoutingPool -> Stream IO Record+_checkRoutingStreamP rp =+ routingStreamP rp ReadAccess "RETURN $x" (H.singleton "x" (PsInteger 1))++_checkSessionReadStream :: Session -> Stream IO Record+_checkSessionReadStream s = sessionReadStream s "RETURN 1"++_checkSessionReadStreamP :: Session -> Stream IO Record+_checkSessionReadStreamP s =+ sessionReadStreamP s "RETURN $x" (H.singleton "x" (PsInteger 1))++_checkSessionWriteStream :: Session -> Stream IO Record+_checkSessionWriteStream s = sessionWriteStream s "RETURN 1"++_checkSessionWriteStreamP :: Session -> Stream IO Record+_checkSessionWriteStreamP s =+ sessionWriteStreamP s "RETURN $x" (H.singleton "x" (PsInteger 1))++-- Decode variant type-level checks++_decoder :: RowDecoder Int64+_decoder = column 0 int64++_checkQueryStreamAs :: Connection -> IO (Stream IO Int64)+_checkQueryStreamAs conn = queryStreamAs _decoder conn "RETURN 1"++_checkQueryStreamPAs :: Connection -> IO (Stream IO Int64)+_checkQueryStreamPAs conn = queryStreamPAs _decoder conn "RETURN $x" (H.singleton "x" (PsInteger 1))++_checkPoolStreamAs :: BoltPool -> Stream IO Int64+_checkPoolStreamAs pool = poolStreamAs _decoder pool "RETURN 1"++_checkPoolStreamPAs :: BoltPool -> Stream IO Int64+_checkPoolStreamPAs pool =+ poolStreamPAs _decoder pool "RETURN $x" (H.singleton "x" (PsInteger 1))++_checkRoutingStreamAs :: RoutingPool -> Stream IO Int64+_checkRoutingStreamAs rp =+ routingStreamAs _decoder rp ReadAccess "RETURN 1"++_checkRoutingStreamPAs :: RoutingPool -> Stream IO Int64+_checkRoutingStreamPAs rp =+ routingStreamPAs _decoder rp ReadAccess "RETURN $x" (H.singleton "x" (PsInteger 1))++_checkSessionReadStreamAs :: Session -> Stream IO Int64+_checkSessionReadStreamAs s = sessionReadStreamAs _decoder s "RETURN 1"++_checkSessionReadStreamPAs :: Session -> Stream IO Int64+_checkSessionReadStreamPAs s =+ sessionReadStreamPAs _decoder s "RETURN $x" (H.singleton "x" (PsInteger 1))++_checkSessionWriteStreamAs :: Session -> Stream IO Int64+_checkSessionWriteStreamAs s = sessionWriteStreamAs _decoder s "RETURN 1"++_checkSessionWriteStreamPAs :: Session -> Stream IO Int64+_checkSessionWriteStreamPAs s =+ sessionWriteStreamPAs _decoder s "RETURN $x" (H.singleton "x" (PsInteger 1))+++tests :: TopSpec+tests = describe "Database.Bolty.Streamly" $ do++ it "queryStream has correct type" $ do+ let _ = _checkQueryStream+ pure ()++ it "queryStreamP has correct type" $ do+ let _ = _checkQueryStreamP+ pure ()++ it "pullStream has correct type" $ do+ let _ = _checkPullStream+ pure ()++ it "Stream type is re-exported" $ do+ let _ = undefined :: Stream IO Int+ pure ()++ it "poolStream has correct type" $ do+ let _ = _checkPoolStream+ pure ()++ it "poolStreamP has correct type" $ do+ let _ = _checkPoolStreamP+ pure ()++ it "routingStream has correct type" $ do+ let _ = _checkRoutingStream+ pure ()++ it "routingStreamP has correct type" $ do+ let _ = _checkRoutingStreamP+ pure ()++ it "sessionReadStream has correct type" $ do+ let _ = _checkSessionReadStream+ pure ()++ it "sessionReadStreamP has correct type" $ do+ let _ = _checkSessionReadStreamP+ pure ()++ it "sessionWriteStream has correct type" $ do+ let _ = _checkSessionWriteStream+ pure ()++ it "sessionWriteStreamP has correct type" $ do+ let _ = _checkSessionWriteStreamP+ pure ()++ it "queryStreamAs has correct type" $ do+ let _ = _checkQueryStreamAs+ pure ()++ it "queryStreamPAs has correct type" $ do+ let _ = _checkQueryStreamPAs+ pure ()++ it "poolStreamAs has correct type" $ do+ let _ = _checkPoolStreamAs+ pure ()++ it "poolStreamPAs has correct type" $ do+ let _ = _checkPoolStreamPAs+ pure ()++ it "routingStreamAs has correct type" $ do+ let _ = _checkRoutingStreamAs+ pure ()++ it "routingStreamPAs has correct type" $ do+ let _ = _checkRoutingStreamPAs+ pure ()++ it "sessionReadStreamAs has correct type" $ do+ let _ = _checkSessionReadStreamAs+ pure ()++ it "sessionReadStreamPAs has correct type" $ do+ let _ = _checkSessionReadStreamPAs+ pure ()++ it "sessionWriteStreamAs has correct type" $ do+ let _ = _checkSessionWriteStreamAs+ pure ()++ it "sessionWriteStreamPAs has correct type" $ do+ let _ = _checkSessionWriteStreamPAs+ pure ()++ -- Verify pool/routing/session functions return Stream (not IO Stream)+ -- This is the key bracketIO design property: streams are pure values.++ it "poolStream returns Stream IO Record (not IO)" $ do+ let _ = poolStream :: BoltPool -> Text -> Stream IO Record+ pure ()++ it "routingStream returns Stream IO Record (not IO)" $ do+ let _ = routingStream :: RoutingPool -> AccessMode -> Text -> Stream IO Record+ pure ()++ it "sessionReadStream returns Stream IO Record (not IO)" $ do+ let _ = sessionReadStream :: Session -> Text -> Stream IO Record+ pure ()++ it "sessionWriteStream returns Stream IO Record (not IO)" $ do+ let _ = sessionWriteStream :: Session -> Text -> Stream IO Record+ pure ()++ it "poolStreamAs returns Stream IO a (not IO)" $ do+ let _ = poolStreamAs :: RowDecoder Int64 -> BoltPool -> Text -> Stream IO Int64+ pure ()++ it "routingStreamAs returns Stream IO a (not IO)" $ do+ let _ = routingStreamAs :: RowDecoder Int64 -> RoutingPool -> AccessMode -> Text -> Stream IO Int64+ pure ()++ it "sessionReadStreamAs returns Stream IO a (not IO)" $ do+ let _ = sessionReadStreamAs :: RowDecoder Int64 -> Session -> Text -> Stream IO Int64+ pure ()++ -- Verify bare-connection functions still return IO (Stream IO Record)+ -- since they perform RUN eagerly before streaming PULL.++ it "queryStream returns IO (Stream IO Record)" $ do+ let _ = queryStream :: Connection -> Text -> IO (Stream IO Record)+ pure ()++ it "queryStreamAs returns IO (Stream IO a)" $ do+ let _ = queryStreamAs :: RowDecoder Int64 -> Connection -> Text -> IO (Stream IO Int64)+ pure ()+++main :: IO ()+main = runSandwichWithCommandLineArgs defaultOptions tests