forked from asonawalla/gazette
/
replica.go
100 lines (83 loc) · 2.76 KB
/
replica.go
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
package journal
import (
log "github.com/sirupsen/logrus"
"github.com/LiveRamp/gazette/pkg/cloudstore"
)
// Replica manages journal components required to serve brokered writes,
// replications, and reads. A Replica instance is capable of switching roles
// at any time (and multiple times), from a pure replica which may serve
// replication requests only, to a broker of the journal.
type Replica struct {
journal Name
// Fragment updates are written into |updates| by |index| and |head|,
// and are consumed by |tail|.
updates chan Fragment
// Watches the long-term storage location for newly available fragments.
index *IndexWatcher
// Serves fragment reads.
tail *Tail
// Serves replicated writes.
head *Head
// Brokers transactions which result in replicated writes to the journal.
broker *Broker
}
func NewReplica(journal Name, localDir string, persister FragmentPersister,
cfs cloudstore.FileSystem) *Replica {
updates := make(chan Fragment, 1)
r := &Replica{
journal: journal,
updates: updates,
index: NewIndexWatcher(journal, cfs, updates).StartWatchingIndex(),
tail: NewTail(journal, updates).StartServingOps(),
head: NewHead(journal, localDir, persister, updates),
broker: NewBroker(journal),
}
// Defer writes until local fragments & the remote index are fully loaded.
go func() {
for _, f := range LocalFragments(localDir, journal) {
updates <- f
}
r.index.WaitForInitialLoad()
log.WithField("journal", journal).Debug("starting head and broker")
r.head.StartServingOps(r.tail.EndOffset())
r.broker.StartServingOps(r.tail.EndOffset())
}()
return r
}
func (r *Replica) Append(op AppendOp) {
r.broker.Append(op)
}
func (r *Replica) Replicate(op ReplicateOp) {
r.head.Replicate(op)
}
func (r *Replica) Read(op ReadOp) {
r.index.WaitForInitialLoad()
r.tail.Read(op)
}
// Switch the Replica into pure-replica mode.
func (r *Replica) StartReplicating(routeToken RouteToken) {
log.WithFields(log.Fields{"journal": r.journal, "route": routeToken}).
Debug("now replicating")
}
// Switch the Replica into broker mode. Appends will be brokered to |peers| with
// the topology captured by |routeToken|.
func (r *Replica) StartBrokeringWithPeers(routeToken RouteToken, peers []Replicator) {
log.WithFields(log.Fields{"journal": r.journal, "route": routeToken}).
Debug("now brokering")
var config BrokerConfig
config.RouteToken = routeToken
config.WriteHead = r.tail.EndOffset()
config.Replicas = append(peers, r.head)
r.broker.UpdateConfig(config)
}
func (r *Replica) Shutdown() {
log.WithField("journal", r.journal).Debug("beginning journal shutdown")
go func() {
r.broker.Stop()
r.head.Stop()
r.index.Stop()
close(r.updates)
r.tail.Stop()
log.WithField("journal", r.journal).Debug("completed journal shutdown")
}()
}