felixfoertsch/syncthing
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 10 Wiki Activity

all: Refactor the protocol/model interface a bit (ref #8981) (#9007)

Browse Source
This commit is contained in:
Jakob Borg
2023-07-29 10:24:44 +02:00
committed by GitHub
parent b806026990
commit 9d21b91124
21 changed files with 431 additions and 359 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/protocol/protocol.go
+79 -29
View File
@@ -123,17 +123,28 @@ var (
type Model interface {
// An index was received from the peer device
Index(deviceID DeviceID, folder string, files []FileInfo) error
Index(conn Connection, folder string, files []FileInfo) error
// An index update was received from the peer device
IndexUpdate(deviceID DeviceID, folder string, files []FileInfo) error
IndexUpdate(conn Connection, folder string, files []FileInfo) error
// A request was made by the peer device
Request(deviceID DeviceID, folder, name string, blockNo, size int32, offset int64, hash []byte, weakHash uint32, fromTemporary bool) (RequestResponse, error)
Request(conn Connection, folder, name string, blockNo, size int32, offset int64, hash []byte, weakHash uint32, fromTemporary bool) (RequestResponse, error)
// A cluster configuration message was received
ClusterConfig(deviceID DeviceID, config ClusterConfig) error
ClusterConfig(conn Connection, config ClusterConfig) error
// The peer device closed the connection or an error occurred
Closed(device DeviceID, err error)
Closed(conn Connection, err error)
// The peer device sent progress updates for the files it is currently downloading
DownloadProgress(deviceID DeviceID, folder string, updates []FileDownloadProgressUpdate) error
DownloadProgress(conn Connection, folder string, updates []FileDownloadProgressUpdate) error
}
// contextLessModel is the Model interface, but without the initial
// Connection parameter. Internal use only.
type contextLessModel interface {
Index(folder string, files []FileInfo) error
IndexUpdate(folder string, files []FileInfo) error
Request(folder, name string, blockNo, size int32, offset int64, hash []byte, weakHash uint32, fromTemporary bool) (RequestResponse, error)
ClusterConfig(config ClusterConfig) error
Closed(err error)
DownloadProgress(folder string, updates []FileDownloadProgressUpdate) error
}
type RequestResponse interface {
@@ -146,7 +157,7 @@ type Connection interface {
Start()
SetFolderPasswords(passwords map[string]string)
Close(err error)
ID() DeviceID
DeviceID() DeviceID
Index(ctx context.Context, folder string, files []FileInfo) error
IndexUpdate(ctx context.Context, folder string, files []FileInfo) error
Request(ctx context.Context, folder string, name string, blockNo int, offset int64, size int, hash []byte, weakHash uint32, fromTemporary bool) ([]byte, error)
@@ -171,8 +182,8 @@ type ConnectionInfo interface {
type rawConnection struct {
ConnectionInfo
id DeviceID
receiver Model
deviceID DeviceID
model contextLessModel
startTime time.Time
cr *countingReader
@@ -229,10 +240,16 @@ const (
// Should not be modified in production code, just for testing.
var CloseTimeout = 10 * time.Second
func NewConnection(deviceID DeviceID, reader io.Reader, writer io.Writer, closer io.Closer, receiver Model, connInfo ConnectionInfo, compress Compression, passwords map[string]string, keyGen *KeyGenerator) Connection {
func NewConnection(deviceID DeviceID, reader io.Reader, writer io.Writer, closer io.Closer, model Model, connInfo ConnectionInfo, compress Compression, passwords map[string]string, keyGen *KeyGenerator) Connection {
// We create the wrapper for the model first, as it needs to be passed
// in at the lowest level in the stack. At the end of construction,
// before returning, we add the connection to cwm so that it can be used
// by the model.
cwm := &connectionWrappingModel{model: model}
// Encryption / decryption is first (outermost) before conversion to
// native path formats.
nm := makeNative(receiver)
nm := makeNative(cwm)
em := newEncryptedModel(nm, newFolderKeyRegistry(keyGen, passwords), keyGen)
// We do the wire format conversion first (outermost) so that the
@@ -241,17 +258,18 @@ func NewConnection(deviceID DeviceID, reader io.Reader, writer io.Writer, closer
ec := newEncryptedConnection(rc, rc, em.folderKeys, keyGen)
wc := wireFormatConnection{ec}
cwm.conn = wc
return wc
}
func newRawConnection(deviceID DeviceID, reader io.Reader, writer io.Writer, closer io.Closer, receiver Model, connInfo ConnectionInfo, compress Compression) *rawConnection {
func newRawConnection(deviceID DeviceID, reader io.Reader, writer io.Writer, closer io.Closer, receiver contextLessModel, connInfo ConnectionInfo, compress Compression) *rawConnection {
cr := &countingReader{Reader: reader}
cw := &countingWriter{Writer: writer}
return &rawConnection{
ConnectionInfo: connInfo,
id: deviceID,
receiver: receiver,
deviceID: deviceID,
model: receiver,
cr: cr,
cw: cw,
closer: closer,
@@ -295,8 +313,8 @@ func (c *rawConnection) Start() {
c.startTime = time.Now().Truncate(time.Second)
}
func (c *rawConnection) ID() DeviceID {
return c.id
func (c *rawConnection) DeviceID() DeviceID {
return c.deviceID
}
// Index writes the list of file information to the connected peer device
@@ -462,7 +480,7 @@ func (c *rawConnection) dispatcherLoop() (err error) {
switch msg := msg.(type) {
case *ClusterConfig:
err = c.receiver.ClusterConfig(c.id, *msg)
err = c.model.ClusterConfig(*msg)
case *Index:
err = c.handleIndex(*msg)
@@ -477,7 +495,7 @@ func (c *rawConnection) dispatcherLoop() (err error) {
c.handleResponse(*msg)
case *DownloadProgress:
err = c.receiver.DownloadProgress(c.id, msg.Folder, msg.Updates)
err = c.model.DownloadProgress(msg.Folder, msg.Updates)
}
if err != nil {
return newHandleError(err, msgContext)
@@ -579,13 +597,13 @@ func (c *rawConnection) readHeader(fourByteBuf []byte) (Header, error) {
}
func (c *rawConnection) handleIndex(im Index) error {
l.Debugf("Index(%v, %v, %d file)", c.id, im.Folder, len(im.Files))
return c.receiver.Index(c.id, im.Folder, im.Files)
l.Debugf("Index(%v, %v, %d file)", c.deviceID, im.Folder, len(im.Files))
return c.model.Index(im.Folder, im.Files)
}
func (c *rawConnection) handleIndexUpdate(im IndexUpdate) error {
l.Debugf("queueing IndexUpdate(%v, %v, %d files)", c.id, im.Folder, len(im.Files))
return c.receiver.IndexUpdate(c.id, im.Folder, im.Files)
l.Debugf("queueing IndexUpdate(%v, %v, %d files)", c.deviceID, im.Folder, len(im.Files))
return c.model.IndexUpdate(im.Folder, im.Files)
}
// checkIndexConsistency verifies a number of invariants on FileInfos received in
@@ -651,7 +669,7 @@ func checkFilename(name string) error {
}
func (c *rawConnection) handleRequest(req Request) {
res, err := c.receiver.Request(c.id, req.Folder, req.Name, int32(req.BlockNo), int32(req.Size), req.Offset, req.Hash, req.WeakHash, req.FromTemporary)
res, err := c.model.Request(req.Folder, req.Name, int32(req.BlockNo), int32(req.Size), req.Offset, req.Hash, req.WeakHash, req.FromTemporary)
if err != nil {
c.send(context.Background(), &Response{
ID: req.ID,
@@ -925,7 +943,7 @@ func (c *rawConnection) internalClose(err error) {
c.closeOnce.Do(func() {
l.Debugln("close due to", err)
if cerr := c.closer.Close(); cerr != nil {
l.Debugln(c.id, "failed to close underlying conn:", cerr)
l.Debugln(c.deviceID, "failed to close underlying conn:", cerr)
}
close(c.closed)
@@ -940,7 +958,7 @@ func (c *rawConnection) internalClose(err error) {
<-c.dispatcherLoopStopped
c.receiver.Closed(c.ID(), err)
c.model.Closed(err)
})
}
@@ -958,11 +976,11 @@ func (c *rawConnection) pingSender() {
case <-ticker.C:
d := time.Since(c.cw.Last())
if d < PingSendInterval/2 {
l.Debugln(c.id, "ping skipped after wr", d)
l.Debugln(c.deviceID, "ping skipped after wr", d)
continue
}
l.Debugln(c.id, "ping -> after", d)
l.Debugln(c.deviceID, "ping -> after", d)
c.ping()
case <-c.closed:
@@ -983,11 +1001,11 @@ func (c *rawConnection) pingReceiver() {
case <-ticker.C:
d := time.Since(c.cr.Last())
if d > ReceiveTimeout {
l.Debugln(c.id, "ping timeout", d)
l.Debugln(c.deviceID, "ping timeout", d)
c.internalClose(ErrTimeout)
}
l.Debugln(c.id, "last read within", d)
l.Debugln(c.deviceID, "last read within", d)
case <-c.closed:
return
@@ -1068,3 +1086,35 @@ func messageContext(msg message) (string, error) {
return "", errors.New("unknown or empty message")
}
}
// connectionWrappingModel takes the Model interface from the model package,
// which expects the Connection as the first parameter in all methods, and
// wraps it to conform to the protocol.contextLessModel interface.
type connectionWrappingModel struct {
conn Connection
model Model
}
func (c *connectionWrappingModel) Index(folder string, files []FileInfo) error {
return c.model.Index(c.conn, folder, files)
}
func (c *connectionWrappingModel) IndexUpdate(folder string, files []FileInfo) error {
return c.model.IndexUpdate(c.conn, folder, files)
}
func (c *connectionWrappingModel) Request(folder, name string, blockNo, size int32, offset int64, hash []byte, weakHash uint32, fromTemporary bool) (RequestResponse, error) {
return c.model.Request(c.conn, folder, name, blockNo, size, offset, hash, weakHash, fromTemporary)
}
func (c *connectionWrappingModel) ClusterConfig(config ClusterConfig) error {
return c.model.ClusterConfig(c.conn, config)
}
func (c *connectionWrappingModel) Closed(err error) {
c.model.Closed(c.conn, err)
}
func (c *connectionWrappingModel) DownloadProgress(folder string, updates []FileDownloadProgressUpdate) error {
return c.model.DownloadProgress(c.conn, folder, updates)
}