Safe Haskell | None |
---|---|
Language | Haskell2010 |
Network.Flink.Internal.Stateful
Synopsis
- class MonadIO m => StatefulFunc s m | m -> s where
- makeConcrete :: (FlinkState s, Message a) => (a -> Function s ()) -> ByteString -> Env -> ToFunction'InvocationBatchRequest -> IO (Either FlinkError (FunctionState ByteString))
- createApp :: FunctionTable -> Application
- flinkServer :: FunctionTable -> Server FlinkApi
- flinkApi :: Proxy FlinkApi
- newtype Function s a = Function {
- runFunction :: ExceptT FlinkError (StateT (FunctionState s) (ReaderT Env IO)) a
- class FlinkState s where
- decodeState :: ByteString -> Either String s
- encodeState :: s -> ByteString
- data FunctionState ctx = FunctionState {}
- data FlinkError
- type FunctionTable = Map (Text, Text) (ByteString, ByteString -> Env -> ToFunction'InvocationBatchRequest -> IO (Either FlinkError (FunctionState ByteString)))
- data Env = Env {}
- newState :: a -> FunctionState a
Documentation
class MonadIO m => StatefulFunc s m | m -> s where Source #
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 Source #
modifyCtx :: (s -> s) -> m () Source #
Arguments
:: Message a | |
=> (Text, Text, Text) | Function address (namespace, type, id) |
-> a | protobuf message to send |
-> m () |
Instances
FlinkState s => StatefulFunc s (Function s) Source # | |
Defined in Network.Flink.Internal.Stateful Methods setInitialCtx :: s -> Function s () insideCtx :: (s -> a) -> Function s a Source # getCtx :: Function s s Source # setCtx :: s -> Function s () Source # modifyCtx :: (s -> s) -> Function s () Source # sendMsg :: Message a => (Text, Text, Text) -> a -> Function s () Source # sendMsgDelay :: Message a => (Text, Text, Text) -> Int -> a -> Function s () Source # sendEgressMsg :: Message a => (Text, Text) -> a -> Function s () Source # |
makeConcrete :: (FlinkState s, Message a) => (a -> Function s ()) -> ByteString -> Env -> ToFunction'InvocationBatchRequest -> IO (Either FlinkError (FunctionState ByteString)) Source #
Takes a function taking an abstract state/message type and converts it to take concrete ByteString
s
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 Source #
Takes function table and creates a wai Application
to serve flink requests
flinkServer :: FunctionTable -> Server FlinkApi Source #
Takes function table and creates a servant Server
to serve flink requests
Monad stack used for the execution of a Flink stateful function Don't reference this directly in your code if possible
Constructors
Function | |
Fields
|
Instances
class FlinkState s where Source #
Provides functions for Flink state SerDe
Methods
decodeState :: ByteString -> Either String s Source #
decodes Flink state types from strict ByteString
s
encodeState :: s -> ByteString Source #
encodes Flink state types to strict ByteString
s
Instances
FlinkState () Source # | |
Defined in Network.Flink.Internal.Stateful Methods decodeState :: ByteString -> Either String () Source # encodeState :: () -> ByteString Source # |
data FunctionState ctx Source #
Constructors
FunctionState | |
Instances
Functor FunctionState Source # | |
Defined in Network.Flink.Internal.Stateful Methods fmap :: (a -> b) -> FunctionState a -> FunctionState b # (<$) :: a -> FunctionState b -> FunctionState a # | |
Show ctx => Show (FunctionState ctx) Source # | |
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) Source # | |
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 Source #
Constructors
MissingInvocationBatch | |
ProtoUnpackError UnpackError | |
ProtoDeserializeError String | |
StateDecodeError String | |
NoSuchFunction (Text, Text) |
Instances
Eq FlinkError Source # | |
Defined in Network.Flink.Internal.Stateful | |
Show FlinkError Source # | |
Defined in Network.Flink.Internal.Stateful Methods showsPrec :: Int -> FlinkError -> ShowS # show :: FlinkError -> String # showList :: [FlinkError] -> ShowS # | |
MonadError FlinkError (Function s) Source # | |
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))) Source #
newState :: a -> FunctionState a Source #