# Getting Started (/connectors/aws/tutorials/getting-started) Get a message flowing through the KubeMQ AWS connector in minutes. You enable the connector, point a standard AWS SDK at the connector's endpoint, create an SQS queue, send a message, and receive it back — all over the genuine AWS SQS wire protocol, with no LocalStack and no KubeMQ SDK. The only change versus a real-AWS app is the **endpoint override**. ## Prerequisites [#prerequisites] * A running **kubemq-server** with the AWS connector **enabled** and reachable on **port 4566** (the enable step is below — all six wire-protocol connectors are opt-in). * One of the AWS SDKs below for your language. There is no KubeMQ SDK; you use the native AWS SDK with only its endpoint overridden. * **Dummy AWS credentials and a region.** The connector's default accept-any mode does not verify the signature value, but the SDK must still form a valid SigV4 request — so an access key, secret, and region are required even though their values are not checked. Every example reads a single convenience variable for the connector endpoint, which maps to both the SQS and SNS endpoint overrides: ```bash export KUBEMQ_AWS_URL="http://localhost:4566" # default; mapped to AWS_ENDPOINT_URL_SQS / _SNS # Dummy credentials + region — STILL required even in accept-any mode. export AWS_ACCESS_KEY_ID="test" export AWS_SECRET_ACCESS_KEY="test" export AWS_REGION="us-east-1" # NOT enforced (default ARN segment "kubemq") ``` ## Enable the connector [#enable-the-connector] The AWS connector is **opt-in — disabled by default.** A stock kubemq-server does **not** serve AWS until you turn it on. Enable it with its enable variable: **Enabling the connector opens a new HTTP listener on port 4566.** That port is **not bound until** you set `CONNECTORS_AWS_ENABLE=true`, and it **must differ** from the server's gRPC/REST/HTTP ports — this is exactly why the connector is opt-in rather than on by default. Until you enable it, no AWS endpoint exists and the SDK cannot connect. To turn it off again, set `CONNECTORS_AWS_ENABLE=false` (a config-only rollback; no data migration). See [Configuration](/connectors/aws/concepts/configuration) for the full settings list. ## How it works [#how-it-works] You override only the endpoint URL on a standard AWS SDK client. `CreateQueue("orders")` registers the queue and maps it to the KubeMQ Queue channel `sqs.orders`; `SendMessage` writes to that channel through the message broker; `ReceiveMessage` returns the message plus a receipt handle; `DeleteMessage` acks it off the queue. *The SQS queue `orders` maps to the KubeMQ Queue channel `sqs.orders`; the broker stores the message and the receive returns it with a node-local receipt handle.* ## Steps [#steps] ### Point the SDK at the connector [#point-the-sdk-at-the-connector] Build a standard AWS SDK SQS client and override only the endpoint URL to the connector's endpoint in `KUBEMQ_AWS_URL` (default `http://localhost:4566`). Supply dummy credentials and a region so the SDK forms a valid SigV4 request. The language tabs run the **complete** round-trip from a single program: create the queue, send one message, receive it, and delete it by receipt handle. ```go 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 form a valid SigV4 request; accept-any mode // checks the signature shape, not its 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. client := sqs.NewFromConfig(cfg, func(o *sqs.Options) { o.BaseEndpoint = aws.String(awsURL()) }) // 1. CreateQueue "orders" -> 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) fmt.Printf("queue ready: %s\n", queueURL) // 2. SendMessage. 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) } // 3. ReceiveMessage (long-poll a few seconds for the message). 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)) // 4. DeleteMessage by receipt handle. if _, err := client.DeleteMessage(ctx, &sqs.DeleteMessageInput{ QueueUrl: aws.String(queueURL), ReceiptHandle: msg.ReceiptHandle, }); err != nil { log.Fatalf("DeleteMessage: %v", err) } fmt.Println("deleted; round-trip complete") } ``` ```python import os import boto3 def make_sqs(): # Override ONLY the endpoint URL; dummy credentials form a valid SigV4 # request (accept-any mode checks the signature shape, not its value). 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() # 1. CreateQueue "orders" -> KubeMQ Queue channel "sqs.orders". queue_url = sqs.create_queue(QueueName="orders")["QueueUrl"] print(f"queue ready: {queue_url}") # 2. SendMessage. sqs.send_message(QueueUrl=queue_url, MessageBody="hello from boto3") # 3. ReceiveMessage (long-poll a few seconds). recv = sqs.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=1, WaitTimeSeconds=5) msg = recv["Messages"][0] print(f"received: {msg['Body']!r}") # 4. DeleteMessage by receipt handle. sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=msg["ReceiptHandle"]) print("deleted; round-trip complete") if __name__ == "__main__": main() ``` ```java 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; dummy credentials form a valid // SigV4 request (accept-any mode checks the signature shape). try (SqsClient sqs = SqsClient.builder() .endpointOverride(URI.create(url)) .region(Region.US_EAST_1) .credentialsProvider(StaticCredentialsProvider.create( AwsBasicCredentials.create("test", "test"))) .build()) { // 1. CreateQueue "orders" -> KubeMQ Queue channel "sqs.orders". String queueUrl = sqs.createQueue(b -> b.queueName("orders")).queueUrl(); System.out.println("queue ready: " + queueUrl); // 2. SendMessage. sqs.sendMessage(b -> b.queueUrl(queueUrl).messageBody("hello from the AWS SDK for Java")); // 3. ReceiveMessage (long-poll a few seconds). Message msg = sqs.receiveMessage(b -> b .queueUrl(queueUrl) .maxNumberOfMessages(1) .waitTimeSeconds(5)) .messages().get(0); System.out.printf("received: %s%n", msg.body()); // 4. DeleteMessage by receipt handle. sqs.deleteMessage(b -> b.queueUrl(queueUrl).receiptHandle(msg.receiptHandle())); System.out.println("deleted; round-trip complete"); } } } ``` ```typescript import { SQSClient, CreateQueueCommand, SendMessageCommand, ReceiveMessageCommand, DeleteMessageCommand, } from "@aws-sdk/client-sqs"; // Override ONLY the endpoint; dummy credentials form a valid SigV4 request // (accept-any mode checks the signature shape, not its value). 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 { // 1. CreateQueue "orders" -> KubeMQ Queue channel "sqs.orders". const created = await sqs.send(new CreateQueueCommand({ QueueName: "orders" })); const queueUrl = created.QueueUrl!; console.log(`queue ready: ${queueUrl}`); // 2. SendMessage. await sqs.send(new SendMessageCommand({ QueueUrl: queueUrl, MessageBody: "hello from the AWS SDK v3" })); // 3. ReceiveMessage (long-poll a few seconds). const recv = await sqs.send( new ReceiveMessageCommand({ QueueUrl: queueUrl, MaxNumberOfMessages: 1, WaitTimeSeconds: 5 }), ); const msg = recv.Messages![0]; console.log(`received: ${msg.Body}`); // 4. DeleteMessage by receipt handle. await sqs.send(new DeleteMessageCommand({ QueueUrl: queueUrl, ReceiptHandle: msg.ReceiptHandle! })); console.log("deleted; round-trip complete"); } main().catch((err) => { console.error(err); process.exit(1); }); ``` ```csharp 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 form a valid // SigV4 request (accept-any mode checks the signature shape, not its value). var config = new AmazonSQSConfig { ServiceURL = url, AuthenticationRegion = "us-east-1" }; using var sqs = new AmazonSQSClient(new BasicAWSCredentials("test", "test"), config); // 1. CreateQueue "orders" -> KubeMQ Queue channel "sqs.orders". var created = await sqs.CreateQueueAsync(new CreateQueueRequest { QueueName = "orders" }); var queueUrl = created.QueueUrl; Console.WriteLine($"queue ready: {queueUrl}"); // 2. SendMessage. await sqs.SendMessageAsync(new SendMessageRequest { QueueUrl = queueUrl, MessageBody = "hello from AWSSDK.NET" }); // 3. ReceiveMessage (long-poll a few seconds). var recv = await sqs.ReceiveMessageAsync(new ReceiveMessageRequest { QueueUrl = queueUrl, MaxNumberOfMessages = 1, WaitTimeSeconds = 5, }); var msg = recv.Messages[0]; Console.WriteLine($"received: {msg.Body}"); // 4. DeleteMessage by receipt handle. await sqs.DeleteMessageAsync(new DeleteMessageRequest { QueueUrl = queueUrl, ReceiptHandle = msg.ReceiptHandle }); Console.WriteLine("deleted; round-trip complete"); ``` ```ruby # 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 form a valid SigV4 request # (accept-any mode checks the signature shape, not its value). sqs = Aws::SQS::Client.new( endpoint: url, region: "us-east-1", credentials: Aws::Credentials.new("test", "test") ) # 1. CreateQueue "orders" -> KubeMQ Queue channel "sqs.orders". queue_url = sqs.create_queue(queue_name: "orders").queue_url puts "queue ready: #{queue_url}" # 2. SendMessage. sqs.send_message(queue_url: queue_url, message_body: "hello from aws-sdk-ruby") # 3. ReceiveMessage (long-poll a few seconds). 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}" # 4. DeleteMessage by receipt handle. sqs.delete_message(queue_url: queue_url, receipt_handle: msg.receipt_handle) puts "deleted; round-trip complete" ``` ```rust 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> { // Override ONLY the endpoint; dummy credentials form a valid SigV4 request // (accept-any mode checks the signature shape, not its value). 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); // 1. CreateQueue "orders" -> 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")?; println!("queue ready: {url}"); // 2. SendMessage. sqs.send_message() .queue_url(&url) .message_body("hello from aws-sdk-rust") .send() .await?; // 3. ReceiveMessage (long-poll a few seconds). 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()); // 4. DeleteMessage by receipt handle. let handle = msg.receipt_handle().ok_or("no receipt handle")?; sqs.delete_message().queue_url(&url).receipt_handle(handle).send().await?; println!("deleted; round-trip complete"); Ok(()) } ``` ### Create an SQS queue and send a message [#create-an-sqs-queue-and-send-a-message] `CreateQueue("orders")` registers the queue in the connector's registry and maps it to the KubeMQ Queue channel `sqs.orders`. The returned queue URL is path-style — `{scheme}://{host}/{AccountId}/orders` (the default `AccountId` is `000000000000`). `SendMessage` writes the body to that channel and returns a `MessageId` and an `MD5OfBody` the connector computes exactly like AWS. ### Receive and acknowledge [#receive-and-acknowledge] `ReceiveMessage` returns the message plus a **receipt handle**, and hides the message for its visibility window. `DeleteMessage(receiptHandle)` acks it off the queue. A successful run prints: ```text queue ready: http://localhost:4566/000000000000/orders received: "hello from the AWS SDK" deleted; round-trip complete ``` Receipt handles and in-flight tracking are **node-local** — a handle minted on one node is rejected on another. In a clustered deployment, put the connector behind a **sticky load balancer** (session affinity). See [Connectivity and security](/connectors/aws/how-to/connectivity-and-security). ## Next steps [#next-steps]