nats live stream fuzzy finder + republish

natsre.sh

#!/usr/bin/env bash
set -euo pipefail

# Notes:
# default macos bash version does not support coproc. Update bash version
# using brew install bash
# requires nats cli
# requires fzf

stream=${STREAM:-router_system}

parse_nats_output() {
  while IFS= read -r line1 && \
    IFS= read -r json_data && \
    IFS= read -r _ && \
    IFS= read -r _; do

    [[ "$line1" =~ stream:\ ([^[:space:]]+) ]] && stream="${BASH_REMATCH[1]}"
    [[ "$line1" =~ seq\ ([0-9]+) ]] && seq="${BASH_REMATCH[1]}"
    [[ "$line1" =~ subject:\ ([^[:space:]]+) ]] && subject="${BASH_REMATCH[1]}"
    [[ "$line1" =~ time:\ ([^[:space:]]+) ]] && time="${BASH_REMATCH[1]}"

    echo "{\"stream\": \"$stream\", \"seq\": $seq, \"subject\": \"$subject\", \"time\": \"$time\", \"data\": $json_data}"
  done
}

cleanup() {
  if [[ -v NATS_PROC ]]; then
    exec {NATS_PROC[0]}<&- 
    exec {NATS_PROC[1]}>&-
  fi
  
  if [[ -v NATS_PROC_PID ]]; then
    nats_pid=$(pgrep -P "$NATS_PROC_PID" nats || true)
    if [[ -n "$nats_pid" ]]; then
      kill "$nats_pid" 2>/dev/null || true
      wait "$nats_pid" 2>/dev/null || true
    fi
    kill "$NATS_PROC_PID" 2>/dev/null || true
    wait "$NATS_PROC_PID" 2>/dev/null || true
  fi
}
trap cleanup EXIT SIGINT SIGTERM

coproc NATS_PROC { nats sub --stream "$stream" "$@" 2>/dev/null; }

chosen=$(fzf --preview 'echo {} | jq --color-output .' < <(parse_nats_output <&"${NATS_PROC[0]}"))
fzf_status=$?

if [ -z "$chosen" ]; then
  exit $fzf_status
fi
subject=$(echo "$chosen" | jq -r '.subject')

echo "> Subject: '$subject'"
echo "> Data:"
echo "$chosen" | jq --color-output

read -r -p "Republish above message? (y/N) " response
if [[ "$response" =~ ^[Yy]$ ]]; then
  stream=$(echo "$chosen" | jq -r '.stream')
  data=$(echo "$chosen" | jq -r -c '.data')
  echo "Republishing..."
  nats pub --jetstream "$subject" "$data"
fi