-
Notifications
You must be signed in to change notification settings - Fork 0
/
event_subprocess_example.py
50 lines (38 loc) · 1.46 KB
/
event_subprocess_example.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
import logging
from concurrent.futures.thread import ThreadPoolExecutor
from camunda.external_task.external_task import ExternalTask
from camunda.external_task.external_task_worker import ExternalTaskWorker
from camunda.utils.log_utils import log_with_context
logger = logging.getLogger(__name__)
default_config = {
"maxTasks": 1,
"lockDuration": 10000,
"asyncResponseTimeout": 30000,
"retries": 3,
"retryTimeout": 5000,
"sleepSeconds": 30,
"isDebug": True,
}
def generic_task_handler(task: ExternalTask):
log_context = {"WORKER_ID": task.get_worker_id(),
"TASK_ID": task.get_task_id(),
"TOPIC": task.get_topic_name()}
log_with_context("executing generic task handler", log_context)
return task.complete()
def main():
configure_logging()
topics = [
("STEP_1", generic_task_handler),
# ("STEP_2", generic_task_handler),
# ("CLEAN_UP", generic_task_handler),
]
executor = ThreadPoolExecutor(max_workers=len(topics))
for index, topic_handler in enumerate(topics):
topic = topic_handler[0]
handler_func = topic_handler[1]
executor.submit(ExternalTaskWorker(worker_id=index, config=default_config).subscribe, topic, handler_func)
def configure_logging():
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[logging.StreamHandler()])
if __name__ == '__main__':
main()