go__sqs_demo/demo.go

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() {
	main2(os.Stdout, os.Args[0], os.Args[1:])
}

func main2(w io.Writer, name string, args []string) {
	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.Fatalf("error in consumer: %v, aborting", err)
	}
	log.Printf("exiting cleanly")
}

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
}