smegmesh/pkg/sync/syncservice.go

99 lines
1.9 KiB
Go
Raw Permalink Normal View History

2023-10-10 21:14:40 +02:00
// sync merges shared state between two nodes
package sync
import (
"context"
"errors"
2023-10-23 19:13:08 +02:00
"io"
2023-10-10 21:14:40 +02:00
"github.com/tim-beatham/wgmesh/pkg/ctrlserver"
"github.com/tim-beatham/wgmesh/pkg/mesh"
2023-10-10 21:14:40 +02:00
"github.com/tim-beatham/wgmesh/pkg/rpc"
)
type SyncServiceImpl struct {
rpc.UnimplementedSyncServiceServer
Server *ctrlserver.MeshCtrlServer
2023-10-10 21:14:40 +02:00
}
// GetMesh: Gets a nodes local mesh configuration as a CRDT
func (s *SyncServiceImpl) GetConf(context context.Context, request *rpc.GetConfRequest) (*rpc.GetConfReply, error) {
mesh := s.Server.MeshManager.GetMesh(request.MeshId)
2023-10-10 21:14:40 +02:00
if mesh == nil {
return nil, errors.New("mesh does not exist")
}
meshBytes := mesh.Save()
reply := rpc.GetConfReply{
Mesh: meshBytes,
}
return &reply, nil
}
// Sync: Pings a node and syncs the mesh configuration with the other node
2023-10-23 19:13:08 +02:00
// SyncMesh: syncs the two streams changes
func (s *SyncServiceImpl) SyncMesh(stream rpc.SyncService_SyncMeshServer) error {
var meshId = ""
var syncer mesh.MeshSyncer = nil
2023-10-10 21:14:40 +02:00
2023-10-23 19:13:08 +02:00
for {
in, err := stream.Recv()
2023-10-10 21:14:40 +02:00
2023-10-23 19:13:08 +02:00
if err == io.EOF {
if syncer != nil {
syncer.Complete()
}
2023-10-23 19:13:08 +02:00
return nil
}
2023-10-10 21:14:40 +02:00
2023-10-23 19:13:08 +02:00
if err != nil {
return err
}
2023-10-10 21:14:40 +02:00
2023-10-23 19:13:08 +02:00
if len(meshId) == 0 {
meshId = in.MeshId
mesh := s.Server.MeshManager.GetMesh(meshId)
if mesh == nil {
return errors.New("mesh does not exist")
}
2023-10-10 21:14:40 +02:00
2023-10-23 19:13:08 +02:00
syncer = mesh.GetSyncer()
} else if meshId != in.MeshId {
return errors.New("differing meshids")
2023-10-23 19:13:08 +02:00
}
if syncer == nil {
return errors.New("syncer should not be nil")
2023-10-23 19:13:08 +02:00
}
msg, moreMessages := syncer.GenerateMessage()
if err = stream.Send(&rpc.SyncMeshReply{Success: true, Changes: msg}); err != nil {
return err
}
if len(in.Changes) != 0 {
if err = syncer.RecvMessage(in.Changes); err != nil {
return err
}
}
if !moreMessages || err == io.EOF {
if syncer != nil {
syncer.Complete()
}
2023-10-23 19:13:08 +02:00
return nil
}
}
}
func NewSyncService(server *ctrlserver.MeshCtrlServer) *SyncServiceImpl {
return &SyncServiceImpl{Server: server}
}