Introduce action render sync

.github/workflows/publish_docker_image.yaml

@@ -6,7 +6,7 @@ on:
       - rc-*
       - hotfix/*
       - release/*
-      - upgrade/libplanet-4.0
+      - feature/issue-117
 env:
   DOCKER_REPO: planetariumhq/market-service
 jobs:

MarketService.Tests/RpcClientTest.cs

@@ -8,6 +8,7 @@
 using Bencodex.Types;
 using Grpc.Core;
 using Lib9c.Model.Order;
+using Lib9c.Renderers;
 using Libplanet.Action.State;
 using Libplanet.Crypto;
 using Libplanet.Types.Assets;
@@ -329,10 +330,15 @@ public RpcClientTest(ITestOutputHelper output)
                 .UseLowerCaseNamingConvention().Options, new DbContextFactorySource<MarketContext>());
 #pragma warning restore EF1001
         var rpcConfigOptions = new RpcConfigOptions {Host = "localhost", Port = 5000};
-        var receiver = new Receiver(new Logger<Receiver>(new LoggerFactory()));
+        var workerOptions = new WorkerOptions
+        {
+            SyncProduct = false,
+            SyncShop = false,
+        };
+        var receiver = new Receiver(new Logger<Receiver>(new LoggerFactory()), new ActionRenderer(), new OptionsWrapper<WorkerOptions>(workerOptions));
         using var logger = _output.BuildLoggerFor<RpcClient>();
         _client = new TestClient(new OptionsWrapper<RpcConfigOptions>(rpcConfigOptions),
-            logger, receiver, _contextFactory, _testService);
+            logger, receiver, _contextFactory, _testService, new ActionRenderer());
     }
 
     [Theory]
@@ -668,8 +674,8 @@ public async Task UpdateProducts(bool legacy)
     private class TestClient : RpcClient
     {
         public TestClient(IOptions<RpcConfigOptions> options, ILogger<RpcClient> logger, Receiver receiver,
-            IDbContextFactory<MarketContext> contextFactory, TestService service) : base(options, logger, receiver,
-            contextFactory)
+            IDbContextFactory<MarketContext> contextFactory, TestService service, ActionRenderer renderer) : base(options, logger, receiver,
+            contextFactory, renderer)
         {
             Service = service;
             var path = "../../../genesis";

MarketService/MarketService.csproj

@@ -31,6 +31,7 @@
     </ItemGroup>
 
     <ItemGroup>
+        <ProjectReference Include="..\lib9c\Lib9c.MessagePack\Lib9c.MessagePack.csproj" />
         <ProjectReference Include="..\lib9c\Lib9c\Lib9c.csproj" />
         <ProjectReference Include="..\MarketService.Response\MarketService.Response.csproj" />
         <ProjectReference Include="..\NineChronicles.RPC.Shared\NineChronicles.RPC.Shared\NineChronicles.RPC.Shared.csproj" />

MarketService/ProductWorker.cs

@@ -54,7 +54,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
             stopWatch.Stop();
             var ts = stopWatch.Elapsed;
             _logger.LogInformation("[ProductWorker]Complete sync product on {BlockIndex}. {TotalElapsed}", _rpcClient.Tip.Index, ts);
-            await Task.Delay(8000, stoppingToken);
+            await Task.Delay(60000 * 5, stoppingToken);
         }
     }
 }

MarketService/Program.cs

@@ -1,4 +1,8 @@
 using System.Text.Json.Serialization;
+using Lib9c.Formatters;
+using Lib9c.Renderers;
+using MessagePack;
+using MessagePack.Resolvers;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.EntityFrameworkCore.Diagnostics;
 
@@ -64,31 +68,43 @@ public void ConfigureServices(IServiceCollection services)
                 .UseLowerCaseNamingConvention()
                 .ConfigureWarnings(w => w.Throw(RelationalEventId.MultipleCollectionIncludeWarning))
         );
-        services.AddSingleton<RpcClient>();
-        services.AddSingleton<Receiver>();
-        services.AddHostedService<RpcService>();
+        services.AddMvc()
+            .AddJsonOptions(
+                options => { options.JsonSerializerOptions.ReferenceHandler = ReferenceHandler.IgnoreCycles; }
+            );
+        var healthChecksBuilder = services.AddHealthChecks()
+            .AddDbContextCheck<MarketContext>();
+
         WorkerOptions workerOptions = new();
         Configuration.GetSection(WorkerOptions.WorkerConfig)
             .Bind(workerOptions);
-        if (workerOptions.SyncShop)
+        var writeMode = workerOptions.SyncProduct || workerOptions.SyncShop;
+        if (writeMode)
         {
-            services.AddHostedService<ShopWorker>();
-        }
+            services
+                .AddHostedService<RpcNodeCheckService>()
+                .AddSingleton<RpcNodeHealthCheck>();
+            services.AddSingleton<RpcClient>();
+            services.AddSingleton<Receiver>();
+            services.AddHostedService<RpcService>();
+            services.AddSingleton<ActionRenderer>();
+            var resolver = MessagePack.Resolvers.CompositeResolver.Create(
+                NineChroniclesResolver.Instance,
+                StandardResolver.Instance
+            );
+            var options = MessagePackSerializerOptions.Standard.WithResolver(resolver);
+            MessagePackSerializer.DefaultOptions = options;
+            if (workerOptions.SyncShop)
+            {
+                services.AddHostedService<ShopWorker>();
+            }
 
-        if (workerOptions.SyncProduct)
-        {
-            services.AddHostedService<ProductWorker>();
+            if (workerOptions.SyncProduct)
+            {
+                services.AddHostedService<ProductWorker>();
+            }
+            healthChecksBuilder.AddCheck<RpcNodeHealthCheck>(nameof(RpcNodeHealthCheck));
         }
-        services.AddMvc()
-            .AddJsonOptions(
-                options => { options.JsonSerializerOptions.ReferenceHandler = ReferenceHandler.IgnoreCycles; }
-            );
-        services
-            .AddHostedService<RpcNodeCheckService>()
-            .AddSingleton<RpcNodeHealthCheck>();
-        services.AddHealthChecks()
-            .AddDbContextCheck<MarketContext>()
-            .AddCheck<RpcNodeHealthCheck>(nameof(RpcNodeHealthCheck));
     }
 
     public void Configure(IApplicationBuilder app, IWebHostEnvironment env)

MarketService/Receiver.cs

@@ -1,6 +1,11 @@
+using System.IO.Compression;
 using Bencodex;
 using Bencodex.Types;
+using Lib9c.Renderers;
 using Libplanet.Types.Blocks;
+using MessagePack;
+using Microsoft.Extensions.Options;
+using Nekoyume.Action;
 using Nekoyume.Shared.Hubs;
 
 namespace MarketService;
@@ -11,14 +16,36 @@ public class Receiver : IActionEvaluationHubReceiver
     public Block PreviousTip;
     private readonly ILogger<Receiver> _logger;
     private readonly Codec _codec = new Codec();
+    private readonly ActionRenderer _actionRenderer;
+    private readonly WorkerOptions _workerOptions;
 
-    public Receiver(ILogger<Receiver> logger)
+    public Receiver(ILogger<Receiver> logger, ActionRenderer actionRenderer, IOptions<WorkerOptions> options)
     {
         _logger = logger;
+        _actionRenderer = actionRenderer;
+        _workerOptions = options.Value;
     }
 
     public void OnRender(byte[] evaluation)
     {
+        if (_workerOptions.SyncProduct || _workerOptions.SyncShop)
+        {
+            using (var cp = new MemoryStream(evaluation))
+            {
+                using (var decompressed = new MemoryStream())
+                {
+                    using (var df = new DeflateStream(cp, CompressionMode.Decompress))
+                    {
+                        df.CopyTo(decompressed);
+                        decompressed.Seek(0, SeekOrigin.Begin);
+                        var dec = decompressed.ToArray();
+                        var ev = MessagePackSerializer.Deserialize<NCActionEvaluation>(dec)
+                            .ToActionEvaluation();
+                        _actionRenderer.ActionRenderSubject.OnNext(ev);
+                    }
+                }
+            }
+        }
     }
 
     public void OnUnrender(byte[] evaluation)