{-# language BangPatterns #-}
{-# language DuplicateRecordFields #-}
{-# language NamedFieldPuns #-}

module Kafka.Assignment.Response.V1
  ( Assignment(..)
  , Ownership(..)
  , parser
  , decode
  ) where

import Control.Monad (when)
import Data.Bytes (Bytes)
import Data.Bytes.Parser (Parser)
import Data.Int (Int16,Int32)
import Data.Primitive (SmallArray)
import Data.Text (Text)
import Kafka.Parser.Context (Context)
import Kafka.Assignment.Request.V1 (Assignment(..),Ownership(..))
import Kafka.Subscription.Response.V1 (parserOwnership)

import qualified Data.Bytes.Parser as Parser
import qualified Kafka.Parser.Context as Ctx
import qualified Kafka.Parser

parser :: Context -> Parser Context s Assignment
parser :: forall s. Context -> Parser Context s Assignment
parser Context
ctx = do
  Int16
version <- Context -> Parser Context s Int16
forall e s. e -> Parser e s Int16
Kafka.Parser.int16 (Field -> Context -> Context
Ctx.Field Field
Ctx.Version Context
ctx)
  -- Since this is the V1 subscription, we require that the serialized
  -- message indicated V1 or greater.
  Bool -> Parser Context s () -> Parser Context s ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int16
version Int16 -> Int16 -> Bool
forall a. Ord a => a -> a -> Bool
<= Int16
0) (Context -> Parser Context s ()
forall e s a. e -> Parser e s a
Parser.fail (Field -> Context -> Context
Ctx.Field Field
Ctx.Version Context
ctx))
  SmallArray Ownership
assignedPartitions <- (Context -> Parser Context s Ownership)
-> Context -> Parser Context s (SmallArray Ownership)
forall s a.
(Context -> Parser Context s a)
-> Context -> Parser Context s (SmallArray a)
Kafka.Parser.array Context -> Parser Context s Ownership
forall s. Context -> Parser Context s Ownership
parserOwnership
    (Field -> Context -> Context
Ctx.Field Field
Ctx.AssignedPartitions Context
ctx) 
  Bytes
userData <- Context -> Parser Context s Bytes
forall s. Context -> Parser Context s Bytes
Kafka.Parser.nonCompactBytes
    (Field -> Context -> Context
Ctx.Field Field
Ctx.UserData Context
ctx)
  Assignment -> Parser Context s Assignment
forall a. a -> Parser Context s a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Assignment{SmallArray Ownership
assignedPartitions :: SmallArray Ownership
$sel:assignedPartitions:Assignment :: SmallArray Ownership
assignedPartitions,Bytes
userData :: Bytes
$sel:userData:Assignment :: Bytes
userData}

-- | Decode allows trailing content since we should be able
-- to parse newer versions of subscription and discard the
-- parts that we do not understand.
decode :: Bytes -> Either Context Assignment
decode :: Bytes -> Either Context Assignment
decode !Bytes
b = (forall s. Parser Context s Assignment)
-> Bytes -> Either Context Assignment
forall e a. (forall s. Parser e s a) -> Bytes -> Either e a
Parser.parseBytesEither (Context -> Parser Context s Assignment
forall s. Context -> Parser Context s Assignment
parser Context
Ctx.Top) Bytes
b