riak 1.1.1.0 → 1.1.2.0

raw patch · 15 files changed

+423/−151 lines, 15 filesdep +asyncdep +processdep +template-haskelldep ~aesonPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: async, process, template-haskell, yaml

Dependency ranges changed: aeson

API changes (from Hackage documentation)

Files

• Changes.md +3/−0
• riak.cabal +15/−11
• src/Network/Riak/Connection/Internal.hs +23/−28
• src/Network/Riak/Request.hs +9/−2
• src/Network/Riak/Response.hs +16/−26
• src/Network/Riak/Search.hs +40/−6
• src/Network/Riak/Tag.hs +18/−2
• src/Network/Riak/Types/Internal.hs +26/−13
• tests/CRDTProperties.hs +3/−3
• tests/Common.hs +0/−17
• tests/Internal.hs +29/−0
• tests/Properties.hs +3/−6
• tests/Test.hs +98/−37
• tests/Utils.hs +84/−0
• tests/dsl/Network/Riak/Admin/DSL.hs +56/−0

@@ -1,3 +1,6 @@+* 1.1.2.0+  - Fixes issue where exceptions were not handled properly with many threads (https://github.com/markhibberd/riak-haskell-client/pull/76)+  - Add / delete indexes * 1.1.1.0   - Fixes for 2 connection leaks on errors.   - Bump upper bound on aeson to <1.1

@@ -1,5 +1,5 @@ name:                riak-version:             1.1.1.0+version:             1.1.2.0 synopsis:            A Haskell client for the Riak decentralized data store description:   A Haskell client library for the Riak decentralized data@@ -99,9 +99,10 @@     Network.Riak.Tag    build-depends:-                aeson                         >= 0.8      && < 1.1,+                aeson                         >= 0.8      && < 1.2,+                async                         >= 2.0.0.0  && < 2.2,                 attoparsec                    >= 0.12.1.6 && < 0.14,-                base                          >= 3        && <5,+                base                          >= 3        && < 5,                 binary,                 blaze-builder                 >= 0.3      && <= 0.5,                 bytestring,@@ -142,32 +143,38 @@ test-suite test   type: exitcode-stdio-1.0   main-is: Test.hs-  hs-source-dirs: tests-  ghc-options: -Wall+  hs-source-dirs: tests, tests/dsl+  ghc-options: -Wall -threaded    if flag(test2i)     cpp-options: -DTEST2I    other-modules:-    Properties     CRDTProperties-    Common+    Internal+    Utils+    Network.Riak.Admin.DSL+    Properties    build-depends:     base,     riak,     riak-protobuf,+    aeson,     bytestring,     containers,     HUnit,+    process,     QuickCheck,     tasty,     tasty-hunit,     tasty-quickcheck,+    template-haskell,     text,     mtl                    >= 2.1,     semigroups             >= 0.16,-    data-default-class     >= 0.0.1+    data-default-class     >= 0.0.1,+    yaml                   >= 0.8.19   benchmark bench@@ -180,6 +187,3 @@     criterion   >= 1.1,     bytestring  >= 0.10,     semigroups  >= 0.16---

@@ -1,4 +1,4 @@-{-# LANGUAGE CPP, OverloadedStrings, RecordWildCards, ScopedTypeVariables, FlexibleContexts #-}+{-# LANGUAGE CPP, OverloadedStrings, RecordWildCards, ScopedTypeVariables, FlexibleContexts, MultiWayIf #-} {-# OPTIONS_GHC -fno-warn-orphans #-}  -- |@@ -37,11 +37,9 @@     , recvResponse_     ) where -import Control.Concurrent (forkIO)-import Control.Concurrent.Chan (newChan, readChan, writeChan)-import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)+import Control.Concurrent.Async (async, waitBoth) import Control.Exception (Exception, IOException, throwIO, bracketOnError)-import Control.Monad (forM_, replicateM, replicateM_)+import Control.Monad (forM_, replicateM) import Data.Binary.Put (Put, putWord32be, runPut) import Data.IORef (newIORef, readIORef, writeIORef) import Data.Int (Int64)@@ -209,12 +207,11 @@ getResponse :: Response a => Connection -> Int64 -> a -> T.MessageTag -> IO a getResponse conn len _ expected = do   tag <- recvGet conn getTag-  case undefined of-   _| tag == expected        -> recvGetN conn (len-1) messageGetM-    | tag == T.ErrorResponse -> throwError =<< recvGetN conn (len-1) messageGetM-    | otherwise ->-        moduleError "getResponse" $ "received unexpected response: expected " ++-                                    show expected ++ ", received " ++ show tag+  if | tag == expected        -> recvGetN conn (len-1) messageGetM+     | tag == T.ErrorResponse -> throwError =<< recvGetN conn (len-1) messageGetM+     | otherwise ->+         moduleError "getResponse" $ "received unexpected response: expected " +++                                     show expected ++ ", received " ++ show tag  -- | Send a request to the server, and receive its response. exchange :: Exchange req resp => Connection -> req -> IO resp@@ -277,12 +274,11 @@ recvCorrectTag :: String -> Connection -> T.MessageTag -> Int64 -> a -> IO a recvCorrectTag func conn expected len v = do   tag <- recvGet conn getTag-  case undefined of-   _| tag == expected -> recvExactly conn (len-1) >> return v-    | tag == T.ErrorResponse -> throwError =<< recvGetN conn len messageGetM-    | otherwise -> moduleError func $-                   "received unexpected response: expected " ++-                   show expected ++ ", received " ++ show tag+  if | tag == expected -> recvExactly conn (len-1) >> return v+     | tag == T.ErrorResponse -> throwError =<< recvGetN conn len messageGetM+     | otherwise -> moduleError func $+                    "received unexpected response: expected " +++                    show expected ++ ", received " ++ show tag  debugRecv :: (a -> String) -> IO a -> IO a #ifdef DEBUG@@ -299,16 +295,16 @@         (Connection -> IO resp) -> Connection -> [req] -> IO [resp] pipe _ _ [] = return [] pipe receive conn@Connection{..} reqs = do-  ch <- newChan   let numReqs = length reqs-  _ <- forkIO . replicateM_ numReqs $ writeChan ch =<< receive conn   let tag = show (messageTag (head reqs))   if Debug.level > 1     then forM_ reqs $ \req -> debug "pipe" $ ">>> " ++ showM req     else debug "pipe" $ ">>> " ++ show numReqs ++ "x " ++ tag-  onIOException ("pipe " ++ tag) .-    sendAll connSock . runPut . mapM_ putRequest $ reqs-  replicateM numReqs $ readChan ch+  receiveResps <- async . replicateM numReqs $ receive conn+  sendReqs <- async . sendAll connSock . runPut . mapM_ putRequest $ reqs+  (_, resps) <- onIOException ("pipe " ++ tag) $+    waitBoth sendReqs receiveResps+  return resps  -- | Send a series of requests to the server, back to back, and -- receive a response for each request sent.  The sending and@@ -331,16 +327,15 @@ pipeline_ :: (Request req) => Connection -> [req] -> IO () pipeline_ _ [] = return () pipeline_ conn@Connection{..} reqs = do-  done <- newEmptyMVar-  _ <- forkIO $ do-         forM_ reqs (recvResponse_ conn . expectedResponse)-         putMVar done ()+  receiveResps <- async $+    forM_ reqs (recvResponse_ conn . expectedResponse)   if Debug.level > 1     then forM_ reqs $ \req -> debug "pipe" $ ">>> " ++ showM req     else debug "pipe" $ ">>> " ++ show (length reqs) ++ "x " ++                         show (messageTag (head reqs))-  sendAll connSock . runPut . mapM_ putRequest $ reqs-  takeMVar done+  sendReqs <- async . sendAll connSock . runPut . mapM_ putRequest $ reqs+  _ <- onIOException "pipeline_" $ waitBoth sendReqs receiveResps+  return ()  onIOException :: String -> IO a -> IO a onIOException func act =

@@ -45,8 +45,11 @@     -- * Map/reduce     , MapReduceRequest     , mapReduce+    -- * Search     , search     , getIndex+    , putIndex+    , deleteIndex     ) where  #if __GLASGOW_HASKELL__ < 710@@ -75,6 +78,8 @@ import qualified Network.Riak.Protocol.GetBucketTypeRequest as GetBucketType import qualified Network.Riak.Protocol.SearchQueryRequest as SearchQueryRequest import qualified Network.Riak.Protocol.YzIndexGetRequest as YzIndex+import qualified Network.Riak.Protocol.YzIndexPutRequest as YzIndex+import qualified Network.Riak.Protocol.YzIndexDeleteRequest as YzIndex  -- | Create a ping request. ping :: PingRequest@@ -251,6 +256,8 @@ getIndex :: Maybe Index -> YzIndex.YzIndexGetRequest getIndex = YzIndex.YzIndexGetRequest --+putIndex :: IndexInfo -> Maybe Timeout -> YzIndex.YzIndexPutRequest+putIndex = YzIndex.YzIndexPutRequest +deleteIndex :: Index -> YzIndex.YzIndexDeleteRequest+deleteIndex = YzIndex.YzIndexDeleteRequest

@@ -1,4 +1,4 @@-{-# LANGUAGE RecordWildCards, CPP, OverloadedStrings #-}+{-# LANGUAGE NamedFieldPuns, RecordWildCards, CPP, OverloadedStrings #-}  -- | -- Module:      Network.Riak.Request@@ -39,8 +39,8 @@ import Network.Riak.Protocol.GetResponse import Network.Riak.Protocol.ListBucketsResponse import Network.Riak.Protocol.PutResponse-import Network.Riak.Protocol.SearchQueryResponse-import Network.Riak.Protocol.SearchDoc+import qualified Network.Riak.Protocol.SearchQueryResponse as Q+import qualified Network.Riak.Protocol.SearchDoc as Q import qualified Network.Riak.Protocol.YzIndexGetResponse as Yz import Network.Riak.Types.Internal hiding (MessageTag(..)) import qualified Network.Riak.Protocol.Link as Link@@ -53,7 +53,9 @@ import Data.Maybe (fromMaybe) import Data.Semigroup import Control.Arrow ((&&&))-import Data.Foldable (toList)+import Control.Monad (join)+import Data.Foldable (foldMap, toList)+import Text.Read (readMaybe)  getClientID :: GetClientIDResponse -> ClientID getClientID = client_id@@ -90,28 +92,16 @@   where go l = l { Link.bucket = unescape <$> Link.bucket l                  , Link.key = unescape <$> Link.key l } -search :: SearchQueryResponse -> [SearchResult]-search = map toSearchResult . toList . docs--toSearchResult :: SearchDoc -> SearchResult-toSearchResult r = SearchResult {-                     bucketType = field "_yz_rt",-                     bucket     = field "_yz_rb",-                     key        = field "_yz_rk",-                     score      = fmap (read . LC.unpack) =<< M.lookup "score" info-                   }-    where-      info :: M.Map L.ByteString (Maybe L.ByteString)-      info = M.fromList . map (Pair.key &&& Pair.value) . toList . fields $ r--      field :: L.ByteString -> L.ByteString-      field name-          = maybe (unexpected $ "field " <> show name <> " has empty value") id-            . maybe (unexpected $ "no " <> show name <> " field") id-            . M.lookup name $ info--      unexpected = unexError "Network.Riak.Response" "search"-+search :: Q.SearchQueryResponse -> SearchResult+search resp =+  SearchResult+    { docs     = fmap (fmap unpair . Q.fields) (Q.docs resp)+    , maxScore = Q.max_score resp+    , numFound = Q.num_found resp+    }+  where+    unpair :: Pair.Pair -> (L.ByteString, Maybe L.ByteString)+    unpair Pair.Pair{Pair.key, Pair.value} = (key, value)  getIndex :: Yz.YzIndexGetResponse -> [IndexInfo] getIndex = toList . Yz.index

@@ -7,23 +7,57 @@ -- -- http://docs.basho.com/riak/2.1.3/dev/using/search/ {-# LANGUAGE CPP #-}-module Network.Riak.Search where+module Network.Riak.Search+  ( IndexInfo+  , SearchResult(..)+  , Score+  , indexInfo+  , getIndex+  , putIndex+  , deleteIndex+  , searchRaw+  ) where  #if __GLASGOW_HASKELL__ <= 708 import           Control.Applicative #endif+import           Data.Sequence (Seq) import           Network.Riak.Connection.Internal+import           Network.Riak.Protocol.Content (Content)+import           Network.Riak.Protocol.YzIndex (YzIndex(YzIndex)) import qualified Network.Riak.Request as Req import qualified Network.Riak.Response as Resp import           Network.Riak.Types.Internal --- | Get an index info for @Just index@, or get all indexes for--- @Nothing@.+-- | 'IndexInfo' smart constructor.+--+-- If 'Nothing', @schema@ defaults to @"_yz_default"@.+--+-- If 'Nothing', @n@ defaults to the default @n@ value for buckets that have not+-- explicitly set the property. In the default installation of @riak@, this is+-- 3 (see https://github.com/basho/riak_core/blob/develop/priv/riak_core.schema).+indexInfo :: Index -> Maybe Schema -> Maybe N -> IndexInfo+indexInfo = YzIndex++-- | Get an index info for @Just index@, or get all indexes for @Nothing@.+--+-- https://docs.basho.com/riak/kv/2.1.4/developing/api/protocol-buffers/yz-index-get/ getIndex :: Connection -> Maybe Index -> IO [IndexInfo] getIndex conn ix = Resp.getIndex <$> exchange conn (Req.getIndex ix) +-- | Create a new index or modify an existing index.+--+-- https://docs.basho.com/riak/kv/2.1.4/developing/api/protocol-buffers/yz-index-put/+putIndex :: Connection -> IndexInfo -> Maybe Timeout -> IO (Seq Content, VClock)+putIndex conn info timeout = Resp.put <$> exchange conn (Req.putIndex info timeout) --- | Search by raw 'SearchQuery' request (a bytestring) using an--- index.-searchRaw :: Connection -> SearchQuery -> Index -> IO [SearchResult]+-- | Delete an index.+--+-- https://docs.basho.com/riak/kv/2.1.4/developing/api/protocol-buffers/yz-index-delete/+deleteIndex :: Connection -> Index -> IO ()+deleteIndex conn ix = exchange_ conn (Req.deleteIndex ix)++-- | Search by raw 'SearchQuery' request (a 'Data.ByteString.Lazy.Bytestring')+-- using an 'Index'.+searchRaw :: Connection -> SearchQuery -> Index -> IO SearchResult searchRaw conn q ix = Resp.search <$> exchange conn (Req.search q ix)

@@ -57,6 +57,8 @@ import Network.Riak.Protocol.SearchQueryResponse import Network.Riak.Protocol.YzIndexGetRequest import Network.Riak.Protocol.YzIndexGetResponse+import Network.Riak.Protocol.YzIndexPutRequest+import Network.Riak.Protocol.YzIndexDeleteRequest import Network.Riak.Types.Internal as Types import Text.ProtocolBuffers.Get (Get, getWord8) @@ -310,6 +312,20 @@  instance Exchange YzIndexGetRequest YzIndexGetResponse +instance Request YzIndexPutRequest where+  expectedResponse _ = Types.YokozunaIndexPutRequest++instance Tagged YzIndexPutRequest where+  messageTag _ = Types.YokozunaIndexPutRequest++instance Exchange YzIndexPutRequest PutResponse++instance Tagged YzIndexDeleteRequest where+    messageTag _ = Types.YokozunaIndexDeleteRequest++instance Request YzIndexDeleteRequest where+    expectedResponse _ = Types.DeleteResponse+ putTag :: MessageTag -> Put putTag m = putWord8 $ message2code HM.! m {-# INLINE putTag #-}@@ -384,8 +400,8 @@  -- (53,CounterGetResp),  (54, Types.YokozunaIndexGetRequest),  (55, Types.YokozunaIndexGetResponse),- -- (56,YokozunaIndexPutReq),- -- (57,YokozunaIndexDeleteReq),+ (56, Types.YokozunaIndexPutRequest),+ (57, Types.YokozunaIndexDeleteRequest),  -- (58,YokozunaSchemaGetReq),  -- (59,YokozunaSchemaGetResp),  -- (60,YokozunaSchemaPutReq),

@@ -1,5 +1,5 @@-{-# LANGUAGE DeriveDataTypeable, FunctionalDependencies, MultiParamTypeClasses,-    RecordWildCards, DeriveGeneric #-}+{-# LANGUAGE BangPatterns, DeriveDataTypeable, FunctionalDependencies,+    MultiParamTypeClasses, RecordWildCards, DeriveGeneric #-}  -- | -- Module:      Network.Riak.Types.Internal@@ -28,14 +28,18 @@     , BucketType     , Key     , Index+    , Schema     , IndexQuery(..)     , IndexValue(..)     , Tag     , SearchQuery     , SearchResult(..)+    , Score     , IndexInfo     , VClock(..)     , Job(..)+    , N+    , Timeout     -- * Quorum management     , Quorum(..)     , DW@@ -57,6 +61,8 @@ import           Data.Digest.Pure.MD5 (md5) import           Data.Hashable (Hashable) import           Data.IORef (IORef)+import           Data.Map (Map)+import           Data.Sequence (Seq) import           Data.Typeable (Typeable) import           Data.Word (Word32) import           GHC.Generics (Generic)@@ -148,6 +154,9 @@ -- | Name of a secondary index type Index = ByteString +-- | Name of an index schema+type Schema = ByteString+ -- | Index query. Can be exact or range, int or bin. Index name should -- not contain the "_bin" or "_int" part, since it's determined from -- data constructor.@@ -180,13 +189,20 @@ -- | Search index info type IndexInfo = YzIndex.YzIndex +-- | N value+--+-- http://docs.basho.com/riak/kv/2.1.4/learn/concepts/replication/+type N = Word32++-- | Timeout in milliseconds+type Timeout = Word32+ -- | Solr search result-data SearchResult = SearchResult {-      bucketType :: BucketType, -- ^ bucket type-      bucket :: Bucket,         -- ^ bucket-      key :: Key,               -- ^ key-      score :: Maybe Score      -- ^ score, if provided-    } deriving (Eq,Show)+data SearchResult = SearchResult+  { docs     :: !(Seq (Seq (ByteString, Maybe ByteString)))+  , maxScore :: !(Maybe Float)+  , numFound :: !(Maybe Word32)+  } deriving (Eq, Ord, Show)  -- | List of (known to us) inbound or outbound message identifiers. data MessageTag = ErrorResponse@@ -225,6 +241,8 @@                 | SearchQueryResponse                 | YokozunaIndexGetRequest                 | YokozunaIndexGetResponse+                | YokozunaIndexPutRequest+                | YokozunaIndexDeleteRequest                   deriving (Eq, Show, Generic)  instance Hashable MessageTag@@ -246,11 +264,6 @@  class (Request req, Response resp) => Exchange req resp     | req -> resp--instance (Tagged a, Tagged b) => Tagged (Either a b) where-    messageTag (Left l)  = messageTag l-    messageTag (Right r) = messageTag r-    {-# INLINE messageTag #-}  -- | A wrapper that keeps Riak vector clocks opaque. newtype VClock = VClock {

@@ -30,7 +30,7 @@ import Test.QuickCheck import Test.QuickCheck.Monadic -import Common+import Utils import Test.Tasty import Test.Tasty.QuickCheck @@ -74,7 +74,7 @@ observeRiak p = Map.fromList . catMaybes <$> observeRiak' (BucketType $ bucketType p)  observeRiak' :: BucketType -> IO [Maybe (Point, C.DataType)]-observeRiak' bt@(BucketType t_) = withSomeConnection $ \c ->+observeRiak' bt@(BucketType t_) = withGlobalConn $ \c ->        sequence [ do r <- C.get c t_ b_ k_                      pure . fmap (p,) $ r                   | b <- values, k <- values,@@ -270,7 +270,7 @@  doRiak :: Action a op =>           Proxy a -> [Op a op] -> IO [Maybe C.DataType]-doRiak p ops = withSomeConnection $ \conn -> do+doRiak p ops = withGlobalConn $ \conn -> do                    --print ops                    (_,_,r) <- runRWST (riak p ops) () conn                    pure r

@@ -1,17 +0,0 @@-module Common (withSomeConnection) where--import qualified Network.Riak.Basic as B-import Network.Riak.Connection.Pool (Pool, create, withConnection)-import Network.Riak.Connection (defaultClient)-import System.IO.Unsafe (unsafePerformIO)---pool :: Pool-{-# NOINLINE pool #-}-pool = unsafePerformIO $-       create defaultClient 1 1 1---- | run action in some riak connection-withSomeConnection :: (B.Connection -> IO a) -> IO a-withSomeConnection = withConnection pool-

@@ -0,0 +1,29 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE TemplateHaskell #-}++-- | GHC stage restriction: have to write Lift in a separate module from where+-- TH is used (Utils.hs)++module Internal where++import Control.Applicative+import Data.Aeson (FromJSON(parseJSON), genericParseJSON)+import Data.Aeson.Types (defaultOptions, fieldLabelModifier)+import Data.Char (toLower)+import GHC.Generics (Generic)+import Language.Haskell.TH.Syntax (Lift(lift))++data Config = Config+  { configHost      :: String+  , configHttpPort  :: Int+  , configProtoPort :: Int+  , configAdmin     :: String+  } deriving Generic++instance Lift Config where+  lift (Config a b c d) = [| Config a b c d |]++instance FromJSON Config where+  parseJSON = genericParseJSON (defaultOptions { fieldLabelModifier = f })+   where+    f ('c':'o':'n':'f':'i':'g':x:xs) = toLower x : xs

@@ -16,7 +16,7 @@ import           Test.QuickCheck.Monadic      (assert, monadicIO, run) import           Test.Tasty import           Test.Tasty.QuickCheck-import           Common+import           Utils  instance Arbitrary L.ByteString where     arbitrary     = L.pack `fmap` arbitrary@@ -36,7 +36,7 @@ t_put_get (QCBucket b) (QCKey k) v =     monadicIO $ assert . uncurry (==) =<< run act   where-    act = withSomeConnection $ \c -> do+    act = withGlobalConn $ \c -> do             p <- Just <$> B.put c btype b k Nothing (binary v) Default Default             r <- B.get c btype b k Default             return (p,r)@@ -48,7 +48,7 @@         r <- run act         assert $ isNothing r     where-      act = withSomeConnection $ \c -> do+      act = withGlobalConn $ \c -> do               _ <- B.put c bt b k Nothing (binary v) Default Default               B.delete c bt b k Default               B.get c bt b k Default@@ -59,6 +59,3 @@  testProperty "t_put_get" t_put_get,  testProperty "put_delete_get" put_delete_get  ]---

@@ -1,6 +1,7 @@-{-# LANGUAGE CPP               #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE ParallelListComp  #-}+{-# LANGUAGE CPP                 #-}+{-# LANGUAGE LambdaCase          #-}+{-# LANGUAGE OverloadedStrings   #-}+{-# LANGUAGE ParallelListComp    #-} {-# LANGUAGE ScopedTypeVariables #-}  module Main where@@ -16,29 +17,69 @@ import           Data.Foldable (toList) import           Data.Semigroup import           Data.Text (Text)+import           Control.Applicative import           Control.Concurrent (threadDelay) import           Control.Exception import qualified Network.Riak as Riak+import           Network.Riak.Admin.DSL import qualified Network.Riak.Basic as B import qualified Network.Riak.Content as B (binary,Content,value)+import           Network.Riak.Connection (exchange) import qualified Network.Riak.CRDT as C import qualified Network.Riak.CRDT.Riak as C+import qualified Network.Riak.Request as Req+import qualified Network.Riak.Response as Resp import qualified Network.Riak.Search as S-import qualified Network.Riak.Cluster as Riak import qualified Network.Riak.JSON as J import           Network.Riak.Resolvable (ResolvableMonoid (..))-import           Network.Riak.Types hiding (key)+import           Network.Riak.Types import qualified Network.Riak.Protocol.ErrorResponse as ER+import qualified Network.Riak.Protocol.SearchQueryRequest as S import qualified Properties import qualified CRDTProperties as CRDT-import           Common+import           Utils import           Test.Tasty import           Test.Tasty.HUnit  main :: IO ()-main = defaultMain tests+main = do+  setup+  defaultMain tests +setup :: IO ()+setup = do+  -- Create a "set-ix" index and wait for it to exist.+  let createIxUrl :: String+      createIxUrl = globalHost ++ ":" ++ show globalHttpPort ++ "/search/index/set-ix" +  shell ("curl -sf -XPUT " ++ createIxUrl ++ " -H 'Content-Type: application/json'")+  let loop =+        try (shell ("curl -sf " ++ createIxUrl)) >>= \case+          Left (_ :: ShellFailure) -> do+            threadDelay (1*1000*1000)+            loop+          Right _ -> pure ()+  loop++  riakAdminWith globalAdmin+    [ waitForService "riak_kv" Nothing+    , waitForService "yokozuna" Nothing++    , bucketTypeCreate "maps" (Just "'{\"props\":{\"datatype\":\"map\"}}'")+    , bucketTypeCreate "sets" (Just "'{\"props\":{\"datatype\":\"set\",\"search_index\":\"set-ix\"}}'")+    , bucketTypeCreate "counters" (Just "'{\"props\":{\"datatype\":\"counter\"}}'")++    , bucketTypeActivate "maps"+    , bucketTypeActivate "sets"+    , bucketTypeActivate "counters"++    , bucketTypeCreate "untyped-1" Nothing+    , bucketTypeCreate "untyped-2" Nothing++    , bucketTypeActivate "untyped-1"+    , bucketTypeActivate "untyped-2"+    ]+ tests :: TestTree tests = testGroup "Tests" [properties,                            integrationalTests,@@ -67,23 +108,23 @@  searches :: TestTree searches = testGroup "Search" [-            search,-            getIndex+            search1,+            search2,+            getIndex,+            putIndex,+            deleteIndex            ]  testClusterSimple :: TestTree-testClusterSimple = testCase "testClusterSimple" $ do-    rc <- Riak.connectToCluster [Riak.defaultClient]-    Riak.inCluster rc B.ping+testClusterSimple = testCase "testClusterSimple" $ withGlobalConn B.ping   testIndexedPutGet :: TestTree testIndexedPutGet = testCase "testIndexedPutGet" $ do-    rc <- Riak.connectToCluster [Riak.defaultClient]     let bt = Nothing         b = "riak-haskell-client-test"         k = "test"-    keys <- Riak.inCluster rc $ \c -> do+    keys <- withGlobalConn $ \c -> do       _ <- J.putIndexed c bt b k [(IndexInt "someindex" 135)] Nothing           (RM (M.fromList [("somekey", "someval")] :: M.Map Text Text))           Default Default@@ -92,14 +133,11 @@  ping'o'death :: TestTree ping'o'death = testCase "ping'o'death" $ replicateM_ 23 ping-    where ping = do-            c <- Riak.connect Riak.defaultClient-            replicateM_ 1024 $ Riak.ping c+    where ping = withGlobalConn (\c -> replicateM_ 1024 (Riak.ping c))   counter :: TestTree-counter = testCase "increment" $ do-              conn <- Riak.connect Riak.defaultClient+counter = testCase "increment" $ withGlobalConn $ \conn -> do               Just (C.DTCounter (C.Counter a)) <- act conn               Just (C.DTCounter (C.Counter b)) <- act conn               assertEqual "inc by 1" 1 (b-a)@@ -108,8 +146,7 @@                  C.get c "counters" "xxx" "yyy"  set :: TestTree-set = testCase "set add" $ do-        conn <- Riak.connect Riak.defaultClient+set = testCase "set add" $ withGlobalConn $ \conn -> do         C.setSendUpdate conn btype buck key [C.SetRemove val]         C.setSendUpdate conn btype buck key [C.SetAdd val]         Just (C.DTSet (C.Set r)) <- C.get conn btype buck key@@ -118,8 +155,7 @@       (btype,buck,key,val) = ("sets","xxx","yyy","foo")  map_ :: TestTree-map_ = testCase "map update" $ do-         conn <- Riak.connect Riak.defaultClient+map_ = testCase "map update" $ withGlobalConn $ \conn -> do          Just (C.DTMap a) <- act conn -- do smth (increment), get          Just (C.DTMap b) <- act conn -- increment, get          assertEqual "map update" (C.modify mapOp a) b -- modify's behaviour should match@@ -159,37 +195,62 @@                            (C.MapCounterOp (C.CounterInc 1))  -search :: TestTree-search = testCase "basic searchRaw" $ do-           conn <- Riak.connect Riak.defaultClient+search1 :: TestTree+search1 = testCase "basic searchRaw" $ withGlobalConn $ \conn -> do            C.sendModify conn btype buck key [C.SetRemove kw]            delay            a <- query conn ("set:" <> kw)-           assertEqual "should not found non-existing" [] a+           assertEqual "should not found non-existing" (S.SearchResult Seq.empty (Just 0.0) (Just 0)) a            C.sendModify conn btype buck key [C.SetAdd kw]            delay            b <- query conn ("set:" <> kw)-           assertBool "searches specific" $ not (null b)+           assertBool "searches specific" $ not (Seq.null (S.docs b))            c <- query conn ("set:*")-           assertBool "searches *" $ not (null c)+           assertBool "searches *" $ not (Seq.null (S.docs c))     where       query conn q = S.searchRaw conn q "set-ix"       (btype,buck,key) = ("sets","xxx","yyy")       kw = "haskell"       delay = threadDelay (1*5000*1000) -- http://docs.basho.com/riak/2.1.3/dev/using/search/#Indexing-Values +search2 :: TestTree+search2 = testCase "search with fl" $ withGlobalConn $ \conn -> do+            let req = (Req.search "set:haskell" "set-ix") { S.fl = Seq.singleton "_yz_rk" }+            resp <- Resp.search <$> exchange conn req+            assertEqual "only returns fl"+              (Seq.singleton (Seq.singleton ("_yz_rk", Just "yyy")))+              (S.docs resp) + getIndex :: TestTree-getIndex = testCase "getIndex" $ do-             conn <- Riak.connect Riak.defaultClient+getIndex = testCase "getIndex" $ withGlobalConn $ \conn -> do              all' <- S.getIndex conn Nothing              one <- S.getIndex conn (Just "set-ix")              assertBool "all indeces" $ not (null all')              assertEqual "set index" 1 (length one) +putIndex :: TestTree+putIndex = testCase "putIndex" $ withGlobalConn $ \conn -> do+             _ <- S.putIndex conn (S.indexInfo "dummy-index" Nothing Nothing) Nothing+             threadDelay 5000000+             one <- S.getIndex conn (Just "dummy-index")+             assertEqual "index was created" 1 (length one)++deleteIndex :: TestTree+deleteIndex = testCase "deleteIndex" $ withGlobalConn $ \conn -> do+  S.deleteIndex conn "dummy-index"+  threadDelay (5*1000*1000)++  _ <- tryJust f (S.getIndex conn (Just "dummy-index"))+  pure ()++  where+    f :: ER.ErrorResponse -> Maybe ()+    f (ER.ErrorResponse "notfound" 0) = Just ()+    f _ = Nothing+ bucketTypes :: TestTree-bucketTypes = testCase "bucketTypes" $ do-             conn <- Riak.connect Riak.defaultClient+bucketTypes = testCase "bucketTypes" $ withGlobalConn $ \conn -> do              [p0,p1,p2] <- sequence [ B.put conn bt b k Nothing o Default Default                                       | bt <- types | o <- [o0,o1,o2] ]              [r0,r1,r2] <- sequence [ B.get conn bt b k Default | bt <- types ]@@ -213,10 +274,10 @@  exceptions :: TestTree exceptions = testGroup "exceptions" [-              testCase "correct put"  . shouldBeOK . withSomeConnection $ put,-              testCase "correct put_" . shouldBeOK . withSomeConnection $ put_,-              testCase "invalid put"  . shouldThrow . withSomeConnection $ putErr,-              testCase "invalid put_" . shouldThrow . withSomeConnection $ put_Err+              testCase "correct put"  . shouldBeOK . withGlobalConn $ put,+              testCase "correct put_" . shouldBeOK . withGlobalConn $ put_,+              testCase "invalid put"  . shouldThrow . withGlobalConn $ putErr,+              testCase "invalid put_" . shouldThrow . withGlobalConn $ put_Err              ]     where       put     = putSome B.put  btype

@@ -0,0 +1,84 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE LambdaCase         #-}+{-# LANGUAGE TemplateHaskell    #-}++module Utils+  ( globalAdmin+  , globalHost+  , globalHttpPort+  , shell+  , withGlobalConn+  , ShellFailure(..)+  ) where++import Control.Applicative+import Control.Exception+import Control.Monad+import Data.Typeable+import Data.Yaml.TH (decodeFile)+import Internal+import System.Exit+import System.IO.Unsafe (unsafePerformIO)+import System.Timeout++import Control.Applicative+import qualified Network.Riak as Riak+import qualified Network.Riak.Basic as B+import Network.Riak.Connection.Pool (Pool, create, withConnection)+import Network.Riak.Connection (defaultClient)++import qualified System.Process as Process+++config :: Config+config = $$(decodeFile "tests/test.yaml")++-- | The global riak-admin string, configured in test.yaml.+globalAdmin :: String+globalAdmin = configAdmin config++globalHost :: String+globalHost = configHost config++globalHttpPort :: Int+globalHttpPort = configHttpPort config++-- | Run action in some Riak connection+withGlobalConn :: (B.Connection -> IO a) -> IO a+withGlobalConn = withConnection pool++-- | The global riak pool that all tests share.+pool :: Pool+pool = unsafePerformIO (create client 1 1 1)+ where+  client = Riak.defaultClient+    { Riak.host = globalHost+    , Riak.port = show (configProtoPort config)+    }+{-# NOINLINE pool #-}++data ShellFailure+  = ShellFailure Int String+  | ShellTimeout String+  deriving (Show, Typeable)++instance Exception ShellFailure++-- | Run a shell command (inheriting stdin, stdout, and stderr), and throw an+-- exception if it fails. Time out after 30 seconds.+shell :: String -> IO ()+shell s =+  timeout (30*1000*1000) act >>= \case+    Nothing -> throw (ShellTimeout s)+    _       -> pure ()+  where+    act :: IO ()+    act =+      bracketOnError+        (do+          (_, _, _, h) <- Process.createProcess (Process.shell s)+          pure h)+        Process.terminateProcess+        (Process.waitForProcess >=> \case+          ExitSuccess -> pure ()+          ExitFailure n -> throw (ShellFailure n s))

@@ -0,0 +1,56 @@+-- | A simple "riak-admin DSL", which is nothing more than strings of stitched+-- together riak-admin shell commands.++module Network.Riak.Admin.DSL+  ( -- * Riak admin commands+    waitForService+  , bucketTypeCreate+  , bucketTypeActivate+    -- * Riak admin runners+  , riakAdmin+  , riakAdminWith+  ) where++import Utils+++-- | Bucket props string, e.g. "{\"props\":{\"n_val\":\"1\"}}"+type BucketProps = String++type BucketTypeName = String++-- | Erlang node name, e.g. "riak@172.17.0.2"+type NodeName = String++-- | riak-admin shell command+type RiakAdmin = String++-- | Riak service name, e.g. "yokozuna"+type ServiceName = String+++-- | riak-admin wait-for-service+waitForService :: ServiceName -> Maybe NodeName -> RiakAdmin+waitForService name node =+  "riak-admin wait-for-service " ++ name ++ maybe "" (" " ++) node++-- | riak-admin bucket-type create+bucketTypeCreate :: BucketTypeName -> Maybe BucketProps -> RiakAdmin+bucketTypeCreate name props =+  "riak-admin bucket-type create " ++ name ++ maybe "" (" " ++) props++-- | riak-admin bucket-type activate+bucketTypeActivate :: BucketTypeName -> RiakAdmin+bucketTypeActivate name = "riak-admin bucket-type activate " ++ name+++-- | Run a list of riak-admin commands locally, and throw an exception if any of+-- them fail.+riakAdmin :: [RiakAdmin] -> IO ()+riakAdmin = riakAdminWith ""++-- | Like 'riakAdmin', but prefix each command with the given 'String' (plus a+-- space).+riakAdminWith :: String -> [RiakAdmin] -> IO ()+riakAdminWith ""     = mapM_ shell+riakAdminWith prefix = mapM_ (\s -> shell (prefix ++ " " ++ s))