add heartbeat logic to dmsg clients

pkg/dmsg/client.go

@@ -60,7 +60,7 @@ func (c *Config) Ensure() {
 func DefaultConfig() *Config {
 	conf := &Config{
 		MinSessions:    DefaultMinSessions,
-		UpdateInterval: DefaultUpdateInterval,
+		UpdateInterval: DefaultUpdateInterval * 5,
 	}
 	return conf
 }
@@ -153,6 +153,8 @@ func (ce *Client) Serve(ctx context.Context) {
 		}
 	}(cancellabelCtx)
 
+	updateEntryLoopOnce := new(sync.Once)
+
 	for {
 		if isClosed(ce.done) {
 			return
@@ -253,6 +255,10 @@ func (ce *Client) Serve(ctx context.Context) {
 				ce.serveWait()
 			}
 		}
+
+		// Only start the update entry loop once we have at least one session established.
+		updateEntryLoopOnce.Do(func() { go ce.updateClientEntryLoop(cancellabelCtx, ce.done, ce.conf.ClientType) })
+
 		// We dial all servers and wait for error or done signal.
 		select {
 		case <-ce.done:

pkg/dmsg/entity_common.go

@@ -251,12 +251,50 @@ func (c *EntityCommon) updateClientEntry(ctx context.Context, done chan struct{}
 		}
 		return c.dc.PostEntry(ctx, entry)
 	}
+
+	// Whether the client's CURRENT delegated servers is the same as what would be advertised.
+	sameSrvPKs := cipher.SamePubKeys(srvPKs, entry.Client.DelegatedServers)
+
+	// No update is needed if delegated servers has no delta, and an entry update is not due.
+	if _, due := c.updateIsDue(); sameSrvPKs && !due {
+		return nil
+	}
+
 	entry.ClientType = clientType
 	entry.Client.DelegatedServers = srvPKs
 	c.log.WithField("entry", entry).Debug("Updating entry.")
 	return c.dc.PutEntry(ctx, c.sk, entry)
 }
 
+func (c *EntityCommon) updateClientEntryLoop(ctx context.Context, done chan struct{}, clientType string) {
+	t := time.NewTimer(c.updateInterval)
+	defer t.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+
+		case <-t.C:
+			if lastUpdate, due := c.updateIsDue(); !due {
+				t.Reset(c.updateInterval - time.Since(lastUpdate))
+				continue
+			}
+
+			c.sessionsMx.Lock()
+			err := c.updateClientEntry(ctx, done, clientType)
+			c.sessionsMx.Unlock()
+
+			if err != nil {
+				c.log.WithError(err).Warn("Failed to update discovery entry.")
+			}
+
+			// Ensure we trigger another update within given 'updateInterval'.
+			t.Reset(c.updateInterval)
+		}
+	}
+}
+
 func (c *EntityCommon) delEntry(ctx context.Context) (err error) {
 
 	entry, err := c.dc.Entry(ctx, c.pk)