go-gun/gun/scoped_put.go

182 lines
5.0 KiB
Go
Raw Normal View History

2019-02-25 04:23:15 +00:00
package gun
import (
"context"
"fmt"
)
type putResultListener struct {
id string
results chan *PutResult
receivedMessages chan *messageReceived
2019-02-25 04:23:15 +00:00
}
type PutResult struct {
Err error
2019-02-25 05:14:26 +00:00
// Nil on error or local put success
Peer *Peer
2019-02-25 04:23:15 +00:00
}
type PutOption interface{}
type putOptionStoreLocalOnly struct{}
func PutOptionStoreLocalOnly() PutOption { return putOptionStoreLocalOnly{} }
type putOptionFailWithoutParent struct{}
func PutOptionFailWithoutParent() PutOption { return putOptionFailWithoutParent{} }
func (s *Scoped) Put(ctx context.Context, val Value, opts ...PutOption) <-chan *PutResult {
// Collect the options
storeLocalOnly := false
failWithoutParent := false
for _, opt := range opts {
switch opt.(type) {
case putOptionStoreLocalOnly:
storeLocalOnly = true
case putOptionFailWithoutParent:
failWithoutParent = true
}
}
ch := make(chan *PutResult, 1)
// Get all the parents
parents := []*Scoped{}
for next := s.parent; next != nil; next = next.parent {
parents = append([]*Scoped{next}, parents...)
}
if len(parents) == 0 {
ch <- &PutResult{Err: ErrLookupOnTopLevel}
close(ch)
return ch
}
// Ask for the soul on the last parent. What this will do is trigger
// lazy soul fetch up the chain. Then we can go through and find who doesn't have a
// cached soul, create one, and store locally.
if soul, err := parents[len(parents)-1].Soul(ctx); err != nil {
ch <- &PutResult{Err: ErrLookupOnTopLevel}
close(ch)
return ch
} else if soul == "" && failWithoutParent {
ch <- &PutResult{Err: fmt.Errorf("Parent not present but required")}
close(ch)
return ch
}
// Now for every parent that doesn't have a cached soul we create one and
// put as part of the message. We accept fetching the cache this way is a bit
// racy.
req := &Message{
ID: randString(9),
Put: make(map[string]*Node),
}
// We know that the first has a soul
prevParentSoul := parents[0].cachedSoul()
2019-02-25 05:14:26 +00:00
currState := StateNow()
2019-02-25 04:23:15 +00:00
for _, parent := range parents[1:] {
parentCachedSoul := parent.cachedSoul()
if parentCachedSoul == "" {
// Create the soul and make it as part of the next put
parentCachedSoul = s.gun.soulGen()
req.Put[prevParentSoul] = &Node{
Metadata: Metadata{
Soul: prevParentSoul,
2019-02-25 05:14:26 +00:00
State: map[string]State{parent.field: currState},
2019-02-25 04:23:15 +00:00
},
Values: map[string]Value{parent.field: ValueRelation(parentCachedSoul)},
}
// Also store locally and set the cached soul
// TODO: Should I not store until the very end just in case it errors halfway
// though? There are no standard cases where it should fail.
if _, err := s.gun.storage.Put(ctx, prevParentSoul, parent.field, ValueRelation(parentCachedSoul), currState, false); err != nil {
2019-02-25 04:23:15 +00:00
ch <- &PutResult{Err: err}
close(ch)
return ch
} else if !parent.setCachedSoul(ValueRelation(parentCachedSoul)) {
ch <- &PutResult{Err: fmt.Errorf("Concurrent cached soul set")}
close(ch)
return ch
}
}
prevParentSoul = parentCachedSoul
}
2019-02-25 05:14:26 +00:00
// Now that we've setup all the parents, we can do this store locally
if _, err := s.gun.storage.Put(ctx, prevParentSoul, s.field, val, currState, false); err != nil {
2019-02-25 04:23:15 +00:00
ch <- &PutResult{Err: err}
close(ch)
return ch
}
// We need an ack for local store and stop if local only
ch <- &PutResult{}
if storeLocalOnly {
close(ch)
return ch
}
// Now, we begin the remote storing
req.Put[prevParentSoul] = &Node{
Metadata: Metadata{
Soul: prevParentSoul,
2019-02-25 05:14:26 +00:00
State: map[string]State{s.field: currState},
2019-02-25 04:23:15 +00:00
},
Values: map[string]Value{s.field: val},
}
// Make a msg chan and register it to listen for acks
msgCh := make(chan *messageReceived)
2019-02-25 04:23:15 +00:00
s.putResultListenersLock.Lock()
s.putResultListeners[ch] = &putResultListener{req.ID, ch, msgCh}
s.putResultListenersLock.Unlock()
2019-02-25 05:14:26 +00:00
s.gun.registerMessageIDListener(req.ID, msgCh)
2019-02-25 04:23:15 +00:00
// Start message listener
go func() {
for {
select {
case <-ctx.Done():
ch <- &PutResult{Err: ctx.Err()}
s.PutDone(ch)
return
case msg, ok := <-msgCh:
if !ok {
return
}
r := &PutResult{Peer: msg.peer}
2019-02-25 04:23:15 +00:00
if msg.Err != "" {
r.Err = fmt.Errorf("Remote error: %v", msg.Err)
} else if msg.OK != 1 {
r.Err = fmt.Errorf("Unexpected remote ok value of %v", msg.OK)
}
safePutResultSend(ch, r)
}
}
}()
// Send async, sending back errors
go func() {
2019-02-25 05:14:26 +00:00
for peerErr := range s.gun.send(ctx, req, nil) {
2019-02-25 04:23:15 +00:00
safePutResultSend(ch, &PutResult{
Err: peerErr.Err,
2019-02-25 05:14:26 +00:00
Peer: peerErr.Peer,
2019-02-25 04:23:15 +00:00
})
}
}()
return ch
}
func (s *Scoped) PutDone(ch <-chan *PutResult) bool {
s.putResultListenersLock.Lock()
l := s.putResultListeners[ch]
delete(s.putResultListeners, ch)
s.putResultListenersLock.Unlock()
if l != nil {
// Unregister the chan
2019-02-25 05:14:26 +00:00
s.gun.unregisterMessageIDListener(l.id)
2019-02-25 04:23:15 +00:00
// Close the message chan and the result chan
close(l.receivedMessages)
close(l.results)
}
return l != nil
}
func safePutResultSend(ch chan<- *PutResult, r *PutResult) {
// Due to the fact that we may send on a closed channel here, we ignore the panic
defer func() { recover() }()
ch <- r
}