You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
375 lines
8.0 KiB
375 lines
8.0 KiB
// ------------------------------------------------------------------------
|
|
// Project atila
|
|
// Active Thing (activething.com) git.activething.com/go
|
|
//
|
|
// File name completion.go
|
|
// Created by DEV
|
|
// Modified 20/02/2024
|
|
//
|
|
// Copyright 2024 activething.com
|
|
// ------------------------------------------------------------------------
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
// ------------------------------------------------------------------------
|
|
|
|
package ctasks
|
|
|
|
import (
|
|
"context"
|
|
"crns"
|
|
"opts"
|
|
"sync"
|
|
"sync/atomic"
|
|
"syns/works"
|
|
"time"
|
|
|
|
)
|
|
|
|
type (
|
|
|
|
Completion struct {
|
|
/*embedded*/context.Context
|
|
mux sync.RWMutex
|
|
wgr sync.WaitGroup
|
|
|
|
parent Context
|
|
expiration *crns.Expiration
|
|
|
|
mode Mode
|
|
state uint32
|
|
pending uint32
|
|
|
|
cid int
|
|
result Result
|
|
|
|
dispatchFn dispatchFnc
|
|
cancelFn func()
|
|
|
|
completers []Completer
|
|
}
|
|
|
|
CompletionFnc func (c *Completion) error
|
|
|
|
dispatchFnc func (*Completion, Completer, int) error
|
|
|
|
|
|
)
|
|
|
|
|
|
func NewCompletion (options ...CompletionFnc) (*Completion,error) {
|
|
|
|
c := &Completion {
|
|
mode : ModeAll,
|
|
cid : CompletionFromSelf,
|
|
dispatchFn: dispatcher,
|
|
}
|
|
for _,o := range options {
|
|
if er := o(c); er != nil {
|
|
return nil,er
|
|
}
|
|
}
|
|
if c.expiration == nil {
|
|
c.expiration = crns.NewExpiration(0,nil)
|
|
}
|
|
return c,nil
|
|
}
|
|
|
|
|
|
func (c *Completion) Mode () Mode {
|
|
return c.mode }
|
|
|
|
func (c *Completion) State () State {
|
|
return State(atomic.LoadUint32(&c.state)) }
|
|
|
|
func (c *Completion) Pending() int {
|
|
return int(atomic.LoadUint32(&c.pending)) }
|
|
|
|
func (c *Completion) IsCompleted () bool {
|
|
return c.State() > StateStarted
|
|
}
|
|
|
|
func (c *Completion) Chrono () crns.Chronometer {
|
|
return c.expiration.Chono() }
|
|
|
|
func (c *Completion) Result () Result {
|
|
c.wgr.Wait()
|
|
return c.result
|
|
}
|
|
|
|
func (c *Completion) Wait () {
|
|
c.wgr.Wait() }
|
|
|
|
|
|
func (c *Completion) AddCompleter (cms ... Completer) error {
|
|
c.mux.Lock()
|
|
if c.State() != StateCreated {
|
|
c.mux.Unlock()
|
|
return G.ErrInvalidState
|
|
}
|
|
for _,cm := range cms {
|
|
c.completers = append(c.completers, cm)
|
|
}
|
|
c.mux.Unlock()
|
|
return nil
|
|
}
|
|
|
|
|
|
func (c *Completion) Cancel () {
|
|
c.notify(CompletionFromRequest, StateCanceled,nil,G.ErrCancelByRequest) }
|
|
|
|
|
|
func (c *Completion) Start (ctx context.Context) error {
|
|
return c.complete(ctx,nil,CompletionFromSelf) }
|
|
|
|
|
|
func (c *Completion) Complete (cmp Context, cid int) {
|
|
_ = c.complete(nil,cmp,cid) }
|
|
|
|
|
|
|
|
func (c *Completion) complete (ctx context.Context, cctx Context, cid int) error {
|
|
c.mux.Lock()
|
|
|
|
if c.State() != StateCreated {
|
|
c.mux.Unlock()
|
|
return G.ErrInvalidState
|
|
}
|
|
|
|
// Nos sirve para determinar de donde va a depender la expiracion
|
|
// en tiempo de la completion en caso de que exista, o bien de la
|
|
// completion padre o bien del contexto.
|
|
//
|
|
var ex crns.Expirer
|
|
|
|
if cctx != nil {
|
|
c.parent = cctx
|
|
c.Context= cctx
|
|
c.cid = cid
|
|
if c.parent.IsCompleted() {
|
|
c.mux.Unlock()
|
|
c.notify(CompletionFromParent, StateCanceled,nil,G.ErrCancelByParent)
|
|
return G.ErrCancelByParent
|
|
}
|
|
ex = c.parent
|
|
} else {
|
|
if ctx == nil {
|
|
c.Context = context.Background()
|
|
} else {
|
|
c.Context = ctx
|
|
if ctx.Err() != nil {
|
|
c.mux.Unlock()
|
|
c.notify(CompletionFromContext, StateCanceled,nil,G.ErrCancelByContext)
|
|
return G.ErrCancelByContext.WrapWithCause(ctx.Err())
|
|
}
|
|
|
|
ch := make(chan struct{})
|
|
c.cancelFn = func(){ close(ch)}
|
|
go func(cc *Completion, ch chan struct{}){
|
|
select {
|
|
case <- ch:
|
|
case <- cc.Context.Done() :
|
|
er := G.ErrCancelByContext.WrapWithCause(cc.Context.Err())
|
|
cc.notify(CompletionFromContext, StateCanceled,nil,er)
|
|
}
|
|
}(c,ch)
|
|
}
|
|
ex = ctx
|
|
}
|
|
|
|
if er := c.expiration.Start(ex); er != nil {
|
|
c.mux.Unlock()
|
|
c.notify(CompletionFromSelf, StateFaulted,nil,er)
|
|
return er
|
|
}
|
|
|
|
atomic.StoreUint32(&c.state,uint32(StateStarted))
|
|
c.wgr.Add(1)
|
|
|
|
go func (c *Completion) {
|
|
ln := len(c.completers)
|
|
if ln == 0 {
|
|
c.mux.Unlock()
|
|
c.notify(CompletionFromSelf, StateCompleted, nil, nil)
|
|
return
|
|
}
|
|
|
|
atomic.StoreUint32(&c.pending,uint32(ln))
|
|
cs := c.completers
|
|
c.mux.Unlock()
|
|
|
|
for ix, cm := range cs {
|
|
if c.IsCompleted() {
|
|
return
|
|
}
|
|
if er := c.dispatchFn(c,cm,ix); er != nil {
|
|
c.notify(CompletionFromSelf, StateFaulted,nil,er)
|
|
return
|
|
}
|
|
}
|
|
}(c)
|
|
|
|
return nil
|
|
}
|
|
|
|
|
|
|
|
func (c *Completion) Notify (frm int, val interface{}, err error) {
|
|
c.notify(frm, StateCompleted,val,err) }
|
|
|
|
|
|
|
|
func (c *Completion) notify (frm int, ste State, val interface{}, err error) {
|
|
c.mux.Lock()
|
|
os := c.State()
|
|
if os > StateStarted {
|
|
c.mux.Unlock()
|
|
return
|
|
}
|
|
|
|
if frm >= CompletionFromChild {
|
|
cn := atomic.AddUint32(&c.pending,-1)
|
|
|
|
switch {
|
|
case c.mode.IsAnyOrResult() && val != nil:
|
|
atomic.StoreUint32(&c.state,uint32(StateCompleted))
|
|
c.result = NewResultValue(frm, val)
|
|
|
|
case c.mode.IsAnyOrError() && err != nil:
|
|
atomic.StoreUint32(&c.state,uint32(StateFaulted))
|
|
c.result = NewResultError(frm, err)
|
|
|
|
case c.mode == ModeTask:
|
|
c.result = NewResult(frm, val, err)
|
|
|
|
default:
|
|
if c.mode == ModeAll {
|
|
if c.result == nil {
|
|
c.result = NewResultList(c.cid, NewResult(frm, val, err))
|
|
} else {
|
|
c.result.(*ResultList).Add(NewResult(frm, val, err))
|
|
}
|
|
}
|
|
if cn == 0 {
|
|
atomic.StoreUint32(&c.state,uint32(StateCompleted))
|
|
} else {
|
|
c.mux.Unlock()
|
|
return
|
|
}
|
|
}
|
|
|
|
} else {
|
|
atomic.StoreUint32(&c.state,uint32(ste))
|
|
c.result = NewResult(frm,val,err)
|
|
}
|
|
|
|
_ = c.expiration.Stop()
|
|
if c.cancelFn != nil {
|
|
c.cancelFn()
|
|
}
|
|
c.wgr.Done()
|
|
c.mux.Unlock()
|
|
|
|
|
|
// Solo notificamos a las completions hijas y al padre en caso de que se haya iniciado
|
|
if os == StateStarted {
|
|
for _, ch := range c.completers {
|
|
// Notificamos solo a los completer que a su vez implementan
|
|
// Context
|
|
if cm, is := ch.(Context); is {
|
|
if !cm.IsCompleted() {
|
|
cm.Notify(CompletionFromParent, nil, G.ErrCancelByParent)
|
|
}
|
|
}
|
|
}
|
|
// Notificamos al padre en caso de que tenga y la completion no venga de el
|
|
if frm != CompletionFromParent && c.parent != nil {
|
|
c.parent.Notify(c.cid,c.result.Value(),c.result.Error())
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
func dispatcher (completion *Completion, completer Completer, cid int) error {
|
|
go completer.Complete(completion,cid)
|
|
return nil
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// WithMode
|
|
// Desc: Función para el constructor de Completion,
|
|
// establece el mode de como se llegara a la completion
|
|
//
|
|
func WithMode (mode Mode) opts.GOptionErrFnc[Completion] {
|
|
return func(c *Completion) error {
|
|
c.mode = mode
|
|
return nil
|
|
}
|
|
}
|
|
|
|
|
|
// WithTimeout
|
|
// Desc: Función para el constructor de Completion
|
|
// Establece el tiempo máximo de la completion
|
|
//
|
|
func WithTimeout (tmo time.Duration) opts.GOptionErrFnc[Completion] {
|
|
return func(c *Completion) error {
|
|
c.expiration = crns.NewExpiration(tmo,func(){
|
|
c.notify(CompletionFromTimer, StateCanceled,nil,G.ErrCancelByTimer)
|
|
})
|
|
return nil
|
|
}
|
|
}
|
|
|
|
|
|
// WithCompleters
|
|
// Desc: Función para el constructor de Completion,
|
|
// añade los objeto a completar en la completion
|
|
//
|
|
func WithCompleters (completers ... Completer) opts.GOptionErrFnc[Completion] {
|
|
return func(c *Completion) error {
|
|
if c.State() != StateCreated {
|
|
return G.ErrInvalidState
|
|
}
|
|
for _, cm := range completers {
|
|
c.completers = append(c.completers, cm)
|
|
}
|
|
return nil
|
|
}
|
|
}
|
|
|
|
|
|
// WithWorkPool
|
|
// Desc: Función para el constructor de Completion,
|
|
// utiliza un pool de trabajo para completar
|
|
// las tareas.
|
|
//
|
|
func WithWorkPool (pool *works.Pool) opts.GOptionErrFnc[Completion] {
|
|
return func(c *Completion) error {
|
|
c.dispatchFn = dispatchFnc(func (cm *Completion, cr Completer, ci int) error {
|
|
return pool.DoWork(c,func(){
|
|
cr.Complete(cm,ci)
|
|
})
|
|
})
|
|
return nil
|
|
}
|
|
}
|
|
|