Today, only two timeouts are supported (instead of many.) We need to add support for timeouts and also allow overrides.

Notes for activity timeouts:

schedule_to_close_timeout = start_to_close_timeout + schedule_to_start_timeout
# schedule_to_start_timeout can be set by the `create` method (if undefined, 5 will be used.)

More information on the timeouts are in the Flow Documentation.

Currently timeout are casted to string – so they can be provided to swf. Instead, we'll cast when we send to swf and not before.

Pass initial workflow execution context custom deciders.

This will allow the decider to make scheduling decisions passed on an initial context.

Likely will involve updating these lines of code:
https://github.com/xethorn/garcon/blob/master/garcon/decider.py#L180-L185

Generators are only able to get context that is set initially. They cannot access context that is dynamically set by the task list before. For example, I have a bootstrap task which get context from a config file, later on, i have another task which query database and update the context from query result right after; The generator in the next task list is not able to get the updated context.

The decider is currently getting all the results from all the tasks, and sends them as an input. SWF has a limit of 32k characters for input/output.

• Have the decider look at the activity and ask for all the required context keys
• Activity can look at all the tasks and get from them the context keys.
• Pluck from the context the required values, and send them as input.

Michael is the founder of CreativeList. Learn more about Creativelist - The Creator Search Engine. It's a search engine specialized in finding designers, photographers, or any type of creatives in big cities like new york, los angeles, san francisco or near you.

Problem

Today, the decider is generated automatically which simplifies how to write simple workflows. However, for workflows that are more complex to write (for instance: workflows with optional activities), it is a lot more difficult to create those using Garcon.

Goal

Add a set of features that make it easy to write more complex workflows (support for loops, conditionals, activity responses).

Draft

• schedule(activity, requires=[], **local_context). The schedule method is responsible for verifying that the activity instance have not been scheduled yet and all the dependencies have completed before scheduling this activity. Schedule can take a local context. This method returns a Result object. Requires is a list of activity that should have completed.
• Result: Representation of a scheduled event. This representation contains the state of the activity, the input of the activity, if completed the response of the activity. Scheduled events can be used as a way to trigger the following activities. Trying to get a value of the response or verifying if a result has processed throws an exception which triggers self.complete on the decider.

Code Sample

import boto.swf.layer2 as swf

from garcon import activity
from garcon import runner
import logging
import random


logger = logging.getLogger(__name__)

domain = 'dev'
name = 'workflow_sample'
create_activity = activity.create(domain, name)

TestActivity1 = create_activity(
    name='o',
    run=runner.Sync(
        lambda context, activity: logger.debug('activity_1')))

TestActivity2 = create_activity(
    name='activity_2',
    run=runner.Async(
        lambda context, activity: logger.debug('activity_2_task_1'),
        lambda context, activity: logger.debug('activity_2_task_2')))

TestActivity3 = create_activity(
    name='activity_3',
    retry=10,
    run=runner.Sync(
        lambda context, activity: logger.debug('activity_3')))

TestActivity4 = create_activity(
    name='activity_4',
    run=runner.Sync(
        lambda context, activity: logger.debug('activity_4')))


def run(execution, schedule):
    activity_1 = schedule(TestActivity)
    activity_2 = schedule(TestActivity2)
    if activity_1: 
       logger.debug('Activity1 has completed')
    activity_3 = schedule(TestActivity3, requires=(activity_1, activity_2))
    activity_4 = schedule(TestActivity4, requires=(activity_3,))

"def timeout" in task.py is not able to modify the timeout of the task. Task times out by default in 10 minutes regardless.