Asynchronous Communication

This package implements an asynchronous messaging system with support for multiple providers (AWS SNS/SQS, Google Cloud Pub/Sub, and RabbitMQ), covering message production and consumption, observability, and error handling features.

Important: Before using the asynchronous communication mechanism, all topic and queue structures must have been previously created in the chosen provider.

Configuration

The system automatically detects the configured cloud provider via environment variables.

AWS / GCP (Default)

By default, the COLIBRI_MESSAGING variable is set to CLOUD_DEFAULT.

RabbitMQ

To use RabbitMQ, configure the following variables:

• COLIBRI_MESSAGING: Set to RABBITMQ.
• RABBITMQ_URL: Service access URL (e.g., amqp://guest:guest@localhost:5672/).

Note: When using RabbitMQ, cloud services (SNS/SQS and Pub/Sub) are ignored.

Initialization

To enable messaging features, add initialization in the main.go function:

// Messaging system initialization
messaging.Initialize()

Main Components

1. Publishers (Producers)

Used to send messages to a specific topic.

// Creating a producer
producer := messaging.NewProducer("TOPIC_NAME")

// Message publication
// The second parameter "action" helps identify the message's purpose
err := producer.Publish(ctx, "create", myMessage)

Features:

• Strong typing support.
• Automatic authentication context propagation.
• Message tracking via UUID.
• Integrated monitoring.

2. Consumers

To consume messages, implement the consumer interface and register it with the system.

// Consumer implementation
type MyConsumer struct{}

func (c *MyConsumer) QueueName() string {
    return "QUEUE_NAME"
}

func (c *MyConsumer) Consume(ctx context.Context, msg *ProviderMessage) error {
    var data MyStructure
    if err := msg.DecodeAndValidateMessage(&data); err != nil {
        return err
    }
    // Message processing logic
    return nil
}

// Consumer registration
messaging.NewConsumer(&MyConsumer{})

3. Message Structure (ProviderMessage)

Messages received by the consumer follow the structure below:

type ProviderMessage struct {
    ID            uuid.UUID
    Origin        string
    Action        string
    Message       any
    AuthContext   *security.AuthenticationContext
    CorrelationID string
}

Besides the public fields, ProviderMessage exposes two methods carrying broker-level metadata that is only populated when the message is consumed:

• Attributes() map[string]string: returns the broker-level attributes/headers received with the message (GCP Pub/Sub message attributes, SQS message + system attributes, RabbitMQ headers). It is nil for messages that were just published and is populated only on consumption. Useful, for example, to read GCP's CloudPubSubDeadLetterSourceDeliveryCount on a dead-letter subscription.
• DeliveryAttempt() *int: returns the broker-reported delivery/receive count (GCP DeliveryAttempt, SQS ApproximateReceiveCount, RabbitMQ x-death count), or nil when the broker doesn't expose it.

func (c *MyConsumer) Consume(ctx context.Context, msg *ProviderMessage) error {
    if attempt := msg.DeliveryAttempt(); attempt != nil && *attempt > 3 {
        // Route to poison-message handling, for example
    }

    origin := msg.Attributes()["origin-service"]
    _ = origin

    return nil
}

To test QueueConsumer implementations without a live broker, use messaging.NewConsumerMessage(action, message, attributes, deliveryAttempt), which builds a ProviderMessage as if it had just been received from a broker, including the values returned by Attributes() and DeliveryAttempt().

Advanced Features

1. Multi-Cloud Support

Complete abstraction for:

• AWS: SNS for topics and SQS for queues.
• Google Cloud: Pub/Sub for topics and subscriptions.
• RabbitMQ: Exchanges and Queues.

2. Observability and Resilience

• Native integration with OpenTelemetry.
• Structured logging and message tracking.
• Dead Letter Queue (DLQ) support for failure handling.

Usage Examples

1. Publishing a New User

type User struct {
    Name  string `json:"name"`
    Email string `json:"email"`
}

func PublishNewUser(ctx context.Context, user User) error {
    producer := messaging.NewProducer("USERS_CREATED")
    return producer.Publish(ctx, "create", user)
}

2. Processing with Authentication Context

func (p *MyConsumer) Consume(ctx context.Context, msg *ProviderMessage) error {
    // The authentication context is automatically populated
    // if provided in the original message metadata.
    tenantID := msg.AuthContext.TenantID
    userID := msg.AuthContext.UserID
    
    // Processing with data isolation or specific permissions
    return nil
}

Best Practices

1. Naming: Use descriptive uppercase names with underscores (e.g., ORDERS_PROCESSED), preferably prefixed by the origin service name.
2. Idempotency: Ensure message processing is idempotent to avoid side effects in case of reprocessing.
3. Validation: Always use msg.DecodeAndValidateMessage to ensure the received payload is as expected.
4. Monitoring: Track queue size and error rate to identify bottlenecks or consumption logic failures.

---