Do eager storage of puts

This commit is contained in:
Chad Retz 2019-02-26 01:10:13 -06:00
parent 3a0fa8ce14
commit 1d987a5e79
3 changed files with 34 additions and 7 deletions

View File

@ -211,17 +211,32 @@ func (g *Gun) startReceiving(peer *Peer) {
}
func (g *Gun) onPeerMessage(ctx context.Context, msg *messageReceived) {
// If we're tracking everything, persist all puts here.
if g.tracking == TrackingEverything {
// TODO:
// * if message-acks are not considered part of a store-all server's storage, then use msg.Ack
// to determine whether we even put here instead of how we do it now.
// * handle gets
// If we're tracking anything, we try to put it (may only be if exists)
if g.tracking != TrackingNothing {
// If we're tracking everything, we persist everything. Otherwise if we're
// only tracking requested, we persist only if it already exists.
putOnlyIfExists := g.tracking == TrackingRequested
for parentSoul, node := range msg.Put {
for field, value := range node.Values {
if state, ok := node.Metadata.State[field]; ok {
// TODO: warn on error or something
g.storage.Put(ctx, parentSoul, field, value, state, false)
// TODO: warn on other error or something
_, err := g.storage.Put(ctx, parentSoul, field, value, state, putOnlyIfExists)
if err == nil {
if msg.storedPuts == nil {
msg.storedPuts = map[string][]string{}
}
msg.storedPuts[parentSoul] = append(msg.storedPuts[parentSoul], field)
}
}
}
}
}
// If there is a listener for this message, use it
if msg.Ack != "" {
g.messageIDListenersLock.RLock()
@ -232,6 +247,7 @@ func (g *Gun) onPeerMessage(ctx context.Context, msg *messageReceived) {
return
}
}
// DAM messages are either requests for our ID or setting of theirs
if msg.DAM != "" {
if msg.PID == "" {
@ -249,6 +265,7 @@ func (g *Gun) onPeerMessage(ctx context.Context, msg *messageReceived) {
}
return
}
// Unhandled message means rebroadcast
g.send(ctx, msg.Message, msg.peer)
}

View File

@ -25,5 +25,5 @@ type messageReceived struct {
*Message
peer *Peer
stored bool
storedPuts map[string][]string
}

View File

@ -137,7 +137,17 @@ func (s *Scoped) fetchRemote(ctx context.Context, ch chan *FetchResult) {
// and only send result if it was an update. Otherwise only do it if we would have done one.
confRes := ConflictResolutionNeverSeenUpdate
if s.gun.tracking == TrackingRequested {
// Wait, wait, we may have already stored this
alreadyStored := false
for _, storedField := range msg.storedPuts[parentSoul] {
if storedField == s.field {
alreadyStored = true
break
}
}
if !alreadyStored {
confRes, r.Err = s.gun.storage.Put(ctx, parentSoul, s.field, newVal, newState, false)
}
} else if lastSeenState > 0 {
confRes = ConflictResolve(lastSeenValue, lastSeenState, newVal, newState, StateNow())
}