SNS Topics

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 loadCfg(ctx context.Context) aws.Config {
	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)
	}
	return cfg
}

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 := loadCfg(ctx)
	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) })

	// 1. CreateTopic (virtual; idempotent).
	topic, err := snsClient.CreateTopic(ctx, &sns.CreateTopicInput{Name: aws.String("notify")})
	if err != nil {
		log.Fatalf("CreateTopic: %v", err)
	}
	topicArn := aws.ToString(topic.TopicArn)
	fmt.Printf("CreateTopic: %s\n", topicArn)

	// 2. Subscribe an SQS queue — auto-confirmed immediately.
	q, err := sqsClient.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String("notify-q")})
	if err != nil {
		log.Fatalf("CreateQueue: %v", err)
	}
	queueURL := aws.ToString(q.QueueUrl)
	attrs, err := sqsClient.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{
		QueueUrl:       aws.String(queueURL),
		AttributeNames: []sqstypes.QueueAttributeName{sqstypes.QueueAttributeNameQueueArn},
	})
	if err != nil {
		log.Fatalf("GetQueueAttributes: %v", err)
	}
	if _, err := snsClient.Subscribe(ctx, &sns.SubscribeInput{
		TopicArn:              aws.String(topicArn),
		Protocol:              aws.String("sqs"),
		Endpoint:              aws.String(attrs.Attributes[string(sqstypes.QueueAttributeNameQueueArn)]),
		ReturnSubscriptionArn: true,
	}); err != nil {
		log.Fatalf("Subscribe: %v", err)
	}
	fmt.Println("Subscribe(sqs): auto-confirmed")

	// 3. Publish — fans out to the subscribed queue.
	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)
	}
	fmt.Printf("Publish: MessageId=%s\n", aws.ToString(pub.MessageId))

	// The queue receives the SNS Notification envelope (default delivery).
	recv, err := sqsClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
		QueueUrl:        aws.String(queueURL),
		WaitTimeSeconds: 5,
	})
	if err != nil {
		log.Fatalf("ReceiveMessage: %v", err)
	}
	var env struct{ Type, Message, MessageId string }
	_ = json.Unmarshal([]byte(aws.ToString(recv.Messages[0].Body)), &env)
	fmt.Printf("Received: Type=%s Message=%q\n", env.Type, env.Message)
}

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 main() -> None:
    sns = make("sns")
    sqs = make("sqs")

    # 1. CreateTopic (virtual; idempotent).
    topic_arn = sns.create_topic(Name="notify")["TopicArn"]
    print(f"CreateTopic -> {topic_arn}")

    # 2. Subscribe an SQS queue — auto-confirmed immediately.
    queue_url = sqs.create_queue(QueueName="notify-q")["QueueUrl"]
    queue_arn = sqs.get_queue_attributes(QueueUrl=queue_url, AttributeNames=["QueueArn"])[
        "Attributes"
    ]["QueueArn"]
    sns.subscribe(TopicArn=topic_arn, Protocol="sqs", Endpoint=queue_arn, ReturnSubscriptionArn=True)
    print("Subscribe(sqs) -> auto-confirmed")

    # 3. Publish — fans out to the subscribed queue (one MessageId).
    pub = sns.publish(TopicArn=topic_arn, Message="hello fan-out")
    print(f"Publish -> MessageId={pub['MessageId']}")

    # The queue receives the SNS Notification envelope (default delivery).
    recv = sqs.receive_message(QueueUrl=queue_url, WaitTimeSeconds=5)
    env = json.loads(recv["Messages"][0]["Body"])
    print(f"Received -> Type={env['Type']} Message={env['Message']!r}")


if __name__ == "__main__":
    main()

import java.net.URI;

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()) {

            // 1. CreateTopic (virtual; idempotent).
            String topicArn = sns.createTopic(b -> b.name("notify")).topicArn();
            System.out.println("CreateTopic -> " + topicArn);

            // 2. Subscribe an SQS queue — auto-confirmed immediately.
            String queueUrl = sqs.createQueue(b -> b.queueName("notify-q")).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));
            System.out.println("Subscribe(sqs) -> auto-confirmed");

            // 3. Publish — fans out to the subscribed queue (one MessageId).
            String messageId = sns.publish(b -> b.topicArn(topicArn).message("hello fan-out"))
                    .messageId();
            System.out.println("Publish -> MessageId=" + messageId);

            // The queue receives the SNS Notification envelope (default delivery).
            var recv = sqs.receiveMessage(b -> b.queueUrl(queueUrl).waitTimeSeconds(5));
            System.out.println("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" },
};

async function main(): Promise<void> {
  const sns = new SNSClient(opts);
  const sqs = new SQSClient(opts);

  // 1. CreateTopic (virtual; idempotent).
  const topic = await sns.send(new CreateTopicCommand({ Name: "notify" }));
  const topicArn = topic.TopicArn!;
  console.log(`CreateTopic -> ${topicArn}`);

  // 2. Subscribe an SQS queue — auto-confirmed immediately.
  const queueUrl = (await sqs.send(new CreateQueueCommand({ QueueName: "notify-q" }))).QueueUrl!;
  const queueArn = (
    await sqs.send(new GetQueueAttributesCommand({ QueueUrl: queueUrl, AttributeNames: ["QueueArn"] }))
  ).Attributes!["QueueArn"]!;
  await sns.send(
    new SubscribeCommand({ TopicArn: topicArn, Protocol: "sqs", Endpoint: queueArn, ReturnSubscriptionArn: true }),
  );
  console.log("Subscribe(sqs) -> auto-confirmed");

  // 3. Publish — fans out to the subscribed queue (one MessageId).
  const pub = await sns.send(new PublishCommand({ TopicArn: topicArn, Message: "hello fan-out" }));
  console.log(`Publish -> MessageId=${pub.MessageId}`);

  // The queue receives the SNS Notification envelope (default delivery).
  const recv = await sqs.send(new ReceiveMessageCommand({ QueueUrl: queueUrl, WaitTimeSeconds: 5 }));
  const env = JSON.parse(recv.Messages![0].Body!);
  console.log(`Received -> Type=${env.Type} Message="${env.Message}"`);
}

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" });

// 1. CreateTopic (virtual; idempotent).
var topic = await sns.CreateTopicAsync(new CreateTopicRequest { Name = "notify" });
Console.WriteLine($"CreateTopic -> {topic.TopicArn}");

// 2. Subscribe an SQS queue — auto-confirmed immediately.
var queueUrl = (await sqs.CreateQueueAsync(new CreateQueueRequest { QueueName = "notify-q" })).QueueUrl;
var queueArn = (await sqs.GetQueueAttributesAsync(new GetQueueAttributesRequest
{
    QueueUrl = queueUrl,
    AttributeNames = ["QueueArn"],
})).QueueARN;
await sns.SubscribeAsync(new SubscribeRequest
{
    TopicArn = topic.TopicArn,
    Protocol = "sqs",
    Endpoint = queueArn,
    ReturnSubscriptionArn = true,
});
Console.WriteLine("Subscribe(sqs) -> auto-confirmed");

// 3. Publish — fans out to the subscribed queue (one MessageId).
var pub = await sns.PublishAsync(new PublishRequest { TopicArn = topic.TopicArn, Message = "hello fan-out" });
Console.WriteLine($"Publish -> MessageId={pub.MessageId}");

// The queue receives the SNS Notification envelope (default delivery).
var recv = await sqs.ReceiveMessageAsync(new ReceiveMessageRequest { QueueUrl = queueUrl, WaitTimeSeconds = 5 });
var env = JsonDocument.Parse(recv.Messages[0].Body).RootElement;
Console.WriteLine($"Received -> Type={env.GetProperty("Type")} Message={env.GetProperty("Message")}");

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

# Remove the SQS-only path-rewriting plugin so requests reach the connector at "/".
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)

# 1. CreateTopic (virtual; idempotent).
topic_arn = sns.create_topic(name: "notify").topic_arn
puts "CreateTopic -> #{topic_arn}"

# 2. Subscribe an SQS queue — auto-confirmed immediately.
queue_url = sqs.create_queue(queue_name: "notify-q").queue_url
queue_arn = sqs.get_queue_attributes(queue_url: queue_url, attribute_names: ["QueueArn"])
               .attributes["QueueArn"]
sns.subscribe(topic_arn: topic_arn, protocol: "sqs", endpoint: queue_arn, return_subscription_arn: true)
puts "Subscribe(sqs) -> auto-confirmed"

# 3. Publish — fans out to the subscribed queue (one MessageId).
pub = sns.publish(topic_arn: topic_arn, message: "hello fan-out")
puts "Publish -> MessageId=#{pub.message_id}"

# The queue receives the SNS Notification envelope (default delivery).
recv = sqs.receive_message(queue_url: queue_url, wait_time_seconds: 5)
env = JSON.parse(recv.messages.first.body)
puts "Received -> Type=#{env['Type']} Message=#{env['Message'].inspect}"

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
}

#[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);

    // 1. CreateTopic (virtual; idempotent).
    let topic_arn = sns.create_topic().name("notify").send().await?.topic_arn.unwrap();
    println!("CreateTopic -> {topic_arn}");

    // 2. Subscribe an SQS queue — auto-confirmed immediately.
    let queue_url = sqs.create_queue().queue_name("notify-q").send().await?.queue_url.unwrap();
    let queue_arn = sqs
        .get_queue_attributes()
        .queue_url(&queue_url)
        .attribute_names(aws_sdk_sqs::types::QueueAttributeName::QueueArn)
        .send()
        .await?
        .attributes
        .and_then(|a| a.get(&aws_sdk_sqs::types::QueueAttributeName::QueueArn).cloned())
        .ok_or("no queue ARN")?;
    sns.subscribe()
        .topic_arn(&topic_arn)
        .protocol("sqs")
        .endpoint(queue_arn)
        .return_subscription_arn(true)
        .send()
        .await?;
    println!("Subscribe(sqs) -> auto-confirmed");

    // 3. Publish — fans out to the subscribed queue (one MessageId).
    let pub_out = sns.publish().topic_arn(&topic_arn).message("hello fan-out").send().await?;
    println!("Publish -> MessageId={}", pub_out.message_id().unwrap_or("<none>"));

    // The queue receives the SNS Notification envelope (default delivery).
    let recv = sqs.receive_message().queue_url(&queue_url).wait_time_seconds(5).send().await?;
    println!("Received -> {}", recv.messages().first().and_then(|m| m.body()).unwrap_or_default());
    Ok(())
}