Publisher Subscriber

PUBSUB_BACKEND=KAFKA# using apache kafka as message broker
PUBSUB_BROKER=localhost:9092
CONSUMER_ID=order-consumer
KAFKA_BATCH_SIZE=1000
KAFKA_BATCH_BYTES=1048576
KAFKA_BATCH_TIMEOUT=300
KAFKA_SASL_MECHANISM=PLAIN
KAFKA_SASL_USERNAME=user
KAFKA_SASL_PASSWORD=password
KAFKA_TLS_CERT_FILE=/path/to/cert.pem
KAFKA_TLS_KEY_FILE=/path/to/key.pem
KAFKA_TLS_CA_CERT_FILE=/path/to/ca.pem
KAFKA_TLS_INSECURE_SKIP_VERIFY=true

#### Docker setup
```shell
docker run --name kafka-1 -p 9092:9092 \
	-e KAFKA_ENABLE_KRAFT=yes \
	-e KAFKA_CFG_PROCESS_ROLES=broker,controller \
	-e KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER \
	-e KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
	-e KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
	-e KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092 \
	-e KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=true \
	-e KAFKA_BROKER_ID=1 \
	-e KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@127.0.0.1:9093 \
	-e ALLOW_PLAINTEXT_LISTENER=yes \
	-e KAFKA_CFG_NODE_ID=1 \
	-v kafka_data:/bitnami \
	bitnami/kafka:3.4

PUBSUB_BACKEND=GOOGLE                   // using Google PubSub as message broker
GOOGLE_PROJECT_ID=project-order         // google projectId where the PubSub is configured
GOOGLE_SUBSCRIPTION_NAME=order-consumer // unique subscription name to identify the subscribing entity

docker pull gcr.io/google.com/cloudsdktool/google-cloud-cli:emulators
docker run --name=gcloud-emulator -d -p 8086:8086 \
	gcr.io/google.com/cloudsdktool/google-cloud-cli:emulators gcloud beta emulators pubsub start --project=test123 \
	--host-port=0.0.0.0:8086

PUBSUB_BACKEND=MQTT            // using MQTT as pubsub
MQTT_HOST=localhost            // broker host URL
MQTT_PORT=1883                 // broker port
MQTT_CLIENT_ID_SUFFIX=test     // suffix to a random generated client-id(uuid v4)

#some additional configs(optional)
MQTT_PROTOCOL=tcp              // protocol for connecting to broker can be tcp, tls, ws or wss
MQTT_MESSAGE_ORDER=true  // config to maintain/retain message publish order, by default this is false
MQTT_USER=username       // authentication username
MQTT_PASSWORD=password   // authentication password

docker run -d \
	--name mqtt \
	-p 8883:8883 \
	-v \
	eclipse-mosquitto:latest <path-to >/mosquitto.conf:/mosquitto/config/mosquitto.conf

PUBSUB_BACKEND=NATS
PUBSUB_BROKER=nats://localhost:4222
NATS_STREAM=mystream
NATS_SUBJECTS=orders.*,shipments.*
NATS_MAX_WAIT=5s
NATS_MAX_PULL_WAIT=500ms
NATS_CONSUMER=my-consumer
NATS_CREDS_FILE=/path/to/creds.json

go get gofr.dev/pkg/gofr/datasources/pubsub/nats

app := gofr.New()

app.AddPubSub(nats.New(nats.Config{
    Server:     "nats://localhost:4222",
    Stream: nats.StreamConfig{
        Stream:   "mystream",
        Subjects: []string{"orders.*", "shipments.*"},
    },
    MaxWait:     5 * time.Second,
    MaxPullWait: 500 * time.Millisecond,
    Consumer:    "my-consumer",
    CredsFile:   "/path/to/creds.json",
}))

docker run -d \
	--name nats \
	-p 4222:4222 \
	-p 8222:8222 \
	-v \
	nats:2.9.16 <path-to >/nats.conf:/nats/config/nats.conf

PUBSUB_BACKEND=REDIS
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_USER=myuser
REDIS_PASSWORD=mypassword
REDIS_DB=0
REDIS_PUBSUB_DB=1
REDIS_TLS_ENABLED=true
REDIS_TLS_CA_CERT=/path/to/ca.pem
REDIS_TLS_CERT=/path/to/cert.pem
REDIS_TLS_KEY=/path/to/key.pem

# Streams mode (default) - requires consumer group
REDIS_STREAMS_CONSUMER_GROUP=my-group
REDIS_STREAMS_CONSUMER_NAME=my-consumer
REDIS_STREAMS_BLOCK_TIMEOUT=5s
REDIS_STREAMS_PEL_RATIO=0.7  # 70% PEL, 30% new messages
REDIS_STREAMS_MAXLEN=1000

# To use PubSub mode instead, set:
# REDIS_PUBSUB_MODE=pubsub

docker run -d \
	--name redis \
	-p 6379:6379 \
	redis:7-alpine

docker run -d \
	--name redis \
	-p 6379:6379 \
	redis:7-alpine redis-server --requirepass mypassword