hascas (empty) → 0.1.0.0
raw patch · 12 files changed
+1313/−0 lines, 12 filesdep +basedep +binarydep +bytestringsetup-changed
Dependencies added: base, binary, bytestring, containers, data-binary-ieee754, hascas, hspec, mtl, network, stm, template-haskell, uuid
Files
- LICENSE +201/−0
- Setup.hs +2/−0
- hascas.cabal +59/−0
- src/Batch.hs +66/−0
- src/CQL.hs +9/−0
- src/Codec.hs +119/−0
- src/Common.hs +291/−0
- src/Derive.hs +47/−0
- src/Driver.hs +131/−0
- src/Encoding.hs +134/−0
- src/Query.hs +115/−0
- test/Spec.hs +139/−0
+ LICENSE view
@@ -0,0 +1,201 @@+ 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:++ (a) You must give any other recipients of the Work or+ Derivative Works a copy of this License; and++ (b) You must cause any modified files to carry prominent notices+ stating that You changed the files; and++ (c) 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++ (d) 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++ APPENDIX: How to apply the Apache License to your work.++ To apply the Apache License to your work, attach the following+ boilerplate notice, with the fields enclosed by brackets "{}"+ replaced with your own identifying information. (Don't include+ the brackets!) The text should be enclosed in the appropriate+ comment syntax for the file format. We also recommend that a+ file or class name and description of purpose be included on the+ same "printed page" as the copyright notice for easier+ identification within third-party archives.++ Copyright {yyyy} {name of copyright owner}++ Licensed under the Apache License, Version 2.0 (the "License");+ you may not use this file except in compliance with the License.+ You may obtain a copy of the License at++ http://www.apache.org/licenses/LICENSE-2.0++ Unless required by applicable law or agreed to in writing, software+ distributed under the License is distributed on an "AS IS" BASIS,+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.+ See the License for the specific language governing permissions and+ limitations under the License.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ hascas.cabal view
@@ -0,0 +1,59 @@+name: hascas+version: 0.1.0.0+synopsis: Cassandra driver for haskell+description: Please see README.md+homepage: https://github.com/eklavya/hascas#readme+license: Apache+license-file: LICENSE+author: Saurabh Rawat+maintainer: saurabh.rawat90@gmail.com+copyright: 2016 Saurabh Rawat+category: Database+build-type: Simple+-- extra-source-files:+cabal-version: >=1.10++library+ hs-source-dirs: src+ exposed-modules: CQL+ other-modules: Driver+ , Codec+ , Batch+ , Encoding+ , Query+ , Common+ , Derive+ build-depends: base >= 4.7 && < 5+ , bytestring+ , binary+ , network+ , uuid+ , containers+ , stm+ , data-binary-ieee754+ , mtl+ , template-haskell+ default-language: Haskell2010++test-suite hascas-test+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: Spec.hs+ build-depends: base+ , hascas+ , bytestring+ , binary+ , network+ , uuid+ , containers+ , stm+ , data-binary-ieee754+ , hspec+ , mtl+ , template-haskell+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ default-language: Haskell2010++source-repository head+ type: git+ location: https://github.com/eklavya/hascas
+ src/Batch.hs view
@@ -0,0 +1,66 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+++module Batch where+++import Control.Applicative+import Control.Concurrent+import Control.Concurrent.MVar+import Control.Concurrent.STM+import Control.Concurrent.STM.TBQueue+import Control.Exception (bracket, catch)+import Control.Monad (forM_, forever, replicateM)+import Control.Monad.Except+import Control.Monad.Reader+import Data.Binary+import Data.Binary.Get+import Data.Binary.IEEE754+import Data.Binary.Put+import Data.Bits+import Data.ByteString+import qualified Data.ByteString.Char8 as C8+import qualified Data.ByteString.Lazy as DBL+import Data.Int+import qualified Data.IntMap as IM+import Data.List+import qualified Data.Map.Strict as DMS++import Common+import Data.Maybe+import Data.Monoid as DM+import Data.UUID+import Debug.Trace+import GHC.Generics (Generic)+import GHC.IO.Handle (hClose, hFlush)+import Network+++instance Batchable LoggedBatch where+ runBatch (LoggedBatch qs) = do+ (Candle driverQ) <- ask+ mvar <- liftIO newEmptyMVar+ let q = encode (0::Int8) <> encode (fromIntegral (Data.List.length qs) :: Int16) <> mconcat qs <> encode SERIAL <> encode (0x00 :: Int8)+ liftIO $ atomically $ writeTBQueue driverQ (LongStr q, 13, mvar)+ res <- liftIO $ takeMVar mvar+ case res of+ Left e -> throwError e+ Right (RRows rows) -> return rows+++-- | Create a prepared batch statement with Prepared query and parameters.+prepBatch :: Prepared -> [Put] -> LoggedBatch+prepBatch (Prepared pid) ks =+ let b = encode (1::Int8) <> encode pid <> encode (fromIntegral (Data.List.length ks) :: Int16) <> mconcat (fmap (addLength . runPut) ks) in+ LoggedBatch [b]+++-- | Create a simple batch statement, with query string and parameters.+batch :: Q -> LoggedBatch+batch (Query (ShortStr q) bs) =+ let b = encode (0::Int8) <> encode (fromIntegral (DBL.length q) :: Int32) <> q <> encode (fromIntegral (Data.List.length bs) :: Int16) <> mconcat (fmap DBL.fromStrict bs) in+ LoggedBatch [b]
+ src/CQL.hs view
@@ -0,0 +1,9 @@+module CQL (module X) where++import Batch as X+import Codec as X+import Common as X+import Derive as X+import Driver as X+import Encoding as X+import Query as X
+ src/Codec.hs view
@@ -0,0 +1,119 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+++module Codec where+++import Control.Applicative+import Control.Concurrent+import Control.Concurrent.MVar+import Control.Concurrent.STM+import Control.Concurrent.STM.TBQueue+import Control.Exception (bracket, catch)+import Control.Monad (forM_, forever, replicateM)+import Data.Binary+import Data.Binary.Get+import Data.Binary.IEEE754+import Data.Binary.Put+import Data.Bits+import Data.ByteString+import qualified Data.ByteString.Char8 as C8+import qualified Data.ByteString.Lazy as DBL+import Data.Int+import qualified Data.IntMap as IM+import qualified Data.Map.Strict as DMS++import Common+import Data.List as List+import Data.Maybe+import Data.Monoid as DM+import qualified Data.Set as Set+import Data.UUID+import Debug.Trace+import GHC.Generics (Generic)+import GHC.IO.Handle (hClose, hFlush)+import Network+++instance Binary CQLDouble where+ get = CQLDouble <$> getFloat64be+ put (CQLDouble d) = putFloat64be d+++instance (Ord k, Binary k, Binary v) => Binary (CQLMap k v) where+ get = do+ zeroLen <- isEmpty+ if zeroLen+ then+ pure $ CQLMap DMS.empty+ else+ do+ num <- get :: Get Int32+ ls <- replicateM (fromIntegral num :: Int) $ do+ r <- get :: Get Bytes+ let Bytes b = r+ let kv = runGet (get :: Get k) b+ r <- get :: Get Bytes+ let Bytes b = r+ let vv = runGet (get :: Get v) b+ return (kv, vv)+ return $ CQLMap $ DMS.fromList ls+ put (CQLMap m) = do+ let n = DMS.size m+ put (fromIntegral n :: Int32)+ forM_ (DMS.toList m) (\(k, v) -> do+ let l = (addLength . runPut . put) k <> (addLength . runPut . put) v+ putLazyByteString l)+++instance (Binary a, Ord a) => Binary (CQLSet a) where+ get = do+ zeroLen <- isEmpty+ if zeroLen+ then+ pure $ CQLSet Set.empty+ else+ do+ num <- get :: Get Int32+ ls <- replicateM (fromIntegral num :: Int) getValue+ return $ CQLSet $ Set.fromList ls+ put (CQLSet s) = do+ let n = Set.size s+ put (fromIntegral n :: Int32)+ forM_ (Set.toList s) (\el -> do+ let l = (addLength . runPut .put) el+ putLazyByteString l)+++getValue :: (Binary a) => Get a+getValue = do+ r <- get :: Get Bytes+ let Bytes b = r+ return $ runGet get b+++instance (Binary a, Ord a) => Binary (CQLList a) where+ get = do+ zeroLen <- isEmpty+ if zeroLen+ then+ pure $ CQLList []+ else+ do+ num <- get :: Get Int32+ ls <- replicateM (fromIntegral num :: Int) getValue+ return $ CQLList ls+ put (CQLList ls) = do+ let n = List.length ls+ put (fromIntegral n :: Int32)+ forM_ ls (\ el -> do+ let l = (addLength . runPut . put) el+ putLazyByteString l)++instance Binary CQLString where+ get = (CQLString . DBL.toStrict) <$> getRemainingLazyByteString+ put (CQLString s) = putByteString s
+ src/Common.hs view
@@ -0,0 +1,291 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+++module Common where+++import Control.Applicative+import Control.Concurrent+import Control.Concurrent.MVar+import Control.Concurrent.STM+import Control.Concurrent.STM.TBQueue+import Control.Exception (bracket, catch)+import Control.Monad (forM_, forever, replicateM)+import Control.Monad.Except+import Control.Monad.Reader+import Data.Binary+import Data.Binary.Get+import Data.Binary.IEEE754+import Data.Binary.Put+import Data.Bits+import Data.ByteString+import qualified Data.ByteString.Char8 as C8+import qualified Data.ByteString.Lazy as DBL+import Data.Int+import qualified Data.IntMap as IM+import Data.List+import qualified Data.Map.Strict as DMS++import Data.Maybe+import Data.Monoid as DM+import qualified Data.Set as Set+import Data.UUID+import Debug.Trace+import GHC.Generics (Generic)+import GHC.IO.Handle (hClose, hFlush)+import Network+import Text.Printf (printf)+++++class Batchable a where+ -- | Run a batch query.+ runBatch :: a -- ^ A batchable query, currently only logged batches are supported.+ -> ExceptT ShortStr (ReaderT Candle IO) [Row]++newtype BatchQuery = BatchQuery DBL.ByteString++data LoggedBatch = LoggedBatch [DBL.ByteString]++instance Monoid LoggedBatch where+ mempty = LoggedBatch []+ mappend (LoggedBatch l1) (LoggedBatch l2) = LoggedBatch (l1 <> l2)++data Candle = Candle (TBQueue (LongStr, Word8, MVar (Either ShortStr Result)))++data Header = Header {+ protocolVersion :: Int8,+ flags :: Int8,+ stream :: Int16,+ opcode :: Int8,+ len :: Int32+}+ deriving(Eq, Show, Generic)++instance Binary Header++newtype ShortStr = ShortStr DBL.ByteString+ deriving(Eq, Ord, Show)++newtype LongStr = LongStr DBL.ByteString+ deriving(Show)++newtype Bytes = Bytes DBL.ByteString+ deriving(Show, Eq, Ord)++newtype ShortBytes = ShortBytes ByteString+ deriving(Show)++-- | Consistency levels.+data Consistency = ANY | ONE | TWO | THREE | QUORUM | ALL | LOCAL_QUORUM | EACH_QUORUM+ | SERIAL | LOCAL_SERIAL | LOCAL_ONE+ deriving(Eq, Generic)++newtype StringMap = StringMap (DMS.Map ShortStr ShortStr)++data OptionId = Ascii+ | Bigint+ | Blob+ | Boolean+ | Counter+ | Decimal+ | Double+ | Float+ | Int+ | Timestamp+ | Uuid+ | Varchar+ | Varint+ | Timeuuid+ | Inet+ | Date+ | Time+ | Smallint+ | Tinyint++data GlobalTableSpec = GlobalTableSpec {+ ksName :: ShortStr,+ tName :: ShortStr+}+ deriving(Show, Generic)++instance Binary GlobalTableSpec++data Metadata = Metadata {+ mflags :: Int32,+ numCols :: Int32,+ gts :: Maybe GlobalTableSpec,+ colSpecs :: [ColumnSpec]+}+ deriving(Show)++data ColumnSpec = ColumnSpec {+ ksname :: Maybe ShortStr,+ tname :: Maybe ShortStr,+ colName :: ShortStr,+ tpe :: Word16,+ elTpe :: Maybe Word16,+ elTpeV :: Maybe Word16+}+ deriving(Show)++data RowContent = RowContent {+ values :: [Bytes]+}+++data Rows = Rows {+ content :: [Row]+}+ deriving(Show)++-- | Row result type.+type Row = DMS.Map CQLString (Word16, Maybe Word16, Maybe Word16, Bytes)+++-- | Auto derivable class to get records from result rows.+class BuildRec a where+ fromRow :: Row -> Maybe a++-- | All field types must implement this class.+class FromCQL a where+ fromCQL :: Row -- ^ Result row+ -> CQLString -- ^ Field name that we want to get+ -> Maybe a++instance Monoid ShortStr where+ mempty = ShortStr ""+ mappend (ShortStr s1) (ShortStr s2) = ShortStr (s1 <> s2)++data Prepared = Prepared ShortBytes+ deriving(Show)++data Result = RRows [Row] | RPrepared ShortBytes+ deriving(Show)+++data Query a b where+ Query :: ShortStr -> [ByteString] -> Query ShortStr [ByteString]++type Q = Query ShortStr [ByteString]++addLength :: DBL.ByteString -> DBL.ByteString+addLength bs = encode (fromIntegral (DBL.length bs) :: Int32) <> bs++getErr :: Get (Int32, ShortStr)+getErr = do+ erc <- get :: Get Int32+ erm <- get :: Get ShortStr+ return (erc, erm)+++instance Binary ShortStr where+ put (ShortStr s) =+ let len = fromIntegral (DBL.length s) :: Int16 in+ do+ put (len :: Int16)+ forM_ (DBL.unpack s) (\c -> put (c :: Word8))++ get = do+ len <- get :: Get Int16+ bs <- replicateM (fromIntegral len ::Int) getWord8+ return $ ShortStr $ DBL.pack bs++++instance Binary LongStr where+ put (LongStr s) =+ let len = fromIntegral (DBL.length s) :: Int32 in+ do+ put (len :: Int32)+ forM_ (DBL.unpack s) (\c -> put (c :: Word8))++ get = do+ len <- get :: Get Int32+ bs <- replicateM (fromIntegral len ::Int) getWord8+ return $ LongStr $ DBL.pack bs++++instance Binary Bytes where+ put (Bytes bs) =+ let len = fromIntegral (DBL.length bs) :: Int32 in+ do+ put (len :: Int32)+ forM_ (DBL.unpack bs) (\c -> put (c :: Word8))++ get = do+ len <- get :: Get Int32+ bs <- replicateM (fromIntegral len ::Int) getWord8+ return $ Bytes $ DBL.pack bs++++instance Binary ShortBytes where+ put (ShortBytes s) =+ let len = fromIntegral (C8.length s) :: Int16 in+ do+ put (len :: Int16)+ forM_ (unpack s) (\c -> put (c :: Word8))++ get = do+ len <- get :: Get Int16+ bs <- replicateM (fromIntegral len ::Int) getWord8+ return $ ShortBytes $ pack bs+++conToWord :: Consistency -> Word16+conToWord c+ | c == ANY = 0x0000+ | c == ONE = 0x0001+ | c == TWO = 0x0002+ | c == THREE = 0x0003+ | c == QUORUM = 0x0004+ | c == ALL = 0x0005+ | c == LOCAL_QUORUM = 0x0006+ | c == EACH_QUORUM = 0x0007+ | c == SERIAL = 0x0008+ | c == LOCAL_SERIAL = 0x0009+ | c == LOCAL_ONE = 0x000A++wordToCon :: Word16 -> Consistency+wordToCon w+ | w == 0x0000 = ANY+ | w == 0x0001 = ONE+ | w == 0x0002 = TWO+ | w == 0x0003 = THREE+ | w == 0x0004 = QUORUM+ | w == 0x0005 = ALL+ | w == 0x0006 = LOCAL_QUORUM+ | w == 0x0007 = EACH_QUORUM+ | w == 0x0008 = SERIAL+ | w == 0x0009 = LOCAL_SERIAL+ | w == 0x000A = LOCAL_ONE++instance Binary Consistency where+ get = do+ c <- get :: Get Word16+ return $ wordToCon c++ put c = put $ conToWord c++newtype CQLDouble = CQLDouble Double+ deriving(Show, Eq, Ord, Generic)++newtype CQLString = CQLString ByteString+ deriving(Show, Eq, Ord, Generic)++newtype CQLMap k v = CQLMap (DMS.Map k v)+ deriving(Show, Eq, Ord, Generic)++newtype CQLSet el = CQLSet (Set.Set el)+ deriving(Show, Eq, Ord, Generic)++newtype CQLList el = CQLList [el]+ deriving(Show, Eq, Ord, Generic)
+ src/Derive.hs view
@@ -0,0 +1,47 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE TemplateHaskell #-}++module Derive(deriveBuildRec) where++import Common hiding (Q)+import Control.Monad+import Data.Binary+import Data.Binary.Get+import Data.ByteString.Lazy as DBL (ByteString)+import Data.Char (toLower)+import Data.Map.Strict (Map, lookup)+import Language.Haskell.TH+import Prelude hiding (lookup)+++genNames :: Int -> Q [Name]+genNames n = replicateM n $ newName "x"++statements :: [Name] -> [(Name, Strict, Type)] -> Name -> [StmtQ]+statements names args r = fmap (\(n, (name, _, tpe)) -> do+ bytesName <- newName "b"+ bindS+ (varP n)+ (infixE+ (Just (infixE (Just (appE (varE 'runGet) (sigE (varE 'get) (appT (conT ''Get) (pure tpe))))) (varE $ mkName ".") (Just (lamE [tupP [wildP,wildP,wildP, conP (mkName "CQL.Bytes") [varP bytesName]]] (varE bytesName)))))+ (varE $ mkName "<$>")+ (Just (appE (appE (varE 'lookup) (appE (conE 'CQLString) (litE (StringL (fmap toLower (nameBase name)))))) (varE r))))+ ) (names `zip` args)++rtrnStmt names conName = [noBindS (appE (varE $ mkName "return") (applyRec (appE (conE conName) (varE $ head names)) (tail names)) )]++applyRec = foldl (\ex nm -> appE ex (varE nm))++-- | Derives BuildRec instances for record types.+deriveBuildRec a = do+ TyConI (DataD _ cName _ constructors _) <- reify a+ case head constructors of+ (RecC conName args) -> do+ names <- genNames $ length args+ rowName <- newName "r"+ [d| instance BuildRec $(conT a) where+ fromRow = $(lamE [varP rowName] (doE (statements names args rowName ++ rtrnStmt names conName))) |]++ _ -> fail "deriveBuildRec: Only simple records supported, is the type you are deriving for, a record type?"
+ src/Driver.hs view
@@ -0,0 +1,131 @@+{-# LANGUAGE GADTs #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+++module Driver (Driver.init) where+++import Control.Applicative+import Control.Concurrent+import Control.Concurrent.MVar+import Control.Concurrent.STM+import Control.Concurrent.STM.TBQueue+import Control.Exception (bracket, catch)+import Control.Monad (forM_, forever, replicateM)+import Data.Binary+import Data.Binary.Get+import Data.Binary.IEEE754+import Data.Binary.Put+import Data.Bits+import Data.ByteString+import qualified Data.ByteString.Char8 as C8+import qualified Data.ByteString.Lazy as DBL+import Data.Int+import qualified Data.IntMap as IM+import Data.List+import qualified Data.Map.Strict as DMS++import Codec+import Common+import Data.Maybe+import Data.Monoid as DM+import Data.UUID+import Debug.Trace+import Encoding+import GHC.Generics (Generic)+import GHC.IO.Handle (Handle, hClose, hFlush)+import Network+++startUp :: Handle -> IO ()+startUp h = do+ let s = encode $ StringMap (DMS.fromList [(ShortStr "CQL_VERSION", ShortStr "3.0.0")])+ let startup = DBL.toStrict $ DBL.pack [4, 0, 0, 1, 1] <> encode (fromIntegral (DBL.length s)::Int32) <> s+ hPut h startup+ header <- hGet h 9+ he <- pure $ runGet (get :: Get Header) $ DBL.fromStrict header+ hGet h (fromIntegral $ len he)+ return ()+++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)+ if opcode he == 0+ then do+ let (erCode, erMsg) = runGet getErr (DBL.fromStrict p)+ mvar <- getResHolder he streams streamMap+ putMVar mvar (Left erMsg)+ else do+ let resultType = runGet (get :: Get Int32) (DBL.fromStrict p)++ 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)++ 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 [])+ return ()+++getResHolder :: Header -> MVar [Int16] -> MVar(IM.IntMap (MVar (Either ShortStr Result))) -> IO (MVar (Either ShortStr Result))+getResHolder he1 streams streamMap = do+ m <- takeMVar streamMap+ let strm = stream he1+ let fm = IM.lookup (fromIntegral strm :: Int) m+ let mvar = fromJust fm+ putMVar streamMap (IM.delete (fromIntegral strm :: Int) m)+ strs <- takeMVar streams+ putMVar streams (strs ++ [strm])+ return 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+ strs <- takeMVar streams+ case Data.List.uncons strs of+ Just (i, strsTail) -> do+ putMVar streams strsTail+ (bs, opc, mvar) <- atomically $ readTBQueue driverQ+ m <- takeMVar streamMap+ let m' = IM.insert (fromIntegral i :: Int) mvar m+ putMVar streamMap m'+ seq m' (return ())+ let bs' = Data.ByteString.pack [4, 0] <> DBL.toStrict (encode i <> DBL.pack [opc] <> encode bs)+ hPut h bs'+ Nothing -> do+ putMVar streams strs+ sendThread h streams streamMap driverQ+ return ()++-- | The first +init :: HostName -> PortID -> IO Candle+init host port = do+ streamNum <- newMVar 0+ streams <- newMVar ([0..32767] :: [Int16])+ streamMap <- newMVar (IM.empty :: (IM.IntMap (MVar (Either ShortStr Result))))+ driverQ <- atomically $ newTBQueue 32768++ h <- connectTo host port+ startUp h+ receiveThread h streams streamMap+ sendThread h streams streamMap driverQ+ return $ Candle driverQ
+ src/Encoding.hs view
@@ -0,0 +1,134 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+++module Encoding where+++import Common+import Control.Applicative+import Control.Concurrent+import Control.Concurrent.MVar+import Control.Concurrent.STM+import Control.Concurrent.STM.TBQueue+import Control.Exception (bracket, catch)+import Control.Monad (forM_, forever, replicateM)+import Data.Binary+import Data.Binary.Get+import Data.Binary.IEEE754+import Data.Binary.Put+import Data.Bits+import Data.ByteString+import qualified Data.ByteString.Char8 as C8+import qualified Data.ByteString.Lazy as DBL+import Data.Int+import qualified Data.IntMap as IM+import Data.List+import qualified Data.Map.Strict as DMS+import Data.Maybe+import Data.Monoid as DM+import qualified Data.Set as Set+import Data.UUID+import Debug.Trace+import GHC.Generics (Generic)+import GHC.IO.Handle (hClose, hFlush)+import Network+++instance Binary StringMap where+ put (StringMap m) = do+ put (fromIntegral (DMS.size m) :: Int16)+ forM_ (DMS.toList m) (\(k, v) -> put k >> put v)++ get = do+ len <- get :: Get Int16+ pl <- replicateM (fromIntegral len :: Int) $ do+ k <- get :: Get ShortStr+ v <- get :: Get ShortStr+ return (k, v)+ return $ StringMap $ DMS.fromList pl+++getColumnSpec ksname tname = do+ colName <- get+ tpe <- get+ case tpe of+ t | t == 32 || t == 34 -> do+ elTpe <- get+ return ColumnSpec {ksname = ksname, tname = tname, colName, tpe, elTpe = Just elTpe, elTpeV = Nothing}+ 33 -> do+ elTpe <- get+ elTpeV <- get+ return ColumnSpec {ksname = ksname, tname = tname, colName, tpe, elTpe = Just elTpe, elTpeV = Just elTpeV}+ _ ->+ return ColumnSpec {ksname = ksname, tname = tname, colName, tpe, elTpe = Nothing, elTpeV = Nothing}+++getMeta :: Get Metadata+getMeta = do+ mflags <- get+ numCols <- get+ if mflags .&. 1 == 1+ then do+ gts <- get :: Get GlobalTableSpec+ colSpecs <- replicateM (fromIntegral numCols :: Int) $ getColumnSpec Nothing Nothing+ return Metadata {mflags, numCols, gts = Just gts, colSpecs}+ else do+ colSpecs <- replicateM (fromIntegral numCols :: Int) $ do+ ksname <- get :: Get ShortStr+ tname <- get :: Get ShortStr+ getColumnSpec (Just ksname) (Just tname)+ return Metadata {mflags, numCols, gts = Nothing, colSpecs}+++getRows :: Get Rows+getRows = do+ meta <- getMeta+ numRows <- get :: Get Int32+ rs <- replicateM (fromIntegral numRows :: Int) $ mapM (\cs -> do+ r <- get :: Get Bytes+ let (ShortStr s) = colName cs+ return (CQLString $ DBL.toStrict s, (tpe cs, elTpe cs, elTpeV cs, r))) $ colSpecs meta+ return Rows { content = fmap DMS.fromList rs}+++instance FromCQL Int8 where+ fromCQL r s = (\(_, _, _, Bytes b) -> runGet (get :: Get Int8) b) <$> DMS.lookup s r++instance FromCQL Int16 where+ fromCQL r s = (\(_, _, _, Bytes b) -> runGet (get :: Get Int16) b) <$> DMS.lookup s r++instance FromCQL Int32 where+ fromCQL r s = (\(_, _, _, Bytes b) -> runGet (get :: Get Int32) b) <$> DMS.lookup s r++instance FromCQL Int64 where+ fromCQL r s = (\(_, _, _, Bytes b) -> runGet (get :: Get Int64) b) <$> DMS.lookup s r++instance FromCQL Double where+ fromCQL r s = (\(_, _, _, Bytes b) -> runGet getFloat64be b) <$> DMS.lookup s r++instance FromCQL Bool where+ fromCQL r s = (\(_, _, _, Bytes b) -> runGet (get :: Get Bool) b) <$> DMS.lookup s r++instance FromCQL UUID where+ fromCQL r s = (\(_, _, _, Bytes b) -> runGet (get :: Get UUID) b) <$> DMS.lookup s r++instance FromCQL CQLString where+ fromCQL r s = (\(_, _, _, Bytes b) -> (CQLString . DBL.toStrict) b) <$> DMS.lookup s r++instance (FromCQL k, FromCQL v, Ord k, Binary k, Binary v) => FromCQL (DMS.Map k v) where+ fromCQL r s = (\(_, kt, vt, Bytes b) -> runGet (if DBL.length b == 0+ then+ pure DMS.empty+ else+ do+ num <- get :: Get Int32+ ls <- replicateM (fromIntegral num :: Int) $ do+ kv <- get :: Get k+ vv <- get :: Get v+ return (kv, vv)+ return $ DMS.fromList ls+ ) b) <$> DMS.lookup s r
+ src/Query.hs view
@@ -0,0 +1,115 @@+{-# LANGUAGE GADTs #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+++module Query where+++import Control.Applicative+import Control.Concurrent+import Control.Concurrent.MVar+import Control.Concurrent.STM+import Control.Concurrent.STM.TBQueue+import Control.Exception (bracket, catch)+import Control.Monad (forM_, forever, replicateM)+import Control.Monad.Except+import Control.Monad.Reader+import Data.Binary+import Data.Binary.Get+import Data.Binary.IEEE754+import Data.Binary.Put+import Data.Bits+import Data.ByteString+import qualified Data.ByteString.Char8 as C8+import qualified Data.ByteString.Lazy as DBL+import Data.Int+import qualified Data.IntMap as IM+import Data.List+import qualified Data.Map.Strict as DMS++import Common+import Data.Maybe+import Data.Monoid as DM+import Data.UUID+import Debug.Trace+import Driver+import GHC.Generics (Generic)+import GHC.IO.Handle (hClose, hFlush)+import Network+++create :: String -> Q+create s = Query (ShortStr $ DBL.fromStrict $ C8.pack "create " <> C8.pack s <> " ") []++drop' :: String -> Q+drop' s = Query (ShortStr $ DBL.fromStrict $ C8.pack "drop " <> C8.pack s <> " ") []++select :: String -> Q+select s = Query (ShortStr $ DBL.fromStrict $ C8.pack "select * from " <> C8.pack s <> " ") []++limit :: Int -> Q+limit i = Query (ShortStr $ " limit " <> DBL.fromStrict (C8.pack (show i))) []++update :: String -> Q+update s = Query (ShortStr $ "update " <> DBL.fromStrict (C8.pack s) <> " set ") []++with :: (Binary k) => String -> k -> Q+with n v = Query (ShortStr (DBL.fromStrict (C8.pack n) <> " = ? ")) [(DBL.toStrict . addLength . runPut . put) v]++delete :: String -> Q+delete s = Query (ShortStr $ "delete from " <> DBL.fromStrict (C8.pack s) <> " ") []++where' :: (Binary k) => String -> k -> Q+where' n v = Query (ShortStr (" where " <> DBL.fromStrict (C8.pack n) <> " = ? ")) [(DBL.toStrict . addLength . runPut . put) v]++and' :: (Binary k) => String -> k -> Q+and' n v = Query (ShortStr (" and " <> DBL.fromStrict (C8.pack n) <> " = ? ")) [(DBL.toStrict . addLength . runPut . put) v]+++prepare :: ByteString -> ExceptT ShortStr (ReaderT Candle IO) Prepared+prepare q = do+ (Candle driverQ) <- ask+ mvar <- liftIO newEmptyMVar+ let len = fromIntegral (C8.length q)::Int32+ let hd = encode len <> DBL.fromStrict q+ liftIO $ atomically $ writeTBQueue driverQ (LongStr hd, 9, mvar)+ res <- liftIO $ takeMVar mvar+ case res of+ Left e -> throwError e+ Right (RPrepared sb) -> return $ Prepared sb+++runCQL :: Consistency -> Q -> ExceptT ShortStr (ReaderT Candle IO) [Row]+runCQL c (Query (ShortStr q) bs) = do+ (Candle driverQ) <- ask+ mvar <- liftIO newEmptyMVar+ let flagAndNum = if Data.List.null bs then encode (0x00 :: Int8) else encode (0x01 :: Int8) <> encode (fromIntegral (Data.List.length bs) :: Int16)+ let q' = q <> encode c <> flagAndNum <> DBL.fromStrict (mconcat bs)+ let len = fromIntegral (DBL.length q)::Int32+ let hd = encode len <> q'+ liftIO $ atomically $ writeTBQueue driverQ (LongStr hd, 7, mvar)+ res <- liftIO $ takeMVar mvar+ case res of+ Left e -> throwError e+ Right (RRows rows) -> return rows+++execCQL :: Consistency -> Prepared -> [Put] -> ExceptT ShortStr (ReaderT Candle IO) [Row]+execCQL c (Prepared pid) ls = do+ (Candle driverQ) <- ask+ mvar <- liftIO newEmptyMVar+ let flag = if Data.List.null ls then 0x00 else 0x01+ let q = encode pid <> (encode c <> encode (flag :: Int8)) <> encode (fromIntegral (Data.List.length ls) :: Int16) <> mconcat (fmap (addLength . runPut) ls)+ liftIO $ atomically $ writeTBQueue driverQ (LongStr q, 10, mvar)+ res <- liftIO $ takeMVar mvar+ case res of+ Left e -> throwError e+ Right (RRows rows) -> return rows+++(#) :: Q -> Q -> Q+(#) (Query s1 bs1) (Query s2 bs2) = Query (s1 <> s2) (bs1 <> bs2)++infixl 7 #
+ test/Spec.hs view
@@ -0,0 +1,139 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TemplateHaskell #-}+++module Main where+++import Control.Applicative+import Control.Concurrent+import Control.Concurrent.MVar+import Control.Concurrent.STM+import Control.Concurrent.STM.TBQueue+import Control.Exception (bracket, catch)+import Control.Monad (forM_, forever, replicateM)+import Control.Monad.Except+import Control.Monad.Reader+import CQL+import Data.Binary+import Data.Binary.Get+import Data.Binary.Put+import Data.Bits+import Data.ByteString+import qualified Data.ByteString.Char8 as C8+import qualified Data.ByteString.Lazy as DBL+import Data.Int+import qualified Data.IntMap as IM+import Data.List+import Data.Map.Strict (lookup)+import qualified Data.Map.Strict as DMS+import Data.Maybe+import Data.Monoid as DM+import Data.Set+import Data.UUID+import Debug.Trace+import GHC.Generics (Generic)+import GHC.IO.Handle (hClose, hFlush)+import Network+import Test.Hspec+++data Emp = Emp {+ empID :: Int64,+ deptID :: Int32,+ alive :: Bool ,+ id :: UUID,+ name :: CQLString,+ salary :: CQLDouble,+ someList :: CQLList Int32,+ someSet :: CQLSet CQLDouble,+ someMap :: CQLMap CQLString CQLString+}+ deriving(Show, Eq)++deriveBuildRec ''Emp+++main :: IO ()+main = do+ ch <- CQL.init "127.0.0.1" (PortNumber 9042)++ hspec $+ describe "driver should be able to" $ do+ it "create a keyspace" $ do+ let q = create "keyspace demodb WITH REPLICATION = {'class' : 'SimpleStrategy','replication_factor': 1}"+ res <- (runReaderT . runExceptT) (runCQL LOCAL_ONE q) ch+ res `shouldBe` Right []++ it "create a table" $ do+ let tableQuery = "TABLE demodb.emp (empID bigint,deptID int,alive boolean,id uuid,name varchar,salary double,"+ ++ "someset set<double>,somelist list<int>,somemap map<text, text>,PRIMARY KEY (empID, deptID))"+ let q = create tableQuery+ res <- (runReaderT . runExceptT) (runCQL LOCAL_ONE q) ch+ res `shouldBe` Right []++ it "execute prepared queries" $ do+ prep <- (runReaderT . runExceptT) (prepare "INSERT INTO demodb.emp (empID,deptID,alive,id,name,salary,somelist,someset,somemap) VALUES (?,?,?,?,?,?,?,?,?)") ch+ case prep of+ Left e -> print e+ Right p -> do+ res <- (runReaderT . runExceptT) (execCQL LOCAL_ONE p [+ put (104::Int64),+ put (15::Int32),+ put True,+ put $ fromJust $ fromString "38d0ceb1-9e3e-427c-bc36-0106398f672b",+ put $ CQLString "Hot Shot",+ put $ CQLDouble 100000.0,+ put ((CQLList [1,2,3,4,5,6]) :: CQLList Int32),+ put ((CQLSet $ fromList [CQLDouble 0.001, CQLDouble 1000.0]) :: CQLSet CQLDouble),+ put $ CQLMap $ DMS.fromList [(CQLString "some", CQLString "Things")]]) ch+ res `shouldBe` Right []++ it "execute prepared queries and get results" $ do+ prep <- (runReaderT . runExceptT) (prepare "select empID, deptID, alive, id, name, salary, someset, somemap, somelist from demodb.emp where empid = ? and deptid = ?") ch+ case prep of+ Left e -> print e+ Right p -> do+ res <- (runReaderT . runExceptT) (execCQL LOCAL_ONE 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 LOCAL_ONE 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+ prep <- (runReaderT . runExceptT) (prepare "INSERT INTO demodb.emp (empID, deptID, alive, id, name, salary) VALUES (?, ?, ?, ?, ?, ?)") ch+ case prep of+ Left e -> print e+ Right p -> do+ let q = batch (update "demodb.emp" # with "name" (CQLString "some name") # where' "empID" (104::Int64) # and' "deptID" (15::Int32)) <>+ batch (update "demodb.emp" # with "alive" False # where' "empID" (104::Int64) # and' "deptID" (15::Int32)) <>+ prepBatch p [+ put (101::Int64),+ put (13::Int32),+ put True,+ put $ fromJust $ fromString "48d0ceb1-9e3e-427c-bc36-0106398f672b",+ put $ CQLString "Hot1 Shot1",+ put $ CQLDouble 10000.0]+ 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 LOCAL_ONE 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+ let q = drop' "table demodb.emp"+ res <- (runReaderT . runExceptT) (runCQL LOCAL_ONE q) ch+ res `shouldBe` Right []++ it "drop a keyspace" $ do+ let q = drop' "keyspace demodb"+ res <- (runReaderT . runExceptT) (runCQL LOCAL_ONE q) ch+ res `shouldBe` Right []