diff --git a/hasbolt.cabal b/hasbolt.cabal --- a/hasbolt.cabal +++ b/hasbolt.cabal @@ -1,13 +1,13 @@ cabal-version: >=1.10 -name: hasbolt -version: 0.1.4.3 -license: BSD3 -license-file: LICENSE -copyright: (c) 2018 Pavel Yakovlev -maintainer: pavel@yakovlev.me -author: Pavel Yakovlev, Martin Heuschober -homepage: https://github.com/zmactep/hasbolt#readme -synopsis: Haskell driver for Neo4j 3+ (BOLT protocol) +name: hasbolt +version: 0.1.4.4 +license: BSD3 +license-file: LICENSE +copyright: (c) 2018 Pavel Yakovlev +maintainer: pavel@yakovlev.me +author: Pavel Yakovlev, Martin Heuschober +homepage: https://github.com/zmactep/hasbolt#readme +synopsis: Haskell driver for Neo4j 3+ (BOLT protocol) description: Haskell driver for Neo4j 3+ (BOLT protocol). . @@ -24,11 +24,12 @@ -TLS/SSL connection . The code was tested with neo4j versions 3.0 — 3.4 and GrapheneDB service -category: Database -build-type: Simple +category: Database +build-type: Simple + source-repository head - type: git + type: git location: https://github.com/zmactep/hasbolt library @@ -37,7 +38,8 @@ Database.Bolt.Lazy Database.Bolt.Lens Database.Bolt.Serialization - hs-source-dirs: src + + hs-source-dirs: src other-modules: Database.Bolt.Value.Type Database.Bolt.Value.Helpers @@ -49,11 +51,12 @@ Database.Bolt.Connection Database.Bolt.Record Database.Bolt.Transaction + default-language: Haskell2010 - ghc-options: -Wall + ghc-options: -Wall build-depends: base >=4.7 && <5, - bytestring >=0.10.8.1 && <0.11, + bytestring >=0.10.8.1 && <0.12, text >=1.2.2.1 && <1.3, containers >=0.5.7.1 && <0.7, binary >=0.8.3.0 && <1.0, @@ -62,23 +65,20 @@ network >=2.6.3.1 && <3.2, connection >=0.2.8 && <0.4, data-default >=0.7.1.1 && <0.8 - + if impl(ghc <8.6) - build-depends: - contravariant >=1.4.1 && <1.6 - + build-depends: contravariant >=1.4.1 && <1.6 + if impl(ghc <8.0) - build-depends: - fail >=4.9 && <5 + build-depends: fail >=4.9 && <5 test-suite hasbolt-test - type: exitcode-stdio-1.0 - main-is: Spec.hs - hs-source-dirs: test - other-modules: - Hex + type: exitcode-stdio-1.0 + main-is: Spec.hs + hs-source-dirs: test + other-modules: Hex default-language: Haskell2010 - ghc-options: -threaded -rtsopts -with-rtsopts=-N + ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: base >=4.8 && <5, hasbolt -any, @@ -86,4 +86,4 @@ QuickCheck >=2.9 && <2.14, text >=1.2.4.0 && <1.3, containers >=0.6.2.1 && <0.7, - bytestring >=0.10.10.0 && <0.11 + bytestring >=0.10.10.1 && <0.11 diff --git a/src/Database/Bolt/Connection.hs b/src/Database/Bolt/Connection.hs --- a/src/Database/Bolt/Connection.hs +++ b/src/Database/Bolt/Connection.hs @@ -20,57 +20,58 @@ import Control.Exception (throwIO) import Control.Monad (void) -import Control.Monad.Trans (MonadIO (..)) -import Control.Monad.Reader (MonadReader (..), runReaderT) import Control.Monad.Except (MonadError (..), runExceptT) -import Data.Text (Text) +import Control.Monad.Reader (MonadReader (..), runReaderT) +import Control.Monad.Trans (MonadIO (..)) import Data.Map.Strict (Map, empty, fromList) +import Data.Text (Text) +import GHC.Stack (HasCallStack) import System.IO.Unsafe (unsafeInterleaveIO) -- |Runs BOLT action on selected pipe -runE :: MonadIO m => Pipe -> BoltActionT m a -> m (Either BoltError a) -runE pipe action = runExceptT (runReaderT (runBoltActionT action) pipe) +runE :: MonadIO m => HasCallStack => Pipe -> BoltActionT m a -> m (Either BoltError a) +runE pipe action = runExceptT (runReaderT (runBoltActionT action) pipe) -- |Runs BOLT action on selected pipe (with errors throw) -run :: MonadIO m => Pipe -> BoltActionT m a -> m a +run :: MonadIO m => HasCallStack => Pipe -> BoltActionT m a -> m a run pipe action = do result <- runE pipe action case result of Right x -> pure x Left r -> liftIO $ throwIO r -- |Runs Cypher query with parameters and returns list of obtained 'Record's. Lazy version -queryP :: MonadIO m => Text -> Map Text Value -> BoltActionT m [Record] +queryP :: MonadIO m => HasCallStack => Text -> Map Text Value -> BoltActionT m [Record] queryP = querySL False -- |Runs Cypher query and returns list of obtained 'Record's. Lazy version -query :: MonadIO m => Text -> BoltActionT m [Record] +query :: MonadIO m => HasCallStack => Text -> BoltActionT m [Record] query cypher = queryP cypher empty -- |Runs Cypher query with parameters and returns list of obtained 'Record's. Strict version -queryP' :: MonadIO m => Text -> Map Text Value -> BoltActionT m [Record] +queryP' :: MonadIO m => HasCallStack => Text -> Map Text Value -> BoltActionT m [Record] queryP' = querySL True -- |Runs Cypher query and returns list of obtained 'Record's. Strict version -query' :: MonadIO m => Text -> BoltActionT m [Record] +query' :: MonadIO m => HasCallStack => Text -> BoltActionT m [Record] query' cypher = queryP' cypher empty -- |Runs Cypher query with parameters and ignores response -queryP_ :: MonadIO m => Text -> Map Text Value -> BoltActionT m () +queryP_ :: MonadIO m => HasCallStack => Text -> Map Text Value -> BoltActionT m () queryP_ cypher params = do void $ sendRequest cypher params ask >>= liftE . discardAll -- |Runs Cypher query and ignores response -query_ :: MonadIO m => Text -> BoltActionT m () +query_ :: MonadIO m => HasCallStack => Text -> BoltActionT m () query_ cypher = queryP_ cypher empty -- Helper functions -querySL :: MonadIO m => Bool -> Text -> Map Text Value -> BoltActionT m [Record] +querySL :: MonadIO m => HasCallStack => Bool -> Text -> Map Text Value -> BoltActionT m [Record] querySL strict cypher params = do keys <- pullKeys cypher params pullRecords strict keys -pullKeys :: MonadIO m => Text -> Map Text Value -> BoltActionT m [Text] +pullKeys :: MonadIO m => HasCallStack => Text -> Map Text Value -> BoltActionT m [Text] pullKeys cypher params = do pipe <- ask status <- sendRequest cypher params liftE $ flush pipe RequestPullAll @@ -80,7 +81,7 @@ mkKeys (ResponseSuccess response) = response `at` "fields" `catchError` \(RecordHasNoKey _) -> pure [] mkKeys x = throwError $ ResponseError (mkFailure x) -pullRecords :: MonadIO m => Bool -> [Text] -> BoltActionT m [Record] +pullRecords :: MonadIO m => HasCallStack => Bool -> [Text] -> BoltActionT m [Record] pullRecords strict keys = do pipe <- ask resp <- liftE $ fetch pipe cases resp @@ -101,7 +102,7 @@ pure (record:rest) -- |Sends request to database and makes an action -sendRequest :: MonadIO m => Text -> Map Text Value -> BoltActionT m Response +sendRequest :: MonadIO m => HasCallStack => Text -> Map Text Value -> BoltActionT m Response sendRequest cypher params = do pipe <- ask liftE $ do diff --git a/src/Database/Bolt/Connection/Connection.hs b/src/Database/Bolt/Connection/Connection.hs --- a/src/Database/Bolt/Connection/Connection.hs +++ b/src/Database/Bolt/Connection/Connection.hs @@ -1,38 +1,64 @@ +{-# LANGUAGE RecordWildCards #-} module Database.Bolt.Connection.Connection where import Control.Applicative (pure, (<$>)) -import Control.Monad (when, forM_) +import Control.Exception (throwIO) +import Control.Monad (forM_, when) import Control.Monad.Trans (MonadIO (..)) import Data.ByteString (ByteString, null) import Data.Default (Default (..)) +import GHC.Stack (HasCallStack, withFrozenCallStack) import Network.Socket (PortNumber) -import Network.Connection (Connection, ConnectionParams (..), connectTo, connectionSetSecure, - initConnectionContext, connectionClose, connectionGetExact, connectionPut) +import Network.Connection (ConnectionParams (..), connectTo, connectionClose, + connectionGetExact, connectionPut, connectionSetSecure, + initConnectionContext) import Prelude hiding (null) +import System.Timeout (timeout) -connect :: MonadIO m => Bool -> String -> PortNumber -> m Connection -connect secure host port = liftIO $ do - ctx <- initConnectionContext - conn <- connectTo ctx ConnectionParams { connectionHostname = host - , connectionPort = port - , connectionUseSecure = Nothing - , connectionUseSocks = Nothing - } - when secure $ connectionSetSecure ctx conn def - pure conn +import Database.Bolt.Connection.Type (BoltError (..), ConnectionWithTimeout (..)) -close :: MonadIO m => Connection -> m () -close = liftIO . connectionClose +connect + :: MonadIO m + => HasCallStack + => Bool + -- ^ Use secure connection + -> String + -- ^ Hostname + -> PortNumber + -> Int + -- ^ Connection and read timeout in seconds + -> m ConnectionWithTimeout +connect secure host port timeSec = liftIO $ do + let timeUsec = 1000000 * timeSec + ctx <- initConnectionContext + conn <- timeoutThrow timeUsec $ + connectTo ctx ConnectionParams { connectionHostname = host + , connectionPort = port + , connectionUseSecure = Nothing + , connectionUseSocks = Nothing + } + when secure $ connectionSetSecure ctx conn def + pure $ ConnectionWithTimeout conn timeUsec -recv :: MonadIO m => Connection -> Int -> m (Maybe ByteString) -recv conn = liftIO . (filterMaybe (not . null) <$>) . connectionGetExact conn +close :: MonadIO m => HasCallStack => ConnectionWithTimeout -> m () +close ConnectionWithTimeout{..} = liftIO $ timeoutThrow cwtTimeoutUsec $ connectionClose cwtConnection + +recv :: MonadIO m => HasCallStack => ConnectionWithTimeout -> Int -> m (Maybe ByteString) +recv ConnectionWithTimeout{..} = liftIO . (filterMaybe (not . null) <$>) . timeoutThrow cwtTimeoutUsec . connectionGetExact cwtConnection where filterMaybe :: (a -> Bool) -> a -> Maybe a filterMaybe p x | p x = Just x | otherwise = Nothing -send :: MonadIO m => Connection -> ByteString -> m () -send conn = liftIO . connectionPut conn +send :: MonadIO m => HasCallStack => ConnectionWithTimeout -> ByteString -> m () +send ConnectionWithTimeout{..} = liftIO . timeoutThrow cwtTimeoutUsec . connectionPut cwtConnection -sendMany :: MonadIO m => Connection -> [ByteString] -> m () -sendMany conn chunks = forM_ chunks $ send conn +sendMany :: MonadIO m => HasCallStack => ConnectionWithTimeout -> [ByteString] -> m () +sendMany conn@ConnectionWithTimeout{..} chunks = liftIO $ forM_ chunks $ timeoutThrow cwtTimeoutUsec . send conn + +timeoutThrow :: HasCallStack => Int -> IO a -> IO a +timeoutThrow timeUsec action = withFrozenCallStack $ do + res <- timeout timeUsec action + case res of + Just a -> return a + Nothing -> throwIO TimeOut diff --git a/src/Database/Bolt/Connection/Pipe.hs b/src/Database/Bolt/Connection/Pipe.hs --- a/src/Database/Bolt/Connection/Pipe.hs +++ b/src/Database/Bolt/Connection/Pipe.hs @@ -17,26 +17,26 @@ import qualified Data.ByteString as B (concat, length, null, splitAt) import Data.Word (Word16) -import Network.Connection (Connection) +import GHC.Stack (HasCallStack) type MonadPipe m = (MonadIO m, MonadError BoltError m) -- |Creates new 'Pipe' instance to use all requests through -connect :: MonadIO m => BoltCfg -> m Pipe +connect :: MonadIO m => HasCallStack => BoltCfg -> m Pipe connect = makeIO connect' where connect' :: MonadPipe m => BoltCfg -> m Pipe - connect' bcfg = do conn <- C.connect (secure bcfg) (host bcfg) (fromIntegral $ port bcfg) + connect' bcfg = do conn <- C.connect (secure bcfg) (host bcfg) (fromIntegral $ port bcfg) (socketTimeout bcfg) let pipe = Pipe conn (maxChunkSize bcfg) handshake pipe bcfg pure pipe -- |Closes 'Pipe' -close :: MonadIO m => Pipe -> m () +close :: MonadIO m => HasCallStack => Pipe -> m () close = C.close . connection -- |Resets current sessions -reset :: MonadIO m => Pipe -> m () +reset :: MonadIO m => HasCallStack => Pipe -> m () reset = makeIO reset' where reset' :: MonadPipe m => Pipe -> m () @@ -46,7 +46,7 @@ throwError ResetFailed -- Helper to make pipe operations in IO -makeIO :: MonadIO m => (a -> ExceptT BoltError m b) -> a -> m b +makeIO :: MonadIO m => HasCallStack => (a -> ExceptT BoltError m b) -> a -> m b makeIO action arg = do actionIO <- runExceptT (action arg) case actionIO of Right x -> pure x @@ -54,13 +54,13 @@ -- = Internal interfaces -ackFailure :: MonadPipe m => Pipe -> m () +ackFailure :: MonadPipe m => HasCallStack => Pipe -> m () ackFailure pipe = flush pipe RequestAckFailure >> void (fetch pipe) -discardAll :: MonadPipe m => Pipe -> m () +discardAll :: MonadPipe m => HasCallStack => Pipe -> m () discardAll pipe = flush pipe RequestDiscardAll >> void (fetch pipe) -flush :: MonadPipe m => Pipe -> Request -> m () +flush :: MonadPipe m => HasCallStack => Pipe -> Request -> m () flush pipe request = do forM_ chunks $ C.sendMany conn . mkChunk C.send conn terminal where bs = pack $ toStructure request @@ -73,7 +73,7 @@ mkChunk chunk = let size = fromIntegral (B.length chunk) :: Word16 in [encodeStrict size, chunk] -fetch :: MonadPipe m => Pipe -> m Response +fetch :: MonadPipe m => HasCallStack => Pipe -> m Response fetch pipe = do bs <- B.concat <$> chunks response <- flip unpackAction bs $ unpackT >>= fromStructure either (throwError . WrongMessageFormat) pure response @@ -89,7 +89,7 @@ -- Helper functions -handshake :: MonadPipe m => Pipe -> BoltCfg -> m () +handshake :: MonadPipe m => HasCallStack => Pipe -> BoltCfg -> m () handshake pipe bcfg = do let conn = connection pipe C.send conn (encodeStrict $ magic bcfg) C.send conn (boltVersionProposal bcfg) @@ -104,7 +104,7 @@ boltVersionProposal :: BoltCfg -> ByteString boltVersionProposal bcfg = B.concat $ encodeStrict <$> [version bcfg, 0, 0, 0] -recvChunk :: MonadPipe m => Connection -> Word16 -> m ByteString +recvChunk :: MonadPipe m => HasCallStack => ConnectionWithTimeout -> Word16 -> m ByteString recvChunk conn size = B.concat <$> helper (fromIntegral size) where helper :: MonadPipe m => Int -> m [ByteString] helper 0 = pure [] diff --git a/src/Database/Bolt/Connection/Type.hs b/src/Database/Bolt/Connection/Type.hs --- a/src/Database/Bolt/Connection/Type.hs +++ b/src/Database/Bolt/Connection/Type.hs @@ -1,3 +1,4 @@ +{-# LANGUAGE GADTs #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE DeriveFunctor #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} @@ -12,10 +13,11 @@ import Control.Monad.Except (MonadError (..), ExceptT (..)) import Data.Default (Default (..)) -import Data.Monoid ((<>)) import Data.Map.Strict (Map) +import Data.Monoid ((<>)) import Data.Text (Text, unpack) import Data.Word (Word16, Word32) +import GHC.Stack (HasCallStack, callStack, prettyCallStack) import Network.Connection (Connection) @@ -38,6 +40,7 @@ | ResponseError ResponseError | RecordHasNoKey Text | NonHasboltError SomeException + | HasCallStack => TimeOut instance Show BoltError where show UnsupportedServerVersion = "Cannot connect: unsupported server version" @@ -49,6 +52,7 @@ show (ResponseError re) = show re show (RecordHasNoKey key) = "Cannot unpack record: key '" <> unpack key <> "' is not presented" show (NonHasboltError msg) = "User error: " <> show msg + show TimeOut = "Operation timeout\n" <> prettyCallStack callStack instance Exception BoltError @@ -70,7 +74,7 @@ , version :: Word32 -- ^'00000001' value , userAgent :: Text -- ^Driver user agent , maxChunkSize :: Word16 -- ^Maximum chunk size of request - , socketTimeout :: Int -- ^Driver socket timeout + , socketTimeout :: Int -- ^Driver socket timeout in seconds , host :: String -- ^Neo4j server hostname , port :: Int -- ^Neo4j server port , user :: Text -- ^Neo4j user @@ -92,8 +96,15 @@ , secure = False } -data Pipe = Pipe { connection :: Connection -- ^Driver connection socket - , mcs :: Word16 -- ^Driver maximum chunk size of request +data ConnectionWithTimeout + = ConnectionWithTimeout + { cwtConnection :: !Connection + , cwtTimeoutUsec :: !Int + -- ^ Timeout in microseconds + } + +data Pipe = Pipe { connection :: ConnectionWithTimeout -- ^Driver connection socket + , mcs :: Word16 -- ^Driver maximum chunk size of request } data AuthToken = AuthToken { scheme :: Text @@ -101,7 +112,7 @@ , credentials :: Text } deriving (Eq, Show) - + data Response = ResponseSuccess { succMap :: Map Text Value } | ResponseRecord { recsList :: [Value] } | ResponseIgnored { ignoreMap :: Map Text Value }