SQS Queues

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 sqsClient(ctx context.Context) *sqs.Client {
	url := "http://localhost:4566"
	if v := os.Getenv("KUBEMQ_AWS_URL"); v != "" {
		url = v
	}
	// Dummy static credentials are mandatory even in accept-any mode: the SDK
	// must form a valid SigV4 request. The region is not enforced.
	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 sqs.NewFromConfig(cfg, func(o *sqs.Options) { o.BaseEndpoint = aws.String(url) })
}

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

	// 1. CreateQueue → registers channel sqs.orders (idempotent on identical attrs).
	created, err := client.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String("orders")})
	if err != nil {
		log.Fatalf("CreateQueue: %v", err)
	}
	queueURL := aws.ToString(created.QueueUrl)
	fmt.Printf("CreateQueue: %s\n", queueURL)

	// 2. SendMessage.
	sent, err := client.SendMessage(ctx, &sqs.SendMessageInput{
		QueueUrl:    aws.String(queueURL),
		MessageBody: aws.String("order #4242 — 3x widget"),
	})
	if err != nil {
		log.Fatalf("SendMessage: %v", err)
	}
	fmt.Printf("SendMessage: MessageId=%s MD5OfBody=%s\n",
		aws.ToString(sent.MessageId), aws.ToString(sent.MD5OfMessageBody))

	// 3. ReceiveMessage (long-poll a few seconds so the message is ready).
	recv, err := client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
		QueueUrl:            aws.String(queueURL),
		MaxNumberOfMessages: 1,
		WaitTimeSeconds:     5,
	})
	if err != nil {
		log.Fatalf("ReceiveMessage: %v", err)
	}
	if len(recv.Messages) != 1 {
		log.Fatalf("expected 1 message, got %d", len(recv.Messages))
	}
	msg := recv.Messages[0]
	fmt.Printf("ReceiveMessage: body=%q\n", aws.ToString(msg.Body))

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

import os

import boto3


def sqs_client():
    url = os.environ.get("KUBEMQ_AWS_URL", "http://localhost:4566")
    # Dummy credentials are mandatory even in accept-any mode (the SDK must form
    # a valid SigV4 request); the region is not enforced.
    return boto3.client(
        "sqs",
        endpoint_url=url,
        region_name="us-east-1",
        aws_access_key_id="test",
        aws_secret_access_key="test",
    )


def main() -> None:
    sqs = sqs_client()

    # 1. CreateQueue → registers channel sqs.orders.
    queue_url = sqs.create_queue(QueueName="orders")["QueueUrl"]
    print(f"CreateQueue   -> {queue_url}")

    # 2. SendMessage.
    body = "hello from boto3"
    send = sqs.send_message(QueueUrl=queue_url, MessageBody=body)
    print(f"SendMessage   -> MessageId={send['MessageId']} MD5OfBody={send['MD5OfMessageBody']}")

    # 3. ReceiveMessage (long-poll so the message is ready).
    recv = sqs.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=1, WaitTimeSeconds=5)
    messages = recv.get("Messages", [])
    if len(messages) != 1:
        raise SystemExit(f"expected 1 message, got {len(messages)}")
    msg = messages[0]
    print(f"ReceiveMessage-> body={msg['Body']!r}")

    # 4. DeleteMessage by receipt handle (acks it off the queue).
    sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=msg["ReceiptHandle"])
    print("DeleteMessage -> acknowledged by receipt handle")


if __name__ == "__main__":
    main()

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

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

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

        // Dummy credentials are mandatory even in accept-any mode (a valid SigV4
        // request must be formed); the region is not enforced.
        try (SqsClient sqs = SqsClient.builder()
                .endpointOverride(URI.create(url))
                .region(Region.US_EAST_1)
                .credentialsProvider(StaticCredentialsProvider.create(
                        AwsBasicCredentials.create("test", "test")))
                .build()) {

            // 1. CreateQueue → registers channel sqs.orders.
            String queueUrl = sqs.createQueue(b -> b.queueName("orders")).queueUrl();
            System.out.println("CreateQueue   -> " + queueUrl);

            // 2. SendMessage.
            SendMessageResponse sent = sqs.sendMessage(b -> b
                    .queueUrl(queueUrl)
                    .messageBody("order #1001"));
            System.out.println("SendMessage   -> MessageId=" + sent.messageId()
                    + " MD5OfBody=" + sent.md5OfMessageBody());

            // 3. ReceiveMessage (long-poll so the message is ready).
            ReceiveMessageResponse recv = sqs.receiveMessage(b -> b
                    .queueUrl(queueUrl)
                    .maxNumberOfMessages(1)
                    .waitTimeSeconds(5));
            List<Message> messages = recv.messages();
            if (messages.size() != 1) {
                throw new IllegalStateException("expected 1 message, got " + messages.size());
            }
            Message msg = messages.get(0);
            System.out.println("ReceiveMessage-> body='" + msg.body() + "'");

            // 4. DeleteMessage by receipt handle (acks it off the queue).
            sqs.deleteMessage(b -> b.queueUrl(queueUrl).receiptHandle(msg.receiptHandle()));
            System.out.println("DeleteMessage -> deleted by receipt handle");
        }
    }
}

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

function sqsClient(): SQSClient {
  // Dummy credentials are mandatory even in accept-any mode (the SDK must form a
  // valid SigV4 request); the region is not enforced.
  return 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> {
  const sqs = sqsClient();

  // 1. CreateQueue → registers channel sqs.orders.
  const created = await sqs.send(new CreateQueueCommand({ QueueName: "orders" }));
  const queueUrl = created.QueueUrl!;
  console.log(`CreateQueue    -> ${queueUrl}`);

  // 2. SendMessage.
  const sent = await sqs.send(
    new SendMessageCommand({ QueueUrl: queueUrl, MessageBody: "order #42: 3x widgets" }),
  );
  console.log(`SendMessage    -> MessageId=${sent.MessageId} MD5OfBody=${sent.MD5OfMessageBody}`);

  // 3. ReceiveMessage (long-poll so the message is ready).
  const recv = await sqs.send(
    new ReceiveMessageCommand({ QueueUrl: queueUrl, MaxNumberOfMessages: 1, WaitTimeSeconds: 5 }),
  );
  const msg = recv.Messages?.[0];
  if (!msg) throw new Error("expected 1 message, got 0");
  console.log(`ReceiveMessage -> body="${msg.Body}"`);

  // 4. DeleteMessage by receipt handle (acks it off the queue).
  await sqs.send(new DeleteMessageCommand({ QueueUrl: queueUrl, ReceiptHandle: msg.ReceiptHandle! }));
  console.log("DeleteMessage  -> acked by receipt handle");
}

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

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

// Dummy credentials are mandatory even in accept-any mode (a valid SigV4 request
// must be formed); the region is not enforced. ServiceURL carries the full
// http://host:port, so leave UseHttp=false to avoid the port being dropped.
var config = new AmazonSQSConfig
{
    ServiceURL = Environment.GetEnvironmentVariable("KUBEMQ_AWS_URL") ?? "http://localhost:4566",
    AuthenticationRegion = "us-east-1",
};
using var sqs = new AmazonSQSClient(new BasicAWSCredentials("test", "test"), config);

// 1. CreateQueue → registers channel sqs.orders.
var created = await sqs.CreateQueueAsync(new CreateQueueRequest { QueueName = "orders" });
Console.WriteLine($"CreateQueue    -> {created.QueueUrl}");

// 2. SendMessage.
var sent = await sqs.SendMessageAsync(new SendMessageRequest
{
    QueueUrl = created.QueueUrl,
    MessageBody = "order #1001",
});
Console.WriteLine($"SendMessage    -> MessageId={sent.MessageId} MD5OfBody={sent.MD5OfMessageBody}");

// 3. ReceiveMessage (long-poll so the message is ready).
var recv = await sqs.ReceiveMessageAsync(new ReceiveMessageRequest
{
    QueueUrl = created.QueueUrl,
    MaxNumberOfMessages = 1,
    WaitTimeSeconds = 5,
});
if (recv.Messages.Count != 1)
    throw new InvalidOperationException($"expected 1 message, got {recv.Messages.Count}");
var msg = recv.Messages[0];
Console.WriteLine($"ReceiveMessage -> body='{msg.Body}'");

// 4. DeleteMessage by receipt handle (acks it off the queue).
await sqs.DeleteMessageAsync(new DeleteMessageRequest
{
    QueueUrl = created.QueueUrl,
    ReceiptHandle = msg.ReceiptHandle,
});
Console.WriteLine("DeleteMessage  -> deleted by receipt handle");

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

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

# Dummy credentials are mandatory even in accept-any mode (the SDK must form a
# valid SigV4 request); the region is not enforced.
sqs = Aws::SQS::Client.new(
  endpoint: url,
  region: "us-east-1",
  access_key_id: "test",
  secret_access_key: "test"
)

# 1. CreateQueue → registers channel sqs.orders.
queue_url = sqs.create_queue(queue_name: "orders").queue_url
puts "CreateQueue   -> #{queue_url}"

# 2. SendMessage.
send_resp = sqs.send_message(queue_url: queue_url, message_body: "order #1138 — 2 widgets")
puts "SendMessage   -> MessageId=#{send_resp.message_id} MD5OfBody=#{send_resp.md5_of_message_body}"

# 3. ReceiveMessage (long-poll so the message is ready).
recv = sqs.receive_message(queue_url: queue_url, max_number_of_messages: 1, wait_time_seconds: 5)
raise "no message received" if recv.messages.empty?

msg = recv.messages.first
puts "ReceiveMessage-> Body=#{msg.body.inspect}"

# 4. DeleteMessage by receipt handle (acks it off the queue).
sqs.delete_message(queue_url: queue_url, receipt_handle: msg.receipt_handle)
puts "DeleteMessage -> ok (acked)"

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

async fn sqs_client() -> aws_sdk_sqs::Client {
    let url = std::env::var("KUBEMQ_AWS_URL").unwrap_or_else(|_| "http://localhost:4566".into());
    // Dummy credentials are mandatory even in accept-any mode (a valid SigV4
    // request must be formed); the region is not enforced.
    let conf = 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;
    aws_sdk_sqs::Client::new(&conf)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let sqs = sqs_client().await;

    // 1. CreateQueue → registers channel sqs.orders.
    let url = sqs
        .create_queue()
        .queue_name("orders")
        .send()
        .await?
        .queue_url
        .ok_or("CreateQueue returned no URL")?;
    println!("CreateQueue   -> {url}");

    // 2. SendMessage.
    let sent = sqs.send_message().queue_url(&url).message_body("order-42").send().await?;
    println!(
        "SendMessage   -> MessageId={} MD5OfBody={}",
        sent.message_id().unwrap_or("<none>"),
        sent.md5_of_message_body().unwrap_or("<none>")
    );

    // 3. ReceiveMessage (long-poll so the message is ready).
    let recv = sqs
        .receive_message()
        .queue_url(&url)
        .max_number_of_messages(1)
        .wait_time_seconds(5)
        .send()
        .await?;
    let messages = recv.messages();
    let msg = messages.first().ok_or("expected 1 message, got 0")?;
    println!("ReceiveMessage-> body='{}'", msg.body().unwrap_or_default());

    // 4. DeleteMessage by receipt handle (acks it off the queue).
    let handle = msg.receipt_handle().ok_or("message has no receipt handle")?;
    sqs.delete_message().queue_url(&url).receipt_handle(handle).send().await?;
    println!("DeleteMessage -> ok (acked)");
    Ok(())
}