task.go

package gws

import (
	"sync"

	"github.com/lxzan/gws/internal"
)

type (
	workerQueue struct {
		mu             sync.Mutex               // &#38145;
		q              internal.Deque[asyncJob] // &#20219;&#21153;&#38431;&#21015;
		maxConcurrency int32                    // &#26368;&#22823;&#24182;&#21457;
		curConcurrency int32                    // &#24403;&#21069;&#24182;&#21457;
	}

	asyncJob func()
)

// newWorkerQueue &#21019;&#24314;&#19968;&#20010;&#20219;&#21153;&#38431;&#21015;
func newWorkerQueue(maxConcurrency int32) *workerQueue {
	c := &workerQueue{
		mu:             sync.Mutex{},
		maxConcurrency: maxConcurrency,
		curConcurrency: 0,
	}
	return c
}

// &#33719;&#21462;&#19968;&#20010;&#20219;&#21153;
func (c *workerQueue) getJob(newJob asyncJob, delta int32) asyncJob {
	c.mu.Lock()
	defer c.mu.Unlock()

	if newJob != nil {
		c.q.PushBack(newJob)
	}
	c.curConcurrency += delta
	if c.curConcurrency >= c.maxConcurrency {
		return nil
	}
	var job = c.q.PopFront()
	if job == nil {
		return nil
	}
	c.curConcurrency++
	return job
}

// &#24490;&#29615;&#25191;&#34892;&#20219;&#21153;
func (c *workerQueue) do(job asyncJob) {
	for job != nil {
		job()
		job = c.getJob(nil, -1)
	}
}

// Push &#36861;&#21152;&#20219;&#21153;, &#26377;&#36164;&#28304;&#31354;&#38386;&#30340;&#35805;&#20250;&#31435;&#21363;&#25191;&#34892;
func (c *workerQueue) Push(job asyncJob) {
	if nextJob := c.getJob(job, 0); nextJob != nil {
		go c.do(nextJob)
	}
}

type channel chan struct{}

func (c channel) add() { c <- struct{}{} }

func (c channel) done() { <-c }

func (c channel) Go(m *Message, f func(*Message) error) error {
	c.add()
	go func() {
		_ = f(m)
		c.done()
	}()
	return nil
}