#!/usr/bin/env bash set -euo pipefail # --- Config (override via env or flags) --- RMQ_USER="${RMQ_USER:-admin}" RMQ_PASS="${RMQ_PASS:-change_me}" RMQ_HOST="${RMQ_HOST:-localhost}" RMQ_PORT="${RMQ_PORT:-15672}" VHOST="${VHOST:-app}" # Mode A: read from a queue (default) QUEUE="${QUEUE:-queue1}" # Mode B: bind temp queue to an exchange + routing key and consume from it EXCHANGE="${EXCHANGE:-}" # e.g. "amq.topic" or "my-exchange". Leave empty to skip binding mode. ROUTING_KEY="${ROUTING_KEY:-#}" # pattern for topic/direct, default is catch-all "#" # Polling interval when no messages SLEEP_SECS="${SLEEP_SECS:-1}" # Pretty-print with jq if available USE_JQ="${USE_JQ:-auto}" # auto|yes|no USE_COLOR="${USE_COLOR:-yes}" # yes|no -> 'yes' forces color with jq -C # Paths APP_ROOT="/opt/containers/ansible-worker/app" NBPLAY="${APP_ROOT}/bin/nbplay" # --- Helpers --- api() { local method="$1"; shift local path="$1"; shift local data="${1:-}" if [[ -n "$data" ]]; then curl -sS -u "$RMQ_USER:$RMQ_PASS" -H "content-type:application/json" -X "$method" "http://$RMQ_HOST:$RMQ_PORT$path" -d "$data" else curl -sS -u "$RMQ_USER:$RMQ_PASS" -H "content-type:application/json" -X "$method" "http://$RMQ_HOST:$RMQ_PORT$path" fi } has_jq() { command -v jq >/dev/null 2>&1; } pretty() { if [[ "$USE_JQ" == "yes" ]] || { [[ "$USE_JQ" == "auto" ]] && has_jq; }; then if [[ "$USE_COLOR" == "yes" ]]; then jq -C .; else jq .; fi else cat fi } urlenc() { printf '%s' "$1" | sed -e 's, ,%20,g' -e 's,/,%2F,g'; } log() { printf '[consumer] %s\n' "$*"; } warn() { printf '[consumer][WARN] %s\n' "$*" >&2; } err() { printf '[consumer][ERROR] %s\n' "$*" >&2; } # Map task_name -> absolute playbook path map_playbook() { case "$1" in uptime) printf '%s/uptime.yml' "${APP_ROOT}" ;; reset_aths) printf '%s/resetradios.yml' "${APP_ROOT}" ;; *) return 1 ;; esac } # Given a JSON object payload, extract fields and dispatch nbplay dispatch_task() { local json="$1" local device task delay playbook device="$(jq -er '.inscope_device // empty' <<<"$json")" || device="" task="$(jq -er '.task_name // empty' <<<"$json")" || task="" delay="$(jq -r '.task_delay // empty' <<<"$json")" || delay="" if [[ -z "$device" || -z "$task" ]]; then warn "payload missing required keys (inscope_device/task_name). Skipping." return 0 fi if ! playbook="$(map_playbook "$task")"; then warn "unknown task_name='$task' for device='$device' (no-op)." return 0 fi # Optional delay: "none" -> no delay; integer -> sleep seconds if [[ -n "$delay" && "$delay" != "none" ]]; then if [[ "$delay" =~ ^[0-9]+$ ]]; then log "delaying ${delay}s before running '${task}' on '${device}'" sleep "$delay" else warn "task_delay value '$delay' not numeric/'none' (ignoring)." fi fi log "→ Running playbook $(basename "$playbook") on host '${device}'" if ! "${NBPLAY}" "$playbook" "$device"; then err "playbook failed (task='${task}', device='${device}')" # do not exit; keep consuming fi } cleanup_queue="" cleanup() { if [[ -n "$cleanup_queue" ]]; then log "Cleaning up temp queue: $cleanup_queue" api DELETE "/api/queues/$(urlenc "$VHOST")/$cleanup_queue" >/dev/null || true fi } trap cleanup EXIT # --- Setup: decide queue source --- if [[ -n "$EXCHANGE" ]]; then cleanup_queue="tmp.$(hostname -s).$$.$(date +%s)" log "Declaring temp queue: $cleanup_queue (exclusive, auto-delete)" api PUT "/api/queues/$(urlenc "$VHOST")/$cleanup_queue" '{ "auto_delete": true, "durable": false, "arguments": {}, "exclusive": true }' >/dev/null log "Binding temp queue to exchange '$EXCHANGE' with routing key '$ROUTING_KEY'" api POST "/api/bindings/$(urlenc "$VHOST")/e/$EXCHANGE/q/$cleanup_queue" "{ \"routing_key\": \"$ROUTING_KEY\", \"arguments\": {} }" >/dev/null QUEUE="$cleanup_queue" log "Consuming from bound temp queue: $QUEUE" else log "Consuming directly from queue: $QUEUE (vhost: $VHOST)" fi log "Press Ctrl+C to stop." # --- Consume loop --- while :; do RESP="$(api POST "/api/queues/$(urlenc "$VHOST")/$QUEUE/get" '{ "count": 1, "ackmode": "ack_requeue_false", "encoding": "auto", "truncate": 1000000 }')" # Empty array => no messages if [[ "$RESP" == "[]" || -z "$RESP" ]]; then sleep "$SLEEP_SECS" continue fi # Process one-by-one (the API returns an array) if has_jq && { [[ "$USE_JQ" == "yes" ]] || [[ "$USE_JQ" == "auto" ]]; }; then echo "$RESP" | jq -c '.[]' | while read -r item; do payload=$(printf '%s' "$item" | jq -r '.payload') rk=$(printf '%s' "$item" | jq -r '.routing_key') ex=$(printf '%s' "$item" | jq -r '.exchange') echo "-----" echo "exchange: ${ex:-\"\"}" echo "routing_key: $rk" echo "payload:" # Decode + pretty print + DISPATCH if jq -e . >/dev/null 2>&1 <<<"$payload"; then ptype="$(printf '%s' "$payload" | jq -r 'type')" if [[ "$ptype" == "string" ]]; then decoded="$(printf '%s' "$payload" | jq -r .)" if jq -e . >/dev/null 2>&1 <<<"$decoded"; then printf '%s' "$decoded" | pretty dispatch_task "$decoded" else printf '%s\n' "$decoded" # not JSON -> no dispatch fi else printf '%s' "$payload" | pretty dispatch_task "$payload" fi else printf '%s\n' "$payload" # not JSON -> no dispatch fi done else # Minimal fallback: print raw response only echo "$RESP" fi done