package buildstep import ( "context" "sync" "time" "git.novox.be/novox/mesh-sdk/go/delivery" ) // FakeAsks is a test double of a seat's workers for the build modules' unit tests: Publish puts an ask on a kind's // worker, Next hands them out in their order, and the fates the puller gives them are counted. Safe for concurrent use. type FakeAsks struct { mu sync.Mutex waiting map[string][][]byte added chan struct{} // Acked, Termed and Naked are the asks' ids or data by fate, InProgress how many heartbeats were given. Acked, Termed, Naked []string InProgress int // Delivered, when set, is the delivery count every ask handed over says (1 when 0): a test of a poisoned ask sets it. Delivered uint64 } // NewFakeAsks is an empty double. func NewFakeAsks() *FakeAsks { return &FakeAsks{waiting: map[string][][]byte{}, added: make(chan struct{}, 1)} } // Publish puts an encoded ask on kind's worker. func (f *FakeAsks) Publish(kind string, data []byte) { f.mu.Lock() f.waiting[kind] = append(f.waiting[kind], data) f.mu.Unlock() select { case f.added <- struct{}{}: default: } } // PublishAsk encodes a and puts it on its kind's worker. func (f *FakeAsks) PublishAsk(a delivery.StepAsk) error { raw, err := delivery.EncodeStepAsk(a) if err != nil { return err } f.Publish(a.Kind, raw) return nil } // Waiting is how many asks wait on kind's worker. func (f *FakeAsks) Waiting(kind string) int { f.mu.Lock() defer f.mu.Unlock() return len(f.waiting[kind]) } // Fates is a copy of the acked, termed and naked lists. func (f *FakeAsks) Fates() (acked, termed, naked []string) { f.mu.Lock() defer f.mu.Unlock() return append([]string(nil), f.Acked...), append([]string(nil), f.Termed...), append([]string(nil), f.Naked...) } // Next is AskSource's. func (f *FakeAsks) Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error) { var deadline <-chan time.Time if wait > 0 { t := time.NewTimer(wait) defer t.Stop() deadline = t.C } for { f.mu.Lock() if q := f.waiting[kind]; len(q) > 0 { data := q[0] f.waiting[kind] = q[1:] f.mu.Unlock() return f.pulled(data), nil } f.mu.Unlock() if wait <= 0 { return nil, nil } select { case <-ctx.Done(): return nil, ctx.Err() case <-deadline: return nil, nil case <-f.added: } } } func (f *FakeAsks) pulled(data []byte) *Pulled { id := string(data) if a, err := delivery.DecodeStepAsk(data); err == nil { id = a.ID } fate := func(list *[]string) func() error { return func() error { f.mu.Lock() *list = append(*list, id) f.mu.Unlock() return nil } } delivered := f.Delivered if delivered == 0 { delivered = 1 } return &Pulled{Data: data, Delivered: delivered, Ack: fate(&f.Acked), Nak: fate(&f.Naked), Term: fate(&f.Termed), InProgress: func() error { f.mu.Lock(); f.InProgress++; f.mu.Unlock(); return nil }} }