diff --git a/Network/Kafka.hs b/Network/Kafka.hs --- a/Network/Kafka.hs +++ b/Network/Kafka.hs @@ -5,13 +5,16 @@ module Network.Kafka where import Control.Applicative -import Control.Exception (bracket) +import Control.Exception (IOException) +import Control.Exception.Lifted (catch) import Control.Lens import Control.Monad (liftM) +import Control.Monad.Except (ExceptT(..), runExceptT, withExceptT, MonadError(..)) import Control.Monad.Trans (liftIO, lift) -import Control.Monad.Trans.Either import Control.Monad.Trans.State import Data.ByteString.Char8 (ByteString) +import Data.List.NonEmpty (NonEmpty(..)) +import qualified Data.List.NonEmpty as NE import Data.Monoid ((<>)) import qualified Data.Pool as Pool import Data.Serialize.Get @@ -19,9 +22,12 @@ import qualified Data.ByteString.Char8 as B import qualified Data.Map as M import qualified Network +import Prelude import Network.Kafka.Protocol +type KafkaAddress = (Host, Port) + data KafkaState = KafkaState { -- | Name to use as a client ID. _stateName :: KafkaString -- | How many acknowledgements are required for producing. @@ -39,23 +45,18 @@ -- | Broker cache , _stateBrokers :: M.Map Leader Broker -- | Connection cache - , _stateConnections :: M.Map Broker (Pool.Pool Handle) + , _stateConnections :: M.Map KafkaAddress (Pool.Pool Handle) -- | Topic metadata cache , _stateTopicMetadata :: M.Map TopicName TopicMetadata - } + -- | Address cache + , _stateAddresses :: NonEmpty KafkaAddress + } deriving (Show) makeLenses ''KafkaState -data KafkaClient = KafkaClient { _kafkaClientState :: KafkaState - , _kafkaClientHandle :: Handle - } - -makeLenses ''KafkaClient - -- | The core Kafka monad. -type Kafka = StateT KafkaClient (EitherT KafkaClientError IO) +type Kafka = StateT KafkaState (ExceptT KafkaClientError IO) -type KafkaAddress = (Host, Port) type KafkaClientId = KafkaString -- | Errors given from the Kafka monad. @@ -68,6 +69,7 @@ -- | Could not find a cached broker for the found leader. | KafkaInvalidBroker Leader | KafkaFailedToFetchMetadata + | KafkaIOException IOException deriving (Eq, Show) -- | Type of response to expect, used for 'KafkaExpected' error. @@ -136,8 +138,8 @@ defaultMaxWaitTime = 0 -- | Create a consumer using default values. -defaultState :: KafkaClientId -> KafkaState -defaultState cid = +mkKafkaState :: KafkaClientId -> KafkaAddress -> KafkaState +mkKafkaState cid addy = KafkaState cid defaultRequiredAcks defaultRequestTimeout @@ -148,50 +150,89 @@ M.empty M.empty M.empty + (addy :| []) --- | Run the underlying Kafka monad at the given leader address and initial state. -runKafka :: KafkaAddress -> KafkaState -> Kafka a -> IO (Either KafkaClientError a) -runKafka (h, p) s k = - bracket (Network.connectTo (h ^. hostString) (p ^. portId)) hClose $ runEitherT . evalStateT k . KafkaClient s +addKafkaAddress :: KafkaAddress -> KafkaState -> KafkaState +addKafkaAddress = over stateAddresses . NE.nub .: cons + where infixr 9 .: + (.:) :: (c -> d) -> (a -> b -> c) -> a -> b -> d + (.:) = (.).(.) +-- | Run the underlying Kafka monad. +runKafka :: KafkaState -> Kafka a -> IO (Either KafkaClientError a) +runKafka s k = runExceptT $ evalStateT k s + -- | Make a request, incrementing the `_stateCorrelationId`. makeRequest :: RequestMessage -> Kafka Request makeRequest m = do - corid <- use (kafkaClientState . stateCorrelationId) - kafkaClientState . stateCorrelationId += 1 - conid <- use (kafkaClientState . stateName) + corid <- use stateCorrelationId + stateCorrelationId += 1 + conid <- use stateName return $ Request (corid, ClientId conid, m) --- | Perform a request and deserialize the response. -doRequest :: Request -> Kafka Response -doRequest r = do - h <- use kafkaClientHandle - doRequest' h r +-- | Catch 'IOException's and wrap them in 'KafkaIOException's. +tryKafkaIO :: IO a -> Kafka a +tryKafkaIO = tryKafka . liftIO -doRequest' :: Handle -> Request -> Kafka Response -doRequest' h r = mapStateT (bimapEitherT KafkaDeserializationError id) $ do - dataLength <- lift . EitherT $ do +-- | Catch 'IOException's and wrap them in 'KafkaIOException's. +tryKafka :: Kafka a -> Kafka a +tryKafka = (`catch` \e -> lift . throwError $ KafkaIOException (e :: IOException)) + +doRequest :: Handle -> Request -> Kafka Response +doRequest h r = do + rawLength <- tryKafkaIO $ do B.hPut h $ requestBytes r hFlush h - rawLength <- B.hGet h 4 - return $ runGet (liftM fromIntegral getWord32be) rawLength - resp <- liftIO $ B.hGet h dataLength - lift . hoistEither $ runGet (getResponse dataLength) resp + B.hGet h 4 + dataLength <- runGetKafka (liftM fromIntegral getWord32be) rawLength + resp <- tryKafkaIO $ B.hGet h dataLength + runGetKafka (getResponse dataLength) resp +runGetKafka :: Get a -> ByteString -> Kafka a +runGetKafka g bs = lift $ withExceptT KafkaDeserializationError $ ExceptT $ return $ runGet g bs + -- | Send a metadata request metadata :: MetadataRequest -> Kafka MetadataResponse -metadata request = do - h <- use kafkaClientHandle - metadata' h request +metadata request = withAnyHandle $ flip metadata' request -- | Send a metadata request metadata' :: Handle -> MetadataRequest -> Kafka MetadataResponse -metadata' handle request = - makeRequest (MetadataRequest request) >>= doRequest' handle >>= expectResponse ExpectedMetadata _MetadataResponse +metadata' h request = + makeRequest (MetadataRequest request) >>= doRequest h >>= expectResponse ExpectedMetadata _MetadataResponse +getTopicPartitionLeader :: TopicName -> Partition -> Kafka Broker +getTopicPartitionLeader t p = do + let s = stateTopicMetadata . at t + tmd <- findMetadataOrElse [t] s KafkaFailedToFetchMetadata + leader <- expect KafkaFailedToFetchMetadata (firstOf $ findPartitionMetadata t . (folded . findPartition p) . partitionMetadataLeader) tmd + use stateBrokers >>= expect (KafkaInvalidBroker leader) (view $ at leader) + +expect :: KafkaClientError -> (a -> Maybe b) -> a -> Kafka b +expect e f = lift . maybe (throwError e) return . f + +-- | Find a leader and partition for the topic. +brokerPartitionInfo :: TopicName -> Kafka [PartitionAndLeader] +brokerPartitionInfo t = do + let s = stateTopicMetadata . at t + tmd <- findMetadataOrElse [t] s KafkaFailedToFetchMetadata + return $ pal <$> tmd ^. partitionsMetadata + where pal d = PartitionAndLeader t (d ^. partitionId) (d ^. partitionMetadataLeader) + +findMetadataOrElse :: [TopicName] -> Getting (Maybe a) KafkaState (Maybe a) -> KafkaClientError -> Kafka a +findMetadataOrElse ts s err = do + maybeFound <- use s + case maybeFound of + Just x -> return x + Nothing -> do + updateMetadatas ts + maybeFound' <- use s + case maybeFound' of + Just x -> return x + Nothing -> lift $ throwError err + -- | Function to give an error when the response seems wrong. expectResponse :: KafkaExpectedResponse -> Getting (Leftmost b) ResponseMessage b -> Response -> Kafka b -expectResponse e p = lift . maybe (left $ KafkaExpected e) return . firstOf (responseMessage . p) +expectResponse e p = expect (KafkaExpected e) (firstOf $ responseMessage . p) -- | Convert an abstract time to a serializable protocol value. protocolTime :: KafkaTime -> Time @@ -208,15 +249,20 @@ -- | Construct a fetch request from the values in the state. fetchRequest :: Offset -> Partition -> TopicName -> Kafka FetchRequest fetchRequest o p topic = do - wt <- use (kafkaClientState . stateWaitTime) - ws <- use (kafkaClientState . stateWaitSize) - bs <- use (kafkaClientState . stateBufferSize) + wt <- use stateWaitTime + ws <- use stateWaitSize + bs <- use stateBufferSize return $ FetchReq (ordinaryConsumerId, wt, ws, [(topic, [(p, o, bs)])]) -- | Execute a fetch request and get the raw fetch response. +fetch' :: Handle -> FetchRequest -> Kafka FetchResponse +fetch' h request = + makeRequest (FetchRequest request) >>= doRequest h >>= expectResponse ExpectedFetch _FetchResponse + +-- | Execute a fetch request and get the raw fetch response. Round-robins the +-- requests to addresses in the 'KafkaState'. fetch :: FetchRequest -> Kafka FetchResponse -fetch request = - makeRequest (FetchRequest request) >>= doRequest >>= expectResponse ExpectedFetch _FetchResponse +fetch request = withAnyHandle $ flip fetch' request -- | Extract out messages with their topics from a fetch response. fetchMessages :: FetchResponse -> [TopicAndMessage] @@ -227,8 +273,10 @@ updateMetadatas ts = do md <- metadata $ MetadataReq ts let (brokers, tmds) = (md ^.. metadataResponseBrokers . folded, md ^.. topicsMetadata . folded) - kafkaClientState . stateBrokers %= \m -> foldr addBroker m brokers - kafkaClientState . stateTopicMetadata %= \m -> foldr addTopicMetadata m tmds + addresses = map broker2address brokers + stateAddresses %= NE.nub . NE.fromList . (++ addresses) . NE.toList + stateBrokers %= \m -> foldr addBroker m brokers + stateTopicMetadata %= \m -> foldr addTopicMetadata m tmds return () where addBroker :: Broker -> M.Map Leader Broker -> M.Map Leader Broker addBroker b = M.insert (Leader . Just $ b ^. brokerNode . nodeId) b @@ -241,26 +289,63 @@ updateAllMetadata :: Kafka () updateAllMetadata = updateMetadatas [] --- | Execute a handler action, creating a new Pool and updating the connections Map if needed. +-- | Execute a Kafka action with a 'Handle' for the given 'Broker', updating +-- the connections cache if needed. +-- +-- When the action throws an 'IOException', it is caught and returned as a +-- 'KafkaIOException' in the Kafka monad. +-- +-- Note that when the given action throws an exception, any state changes will +-- be discarded. This includes both 'IOException's and exceptions thrown by +-- 'throwError' from 'Control.Monad.Except'. withBrokerHandle :: Broker -> (Handle -> Kafka a) -> Kafka a -withBrokerHandle broker f = do - conns <- use (kafkaClientState . stateConnections) - let foundPool = conns ^. at broker +withBrokerHandle broker = withAddressHandle (broker2address broker) + +-- | Execute a Kafka action with a 'Handle' for the given 'KafkaAddress', +-- updating the connections cache if needed. +-- +-- When the action throws an 'IOException', it is caught and returned as a +-- 'KafkaIOException' in the Kafka monad. +-- +-- Note that when the given action throws an exception, any state changes will +-- be discarded. This includes both 'IOException's and exceptions thrown by +-- 'throwError' from 'Control.Monad.Except'. +withAddressHandle :: KafkaAddress -> (Handle -> Kafka a) -> Kafka a +withAddressHandle address kafkaAction = do + conns <- use stateConnections + let foundPool = conns ^. at address pool <- case foundPool of Nothing -> do - newPool <- liftIO $ mkPool broker - kafkaClientState . stateConnections .= (at broker ?~ newPool $ conns) + newPool <- tryKafkaIO $ mkPool address + stateConnections .= (at address ?~ newPool $ conns) return newPool Just p -> return p - Pool.withResource pool f - where mkPool :: Broker -> IO (Pool.Pool Handle) - mkPool b = Pool.createPool (createHandle b) hClose 1 10 1 - createHandle b = do - let h = b ^. brokerHost ^. hostString - p = b ^. brokerPort ^. portId - Network.connectTo h p + tryKafka $ Pool.withResource pool kafkaAction + where + mkPool :: KafkaAddress -> IO (Pool.Pool Handle) + mkPool a = Pool.createPool (createHandle a) hClose 1 10 1 + where createHandle (h, p) = Network.connectTo (h ^. hostString) (p ^. portId) +broker2address :: Broker -> KafkaAddress +broker2address broker = (,) (broker ^. brokerHost) (broker ^. brokerPort) +-- | Like 'withAddressHandle', but round-robins the addresses in the 'KafkaState'. +-- +-- When the action throws an 'IOException', it is caught and returned as a +-- 'KafkaIOException' in the Kafka monad. +-- +-- Note that when the given action throws an exception, any state changes will +-- be discarded. This includes both 'IOException's and exceptions thrown by +-- 'throwError' from 'Control.Monad.Except'. +withAnyHandle :: (Handle -> Kafka a) -> Kafka a +withAnyHandle f = do + (addy :| _) <- use stateAddresses + x <- withAddressHandle addy f + stateAddresses %= rotate + return x + where rotate :: NonEmpty a -> NonEmpty a + rotate = NE.fromList . rotate' 1 . NE.toList + rotate' n xs = zipWith const (drop n (cycle xs)) xs -- * Offsets @@ -272,11 +357,16 @@ , _maxNumOffsets :: MaxNumberOfOffsets } --- TODO: Properly look up the offset via the partition. -- | Get the first found offset. getLastOffset :: KafkaTime -> Partition -> TopicName -> Kafka Offset -getLastOffset m p t = - makeRequest (offsetRequest [(TopicAndPartition t p, PartitionOffsetRequestInfo m 1)]) >>= doRequest >>= maybe (StateT . const $ left KafkaNoOffset) return . firstOf (responseMessage . _OffsetResponse . offsetResponseOffset p) +getLastOffset m p t = do + broker <- getTopicPartitionLeader t p + withBrokerHandle broker (\h -> getLastOffset' h m p t) + +-- | Get the first found offset. +getLastOffset' :: Handle -> KafkaTime -> Partition -> TopicName -> Kafka Offset +getLastOffset' h m p t = + makeRequest (offsetRequest [(TopicAndPartition t p, PartitionOffsetRequestInfo m 1)]) >>= doRequest h >>= maybe (StateT . const $ throwError KafkaNoOffset) return . firstOf (responseMessage . _OffsetResponse . offsetResponseOffset p) -- | Create an offset request. offsetRequest :: [(TopicAndPartition, PartitionOffsetRequestInfo)] -> RequestMessage diff --git a/Network/Kafka/Producer.hs b/Network/Kafka/Producer.hs --- a/Network/Kafka/Producer.hs +++ b/Network/Kafka/Producer.hs @@ -1,10 +1,8 @@ module Network.Kafka.Producer where -import Prelude hiding ((!!)) import Control.Applicative import Control.Lens -import Control.Monad.Trans (liftIO, lift) -import Control.Monad.Trans.Either +import Control.Monad.Trans (liftIO) import Data.ByteString.Char8 (ByteString) import qualified Data.Digest.Murmur32 as Murmur32 import Data.List.Safe ((!!)) @@ -12,6 +10,7 @@ import System.IO import qualified Data.Map as M import System.Random (getStdRandom, randomR) +import Prelude hiding ((!!)) import Network.Kafka import Network.Kafka.Protocol @@ -21,7 +20,7 @@ -- | Execute a produce request and get the raw preduce response. produce :: Handle -> ProduceRequest -> Kafka ProduceResponse produce handle request = - makeRequest (ProduceRequest request) >>= doRequest' handle >>= expectResponse ExpectedProduce _ProduceResponse + makeRequest (ProduceRequest request) >>= doRequest handle >>= expectResponse ExpectedProduce _ProduceResponse -- | Construct a produce request with explicit arguments. produceRequest :: RequiredAcks -> Timeout -> [(TopicAndPartition, MessageSet)] -> ProduceRequest @@ -65,32 +64,12 @@ -- | Execute a produce request using the values in the state. send :: Leader -> [(TopicAndPartition, MessageSet)] -> Kafka ProduceResponse send l ts = do - let s = kafkaClientState . stateBrokers . at l + let s = stateBrokers . at l topicNames = map (_tapTopic . fst) ts broker <- findMetadataOrElse topicNames s (KafkaInvalidBroker l) - requiredAcks <- use (kafkaClientState . stateRequiredAcks) - requestTimeout <- use (kafkaClientState . stateRequestTimeout) + requiredAcks <- use stateRequiredAcks + requestTimeout <- use stateRequestTimeout withBrokerHandle broker $ \handle -> produce handle $ produceRequest requiredAcks requestTimeout ts - --- | Find a leader and partition for the topic. -brokerPartitionInfo :: TopicName -> Kafka [PartitionAndLeader] -brokerPartitionInfo t = do - let s = kafkaClientState . stateTopicMetadata . at t - tmd <- findMetadataOrElse [t] s KafkaFailedToFetchMetadata - return $ pal <$> tmd ^. partitionsMetadata - where pal d = PartitionAndLeader t (d ^. partitionId) (d ^. partitionMetadataLeader) - -findMetadataOrElse :: [TopicName] -> Getting (Maybe a) KafkaClient (Maybe a) -> KafkaClientError -> Kafka a -findMetadataOrElse ts s err = do - maybeFound <- use s - case maybeFound of - Just x -> return x - Nothing -> do - updateMetadatas ts - maybeFound' <- use s - case maybeFound' of - Just x -> return x - Nothing -> lift $ left $ err getRandPartition :: [PartitionAndLeader] -> Kafka (Maybe PartitionAndLeader) getRandPartition ps = diff --git a/Network/Kafka/Protocol.hs b/Network/Kafka/Protocol.hs --- a/Network/Kafka/Protocol.hs +++ b/Network/Kafka/Protocol.hs @@ -5,7 +5,7 @@ module Network.Kafka.Protocol where -import Control.Applicative (Applicative(..), Alternative(..), (<$>), (<*>)) +import Control.Applicative import Control.Category (Category(..)) import Control.Lens import Control.Monad (replicateM, liftM, liftM2, liftM3, liftM4, liftM5) diff --git a/milena.cabal b/milena.cabal --- a/milena.cabal +++ b/milena.cabal @@ -4,7 +4,7 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change -version: 0.3.0.0 +version: 0.4.0.0 synopsis: A Kafka client for Haskell. description: The protocol module is stable (the only changes will be to support changes in the Kafka protocol). The API is functional but subject to change. @@ -25,13 +25,13 @@ source-repository this type: git location: https://github.com/tylerholien/milena.git - tag: 0.3.0.0 + tag: 0.4.0.0 library default-language: Haskell2010 ghc-options: -Wall exposed-modules: Network.Kafka, - Network.Kafka.Protocol + Network.Kafka.Protocol, Network.Kafka.Producer build-depends: base >=4.7 && <5, mtl >=2.1 && <2.3, @@ -40,14 +40,14 @@ network >=2.4 && <2.7, digest >=0.0.1.0 && <0.1, containers >=0.5 && <0.6, - either >=4.3 && <4.4, random >=1.0 && <1.2, transformers >=0.3 && <0.5, lens >=4.4 && <4.13, resource-pool >=0.2.3.2 && <0.3, lifted-base >=0.2.3.6 && <0.3, murmur-hash >=0.1.0.8 && <0.2, - listsafe >=0.1.0.1 && <0.2 + listsafe >=0.1.0.1 && <0.2, + semigroups >=0.16.2.2 && <0.17 test-suite test default-language: Haskell2010 @@ -58,7 +58,24 @@ network, QuickCheck, bytestring, - hspec + lens, + semigroups, + tasty, + tasty-hspec, + tasty-quickcheck hs-source-dirs: test main-is: tests.hs type: exitcode-stdio-1.0 + +executable exkhs + hs-source-dirs: src + main-is: Main.hs + default-language: Haskell2010 + build-depends: base, + milena, + bytestring, + cereal, + mtl, + lifted-base, + lens, + pretty-show diff --git a/src/Main.hs b/src/Main.hs new file mode 100644 --- /dev/null +++ b/src/Main.hs @@ -0,0 +1,42 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE UnicodeSyntax #-} +{-# LANGUAGE ExplicitForAll #-} + +module Main where + +import qualified Data.ByteString.Char8 as C +import Control.Lens +import Control.Monad (forM) +import Control.Monad.Except (catchError) +import Network.Kafka +import Network.Kafka.Protocol +import System.Environment (getArgs) +-- import Text.Show.Pretty (ppShow) + +main ∷ IO () +main = do + h:portString:_ ← getArgs + let host = Host (KString (C.pack h)) + port = Port (read portString) + topic = "open_channel" + state = mkKafkaState "command-line-test-client" (host, port) + result ← runKafka state $ do + -- md ← metadata (MetadataReq [topic]) + -- putStrLnM (ppShow md) + topicPartitionList ← brokerPartitionInfo topic + forM topicPartitionList $ \(PartitionAndLeader { _palLeader, _palTopic, _palPartition }) → do + let s = stateBrokers . at _palLeader + broker ← findMetadataOrElse [topic] s (KafkaInvalidBroker _palLeader) + flip catchError (return . Left) $ do + result ← withBrokerHandle broker $ \handle → do + offset ← getLastOffset' handle EarliestTime _palPartition topic + fetchRequest offset _palPartition topic >>= fetch' handle + return $ Right result + print result + +-- tmd :: TopicMetadata +-- tmd = TopicMetadata (NoError, "omfg", [PartitionMetadata (NoError, 0, Leader (Just 1), Replicas [], Isr []), PartitionMetadata (NoError, 1, Leader (Just 1), Replicas [], Isr [])]) +-- newtype TopicMetadata = TopicMetadata { _topicMetadataFields :: (KafkaError, TopicName, [PartitionMetadata]) } deriving (Show, Eq, Deserializable) +-- newtype PartitionMetadata = PartitionMetadata { _partitionMetadataFields :: (KafkaError, Partition, Leader, Replicas, Isr) } deriving (Show, Eq, Deserializable) diff --git a/test/tests.hs b/test/tests.hs --- a/test/tests.hs +++ b/test/tests.hs @@ -3,18 +3,28 @@ module Main where import Data.Functor -import Data.Either (isRight) +import Data.Either (isRight, isLeft) +import qualified Data.List.NonEmpty as NE +import Control.Lens +import Control.Monad.Except (catchError, throwError) +import Control.Monad.Trans (liftIO) import Network.Kafka import Network.Kafka.Producer import Network.Kafka.Protocol (Leader(..)) -import Test.Hspec -import Test.Hspec.QuickCheck +import Test.Tasty +import Test.Tasty.Hspec +import Test.Tasty.QuickCheck import qualified Data.ByteString.Char8 as B +import Prelude + main :: IO () -main = hspec $ do +main = testSpec "the specs" specs >>= defaultMain + +specs :: Spec +specs = do let topic = "milena-test" - run = runKafka ("localhost", 9092) $ defaultState "milena-test-client" + run = runKafka $ mkKafkaState "milena-test-client" ("localhost", 9092) byteMessages = fmap (TopicAndMessage topic . makeMessage . B.pack) describe "can talk to local Kafka server" $ do @@ -22,6 +32,13 @@ result <- run . produceMessages $ byteMessages ms result `shouldSatisfy` isRight + prop "can produce multiple messages" $ \(ms, ms') -> do + result <- run $ do + r1 <- produceMessages $ byteMessages ms + r2 <- produceMessages $ byteMessages ms' + return $ r1 ++ r2 + result `shouldSatisfy` isRight + prop "can fetch messages" $ do result <- run $ do offset <- getLastOffset EarliestTime 0 topic @@ -37,3 +54,30 @@ void $ send leader [(TopicAndPartition topic 0, groupMessagesToSet messages)] fmap tamPayload . fetchMessages <$> (fetch =<< fetchRequest offset 0 topic) result `shouldBe` Right (tamPayload <$> messages) + + describe "withAddressHandle" $ do + it "turns 'IOException's into 'KafkaClientError's" $ do + result <- run $ withAddressHandle ("localhost", 9092) (\_ -> liftIO $ ioError $ userError "SOMETHING WENT WRONG!") :: IO (Either KafkaClientError ()) + result `shouldSatisfy` isLeft + + it "discards monadic effects when exceptions are thrown" $ do + result <- run $ do + stateName .= "expected" + _ <- flip catchError (return . Left) $ withAddressHandle ("localhost", 9092) $ \_ -> do + stateName .= "changed" + _ <- throwError KafkaFailedToFetchMetadata + n <- use stateName + return (Right n) + use stateName + result `shouldBe` Right "expected" + + describe "updateMetadatas" $ + it "de-dupes _stateAddresses" $ do + result <- run $ do + stateAddresses %= NE.cons ("localhost", 9092) + updateMetadatas [] + use stateAddresses + result `shouldBe` fmap NE.nub result + +prop :: Testable prop => String -> prop -> SpecWith () +prop s = it s . property