@@ -0,0 +1,1465 @@
+#!/usr/bin/env python3%0A%0Aimport concurrent.futures%0Aimport threading%0Aimport queue%0Aimport time%0Aimport json%0A%0A%0Adef main():%0A    incoming = queue.Queue()%0A    outgoing = queue.Queue()%0A%0A    receiver = threading.Thread(target=receive, args=(incoming,), daemon=True)%0A    processor = threading.Thread(target=process, args=(incoming, outgoing), daemon=True)%0A    distributor = threading.Thread(target=distribute, args=(outgoing,))%0A%0A    receiver.start()%0A    processor.start()%0A    distributor.start()%0A%0A    distributor.join()%0A%0A%0Adef receive(incoming):%0A    with open('input.json', 'r') as file:%0A        for line in file:%0A            event = json.loads(line)%0A            print('%3E %7B%7D'.format(event%5B'id'%5D))%0A            incoming.put(event)%0A%0A%0Adef process(incoming, outgoing):%0A    with concurrent.futures.ProcessPoolExecutor() as executor:%0A        while True:%0A            event = incoming.get()%0A            future = executor.submit(process_change, event)%0A            outgoing.put(future)%0A%0A%0Adef process_change(event):%0A    start = time.time()%0A    while time.time() - start %3C event%5B'wait'%5D:%0A        time.sleep(0.0001)%0A    print('processed %7B%7D'.format(event%5B'id'%5D))%0A    return event%0A%0A%0Adef distribute(outgoing):%0A    start = time.time()%0A    while True:%0A        event = outgoing.get().result()%0A        print('%3C %7B%7D'.format(event%5B'id'%5D))%0A        if event%5B'id'%5D == 1234:%0A            print('done! elapsed time: %7B%7D'.format(time.time() - start))%0A            break%0A%0A%0Aif __name__ == '__main__':%0A    main()%0A

Create event_processor.py