queue-oncomplete-routine.go

package main

import (
	"log"
	"strings"
	"sync"
	"time"
)

type workdata struct {
	s string
	i int
}

type j struct {
	wd    *workdata
	doneC chan struct{}
}

var wg = &sync.WaitGroup{}
var pipeline = make(chan j, 100)

func workFn(jj j, cb func(j)) {

	defer cb(jj)

	time.Sleep(time.Second)
	jj.wd.s = strings.ToLower(jj.wd.s)
	jj.wd.i += 100
}

// workReader is an example of a sync/async delegator pattern.
// A real world use case for this might be a SQL ORM or otherwise database r/w scenario,
// where READs are either unlimited or can and will be managed in another system,
// but where WRITEs are either already blocking (and will cause a connection timeout if
// queued too deep) or the order/result of WRITEs is required syncronously by the caller, eg.
// to get an auto-assigned 'ID' for the creation of a new row, or to depend on an error or not.
func workReader(inC <-chan j) {
	onDone := func(jj j) {

		defer wg.Done()

		if jj.doneC != nil {
			log.Println("sync done", jj.wd)
			jj.doneC <- struct{}{}
		} else {
			log.Println("async done", jj.wd)
		}
	}

	for jj := range inC {
		jj := jj
		if jj.doneC == nil {
			go workFn(jj, onDone)
		} else {
			workFn(jj, onDone)
		}
	}

}

func jRun(w *workdata, await bool) {
	wg.Add(1)

	var out = make(chan struct{}, 1)
	var jj = j{w, out}

	if !await {
		jj.doneC = nil
		out <- struct{}{}
	}

	pipeline <- jj
	<-out
	close(out)
}

func main() {

	go workReader(pipeline)

	for i := 0; i < 100; i++ {
		if i%5 == 0 {
			syncWD := &workdata{
				"hElLO",
				i,
			}
			log.Println("run sync", i)
			jRun(syncWD, true)
			log.Println("done sync", syncWD)
			continue
		}
		log.Println("run async", i)
		jRun(&workdata{
			"hElLO",
			i,
		}, false)
	}

	wg.Wait()
}