carlos / platform Public

Download this file

1	package host
2
3	import (
4	"context"
5	"github.com/carlosframework/platform/internal/hostbox"
6	"github.com/carlosframework/platform/internal/hostwrite"
7	"github.com/carlosframework/platform/internal/registry"
8	"github.com/carlosframework/platform/internal/schedule"
9	)
10
11	func ClaimTimerDelivery(set *ClaimProducerSet, newID func() string) schedule.DeliveryAdmission {
12	return func(ctx context.Context, work hostwrite.Work, rt registry.Route, t registry.Timer) (context.Context, func(), error) {
13	if set == nil || newID == nil {
14	return nil, nil, hostwrite.ErrUnavailable
15	}
16	selected, err := hostbox.Selected(work)
17	if err != nil {
18	return nil, nil, err
19	}
20	if selected.Host != rt.Host || t.Host != rt.Host || selected.Captured.Incarnation != rt.ClaimIncarnation || selected.Captured.Stamp != rt.ClaimStamp || t.ClaimIncarnation != rt.ClaimIncarnation || t.ClaimStamp != rt.ClaimStamp {
21	return nil, nil, hostbox.ErrChanged
22	}
23	set.mu.Lock()
24	var chosen *ClaimProducers
25	for _, p := range set.items {
26	if p.target.Host == selected.Host && p.target.Incarnation == selected.Captured.Incarnation && p.target.Stamp == selected.Captured.Stamp && p.target.Session == selected.Session && p.target.Box == selected.Box {
27	if chosen != nil {
28	set.mu.Unlock()
29	return nil, nil, hostbox.ErrChanged
30	}
31	chosen = p
32	}
33	}
34	set.mu.Unlock()
35	if chosen == nil {
36	return nil, nil, hostwrite.ErrUnavailable
37	}
38	var child context.Context
39	var finish func(error)
40	err = work.With(ctx, hostwrite.Identity{}, func() error {
41	var e error
42	child, finish, e = chosen.Register(ctx, chosen.target, "timer-"+newID())
43	return e
44	})
45	if err != nil {
46	return nil, nil, err
47	}
48	return child, func() { finish(nil) }, nil
49	}
50	}
51