DynamicData.Zmq

DynamicData.Zmq is a classic ZeroMQ (using NetMQ) pub/sub platform using a broker for message control. Read operations on the system state are done on a local cache while write operations are done at the broker level. The broker itself act as an event database.

Each producer generates events or commands on a given subject (e.g, price changed event for a currency pair). The subject itself is as such a stream of events, that is an aggregate.

    public class CurrencyPair : AggregateBase<string>
    {

        public CcyPairState State { get; set; }

        public double Ask { get; set; }
        public double Bid { get; set; }
        public double Mid { get; set; }
        public double Spread { get; set; }

        public override string ToString()
        {
            return $"{this.Id}({AppliedEvents.Count()} event(s))";
        }
    }

    public interface IAggregate
    {
        IEnumerable<IEvent> AppliedEvents { get; }
        void Apply(IEvent @event);
    }

    public interface IAggregate<TKey> : IAggregate
    {
        TKey Id { get; set; }
        int Version { get; set; }
        void Apply<TAggregate>(IEvent<TKey, TAggregate> @event) where TAggregate : IAggregate<TKey>;
    }

Each event is defined by routable properties which serialize as a ZeroMQ readable subject (e.g, my event should be serialized as {CcyPair}.{Market} or {CcyPair}.{Market}.{Counterparty}), with the aggregate id always in the first position) in order to allow a ZeroMQ filter at the socket level.

    public class ChangeCcyPairPrice : CommandBase<string, CurrencyPair>
    {
        public ChangeCcyPairPrice(string ccyPairId, string market, double ask, double bid, double mid, double spread): base(ccyPairId)
        {
            Ask = ask;
            Bid = bid;
            Mid = mid;
            Spread = spread;
            Market = market;
        }

        public double Ask { get; set; }
        public double Bid { get; set; }
        public double Mid { get; set; }
        public double Spread { get; set; }

        [RoutingPosition(0)]
        public string Market { get; set; }

        public override void Apply(CurrencyPair aggregate)
        {
            aggregate.Ask = Ask;
            aggregate.Bid = Bid;
            aggregate.Mid = Mid;
            aggregate.Spread = Spread;
        }
    }

The broker captures the event, gives it an id and a position in the stream, and save it before forwarding it to its subscribers.

    public interface IEventCache
    {
        Task<IEventId> AppendToStream(string subject, byte[] payload);
        Task<IEnumerable<IEventMessage>> GetAllStreams();
        Task<IEnumerable<IEventMessage>> GetStream(string streamId);
        Task<IEnumerable<IEventMessage>> GetStreamBySubject(string subject);
        Task Clear();
    }

    public interface IEventIdProvider
    {
        Task Reset();
        IEventId Next(string streamName, string subject);
    }}

The subscriber holds a local cache, using DynamicData, where events are applied on aggregates (e.g, a ChangePriceEvent on the EUR/USD currency pair). DynamicData ObservableCache allows then to process the changes to internal subscribers, using Rx.

    public interface IDynamicCache<TKey, TAggregate> : IActor where TAggregate : IAggregate<TKey>
    {
        IObservableCache<TAggregate, TKey> OnItemChanged { get; }
        IEnumerable<TAggregate> Items { get; }
        IObservable<DynamicCacheState> OnStateChanged { get; }
        DynamicCacheState CacheState { get; }
        bool IsStaled { get; }
        IObservable<bool> OnStaled { get; }
        bool IsCaughtingUp { get; }
        IObservable<bool> OnCaughtingUp { get; }

    }

The subscriber manages disconnection and cache rebuilding. The broker keeps a "state of the world" router socket open to get the current application state which is then reconciled locally with the events received during the caughting-up process.

The E2E tests give a bird eye view of the process.

The BrokerageService and DynamicCache class hold the core of the logic explained above.

DynamicData.Zmq exposes a .NET Core MVC API (here using Scrutor for DI), but could be used withing any .NET application.

var dynamicCacheBuilder = services.AddDynamicCacheService<string, CurrencyPair>(configuration =>
	{
            configuration.UseEventCache<InMemoryEventCache>();
            configuration.UseEventIdProvider<InMemoryEventIdProvider>();
            configuration.UseSerializer<JsonNetSerializer>();
	});

    dynamicCacheBuilder
			.AddDynamicCache(configuration =>
                            {
                                configuration.HeartbeatEndpoint = HeartbeatEndpoint;
                                configuration.StateOfTheWorldEndpoint = StateOfTheWorldEndpoint;
                                configuration.SubscriptionEndpoint = ToSubscribersEndpoint;
                            })
                        .AddBroker(configuration =>
                            {
                                configuration.HeartbeatEndpoint = HeartbeatEndpoint;
                                configuration.StateOfTheWorldEndpoint = StateOfTheWorldEndpoint;
                                configuration.ToSubscribersEndpoint = ToSubscribersEndpoint;
                                configuration.ToPublisherEndpoint = ToPublishersEndpoint;
                            });


     dynamicCacheBuilder.AddProducer<string, CurrencyPair, Market, MarketConfiguration>(configuration =>
              {
                  configuration.HeartbeatEndpoint = HeartbeatEndpoint;
                  configuration.BrokerEndpoint = ToPublishersEndpoint;
              });


     services.Scan(scan => scan.FromEntryAssembly()
                               .AddClasses(classes => classes.AssignableTo<IHostedService>())
                               .AsImplementedInterfaces());

Then, host it in your .NET Core MVC application.

 public class HostCache : IHostedService
    {
        private readonly IDynamicCache<string, CurrencyPair> _cache;
        private readonly ILogger<HostCache> _logger;

        public HostCache(ILogger<HostCache> logger, IDynamicCache<string, CurrencyPair> cache)
        {
            _cache = cache;
            _logger = logger;
        }

        public async Task StartAsync(CancellationToken cancellationToken)
        {
            _cache.OnItemChanged
                  .Connect()
                  .ToCollection()
                  .Scan(CcyPairPrices.Default, (previous, obs) =>
                    {
                        var prices = new CcyPairPrices();

                        foreach (var o in obs)
                        {
                            prices.Prices.Add(new Price()
                            {
                                CcyPair = o.Id,
                                Ask = o.Ask,
                                Bid = o.Bid,
                                EventCount = o.AppliedEvents.Count()
                            });
                        }

                        return prices;

                    })
                  .Subscribe(state =>
                {
                    _logger.LogInformation(state.ToString());
                }
        );


            await _cache.Run();
        }

        public async Task StopAsync(CancellationToken cancellationToken)
        {
            await _cache.Destroy();
        }