Fan-Out (SNS → SQS)

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/credentials"
	"github.com/aws/aws-sdk-go-v2/service/sns"
	"github.com/aws/aws-sdk-go-v2/service/sqs"
	sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	url := os.Getenv("KUBEMQ_AWS_URL")
	if url == "" {
		url = "http://localhost:4566"
	}
	cfg, err := config.LoadDefaultConfig(ctx,
		config.WithRegion("us-east-1"),
		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test", "test", "")),
	)
	if err != nil {
		log.Fatalf("load config: %v", err)
	}
	snsClient := sns.NewFromConfig(cfg, func(o *sns.Options) { o.BaseEndpoint = aws.String(url) })
	sqsClient := sqs.NewFromConfig(cfg, func(o *sqs.Options) { o.BaseEndpoint = aws.String(url) })

	topic, _ := snsClient.CreateTopic(ctx, &sns.CreateTopicInput{Name: aws.String("notify")})
	topicArn := aws.ToString(topic.TopicArn)

	// Two target queues, each subscribed (auto-confirmed).
	urlA, arnA := makeQueue(ctx, sqsClient, "q-a")
	urlB, arnB := makeQueue(ctx, sqsClient, "q-b")
	subscribe(ctx, snsClient, topicArn, arnA)
	subscribe(ctx, snsClient, topicArn, arnB)

	// One Publish fans out to both queues with a shared MessageId.
	pub, err := snsClient.Publish(ctx, &sns.PublishInput{
		TopicArn: aws.String(topicArn),
		Message:  aws.String("hello fan-out"),
	})
	if err != nil {
		log.Fatalf("Publish: %v", err)
	}
	msgID := aws.ToString(pub.MessageId)
	fmt.Printf("Publish: MessageId=%s\n", msgID)

	for name, qURL := range map[string]string{"q-a": urlA, "q-b": urlB} {
		env := receiveEnvelope(ctx, sqsClient, qURL)
		if env.MessageId != msgID {
			log.Fatalf("FAIL: %s MessageId=%q != publish %q", name, env.MessageId, msgID)
		}
		fmt.Printf("%s received the publish (MessageId=%s)\n", name, env.MessageId)
	}
}

func makeQueue(ctx context.Context, c *sqs.Client, name string) (url, arn string) {
	created, _ := c.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(name)})
	url = aws.ToString(created.QueueUrl)
	out, _ := c.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{
		QueueUrl:       aws.String(url),
		AttributeNames: []sqstypes.QueueAttributeName{sqstypes.QueueAttributeNameQueueArn},
	})
	return url, out.Attributes[string(sqstypes.QueueAttributeNameQueueArn)]
}

func subscribe(ctx context.Context, c *sns.Client, topicArn, queueArn string) {
	if _, err := c.Subscribe(ctx, &sns.SubscribeInput{
		TopicArn: aws.String(topicArn), Protocol: aws.String("sqs"),
		Endpoint: aws.String(queueArn), ReturnSubscriptionArn: true,
	}); err != nil {
		log.Fatalf("Subscribe: %v", err)
	}
}

type notification struct{ MessageId, Message string }

func receiveEnvelope(ctx context.Context, c *sqs.Client, queueURL string) notification {
	recv, _ := c.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
		QueueUrl: aws.String(queueURL), WaitTimeSeconds: 5, MaxNumberOfMessages: 1,
	})
	var env notification
	_ = json.Unmarshal([]byte(aws.ToString(recv.Messages[0].Body)), &env)
	return env
}

import json
import os

import boto3


def make(service: str):
    return boto3.client(
        service,
        endpoint_url=os.environ.get("KUBEMQ_AWS_URL", "http://localhost:4566"),
        region_name="us-east-1",
        aws_access_key_id="test",
        aws_secret_access_key="test",
    )


def make_queue(sqs, name: str):
    url = sqs.create_queue(QueueName=name)["QueueUrl"]
    arn = sqs.get_queue_attributes(QueueUrl=url, AttributeNames=["QueueArn"])["Attributes"]["QueueArn"]
    return url, arn


def main() -> None:
    sns = make("sns")
    sqs = make("sqs")

    topic_arn = sns.create_topic(Name="notify")["TopicArn"]

    # Two target queues, each subscribed (auto-confirmed).
    url_a, arn_a = make_queue(sqs, "q-a")
    url_b, arn_b = make_queue(sqs, "q-b")
    for arn in (arn_a, arn_b):
        sns.subscribe(TopicArn=topic_arn, Protocol="sqs", Endpoint=arn, ReturnSubscriptionArn=True)

    # One Publish fans out to both queues with a shared MessageId.
    msg_id = sns.publish(TopicArn=topic_arn, Message="hello fan-out")["MessageId"]
    print(f"Publish -> MessageId={msg_id}")

    for name, url in (("q-a", url_a), ("q-b", url_b)):
        recv = sqs.receive_message(QueueUrl=url, WaitTimeSeconds=5, MaxNumberOfMessages=1)
        env = json.loads(recv["Messages"][0]["Body"])
        assert env["MessageId"] == msg_id  # same MessageId across deliveries
        print(f"{name} received the publish (MessageId={env['MessageId']})")


if __name__ == "__main__":
    main()

import java.net.URI;
import java.util.List;
import java.util.Map;

import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.sns.SnsClient;
import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.QueueAttributeName;

public final class Main {
    public static void main(String[] args) {
        String url = System.getenv().getOrDefault("KUBEMQ_AWS_URL", "http://localhost:4566");
        StaticCredentialsProvider creds = StaticCredentialsProvider.create(
                AwsBasicCredentials.create("test", "test"));

        try (SnsClient sns = SnsClient.builder().endpointOverride(URI.create(url))
                .region(Region.US_EAST_1).credentialsProvider(creds).build();
             SqsClient sqs = SqsClient.builder().endpointOverride(URI.create(url))
                .region(Region.US_EAST_1).credentialsProvider(creds).build()) {

            String topicArn = sns.createTopic(b -> b.name("notify")).topicArn();

            // Two target queues, each subscribed (auto-confirmed).
            Map<String, String> queues = Map.of("q-a", "", "q-b", "");
            for (String name : List.of("q-a", "q-b")) {
                String queueUrl = sqs.createQueue(b -> b.queueName(name)).queueUrl();
                String queueArn = sqs.getQueueAttributes(b -> b
                        .queueUrl(queueUrl).attributeNames(QueueAttributeName.QUEUE_ARN))
                        .attributes().get(QueueAttributeName.QUEUE_ARN);
                sns.subscribe(b -> b.topicArn(topicArn).protocol("sqs").endpoint(queueArn)
                        .returnSubscriptionArn(true));
                queues = new java.util.HashMap<>(queues);
                queues.put(name, queueUrl);
            }

            // One Publish fans out to both queues with a shared MessageId.
            String messageId = sns.publish(b -> b.topicArn(topicArn).message("hello fan-out"))
                    .messageId();
            System.out.println("Publish -> MessageId=" + messageId);

            for (Map.Entry<String, String> e : queues.entrySet()) {
                var recv = sqs.receiveMessage(b -> b.queueUrl(e.getValue())
                        .waitTimeSeconds(5).maxNumberOfMessages(1));
                System.out.println(e.getKey() + " received: " + recv.messages().get(0).body());
            }
        }
    }
}

import { SNSClient, CreateTopicCommand, SubscribeCommand, PublishCommand } from "@aws-sdk/client-sns";
import {
  SQSClient,
  CreateQueueCommand,
  GetQueueAttributesCommand,
  ReceiveMessageCommand,
} from "@aws-sdk/client-sqs";

const url = process.env["KUBEMQ_AWS_URL"] ?? "http://localhost:4566";
const opts = { endpoint: url, region: "us-east-1", credentials: { accessKeyId: "test", secretAccessKey: "test" } };
const sns = new SNSClient(opts);
const sqs = new SQSClient(opts);

async function makeQueue(name: string): Promise<{ url: string; arn: string }> {
  const url = (await sqs.send(new CreateQueueCommand({ QueueName: name }))).QueueUrl!;
  const arn = (
    await sqs.send(new GetQueueAttributesCommand({ QueueUrl: url, AttributeNames: ["QueueArn"] }))
  ).Attributes!["QueueArn"]!;
  return { url, arn };
}

async function main(): Promise<void> {
  const topicArn = (await sns.send(new CreateTopicCommand({ Name: "notify" }))).TopicArn!;

  // Two target queues, each subscribed (auto-confirmed).
  const a = await makeQueue("q-a");
  const b = await makeQueue("q-b");
  for (const q of [a, b]) {
    await sns.send(new SubscribeCommand({ TopicArn: topicArn, Protocol: "sqs", Endpoint: q.arn, ReturnSubscriptionArn: true }));
  }

  // One Publish fans out to both queues with a shared MessageId.
  const msgId = (await sns.send(new PublishCommand({ TopicArn: topicArn, Message: "hello fan-out" }))).MessageId;
  console.log(`Publish -> MessageId=${msgId}`);

  for (const [name, q] of [["q-a", a], ["q-b", b]] as const) {
    const recv = await sqs.send(new ReceiveMessageCommand({ QueueUrl: q.url, WaitTimeSeconds: 5, MaxNumberOfMessages: 1 }));
    const env = JSON.parse(recv.Messages![0].Body!);
    console.log(`${name} received the publish (MessageId=${env.MessageId})`);
  }
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

using System.Text.Json;
using Amazon.Runtime;
using Amazon.SimpleNotificationService;
using Amazon.SimpleNotificationService.Model;
using Amazon.SQS;
using Amazon.SQS.Model;

var url = Environment.GetEnvironmentVariable("KUBEMQ_AWS_URL") ?? "http://localhost:4566";
var creds = new BasicAWSCredentials("test", "test");
using var sns = new AmazonSimpleNotificationServiceClient(creds,
    new AmazonSimpleNotificationServiceConfig { ServiceURL = url, AuthenticationRegion = "us-east-1" });
using var sqs = new AmazonSQSClient(creds,
    new AmazonSQSConfig { ServiceURL = url, AuthenticationRegion = "us-east-1" });

var topicArn = (await sns.CreateTopicAsync(new CreateTopicRequest { Name = "notify" })).TopicArn;

async Task<(string url, string arn)> MakeQueue(string name)
{
    var queueUrl = (await sqs.CreateQueueAsync(new CreateQueueRequest { QueueName = name })).QueueUrl;
    var arn = (await sqs.GetQueueAttributesAsync(new GetQueueAttributesRequest
    {
        QueueUrl = queueUrl,
        AttributeNames = ["QueueArn"],
    })).QueueARN;
    return (queueUrl, arn);
}

// Two target queues, each subscribed (auto-confirmed).
var a = await MakeQueue("q-a");
var b = await MakeQueue("q-b");
foreach (var q in new[] { a, b })
{
    await sns.SubscribeAsync(new SubscribeRequest
    {
        TopicArn = topicArn, Protocol = "sqs", Endpoint = q.arn, ReturnSubscriptionArn = true,
    });
}

// One Publish fans out to both queues with a shared MessageId.
var msgId = (await sns.PublishAsync(new PublishRequest { TopicArn = topicArn, Message = "hello fan-out" })).MessageId;
Console.WriteLine($"Publish -> MessageId={msgId}");

foreach (var (name, q) in new[] { ("q-a", a), ("q-b", b) })
{
    var recv = await sqs.ReceiveMessageAsync(new ReceiveMessageRequest
    {
        QueueUrl = q.url, WaitTimeSeconds = 5, MaxNumberOfMessages = 1,
    });
    var env = JsonDocument.Parse(recv.Messages[0].Body).RootElement;
    Console.WriteLine($"{name} received the publish (MessageId={env.GetProperty("MessageId")})");
}

# frozen_string_literal: true
require "aws-sdk-sns"
require "aws-sdk-sqs"
require "json"

Aws::SQS::Client.remove_plugin(Aws::SQS::Plugins::QueueUrls)

url = ENV.fetch("KUBEMQ_AWS_URL", "http://localhost:4566")
opts = { endpoint: url, region: "us-east-1", access_key_id: "test", secret_access_key: "test" }
sns = Aws::SNS::Client.new(opts)
sqs = Aws::SQS::Client.new(opts)

topic_arn = sns.create_topic(name: "notify").topic_arn

# Two target queues, each subscribed (auto-confirmed).
def make_queue(sqs, name)
  url = sqs.create_queue(queue_name: name).queue_url
  arn = sqs.get_queue_attributes(queue_url: url, attribute_names: ["QueueArn"]).attributes["QueueArn"]
  [url, arn]
end

url_a, arn_a = make_queue(sqs, "q-a")
url_b, arn_b = make_queue(sqs, "q-b")
[arn_a, arn_b].each do |arn|
  sns.subscribe(topic_arn: topic_arn, protocol: "sqs", endpoint: arn, return_subscription_arn: true)
end

# One Publish fans out to both queues with a shared MessageId.
msg_id = sns.publish(topic_arn: topic_arn, message: "hello fan-out").message_id
puts "Publish -> MessageId=#{msg_id}"

{ "q-a" => url_a, "q-b" => url_b }.each do |name, queue_url|
  recv = sqs.receive_message(queue_url: queue_url, wait_time_seconds: 5, max_number_of_messages: 1)
  env = JSON.parse(recv.messages.first.body)
  puts "#{name} received the publish (MessageId=#{env['MessageId']})"
end

use aws_config::{BehaviorVersion, Region};
use aws_credential_types::Credentials;
use std::error::Error;

async fn config() -> aws_config::SdkConfig {
    let url = std::env::var("KUBEMQ_AWS_URL").unwrap_or_else(|_| "http://localhost:4566".into());
    aws_config::defaults(BehaviorVersion::latest())
        .region(Region::new("us-east-1"))
        .credentials_provider(Credentials::new("test", "test", None, None, "static"))
        .endpoint_url(url)
        .load()
        .await
}

async fn make_queue(sqs: &aws_sdk_sqs::Client, name: &str) -> (String, String) {
    let url = sqs.create_queue().queue_name(name).send().await.unwrap().queue_url.unwrap();
    let arn = sqs
        .get_queue_attributes()
        .queue_url(&url)
        .attribute_names(aws_sdk_sqs::types::QueueAttributeName::QueueArn)
        .send()
        .await
        .unwrap()
        .attributes
        .and_then(|a| a.get(&aws_sdk_sqs::types::QueueAttributeName::QueueArn).cloned())
        .unwrap();
    (url, arn)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let conf = config().await;
    let sns = aws_sdk_sns::Client::new(&conf);
    let sqs = aws_sdk_sqs::Client::new(&conf);

    let topic_arn = sns.create_topic().name("notify").send().await?.topic_arn.unwrap();

    // Two target queues, each subscribed (auto-confirmed).
    let (url_a, arn_a) = make_queue(&sqs, "q-a").await;
    let (url_b, arn_b) = make_queue(&sqs, "q-b").await;
    for arn in [arn_a, arn_b] {
        sns.subscribe()
            .topic_arn(&topic_arn)
            .protocol("sqs")
            .endpoint(arn)
            .return_subscription_arn(true)
            .send()
            .await?;
    }

    // One Publish fans out to both queues with a shared MessageId.
    let msg_id = sns.publish().topic_arn(&topic_arn).message("hello fan-out").send().await?
        .message_id.unwrap_or_default();
    println!("Publish -> MessageId={msg_id}");

    for (name, url) in [("q-a", url_a), ("q-b", url_b)] {
        let recv = sqs.receive_message().queue_url(&url).wait_time_seconds(5).max_number_of_messages(1)
            .send().await?;
        println!("{name} received: {}", recv.messages().first().and_then(|m| m.body()).unwrap_or_default());
    }
    Ok(())
}