Add remaining eight broker providers and documentation

README.MD

@@ -63,7 +63,7 @@ That gives you intelligent routing without putting an LLM call in the `Publish()
 - **Lifecycle hooks** with `OnBefore`, `OnAfter`, and `OnError`.
 - **Back-pressure policies**: `DropOldest`, `Block`, or `ReturnError`.
 - **Dead-letter queue** with replay support and panic recovery.
-- **Redis Streams, RabbitMQ, NATS JetStream, and Kafka EventStore options** for cross-process delivery.
+- **12 official EventStore broker options**: Redis Streams, RabbitMQ, NATS JetStream, Kafka, SQS, SNS, GCP Pub/Sub, Azure Service Bus, Pulsar, MQTT, PostgreSQL, and NSQ.
 - **Transactions** with local buffering, transport-aware commit, and optional Rules/Cache/Jev event selection.
 - **Scheduling** with absolute/relative timers and optional Rules/Cache/Jev event selection.
 - **Metrics** for published, processed, and failed events.
@@ -762,6 +762,137 @@ Kafka consumer offsets are committed synchronously only after successful event
 handling. With the default hash balancer, the string projection is used as the
 message key, preserving ordering for a projection within a partition.
 
+## Remaining official brokers
+
+All remaining official transports use the same store-first API:
+
+```go
+store := GoEventBus.NewEventStore(
+    &dispatcher,
+    1024,
+    GoEventBus.Block,
+    GoEventBus.WithSQS(sqsConfig), // swap for any provider below
+)
+```
+
+| Broker | EventStore option | Consume semantics |
+|---|---|---|
+| AWS SQS | `WithSQS` | long-poll; delete only after successful handling |
+| AWS SNS | `WithSNS` | publish through SNS; consumption delegates to a real subscriber transport such as SQS |
+| Google Cloud Pub/Sub | `WithGCPPubSub` | Ack on success, Nack on decode/handler failure |
+| Azure Service Bus | `WithAzureServiceBus` | PeekLock; Complete on success, Abandon on failure |
+| Apache Pulsar | `WithPulsar` | durable subscription; Ack on success, Nack on failure |
+| MQTT | `WithMQTT` | QoS 0/1/2; broker ACK behavior follows MQTT/Paho semantics |
+| PostgreSQL | `WithPostgres` | persistent table + `FOR UPDATE SKIP LOCKED`; delete on success, delayed retry on failure |
+| NSQ | `WithNSQ` | channel-based consumption; handler errors are requeued by go-nsq |
+
+Broker-backed events still require string projections, custom `EventCodec`
+implementations are supported by every provider, and `Drain` / `Close`
+closes provider-owned resources. Injected cloud clients/connections remain
+application-owned.
+
+### AWS SQS
+
+```go
+store := GoEventBus.NewEventStore(
+    &dispatcher, 1024, GoEventBus.Block,
+    GoEventBus.WithSQS(GoEventBus.SQSProviderConfig{
+        Client:   sqsClient,
+        QueueURL: queueURL,
+    }),
+)
+```
+
+FIFO queues can set `FIFO`, `MessageGroupID`, and an optional
+`MessageDeduplicationID` callback.
+
+### AWS SNS
+
+```go
+sqsSubscriber, _ := GoEventBus.NewSQSProvider(sqsConfig)
+
+store := GoEventBus.NewEventStore(
+    &dispatcher, 1024, GoEventBus.Block,
+    GoEventBus.WithSNS(GoEventBus.SNSProviderConfig{
+        Client:     snsClient,
+        TopicARN:   topicARN,
+        Subscriber: sqsSubscriber,
+    }),
+)
+```
+
+SNS is not modeled as a queue. `Consume` requires an actual subscriber
+transport such as SQS, matching SNS delivery semantics.
+
+### Google Cloud Pub/Sub
+
+```go
+GoEventBus.WithGCPPubSub(GoEventBus.GCPPubSubProviderConfig{
+    Client:         pubsubClient,
+    Topic:          "events",
+    Subscription:   "billing",
+    EnableOrdering: true,
+})
+```
+
+### Azure Service Bus
+
+```go
+GoEventBus.WithAzureServiceBus(GoEventBus.AzureServiceBusProviderConfig{
+    Client: sbClient,
+    Queue:  "billing",
+})
+```
+
+For topic subscriptions, set `Topic` and `Subscription` instead of `Queue`.
+
+### Apache Pulsar
+
+```go
+GoEventBus.WithPulsar(GoEventBus.PulsarProviderConfig{
+    Client:       pulsarClient,
+    Topic:        "persistent://public/default/events",
+    Subscription: "billing",
+})
+```
+
+### MQTT
+
+```go
+GoEventBus.WithMQTT(GoEventBus.MQTTProviderConfig{
+    Options:     mqttOptions,
+    TopicPrefix: "events/",
+    QoS:         1,
+})
+```
+
+An injected MQTT client remains caller-owned. When `Options` is supplied,
+GoEventBus creates and owns the client. Automatic reconnect and persistent
+sessions are configured through Paho's `ClientOptions`.
+
+### PostgreSQL
+
+```go
+GoEventBus.WithPostgres(GoEventBus.PostgresProviderConfig{
+    ConnectionString: os.Getenv("DATABASE_URL"),
+    Table:            "goeventbus_events",
+})
+```
+
+The PostgreSQL provider uses a durable table-backed queue. Consumers claim one
+row with `FOR UPDATE SKIP LOCKED`; successful processing deletes the row,
+while failures increment the attempt count and delay the next delivery.
+
+### NSQ
+
+```go
+GoEventBus.WithNSQ(GoEventBus.NSQProviderConfig{
+    NSQDAddress: "127.0.0.1:4150",
+    Topic:       "events",
+    Channel:     "billing",
+})
+```
+
 Providers use `JSONCodec` by default. Supply a custom `DecodePayload` when
 consumers need concrete payload types rather than generic JSON values.
 

azure_servicebus_provider.go

@@ -0,0 +1,167 @@
+package GoEventBus
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"sync/atomic"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus"
+)
+
+type AzureServiceBusProviderConfig struct {
+	Client *azservicebus.Client
+
+	// Configure Queue for queue semantics, or Topic for topic publishing.
+	// Topic consumption additionally requires Subscription.
+	Queue        string
+	Topic        string
+	Subscription string
+
+	Codec EventCodec
+}
+
+type AzureServiceBusProvider struct {
+	client   *azservicebus.Client
+	sender   *azservicebus.Sender
+	receiver *azservicebus.Receiver
+	codec    EventCodec
+
+	closeMu sync.Mutex
+	closed  atomic.Bool
+}
+
+func WithAzureServiceBus(config AzureServiceBusProviderConfig) EventStoreOption {
+	return withProviderFactory(func() (Provider, error) { return NewAzureServiceBusProvider(config) })
+}
+
+func NewAzureServiceBusProvider(config AzureServiceBusProviderConfig) (*AzureServiceBusProvider, error) {
+	if config.Client == nil {
+		return nil, errors.New("goeventbus: Azure Service Bus client must not be nil")
+	}
+	if config.Queue == "" && config.Topic == "" {
+		return nil, errors.New("goeventbus: Azure Service Bus queue or topic is required")
+	}
+	if config.Queue != "" && config.Topic != "" {
+		return nil, errors.New("goeventbus: Azure Service Bus queue and topic are mutually exclusive")
+	}
+	if config.Codec == nil {
+		config.Codec = JSONCodec{}
+	}
+
+	entity := config.Queue
+	if entity == "" {
+		entity = config.Topic
+	}
+	sender, err := config.Client.NewSender(entity, nil)
+	if err != nil {
+		return nil, fmt.Errorf("goeventbus: create Azure Service Bus sender: %w", err)
+	}
+
+	var receiver *azservicebus.Receiver
+	if config.Queue != "" {
+		receiver, err = config.Client.NewReceiverForQueue(config.Queue, &azservicebus.ReceiverOptions{
+			ReceiveMode: azservicebus.ReceiveModePeekLock,
+		})
+	} else if config.Subscription != "" {
+		receiver, err = config.Client.NewReceiverForSubscription(config.Topic, config.Subscription, &azservicebus.ReceiverOptions{
+			ReceiveMode: azservicebus.ReceiveModePeekLock,
+		})
+	}
+	if err != nil {
+		_ = sender.Close(context.Background())
+		return nil, fmt.Errorf("goeventbus: create Azure Service Bus receiver: %w", err)
+	}
+
+	return &AzureServiceBusProvider{
+		client:   config.Client,
+		sender:   sender,
+		receiver: receiver,
+		codec:    config.Codec,
+	}, nil
+}
+
+func (p *AzureServiceBusProvider) Publish(ctx context.Context, e Event) error {
+	if p.closed.Load() {
+		return ErrProviderClosed
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	if _, err := remoteProjection(e); err != nil {
+		return err
+	}
+	payload, err := p.codec.Encode(e)
+	if err != nil {
+		return err
+	}
+	if err := p.sender.SendMessage(ctx, &azservicebus.Message{Body: payload}, nil); err != nil {
+		return fmt.Errorf("goeventbus: publish Azure Service Bus event: %w", err)
+	}
+	return nil
+}
+
+func (p *AzureServiceBusProvider) Consume(ctx context.Context, consumer EventConsumer) error {
+	if consumer == nil {
+		return ErrNilConsumer
+	}
+	if p.closed.Load() {
+		return ErrProviderClosed
+	}
+	if p.receiver == nil {
+		return errors.New("goeventbus: Azure Service Bus receiver is not configured")
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
+	for {
+		if p.closed.Load() {
+			return ErrProviderClosed
+		}
+		messages, err := p.receiver.ReceiveMessages(ctx, 1, nil)
+		if err != nil {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+			if p.closed.Load() {
+				return ErrProviderClosed
+			}
+			return fmt.Errorf("goeventbus: receive Azure Service Bus event: %w", err)
+		}
+		for _, msg := range messages {
+			event, err := p.codec.Decode(msg.Body)
+			if err != nil {
+				_ = p.receiver.AbandonMessage(ctx, msg, nil)
+				return fmt.Errorf("goeventbus: decode Azure Service Bus event: %w", err)
+			}
+			if err := consumer(ctx, event); err != nil {
+				_ = p.receiver.AbandonMessage(ctx, msg, nil)
+				return err
+			}
+			if err := p.receiver.CompleteMessage(ctx, msg, nil); err != nil {
+				return fmt.Errorf("goeventbus: complete Azure Service Bus event: %w", err)
+			}
+		}
+	}
+}
+
+func (p *AzureServiceBusProvider) Close() error {
+	if p.closed.Swap(true) {
+		return nil
+	}
+	p.closeMu.Lock()
+	defer p.closeMu.Unlock()
+	ctx := context.Background()
+	var errs []error
+	if p.receiver != nil {
+		errs = append(errs, p.receiver.Close(ctx))
+	}
+	if p.sender != nil {
+		errs = append(errs, p.sender.Close(ctx))
+	}
+	return errors.Join(errs...)
+}
+
+var _ Provider = (*AzureServiceBusProvider)(nil)

broker_options_test.go

@@ -3,6 +3,8 @@ package GoEventBus
 import (
 	"errors"
 	"testing"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
 )
 
 func TestWithNATSJetStreamRejectsNilConnectionLazily(t *testing.T) {
@@ -63,3 +65,54 @@ func TestKafkaRejectsNonStringProjectionBeforeNetwork(t *testing.T) {
 		t.Fatalf("expected ErrStringProjection, got %v", err)
 	}
 }
+
+
+func TestRemainingBrokerOptionsValidateLazily(t *testing.T) {
+	tests := []struct {
+		name string
+		opt  EventStoreOption
+		want string
+	}{
+		{"sqs", WithSQS(SQSProviderConfig{}), "goeventbus: SQS client must not be nil"},
+		{"sns", WithSNS(SNSProviderConfig{}), "goeventbus: SNS client must not be nil"},
+		{"gcp-pubsub", WithGCPPubSub(GCPPubSubProviderConfig{}), "goeventbus: GCP Pub/Sub client must not be nil"},
+		{"azure-service-bus", WithAzureServiceBus(AzureServiceBusProviderConfig{}), "goeventbus: Azure Service Bus client must not be nil"},
+		{"pulsar", WithPulsar(PulsarProviderConfig{}), "goeventbus: Pulsar client must not be nil"},
+		{"mqtt", WithMQTT(MQTTProviderConfig{}), "goeventbus: MQTT client or client options are required"},
+		{"postgres", WithPostgres(PostgresProviderConfig{}), "goeventbus: PostgreSQL pool or connection string is required"},
+		{"nsq", WithNSQ(NSQProviderConfig{}), "goeventbus: NSQ nsqd address must not be empty"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var dispatcher Dispatcher
+			store := NewEventStore(&dispatcher, 8, Block, tt.opt)
+			err := store.PublishToProvider(nil, Event{Projection: "order.created"})
+			if err == nil || err.Error() != tt.want {
+				t.Fatalf("expected %q, got %v", tt.want, err)
+			}
+		})
+	}
+}
+
+func TestPostgresRejectsUnsafeTableName(t *testing.T) {
+	_, err := NewPostgresProvider(PostgresProviderConfig{
+		ConnectionString: "postgres://localhost/test",
+		Table:            "events;drop table users",
+	})
+	if err == nil {
+		t.Fatal("expected invalid table name error")
+	}
+}
+
+func TestMQTTRejectsInvalidQoS(t *testing.T) {
+	_, err := NewMQTTProvider(MQTTProviderConfig{
+		Client: nil,
+		Options: new(mqtt.ClientOptions),
+		Topic: "events",
+		QoS: 3,
+	})
+	if err == nil {
+		t.Fatal("expected invalid MQTT QoS error")
+	}
+}