batch/safe.go at master · nikandfor/batch · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
package batch

import "context"

type (
	Batch[Res any] struct {
		c *Controller[Res]

		noCopy noCopy
		state  byte
	}

	noCopy struct{}
)

const (
	stateNew = iota
	stateQueued
	stateEntered
	stateCommitted
	stateExited = stateNew

	usage = "By -> defer Exit -> [QueueIn] -> Enter -> Cancel/Commit/return"
)

func (c *Controller[Res]) Batch() Batch[Res] {
	return Batch[Res]{
		c: c,
	}
}

func (c *Controller[Res]) QueueInBatch() Batch[Res] {
	c.queue.In()

	return Batch[Res]{
		c:     c,
		state: stateQueued,
	}
}

func (b *Batch[Res]) QueueIn() int {
	if b.state != stateNew {
		panic(usage)
	}

	b.state = stateQueued

	return b.c.queue.In()
}

func (b *Batch[Res]) Enter(blocking bool) int {
	switch b.state {
	case stateNew:
		b.QueueIn()
	case stateQueued:
	default:
		panic(usage)
	}

	idx := b.c.Enter(blocking)
	if idx >= 0 {
		b.state = stateEntered
	} else {
		b.state = stateNew
	}

	return idx
}

func (b *Batch[Res]) Trigger() {
	b.c.Trigger()
}

func (b *Batch[Res]) Cancel(ctx context.Context, err error) (Res, error) {
	if b.state != stateEntered {
		panic(usage)
	}

	b.state = stateCommitted

	return b.c.Cancel(ctx, err)
}

func (b *Batch[Res]) Commit(ctx context.Context) (Res, error) {
	return b.CommitFunc(ctx, b.c.Committer)
}

func (b *Batch[Res]) CommitFunc(ctx context.Context, f CommitFunc[Res]) (Res, error) {
	if b.state != stateEntered {
		panic(usage)
	}

	b.state = stateCommitted

	return b.c.CommitFunc(ctx, f)
}

func (b *Batch[Res]) Exit() int {
	return b.ExitErr(nil)
}

func (b *Batch[Res]) ExitErr(errp *error) (idx int) {
	idx = -1
	s := b.state
	b.state = stateExited

	switch s {
	case stateNew:
	case stateQueued:
		b.c.queue.Out()
		b.c.Notify()
	case stateEntered, stateCommitted:
		idx = b.c.ExitErr(errp)
	default:
		panic(usage)
	}

	return idx
}

func (noCopy) Lock()   {}
func (noCopy) Unlock() {}