add: ping service to sdk

src/cs/Bts.Spark.Api/Services/SparkApiServices.cs

@@ -8,6 +8,7 @@ using bts.api.core;
 using bts.api.orderman;
 using bts.api.position;
 using bts.co2;
+using bts.metrics;
 using bts.md;
 
 using Microsoft.Extensions.Configuration;
@@ -86,6 +87,7 @@ public static class SparkApiServices {
          services.AddSparkdService<OrderService>();
          services.AddSingleton<IOrdersClient, OrdersClient>();
       }
+      services.AddSparkdService<ShellLatencyService>();
       if (configurationSection.GetValue("EnablePosition", true)) {
          services.TryAddSingleton<ISecurityResolver>(ActivatorUtilities.GetServiceOrCreateInstance<PassiveSecurityResolver>);
          services.AddSparkdService<PositionService>()
@@ -176,6 +178,7 @@ public static class SparkApiServices {
          services.AddSparkdService<OrderService>();
          services.AddSingleton<IOrdersClient, OrdersClient>();
       }
+      services.AddSparkdService<ShellLatencyService>();
       if (isPositionsEnabled) {
          services.AddSparkdService<PositionService>();
       }

src/cs/Bts.Spark.Sdk/AuthenticatedSimpleSparkClient.cs

@@ -3,6 +3,7 @@ namespace Bts.Spark.Sdk;
 using bts.api;
 using bts.api.core;
 using bts.api.orderman;
+using bts.metrics;
 
 using global::Spark.Api.Admin;
 using global::Spark.Api.MarketData;
@@ -28,6 +29,7 @@ public class AuthenticatedSimpleSparkClient : ISimpleSparkClient, IHostedService
    readonly IOrdersClient ordersClient;
    readonly IAdminClient adminClient;
    readonly ISecurityResolver securityResolver;
+   readonly ShellLatencyService shellLatencyService;
    readonly ISparkLoginService loginService;
    readonly SimpleSparkClientConfig config;
    readonly ILogger<AuthenticatedSimpleSparkClient> logger;
@@ -39,6 +41,7 @@ public class AuthenticatedSimpleSparkClient : ISimpleSparkClient, IHostedService
       IOrdersClient ordersClient,
       IAdminClient adminClient,
       ISecurityResolver securityResolver,
+      ShellLatencyService shellLatencyService,
       ISparkLoginService loginService,
       IOptions<SimpleSparkClientConfig> config,
       ILogger<AuthenticatedSimpleSparkClient> logger) {
@@ -46,6 +49,7 @@ public class AuthenticatedSimpleSparkClient : ISimpleSparkClient, IHostedService
       this.ordersClient = ordersClient;
       this.adminClient = adminClient;
       this.securityResolver = securityResolver;
+      this.shellLatencyService = shellLatencyService;
       this.loginService = loginService;
       this.config = config.Value;
       this.logger = logger;
@@ -61,7 +65,7 @@ public class AuthenticatedSimpleSparkClient : ISimpleSparkClient, IHostedService
       }
       cancellationToken.ThrowIfCancellationRequested();
       var user = loginRes.user.Value;
-      simpleClient = new(user.name, user.id, mdClient, ordersClient, adminClient, securityResolver, config.OrderPoolCapacity, config.OrderPoolReplenishInterval);
+      simpleClient = new(user.name, user.id, mdClient, ordersClient, shellLatencyService, adminClient, securityResolver, config.OrderPoolCapacity, config.OrderPoolReplenishInterval);
       simpleClientSubject.OnNext(simpleClient);
       logger.LogInformation("SimpleSparkClient started");
    }
@@ -97,6 +101,9 @@ public class AuthenticatedSimpleSparkClient : ISimpleSparkClient, IHostedService
    public IObservable<TableUpdate<PriceTicker, uint>> PriceTicker(SecurityCriteria? securityCriteria) =>
       simpleClient?.PriceTicker(securityCriteria) ?? simpleClientSubject.Take(1).Select(c => c.PriceTicker(securityCriteria)).FlattenSeries();
 
+   public Task<bool> PingAsync() =>
+      simpleClient?.PingAsync() ?? pingBeforeStartAsync();
+
    public Task<Response> CreateOrder(OrderDefinition orderDef, double quantity, double price) =>
       orderOp(cli => cli.CreateOrder(orderDef, quantity, price));
 
@@ -134,4 +141,15 @@ public class AuthenticatedSimpleSparkClient : ISimpleSparkClient, IHostedService
       async Task<T> LateImpl() =>
          await op(await simpleClientSubject.FirstAsync());
    }
+
+   async Task<bool> pingBeforeStartAsync() {
+      var request = new BufferMessage() {
+         Sequence = (ulong)DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
+         Data = "spark-sdk-ping"
+      };
+      var result = await shellLatencyService.ping_async(request);
+      return result.success &&
+         result.value.Sequence == request.Sequence &&
+         result.value.Data == request.Data;
+   }
 }

src/cs/Bts.Spark.Sdk/ExecutionFacadeSimpleSparkClient.cs

@@ -48,6 +48,9 @@ public class ExecutionFacadeSimpleSparkClient : ISimpleSparkClient {
    public IObservable<TableUpdate<PriceTicker, uint>> PriceTicker(SecurityCriteria? securityCriteria) =>
       sparkClient.PriceTicker(securityCriteria);
 
+   public Task<bool> PingAsync() =>
+      sparkClient.PingAsync();
+
    public Task<Response> CreateOrder(OrderDefinition orderDef, double quantity, double price) {
       logger.LogInformation("CreateOrder {account} {tag} {securityId} {side} {quantity} @ {price}", orderDef.Account, orderDef.Tag, orderDef.SecurityKey, orderDef.Side, quantity, price);
       return Task.FromResult(facadeResponse());

src/cs/Bts.Spark.Sdk/ISimpleSparkClient.cs

@@ -12,6 +12,8 @@ using mbts.spark;
 using System;
 
 public interface ISimpleSparkClient {
+   Task<bool> PingAsync();
+
    // Market Data
    IObservable<TableUpdate<SecurityImmutable, uint>> Securities(SecurityCriteria? securityCriteria);
    IObservable<TableUpdate<BookRecord, uint>> Books(SecurityCriteria? securityCriteria);

src/cs/Bts.Spark.Sdk/SimpleSparkClient.cs

@@ -16,12 +16,14 @@ using global::Spark.Core.Disposables;
 using global::Spark.Core.Reactive;
 
 using mbts.spark;
+using bts.metrics;
 
 using System.Reactive.Linq;
 
 public class SimpleSparkClient : ISimpleSparkClient, IDisposable {
    readonly IMarketDataClient mdClient;
    readonly IOrdersClient ordersClient;
+   readonly ShellLatencyService shellLatencyService;
    readonly OrderPool limitOrderPool;
    readonly OrderPool marketOrderPool;
    readonly TableUpdateDemuxer<OrderRecordTupleWithRoot, uint, uint> orderSecurityDmx;
@@ -46,6 +48,7 @@ public class SimpleSparkClient : ISimpleSparkClient, IDisposable {
       uint userId,
       IMarketDataClient mdClient,
       IOrdersClient ordersClient,
+      ShellLatencyService shellLatencyService,
       IAdminClient adminClient,
       ISecurityResolver securityResolver,
       int orderPoolCapacity,
@@ -53,6 +56,7 @@ public class SimpleSparkClient : ISimpleSparkClient, IDisposable {
    ) {
       this.mdClient = mdClient;
       this.ordersClient = ordersClient;
+      this.shellLatencyService = shellLatencyService;
       limitOrderPool = new(orderPoolCapacity, () => new() { User = userName, UserId = userId, TimeInForce = TimeInForce.Gtc, CustomDef = new() { Limit = new() } });
       marketOrderPool = new(orderPoolCapacity, () => new() { User = userName, UserId = userId, TimeInForce = TimeInForce.Gtc, CustomDef = new() { Market = new() } });
       orderSecurityDmx = new TableUpdateDemuxer<OrderRecordTupleWithRoot, uint, uint>(o => o.order.order.Key, o => o.security_id);
@@ -140,6 +144,17 @@ public class SimpleSparkClient : ISimpleSparkClient, IDisposable {
       }
    }
 
+   public async Task<bool> PingAsync() {
+      var request = new BufferMessage() {
+         Sequence = (ulong)DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
+         Data = "spark-sdk-ping"
+      };
+      var result = await shellLatencyService.ping_async(request);
+      return result.success &&
+         result.value.Sequence == request.Sequence &&
+         result.value.Data == request.Data;
+   }
+
    public IObservable<TableUpdate<BookRecord, uint>> Books(SecurityCriteria? securityCriteria) {
       var booksSource = this.booksSource;
       var secFilter = securityCriteria?.ToPredicate();