def publish_processed_network_packets(<EOL>name="<STR_LIT>",<EOL>task_queue=None,<EOL>result_queue=None,<EOL>need_response=False,<EOL>shutdown_msg="<STR_LIT>"):
<EOL>filter_key = ev("<STR_LIT>",<EOL>INCLUDED_IGNORE_KEY)<EOL>forward_host = ev("<STR_LIT>", "<STR_LIT:127.0.0.1>")<EOL>forward_port = int(ev("<STR_LIT>", "<STR_LIT>"))<EOL>include_filter_key = ev("<STR_LIT>", "<STR_LIT>")<EOL>if not include_filter_key and filter_key:<EOL><INDENT>include_filter_key = filter_key<EOL><DEDENT>filter_keys = [filter_key]<EOL>log.info(("<STR_LIT>"<EOL>"<STR_LIT>"<EOL>"<STR_LIT>")<EOL>.format(name,<EOL>forward_host,<EOL>forward_port,<EOL>include_filter_key,<EOL>filter_key))<EOL>forward_skt = None<EOL>not_done = True<EOL>while not_done:<EOL><INDENT>if not forward_skt:<EOL><INDENT>forward_skt = connect_forwarder(<EOL>forward_host=forward_host,<EOL>forward_port=forward_port)<EOL><DEDENT>next_task = task_queue.get()<EOL>if next_task:<EOL><INDENT>if str(next_task) == shutdown_msg:<EOL><INDENT>log.info(("<STR_LIT>"<EOL>"<STR_LIT>")<EOL>.format(name,<EOL>next_task))<EOL>task_queue.task_done()<EOL>break<EOL><DEDENT>try:<EOL><INDENT>log.debug(("<STR_LIT>")<EOL>.format(name))<EOL>source = next_task.source<EOL>packet = next_task.payload<EOL>if not packet:<EOL><INDENT>log.error(("<STR_LIT>"<EOL>"<STR_LIT>")<EOL>.format(name,<EOL>next_task))<EOL>break<EOL><DEDENT>log.debug(("<STR_LIT>")<EOL>.format(name,<EOL>source))<EOL>network_data = parse_network_data(<EOL>data_packet=packet,<EOL>include_filter_key=include_filter_key,<EOL>filter_keys=filter_keys)<EOL>if network_data["<STR_LIT:status>"] == VALID:<EOL><INDENT>if network_data["<STR_LIT>"] == TCPor network_data["<STR_LIT>"] == UDPor network_data["<STR_LIT>"] == ARPor network_data["<STR_LIT>"] == ICMP:<EOL><INDENT>log.info(("<STR_LIT>"<EOL>"<STR_LIT>")<EOL>.format(name,<EOL>network_data["<STR_LIT:id>"],<EOL>network_data["<STR_LIT>"],<EOL>network_data["<STR_LIT>"]))<EOL>if not forward_skt:<EOL><INDENT>forward_skt = connect_forwarder(<EOL>forward_host=forward_host,<EOL>forward_port=forward_port)<EOL><DEDENT>if forward_skt:<EOL><INDENT>if network_data["<STR_LIT>"]:<EOL><INDENT>sent = False<EOL>while not sent:<EOL><INDENT>try:<EOL><INDENT>log.info("<STR_LIT>".format(<EOL>network_data["<STR_LIT>"]))<EOL>send_msg(<EOL>forward_skt,<EOL>network_data["<STR_LIT>"]<EOL>.encode("<STR_LIT:utf-8>"))<EOL>sent = True<EOL><DEDENT>except Exception as e:<EOL><INDENT>sent = False<EOL>time.sleep(<NUM_LIT:0.5>)<EOL>try:<EOL><INDENT>forward_skt.close()<EOL>forward_skt = None<EOL><DEDENT>except Exception as w:<EOL><INDENT>forward_skt = None<EOL><DEDENT>forward_skt = connect_forwarder(<EOL>forward_host=forward_host,<EOL>forward_port=forward_port)<EOL><DEDENT><DEDENT>log.info("<STR_LIT>".format(<EOL>network_data["<STR_LIT>"]))<EOL>if need_response:<EOL><INDENT>log.info("<STR_LIT>")<EOL>cdr_res = forward_skt.recv(<NUM_LIT>)<EOL>log.info(("<STR_LIT>")<EOL>.format(cdr_res))<EOL><DEDENT><DEDENT>else:<EOL><INDENT>log.info(("<STR_LIT>"<EOL>"<STR_LIT>")<EOL>.format(<EOL>name,<EOL>network_data["<STR_LIT>"],<EOL>network_data["<STR_LIT>"],<EOL>network_data["<STR_LIT:status>"]))<EOL><DEDENT><DEDENT><DEDENT>else:<EOL><INDENT>log.info(("<STR_LIT>"<EOL>"<STR_LIT>")<EOL>.format(name,<EOL>network_data["<STR_LIT:id>"],<EOL>network_data["<STR_LIT>"],<EOL>network_data["<STR_LIT:status>"]))<EOL><DEDENT><DEDENT>elif network_data["<STR_LIT:status>"] == FILTERED:<EOL><INDENT>log.info(("<STR_LIT>")<EOL>.format(name,<EOL>network_data["<STR_LIT>"],<EOL>network_data["<STR_LIT:status>"]))<EOL><DEDENT>else:<EOL><INDENT>if network_data["<STR_LIT:status>"] == INVALID:<EOL><INDENT>log.info(("<STR_LIT>"<EOL>"<STR_LIT>")<EOL>.format(name,<EOL>network_data["<STR_LIT:id>"],<EOL>network_data["<STR_LIT>"],<EOL>network_data["<STR_LIT:error>"],<EOL>network_data["<STR_LIT:status>"]))<EOL><DEDENT>else:<EOL><INDENT>log.info(("<STR_LIT>"<EOL>"<STR_LIT>")<EOL>.format(name,<EOL>network_data["<STR_LIT:id>"],<EOL>network_data["<STR_LIT>"],<EOL>network_data["<STR_LIT:error>"],<EOL>network_data["<STR_LIT:status>"]))<EOL><DEDENT><DEDENT><DEDENT>except KeyboardInterrupt as k:<EOL><INDENT>log.info(("<STR_LIT>")<EOL>.format(name))<EOL>break<EOL><DEDENT>except Exception as e:<EOL><INDENT>log.error(("<STR_LIT>"<EOL>"<STR_LIT>")<EOL>.format(name,<EOL>e))<EOL>break<EOL><DEDENT><DEDENT>log.info(("<STR_LIT>")<EOL>.format(name, next_task))<EOL>task_queue.task_done()<EOL>if need_response:<EOL><INDENT>answer = "<STR_LIT>".format(next_task())<EOL>result_queue.put(answer)<EOL><DEDENT><DEDENT>if forward_skt:<EOL><INDENT>try:<EOL><INDENT>forward_skt.close()<EOL>log.info("<STR_LIT>")<EOL>forward_skt = None<EOL><DEDENT>except Exception:<EOL><INDENT>log.info("<STR_LIT>")<EOL><DEDENT><DEDENT>log.info("<STR_LIT>".format(name))<EOL>return<EOL>
# Redis/RabbitMQ/SQS messaging endpoints for pub-sub
routing_key = ev("PUBLISH_EXCHANGE",
                 "reporting.accounts")
queue_name = ev("PUBLISH_QUEUE",
                "reporting.accounts")
auth_url = ev("PUB_BROKER_URL",
              "redis://localhost:6379/15")
serializer = "json"