lib: Add util.Service as suture.Service template (fixes #5801) (#5806)

lib/relay/client/client.go

@@ -9,6 +9,10 @@ import (
 	"time"
 
 	"github.com/syncthing/syncthing/lib/relay/protocol"
+	"github.com/syncthing/syncthing/lib/sync"
+	"github.com/syncthing/syncthing/lib/util"
+
+	"github.com/thejerf/suture"
 )
 
 type relayClientFactory func(uri *url.URL, certs []tls.Certificate, invitations chan protocol.SessionInvitation, timeout time.Duration) RelayClient
@@ -22,8 +26,7 @@ var (
 )
 
 type RelayClient interface {
-	Serve()
-	Stop()
+	suture.Service
 	Error() error
 	Latency() time.Duration
 	String() string
@@ -39,3 +42,42 @@ func NewClient(uri *url.URL, certs []tls.Certificate, invitations chan protocol.
 
 	return factory(uri, certs, invitations, timeout), nil
 }
+
+type commonClient struct {
+	util.ServiceWithError
+
+	invitations              chan protocol.SessionInvitation
+	closeInvitationsOnFinish bool
+	mut                      sync.RWMutex
+}
+
+func newCommonClient(invitations chan protocol.SessionInvitation, serve func(chan struct{}) error) commonClient {
+	c := commonClient{
+		invitations: invitations,
+		mut:         sync.NewRWMutex(),
+	}
+	newServe := func(stop chan struct{}) error {
+		defer c.cleanup()
+		return serve(stop)
+	}
+	c.ServiceWithError = util.AsServiceWithError(newServe)
+	if c.invitations == nil {
+		c.closeInvitationsOnFinish = true
+		c.invitations = make(chan protocol.SessionInvitation)
+	}
+	return c
+}
+
+func (c *commonClient) cleanup() {
+	c.mut.Lock()
+	if c.closeInvitationsOnFinish {
+		close(c.invitations)
+	}
+	c.mut.Unlock()
+}
+
+func (c *commonClient) Invitations() chan protocol.SessionInvitation {
+	c.mut.RLock()
+	defer c.mut.RUnlock()
+	return c.invitations
+}

lib/relay/client/dynamic.go

@@ -14,45 +14,29 @@ import (
 	"github.com/syncthing/syncthing/lib/osutil"
 	"github.com/syncthing/syncthing/lib/rand"
 	"github.com/syncthing/syncthing/lib/relay/protocol"
-	"github.com/syncthing/syncthing/lib/sync"
 )
 
 type dynamicClient struct {
-	pooladdr                 *url.URL
-	certs                    []tls.Certificate
-	invitations              chan protocol.SessionInvitation
-	closeInvitationsOnFinish bool
-	timeout                  time.Duration
+	commonClient
+
+	pooladdr *url.URL
+	certs    []tls.Certificate
+	timeout  time.Duration
 
-	mut    sync.RWMutex
-	err    error
 	client RelayClient
-	stop   chan struct{}
 }
 
 func newDynamicClient(uri *url.URL, certs []tls.Certificate, invitations chan protocol.SessionInvitation, timeout time.Duration) RelayClient {
-	closeInvitationsOnFinish := false
-	if invitations == nil {
-		closeInvitationsOnFinish = true
-		invitations = make(chan protocol.SessionInvitation)
-	}
-	return &dynamicClient{
-		pooladdr:                 uri,
-		certs:                    certs,
-		invitations:              invitations,
-		closeInvitationsOnFinish: closeInvitationsOnFinish,
-		timeout:                  timeout,
-
-		mut: sync.NewRWMutex(),
+	c := &dynamicClient{
+		pooladdr: uri,
+		certs:    certs,
+		timeout:  timeout,
 	}
+	c.commonClient = newCommonClient(invitations, c.serve)
+	return c
 }
 
-func (c *dynamicClient) Serve() {
-	defer c.cleanup()
-	c.mut.Lock()
-	c.stop = make(chan struct{})
-	c.mut.Unlock()
-
+func (c *dynamicClient) serve(stop chan struct{}) error {
 	uri := *c.pooladdr
 
 	// Trim off the `dynamic+` prefix
@@ -63,8 +47,7 @@ func (c *dynamicClient) Serve() {
 	data, err := http.Get(uri.String())
 	if err != nil {
 		l.Debugln(c, "failed to lookup dynamic relays", err)
-		c.setError(err)
-		return
+		return err
 	}
 
 	var ann dynamicAnnouncement
@@ -72,8 +55,7 @@ func (c *dynamicClient) Serve() {
 	data.Body.Close()
 	if err != nil {
 		l.Debugln(c, "failed to lookup dynamic relays", err)
-		c.setError(err)
-		return
+		return err
 	}
 
 	var addrs []string
@@ -87,22 +69,26 @@ func (c *dynamicClient) Serve() {
 		addrs = append(addrs, ruri.String())
 	}
 
+	defer func() {
+		c.mut.RLock()
+		if c.client != nil {
+			c.client.Stop()
+		}
+		c.mut.RUnlock()
+	}()
+
 	for _, addr := range relayAddressesOrder(addrs) {
 		select {
-		case <-c.stop:
+		case <-stop:
 			l.Debugln(c, "stopping")
-			c.setError(nil)
-			return
+			return nil
 		default:
 			ruri, err := url.Parse(addr)
 			if err != nil {
 				l.Debugln(c, "skipping relay", addr, err)
 				continue
 			}
-			client, err := NewClient(ruri, c.certs, c.invitations, c.timeout)
-			if err != nil {
-				continue
-			}
+			client := newStaticClient(ruri, c.certs, c.invitations, c.timeout)
 			c.mut.Lock()
 			c.client = client
 			c.mut.Unlock()
@@ -115,24 +101,14 @@ func (c *dynamicClient) Serve() {
 		}
 	}
 	l.Debugln(c, "could not find a connectable relay")
-	c.setError(fmt.Errorf("could not find a connectable relay"))
-}
-
-func (c *dynamicClient) Stop() {
-	c.mut.RLock()
-	defer c.mut.RUnlock()
-	close(c.stop)
-	if c.client == nil {
-		return
-	}
-	c.client.Stop()
+	return fmt.Errorf("could not find a connectable relay")
 }
 
 func (c *dynamicClient) Error() error {
 	c.mut.RLock()
 	defer c.mut.RUnlock()
 	if c.client == nil {
-		return c.err
+		return c.Error()
 	}
 	return c.client.Error()
 }
@@ -159,28 +135,6 @@ func (c *dynamicClient) URI() *url.URL {
 	return c.client.URI()
 }
 
-func (c *dynamicClient) Invitations() chan protocol.SessionInvitation {
-	c.mut.RLock()
-	inv := c.invitations
-	c.mut.RUnlock()
-	return inv
-}
-
-func (c *dynamicClient) cleanup() {
-	c.mut.Lock()
-	if c.closeInvitationsOnFinish {
-		close(c.invitations)
-		c.invitations = make(chan protocol.SessionInvitation)
-	}
-	c.mut.Unlock()
-}
-
-func (c *dynamicClient) setError(err error) {
-	c.mut.Lock()
-	c.err = err
-	c.mut.Unlock()
-}
-
 // This is the announcement received from the relay server;
 // {"relays": [{"url": "relay://10.20.30.40:5060"}, ...]}
 type dynamicAnnouncement struct {

lib/relay/client/static.go

@@ -12,88 +12,54 @@ import (
 	"github.com/syncthing/syncthing/lib/dialer"
 	syncthingprotocol "github.com/syncthing/syncthing/lib/protocol"
 	"github.com/syncthing/syncthing/lib/relay/protocol"
-	"github.com/syncthing/syncthing/lib/sync"
 )
 
 type staticClient struct {
-	uri         *url.URL
-	invitations chan protocol.SessionInvitation
+	commonClient
 
-	closeInvitationsOnFinish bool
+	uri *url.URL
 
 	config *tls.Config
 
 	messageTimeout time.Duration
 	connectTimeout time.Duration
 
-	stop    chan struct{}
-	stopped chan struct{}
-	stopMut sync.RWMutex
-
 	conn *tls.Conn
 
-	mut       sync.RWMutex
-	err       error
 	connected bool
 	latency   time.Duration
 }
 
 func newStaticClient(uri *url.URL, certs []tls.Certificate, invitations chan protocol.SessionInvitation, timeout time.Duration) RelayClient {
-	closeInvitationsOnFinish := false
-	if invitations == nil {
-		closeInvitationsOnFinish = true
-		invitations = make(chan protocol.SessionInvitation)
-	}
-
-	stopped := make(chan struct{})
-	close(stopped) // not yet started, don't block on Stop()
-	return &staticClient{
-		uri:         uri,
-		invitations: invitations,
-
-		closeInvitationsOnFinish: closeInvitationsOnFinish,
+	c := &staticClient{
+		uri: uri,
 
 		config: configForCerts(certs),
 
 		messageTimeout: time.Minute * 2,
 		connectTimeout: timeout,
-
-		stop:    make(chan struct{}),
-		stopped: stopped,
-		stopMut: sync.NewRWMutex(),
-
-		mut: sync.NewRWMutex(),
 	}
+	c.commonClient = newCommonClient(invitations, c.serve)
+	return c
 }
 
-func (c *staticClient) Serve() {
-	defer c.cleanup()
-	c.stopMut.Lock()
-	c.stop = make(chan struct{})
-	c.stopped = make(chan struct{})
-	c.stopMut.Unlock()
-	defer close(c.stopped)
-
+func (c *staticClient) serve(stop chan struct{}) error {
 	if err := c.connect(); err != nil {
 		l.Infof("Could not connect to relay %s: %s", c.uri, err)
-		c.setError(err)
-		return
+		return err
 	}
 
 	l.Debugln(c, "connected", c.conn.RemoteAddr())
+	defer c.disconnect()
 
 	if err := c.join(); err != nil {
-		c.conn.Close()
 		l.Infof("Could not join relay %s: %s", c.uri, err)
-		c.setError(err)
-		return
+		return err
 	}
 
 	if err := c.conn.SetDeadline(time.Time{}); err != nil {
-		c.conn.Close()
 		l.Infoln("Relay set deadline:", err)
-		c.setError(err)
-		return
+		return err
 	}
 
 	l.Infof("Joined relay %s://%s", c.uri.Scheme, c.uri.Host)
@@ -106,12 +72,10 @@ func (c *staticClient) Serve() {
 	messages := make(chan interface{})
 	errors := make(chan error, 1)
 
-	go messageReader(c.conn, messages, errors)
+	go messageReader(c.conn, messages, errors, stop)
 
 	timeout := time.NewTimer(c.messageTimeout)
 
-	c.stopMut.RLock()
-	defer c.stopMut.RUnlock()
 	for {
 		select {
 		case message := <-messages:
@@ -122,11 +86,9 @@ func (c *staticClient) Serve() {
 			case protocol.Ping:
 				if err := protocol.WriteMessage(c.conn, protocol.Pong{}); err != nil {
 					l.Infoln("Relay write:", err)
-					c.setError(err)
-					c.disconnect()
-				} else {
-					l.Debugln(c, "sent pong")
+					return err
 				}
+				l.Debugln(c, "sent pong")
 
 			case protocol.SessionInvitation:
 				ip := net.IP(msg.Address)
@@ -137,52 +99,28 @@ func (c *staticClient) Serve() {
 
 			case protocol.RelayFull:
 				l.Infof("Disconnected from relay %s due to it becoming full.", c.uri)
-				c.setError(fmt.Errorf("Relay full"))
-				c.disconnect()
+				return fmt.Errorf("relay full")
 
 			default:
 				l.Infoln("Relay: protocol error: unexpected message %v", msg)
-				c.setError(fmt.Errorf("protocol error: unexpected message %v", msg))
-				c.disconnect()
+				return fmt.Errorf("protocol error: unexpected message %v", msg)
 			}
 
-		case <-c.stop:
+		case <-stop:
 			l.Debugln(c, "stopping")
-			c.setError(nil)
-			c.disconnect()
+			return nil
 
-		// We always exit via this branch of the select, to make sure the
-		// the reader routine exits.
 		case err := <-errors:
-			close(errors)
-			close(messages)
-			c.mut.Lock()
-			if c.connected {
-				c.conn.Close()
-				c.connected = false
-				l.Infof("Disconnecting from relay %s due to error: %s", c.uri, err)
-				c.err = err
-			} else {
-				c.err = nil
-			}
-			c.mut.Unlock()
-			return
+			l.Infof("Disconnecting from relay %s due to error: %s", c.uri, err)
+			return err
 
 		case <-timeout.C:
 			l.Debugln(c, "timed out")
-			c.disconnect()
-			c.setError(fmt.Errorf("timed out"))
+			return fmt.Errorf("timed out")
 		}
 	}
 }
 
-func (c *staticClient) Stop() {
-	c.stopMut.RLock()
-	close(c.stop)
-	<-c.stopped
-	c.stopMut.RUnlock()
-}
-
 func (c *staticClient) StatusOK() bool {
 	c.mut.RLock()
 	con := c.connected
@@ -205,22 +143,6 @@ func (c *staticClient) URI() *url.URL {
 	return c.uri
 }
 
-func (c *staticClient) Invitations() chan protocol.SessionInvitation {
-	c.mut.RLock()
-	inv := c.invitations
-	c.mut.RUnlock()
-	return inv
-}
-
-func (c *staticClient) cleanup() {
-	c.mut.Lock()
-	if c.closeInvitationsOnFinish {
-		close(c.invitations)
-		c.invitations = make(chan protocol.SessionInvitation)
-	}
-	c.mut.Unlock()
-}
-
 func (c *staticClient) connect() error {
 	if c.uri.Scheme != "relay" {
 		return fmt.Errorf("Unsupported relay schema: %v", c.uri.Scheme)
@@ -261,19 +183,6 @@ func (c *staticClient) disconnect() {
 	c.conn.Close()
 }
 
-func (c *staticClient) setError(err error) {
-	c.mut.Lock()
-	c.err = err
-	c.mut.Unlock()
-}
-
-func (c *staticClient) Error() error {
-	c.mut.RLock()
-	err := c.err
-	c.mut.RUnlock()
-	return err
-}
-
 func (c *staticClient) join() error {
 	if err := protocol.WriteMessage(c.conn, protocol.JoinRelayRequest{}); err != nil {
 		return err
@@ -332,13 +241,17 @@ func performHandshakeAndValidation(conn *tls.Conn, uri *url.URL) error {
 	return nil
 }
 
-func messageReader(conn net.Conn, messages chan<- interface{}, errors chan<- error) {
+func messageReader(conn net.Conn, messages chan<- interface{}, errors chan<- error, stop chan struct{}) {
 	for {
 		msg, err := protocol.ReadMessage(conn)
 		if err != nil {
 			errors <- err
 			return
 		}
-		messages <- msg
+		select {
+		case messages <- msg:
+		case <-stop:
+			return
+		}
 	}
 }