diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/hascas.cabal b/hascas.cabal new file mode 100644 --- /dev/null +++ b/hascas.cabal @@ -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 diff --git a/src/Batch.hs b/src/Batch.hs new file mode 100644 --- /dev/null +++ b/src/Batch.hs @@ -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] diff --git a/src/CQL.hs b/src/CQL.hs new file mode 100644 --- /dev/null +++ b/src/CQL.hs @@ -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 diff --git a/src/Codec.hs b/src/Codec.hs new file mode 100644 --- /dev/null +++ b/src/Codec.hs @@ -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 diff --git a/src/Common.hs b/src/Common.hs new file mode 100644 --- /dev/null +++ b/src/Common.hs @@ -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) diff --git a/src/Derive.hs b/src/Derive.hs new file mode 100644 --- /dev/null +++ b/src/Derive.hs @@ -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?" diff --git a/src/Driver.hs b/src/Driver.hs new file mode 100644 --- /dev/null +++ b/src/Driver.hs @@ -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 diff --git a/src/Encoding.hs b/src/Encoding.hs new file mode 100644 --- /dev/null +++ b/src/Encoding.hs @@ -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 diff --git a/src/Query.hs b/src/Query.hs new file mode 100644 --- /dev/null +++ b/src/Query.hs @@ -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 # diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -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,somelist list,somemap map,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 []