continuum-client (empty) → 0.1.0.0
raw patch · 6 files changed
+495/−0 lines, 6 filesdep +basedep +bytestringdep +cerealsetup-changed
Dependencies added: base, bytestring, cereal, containers, mtl, nanomsg-haskell, time
Files
- LICENSE +0/−0
- Setup.hs +2/−0
- continuum-client.cabal +28/−0
- src/Continuum/Client/Base.hs +53/−0
- src/Continuum/Common/Serialization.hs +218/−0
- src/Continuum/Common/Types.hs +194/−0
+ LICENSE view
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ continuum-client.cabal view
@@ -0,0 +1,28 @@+name: continuum-client+stability: experimental+description: Continuum Database Client+version: 0.1.0.0+license: MIT+license-file: LICENSE+author: Alex Petrov+maintainer: alexp@coffeenco.de+build-type: Simple+cabal-version: >=1.10++source-repository head+ type: git+ location: https://github.com/ifesdjeen/continuum.git++library+ default-language: Haskell2010+ exposed-modules: Continuum.Client.Base+ , Continuum.Common.Types+ , Continuum.Common.Serialization+ hs-source-dirs: src+ build-depends: base >= 4 && < 5+ , containers >0.5+ , bytestring >= 0.10.4.0+ , cereal >= 0.4.0.1+ , nanomsg-haskell >= 0.2.2+ , time >=1.4.2+ , mtl >2.1
+ src/Continuum/Client/Base.hs view
@@ -0,0 +1,53 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleContexts #-}++module Continuum.Client.Base+ (module Continuum.Client.Base+ , module Continuum.Common.Types+ )+ where++import qualified Nanomsg as N+import Data.Serialize (encode, decode)+import qualified Continuum.Common.Serialization as S+import Continuum.Common.Types++data ContinuumClient = ContinuumClient+ { clientSocket :: N.Socket N.Req+ , clientEndpoint :: N.Endpoint }++connect :: String+ -> String+ -> IO ContinuumClient++connect host port = do+ socket <- N.socket N.Req+ endpoint <- N.connect socket ("tcp://" ++ host ++ ":" ++ port)+ return ContinuumClient { clientSocket = socket+ , clientEndpoint = endpoint}++disconnect :: ContinuumClient -> IO ()+disconnect client = do+ N.shutdown (clientSocket client) (clientEndpoint client)+ N.close (clientSocket client)+ return ()++executeQuery :: ContinuumClient+ -> DbName+ -> SelectQuery+ -> IO (DbErrorMonad DbResult)+executeQuery client dbName query = do+ let sock = (clientSocket client)+ _ <- N.send sock (encode (Select dbName query))+ bs <- N.recv sock+ return $ (S.decodeDbResult bs)++sendRequest :: ContinuumClient+ -> Request+ -> IO (DbErrorMonad DbResult)+sendRequest client request = do+ let sock = (clientSocket client)+ _ <- N.send sock (encode request)+ bs <- N.recv sock+ return $ (S.decodeDbResult bs)
+ src/Continuum/Common/Serialization.hs view
@@ -0,0 +1,218 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Continuum.Common.Serialization where++import Foreign+import Continuum.Common.Types++import Data.Serialize as S+import qualified Data.ByteString as B+import qualified Data.Map as Map++import Control.Applicative ( (<$>) )+import Data.List ( elemIndex )+import Data.Maybe ( isJust, fromJust, catMaybes )+import Control.Monad.Except ( forM_, throwError )+import Control.Monad ( join )++-- import Debug.Trace++data Success = Success++-- validate :: DbSchema -> DbRecord -> Either String Success+-- validate = error "Not Implemented"++-- |+-- | ENCODING+-- |++encodeRecord :: DbSchema -> DbRecord -> Integer -> (B.ByteString, B.ByteString)+encodeRecord schema (DbRecord timestamp vals) sid = (encodeKey, encodeValue)+ where encodeKey = B.concat [(packWord64 timestamp), (packWord64 sid)]+ encodedParts = fmap fastEncodeValue $ catMaybes $ (\x -> Map.lookup x vals) <$> (fields schema)+ lengths = B.length <$> encodedParts+ encodeValue = runPut $ do+ -- Change to B.Pack+ forM_ lengths (putWord8 . fromIntegral)+ forM_ encodedParts putByteString+ -- encode . catMaybes $ fmap (\x -> Map.lookup x vals) (fields schema)++encodeBeginTimestamp :: Integer -> B.ByteString+encodeBeginTimestamp timestamp =+ B.concat [(packWord64 timestamp), (packWord64 0)]++encodeEndTimestamp :: Integer -> B.ByteString+encodeEndTimestamp timestamp =+ B.concat [(packWord64 timestamp), (packWord64 999999)]++encodeSchema :: DbSchema -> B.ByteString+encodeSchema = encode++decodeSchema :: (DbName, B.ByteString)+ -> DbErrorMonad DbResult+decodeSchema (dbName, encodedSchema) =+ case (decode encodedSchema) of+ (Left err) -> throwError $ SchemaDecodingError err+ (Right schema) -> return $ DbSchemaResult (dbName, schema)++decodeQuery :: B.ByteString+ -> DbErrorMonad SelectQuery+decodeQuery encodedQuery =+ case (decode encodedQuery) of+ (Left err) -> throwError $ SchemaDecodingError err+ (Right query) -> return query++decodeDbResult :: B.ByteString+ -> DbErrorMonad DbResult+decodeDbResult encodedDbResult =+ case (decode encodedDbResult) of+ (Left err) -> throwError $ SchemaDecodingError err+ (Right query) -> join $ return query+++-- |+-- | DECODING+-- |++decodeRecord :: Decoding+ -> DbSchema+ -> Decoder++decodeRecord (Field field) schema !(k, bs) = do+ decodedK <- decodeKey k+ decodedVal <- if isJust idx+ then decodeFieldByIndex schema indices (fromJust idx) bs+ else throwError FieldNotFoundError+ return $! FieldRes (decodedK, decodedVal)+ where idx = elemIndex field (fields schema)+ indices = decodeIndexes schema bs++decodeRecord Record schema !(k, bs) = do+ timestamp <- decodeKey k+ decodedValue <- decodeValues schema bs+ return $! RecordRes $ DbRecord timestamp (Map.fromList $ zip (fields schema) decodedValue)++decodeRecord (Fields flds) schema (k, bs) = do+ decodedK <- decodeKey k+ decodedVal <- if isJust idxs+ then mapM (\idx -> decodeFieldByIndex schema (decodeIndexes schema bs) idx bs) (fromJust idxs)+ else throwError FieldNotFoundError+ return $! FieldsRes (decodedK, decodedVal)+ where+ {-# INLINE idxs #-}+ idxs = mapM (`elemIndex` (fields schema)) flds+{-# INLINE decodeRecord #-}++-- |+-- | DECODING Utility Functions+-- |++decodeKey :: B.ByteString -> DbErrorMonad Integer+decodeKey x = return $ unpackWord64 (B.take 8 x)+{-# INLINE decodeKey #-}++decodeChunkKey :: (B.ByteString, B.ByteString) -> DbErrorMonad DbResult+decodeChunkKey (x, _) = return $ KeyRes (unpackWord64 (B.take 8 x))++{-# INLINE decodeChunkKey #-}++decodeIndexes :: DbSchema -> B.ByteString -> [Int]+decodeIndexes schema bs =+ map fromIntegral $ B.unpack $ B.take (length (fields schema)) bs+{-# INLINE decodeIndexes #-}++slide :: [a] -> [(a, a)]+slide (f:s:xs) = (f,s) : slide (s:xs)+slide _ = []++decodeValues :: DbSchema -> B.ByteString -> DbErrorMonad [DbValue]+decodeValues schema bs = mapM (\(t, s) -> fastDecodeValue t s) (zip (schemaTypes schema) bytestrings)+ where+ indices = decodeIndexes schema bs+ bytestrings = snd (foldl step (B.drop (length indices) bs, []) indices)+ step (remaining', acc) n = (B.drop n remaining', acc ++ [B.take n remaining'])++-- | Decodes field by index+decodeFieldByIndex :: DbSchema+ -> [Int]+ -> Int+ -> B.ByteString+ -> DbErrorMonad DbValue+decodeFieldByIndex schema indices idx bs = fastDecodeValue ((schemaTypes schema) !! idx) bytestring+ where+ {-# INLINE bytestring #-}+ bytestring = B.take (indices !! idx) $ B.drop startFrom bs+ {-# INLINE startFrom #-}+ startFrom = (length indices) + (sum $ take idx indices)+{-# INLINE decodeFieldByIndex #-}+++-- |+-- | "FAST" CUSTOM SERIALIZATION+-- |++fastEncodeValue :: DbValue -> B.ByteString+fastEncodeValue (DbInt value) = packWord64 value+fastEncodeValue (DbString value) = value++fastDecodeValue :: DbType -> B.ByteString -> DbErrorMonad DbValue+fastDecodeValue DbtInt bs = return $ DbInt (unpackWord64 bs)+fastDecodeValue DbtString bs = return $ DbString bs++packWord64 :: Integer -> B.ByteString+packWord64 i =+ let w = (fromIntegral i :: Word64)+ in B.pack [ (fromIntegral (w `shiftR` 56) :: Word8)+ , (fromIntegral (w `shiftR` 48) :: Word8)+ , (fromIntegral (w `shiftR` 40) :: Word8)+ , (fromIntegral (w `shiftR` 32) :: Word8)+ , (fromIntegral (w `shiftR` 24) :: Word8)+ , (fromIntegral (w `shiftR` 16) :: Word8)+ , (fromIntegral (w `shiftR` 8) :: Word8)+ , (fromIntegral (w) :: Word8)]+{-# INLINE packWord64 #-}++unpackWord64 :: B.ByteString -> Integer+unpackWord64 s =+ (fromIntegral (s `B.index` 0) `shiftL` 56) .|.+ (fromIntegral (s `B.index` 1) `shiftL` 48) .|.+ (fromIntegral (s `B.index` 2) `shiftL` 40) .|.+ (fromIntegral (s `B.index` 3) `shiftL` 32) .|.+ (fromIntegral (s `B.index` 4) `shiftL` 24) .|.+ (fromIntegral (s `B.index` 5) `shiftL` 16) .|.+ (fromIntegral (s `B.index` 6) `shiftL` 8) .|.+ (fromIntegral (s `B.index` 7) )+{-# INLINE unpackWord64 #-}++-- Group Queries++byField :: B.ByteString -> DbRecord -> DbValue+byField f (DbRecord _ m) = fromJust $ Map.lookup f m++byFieldMaybe :: B.ByteString -> DbRecord -> Maybe DbValue+byFieldMaybe f (DbRecord _ m) = Map.lookup f m++byTime :: Integer -> DbRecord -> Integer+byTime interval (DbRecord t _) = interval * (t `quot` interval)+++unpackString :: DbValue -> B.ByteString+unpackString (DbString i) = i+unpackString _ = error "Can't unpack Int"++unpackInt :: DbValue -> Integer+unpackInt (DbInt i) = i+unpackInt _ = error "Can't unpack Int"++unpackFloat :: DbValue -> Float+unpackFloat (DbFloat i) = i+unpackFloat _ = error "Can't unpack Float"++unpackDouble :: DbValue -> Double+unpackDouble (DbDouble i) = i+unpackDouble _ = error "Can't unpack Double"
+ src/Continuum/Common/Types.hs view
@@ -0,0 +1,194 @@+{-# LANGUAGE DeriveGeneric #-}++module Continuum.Common.Types where++import Data.ByteString ( ByteString )+import GHC.Generics ( Generic )+import qualified Data.Serialize as S+import qualified Data.Map as Map++-- |+-- | ALIASES+-- |++type DbName = ByteString++type Decoder = (ByteString, ByteString) -> DbErrorMonad DbResult++-- |+-- | INTERNAL DB TYPES+-- |++data DbType =+ DbtInt+ | DbtDouble+ | DbtString+ deriving(Show, Generic, Eq)++instance S.Serialize DbType++-- |+-- | DB Error+-- |++data DbError =+ IndexesDecodeError String+ | FieldDecodeError String ByteString+ | ValuesDecodeError String+ | ValueDecodeError String+ | KeyDecodeError String+ | DecodeFieldByIndexError String [Int]+ | FieldNotFoundError+ | NoSuchDatabaseError+ | NoAggregatorAvailable+ | SchemaDecodingError String+ | OtherError+ deriving (Show, Eq, Ord, Generic)++instance S.Serialize DbError++type DbErrorMonad = Either DbError+++-- |+-- | DB VALUE+-- |++data DbValue =+ EmptyValue+ | DbInt Integer+ | DbFloat Float+ | DbDouble Double+ | DbString ByteString+ | DbTimestamp Integer+ | DbSequenceId Integer+ -- DbList [DbValue]+ -- DbMap [(DbValue, DbValue)]+ deriving (Show, Eq, Ord, Generic)++instance S.Serialize DbValue++data DbRecord =+ DbRecord Integer (Map.Map ByteString DbValue)+ deriving(Generic, Show, Eq)++instance S.Serialize DbRecord++-- | Creates a DbRecord from Timestamp and Key/Value pairs+--+makeRecord :: Integer -> [(ByteString, DbValue)] -> DbRecord+makeRecord timestamp vals = DbRecord timestamp (Map.fromList vals)++-- |+-- | DB RESULT+-- |++data DbResult =+ EmptyRes+ | ErrorRes DbError+ | KeyRes Integer+ | RecordRes DbRecord+ | FieldRes (Integer, DbValue)+ | FieldsRes (Integer, [DbValue])+ | CountStep Integer+ | CountRes Integer+ | GroupRes (Map.Map DbValue DbResult)+ | DbResults [DbResult]+ | DbSchemaResult (DbName, DbSchema)+ deriving(Generic, Show, Eq)++instance S.Serialize DbResult++-- |+-- | RANGE+-- |++-- Maybe someday we'll need a ByteBuffer scan ranges. For now all keys are always+-- integers. Maybe iterators for something like indexes should be done somehow+-- differently not to make that stuff even more complex.+data ScanRange =+ OpenEnd Integer+ | SingleKey Integer+ | KeyRange Integer Integer+ | EntireKeyspace+ deriving(Show, Generic)++instance S.Serialize ScanRange++-- |+-- | AGGREGATES+-- |++data Decoding =+ Field ByteString+ | Fields [ByteString]+ | Key+ | Record+ deriving(Generic, Show)++instance S.Serialize Decoding++-- |+-- | DB SCHEMA+-- |++data DbSchema = DbSchema+ { fieldMappings :: Map.Map ByteString Int+ , fields :: [ByteString]+ , indexMappings :: Map.Map Int ByteString+ , schemaMappings :: Map.Map ByteString DbType+ , schemaTypes :: [DbType]+ } deriving (Generic, Show, Eq)++instance S.Serialize DbSchema++-- | Creates a DbSchema out of Schema Definition (name/type pairs)+--+makeSchema :: [(ByteString, DbType)] -> DbSchema+makeSchema stringTypeList =+ DbSchema { fieldMappings = fMappings+ , fields = fields'+ , schemaMappings = Map.fromList stringTypeList+ , indexMappings = iMappings+ , schemaTypes = schemaTypes'}+ where fields' = fmap fst stringTypeList+ schemaTypes' = fmap snd stringTypeList+ fMappings = Map.fromList $ zip fields' iterateFrom0+ iMappings = Map.fromList $ zip iterateFrom0 fields'+ iterateFrom0 = (iterate (1+) 0)++-- |+-- | QUERIES+-- |++data SelectQuery =+ Count+ -- | Distinct+ -- | Min+ -- | Max+ | Group SelectQuery+ | FetchAll+ deriving (Generic, Show)++instance S.Serialize SelectQuery++-- |+-- | External Protocol Specification+-- |++-- TODO: Split client and server requests++data Node = Node String String+ deriving(Generic, Show, Eq, Ord)++instance S.Serialize Node++data Request =+ Shutdown+ | Insert DbName DbRecord+ | CreateDb DbName DbSchema+ -- TODO: Add ByteString here, never encode it inside of SelectQuery itself+ | Select DbName SelectQuery+ deriving(Generic, Show)++instance S.Serialize Request