queueservice

package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/faustbrian/go-correlation"
	"github.com/faustbrian/go-queue/core"
	"github.com/faustbrian/go-queue/queueservice"
)

type typedWorkerResource struct{}

type typedDelivery string

func (delivery typedDelivery) Bytes() []byte   { return []byte(delivery) }
func (delivery typedDelivery) Payload() []byte { return []byte(delivery) }

func main() {
	factory, _ := correlation.NewFactory(correlation.FactoryOptions{})
	worker, err := queueservice.NewLifecycleWorker(
		queueservice.LifecycleWorkerOptions[*typedWorkerResource]{
			Name:        "orders-worker",
			Resource:    &typedWorkerResource{},
			Correlation: factory,
			Handler: func(_ context.Context, task core.TaskMessage) error {
				fmt.Println(string(task.Payload()))
				return nil
			},
			Run: func(
				ctx context.Context,
				_ *typedWorkerResource,
				handler queueservice.Handler,
			) error {
				return handler(ctx, typedDelivery("delivery"))
			},
			Shutdown: func(context.Context, *typedWorkerResource) error {
				return nil
			},
		},
	)
	if err != nil {
		return
	}
	plan := worker.Plan()
	if err = plan.Components[0].Start(context.Background()); err != nil {
		return
	}
	if err = plan.Tasks[0].Run(context.Background()); !errors.Is(err, queueservice.ErrWorkerExited) {
		return
	}
	stopContext, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err = plan.Components[0].Stop(stopContext); err != nil {
		return
	}

}

Output:
delivery

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/faustbrian/go-correlation"
	queue "github.com/faustbrian/go-queue"
	"github.com/faustbrian/go-queue/core"
	"github.com/faustbrian/go-queue/job"
	"github.com/faustbrian/go-queue/queueservice"
	"go.opentelemetry.io/otel/propagation"
)

type payload string

func (value payload) Bytes() []byte { return []byte(value) }

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

	factory, _ := correlation.NewFactory(correlation.FactoryOptions{})
	handled := make(chan string, 1)
	handler, err := queueservice.NewHandler(queueservice.HandlerOptions{
		Correlation:     factory,
		TrustedMetadata: true,
		TracePropagator: propagation.TraceContext{},
		Handler: func(_ context.Context, task core.TaskMessage) error {
			handled <- string(task.Payload())
			return nil
		},
	})
	if err != nil {
		return
	}
	concrete, err := queue.NewQueue(
		queue.WithWorker(queue.NewRing(queue.WithFn(handler))),
		queue.WithWorkerCount(1),
	)
	if err != nil {
		return
	}
	worker, err := queueservice.NewWorker(queueservice.WorkerOptions{
		Name: "jobs-worker", Queue: concrete,
	})
	if err != nil {
		return
	}
	producer, err := queueservice.NewProducer(
		queueservice.ProducerOptions[*queue.Queue]{
			Name: "jobs-producer", Resource: concrete, Correlation: factory,
			TracePropagator: propagation.TraceContext{},
			Publish: func(
				_ context.Context,
				resource *queue.Queue,
				message core.QueuedMessage,
				options ...job.AllowOption,
			) error {
				return resource.Queue(message, options...)
			},
		},
	)
	if err != nil {
		return
	}

	workerComponent := worker.Component()
	producerComponent := producer.Component()
	if err = workerComponent.Start(ctx); err != nil {
		return
	}
	if err = producerComponent.Start(ctx); err != nil {
		return
	}
	parent, _ := factory.Start()
	if _, err = producer.Publish(
		correlation.WithValues(ctx, parent),
		payload("delivery"),
	); err != nil {
		return
	}
	select {
	case value := <-handled:
		fmt.Println(value)
	case <-ctx.Done():
		return
	}
	if err = producerComponent.Stop(ctx); err != nil {
		return
	}
	if err = workerComponent.Stop(ctx); err != nil {
		return
	}

}

Output:
delivery