InternalShared.hs

module Haskakafka.InternalShared
where

import           Control.Exception
import           Control.Monad
import qualified Data.ByteString                as BS
import qualified Data.ByteString.Internal       as BSI
import           Foreign
import           Foreign.C.Error
import           Haskakafka.InternalRdKafka
import           Haskakafka.InternalRdKafkaEnum
import           Haskakafka.InternalTypes

word8PtrToBS :: Int -> Word8Ptr -> IO BS.ByteString
word8PtrToBS len ptr = BSI.create len $ \bsptr ->
    BSI.memcpy bsptr ptr len

fromMessagePtr :: RdKafkaMessageTPtr -> IO (Either KafkaError KafkaMessage)
fromMessagePtr ptr =
    withForeignPtr ptr $ \realPtr ->
    if realPtr == nullPtr then liftM (Left . kafkaRespErr) getErrno
    else do
        s <- peek realPtr
        msg <- if err'RdKafkaMessageT s /= RdKafkaRespErrNoError
            then return $ Left . KafkaResponseError $ err'RdKafkaMessageT s
            else Right <$> fromMessageStorable s
        rdKafkaMessageDestroy realPtr
        return msg

fromMessageStorable :: RdKafkaMessageT -> IO KafkaMessage
fromMessageStorable s = do
    payload <- word8PtrToBS (len'RdKafkaMessageT s) (payload'RdKafkaMessageT s)
    topic   <- newForeignPtr_ (topic'RdKafkaMessageT s) >>= rdKafkaTopicName

    key <- if key'RdKafkaMessageT s == nullPtr
               then return Nothing
               else liftM Just $ word8PtrToBS (keyLen'RdKafkaMessageT s) (key'RdKafkaMessageT s)

    return $ KafkaMessage
             topic
             (partition'RdKafkaMessageT s)
             (offset'RdKafkaMessageT s)
             payload
             key

kafkaRespErr :: Errno -> KafkaError
kafkaRespErr (Errno num) = KafkaResponseError $ rdKafkaErrno2err (fromIntegral num)
{-# INLINE kafkaRespErr #-}

throwOnError :: IO (Maybe String) -> IO ()
throwOnError action = do
    m <- action
    case m of
        Just e -> throw $ KafkaError e
        Nothing -> return ()

hasError :: KafkaError -> Bool
hasError err = case err of
    KafkaResponseError RdKafkaRespErrNoError -> False
    _ -> True
{-# INLINE hasError #-}

kafkaErrorToEither :: KafkaError -> Either KafkaError ()
kafkaErrorToEither err = case err of
    KafkaResponseError RdKafkaRespErrNoError -> Right ()
    _ -> Left err
{-# INLINE kafkaErrorToEither #-}

offsetToInt64 :: KafkaOffset -> Int64
offsetToInt64 o = case o of
    KafkaOffsetBeginning -> -2
    KafkaOffsetEnd       -> -1
    KafkaOffset off      -> off
    KafkaOffsetStored    -> -1000
    KafkaOffsetInvalid   -> -1001
    KafkaOffsetTail i    -> -2000 - i
{-# INLINE offsetToInt64 #-}

int64ToOffset :: Int64 -> KafkaOffset
int64ToOffset o
    | o == -2    = KafkaOffsetBeginning
    | o == -1    = KafkaOffsetEnd
    | o == -1000 = KafkaOffsetStored
    | o >= 0     = KafkaOffset o
    | o <= -2000 = KafkaOffsetTail (-2000 - o)
    | otherwise  = KafkaOffsetInvalid
{-# INLINE int64ToOffset #-}