AWS (SQS & SNS)

package main

import (
	"context"
	"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/sqs"
)

func awsURL() string {
	if v := os.Getenv("KUBEMQ_AWS_URL"); v != "" {
		return v
	}
	return "http://localhost:4566"
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()

	// Dummy static credentials are required so the SDK forms a valid SigV4
	// request; the connector's accept-any mode does not verify their value.
	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)
	}

	// Override ONLY the endpoint URL — everything else is a normal AWS SDK app.
	client := sqs.NewFromConfig(cfg, func(o *sqs.Options) {
		o.BaseEndpoint = aws.String(awsURL())
	})

	// CreateQueue "orders" maps to KubeMQ Queue channel "sqs.orders".
	created, err := client.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String("orders")})
	if err != nil {
		log.Fatalf("CreateQueue: %v", err)
	}
	queueURL := aws.ToString(created.QueueUrl)

	if _, err := client.SendMessage(ctx, &sqs.SendMessageInput{
		QueueUrl:    aws.String(queueURL),
		MessageBody: aws.String("hello from the AWS SDK"),
	}); err != nil {
		log.Fatalf("SendMessage: %v", err)
	}

	recv, err := client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
		QueueUrl:            aws.String(queueURL),
		MaxNumberOfMessages: 1,
		WaitTimeSeconds:     5,
	})
	if err != nil || len(recv.Messages) != 1 {
		log.Fatalf("ReceiveMessage: %v (got %d)", err, len(recv.Messages))
	}
	msg := recv.Messages[0]
	fmt.Printf("received: %q\n", aws.ToString(msg.Body))

	// DeleteMessage acks the message off the queue by its receipt handle.
	if _, err := client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
		QueueUrl:      aws.String(queueURL),
		ReceiptHandle: msg.ReceiptHandle,
	}); err != nil {
		log.Fatalf("DeleteMessage: %v", err)
	}
}

import os

import boto3

QUEUE = "orders"


def make_sqs():
    # Override ONLY the endpoint URL; dummy credentials are still required so
    # boto3 forms a valid SigV4 request (accept-any mode checks shape only).
    return boto3.client(
        "sqs",
        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:
    sqs = make_sqs()

    # CreateQueue "orders" maps to KubeMQ Queue channel "sqs.orders".
    queue_url = sqs.create_queue(QueueName=QUEUE)["QueueUrl"]

    sqs.send_message(QueueUrl=queue_url, MessageBody="hello from boto3")

    recv = sqs.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=1, WaitTimeSeconds=5)
    msg = recv["Messages"][0]
    print(f"received: {msg['Body']!r}")

    # DeleteMessage acks the message off the queue by its receipt handle.
    sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=msg["ReceiptHandle"])


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.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.Message;

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

        // endpointOverride is the only change versus a real-AWS app; dummy
        // credentials are still required to form a valid SigV4 request.
        try (SqsClient sqs = SqsClient.builder()
                .endpointOverride(URI.create(url))
                .region(Region.US_EAST_1)
                .credentialsProvider(StaticCredentialsProvider.create(
                        AwsBasicCredentials.create("test", "test")))
                .build()) {

            // CreateQueue "orders" maps to KubeMQ Queue channel "sqs.orders".
            String queueUrl = sqs.createQueue(b -> b.queueName("orders")).queueUrl();

            sqs.sendMessage(b -> b.queueUrl(queueUrl).messageBody("hello from the AWS SDK for Java"));

            Message msg = sqs.receiveMessage(b -> b
                    .queueUrl(queueUrl)
                    .maxNumberOfMessages(1)
                    .waitTimeSeconds(5))
                    .messages().get(0);
            System.out.printf("received: %s%n", msg.body());

            // DeleteMessage acks the message off the queue by its receipt handle.
            sqs.deleteMessage(b -> b.queueUrl(queueUrl).receiptHandle(msg.receiptHandle()));
        }
    }
}

import {
  SQSClient,
  CreateQueueCommand,
  SendMessageCommand,
  ReceiveMessageCommand,
  DeleteMessageCommand,
} from "@aws-sdk/client-sqs";

const QUEUE = "orders";

// Override ONLY the endpoint; dummy credentials are still required so the SDK
// forms a valid SigV4 request (accept-any mode checks the signature shape).
const sqs = new SQSClient({
  endpoint: process.env["KUBEMQ_AWS_URL"] ?? "http://localhost:4566",
  region: "us-east-1",
  credentials: { accessKeyId: "test", secretAccessKey: "test" },
});

async function main(): Promise<void> {
  // CreateQueue "orders" maps to KubeMQ Queue channel "sqs.orders".
  const created = await sqs.send(new CreateQueueCommand({ QueueName: QUEUE }));
  const queueUrl = created.QueueUrl!;

  await sqs.send(new SendMessageCommand({ QueueUrl: queueUrl, MessageBody: "hello from the AWS SDK v3" }));

  const recv = await sqs.send(
    new ReceiveMessageCommand({ QueueUrl: queueUrl, MaxNumberOfMessages: 1, WaitTimeSeconds: 5 }),
  );
  const msg = recv.Messages![0];
  console.log(`received: ${msg.Body}`);

  // DeleteMessage acks the message off the queue by its receipt handle.
  await sqs.send(new DeleteMessageCommand({ QueueUrl: queueUrl, ReceiptHandle: msg.ReceiptHandle! }));
}

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

using Amazon.Runtime;
using Amazon.SQS;
using Amazon.SQS.Model;

var url = Environment.GetEnvironmentVariable("KUBEMQ_AWS_URL") ?? "http://localhost:4566";

// ServiceURL carries the full http://host:port; dummy credentials are still
// required to form a valid SigV4 request (accept-any mode checks shape only).
var config = new AmazonSQSConfig { ServiceURL = url, AuthenticationRegion = "us-east-1" };
using var sqs = new AmazonSQSClient(new BasicAWSCredentials("test", "test"), config);

const string queueName = "orders";

// CreateQueue "orders" maps to KubeMQ Queue channel "sqs.orders".
var created = await sqs.CreateQueueAsync(new CreateQueueRequest { QueueName = queueName });
var queueUrl = created.QueueUrl;

await sqs.SendMessageAsync(new SendMessageRequest
{
    QueueUrl = queueUrl,
    MessageBody = "hello from AWSSDK.NET",
});

var recv = await sqs.ReceiveMessageAsync(new ReceiveMessageRequest
{
    QueueUrl = queueUrl,
    MaxNumberOfMessages = 1,
    WaitTimeSeconds = 5,
});
var msg = recv.Messages[0];
Console.WriteLine($"received: {msg.Body}");

// DeleteMessage acks the message off the queue by its receipt handle.
await sqs.DeleteMessageAsync(new DeleteMessageRequest
{
    QueueUrl = queueUrl,
    ReceiptHandle = msg.ReceiptHandle,
});

# frozen_string_literal: true

require "aws-sdk-sqs"

# The Ruby SQS plugin rewrites the request endpoint to the full QueueUrl path,
# which the single-endpoint connector rejects — remove it so requests stay on
# the configured base endpoint (as boto3 and aws-sdk-go-v2 do).
Aws::SQS::Client.remove_plugin(Aws::SQS::Plugins::QueueUrls)

url = ENV.fetch("KUBEMQ_AWS_URL", "http://localhost:4566")

# Override ONLY the endpoint; dummy credentials are still required so the SDK
# forms a valid SigV4 request (accept-any mode checks the signature shape).
sqs = Aws::SQS::Client.new(
  endpoint: url,
  region: "us-east-1",
  credentials: Aws::Credentials.new("test", "test")
)

# CreateQueue "orders" maps to KubeMQ Queue channel "sqs.orders".
queue_url = sqs.create_queue(queue_name: "orders").queue_url

sqs.send_message(queue_url: queue_url, message_body: "hello from aws-sdk-ruby")

recv = sqs.receive_message(queue_url: queue_url, max_number_of_messages: 1, wait_time_seconds: 5)
msg = recv.messages.first
puts "received: #{msg.body.inspect}"

# DeleteMessage acks the message off the queue by its receipt handle.
sqs.delete_message(queue_url: queue_url, receipt_handle: msg.receipt_handle)

use aws_config::BehaviorVersion;
use aws_sdk_sqs::config::Credentials;
use aws_sdk_sqs::config::Region;
use std::error::Error;

fn aws_url() -> String {
    std::env::var("KUBEMQ_AWS_URL").unwrap_or_else(|_| "http://localhost:4566".to_string())
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    // Override ONLY the endpoint; dummy credentials are still required so the
    // SDK forms a valid SigV4 request (accept-any mode checks shape only).
    let creds = Credentials::new("test", "test", None, None, "kubemq-aws");
    let conf = aws_config::defaults(BehaviorVersion::latest())
        .region(Region::new("us-east-1"))
        .credentials_provider(creds)
        .endpoint_url(aws_url())
        .load()
        .await;
    let sqs = aws_sdk_sqs::Client::new(&conf);

    // CreateQueue "orders" maps to KubeMQ Queue channel "sqs.orders".
    sqs.create_queue().queue_name("orders").send().await?;
    let url = sqs
        .get_queue_url()
        .queue_name("orders")
        .send()
        .await?
        .queue_url
        .ok_or("GetQueueUrl returned no URL")?;

    sqs.send_message()
        .queue_url(&url)
        .message_body("hello from aws-sdk-rust")
        .send()
        .await?;

    let received = sqs
        .receive_message()
        .queue_url(&url)
        .max_number_of_messages(1)
        .wait_time_seconds(5)
        .send()
        .await?;
    let msg = &received.messages()[0];
    println!("received: {}", msg.body().unwrap_or_default());

    // DeleteMessage acks the message off the queue by its receipt handle.
    let handle = msg.receipt_handle().ok_or("no receipt handle")?;
    sqs.delete_message().queue_url(&url).receipt_handle(handle).send().await?;
    Ok(())
}