How to configure stream-loader.py to send "Info" messages to RabbitMQ

[Lohrax]

If I have an existing system and I now want to direct withInfo messages out, what variables need to be configured to enable that?

[docktermj]

To have stream-loader.py send "Info" messages to a RabbitMQ queue...

Use proper subcommand

The proper subcommand to instruct stream-loader.py to send Info messages to RabbitMQ is rabbitmq-withinfo.

This can be found by running the following stream-loader.py "help" command. Example:

$ ./stream-loader.py --help

usage: stream-loader.py [-h]
                        {azure-queue,azure-queue-withinfo,kafka,kafka-withinfo,rabbitmq,rabbitmq-withinfo,sleep,sqs,sqs-withinfo,url,version,docker-acceptance-test}
                        ...

Initialize Senzing installation. For subcommand help, run 'stream-loader.py
<subcommand> --help' For more information, see
https://github.com/Senzing/stream-loader

positional arguments:
  {azure-queue,azure-queue-withinfo,kafka,kafka-withinfo,rabbitmq,rabbitmq-withinfo,sleep,sqs,sqs-withinfo,url,version,docker-acceptance-test}
                        Subcommands (SENZING_SUBCOMMAND):
    azure-queue         Read JSON Lines from Azure Message Bus queue.
    azure-queue-withinfo
                        Read JSON Lines from Azure Message Bus queue.
    kafka               Read JSON Lines from Apache Kafka topic.
    kafka-withinfo      Read JSON Lines from Apache Kafka topic. Return info
                        to a queue.
    rabbitmq            Read JSON Lines from RabbitMQ queue.
    rabbitmq-withinfo   Read JSON Lines from RabbitMQ queue. Return info to a
                        queue.
    sleep               Do nothing but sleep. For Docker testing.
    sqs                 Read JSON Lines from AWS SQS queue.
    sqs-withinfo        Read JSON Lines from AWS SQS queue. Return info to a
                        queue.
    url                 Read JSON Lines from URL-addressable file.
    version             Print version of program.
    docker-acceptance-test
                        For Docker acceptance testing.

optional arguments:
  -h, --help            show this help message and exit

Find command options / environment variables

The environment variables used to configure stream-loader.py for sending "info" messages to RabbitMQ are:

• SENZING_RABBITMQ_INFO_EXCHANGE
• SENZING_RABBITMQ_INFO_HOST
• SENZING_RABBITMQ_INFO_PASSWORD
• SENZING_RABBITMQ_INFO_QUEUE
• SENZING_RABBITMQ_INFO_ROUTING_KEY
• SENZING_RABBITMQ_INFO_USERNAME
• SENZING_RABBITMQ_INFO_VIRTUAL_HOST

$ ./stream-loader.py rabbitmq-withinfo \--help

usage: stream-loader.py rabbitmq-withinfo [-h] 
                                          [--rabbitmq-failure-exchange SENZING_RABBITMQ_FAILURE_EXCHANGE]
                                          [--rabbitmq-failure-host SENZING_RABBITMQ_FAILURE_HOST]
                                          [--rabbitmq-failure-password SENZING_RABBITMQ_FAILURE_PASSWORD]
                                          [--rabbitmq-failure-port SENZING_RABBITMQ_FAILURE_PORT]
                                          [--rabbitmq-failure-queue SENZING_RABBITMQ_FAILURE_QUEUE]
                                          [--rabbitmq-failure-routing-key SENZING_RABBITMQ_FAILURE_ROUTING_KEY]
                                          [--rabbitmq-failure-username SENZING_RABBITMQ_FAILURE_USERNAME]
                                          [--rabbitmq-failure-virtual-host SENZING_RABBITMQ_FAILURE_VIRTUAL_HOST]
                                          [--rabbitmq-info-exchange SENZING_RABBITMQ_INFO_EXCHANGE]
                                          [--rabbitmq-info-host SENZING_RABBITMQ_INFO_HOST]
                                          [--rabbitmq-info-password SENZING_RABBITMQ_INFO_PASSWORD]
                                          [--rabbitmq-info-port SENZING_RABBITMQ_INFO_PORT]
                                          [--rabbitmq-info-queue SENZING_RABBITMQ_INFO_QUEUE]
                                          [--rabbitmq-info-routing-key SENZING_RABBITMQ_INFO_ROUTING_KEY]
                                          [--rabbitmq-info-username SENZING_RABBITMQ_INFO_USERNAME]
                                          [--rabbitmq-info-virtual-host SENZING_RABBITMQ_INFO_VIRTUAL_HOST]
                                          [--debug]
                                          [--delay-in-seconds SENZING_DELAY_IN_SECONDS]
                                          [--engine-configuration-json SENZING_ENGINE_CONFIGURATION_JSON]
                                          [--entity-type SENZING_ENTITY_TYPE]
                                          [--license-base64-encoded SENZING_LICENSE_BASE64_ENCODED]
                                          [--monitoring-period-in-seconds SENZING_MONITORING_PERIOD_IN_SECONDS]
                                          [--stream-loader-directive-name SENZING_STREAM_LOADER_DIRECTIVE_NAME]
                                          [--threads-per-process SENZING_THREADS_PER_PROCESS]
                                          [--rabbitmq-exchange SENZING_RABBITMQ_EXCHANGE]
                                          [--rabbitmq-heartbeat-in-seconds SENZING_RABBITMQ_HEARTBEAT_IN_SECONDS]
                                          [--rabbitmq-host SENZING_RABBITMQ_HOST]
                                          [--rabbitmq-password SENZING_RABBITMQ_PASSWORD]
                                          [--rabbitmq-port SENZING_RABBITMQ_PORT]
                                          [--rabbitmq-prefetch-count SENZING_RABBITMQ_PREFETCH_COUNT]
                                          [--rabbitmq-queue SENZING_RABBITMQ_QUEUE]
                                          [--rabbitmq-reconnect-delay-in-seconds SENZING_RABBITMQ_RECONNECT_DELAY_IN_SECONDS]
                                          [--rabbitmq-reconnect-number-of-retries SENZING_RABBITMQ_RECONNECT_NUMBER_OF_RETRIES]
                                          [--rabbitmq-use-existing-entities SENZING_RABBITMQ_USE_EXISTNG_ENTITIES]
                                          [--rabbitmq-username SENZING_RABBITMQ_USERNAME]
                                          [--rabbitmq-virtual-host SENZING_RABBITMQ_VIRTUAL_HOST]

optional arguments:
  -h, --help            show this help message and exit
  --rabbitmq-failure-exchange SENZING_RABBITMQ_FAILURE_EXCHANGE
                        RabbitMQ exchange for failures. Default: SENZING_RABBITMQ_EXCHANGE
  --rabbitmq-failure-host SENZING_RABBITMQ_FAILURE_HOST
                        RabbitMQ host. Default: SENZING_RABBITMQ_HOST
  --rabbitmq-failure-password SENZING_RABBITMQ_FAILURE_PASSWORD
                        RabbitMQ password. Default: SENZING_RABBITMQ_PASSWORD
  --rabbitmq-failure-port SENZING_RABBITMQ_FAILURE_PORT
                        RabbitMQ port. Default: SENZING_RABBITMQ_PORT
  --rabbitmq-failure-queue SENZING_RABBITMQ_FAILURE_QUEUE
                        RabbitMQ queue for failures. Default: senzing-rabbitmq-failure-queue
  --rabbitmq-failure-routing-key SENZING_RABBITMQ_FAILURE_ROUTING_KEY
                        RabbitMQ routing key for failures. Default: senzing.failure
  --rabbitmq-failure-username SENZING_RABBITMQ_FAILURE_USERNAME
                        RabbitMQ username. Default: SENZING_RABBITMQ_USERNAME
  --rabbitmq-failure-virtual-host SENZING_RABBITMQ_FAILURE_VIRTUAL_HOST
                        RabbitMQ virtual host. Default: SENZING_RABBITMQ_VIRTUAL_HOST
  --rabbitmq-info-exchange SENZING_RABBITMQ_INFO_EXCHANGE
                        RabbitMQ exchange for info. Default: SENZING_RABBITMQ_EXCHANGE
  --rabbitmq-info-host SENZING_RABBITMQ_INFO_HOST
                        RabbitMQ host. Default: SENZING_RABBITMQ_HOST
  --rabbitmq-info-password SENZING_RABBITMQ_INFO_PASSWORD
                        RabbitMQ password. Default: SENZING_RABBITMQ_PASSWORD
  --rabbitmq-info-port SENZING_RABBITMQ_INFO_PORT
                        RabbitMQ host. Default: SENZING_RABBITMQ_PORT
  --rabbitmq-info-queue SENZING_RABBITMQ_INFO_QUEUE
                        RabbitMQ queue for info. Default: senzing-rabbitmq-info-queue
  --rabbitmq-info-routing-key SENZING_RABBITMQ_INFO_ROUTING_KEY
                        RabbitMQ routing key for info. Default: senzing-rabbitmq-info-routing-key
  --rabbitmq-info-username SENZING_RABBITMQ_INFO_USERNAME
                        RabbitMQ username. Default: SENZING_RABBITMQ_USERNAME
  --rabbitmq-info-virtual-host SENZING_RABBITMQ_INFO_VIRTUAL_HOST
                        RabbitMQ virtual host. Default: SENZING_RABBITMQ_VIRTUAL_HOST
  --debug               Enable debugging. (SENZING_DEBUG) Default: False
  --delay-in-seconds SENZING_DELAY_IN_SECONDS
                        Delay before processing in seconds. DEFAULT: 0
  --engine-configuration-json SENZING_ENGINE_CONFIGURATION_JSON
                        Advanced Senzing engine configuration. Default: none
  --entity-type SENZING_ENTITY_TYPE
                        Entity type.
  --license-base64-encoded SENZING_LICENSE_BASE64_ENCODED
                        Base64 encoding of a Senzing license. Default: none
  --monitoring-period-in-seconds SENZING_MONITORING_PERIOD_IN_SECONDS
                        Period, in seconds, between monitoring reports. Default: 600
  --stream-loader-directive-name SENZING_STREAM_LOADER_DIRECTIVE_NAME
                        Advanced: The JSON key in messages that direct stream-loader behavior. Default: senzingStreamLoader
  --threads-per-process SENZING_THREADS_PER_PROCESS
                        Number of threads per process. Default: 4
  --rabbitmq-exchange SENZING_RABBITMQ_EXCHANGE
                        RabbitMQ exchange. Default: senzing-rabbitmq-exchange
  --rabbitmq-heartbeat-in-seconds SENZING_RABBITMQ_HEARTBEAT_IN_SECONDS
                        RabbitMQ heartbeat. Default: 60
  --rabbitmq-host SENZING_RABBITMQ_HOST
                        RabbitMQ host. Default: localhost:5672
  --rabbitmq-password SENZING_RABBITMQ_PASSWORD
                        RabbitMQ password. Default: bitnami
  --rabbitmq-port SENZING_RABBITMQ_PORT
                        RabbitMQ port. Default: 5672
  --rabbitmq-prefetch-count SENZING_RABBITMQ_PREFETCH_COUNT
                        RabbitMQ prefetch-count. Default: 50
  --rabbitmq-queue SENZING_RABBITMQ_QUEUE
                        RabbitMQ queue. Default: senzing-rabbitmq-queue
  --rabbitmq-reconnect-delay-in-seconds SENZING_RABBITMQ_RECONNECT_DELAY_IN_SECONDS
                        The time (in seconds) to wait between attempts to reconnect to the RabbitMQ broker. Default: 60
  --rabbitmq-reconnect-number-of-retries SENZING_RABBITMQ_RECONNECT_NUMBER_OF_RETRIES
                        The number of times to try reconnecting a dropped connection to the RabbitMQ broker. Default: 10
  --rabbitmq-use-existing-entities SENZING_RABBITMQ_USE_EXISTNG_ENTITIES
                        Connect to an existing queue using its settings. An error is thrown if the queue does not exist. If False, it will create a queue
                        if one does not exist with the specified name. If it exists, then it will attempt to connect, checking the settings match. Default:
                        True
  --rabbitmq-username SENZING_RABBITMQ_USERNAME
                        RabbitMQ username. Default: user
  --rabbitmq-virtual-host SENZING_RABBITMQ_VIRTUAL_HOST
                        RabbitMQ virtual host. Default: None, which will use the RabbitMQ defined default virtual host

Docker

In docker run ... notation. Example:

sudo docker run \
  :
  --env SENZING_RABBITMQ_HOST=senzing-rabbitmq \
  --env SENZING_RABBITMQ_INFO_QUEUE=senzing-rabbitmq-info-queue \
  --env SENZING_RABBITMQ_PASSWORD=${RABBITMQ_PASSWORD} \
  --env SENZING_RABBITMQ_USERNAME=${RABBITMQ_USERNAME} \
  --env SENZING_SUBCOMMAND: rabbitmq-withinfo \
  :
  senzing/stream-loader

Note that if both the input queue and the info queue use the same RabbitMQ server, SENZING_RABBITMQ_HOST, SENZING_RABBITMQ_PASSWORD, and SENZING_RABBITMQ_USERNAME, will apply to both queues.

Docker-compose

In docker-compose.yaml format, it looks like this.

More

For more, see stream-loader.

[Lohrax]

TY, you guys ROCK