{-# language BangPatterns #-}
{-# language LambdaCase #-}
{-# language TypeApplications #-}

-- | Adapted from kafka documentation. All fixed-width types use big-endian
-- encoding.
--
-- > BOOLEAN    | Single byte. Must be 0 or 1.
-- > INT8       | Signed 8-bit integer. Exactly 1 byte.
-- > INT16      | Signed 16-bit integer. Exactly 2 bytes.
-- > INT32      | Signed 32-bit integer. Exactly 4 bytes.
-- > INT64      | Signed 64-bit integer. Exactly 8 bytes.
-- > UINT32     | Unsigned 32-bit integer. Exactly 4 bytes.
-- > VARINT     | Signed 32-bit integer. LEB128 with zigzag.
-- > VARLONG    | Single 64-bit integer. LEB128 with zigzag.
-- > UUID       | 16 bytes
-- > FLOAT64    | Double-precision 64-bit format IEEE 754 value. Exactly 8 bytes.
-- > STRING                  | N is given as INT16. Then N bytes follow (UTF-8 sequence).
-- > COMPACT_STRING          | N+1 is given as an UNSIGNED_VARINT. Then N bytes follow (UTF-8 sequence).
-- > NULLABLE_STRING         | Same as STRING, but N = -1 means null
-- > COMPACT_NULLABLE_STRING | Same as COMPACT_STRING but N+1 = 0 means null.
-- > BYTES                   | N given as INT32. Then N bytes follow.
-- > COMPACT_BYTES           | N+1 given as UNSIGNED_VARINT. Then N bytes follow.
-- > NULLABLE_BYTES          | Same as BYTES but N = -1 means null.
-- > COMPACT_NULLABLE_BYTES  | Same as COMPACT_BYTES but N+1 = 0 means null.
-- > RECORDS                 | See official documentation
-- > ARRAY                   | Sequence of objects. First, N is given as INT32. Then N instances of type T follow. When N = -1, it means null.
-- > COMPACT_ARRAY           | Sequence of objects, First, N+1 is given as UNSIGNED_VARINT. Then N instances of type T follow. When N+1 = 0, it means null.
module Kafka.Builder
  ( Builder
  , nullableString
  , compactNullableString
  , compactString
  , compactBytes
  , nonCompactBytes
  , string
  , array
  , compactArray
  , compactNullableArray
  , int8
  , int16
  , int32
  , int32Array
  , int64
  , compactInt32Array
  , word128
  , varWordNative
  , varIntNative
  , varInt32
  , varInt64
  , boolean
    -- * Re-exports
  , word8
  , copy
  , fromBounded
  , run
  , consLength
  , bytes
  , Builder.runOnto
  , Builder.runOntoLength
  , Builder.chunks
  ) where

import Data.Bytes.Builder (Builder,fromBounded,run,word8,consLength,copy,bytes)
import Data.Int (Int8,Int16,Int32,Int64)
import Data.Primitive (SmallArray,PrimArray)
import Data.Text (Text)
import Data.Bytes (Bytes)
import Data.Bytes.Chunks (Chunks)
import Data.WideWord (Word128)
import qualified Data.Bytes as Bytes
import qualified Data.Bytes.Builder as Builder
import qualified Data.Bytes.Text.Utf8 as Utf8
import qualified Data.Primitive as PM

-- Implementation Note: We unconditionally copy the string since kafka strings
-- are typically small (less than 255 bytes). Topic names cannot even be more
-- than 255 bytes.
nullableString :: Maybe Text -> Builder
nullableString :: Maybe Text -> Builder
nullableString = \case
  Maybe Text
Nothing -> Int16 -> Builder
Builder.int16BE (-Int16
1)
  Just Text
s -> Text -> Builder
string Text
s

compactNullableString :: Maybe Text -> Builder
compactNullableString :: Maybe Text -> Builder
compactNullableString = \case
  Maybe Text
Nothing -> Word8 -> Builder
Builder.word8 Word8
0
  Just Text
s ->
    let b :: Bytes
b = Text -> Bytes
Utf8.fromText Text
s
     in Word -> Builder
Builder.wordLEB128 (forall a b. (Integral a, Num b) => a -> b
fromIntegral @Int @Word (Bytes -> Int
Bytes.length Bytes
b Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1)) Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<> Bytes -> Builder
Builder.copy Bytes
b

compactNullableArray :: (a -> Builder) -> Maybe (SmallArray a) -> Builder
{-# inline compactNullableArray #-}
compactNullableArray :: forall a. (a -> Builder) -> Maybe (SmallArray a) -> Builder
compactNullableArray a -> Builder
f Maybe (SmallArray a)
m = case Maybe (SmallArray a)
m of
  Maybe (SmallArray a)
Nothing -> Word8 -> Builder
Builder.word8 Word8
0
  Just SmallArray a
xs -> Word -> Builder
Builder.wordLEB128 (forall a b. (Integral a, Num b) => a -> b
fromIntegral @Int @Word (Int
1 Int -> Int -> Int
forall a. Num a => a -> a -> a
+ SmallArray a -> Int
forall a. SmallArray a -> Int
PM.sizeofSmallArray SmallArray a
xs)) Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<> (a -> Builder) -> SmallArray a -> Builder
forall m a. Monoid m => (a -> m) -> SmallArray a -> m
forall (t :: * -> *) m a.
(Foldable t, Monoid m) =>
(a -> m) -> t a -> m
foldMap a -> Builder
f SmallArray a
xs

compactString :: Text -> Builder
compactString :: Text -> Builder
compactString Text
s =
  let b :: Bytes
b = Text -> Bytes
Utf8.fromText Text
s
   in Bytes -> Builder
compactBytes Bytes
b

compactBytes :: Bytes -> Builder
compactBytes :: Bytes -> Builder
compactBytes Bytes
b = Word -> Builder
Builder.wordLEB128 (forall a b. (Integral a, Num b) => a -> b
fromIntegral @Int @Word (Bytes -> Int
Bytes.length Bytes
b Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1)) Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<> Bytes -> Builder
Builder.copy Bytes
b

string :: Text -> Builder
string :: Text -> Builder
string !Text
s = 
  let b :: Bytes
b = Text -> Bytes
Utf8.fromText Text
s
   in Int16 -> Builder
Builder.int16BE (forall a b. (Integral a, Num b) => a -> b
fromIntegral @Int @Int16 (Bytes -> Int
Bytes.length Bytes
b)) Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<> Bytes -> Builder
Builder.copy Bytes
b

nonCompactBytes :: Bytes -> Builder
nonCompactBytes :: Bytes -> Builder
nonCompactBytes !Bytes
b = 
  Int32 -> Builder
Builder.int32BE (forall a b. (Integral a, Num b) => a -> b
fromIntegral @Int @Int32 (Bytes -> Int
Bytes.length Bytes
b))
  Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<>
  Bytes -> Builder
Builder.copy Bytes
b

-- | Encode the length as @int32@. Then, encode all the elements one after another.
-- Does not support nullable array.
array :: (a -> Builder) -> SmallArray a -> Builder
{-# inline array #-}
array :: forall a. (a -> Builder) -> SmallArray a -> Builder
array a -> Builder
f !SmallArray a
xs = Int32 -> Builder
Builder.int32BE (forall a b. (Integral a, Num b) => a -> b
fromIntegral @Int @Int32 (SmallArray a -> Int
forall a. SmallArray a -> Int
PM.sizeofSmallArray SmallArray a
xs)) Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<> (a -> Builder) -> SmallArray a -> Builder
forall m a. Monoid m => (a -> m) -> SmallArray a -> m
forall (t :: * -> *) m a.
(Foldable t, Monoid m) =>
(a -> m) -> t a -> m
foldMap a -> Builder
f SmallArray a
xs

-- | Not nullable.
compactArray :: (a -> Builder) -> SmallArray a -> Builder
{-# inline compactArray #-}
compactArray :: forall a. (a -> Builder) -> SmallArray a -> Builder
compactArray a -> Builder
f !SmallArray a
xs = Word -> Builder
Builder.wordLEB128 (forall a b. (Integral a, Num b) => a -> b
fromIntegral @Int @Word (Int
1 Int -> Int -> Int
forall a. Num a => a -> a -> a
+ SmallArray a -> Int
forall a. SmallArray a -> Int
PM.sizeofSmallArray SmallArray a
xs)) Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<> (a -> Builder) -> SmallArray a -> Builder
forall m a. Monoid m => (a -> m) -> SmallArray a -> m
forall (t :: * -> *) m a.
(Foldable t, Monoid m) =>
(a -> m) -> t a -> m
foldMap a -> Builder
f SmallArray a
xs

-- x compactArrayChunks :: (a -> Chunks) -> SmallArray a -> Chunks
-- x {-# inline compactArrayChunks #-}
-- x compactArrayChunks f !xs = Builder.runOnto 20
-- x   (Builder.wordLEB128 (fromIntegral @Int @Word (1 + PM.sizeofSmallArray xs)))
-- x   (foldMap f xs)

int8 :: Int8 -> Builder
{-# inline int8 #-}
int8 :: Int8 -> Builder
int8 = Word8 -> Builder
Builder.word8 (Word8 -> Builder) -> (Int8 -> Word8) -> Int8 -> Builder
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Int8 -> Word8
forall a b. (Integral a, Num b) => a -> b
fromIntegral

int16 :: Int16 -> Builder
{-# inline int16 #-}
int16 :: Int16 -> Builder
int16 = Int16 -> Builder
Builder.int16BE

int32 :: Int32 -> Builder
{-# inline int32 #-}
int32 :: Int32 -> Builder
int32 = Int32 -> Builder
Builder.int32BE

int64 :: Int64 -> Builder
{-# inline int64 #-}
int64 :: Int64 -> Builder
int64 = Int64 -> Builder
Builder.int64BE

word128 :: Word128 -> Builder
{-# inline word128 #-}
word128 :: Word128 -> Builder
word128 = Word128 -> Builder
Builder.word128BE

varWordNative :: Word -> Builder
{-# inline varWordNative #-}
varWordNative :: Word -> Builder
varWordNative = Word -> Builder
Builder.wordLEB128

varIntNative :: Int -> Builder
{-# inline varIntNative #-}
varIntNative :: Int -> Builder
varIntNative = Int -> Builder
Builder.intLEB128

varInt64 :: Int64 -> Builder
{-# inline varInt64 #-}
varInt64 :: Int64 -> Builder
varInt64 = Int64 -> Builder
Builder.int64LEB128

varInt32 :: Int32 -> Builder
{-# inline varInt32 #-}
varInt32 :: Int32 -> Builder
varInt32 = Int32 -> Builder
Builder.int32LEB128

boolean :: Bool -> Builder
boolean :: Bool -> Builder
boolean Bool
b = case Bool
b of
  Bool
False -> Word8 -> Builder
Builder.word8 Word8
0
  Bool
True -> Word8 -> Builder
Builder.word8 Word8
1

int32Array :: PrimArray Int32 -> Builder
int32Array :: PrimArray Int32 -> Builder
int32Array !PrimArray Int32
x =
  Int32 -> Builder
Builder.int32BE (forall a b. (Integral a, Num b) => a -> b
fromIntegral @Int @Int32 Int
n)
  Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<>
  PrimArray Int32 -> Int -> Int -> Builder
Builder.int32ArrayBE PrimArray Int32
x Int
0 Int
n
  where
  !n :: Int
n = PrimArray Int32 -> Int
forall a. Prim a => PrimArray a -> Int
PM.sizeofPrimArray PrimArray Int32
x

compactInt32Array :: PrimArray Int32 -> Builder
compactInt32Array :: PrimArray Int32 -> Builder
compactInt32Array !PrimArray Int32
x =
  Word -> Builder
Builder.wordLEB128 (forall a b. (Integral a, Num b) => a -> b
fromIntegral @Int @Word (Int
1 Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
n))
  Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<>
  PrimArray Int32 -> Int -> Int -> Builder
Builder.int32ArrayBE PrimArray Int32
x Int
0 Int
n
  where
  !n :: Int
n = PrimArray Int32 -> Int
forall a. Prim a => PrimArray a -> Int
PM.sizeofPrimArray PrimArray Int32
x