hascas 1.0.0 → 1.1.0

raw patch · 5 files changed

+95/−67 lines, 5 filesdep ~basePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• hascas.cabal +1/−1
• src/CQL.hs +10/−3
• src/Driver.hs +74/−54
• src/Query.hs +5/−4
• test/Spec.hs +5/−5

@@ -1,5 +1,5 @@ name:                hascas-version:             1.0.0+version:             1.1.0 synopsis:            Cassandra driver for haskell description:         This is a cassandra driver. homepage:            https://github.com/eklavya/hascas#readme

@@ -22,13 +22,13 @@ Initialize the driver by calling  @-  CQL.init host port+  CQL.init host port retryInterval @  for example  @-  CQL.init "127.0.0.1" (PortNumber 9042)+  CQL.init "127.0.0.1" (PortNumber 9042) (RetryInterval 1000000) @  Example:@@ -51,7 +51,7 @@  main :: IO () main = do-    candle <- CQL.init "127.0.0.1" (PortNumber 9042)+    candle <- CQL.init "127.0.0.1" (PortNumber 9042) (RetryInterval 1000000)      res <- flip runReaderT candle $ runExceptT $ do       let q = create "keyspace demodb WITH REPLICATION = {'class' : 'SimpleStrategy','replication_factor': 1}"@@ -121,6 +121,8 @@ module CQL (            Driver.init,            Consistency(..),+           Candle,+           Driver.RetryInterval(..),            -- ** Data Types            Rows,            Row,@@ -130,6 +132,8 @@            CQLMap(..),            CQLSet(..),            CQLList(..),+           ShortStr(..),+           LongStr(..),            -- ** Auto derive conversion for record types            deriveBuildRec,            fromRow,@@ -145,10 +149,13 @@            where',            and',            -- ** Prepare and run queries+           Prepared,            runCQL,            prepare,            execCQL,            -- ** Batching+           LoggedBatch,+           BatchQuery,            batch,            prepBatch,            runBatch) where

@@ -4,7 +4,7 @@ {-# LANGUAGE ScopedTypeVariables #-}  -module Driver (Driver.init, Driver.allConnections) where+module Driver (Driver.init, Driver.allConnections, Driver.RetryInterval(..)) where   import           Codec@@ -14,7 +14,7 @@ import           Control.Concurrent.MVar import           Control.Concurrent.STM import           Control.Concurrent.STM.TBQueue-import           Control.Exception              (bracket, catch, mask)+import           Control.Exception.Safe         (bracket, catchAny, mask) import           Control.Monad                  (forM_, forever, replicateM) import           Control.Monad.Except import           Data.Binary@@ -33,6 +33,7 @@ import           Data.Maybe import           Data.Monoid                    as DM import           Data.Set+import           Data.Traversable import           Data.UUID import           Debug.Trace import           Encoding@@ -72,38 +73,43 @@   return mvar  -receiveThread :: Handle -> MVar [Int16] -> MVar(IM.IntMap (MVar (Either ShortStr Result))) -> IO ()-receiveThread h streams streamMap = do-  forkIO $ forever $ do-        header <- hGet h 9-        he <- pure $ runGet (get :: Get Header) $ DBL.fromStrict header-        p <- hGet h (fromIntegral $ len he)-        case opcode he of-          0 -> do-            let (erCode, erMsg) = runGet getErr (DBL.fromStrict p)-            mvar <- getResHolder he streams streamMap-            putMVar mvar (Left erMsg)-          8 -> do-            let resultType = runGet (get :: Get Int32) (DBL.fromStrict p)+receiveThread :: HostName -> PortID -> RetryInterval -> Handle -> MVar [Int16] -> MVar(IM.IntMap (MVar (Either ShortStr Result))) -> IO ()+receiveThread host port ri h streams streamMap = do+  forkIO $ catchAny (forever $ do+                                header <- hGet h 9+                                he <- pure $ runGet (get :: Get Header) $ DBL.fromStrict header+                                p <- hGet h (fromIntegral $ len he)+                                case opcode he of+                                  0 -> do+                                    let (erCode, erMsg) = runGet getErr (DBL.fromStrict p)+                                    mvar <- getResHolder he streams streamMap+                                    putMVar mvar (Left erMsg)+                                  8 -> do+                                    let resultType = runGet (get :: Get Int32) (DBL.fromStrict p) -            case resultType of-              1 -> do-                mvar <- getResHolder he streams streamMap-                putMVar mvar (Right $ RRows [])+                                    case resultType of+                                      1 -> do+                                        mvar <- getResHolder he streams streamMap+                                        putMVar mvar (Right $ RRows []) -              2 -> do-                let rows = content $ runGet getRows (DBL.fromStrict $ C8.drop 4 p)-                mvar <- getResHolder he streams streamMap-                putMVar mvar (Right $ RRows rows)+                                      2 -> do+                                        let rows = content $ runGet getRows (DBL.fromStrict $ C8.drop 4 p)+                                        mvar <- getResHolder he streams streamMap+                                        putMVar mvar (Right $ RRows rows) -              4 -> do-                let prep = runGet (get :: Get ShortBytes) (DBL.fromStrict $ C8.drop 4 p)-                mvar <- getResHolder he streams streamMap-                putMVar mvar (Right $ RPrepared prep)+                                      4 -> do+                                        let prep = runGet (get :: Get ShortBytes) (DBL.fromStrict $ C8.drop 4 p)+                                        mvar <- getResHolder he streams streamMap+                                        putMVar mvar (Right $ RPrepared prep) -              5 -> do-                mvar <- getResHolder he streams streamMap-                putMVar mvar (Right $ RRows [])+                                      5 -> do+                                        mvar <- getResHolder he streams streamMap+                                        putMVar mvar (Right $ RRows [])) (\e -> do+                                                                                  print "error in receiving"+                                                                                  sm <- takeMVar streamMap+                                                                                  nm <- sequence $ fmap (\mvar -> tryPutMVar mvar (Left $ ShortStr $ DBL.fromStrict $ C8.pack $ "connection to node " <> show host <> " was lost")) sm+                                                                                  putMVar streamMap sm)+           -- 12 -> do           --   print "something happened"           --   print $ runGet (get :: Get ShortStr) (DBL.fromStrict p)@@ -122,8 +128,8 @@           return mvar  -someDef :: Handle -> MVar [Int16] -> MVar (IM.IntMap (MVar (Either ShortStr Result))) -> (LongStr, Word8, MVar (Either ShortStr Result)) -> IO ()-someDef h streams streamMap (bs, opc, mvar) = do+writeQuery :: Handle -> MVar [Int16] -> MVar (IM.IntMap (MVar (Either ShortStr Result))) -> (LongStr, Word8, MVar (Either ShortStr Result)) -> IO ()+writeQuery h streams streamMap (bs, opc, mvar) = do       strs <- takeMVar streams       case Data.List.uncons strs of         Just (i, strsTail) -> do@@ -137,26 +143,32 @@          Nothing -> do           putMVar streams strs-          someDef h streams streamMap (bs, opc, mvar)+          writeQuery h streams streamMap (bs, opc, mvar)  -sendThread :: Handle -> MVar [Int16] -> MVar(IM.IntMap (MVar (Either ShortStr Result))) -> TBQueue (LongStr, Word8, MVar (Either ShortStr Result)) -> IO ()-sendThread h streams streamMap driverQ = do-  forkIO $ forever $ do+sendThread :: HostName -> PortID -> RetryInterval -> Handle -> MVar [Int16] -> MVar(IM.IntMap (MVar (Either ShortStr Result))) -> TBQueue (LongStr, Word8, MVar (Either ShortStr Result)) -> IO ()+sendThread host port ri h streams streamMap driverQ = do+  forkIO $ catchAny (forever $ do           (bs, opc, mvar) <- atomically $ readTBQueue driverQ           when (opc == 9) $ do              pq <- takeMVar prepareQueries              putMVar prepareQueries (Data.Set.insert bs pq)              cons <- readMVar allConnections-             mvars <- forM cons $ \(_, h', str, strMap) -> do+             mvars <- forM cons $ \(_, _, h', str, strMap) -> do                mv <- newEmptyMVar-               someDef h' str strMap (bs, opc, mv)+               writeQuery h' str strMap (bs, opc, mv)                return mv              forkIO $ do                ls <- sequence $ fmap (\mva -> do {a <- takeMVar mva; return a}) mvars                putMVar mvar $ Data.List.foldl' (\b a -> if isLeft a then a else b) (Data.List.head ls) (Data.List.tail ls)              return ()-          when (opc /= 9) $ someDef h streams streamMap (bs, opc, mvar)+          when (opc /= 9) $ writeQuery h streams streamMap (bs, opc, mvar)) (\e -> do+                                                                                  sm <- takeMVar streamMap+                                                                                  nm <- sequence $ fmap (\mvar -> tryPutMVar mvar (Left $ ShortStr $ DBL.fromStrict $ C8.pack $ "connection to node " <> show host <> " was lost")) sm+                                                                                  putMVar streamMap sm+                                                                                  (host, port, h, streams, streamMap) <- setupNode host port ri+                                                                                  sendThread host port ri h streams streamMap driverQ+                                                                                  receiveThread host port ri h streams streamMap)   return ()  @@ -186,14 +198,29 @@   {-# NOINLINE allConnections #-}-allConnections :: MVar [(HostName, Handle, MVar [Int16], MVar(IM.IntMap (MVar (Either ShortStr Result))))]+allConnections :: MVar [(HostName, PortID, Handle, MVar [Int16], MVar(IM.IntMap (MVar (Either ShortStr Result))))] allConnections = unsafePerformIO $ newMVar []  +setupNode peer port rInt@(RetryInterval ri) = catchAny (do+  streams <- newMVar ([0..32767] :: [Int16])+  streamMap <- newMVar (IM.empty :: (IM.IntMap (MVar (Either ShortStr Result))))+  h <- connectTo peer port+  startUp h+  return (peer, port, h, streams, streamMap)) (\e -> do+                                                  threadDelay ri+                                                  setupNode peer port rInt)++++newtype RetryInterval = RetryInterval Int++ -- | The first function you need to call. It initializes the driver and connects to the cluster. -- You only need to specify one node from your cluster here.-init :: HostName -> PortID -> ExceptT ShortStr IO Candle-init host port = do+-- Retryinterval is the interval with which connection to a node will be retried in case of disconnection.+init :: HostName -> PortID -> RetryInterval -> ExceptT ShortStr IO Candle+init host port ri = do     streamNum <- liftIO $ newMVar 0     streams <- liftIO $ newMVar ([0..32767] :: [Int16])     streamMap <- liftIO $ newMVar (IM.empty :: (IM.IntMap (MVar (Either ShortStr Result))))@@ -208,18 +235,11 @@       Left err -> throwError err       Right rows -> do         let peers = (\(CQLString p) -> Data.List.concat <$> Data.List.intersperse "." $ fmap show (unpack p)) <$> catMaybes (fmap (\r -> fromCQL r (CQLString "peer")::Maybe CQLString) rows)-        oCons <- liftIO $ sequence $ fmap (\peer -> do-                        streams <- newMVar ([0..32767] :: [Int16])-                        streamMap <- newMVar (IM.empty :: (IM.IntMap (MVar (Either ShortStr Result))))-                        h <- connectTo peer port-                        startUp h-                        -- registerForEvents h-                        return (peer, h, streams, streamMap)-                    ) peers-        let connections = (host, h, streams, streamMap) : oCons+        oCons <- liftIO $ sequence $ fmap (\peer -> setupNode peer port ri) peers+        let connections = (host, port, h, streams, streamMap) : oCons         allCon <- liftIO $ takeMVar allConnections         liftIO $ putMVar allConnections connections-        forM_ connections (\(_, h, streams, streamMap) -> do-          liftIO $ receiveThread h streams streamMap-          liftIO $ sendThread h streams streamMap driverQ)+        forM_ connections (\(host, port, h, streams, streamMap) -> do+          liftIO $ receiveThread host port ri h streams streamMap+          liftIO $ sendThread host port ri h streams streamMap driverQ)         return $ Candle driverQ

@@ -40,6 +40,7 @@ import           Network  + create :: String -> Q create s = Query (ShortStr $ DBL.fromStrict $ C8.pack "create " <> C8.pack s <> " ") [] @@ -70,7 +71,7 @@  -- | Prepare a query, returns a prepared query which can be fed to execCQL for execution. prepare :: ByteString -> ExceptT ShortStr (ReaderT Candle IO) Prepared-prepare q = do+prepare q  = do   (Candle driverQ) <- ask   mvar <- liftIO newEmptyMVar   let len = fromIntegral (C8.length q)::Int32@@ -78,7 +79,7 @@   liftIO $ atomically $ writeTBQueue driverQ (LongStr hd, 9, mvar)   res <- liftIO $ takeMVar mvar   case res of-      Left e -> throwError e+      Left e               -> throwError e       Right (RPrepared sb) -> return $ Prepared sb  @@ -94,7 +95,7 @@   liftIO $ atomically $ writeTBQueue driverQ (LongStr hd, 7, mvar)   res <- liftIO $ takeMVar mvar   case res of-    Left e -> throwError e+    Left e             -> throwError e     Right (RRows rows) -> return rows  @@ -108,7 +109,7 @@   liftIO $ atomically $ writeTBQueue driverQ (LongStr q, 10, mvar)   res <- liftIO $ takeMVar mvar   case res of-    Left e -> throwError e+    Left e             -> throwError e     Right (RRows rows) -> return rows  

@@ -60,7 +60,7 @@  main :: IO () main = do-    conRes <- runExceptT $ CQL.init "127.0.0.1" (PortNumber 9042)+    conRes <- runExceptT $ CQL.init "127.0.0.1" (PortNumber 9042) (RetryInterval 1000000)     case conRes of       Right ch ->         hspec $ before (threadDelay 1000000) (describe "driver should be able to" $ do@@ -83,7 +83,7 @@               case prep of                 Left e -> print e                 Right p -> do-                  res <- (runReaderT . runExceptT) (execCQL QUORUM p [+                  res <- (runReaderT . runExceptT) (execCQL ALL p [                     put (104::Int64),                     put (15::Int32),                     put True,@@ -100,14 +100,14 @@               case prep of                 Left e -> print e                 Right p -> do-                  res <- (runReaderT . runExceptT) (execCQL QUORUM p [+                  res <- (runReaderT . runExceptT) (execCQL ALL p [                     put (104::Int64),                     put (15::Int32)]) ch                   fmap fromRow <$> res `shouldBe` Right [Just (Emp {empID = 104, deptID = 15, alive = True, Main.id = fromJust $ fromString $ "38d0ceb1-9e3e-427c-bc36-0106398f672b", name = CQLString "Hot Shot", salary = CQLDouble 100000.0, someList = CQLList [1,2,3,4,5,6], someSet = CQLSet (fromList [CQLDouble 1.0e-3,CQLDouble 1000.0]), someMap = CQLMap $ DMS.fromList [(CQLString "some",CQLString "Things")]})]              it "select rows from table" $ do               let q = select "demodb.emp" # where' "empID" (104::Int64) # and' "deptID" (15::Int32)-              res <- (runReaderT . runExceptT) (runCQL QUORUM q) ch+              res <- (runReaderT . runExceptT) (runCQL ALL q) ch               fmap fromRow <$> res `shouldBe` Right [Just (Emp {empID = 104, deptID = 15, alive = True, Main.id = fromJust $ fromString $ "38d0ceb1-9e3e-427c-bc36-0106398f672b", name = CQLString "Hot Shot", salary = CQLDouble 100000.0, someList = CQLList [1,2,3,4,5,6], someSet = CQLSet (fromList [CQLDouble 1.0e-3,CQLDouble 1000.0]), someMap = CQLMap $ DMS.fromList [(CQLString "some",CQLString "Things")]})]              it "batch queries" $ do@@ -127,7 +127,7 @@                   res <- (runReaderT . runExceptT) (runBatch q) ch                   res `shouldBe` Right []                   let q = select "demodb.emp" # where' "empID" (101::Int64) # and' "deptID" (13::Int32)-                  res <- (runReaderT . runExceptT) (runCQL QUORUM q) ch+                  res <- (runReaderT . runExceptT) (runCQL ALL q) ch                   fmap fromRow <$> res `shouldBe` Right [Just (Emp {empID = 101, deptID = 13, alive = True, Main.id = fromJust $ fromString $ "48d0ceb1-9e3e-427c-bc36-0106398f672b", name = CQLString "Hot1 Shot1", salary = CQLDouble 10000.0, someList = CQLList [], someSet = CQLSet Data.Set.empty, someMap = CQLMap DMS.empty})]              it "drop a table" $ do