Network.Flink.Internal.Stateful

Documentation

class MonadIO m => StatefulFunc s m | m -> s where

Used to represent all Flink stateful function capabilities.

Contexts are received from Flink and deserialized into s all modifications to state are shipped back to Flink at the end of the batch to be persisted.

Message passing is also queued up and passed back at the end of the current batch.

Example of a stateless function (done by setting s to ()) that adds one to a number and puts the protobuf response on Kafka via an egress message:

adder :: StatefulFunc () m => AdderRequest -> m ()
adder msg = sendEgressMsg ("adder", "added") (kafkaRecord "added" name added)
  where
    num = msg ^. AdderRequest.num
    added = defMessage & AdderResponse.num .~ (num + 1)

Example of a stateful function:

newtype GreeterState = GreeterState
  { greeterStateCount :: Int
  }
  deriving (Generic, Show, ToJSON, FromJSON)

instance FlinkState GreeterState where
  decodeState = eitherDecode . BSL.fromStrict
  encodeState = BSL.toStrict . Data.Aeson.encode

counter :: StatefulFunc GreeterState m => EX.GreeterRequest -> m ()
counter msg = do
  newCount <- (+ 1) <$> insideCtx greeterStateCount
  let respMsg = "Saw " <> T.unpack name <> " " <> show newCount <> " time(s)"

  sendEgressMsg ("greeting", "greets") (kafkaRecord "greets" name $ response (T.pack respMsg))
  modifyCtx (old -> old {greeterStateCount = newCount})
  where
    name = msg ^. EX.name
    response :: Text -> EX.GreeterResponse
    response greeting =
      defMessage
        & EX.greeting .~ greeting

This will respond to each event by counting how many times it has been called for the name it was passed. The final state is taken and sent back to Flink. Failures of any kind will cause state to rollback to previous values seamlessly without double counting.

Minimal complete definition

setInitialCtx, insideCtx, getCtx, setCtx, modifyCtx, sendMsg, sendMsgDelay, sendEgressMsg

Methods

insideCtx :: (s -> a) -> m a

getCtx :: m s

setCtx :: s -> m ()

modifyCtx :: (s -> s) -> m ()

sendMsg

Arguments

:: Message a

=> (Text, Text, Text)

Function address (namespace, type, id)

-> a

protobuf message to send

-> m ()

sendMsgDelay

Arguments

:: Message a

=> (Text, Text, Text)

Function address (namespace, type, id)

-> Int

delay before message send

-> a

protobuf message to send

-> m ()

sendEgressMsg

Arguments

:: Message a

=> (Text, Text)

egress address (namespace, type)

-> a

protobuf message to send (should be a Kafka or Kinesis protobuf record)

-> m ()

Instances

FlinkState s => StatefulFunc s (Function s)

Defined in Network.Flink.Internal.Stateful

Methods

setInitialCtx :: s -> Function s ()

insideCtx :: (s -> a) -> Function s a

getCtx :: Function s s

setCtx :: s -> Function s ()

modifyCtx :: (s -> s) -> Function s ()

sendMsg :: Message a => (Text, Text, Text) -> a -> Function s ()

sendMsgDelay :: Message a => (Text, Text, Text) -> Int -> a -> Function s ()

sendEgressMsg :: Message a => (Text, Text) -> a -> Function s ()

makeConcrete :: (FlinkState s, Message a) => (a -> Function s ()) -> ByteString -> Env -> ToFunction'InvocationBatchRequest -> IO (Either FlinkError (FunctionState ByteString))

Takes a function taking an abstract state/message type and converts it to take concrete ByteStrings This allows each function in the FunctionTable to take its own individual type of state and just expose a function accepting ByteString to the library code.

createApp :: FunctionTable -> Application

Takes function table and creates a wai Application to serve flink requests

flinkServer :: FunctionTable -> Server FlinkApi

Takes function table and creates a servant Server to serve flink requests

flinkApi :: Proxy FlinkApi

newtype Function s a

Monad stack used for the execution of a Flink stateful function Don't reference this directly in your code if possible

Constructors

Function

Fields

• runFunction :: ExceptT FlinkError (StateT (FunctionState s) (ReaderT Env IO)) a

Instances

MonadReader Env (Function s)

Defined in Network.Flink.Internal.Stateful

Methods

ask :: Function s Env

local :: (Env -> Env) -> Function s a -> Function s a

reader :: (Env -> a) -> Function s a

MonadError FlinkError (Function s)

Defined in Network.Flink.Internal.Stateful

Methods

throwError :: FlinkError -> Function s a

catchError :: Function s a -> (FlinkError -> Function s a) -> Function s a

FlinkState s => StatefulFunc s (Function s)

Defined in Network.Flink.Internal.Stateful

Methods

setInitialCtx :: s -> Function s ()

insideCtx :: (s -> a) -> Function s a

getCtx :: Function s s

setCtx :: s -> Function s ()

modifyCtx :: (s -> s) -> Function s ()

sendMsg :: Message a => (Text, Text, Text) -> a -> Function s ()

sendMsgDelay :: Message a => (Text, Text, Text) -> Int -> a -> Function s ()

sendEgressMsg :: Message a => (Text, Text) -> a -> Function s ()

Monad (Function s)

Defined in Network.Flink.Internal.Stateful

Methods

(>>=) :: Function s a -> (a -> Function s b) -> Function s b

(>>) :: Function s a -> Function s b -> Function s b

return :: a -> Function s a

Functor (Function s)

Defined in Network.Flink.Internal.Stateful

Methods

fmap :: (a -> b) -> Function s a -> Function s b

(<$) :: a -> Function s b -> Function s a

Applicative (Function s)

Defined in Network.Flink.Internal.Stateful

Methods

pure :: a -> Function s a

(<*>) :: Function s (a -> b) -> Function s a -> Function s b

liftA2 :: (a -> b -> c) -> Function s a -> Function s b -> Function s c

(*>) :: Function s a -> Function s b -> Function s b

(<*) :: Function s a -> Function s b -> Function s a

MonadIO (Function s)

Defined in Network.Flink.Internal.Stateful

Methods

liftIO :: IO a -> Function s a

MonadState (FunctionState s) (Function s)

Defined in Network.Flink.Internal.Stateful

Methods

get :: Function s (FunctionState s)

put :: FunctionState s -> Function s ()

state :: (FunctionState s -> (a, FunctionState s)) -> Function s a

class FlinkState s where

Provides functions for Flink state SerDe

Methods

decodeState :: ByteString -> Either String s

decodes Flink state types from strict ByteStrings

encodeState :: s -> ByteString

encodes Flink state types to strict ByteStrings

Instances

FlinkState ()

Defined in Network.Flink.Internal.Stateful

Methods

decodeState :: ByteString -> Either String ()

encodeState :: () -> ByteString

data FunctionState ctx

Constructors

FunctionState

Fields

• functionStateCtx :: ctx
• functionStateMutated :: Bool
• functionStateInvocations :: Seq FromFunction'Invocation
• functionStateDelayedInvocations :: Seq FromFunction'DelayedInvocation
• functionStateEgressMessages :: Seq FromFunction'EgressMessage

Instances

Functor FunctionState

Defined in Network.Flink.Internal.Stateful

Methods

fmap :: (a -> b) -> FunctionState a -> FunctionState b

(<$) :: a -> FunctionState b -> FunctionState a

Show ctx => Show (FunctionState ctx)

Defined in Network.Flink.Internal.Stateful

Methods

showsPrec :: Int -> FunctionState ctx -> ShowS

show :: FunctionState ctx -> String

showList :: [FunctionState ctx] -> ShowS

MonadState (FunctionState s) (Function s)

Defined in Network.Flink.Internal.Stateful

Methods

get :: Function s (FunctionState s)

put :: FunctionState s -> Function s ()

state :: (FunctionState s -> (a, FunctionState s)) -> Function s a

data FlinkError

Constructors

MissingInvocationBatch

ProtoUnpackError UnpackError

ProtoDeserializeError String

StateDecodeError String

NoSuchFunction (Text, Text)

Instances

Eq FlinkError

Defined in Network.Flink.Internal.Stateful

Methods

(==) :: FlinkError -> FlinkError -> Bool

(/=) :: FlinkError -> FlinkError -> Bool

Show FlinkError

Defined in Network.Flink.Internal.Stateful

Methods

showsPrec :: Int -> FlinkError -> ShowS

show :: FlinkError -> String

showList :: [FlinkError] -> ShowS

MonadError FlinkError (Function s)

Defined in Network.Flink.Internal.Stateful

Methods

throwError :: FlinkError -> Function s a

catchError :: Function s a -> (FlinkError -> Function s a) -> Function s a

type FunctionTable = Map (Text, Text) (ByteString, ByteString -> Env -> ToFunction'InvocationBatchRequest -> IO (Either FlinkError (FunctionState ByteString)))

data Env

Constructors

Env

Fields

• envFunctionNamespace :: Text
• envFunctionType :: Text
• envFunctionId :: Text

Instances

Show Env

Defined in Network.Flink.Internal.Stateful

Methods

showsPrec :: Int -> Env -> ShowS

show :: Env -> String

showList :: [Env] -> ShowS

MonadReader Env (Function s)

Defined in Network.Flink.Internal.Stateful

Methods

ask :: Function s Env

local :: (Env -> Env) -> Function s a -> Function s a

reader :: (Env -> a) -> Function s a

newState :: a -> FunctionState a