Merge branch 'pluralsight'

.gitignore

@@ -1,2 +1,2 @@
 /.idea
-/volume
+volume

.run/Consumer.run.xml

@@ -0,0 +1,15 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="Consumer" type="GoApplicationRunConfiguration" factoryName="Go Application">
+    <module name="sqs_demo" />
+    <working_directory value="$PROJECT_DIR$" />
+    <parameters value="-profile=sqs-tutorial -url=https://sqs.eu-west-3.amazonaws.com" />
+    <envs>
+      <env name="AWS_PROFILE" value="sqs-tutorial" />
+    </envs>
+    <kind value="PACKAGE" />
+    <package value="code.osinet.fr/fgm/sqs_demo/cmd/consumer/" />
+    <directory value="$PROJECT_DIR$" />
+    <filePath value="$PROJECT_DIR$/demo.go" />
+    <method v="2" />
+  </configuration>
+</component>

back/aws.go

@@ -1,30 +0,0 @@
-package main
-
-import (
-	"fmt"
-
-	"github.com/aws/aws-sdk-go-v2/aws"
-)
-
-type endpointResolver struct {
-	region string
-}
-
-func (e endpointResolver) ResolveEndpoint(service, region string, options ...interface{}) (aws.Endpoint, error) {
-	if service != `SQS` {
-		return aws.Endpoint{}, fmt.Errorf("trying to resolve non-SQS service: %s", service)
-	}
-	ep := aws.Endpoint{
-		URL:               "http://localhost:4566/",
-		HostnameImmutable: false,
-		PartitionID:       "000000000000",
-		SigningName:       "",
-		SigningRegion:     e.region,
-		SigningMethod:     "",
-		Source:            0,
-	}
-	if region != "" {
-		ep.URL = fmt.Sprintf("https://sqs.%s.amazonaws.com/", region)
-	}
-	return ep, nil
-}

back/cmd/consumer/main.go

@@ -0,0 +1,56 @@
+package main
+
+import (
+	"context"
+	"io"
+	"log"
+	"os"
+
+	"github.com/fgm/izidic"
+
+	"code.osinet.fr/fgm/sqs_demo/services"
+)
+
+func main() {
+	os.Exit(int(main2(os.Stdout, os.Args[0], os.Args[1:])))
+}
+
+func main2(w io.Writer, name string, args []string) (exitCode byte) {
+	ctx := context.Background()
+	dic := Resolve(w, name, args)
+	lister := dic.MustService(services.SvcLister).(func(ctx context.Context) string)
+	qURL := lister(ctx)
+
+	if false {
+		receiver := dic.MustService(services.SvcReceiver).(func(ctx context.Context, qURL string))
+		receiver(ctx, qURL)
+	}
+	consumer := dic.MustService(services.SvcConsumer).(func(ctx context.Context, qURL string) error)
+	err := consumer(ctx, qURL)
+	if err != nil {
+		log.Printf("error in consumer: %v, aborting", err)
+		exitCode = 1
+	}
+
+	log.Printf("exiting cleanly")
+	return 0
+}
+
+func Resolve(w io.Writer, name string, args []string) *izidic.Container {
+	dic := izidic.New()
+	dic.Store(services.PName, name)
+	dic.Store(services.PArgs, args)
+	dic.Store(services.PWriter, w)
+	dic.Store("handler", services.Handler(HandleDummy))
+
+	dic.Register(services.SvcClient, services.SQSClientService)
+	dic.Register(services.SvcConsumer, services.ConsumerService)
+	dic.Register(services.SvcFlags, services.FlagsService)
+	dic.Register(services.SvcLister, services.ListerService)
+	dic.Register(services.SvcLogger, services.LoggerService)
+	dic.Register(services.SvcReceiver, services.ReceiverService)
+
+	dic.MustService(services.SvcFlags) // Store generated params before freeze.
+	dic.Freeze()
+	return dic
+}

back/cmd/producer/main.go

@@ -0,0 +1,46 @@
+package main
+
+import (
+	"context"
+	"io"
+	"log"
+	"os"
+
+	"github.com/fgm/izidic"
+
+	"code.osinet.fr/fgm/sqs_demo/services"
+)
+
+func main() {
+	os.Exit(int(main2(os.Stdout, os.Args[0], os.Args[1:])))
+}
+
+func main2(w io.Writer, name string, args []string) (exitCode byte) {
+	ctx := context.Background()
+	dic := Resolve(w, name, args)
+	lister := dic.MustService(services.SvcLister).(func(ctx context.Context) string)
+	qURL := lister(ctx)
+
+	producer := dic.MustService(services.SvcProducer).(func(ctx context.Context, qName string))
+	producer(ctx, qURL)
+
+	log.Printf("exiting cleanly")
+	return 0
+}
+
+func Resolve(w io.Writer, name string, args []string) *izidic.Container {
+	dic := izidic.New()
+	dic.Store(services.PName, name)
+	dic.Store(services.PArgs, args)
+	dic.Store(services.PWriter, w)
+
+	dic.Register(services.SvcClient, services.SQSClientService)
+	dic.Register(services.SvcFlags, services.FlagsService)
+	dic.Register(services.SvcLister, services.ListerService)
+	dic.Register(services.SvcLogger, services.LoggerService)
+	dic.Register(services.SvcProducer, services.ProducerService)
+
+	dic.MustService(services.SvcFlags) // Store generated params before freeze.
+	dic.Freeze()
+	return dic
+}

back/cmd/redriver/main.go

@@ -0,0 +1,46 @@
+package main
+
+import (
+	"context"
+	"io"
+	"log"
+	"os"
+
+	"github.com/fgm/izidic"
+
+	services2 "code.osinet.fr/fgm/sqs_demo/back/services"
+)
+
+func main() {
+	os.Exit(int(main2(os.Stdout, os.Args[0], os.Args[1:])))
+}
+
+func main2(w io.Writer, name string, args []string) (exitCode byte) {
+	ctx := context.Background()
+	dic := Resolve(w, name, args)
+	lister := dic.MustService(services2.SvcLister).(func(ctx context.Context) string)
+	qURL := lister(ctx)
+
+	producer := dic.MustService(services2.SvcProducer).(func(ctx context.Context, qName string))
+	producer(ctx, qURL)
+
+	log.Printf("exiting cleanly")
+	return 0
+}
+
+func Resolve(w io.Writer, name string, args []string) *izidic.Container {
+	dic := izidic.New()
+	dic.Store(services2.PName, name)
+	dic.Store(services2.PArgs, args)
+	dic.Store(services2.PWriter, w)
+
+	dic.Register(services2.SvcClient, services2.SQSClientService)
+	dic.Register(services2.SvcFlags, services2.FlagsService)
+	dic.Register(services2.SvcLister, services2.ListerService)
+	dic.Register(services2.SvcLogger, services2.LoggerService)
+	dic.Register(services2.SvcProducer, services2.ProducerService)
+
+	dic.MustService(services2.SvcFlags) // Store generated params before freeze.
+	dic.Freeze()
+	return dic
+}

back/demo.go

@@ -1,132 +0,0 @@
-package main
-
-import (
-	"context"
-	"encoding/json"
-	"flag"
-	"fmt"
-	"io"
-	"log"
-	"os"
-
-	"github.com/aws/aws-sdk-go-v2/config"
-	"github.com/aws/aws-sdk-go-v2/service/sqs"
-	"github.com/fgm/izidic"
-)
-
-func main() {
-	os.Exit(int(main2(os.Stdout, os.Args[0], os.Args[1:])))
-}
-
-func main2(w io.Writer, name string, args []string) byte {
-	ctx := context.Background()
-	dic := resolve(w, name, args)
-	lister := dic.MustService("lister").(func(ctx context.Context) string)
-	receiver := dic.MustService("receiver").(func(ctx context.Context, qURL string))
-	consumer := dic.MustService("consumer").(func(ctx context.Context, qURL string) error)
-	qURL := lister(ctx)
-	receiver(ctx, qURL)
-	err := consumer(ctx, qURL)
-	if err != nil {
-		log.Printf("error in consumer: %v, aborting", err)
-		return 1
-	}
-	log.Printf("exiting cleanly")
-	return 0
-}
-
-func resolve(w io.Writer, name string, args []string) *izidic.Container {
-	dic := izidic.New()
-	dic.Store("name", name)
-	dic.Store("args", args)
-	dic.Store("writer", w)
-	dic.Store("handler", Handler(HandleDummy))
-	dic.Register("flags", flagsService)
-	dic.Register("logger", loggerService)
-	dic.Register("sqs", sqsClientService)
-	dic.Register("lister", listerService)
-	dic.Register("receiver", receiverService)
-	dic.Register("sender", senderService)
-	dic.Register("consumer", consumerService)
-
-	dic.MustService("flags") // Store generated params before freeze.
-	dic.Freeze()
-	return dic
-}
-
-func flagsService(dic *izidic.Container) (any, error) {
-	fs := flag.NewFlagSet(dic.MustParam("name").(string), flag.ContinueOnError)
-	profile := fs.String("profile", "test-profile", "The AWS profile")
-	region := fs.String("region", "eu-west-3", "The AWS region to connect to")
-	qName := fs.String("queue-name", "dummy-queue", "The queue name")
-	sqsURL := fs.String("url", "http://localhost:4566", "The SQS endpoint URL")
-	if err := fs.Parse(dic.MustParam("args").([]string)); err != nil {
-		return nil, fmt.Errorf("cannot obtain CLI args")
-	}
-
-	dic.Store("profile", *profile)
-	dic.Store("region", *region)
-	dic.Store("url", *sqsURL)
-	dic.Store("queue-name", *qName)
-	return fs, nil
-}
-
-// loggerService is an izidic.Service also containing a one-time initialization action.
-func loggerService(dic *izidic.Container) (any, error) {
-	w := dic.MustParam("writer").(io.Writer)
-	log.SetOutput(w) // Support dependency code not taking an injected logger.
-	logger := log.New(w, "", log.LstdFlags)
-	return logger, nil
-}
-
-func sqsClientService(dic *izidic.Container) (any, error) {
-	ctx := context.Background()
-	profile := dic.MustParam("profile").(string)
-	region := dic.MustParam("region").(string)
-	epr := endpointResolver{region: region}
-	cfg, err := config.LoadDefaultConfig(ctx,
-		config.WithRegion(region),
-		config.WithSharedConfigProfile(profile),
-		config.WithEndpointResolverWithOptions(epr),
-	)
-	if err != nil {
-		return nil, fmt.Errorf("failed loading default AWS config: %w", err)
-	}
-
-	client := sqs.NewFromConfig(cfg)
-	return client, nil
-}
-
-func listerService(dic *izidic.Container) (any, error) {
-	cli := dic.MustService("sqs").(*sqs.Client)
-	w := dic.MustParam("writer").(io.Writer)
-	return func(ctx context.Context) string {
-		return lister(ctx, w, cli)
-	}, nil
-}
-
-func receiverService(dic *izidic.Container) (any, error) {
-	cli := dic.MustService("sqs").(*sqs.Client)
-	w := dic.MustParam("writer").(io.Writer)
-	return func(ctx context.Context, qURL string) {
-		receiver(ctx, w, cli, qURL)
-	}, nil
-}
-
-func senderService(dic *izidic.Container) (any, error) {
-	cli := dic.MustService("sqs").(*sqs.Client)
-	w := dic.MustParam("writer").(io.Writer)
-	return func(ctx context.Context, qName string) {
-		sender(ctx, w, cli, qName)
-	}, nil
-}
-
-func consumerService(dic *izidic.Container) (any, error) {
-	cli := dic.MustService("sqs").(*sqs.Client)
-	w := dic.MustParam("writer").(io.Writer)
-	hdl := dic.MustParam("handler").(Handler)
-	enc := json.NewEncoder(w)
-	return func(ctx context.Context, qURL string) error {
-		return consumer(ctx, w, enc, cli, qURL, hdl)
-	}, nil
-}

back/producer.go

@@ -1,12 +0,0 @@
-package main
-
-import (
-	"context"
-	"io"
-
-	"github.com/aws/aws-sdk-go-v2/service/sqs"
-)
-
-func sender(ctx context.Context, w io.Writer, client *sqs.Client, qName string) {
-
-}

back/services/client.go

@@ -0,0 +1,52 @@
+package services
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	"github.com/fgm/izidic"
+)
+
+func SQSClientService(dic *izidic.Container) (any, error) {
+	ctx := context.Background()
+	profile := dic.MustParam(PProfile).(string)
+	region := dic.MustParam(PRegion).(string)
+	epr := endpointResolver{region: region}
+	cfg, err := config.LoadDefaultConfig(ctx,
+		config.WithRegion(region),
+		config.WithSharedConfigProfile(profile),
+		config.WithEndpointResolverWithOptions(epr),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed loading default AWS config: %w", err)
+	}
+
+	client := sqs.NewFromConfig(cfg)
+	return client, nil
+}
+
+type endpointResolver struct {
+	region string
+}
+
+func (e endpointResolver) ResolveEndpoint(service, region string, options ...interface{}) (aws.Endpoint, error) {
+	if service != `SQS` {
+		return aws.Endpoint{}, fmt.Errorf("trying to Resolve non-SQS service: %s", service)
+	}
+	ep := aws.Endpoint{
+		URL:               "http://localhost:4566/",
+		HostnameImmutable: false,
+		PartitionID:       "000000000000",
+		SigningName:       "",
+		SigningRegion:     e.region,
+		SigningMethod:     "",
+		Source:            0,
+	}
+	if region != "" {
+		ep.URL = fmt.Sprintf("https://sqs.%s.amazonaws.com/", region)
+	}
+	return ep, nil
+}

back/consumer.go → back/services/consumer.go

@@ -1,4 +1,4 @@
-package main
+package services
 
 import (
 	"context"
@@ -11,41 +11,39 @@ import (
 	"strconv"
 	"time"
 
-	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/sqs"
 	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
 	"github.com/davecgh/go-spew/spew"
+	"github.com/fgm/izidic"
 	"github.com/google/uuid"
-	"gopkg.in/yaml.v2"
 )
 
-func lister(ctx context.Context, w io.Writer, client *sqs.Client) string {
-	lqo, err := client.ListQueues(ctx, &sqs.ListQueuesInput{
-		MaxResults:      aws.Int32(10),
-		NextToken:       nil,
-		QueueNamePrefix: aws.String(""),
-	})
-	if err != nil {
-		log.Fatalf("failed listing queues: %v", err)
-	}
-	y := yaml.NewEncoder(w)
-	y.Encode(lqo.QueueUrls)
-	return lqo.QueueUrls[0]
+type Event struct {
+	MessageID uuid.UUID `json:"MessageId"`
+	BodySum   string    `json:"MD5OfBody"`
+	AttrSum   string    `json:"MD5MofMessageAttributes"`
+
+	EventAttributes
+	MessageAttributes map[string]types.MessageAttributeValue
+	Body              []byte
 }
 
-func receiver(ctx context.Context, w io.Writer, client *sqs.Client, qURL string) {
-	rmi := sqs.ReceiveMessageInput{
-		QueueUrl:              &qURL,
-		AttributeNames:        []types.QueueAttributeName{"All"},
-		MessageAttributeNames: []string{"All"},
-		VisibilityTimeout:     0,
-		WaitTimeSeconds:       0,
+func (e Event) IsRetryable() bool {
+	maybeRetry, ok := e.MessageAttributes["retry"]
+	if !ok {
+		return false
 	}
-	msg, err := client.ReceiveMessage(ctx, &rmi)
-	if err != nil {
-		log.Fatalf("failed receiving from queue %s: %v", err)
+	if maybeRetry.DataType == nil || *maybeRetry.DataType != "String" || maybeRetry.StringValue == nil {
+		return false
 	}
-	spew.Fdump(w, msg.Messages)
+	return *maybeRetry.StringValue == "1"
+}
+
+type EventAttributes struct {
+	SenderID                         string    `json:"SenderId"`
+	SentTime                         time.Time `json:"SentTimestamp"`
+	ApproximateReceiveCount          int       `json:"ApproximateReceiveCount"`
+	ApproximateFirstReceiveTimestamp time.Time `json:"ApproximateFirstReceiveTimestamp"`
 }
 
 type Handler func(ctx context.Context, enc *json.Encoder, msgID uuid.UUID, sent time.Time, input []byte, meta map[string]types.MessageAttributeValue) error
@@ -59,7 +57,25 @@ func (m message) String() string {
 	return *m.MessageId
 }
 
-func consumer(ctx context.Context, w io.Writer, enc *json.Encoder, client *sqs.Client, qURL string, hdl Handler) error {
+func ConsumerService(dic *izidic.Container) (any, error) {
+	cli := dic.MustService("sqs").(*sqs.Client)
+	w := dic.MustParam(PWriter).(io.Writer)
+	hdl := dic.MustParam(PHandler).(Handler)
+	enc := json.NewEncoder(w)
+	return func(ctx context.Context, qURL string) error {
+		return consumeMessage(ctx, w, enc, cli, qURL, hdl)
+	}, nil
+}
+
+func ReceiverService(dic *izidic.Container) (any, error) {
+	cli := dic.MustService("sqs").(*sqs.Client)
+	w := dic.MustParam(PWriter).(io.Writer)
+	return func(ctx context.Context, qURL string) {
+		receiveMessage(ctx, w, cli, qURL)
+	}, nil
+}
+
+func consumeMessage(ctx context.Context, _ io.Writer, enc *json.Encoder, client *sqs.Client, qURL string, hdl Handler) error {
 	rmi := sqs.ReceiveMessageInput{
 		QueueUrl:              &qURL,
 		AttributeNames:        []types.QueueAttributeName{"All"},
@@ -75,7 +91,7 @@ func consumer(ctx context.Context, w io.Writer, enc *json.Encoder, client *sqs.C
 			return fmt.Errorf("failed receiving from queue: %w, aborting", err)
 		}
 		if len(recv.Messages) == 0 {
-			fmt.Fprintf(w, "No message with %d seconds timeout\n", rmi.WaitTimeSeconds)
+			log.Printf("No message with %d seconds timeout\n", rmi.WaitTimeSeconds)
 			continue
 		}
 		if len(recv.Messages) != 1 {
@@ -84,41 +100,44 @@ func consumer(ctx context.Context, w io.Writer, enc *json.Encoder, client *sqs.C
 		msg := message(recv.Messages[0])
 		evt, err := validateMessage(msg)
 		if err != nil {
-			fmt.Fprintf(w, "invalid message %s: %w, dropping it anyway", msg, err)
+			log.Printf("invalid message %s: %v, dropping it anyway", msg, err)
 		} else {
 			if err := hdl(ctx, enc, evt.MessageID, evt.SentTime, evt.Body, evt.MessageAttributes); err != nil {
-				fmt.Fprintf(w, "Error processing message: %s: %v, dropping it anyway\n", msg, err)
+				log.Printf("message %s failed processing : %v, dropping it anyway\n", msg, err)
 			} else {
-				fmt.Fprintf(w, "Message %s processed successfully\n", msg)
+				log.Printf("message %s processed successfully\n", msg)
 			}
 		}
-		dmi := sqs.DeleteMessageInput{
-			QueueUrl:      &qURL,
-			ReceiptHandle: msg.ReceiptHandle,
-		}
-		_, err = client.DeleteMessage(ctx, &dmi)
-		if err != nil {
-			fmt.Fprintf(w, "Error deleting message %s after successful processing: %v\n", msg, err)
-			continue
+		if evt.IsRetryable() {
+			log.Printf("message %s not deleted, for retry", msg)
+		} else {
+			dmi := sqs.DeleteMessageInput{
+				QueueUrl:      &qURL,
+				ReceiptHandle: msg.ReceiptHandle,
+			}
+			_, err = client.DeleteMessage(ctx, &dmi)
+			if err != nil {
+				log.Printf("Error deleting message %s after successful processing: %v\n", msg, err)
+				continue
+			}
+			log.Printf("message %s deleted after processing\n", msg)
 		}
-		fmt.Fprintf(w, "Deleted processed message %s\n", msg)
 	}
 }
 
-type EventAttributes struct {
-	SenderID                         string    `json:"SenderId"`
-	SentTime                         time.Time `json:"SentTimestamp"`
-	ApproximateReceiveCount          int       `json:"ApproximateReceiveCount"`
-	ApproximateFirstReceiveTimestamp time.Time `json:"ApproximateFirstReceiveTimestamp"`
-}
-type Event struct {
-	MessageID uuid.UUID `json:"MessageId"`
-	BodySum   string    `json:"MD5OfBody"`
-	AttrSum   string    `json:"MD5MofMessageAttributes"`
-
-	EventAttributes
-	MessageAttributes map[string]types.MessageAttributeValue
-	Body              []byte
+func receiveMessage(ctx context.Context, w io.Writer, client *sqs.Client, qURL string) {
+	rmi := sqs.ReceiveMessageInput{
+		QueueUrl:              &qURL,
+		AttributeNames:        []types.QueueAttributeName{"All"},
+		MessageAttributeNames: []string{"All"},
+		VisibilityTimeout:     1,
+		WaitTimeSeconds:       5,
+	}
+	msg, err := client.ReceiveMessage(ctx, &rmi)
+	if err != nil {
+		log.Fatalf("failed receiving from queue %s: %v", err)
+	}
+	spew.Fdump(w, msg.Messages)
 }
 
 func validateMessage(msg message) (*Event, error) {

back/services/lister.go

@@ -0,0 +1,34 @@
+package services
+
+import (
+	"context"
+	"io"
+	"log"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	"github.com/fgm/izidic"
+	"gopkg.in/yaml.v2"
+)
+
+func ListerService(dic *izidic.Container) (any, error) {
+	cli := dic.MustService("sqs").(*sqs.Client)
+	w := dic.MustParam(PWriter).(io.Writer)
+	return func(ctx context.Context) string {
+		return lister(ctx, w, cli)
+	}, nil
+}
+
+func lister(ctx context.Context, w io.Writer, client *sqs.Client) string {
+	lqo, err := client.ListQueues(ctx, &sqs.ListQueuesInput{
+		MaxResults:      aws.Int32(10),
+		NextToken:       nil,
+		QueueNamePrefix: aws.String(""),
+	})
+	if err != nil {
+		log.Fatalf("failed listing queues: %v", err)
+	}
+	y := yaml.NewEncoder(w)
+	y.Encode(lqo.QueueUrls)
+	return lqo.QueueUrls[0]
+}

back/services/producer.go

@@ -0,0 +1,71 @@
+package services
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"log"
+	"math/rand"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
+	"github.com/fgm/izidic"
+)
+
+func ProducerService(dic *izidic.Container) (any, error) {
+	cli := dic.MustService("sqs").(*sqs.Client)
+	w := dic.MustParam(PWriter).(io.Writer)
+	return func(ctx context.Context, qName string) {
+		senderHandler(ctx, w, cli, qName)
+	}, nil
+}
+
+func senderHandler(ctx context.Context, _ io.Writer, client *sqs.Client, qURL string) {
+	pinger := time.NewTicker(2 * time.Second)
+	for range pinger.C {
+		sendOne(ctx, client, qURL)
+	}
+}
+
+func sendOne(ctx context.Context, client *sqs.Client, qURL string) {
+	data := map[string]int{"a": rand.Int()}
+	data["b"] = data["a"] + 1
+	bs, _ := json.Marshal(data)
+	body := string(bs)
+
+	ma := map[string]types.MessageAttributeValue{
+		"x": {
+			DataType:    aws.String("String"),
+			StringValue: aws.String("a string value"),
+		},
+		"y": {
+			DataType:    aws.String("Number"),
+			StringValue: aws.String("42"),
+		},
+		"retry": {
+			DataType:    aws.String("String"),
+			StringValue: aws.String("1"),
+		},
+	}
+
+	smr := sqs.SendMessageInput{
+		MessageBody:             &body,
+		QueueUrl:                &qURL,
+		DelaySeconds:            0,
+		MessageAttributes:       ma,
+		MessageDeduplicationId:  nil,
+		MessageGroupId:          nil,
+		MessageSystemAttributes: nil,
+	}
+	smo, err := client.SendMessage(ctx, &smr)
+	if err != nil {
+		log.Printf("failed producing message: %v", err)
+	} else if smo.MessageId == nil {
+		log.Println("message produced with a nil ID")
+	} else {
+		log.Printf("message %s produced with %#v", *smo.MessageId, data)
+	}
+
+}

back/services/redriver.go

@@ -0,0 +1,58 @@
+package services
+
+import (
+	"context"
+	"io"
+
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	"github.com/fgm/izidic"
+)
+
+type CreateMoveTaskOutput struct {
+	Status                           *string // Running
+	SourceARN                        *string
+	ApproximateNumberOfMessagesMoved int
+}
+
+type Redriver interface {
+	CreateMoveTask()
+	ListMoveTasks() []any
+}
+
+type redriver struct {
+}
+
+func (r redriver) CreateMoveTask() {
+	// TODO implement me
+	// u := url.URL{
+	// 	Scheme:      "https",
+	// 	User:        nil,
+	// 	Host:        "",
+	// 	Path:        "",
+	// 	RawPath:     "",
+	// 	OmitHost:    false,
+	// 	ForceQuery:  false,
+	// 	RawQuery:    "",
+	// 	Fragment:    "",
+	// 	RawFragment: "",
+	// }
+	/*
+		Action=CreateMoveTask
+		&SourceArn=arn%3Aaws%3Asqs%3Aeu-west-3%3A751146239996%3Atest-dlq
+		&TaskName=079228fe-098b-436d-a2f4-cb4d29ebb55a
+		&Version=2012-11-05
+	*/
+}
+
+func (r redriver) ListMoveTasks() []any {
+	// TODO implement me
+	panic("implement me")
+}
+
+func RedriverService(dic *izidic.Container) (any, error) {
+	cli := dic.MustService(SvcClient).(*sqs.Client)
+	w := dic.MustParam(PWriter).(io.Writer)
+	return func(ctx context.Context, qName string) {
+		senderHandler(ctx, w, cli, qName)
+	}, nil
+}

back/services/services.go

@@ -0,0 +1,55 @@
+package services
+
+import (
+	"flag"
+	"fmt"
+	"io"
+	"log"
+
+	"github.com/fgm/izidic"
+)
+
+const (
+	PArgs    = "args"
+	PHandler = "handler"
+	PName    = "name"
+	PProfile = "profile"
+	PQName   = "queue-name"
+	PRegion  = "region"
+	PURL     = "url"
+	PWriter  = "writer"
+
+	SvcClient   = "sqs"
+	SvcConsumer = "consumeMessage"
+	SvcFlags    = "flags"
+	SvcLister   = "lister"
+	SvcLogger   = "logger"
+	SvcProducer = "producer"
+	SvcReceiver = "receiver"
+	SvcRedriver = "redriver"
+)
+
+func FlagsService(dic *izidic.Container) (any, error) {
+	fs := flag.NewFlagSet(dic.MustParam(PName).(string), flag.ContinueOnError)
+	profile := fs.String(PProfile, "test-profile", "The AWS profile")
+	region := fs.String(PRegion, "eu-west-3", "The AWS region to connect to")
+	qName := fs.String(PQName, "dummy-queue", "The queue name")
+	sqsURL := fs.String(PURL, "http://localhost:4566", "The SQS endpoint URL")
+	if err := fs.Parse(dic.MustParam(PArgs).([]string)); err != nil {
+		return nil, fmt.Errorf("cannot obtain CLI args")
+	}
+
+	dic.Store(PProfile, *profile)
+	dic.Store(PRegion, *region)
+	dic.Store(PURL, *sqsURL)
+	dic.Store(PQName, *qName)
+	return fs, nil
+}
+
+// LoggerService is an izidic.Service also containing a one-time initialization action.
+func LoggerService(dic *izidic.Container) (any, error) {
+	w := dic.MustParam(PWriter).(io.Writer)
+	log.SetOutput(w) // Support dependency code not taking an injected logger.
+	logger := log.New(w, "", log.LstdFlags)
+	return logger, nil
+}