chore: use HasKafka in favor of KAdmin

Author: JoranVanBelle

hw-kafka-client.cabal

@@ -55,9 +55,8 @@ library
   build-tool-depends:   c2hs:c2hs
   if impl(ghc <8.0)
     build-depends:      semigroups
-  exposed-modules:      Kafka.Admin
-                        Kafka.Admin.AdminProperties
-                        Kafka.Admin.Types
+  exposed-modules:      Kafka.Topic
+                        Kafka.Topic.Types
                         Kafka.Consumer
                         Kafka.Consumer.ConsumerProperties
                         Kafka.Consumer.Subscription

src/Kafka/Admin/AdminProperties.hs

@@ -1161,11 +1161,10 @@ rdKafkaErrorIsRetriable ptr = boolFromCInt <$> rdKafkaErrorIsRetriable' ptr
 rdKafkaErrorTxnRequiresAbort :: RdKafkaErrorTPtr -> IO Bool
 rdKafkaErrorTxnRequiresAbort ptr = boolFromCInt <$> rdKafkaErrorTxnRequiresAbort' ptr
 
--- Admin API
+-- Topics
 {#enum rd_kafka_admin_op_t as ^ {underscoreToCase} deriving (Show, Eq) #}
 
-data RdKafkaTopicResultT
-{#pointer *rd_kafka_topic_result_t as RdKafkaTopicResultTPtr foreign -> RdKafkaTopicResultT #}
+
 
 data RdKafkaAdminOptionsT
 {#pointer *rd_kafka_AdminOptions_t as RdKafkaAdminOptionsTPtr foreign -> RdKafkaAdminOptionsT #}

src/Kafka/Internal/RdKafka.chs

@@ -1,60 +1,42 @@
-module Kafka.Admin(
+module Kafka.Topic(
 module X
-, newKAdmin
 , createTopic
 , deleteTopic
-, closeKAdmin
 ) where
 
-import Control.Monad
-import Control.Monad.IO.Class
-import Data.Text
-import Data.List.NonEmpty
-import qualified Data.List.NonEmpty as NEL
-import qualified Data.Text as T
+import           Control.Monad.IO.Class
+import           Data.List.NonEmpty
+import qualified Data.List.NonEmpty     as NEL
+import           Data.Text
+import qualified Data.Text              as T
 
 import Kafka.Internal.RdKafka
 import Kafka.Internal.Setup
 
-import Kafka.Types as X
-import Kafka.Admin.AdminProperties as X
-import Kafka.Admin.Types as X
+import Kafka.Topic.Types as X
+import Kafka.Types       as X
 
-newKAdmin ::( MonadIO m )
-          => AdminProperties      
-          -> m (Either KafkaError KAdmin)
-newKAdmin properties = liftIO $ do
-  kafkaConfig@(KafkaConf kafkaConf' _ _) <- kafkaConf ( KafkaProps $ adminProps properties)
-  maybeKafka <- newRdKafkaT RdKafkaConsumer kafkaConf'
-  case maybeKafka of
-    Left err -> pure $ Left $ KafkaError err
-    Right kafka -> pure $ Right $ KAdmin (Kafka kafka) kafkaConfig
-
-closeKAdmin :: KAdmin 
-               -> IO ()
-closeKAdmin ka = void $ rdKafkaConsumerClose (getRdKafka ka)
 --- CREATE TOPIC ---
-createTopic :: KAdmin
-              -> NewTopic
-              -> IO (Either KafkaError TopicName)
-createTopic kAdmin topic = liftIO $ do
-  let kafkaPtr = getRdKafka kAdmin
+createTopic :: HasKafka k => k -> NewTopic -> IO (Either KafkaError TopicName)
+createTopic k topic = do
+  let kafkaPtr = getRdKafka k
   queue <- newRdKafkaQueue kafkaPtr
-  opts <- newRdKAdminOptions kafkaPtr RdKafkaAdminOpAny 
+  opts <- newRdKAdminOptions kafkaPtr RdKafkaAdminOpAny
 
   topicRes <- withNewTopic topic $ \topic' -> rdKafkaCreateTopic kafkaPtr topic' opts queue
   case topicRes of
-    Left err -> do 
+    Left err -> do
       pure $ Left (NEL.head err)
     Right _ -> do
       pure $ Right $ topicName topic
 
 --- DELETE TOPIC ---
-deleteTopic :: KAdmin
+deleteTopic :: HasKafka k
+              => k
               -> TopicName
               -> IO (Either KafkaError TopicName)
-deleteTopic kAdmin topic = liftIO $ do
-  let kafkaPtr = getRdKafka kAdmin
+deleteTopic k topic = liftIO $ do
+  let kafkaPtr = getRdKafka k
   queue <- newRdKafkaQueue kafkaPtr
   opts <- newRdKAdminOptions kafkaPtr RdKafkaAdminOpAny
 
@@ -65,15 +47,15 @@ deleteTopic kAdmin topic = liftIO $ do
     Right _ -> do
       pure $ Right topic
 
-withNewTopic :: NewTopic 
+withNewTopic :: NewTopic
                 -> (RdKafkaNewTopicTPtr ->  IO a)
                 -> IO (Either (NonEmpty KafkaError) a)
 withNewTopic t transform = do
   mkNewTopicRes <- mkNewTopic t newTopicPtr
   case mkNewTopicRes of
     Left err -> do
       return $ Left err
-    Right topic -> do 
+    Right topic -> do
       res <- transform topic
       return $ Right res
 
@@ -93,23 +75,23 @@ newTopicPtr :: NewTopic -> IO (Either KafkaError RdKafkaNewTopicTPtr)
 newTopicPtr topic = do
   ptrRes <- newRdKafkaNewTopic (unpack $ unTopicName $ topicName topic) (unPartitionCount $ topicPartitionCount topic) (unReplicationFactor $ topicReplicationFactor topic)
   case ptrRes of
-    Left str -> pure $ Left (KafkaError $ T.pack str)
+    Left str  -> pure $ Left (KafkaError $ T.pack str)
     Right ptr -> pure $ Right ptr
 
 oldTopicPtr :: TopicName -> IO (Either KafkaError RdKafkaDeleteTopicTPtr)
 oldTopicPtr tName = do
   res <- newRdKafkaDeleteTopic $ unpack . unTopicName $ tName
   case res of
-    Left str -> pure $ Left (KafkaError $ T.pack str)
+    Left str  -> pure $ Left (KafkaError $ T.pack str)
     Right ptr -> pure $ Right ptr
 
-mkNewTopic :: NewTopic 
+mkNewTopic :: NewTopic
               -> (NewTopic -> IO (Either KafkaError a))
               -> IO (Either (NonEmpty KafkaError) a)
 mkNewTopic topic create = do
   res <- create topic
   case res of
-    Left err -> pure $ Left (NEL.singleton err)
+    Left err       -> pure $ Left (singletonList err)
     Right resource -> pure $ Right resource
 
 rmOldTopic :: TopicName
@@ -118,5 +100,8 @@ rmOldTopic :: TopicName
 rmOldTopic tName remove = do
   res <- remove tName
   case res of
-    Left err -> pure $ Left (NEL.singleton err)
+    Left err       -> pure $ Left (singletonList err)
     Right resource -> pure $ Right resource
+
+singletonList :: a -> NonEmpty a
+singletonList x = x :| []

src/Kafka/Admin.hs renamed to src/Kafka/Topic.hs

@@ -1,27 +1,12 @@
-module Kafka.Admin.Types (
-KAdmin(..)
-, PartitionCount (..)
+module Kafka.Topic.Types (
+PartitionCount (..)
 , ReplicationFactor (..)
 , NewTopic (..)
 ) where
 
 import Data.Map
 
 import Kafka.Types
-import Kafka.Internal.Setup
-
-data KAdmin = KAdmin {
-  kaKafkaPtr  :: !Kafka
-  , kaKafkaConf :: !KafkaConf
-}
-
-instance HasKafka KAdmin where
-  getKafka = kaKafkaPtr
-  {-# INLINE getKafka #-}
-
-instance HasKafkaConf KAdmin where
-  getKafkaConf = kaKafkaConf
-  {-# INLINE getKafkaConf #-}
 
 newtype PartitionCount = PartitionCount { unPartitionCount :: Int } deriving (Show, Eq)
 newtype ReplicationFactor = ReplicationFactor { unReplicationFactor :: Int } deriving (Show, Eq)

src/Kafka/Admin/Types.hs renamed to src/Kafka/Topic/Types.hs

@@ -18,7 +18,7 @@ import Kafka.Consumer
 import qualified Data.Text as T
 import Kafka.Metadata
 import Kafka.Producer
-import Kafka.Admin
+import Kafka.Topic
 import Kafka.TestEnv
 import Test.Hspec
 
@@ -174,20 +174,19 @@ spec = do
                   forM_ res $ \rcs ->
                     forM_ rcs ((`shouldBe` Set.fromList (headersToList testHeaders)) . Set.fromList . headersToList . crHeaders)
 
-    describe "Kafka.Admin.Spec" $ do
+    describe "Kafka.Topic.Spec" $ do
         let topicName = addRandomChars "admin.topic.created." 5
 
         topicsMVar <- runIO newEmptyMVar
 
-        specWithAdmin "Create topic" $ do
-    
-          it "should create a new topic" $ \(admin :: KAdmin) -> do
+        specWithConsumer "Read all topics" consumerProps $ do
+
+          it "should create a topic" $ \(consumer :: KafkaConsumer) -> do
               tName <- topicName
               let newTopic = mkNewTopic (TopicName ( T.pack(tName) ))
-              result <- createTopic admin newTopic
+              result <- createTopic consumer newTopic
               result `shouldSatisfy` isRight
 
-        specWithConsumer "Read all topics" consumerProps $ do
 
           it "should return all the topics" $ \(consumer :: KafkaConsumer) -> do
             res <- allTopicsMetadata consumer (Timeout 1000)
@@ -205,13 +204,12 @@ spec = do
             topicsLen `shouldSatisfy` (>0)
             hasTopic `shouldBe` True
 
-        specWithAdmin "Remove topics" $ do
-          
-          it "should delete all the topics currently existing" $ \(admin ::KAdmin) -> do
+          it "should delete all the topics currently existing" $ \(consumer :: KafkaConsumer) -> do
             topics <- takeMVar topicsMVar
             forM_ topics $ \topic -> do
-              result <- deleteTopic admin topic
+              result <- deleteTopic consumer topic
               result `shouldSatisfy` isRight
+
 ----------------------------------------------------------------------------------------------------------------
 
 data ReadState = Skip | Read

tests-it/Kafka/IntegrationSpec.hs

@@ -15,7 +15,6 @@ import qualified System.Random as Rnd
 import Control.Concurrent
 import Kafka.Consumer     as C
 import Kafka.Producer     as P
-import Kafka.Admin        as A
 
 import Test.Hspec
 
@@ -58,9 +57,6 @@ producerProps =  P.brokersList [brokerAddress]
               <> P.setCallback (logCallback (\l s1 s2 -> print $ "[Producer] " <> show l <> ": " <> s1 <> ", " <> s2))
               <> P.setCallback (errorCallback (\e r -> print $ "[Producer] " <> show e <> ": " <> r))
 
-adminProperties :: AdminProperties
-adminProperties = A.brokers [brokerAddress]
-
 testSubscription :: TopicName -> Subscription
 testSubscription t = topics [t]
               <> offsetReset Earliest
@@ -80,9 +76,6 @@ mkConsumerWith props = do
       (RebalanceAssign _) -> putMVar var True
       _                   -> pure ()
 
-mkAdmin :: IO KAdmin
-mkAdmin = newKAdmin adminProperties >>= \(Right k) -> pure k
-
 specWithConsumer :: String -> ConsumerProperties -> SpecWith KafkaConsumer -> Spec
 specWithConsumer s p f =
   beforeAll (mkConsumerWith p)
@@ -97,9 +90,3 @@ specWithKafka s p f =
   beforeAll ((,) <$> mkConsumerWith p <*> mkProducer)
     $ afterAll (\(consumer, producer) -> void $ closeProducer producer >> closeConsumer consumer)
     $ describe s f
-
-specWithAdmin :: String -> SpecWith KAdmin -> Spec
-specWithAdmin s f = 
-  beforeAll mkAdmin 
-  $ afterAll (void . closeKAdmin) 
-  $ describe s f