SyncFramework – Adding network support

by Joche Ojeda | Oct 17, 2021 | Data Synchronization

So far, all our test exists inside the same process, so they communicate through variables, in real-life scenarios nodes won’t exist in the same process and most of the time not even in the same location.

The easiest and most standard way to implement client-server communication in the dotnet world is a rest API, so let’s define an API client and a server service, lets start with the API client

ISyncFrameworkClient

https://github.com/egarim/SyncFramework/tree/main/src/BIT.Data.Sync/Client/ISyncFrameworkClient.cs

public interface ISyncFrameworkClient
{
    Task<List<Delta>> FetchAsync(Guid startindex, string identity, CancellationToken cancellationToken);
    Task PushAsync(IEnumerable<IDelta> Deltas, CancellationToken cancellationToken);
}

As you can see in the code above, this interface is really simple because the client only implements 2 operations

• Fetch: downloads the deltas from the server
• Push: upload the deltas from our delta store to the server

Now we should create an implementation for this interface, see the code below

https://github.com/egarim/SyncFramework/tree/main/src/BIT.Data.Sync/Client/SyncFrameworkHttpClient.cs

public class SyncFrameworkHttpClient : ISyncFrameworkClient
{
    HttpClient _httpClient;
    public string DeltaStoreId { get; }
    public SyncFrameworkHttpClient(HttpClient httpClient,string NodeId)
    {
        this.DeltaStoreId = NodeId;
        _httpClient = httpClient;
        _httpClient.DefaultRequestHeaders.Add("NodeId", NodeId);
      
        this.DeltaStoreId = NodeId;
    }
    public SyncFrameworkHttpClient(string BaseAddress, string DeltaStoreId):this(new HttpClient() { BaseAddress=new Uri(BaseAddress)},DeltaStoreId)
    {
       
    }
    public virtual async Task PushAsync(IEnumerable<IDelta> Deltas, CancellationToken cancellationToken  = default)
    {
       
        try
        {
            List<Delta> toserialzie = new List<Delta>();
            foreach (IDelta delta in Deltas)
            {
                toserialzie.Add(new Delta(delta));
            }
            cancellationToken.ThrowIfCancellationRequested();

            DataContractJsonSerializer js = new DataContractJsonSerializer(typeof(List<Delta>));
            MemoryStream msObj = new MemoryStream();
            js.WriteObject(msObj, toserialzie);
            msObj.Position = 0;
            StreamReader sr = new StreamReader(msObj);
            string jsonDeltas = sr.ReadToEnd();

            var data = new StringContent(jsonDeltas, Encoding.UTF8, "application/json");
            await _httpClient.PostAsync("/Sync/Push", data, cancellationToken).ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            var message = ex.Message;
            throw;
        }
      
    }
    public virtual async Task<List<Delta>> FetchAsync(Guid startindex, string identity, CancellationToken cancellationToken = default)
    {
        var QueryParams = new Dictionary<string, string>();
        QueryParams.Add(nameof(startindex), startindex.ToString());
        QueryParams.Add(nameof(identity), identity);
        cancellationToken.ThrowIfCancellationRequested();
        var query = HttpUtility.ParseQueryString("");
        foreach (KeyValuePair<string, string> CurrentParam in QueryParams)
        {
            query[CurrentParam.Key] = CurrentParam.Value;
        }
        var reponse = await _httpClient.GetStringAsync($"/Sync/Fetch?{query.ToString()}").ConfigureAwait(false);

        using (var ms = new MemoryStream(Encoding.Unicode.GetBytes(reponse)))
        {

            DataContractJsonSerializer deserializer = new DataContractJsonSerializer(typeof(List<Delta>));
            List<Delta> Deltas = (List<Delta>)deserializer.ReadObject(ms);

            return Deltas;
        }

        //List<Delta> Deltas = JsonConvert.DeserializeObject<List<Delta>>(reponse);
        return null;

    }

}

it’s an implementation of the ISyncFrameworkClient interface using HTTP communication

• Fetch: uses an HTTP get request
• Push: uses an HTTP post request

Also, the “nodeid” header is added to the request, you will understand why when we implement the server part.

Now that we have defined the contract for the client and also provided the base implementation using an HTTP client, its time to define what a client node is, please take a look at the code below

ISyncClientNodeExtensions

https://github.com/egarim/SyncFramework/tree/main/src/BIT.Data.Sync/Client/ISyncClientNodeExtensions.cs

    public static class ISyncClientNodeExtensions
    {

        public static async Task<List<Delta>> FetchAsync(this ISyncClientNode instance, CancellationToken cancellationToken = default)
        {
            cancellationToken.ThrowIfCancellationRequested();
            var LastDetalIndex = await instance.DeltaStore.GetLastProcessedDeltaAsync(cancellationToken).ConfigureAwait(false);
            return await instance.SyncFrameworkClient.FetchAsync(LastDetalIndex, instance.Identity, cancellationToken).ConfigureAwait(false);
        }
        public static async Task PullAsync(this ISyncClientNode instance, CancellationToken cancellationToken = default)
        {
            cancellationToken.ThrowIfCancellationRequested();
            var Deltas = await instance.FetchAsync(cancellationToken).ConfigureAwait(false);
            if (Deltas.Any())
            {
                await instance.DeltaProcessor.ProcessDeltasAsync(Deltas, cancellationToken).ConfigureAwait(false);
                Guid index = Deltas.Max(d => d.Index);
                await instance.DeltaStore.SetLastProcessedDeltaAsync(index, cancellationToken).ConfigureAwait(false);
            }

        }
        public static async Task PushAsync(this ISyncClientNode instance, CancellationToken cancellationToken = default)
        {
            cancellationToken.ThrowIfCancellationRequested();
            var LastPushedDelta = await instance.DeltaStore.GetLastPushedDeltaAsync(cancellationToken).ConfigureAwait(false);
            var Deltas = await instance.DeltaStore.GetDeltasAsync(LastPushedDelta,cancellationToken).ConfigureAwait(false);
            if (Deltas.Any())
            {
                var Max = Deltas.Max(d => d.Index);
                await instance.SyncFrameworkClient.PushAsync(Deltas, cancellationToken).ConfigureAwait(false);
                await instance.DeltaStore.SetLastPushedDeltaAsync(Max,cancellationToken).ConfigureAwait(false);
            }



        }
    }

so, this is how the SyncClientNode is structured

Let’s move to the server-side now, here the idea is to be able to host multiple delta store and delta processors and also to be able to introduce custom logic either saving the deltas into the delta store or processing the deltas into a data object

ISyncServerNode

https://github.com/egarim/SyncFramework/blob/main/src/BIT.Data.Sync/Server/ISyncServerNode.cs

public interface ISyncServerNode
{
    string NodeId { get; set; }
    Task SaveDeltasAsync(IEnumerable<IDelta> deltas, CancellationToken cancellationToken);
    Task<IEnumerable<IDelta>> GetDeltasAsync(Guid startindex, string identity, CancellationToken cancellationToken);
    Task ProcessDeltasAsync(IEnumerable<IDelta> deltas, CancellationToken cancellationToken);
}

Now we need to define a server, so here is the interface for the SyncServer

ISyncServer

https://github.com/egarim/SyncFramework/blob/main/src/BIT.Data.Sync/Server/ISyncServer.cs

public interface ISyncServer
{
    IEnumerable<ISyncServerNode> Nodes { get; }
    Task<IEnumerable<IDelta>> GetDeltasAsync(string name, Guid startindex, string identity, CancellationToken cancellationToken);
    Task ProcessDeltasAsync(string Name, IEnumerable<IDelta> deltas, CancellationToken cancellationToken);
    Task SaveDeltasAsync(string name, IEnumerable<IDelta> deltas, CancellationToken cancellationToken);
}

As you can see, the members are almost the same as the sync node, this design allows us to have more than one node on the server-side

Here is the implementation of the SyncServer

SyncServer

https://github.com/egarim/SyncFramework/blob/main/src/BIT.Data.Sync/Server/SyncServer.cs

public class SyncServer: ISyncServer
{


    IEnumerable<ISyncServerNode> _Nodes;
   
    public SyncServer(params ISyncServerNode[] Nodes)
    {
        this._Nodes = Nodes;
    }

    public IEnumerable<ISyncServerNode> Nodes => _Nodes;

   

    public async Task<IEnumerable<IDelta>> GetDeltasAsync(string NodeId, Guid startindex, string identity, CancellationToken cancellationToken)
    {
        cancellationToken.ThrowIfCancellationRequested();
        ISyncServerNode Node = GetNode(NodeId);
        if (Node != null)
        {
            return await Node.GetDeltasAsync(startindex, identity, cancellationToken).ConfigureAwait(false);
        }

        IEnumerable<IDelta> result = new List<IDelta>();
        return result;
    }
    public Task ProcessDeltasAsync(string NodeId, IEnumerable<IDelta> deltas, CancellationToken cancellationToken)
    {
        cancellationToken.ThrowIfCancellationRequested();
        ISyncServerNode Node = GetNode(NodeId);
        if (Node != null)
        {
            return Node.ProcessDeltasAsync(deltas, cancellationToken);
        }
        return null;

    }

    private ISyncServerNode GetNode(string NodeId)
    {
        return Nodes.FirstOrDefault(node => node.NodeId == NodeId);
    }

    public Task SaveDeltasAsync(string NodeId, IEnumerable<IDelta> deltas, CancellationToken cancellationToken)
    {
        cancellationToken.ThrowIfCancellationRequested();

        ISyncServerNode Node = GetNode(NodeId);

        if (Node != null)
        {
            return Node.SaveDeltasAsync(deltas, cancellationToken);
        }
        return Task.CompletedTask;
    }
}

the following pictures show the 2 possible server implementations

and that’s it for this post, in the next post I will show the test cases for this implementation

SyncFramework – Planning the first implementation

by Joche Ojeda | Oct 12, 2021 | Data Synchronization

Well, it’s time to create our first implementation, first, we need a place to store the deltas generated in the process of tracking changes in a data object.

To keep the Implementation simple, we will create a delta store that saves the deltas in memory. This delta store can also be used for testing purposes

MemoryDeltaStore

https://github.com/egarim/SyncFramework/blob/main/src/BIT.Data.Sync/Imp/MemoryDeltaStore.cs

public class MemoryDeltaStore : BIT.Data.Sync.DeltaStoreBase
   {
       IList<IDelta> Deltas;

       public MemoryDeltaStore(IEnumerable<IDelta> Deltas)
       {
           this.Deltas = new List<IDelta>(Deltas);

       }


       protected MemoryDeltaStore()
       {

       }
       //TODO fix the use of MemoryDb
       public MemoryDeltaStore(DeltaStoreSettings deltaStoreSettings) : base(deltaStoreSettings)
       {

       }

       public async override Task SaveDeltasAsync(IEnumerable<IDelta> deltas, CancellationToken cancellationToken = default)
       {
           cancellationToken.ThrowIfCancellationRequested();
           foreach (IDelta delta in deltas)
           {
               cancellationToken.ThrowIfCancellationRequested();
               Deltas.Add(new Delta(delta));
           }
       }

       public override Task<IEnumerable<IDelta>> GetDeltasFromOtherNodes(Guid startindex, string identity, CancellationToken cancellationToken = default)
       {
           cancellationToken.ThrowIfCancellationRequested();
           var result = Deltas.Where(d => d.Index.CompareTo(startindex) > 0 && string.Compare(d.Identity, identity, StringComparison.Ordinal) != 0);
           return Task.FromResult(result.Cast<IDelta>());
       }
       public override Task<IEnumerable<IDelta>> GetDeltasAsync(Guid startindex, CancellationToken cancellationToken = default)
       {
           cancellationToken.ThrowIfCancellationRequested();
           return Task.FromResult(Deltas.Where(d => d.Index.CompareTo(startindex) > 0).ToList().Cast<IDelta>());
       }
       Guid LastProcessedDelta;
       public override async Task<Guid> GetLastProcessedDeltaAsync(CancellationToken cancellationToken = default)
       {
           return LastProcessedDelta;
       }

       public override async Task SetLastProcessedDeltaAsync(Guid Index, CancellationToken cancellationToken = default)
       {
           cancellationToken.ThrowIfCancellationRequested();
           LastProcessedDelta = Index;


       }

      
       Guid LastPushedDelta;
       public async override Task<Guid> GetLastPushedDeltaAsync(CancellationToken cancellationToken)
       {
           return LastPushedDelta;
       }

       public async override Task SetLastPushedDeltaAsync(Guid Index, CancellationToken cancellationToken = default)
       {
           cancellationToken.ThrowIfCancellationRequested();
           LastPushedDelta = Index;


       }

       public async override Task<int> GetDeltaCountAsync(Guid startindex, CancellationToken cancellationToken = default)
       {
           cancellationToken.ThrowIfCancellationRequested();
           return Deltas.Count(d => d.Index.CompareTo(startindex) > 0);
       }

       public async override Task PurgeDeltasAsync(CancellationToken cancellationToken = default)
       {
           cancellationToken.ThrowIfCancellationRequested();
           Deltas.Clear();
          

       }
   }

Now that we have a delta store in place, we need a data object, something that we can use to generate data and track how the data is changing, so again for test purposes, I have implemented a small in-memory database

SimpleDatabase

https://github.com/egarim/SyncFramework/blob/main/src/BIT.Data.Sync/Imp/SimpleDatabase.cs

public class SimpleDatabase 
   {
       public IDeltaProcessor DeltaProcessor { get; set; }
       public string Identity { get; set; }
       public IDeltaStore DeltaStore { get; set; }
       public SimpleDatabase(IDeltaStore deltaStore, string identity,  List<SimpleDatabaseRecord> Data)
       {
         
           Identity = identity;
           DeltaStore = deltaStore;
           this.Data= Data;
       }
       List<SimpleDatabaseRecord> Data;
       public async void Update(SimpleDatabaseRecord Instance)
       {
           var ObjectToUpdate = Data.FirstOrDefault(x => x.Key == Instance.Key);
           if (ObjectToUpdate != null)
           {
               var Index = Data.IndexOf(ObjectToUpdate);
               Data[Index] = Instance;
               SimpleDatabaseModification item = new SimpleDatabaseModification(OperationType.Update, Instance);
               await SaveDelta(item);
           }
         
       }

       private async Task SaveDelta(SimpleDatabaseModification item)
       {
           var Delta = DeltaStore.CreateDelta(Identity,item);
           await DeltaStore.SaveDeltasAsync(new List<IDelta>() { Delta }, default);
       }

       public void Delete(SimpleDatabaseRecord Instance)
       {
           var ObjectToDelete=  Data.FirstOrDefault(x=>x.Key==Instance.Key);
           if(ObjectToDelete!=null)
           {
               Data.Remove(ObjectToDelete);
              
           }
          
       }
       public async Task Add(SimpleDatabaseRecord Instance)
       {
           Data.Add(Instance);
          
           SimpleDatabaseModification item = new SimpleDatabaseModification(OperationType.Add, Instance);
           await SaveDelta(item);
       }
   }

In the class above I have implemented methods to add, delete and update a record. Inside each method I create an instance of an object called SimpleDatabaseModification, I used that object to keep track of which operation is happening and keep a copy of the instance being handle at the moment, that is what we are going to save as a delta.

SimpleDatabaseModification

https://github.com/egarim/SyncFramework/blob/main/src/BIT.Data.Sync/Imp/SimpleDatabaseModification.cs

     public class SimpleDatabaseModification
    {
        public OperationType Operation { get; set; }
        public SimpleDatabaseModification(OperationType operation, SimpleDatabaseRecord record)
        {
            Operation = operation;
            Record = record;
        }
        public SimpleDatabaseRecord Record { get; set; }
    }

Now since the SimpleDatabase is saving the records on a list the next step is to create a processor that gets the information out of the delta and use it to recreate that list, so here is the delta processor

SimpleDatabaseDeltaProcessor

https://github.com/egarim/SyncFramework/blob/main/src/BIT.Data.Sync/Imp/SimpleDatabaseDeltaProcessor.cs

    public class SimpleDatabaseDeltaProcessor :DeltaProcessorBase 
    {
        
        List<SimpleDatabaseRecord> _CurrentText;
        public SimpleDatabaseDeltaProcessor(DeltaStoreSettings deltaStoreSettings, List<SimpleDatabaseRecord> CurrentData) : base(deltaStoreSettings)
        {
            _CurrentText= CurrentData;
        }
        public override Task ProcessDeltasAsync(IEnumerable<IDelta> deltas, CancellationToken cancellationToken)
        {
           
            cancellationToken.ThrowIfCancellationRequested();
            foreach (IDelta delta in deltas)
            {
                cancellationToken.ThrowIfCancellationRequested();
                var Modification= this.GetDeltaOperations<SimpleDatabaseModification>(delta);
                switch (Modification.Operation)
                {
                    case OperationType.Add:
                        this._CurrentText.Add(Modification.Record);
                        break;
                    case OperationType.Delete:
                        var ObjectToDelete=  this._CurrentText.FirstOrDefault(x=>x.Key==Modification.Record.Key);
                        this._CurrentText.Remove(ObjectToDelete);
                        break;
                    case OperationType.Update:
                        var ObjectToUpdate = this._CurrentText.FirstOrDefault(x => x.Key == Modification.Record.Key);
                        var Index= this._CurrentText.IndexOf(ObjectToUpdate);
                        this._CurrentText[Index] = Modification.Record;
                        break;
                }
              
                
            }
            return Task.CompletedTask;
            
        }
    }

Well, that is for this post, in the next post we will create some test scenarios to test our implementations