def messaging()

in python/showcase/showcase/__main__.py [0:0]

• 15 lines of code
• 7 McCabe index (conditional complexity)

def messaging(context: Context, message: Message):
    # You send messages to functions simply by specifying the target function's typename
    # and the target instance id for that for function; StateFun handles the routing for you,
    # without the need of any means for service discovery.
    target = "showcase/types"

    # you can directly send primitive type values as messages, ...
    context.send(message_builder(target_typename=target, target_id="0", bool_value=True))
    context.send(message_builder(target_typename=target, target_id="1", int_value=123))
    context.send(message_builder(target_typename=target, target_id="2", float_value=3.14159e+11))
    context.send(message_builder(target_typename=target, target_id="3", str_value="Hello world"))
    context.send(message_builder(target_typename=target, target_id="4", double_value=1.23))
    context.send(message_builder(target_typename=target, target_id="5", long_value=123456789))
    # ... or, in general, a value of any custom defined type.
    context.send(
        message_builder(target_typename=target, target_id="6", value={"name": "Joe"}, value_type=GREET_JSON_TYPE))

    # or extract the payload directly
    print(f"The payload's type is {message.value_typename()} and the raw bytes are {message.raw_value()}")

    # You can send messages to any function, including yourself!
    me = context.address
    context.send(message_builder(target_typename=me.typename, target_id=me.id, str_value="hello!"))

    # Additionally, you may ask StateFun to send out a message after a specified delay.
    # A common usage pattern is to send delayed messages to yourself to model timer triggers.
    context.send_after(timedelta(minutes=10),
                       message_builder(target_typename=target, target_id="7", str_value="Hello from the future"))