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 deserializeBytesd 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.

Minimal complete definition

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

Methods

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

getCtx :: m s

setCtx :: s -> m ()

modifyCtx :: (s -> s) -> 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 ()

sendMsg

Arguments

:: Serde a

=> Address

Function address (namespace, type, id)

-> a

message to send

-> m ()

sendMsgDelay

Arguments

:: Serde a

=> Address

Function address (namespace, type, id)

-> NominalDiffTime

delay before message send

-> a

message to send

-> m ()

returns cancelation token with which delivery of the message could be canceled

Instances

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 ()

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

sendMsg :: Serde a => Address -> a -> Function s ()

sendMsgDelay :: Serde a => Address -> NominalDiffTime -> a -> Function s ()

flinkWrapper :: forall a s. (Serde a, Serde s) => s -> Expiration -> (a -> Function s ()) -> FuncExec

Takes a function taking an arbitrary state type and converts it to take 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

data Address

Constructors

Address FuncType Text

Bundled Patterns

pattern Address' :: Text -> Text -> Text -> Address

data FuncType

Constructors

FuncType Text Text

Instances

Eq FuncType

Defined in Network.Flink.Internal.Stateful

Methods

(==) :: FuncType -> FuncType -> Bool

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

Ord FuncType

Defined in Network.Flink.Internal.Stateful

Methods

compare :: FuncType -> FuncType -> Ordering

(<) :: FuncType -> FuncType -> Bool

(<=) :: FuncType -> FuncType -> Bool

(>) :: FuncType -> FuncType -> Bool

(>=) :: FuncType -> FuncType -> Bool

max :: FuncType -> FuncType -> FuncType

min :: FuncType -> FuncType -> FuncType

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

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 ()

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

sendMsg :: Serde a => Address -> a -> Function s ()

sendMsgDelay :: Serde a => Address -> NominalDiffTime -> 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 Serde a where

Methods

tpName :: Proxy a -> Text

Type name

deserializeBytes :: ByteString -> Either String a

decodes types from strict ByteStrings

serializeBytes :: a -> ByteString

encodes types to strict ByteStrings

Instances

Serde ()

Defined in Network.Flink.Internal.Stateful

Methods

tpName :: Proxy () -> Text

deserializeBytes :: ByteString -> Either String ()

serializeBytes :: () -> ByteString

Serde ByteString

Defined in Network.Flink.Internal.Stateful

Methods

tpName :: Proxy ByteString -> Text

deserializeBytes :: ByteString -> Either String ByteString

serializeBytes :: ByteString -> ByteString

Serde ByteString

Defined in Network.Flink.Internal.Stateful

Methods

tpName :: Proxy ByteString -> Text

deserializeBytes :: ByteString0 -> Either String ByteString

serializeBytes :: ByteString -> ByteString0

Json a => Serde (JsonSerde a)

Defined in Network.Flink.Internal.Stateful

Methods

tpName :: Proxy (JsonSerde a) -> Text

deserializeBytes :: ByteString -> Either String (JsonSerde a)

serializeBytes :: JsonSerde a -> ByteString

Message a => Serde (ProtoSerde a)

Defined in Network.Flink.Internal.Stateful

Methods

tpName :: Proxy (ProtoSerde a) -> Text

deserializeBytes :: ByteString -> Either String (ProtoSerde a)

serializeBytes :: ProtoSerde a -> 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

ProtodeserializeBytesError String

InvalidTypePassedError Text Text

EmptyArgumentPassed

StateDecodeError String

MessageDecodeError String

ProtoMessageDecodeError UnpackError

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 FuncType FuncExec

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

data Expiration

Constructors

Expiration

Fields

• emode :: ExpirationMode
• expireAfterMillis :: NominalDiffTime

Instances

Eq Expiration

Defined in Network.Flink.Internal.Stateful

Methods

(==) :: Expiration -> Expiration -> Bool

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

Show Expiration

Defined in Network.Flink.Internal.Stateful

Methods

showsPrec :: Int -> Expiration -> ShowS

show :: Expiration -> String

showList :: [Expiration] -> ShowS

data ExpirationMode

Constructors

NONE

AFTER_WRITE

AFTER_CALL

Instances

Eq ExpirationMode

Defined in Network.Flink.Internal.Stateful

Methods

(==) :: ExpirationMode -> ExpirationMode -> Bool

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

Show ExpirationMode

Defined in Network.Flink.Internal.Stateful

Methods

showsPrec :: Int -> ExpirationMode -> ShowS

show :: ExpirationMode -> String

showList :: [ExpirationMode] -> ShowS

newState :: a -> FunctionState a

newtype ProtoSerde a

Constructors

ProtoSerde

Fields

• getProto :: a

Instances

Functor ProtoSerde

Defined in Network.Flink.Internal.Stateful

Methods

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

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

Message a => Serde (ProtoSerde a)

Defined in Network.Flink.Internal.Stateful

Methods

tpName :: Proxy (ProtoSerde a) -> Text

deserializeBytes :: ByteString -> Either String (ProtoSerde a)

serializeBytes :: ProtoSerde a -> ByteString

newtype JsonSerde a

Constructors

JsonSerde

Fields

• getJson :: a

Instances

Functor JsonSerde

Defined in Network.Flink.Internal.Stateful

Methods

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

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

Json a => Serde (JsonSerde a)

Defined in Network.Flink.Internal.Stateful

Methods

tpName :: Proxy (JsonSerde a) -> Text

deserializeBytes :: ByteString -> Either String (JsonSerde a)

serializeBytes :: JsonSerde a -> ByteString

jsonState :: Json s => Function s () -> Function (JsonSerde s) ()

Convenience function for wrapping state in newtype for JSON serialization

protoState :: Message s => Function s () -> Function (ProtoSerde s) ()

Convenience function for wrapping state in newtype for Protobuf serialization

sendProtoMsg :: (StatefulFunc s m, Message a) => Address -> a -> m ()

Convinience function to send protobuf messages

sendProtoMsgDelay :: (StatefulFunc s m, Message a) => Address -> NominalDiffTime -> a -> m ()

Convinience function to send delayed protobuf messages