{ "cells": [ { "cell_type": "markdown", "id": "informative-conservation", "metadata": {}, "source": [ "# flow API Example\n", "\n", "The [sarracenia.flow class](../Reference/code.rst#module-sarracenia.flow) provides built in accept/reject filtering for messages, supports built-in downloading in several protocols, retries on failure, and allows the creation of callbacks, to customize processing.\n", "\n", "You need to provide a configuration as an argument when instantiating a subscriber.\n", "the _sarracenia.config.no_file_config()_ returns an empty configuration without consulting\n", "any of the sr3 configuration file tree.\n", "\n", "After adding the modifications needed to the configuration, the subscriber is then initiated and run." ] }, { "cell_type": "code", "execution_count": 2, "id": "weekly-terminology", "metadata": { "scrolled": true }, "outputs": [], "source": [ "!mkdir /tmp/flow_demo" ] }, { "cell_type": "markdown", "id": "exterior-folks", "metadata": {}, "source": [ "make a directory for the files you are going to download.\n", "the root of the directory tree to must exist." ] }, { "cell_type": "code", "execution_count": 3, "id": "aggregate-election", "metadata": {}, "outputs": [], "source": [ "import re\n", "import sarracenia.config\n", "from sarracenia.flow.subscribe import Subscribe\n", "import sarracenia.flowcb\n", "import sarracenia.credentials\n", "\n", "cfg = sarracenia.config.no_file_config()\n", "\n", "cfg.broker = sarracenia.credentials.Credential('amqps://anonymous:anonymous@hpfx.collab.science.gc.ca')\n", "cfg.topicPrefix = [ 'v02', 'post']\n", "cfg.component = 'subscribe'\n", "cfg.config = 'flow_demo'\n", "cfg.action = 'start'\n", "cfg.bindings = [ ('xpublic', ['v02', 'post'], ['*', 'WXO-DD', 'observations', 'swob-ml', '#' ]) ]\n", "cfg.queueName='q_anonymous.subscriber_test2'\n", "cfg.download=True\n", "cfg.batch=1\n", "cfg.messageCountMax=5\n", "\n", "# set the instance number for the flow class.\n", "cfg.no=0\n", "\n", "# set other settings based on provided ones, so it is ready for use.\n", "\n", "cfg.finalize()\n", "\n", "# accept/reject patterns:\n", "pattern=\".*\"\n", "# to_match, write_to_dir, DESTFN, regex_to_match, accept=True,mirror,strip, pstrip,flatten\n", "cfg.masks= [ ( pattern, \"/tmp/flow_demo\", None, re.compile(pattern), True, False, False, False, '/' ) ]\n", "\n", "\n" ] }, { "cell_type": "markdown", "id": "legitimate-necessity", "metadata": {}, "source": [ "\n", "## starters.\n", "the broker, bindings, and queueName settings are explained in the moth notebook.\n", "\n", "## cfg.download\n", "\n", "Whether you want the flow to download the files corresponding to the messages.\n", "If true, then it will download the files.\n", "\n", "## cfg.batch\n", "\n", "Messages are processed in batches. The number of messages to retrieve per call to newMessages()\n", "is limited by the _batch_ setting. We set it to 1 here so you can see each file being downloaded immediately when the corresponding message is downloaded. you can leave this blank, and it defaults to 25. Settings are matter of taste and use case.\n", "\n", "## cfg.messageCountMax\n", "\n", "Normally we just leave this setting at it's default (0) which has no effect on processing.\n", "for demonstration purposes, we limit the number of messages the subscriber will process with this setting.\n", "after _messageCountMax_ messages have been received, stop processing.\n", "\n", "\n", "## cfg.masks\n", "masks are a compiled form of accept/reject directives. a relPath is compared to the regex in the mask.\n", "If the regex matches, and accept is true, then the message is accepted for further processing.\n", "If the regex matches, but accept is False, then processing of the message is stopped (the message is rejected.)\n", "\n", "masks are a tuple. the meaning can be looked up in the sr3(1) man page.\n", "\n", "* pattern_string, the input regular expression string, to be compiled by re routines.\n", "* directory, where to put the files downloaded (root of the tree, when mirroring)\n", "* fn, transformation of filename to do. None is the 99% use case.\n", "* regex, compiled regex version of the pattern_string\n", "* accept(True/False), if pattern matches then accept message for further processing.\n", "* mirror(True/False), when downloading build a complete tree to mirror the source, or just dump in directory\n", "* strip(True/False), modify the relpath by stripping entries from the left.\n", "* pstrip(True/False), strip entries based on patterm\n", "* flatten(char ... '/' means do not flatten.) )\n", "\n", "## cfg.no, cfg.pid_filename\n", "\n", "These settings are needed because they would ordinarily be set by the sarracenia.instance class which is\n", "normally used to launch flows. They allow setting up of run-time paths for retry_queues, and statefiles,\n", "to remember settings if need be between runs.\n" ] }, { "cell_type": "code", "execution_count": 4, "id": "musical-discrimination", "metadata": { "scrolled": false }, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "2024-01-29 15:00:37,351 [INFO] sarracenia.flow loadCallbacks flowCallback plugins to load: ['sarracenia.flowcb.gather.message.Message', 'sarracenia.flowcb.retry.Retry', 'sarracenia.flowcb.housekeeping.resources.Resources', 'log']\n", "2024-01-29 15:00:37,354 [DEBUG] sarracenia.flowcb.retry __init__ sr_retry __init__\n", "2024-01-29 15:00:37,354 [DEBUG] sarracenia.config add_option []0 retry_driver declared as type: value:disk\n", "2024-01-29 15:00:37,355 [DEBUG] sarracenia.diskqueue __init__ work_retry_00 __init__\n", "2024-01-29 15:00:37,357 [DEBUG] sarracenia.config add_option []0 MemoryMax declared as type: value:0\n", "2024-01-29 15:00:37,357 [DEBUG] sarracenia.config add_option []0 MemoryBaseLineFile declared as type: value:100\n", "2024-01-29 15:00:37,358 [DEBUG] sarracenia.config add_option []0 MemoryMultiplier declared as type: value:3\n", "2024-01-29 15:00:37,359 [DEBUG] sarracenia.config add_option []0 logEvents declared as type: value:{'after_work', 'on_housekeeping', 'after_accept', 'after_post'}\n", "2024-01-29 15:00:37,359 [DEBUG] sarracenia.config add_option []0 logMessageDump declared as type: value:False\n", "2024-01-29 15:00:37,359 [INFO] sarracenia.flowcb.log __init__ subscribe initialized with: logEvents: {'after_work', 'on_housekeeping', 'after_accept', 'after_post'}, logMessageDump: False\n", "2024-01-29 15:00:37,360 [DEBUG] sarracenia.config check_undeclared_options missing defaults: {'post_exchangeSplit', 'follow_symlinks', 'topic', 'post_exchange', 'reconnect', 'sendTo', 'pollUrl', 'logMessageDump', 'MemoryBaseLineFile', 'exchange_suffix', 'force_polling', 'exchangeSplit', 'post_topic', 'save', 'inplace', 'retry_driver', 'post_on_start', 'header', 'blocksize', 'restore', 'MemoryMultiplier', 'report_exchange', 'cluster', 'nodupe_basis', 'post_exchangeSuffix', 'identity', 'MemoryMax', 'count', 'notify_only', 'feeder', 'realpathFilter'}\n", "2024-01-29 15:00:37,360 [INFO] sarracenia.flow run callbacks loaded: ['sarracenia.flowcb.gather.message.Message', 'sarracenia.flowcb.retry.Retry', 'sarracenia.flowcb.housekeeping.resources.Resources', 'log']\n", "2024-01-29 15:00:37,360 [INFO] sarracenia.flow run pid: 3567801 subscribe/flow_demo instance: 0\n", "2024-01-29 15:00:37,448 [DEBUG] amqp _on_start Start from server, version: 0.9, properties: {'capabilities': {'publisher_confirms': True, 'exchange_exchange_bindings': True, 'basic.nack': True, 'consumer_cancel_notify': True, 'connection.blocked': True, 'consumer_priorities': True, 'authentication_failure_close': True, 'per_consumer_qos': True, 'direct_reply_to': True}, 'cluster_name': 'rabbit@hpfx2.collab.science.gc.ca', 'copyright': 'Copyright (c) 2007-2022 VMware, Inc. or its affiliates.', 'information': 'Licensed under the MPL 2.0. Website: https://rabbitmq.com', 'platform': 'Erlang/OTP 24.2.1', 'product': 'RabbitMQ', 'version': '3.9.13'}, mechanisms: [b'PLAIN', b'AMQPLAIN'], locales: ['en_US']\n", "2024-01-29 15:00:37,493 [DEBUG] amqp __init__ using channel_id: 1\n", "2024-01-29 15:00:37,514 [DEBUG] amqp _on_open_ok Channel open\n", "2024-01-29 15:00:37,514 [DEBUG] amqp __init__ using channel_id: 2\n", "2024-01-29 15:00:37,535 [DEBUG] amqp _on_open_ok Channel open\n", "2024-01-29 15:00:37,634 [DEBUG] amqp _on_start Start from server, version: 0.9, properties: {'capabilities': {'publisher_confirms': True, 'exchange_exchange_bindings': True, 'basic.nack': True, 'consumer_cancel_notify': True, 'connection.blocked': True, 'consumer_priorities': True, 'authentication_failure_close': True, 'per_consumer_qos': True, 'direct_reply_to': True}, 'cluster_name': 'rabbit@hpfx2.collab.science.gc.ca', 'copyright': 'Copyright (c) 2007-2022 VMware, Inc. or its affiliates.', 'information': 'Licensed under the MPL 2.0. Website: https://rabbitmq.com', 'platform': 'Erlang/OTP 24.2.1', 'product': 'RabbitMQ', 'version': '3.9.13'}, mechanisms: [b'PLAIN', b'AMQPLAIN'], locales: ['en_US']\n", "2024-01-29 15:00:37,681 [DEBUG] amqp __init__ using channel_id: 1\n", "2024-01-29 15:00:37,699 [DEBUG] amqp _on_open_ok Channel open\n", "2024-01-29 15:00:37,699 [DEBUG] amqp __init__ using channel_id: 2\n", "2024-01-29 15:00:37,730 [DEBUG] amqp _on_open_ok Channel open\n", "2024-01-29 15:00:37,749 [INFO] sarracenia.moth.amqp _queueDeclare queue declared q_anonymous.subscriber_test2 (as: amqps://anonymous@hpfx.collab.science.gc.ca), (messages waiting: 50000)\n", "2024-01-29 15:00:37,749 [INFO] sarracenia.moth.amqp getSetup binding q_anonymous.subscriber_test2 with v02.post.*.WXO-DD.observations.swob-ml.# to xpublic (as: amqps://anonymous@hpfx.collab.science.gc.ca)\n", "2024-01-29 15:00:37,765 [DEBUG] sarracenia.moth.amqp getSetup getSetup ... Done!\n", "2024-01-29 15:00:37,786 [DEBUG] sarracenia.moth.amqp getNewMessage new msg: {'_format': 'v02', '_deleteOnPost': {'source', '_format', 'exchange', 'subtopic', 'local_offset', 'ack_id'}, 'sundew_extension': 'DMS:WXO_RENAMED_SWOB2:MSC:XML::20240129174355', 'from_cluster': 'DDSR.CMC', 'to_clusters': 'ALL', 'filename': 'msg_ddsr-WXO-DD_f2884f4dfeb89a44ec2ccbcc1c154702:DMS:WXO_RENAMED_SWOB2:MSC:XML::20240129174355', 'source': 'anonymous', 'mtime': '20240129T174356.779', 'atime': '20240129T174356.779', 'pubTime': '20240129T174356.779', 'baseUrl': 'https://hpfx.collab.science.gc.ca', 'relPath': '/20240129/WXO-DD/observations/swob-ml/20240129/CXCK/2024-01-29-1743-CXCK-AUTO-minute-swob.xml', 'subtopic': ['20240129', 'WXO-DD', 'observations', 'swob-ml', '20240129', 'CXCK'], 'identity': {'method': 'md5', 'value': 'sZvG3KgpfENZc15YSMHvbQ=='}, 'size': 9326, 'exchange': 'xpublic', 'ack_id': 1, 'local_offset': 0}\n", "2024-01-29 15:00:37,787 [INFO] sarracenia.flowcb.log after_accept accepted: (lag: 8201.01 ) https://hpfx.collab.science.gc.ca /20240129/WXO-DD/observations/swob-ml/20240129/CXCK/2024-01-29-1743-CXCK-AUTO-minute-swob.xml\n", "2024-01-29 15:00:37,787 [INFO] sarracenia.flow run now active on vip ['AnyAddressIsFine']\n", "2024-01-29 15:00:37,788 [INFO] sarracenia.flow do_download missing destination directories, makedirs: /tmp/flow_demo/20240129/WXO-DD/observations/swob-ml/20240129/CXCK \n", "2024-01-29 15:00:37,789 [DEBUG] sarracenia.config add_option []0 accelWgetCommand declared as type: value:/usr/bin/wget %s -o - -O %d\n", "2024-01-29 15:00:37,887 [INFO] sarracenia.flowcb.log after_work downloaded ok: /tmp/flow_demo/20240129/WXO-DD/observations/swob-ml/20240129/CXCK/2024-01-29-1743-CXCK-AUTO-minute-swob.xml \n", "2024-01-29 15:00:37,912 [DEBUG] sarracenia.moth.amqp getNewMessage new msg: {'_format': 'v02', '_deleteOnPost': {'source', '_format', 'exchange', 'subtopic', 'local_offset', 'ack_id'}, 'sundew_extension': 'DMS:WXO_RENAMED_SWOB2:MSC:XML::20240129174355', 'from_cluster': 'DDSR.CMC', 'to_clusters': 'ALL', 'filename': 'msg_ddsr-WXO-DD_fc8051d6b19291e9b02b8da5f6fc3d2f:DMS:WXO_RENAMED_SWOB2:MSC:XML::20240129174355', 'source': 'anonymous', 'mtime': '20240129T174356.779', 'atime': '20240129T174356.779', 'pubTime': '20240129T174356.779', 'baseUrl': 'https://hpfx.collab.science.gc.ca', 'relPath': '/20240129/WXO-DD/observations/swob-ml/20240129/CZKD/2024-01-29-1743-CZKD-AUTO-minute-swob.xml', 'subtopic': ['20240129', 'WXO-DD', 'observations', 'swob-ml', '20240129', 'CZKD'], 'identity': {'method': 'md5', 'value': 'yU3e4yc2eVtN+qwiiohaLQ=='}, 'size': 9440, 'exchange': 'xpublic', 'ack_id': 2, 'local_offset': 0}\n", "2024-01-29 15:00:37,912 [INFO] sarracenia.flowcb.log after_accept accepted: (lag: 8201.13 ) https://hpfx.collab.science.gc.ca /20240129/WXO-DD/observations/swob-ml/20240129/CZKD/2024-01-29-1743-CZKD-AUTO-minute-swob.xml\n", "2024-01-29 15:00:37,913 [INFO] sarracenia.flow do_download missing destination directories, makedirs: /tmp/flow_demo/20240129/WXO-DD/observations/swob-ml/20240129/CZKD \n", "2024-01-29 15:00:38,000 [INFO] sarracenia.flowcb.log after_work downloaded ok: /tmp/flow_demo/20240129/WXO-DD/observations/swob-ml/20240129/CZKD/2024-01-29-1743-CZKD-AUTO-minute-swob.xml \n", "2024-01-29 15:00:38,024 [DEBUG] sarracenia.moth.amqp getNewMessage new msg: {'_format': 'v02', '_deleteOnPost': {'source', '_format', 'exchange', 'subtopic', 'local_offset', 'ack_id'}, 'sundew_extension': 'DMS:WXO_RENAMED_SWOB2:MSC:XML::20240129174355', 'from_cluster': 'DDSR.CMC', 'to_clusters': 'ALL', 'filename': 'msg_ddsr-WXO-DD_fe4c49c3c2cc0493ae7473d321d25199:DMS:WXO_RENAMED_SWOB2:MSC:XML::20240129174355', 'source': 'anonymous', 'mtime': '20240129T174356.779', 'atime': '20240129T174356.779', 'pubTime': '20240129T174356.779', 'baseUrl': 'https://hpfx.collab.science.gc.ca', 'relPath': '/20240129/WXO-DD/observations/swob-ml/20240129/CVBB/2024-01-29-1743-CVBB-AUTO-minute-swob.xml', 'subtopic': ['20240129', 'WXO-DD', 'observations', 'swob-ml', '20240129', 'CVBB'], 'identity': {'method': 'md5', 'value': 'Hwu7CE6asjaQMz7veEmUXA=='}, 'size': 8399, 'exchange': 'xpublic', 'ack_id': 3, 'local_offset': 0}\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "2024-01-29 15:00:38,025 [INFO] sarracenia.flowcb.log after_accept accepted: (lag: 8201.25 ) https://hpfx.collab.science.gc.ca /20240129/WXO-DD/observations/swob-ml/20240129/CVBB/2024-01-29-1743-CVBB-AUTO-minute-swob.xml\n", "2024-01-29 15:00:38,025 [INFO] sarracenia.flow do_download missing destination directories, makedirs: /tmp/flow_demo/20240129/WXO-DD/observations/swob-ml/20240129/CVBB \n", "2024-01-29 15:00:38,114 [INFO] sarracenia.flowcb.log after_work downloaded ok: /tmp/flow_demo/20240129/WXO-DD/observations/swob-ml/20240129/CVBB/2024-01-29-1743-CVBB-AUTO-minute-swob.xml \n", "2024-01-29 15:00:38,139 [DEBUG] sarracenia.moth.amqp getNewMessage new msg: {'_format': 'v02', '_deleteOnPost': {'source', '_format', 'exchange', 'subtopic', 'local_offset', 'ack_id'}, 'sundew_extension': 'DMS:WXO_RENAMED_SWOB2:MSC:XML::20240129174356', 'from_cluster': 'DDSR.CMC', 'to_clusters': 'ALL', 'filename': 'msg_ddsr-WXO-DD_8067f0a1a5b4711ab86e481341b26590:DMS:WXO_RENAMED_SWOB2:MSC:XML::20240129174356', 'source': 'anonymous', 'mtime': '20240129T174357.781', 'atime': '20240129T174357.781', 'pubTime': '20240129T174357.781', 'baseUrl': 'https://hpfx.collab.science.gc.ca', 'relPath': '/20240129/WXO-DD/observations/swob-ml/20240129/CWLJ/2024-01-29-1743-CWLJ-AUTO-minute-swob.xml', 'subtopic': ['20240129', 'WXO-DD', 'observations', 'swob-ml', '20240129', 'CWLJ'], 'identity': {'method': 'md5', 'value': 'uDrzi9GLNnhEgGvSylHu9g=='}, 'size': 9428, 'exchange': 'xpublic', 'ack_id': 4, 'local_offset': 0}\n", "2024-01-29 15:00:38,140 [INFO] sarracenia.flowcb.log after_accept accepted: (lag: 8200.36 ) https://hpfx.collab.science.gc.ca /20240129/WXO-DD/observations/swob-ml/20240129/CWLJ/2024-01-29-1743-CWLJ-AUTO-minute-swob.xml\n", "2024-01-29 15:00:38,141 [INFO] sarracenia.flow do_download missing destination directories, makedirs: /tmp/flow_demo/20240129/WXO-DD/observations/swob-ml/20240129/CWLJ \n", "2024-01-29 15:00:38,242 [INFO] sarracenia.flowcb.log after_work downloaded ok: /tmp/flow_demo/20240129/WXO-DD/observations/swob-ml/20240129/CWLJ/2024-01-29-1743-CWLJ-AUTO-minute-swob.xml \n", "2024-01-29 15:00:38,262 [DEBUG] sarracenia.moth.amqp getNewMessage new msg: {'_format': 'v02', '_deleteOnPost': {'source', '_format', 'exchange', 'subtopic', 'local_offset', 'ack_id'}, 'sundew_extension': 'DMS:WXO_RENAMED_SWOB2:MSC:XML::20240129174355', 'from_cluster': 'DDSR.CMC', 'to_clusters': 'ALL', 'filename': 'msg_ddsr-WXO-DD_6f203257347d4f090abc1d7557864cb7:DMS:WXO_RENAMED_SWOB2:MSC:XML::20240129174355', 'source': 'anonymous', 'mtime': '20240129T174357.267', 'atime': '20240129T174357.267', 'pubTime': '20240129T174357.267', 'baseUrl': 'https://hpfx.collab.science.gc.ca', 'relPath': '/20240129/WXO-DD/observations/swob-ml/20240129/CAMS/2024-01-29-1743-CAMS-AUTO-minute-swob.xml', 'subtopic': ['20240129', 'WXO-DD', 'observations', 'swob-ml', '20240129', 'CAMS'], 'identity': {'method': 'md5', 'value': 'H/h4jm6MTzMSp1oCeDS1jA=='}, 'size': 9826, 'exchange': 'xpublic', 'ack_id': 5, 'local_offset': 0}\n", "2024-01-29 15:00:38,263 [INFO] sarracenia.flowcb.log after_accept accepted: (lag: 8201.00 ) https://hpfx.collab.science.gc.ca /20240129/WXO-DD/observations/swob-ml/20240129/CAMS/2024-01-29-1743-CAMS-AUTO-minute-swob.xml\n", "2024-01-29 15:00:38,263 [INFO] sarracenia.flow do_download missing destination directories, makedirs: /tmp/flow_demo/20240129/WXO-DD/observations/swob-ml/20240129/CAMS \n", "2024-01-29 15:00:38,356 [INFO] sarracenia.flowcb.log after_work downloaded ok: /tmp/flow_demo/20240129/WXO-DD/observations/swob-ml/20240129/CAMS/2024-01-29-1743-CAMS-AUTO-minute-swob.xml \n", "2024-01-29 15:00:38,357 [INFO] sarracenia.flow please_stop ok, telling 4 callbacks about it.\n", "2024-01-29 15:00:38,357 [INFO] sarracenia.flow run starting last pass (without gather) through loop for cleanup.\n", "2024-01-29 15:00:38,358 [INFO] sarracenia.flow please_stop ok, telling 4 callbacks about it.\n", "2024-01-29 15:00:38,359 [INFO] sarracenia.flow run on_housekeeping pid: 3567801 subscribe/flow_demo instance: 0\n", "2024-01-29 15:00:38,359 [INFO] sarracenia.flowcb.gather.message on_housekeeping messages: good: 5 bad: 0 bytes: 730 Bytes average: 146 Bytes\n", "2024-01-29 15:00:38,359 [INFO] sarracenia.flowcb.retry on_housekeeping on_housekeeping\n", "2024-01-29 15:00:38,360 [INFO] sarracenia.diskqueue on_housekeeping work_retry_00 on_housekeeping\n", "2024-01-29 15:00:38,361 [INFO] sarracenia.diskqueue on_housekeeping No retry in list\n", "2024-01-29 15:00:38,361 [INFO] sarracenia.diskqueue on_housekeeping on_housekeeping elapse 0.000548\n", "2024-01-29 15:00:38,361 [INFO] sarracenia.diskqueue on_housekeeping post_retry_000 on_housekeeping\n", "2024-01-29 15:00:38,362 [INFO] sarracenia.diskqueue on_housekeeping No retry in list\n", "2024-01-29 15:00:38,362 [INFO] sarracenia.diskqueue on_housekeeping on_housekeeping elapse 0.000741\n", "2024-01-29 15:00:38,363 [INFO] sarracenia.flowcb.housekeeping.resources on_housekeeping Current Memory cpu_times: user=0.76 system=0.17\n", "2024-01-29 15:00:38,363 [INFO] sarracenia.flowcb.housekeeping.resources on_housekeeping Current mem usage: 790.2 MiB, accumulating count (5 or 5/100 so far) before self-setting threshold\n", "2024-01-29 15:00:38,364 [INFO] sarracenia.flowcb.log stats version: 3.00.51rc6, started: a second ago, last_housekeeping: 1.0 seconds ago \n", "2024-01-29 15:00:38,364 [INFO] sarracenia.flowcb.log stats messages received: 5, accepted: 5, rejected: 0 rate accepted: 100.0% or 5.0 m/s\n", "2024-01-29 15:00:38,364 [INFO] sarracenia.flowcb.log stats files transferred: 5 bytes: 45.3 KiB rate: 45.1 KiB/sec\n", "2024-01-29 15:00:38,365 [INFO] sarracenia.flowcb.log stats lag: average: 8200.95, maximum: 8201.25 \n", "2024-01-29 15:00:38,366 [INFO] sarracenia.flowcb.log on_housekeeping housekeeping\n", "2024-01-29 15:00:38,366 [INFO] sarracenia.flow run clean stop from run loop\n", "2024-01-29 15:00:38,385 [DEBUG] amqp collect Closed channel #1\n", "2024-01-29 15:00:38,386 [DEBUG] amqp collect Closed channel #2\n", "2024-01-29 15:00:38,386 [INFO] sarracenia.flowcb.gather.message on_stop closing\n", "2024-01-29 15:00:38,386 [INFO] sarracenia.flow close flow/close completed cleanly pid: 3567801 subscribe/flow_demo instance: 0\n" ] } ], "source": [ "subscriber = sarracenia.flow.subscribe.Subscribe( cfg )\n", "\n", "subscriber.run()" ] }, { "cell_type": "markdown", "id": "passive-biotechnology", "metadata": {}, "source": [ "## Conclusion:\n", "\n", "With the sarracenia.flow class, an async method of operation is supported, it can be customized using flowcb (flow callback) class to introduce specific processing at specific times. It is just like invocation of a single instance from the command line, except all configuration is done within python by setting cfg fields, rather than using the configuration language.\n", "\n", "What is lost vs. using the command line tool: \n", "\n", "* ability to use the configuration language (slightly simpler than assigning values to the cfg object) \n", "* easy running of multiple instances, \n", "* co-ordinated monitoring of the instances (restarts on failure, and a programmable number of subscribers started per configuration.) \n", "* log file management.\n", "\n", "The command line tool provides those additional features." ] } ], "metadata": { "kernelspec": { "display_name": "Python 3 (ipykernel)", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.10.12" } }, "nbformat": 4, "nbformat_minor": 5 }