A Simple Queue Implementation in Golang with channels

  // Queue holds name, list of jobs and context with cancel.
  type Queue struct {
     name   string
     jobs   chan Job
     ctx    context.Context
     cancel context.CancelFunc
  }

  // Job - holds logic to perform some operations during queue execution.
  type Job struct {
    Name   string
    Action func() error // A function that should be executed when the job is running.
  }

  // NewQueue instantiates new queue.
  func NewQueue(name string) *Queue {
    ctx, cancel := context.WithCancel(context.Background())

    return &Queue{
       jobs:   make(chan Job),
       name:   name,
       ctx:    ctx,
       cancel: cancel,
    }
  }

// AddJobs adds jobs to the queue and cancels channel.
func (q *Queue) AddJobs(jobs []Job) {
  var wg sync.WaitGroup
  wg.Add(len(jobs))

  for _, job := range jobs {
    // Goroutine which adds job to the queue.
    go func(job Job) {
      q.AddJob(job)
      wg.Done()
    }(job)
  }

  go func() {
    wg.Wait()
    // Cancel queue channel, when all goroutines were done.
    q.cancel()
  }()
}

// AddJob sends job to the channel.
func (q *Queue) AddJob(job Job) {
  q.jobs <- job
  log.Printf("New job %s added to %s queue", job.Name, q.name)
}

// Run performs job execution.
func (j Job) Run() error {
  log.Printf("Job running: %s", j.GetName())

  err := j.Action()
  if err != nil {
    return err
  }

  return nil
}

// Worker responsible for queue serving.
type Worker struct {
  Queue *Queue
}

// NewWorker initializes a new Worker.
func NewWorker(queue *Queue) *Worker {
  return &Worker{
    Queue: queue,
  }
}

// DoWork processes jobs from the queue (jobs channel).
func (w *Worker) DoWork() bool {
  for {
    select {
      // if context was canceled.
      case <-w.Queue.ctx.Done():
        log.Printf("Work done in queue %s: %s!", w.Queue.name, w.Queue.ctx.Err())
        return true
      // if job received.
      case job := <-w.Queue.jobs:
        err := job.Run()
        if err != nil {
          log.Print(err)
          continue
        }
    }
  }
}

package main

import (
	"fmt"
	"github.com/alexsergivan/blog-examples/queue/queue"
)

// Our products storage.
var products = []string{
	"books",
	"computers",
}

func main()  {
    // New products, which we need to add to our products storage.
	newProducts := []string{
		"apples",
		"oranges",
		"wine",
		"bread",
		"orange juice",
	}
    // New queue initialization.
    productsQueue := queue.NewQueue("NewProducts")
	var jobs []queue.Job

    // Range over new products.
	for _, newProduct := range newProducts {
        // We need to do this, because variables declared in for loops are passed by reference.
        // Otherwise, our closure will always receive the last item from the newProducts.
        product := newProduct
        // Defining of the closure, where we add a new product to our simple storage (products slice)
		action := func() error {
			products = append(products, product)
			return nil
        }
        // Append job to jobs slice.
		jobs = append(jobs, queue.Job{
			Name:   fmt.Sprintf("Importing new product: %s", newProduct),
			Action: action,
		})
	}

    // Adds jobs to the queue.
	productsQueue.AddJobs(jobs)

    // Defines a queue worker, which will execute our queue.
    worker := queue.NewWorker(productsQueue)
    // Execute jobs in queue.
    worker.DoWork()

    // Prints products storage after queue execution.
	defer fmt.Print(products)
}

2021/01/12 22:10:33 New job Importing new product: apples added to NewProducts queue
2021/01/12 22:10:33 Job running: Importing new product: apples
2021/01/12 22:10:33 Job running: Importing new product: wine
2021/01/12 22:10:33 Job running: Importing new product: oranges
2021/01/12 22:10:33 Job running: Importing new product: bread
2021/01/12 22:10:33 Job running: Importing new product: orange juice
2021/01/12 22:10:33 New job Importing new product: orange juice added to NewProducts queue
2021/01/12 22:10:33 New job Importing new product: oranges added to NewProducts queue
2021/01/12 22:10:33 New job Importing new product: bread added to NewProducts queue
2021/01/12 22:10:33 New job Importing new product: wine added to NewProducts queue
2021/01/12 22:10:33 Work done in queue NewProducts: context canceled!
[books computers apples wine oranges bread orange juice]%