__init__.py 文件源码

def main():
    args = _get_args()
    level = args['log_level'].upper()
    level = log_levels.get(level, logging.INFO)
    log = setup_logging(log_level=level)

    import core, utils, server

    config_file = args['CONFIG']
    if not utils.file_exists(config_file):
        log.error('File %s does not exist' % (config_file))
        sys.exit(1)

    log.info('Validating config')
    try:
        core.Engine.validate_config(config_file)
    except core.ConfigValidationError as ex:
        log.error('Invalid config. %s' % (str(ex)))
        sys.exit(1)

    log.info('Initializing xFlow engine')
    engine = core.Engine(config_file)
    log.info('Config is valid')

    # Run as server
    if args['s']:
        logging.info('Configuring xFlow Engine')
        engine.configure()
        app = server.create_app(engine)
        logging.info('Running as server')
        app.run(host='0.0.0.0', port=80, server='waitress', loglevel='warning')

    # Configure the lambdas, streams and subscriptions
    if args['c']:
        logging.info('Configuring xFlow Engine')
        engine.configure()
        logging.info('xFlow Engine configured')

    # Publish json data to stream
    if args['p']:
        stream = args['p'][0]
        data = args['p'][1]
        log.info('\n\n\nPublishing to stream: %s\n\nData: %s' % (stream, data))
        try:
            engine.publish(stream, data)
            log.info('Published')
        except core.KinesisStreamDoesNotExist:
            sys.exit(1)

    # Track a workflow
    if args['t']:
        workflow_id = args['t'][0]
        execution_id = args['t'][1]
        log.info("\n\n\nTracking workflow, workflow_id=%s, execution_id=%s" % (workflow_id, execution_id))
        try:
            tracking_info = engine.track(workflow_id, execution_id)
            print json.dumps(tracking_info, indent=4)
        except (core.CloudWatchStreamDoesNotExist,
                core.WorkflowDoesNotExist,
                core.CloudWatchLogDoesNotExist):
            sys.exit(1)