func (s *SubQ) Wait(ctx context.Context, qs []string, pollWait time.Duration, condition func() bool) error {
	if condition == nil {
		condition = makeDefaultCondition()
	}
	wait := func() <-chan time.Time {
		if pollWait > 0 {
			return time.After(pollWait)
		}
		return nil
	}
	var listeners []*sub
	for _, q := range qs {
		qi := s.addListener(q)
		defer qi.Done()
		listeners = append(listeners, qi)
	}
	for !condition() {
		if len(listeners) == 1 {
			select {
			case <-wait():
			case <-listeners[0].Ch():
			case <-ctx.Done():
				return fmt.Errorf("subq wait: %w", ctx.Err())
			}
			continue
		}
		cases := []reflect.SelectCase{
			{
				Dir:  reflect.SelectRecv,
				Chan: reflect.ValueOf(ctx.Done()),
			},
			{
				Dir:  reflect.SelectRecv,
				Chan: reflect.ValueOf(wait()),
			},
		}
		for _, listener := range listeners {
			cases = append(cases, reflect.SelectCase{
				Dir:  reflect.SelectRecv,
				Chan: reflect.ValueOf(listener.Ch()),
			})
		}
		chosen, _, _ := reflect.Select(cases)
		if chosen == 0 {
			return fmt.Errorf("subq wait multi: %w", ctx.Err())
		}
	}
	return nil
}
Wait waits on the given queues until something is notified on one or the
context expires, whichever comes first. This is the basic behavior when
pollWait is 0 and condition is nil.

If condition is not nil, it is called immediately. If it returns true, then the
wait is satisfied and the function exits with a nil error. If it returns
false, then the function begins to wait for either a notification, a context
cancelation, or the expiration of pollWait.

Note that condition should execute quickly and not block. It should test the
condition as fast as it can and return. Otherwise "Notify" might busy-wait
for a while, which is obviously not good.

When pollWait expires or a suitable notification arrives, condition is called
again, and the above process repeats.

If pollWait is 0, the only way to check condition again is if the channel is
notified. Otherwise Wait terminates with an error.

This implementation allows you to attempt a polling operation, then wait for
notification that the next one is likely to succeed, then check again just
in case you got scooped by another process, repeating until something is
truly available.

Note that condition is called directly from this function, so if it needs a
context, it can simply close over the same one passed in here.
package subq
import (
	"context"
	"fmt"
	"log"
	"reflect"
	"sync"
	"time"
)
type SubQ struct {
	sync.Mutex
	qs map[string]*sub
}
type sub struct {
	sync.Mutex
	listeners int
	ch        chan string
}
func (s *sub) Ch() chan string {
	if s == nil {
		return nil
	}
	defer un(lock(s))
	return s.ch
}
func (s *sub) Add() {
	defer un(lock(s))
	s.listeners++
}
func (s *sub) Done() {
	defer un(lock(s))
	if s.listeners == 0 {
		log.Fatal("Done before Add")
	}
	s.listeners--
}
func (s *sub) Reserved() bool {
	if s == nil {
		return false
	}
	defer un(lock(s))
	return s.listeners != 0
}
func lock(l sync.Locker) func() {
	l.Lock()
	return l.Unlock
}
func un(f func()) {
	f()
}
func New() *SubQ {
	return &SubQ{
		qs: make(map[string]*sub),
	}
}
func (s *SubQ) Notify(q string) {
	defer un(lock(s))
	qInfo := s.qs[q]
	if !qInfo.Reserved() {
		return
	}
	go func() {
		for {
			select {
			case qInfo.Ch() <- q:
				return
			case <-time.After(500 * time.Millisecond):
				if !qInfo.Reserved() {
					return
				}
			}
		}
	}()
}
func makeDefaultCondition() func() bool {
	var done bool
	return func() bool {
		val := done
		done = true
		return val
	}
}
func (s *SubQ) tryCollect(q string) bool {
	defer un(lock(s))
	if !s.qs[q].Reserved() {
		delete(s.qs, q)
		return true
	}
	return false
}
func (s *SubQ) addListener(q string) *sub {
	defer un(lock(s))
	qs := s.qs
	if qs[q] != nil {
		qs[q].Add()
		return qs[q]
	}
	qs[q] = &sub{ch: make(chan string)}
	qs[q].Add()
	go func() {
		for !s.tryCollect(q) {
			time.Sleep(1 * time.Second)
		}
	}()
	return qs[q]
}