¡Hola, Habr! Decidí alejarme de Scala, Idris y otros FP por un tiempo y hablar un poco sobre Event Store, una base de datos en la que los eventos se pueden guardar en flujos de eventos. Como en el buen libro antiguo, también tenemos Mosqueteros de hecho 4 y el cuarto es DDD. Primero, uso Event Storming para seleccionar comandos, eventos y entidades asociadas con ellos. Luego, sobre su base, guardaré el estado del objeto y lo restauraré. Haré un TodoList regular en este artículo. Para más detalles, bienvenido bajo cat.
Contenido
- Los tres mosqueteros: abastecimiento de eventos, asalto de eventos y tienda de eventos: ingrese a la batalla: parte 1: probar la tienda de eventos de DB
Enlaces
Fuentes
Imágenes Imagen de la ventana acoplable
Tienda de
eventos Soucing de eventos Tormenta de
eventos
En realidad, Event Store es una base de datos diseñada para almacenar eventos. También sabe cómo crear suscripciones a eventos para que puedan ser procesados de alguna manera. También hay proyecciones que también reaccionan a los eventos y, en base a ellos, acumulan algunos datos. Por ejemplo, durante el evento TodoCreated, puede aumentar algún tipo de contador de recuento en la proyección. Por ahora, en esta parte, usaré Event Store como Db de lectura y escritura. Además, en los siguientes artículos, crearé una base de datos separada para leer en la que se escribirán los datos en función de los eventos almacenados en la base de datos para escribir en el almacén de eventos. También habrá un ejemplo de cómo hacer un "viaje en el tiempo" haciendo retroceder el sistema al estado que tenía en el pasado.
Y entonces comencemos Event Stroming. Por lo general, para su implementación, se reúnen todas las personas interesadas y expertos que cuentan qué eventos en el área temática simulará el software. Por ejemplo, para el software de la planta - Producto fabricado. Para el juego - Daño recibido. Para software financiero: dinero acreditado en la cuenta, etc. Dado que nuestra área temática es tan simple como TodoList, tendremos pocos eventos. Entonces, escribamos los eventos de nuestra área temática (dominio) en la pizarra.
Ahora agreguemos los comandos que activan estos eventos.
A continuación, agrupemos estos eventos y comandos alrededor de la entidad con un cambio en el estado al que están asociados.
Mis comandos simplemente se convertirán en nombres de métodos de servicio. Vayamos a la implementación.
Primero, describamos los eventos en código.
public interface IDomainEvent
{
// . id Event Strore
Guid EventId { get; }
// . Event Store
long EventNumber { get; set; }
}
public sealed class TodoCreated : IDomainEvent
{
//Id Todo
public Guid Id { get; set; }
// Todo
public string Name { get; set; }
public Guid EventId => Id;
public long EventNumber { get; set; }
}
public sealed class TodoRemoved : IDomainEvent
{
public Guid EventId { get; set; }
public long EventNumber { get; set; }
}
public sealed class TodoCompleted: IDomainEvent
{
public Guid EventId { get; set; }
public long EventNumber { get; set; }
}
Ahora nuestro núcleo es una entidad:
public sealed class Todo : IEntity<TodoId>
{
private readonly List<IDomainEvent> _events;
public static Todo CreateFrom(string name)
{
var id = Guid.NewGuid();
var e = new List<IDomainEvent>(){new TodoCreated()
{
Id = id,
Name = name
}};
return new Todo(new TodoId(id), e, name, false);
}
public static Todo CreateFrom(IEnumerable<IDomainEvent> events)
{
var id = Guid.Empty;
var name = String.Empty;
var completed = false;
var ordered = events.OrderBy(e => e.EventNumber).ToList();
if (ordered.Count == 0)
return null;
foreach (var @event in ordered)
{
switch (@event)
{
case TodoRemoved _:
return null;
case TodoCreated created:
name = created.Name;
id = created.Id;
break;
case TodoCompleted _:
completed = true;
break;
default: break;
}
}
if (id == default)
return null;
return new Todo(new TodoId(id), new List<IDomainEvent>(), name, completed);
}
private Todo(TodoId id, List<IDomainEvent> events, string name, bool isCompleted)
{
Id = id;
_events = events;
Name = name;
IsCompleted = isCompleted;
Validate();
}
public TodoId Id { get; }
public IReadOnlyList<IDomainEvent> Events => _events;
public string Name { get; }
public bool IsCompleted { get; private set; }
public void Complete()
{
if (!IsCompleted)
{
IsCompleted = true;
_events.Add(new TodoCompleted()
{
EventId = Guid.NewGuid()
});
}
}
public void Delete()
{
_events.Add(new TodoRemoved()
{
EventId = Guid.NewGuid()
});
}
private void Validate()
{
if (Events == null)
throw new ApplicationException(" ");
if (string.IsNullOrWhiteSpace(Name))
throw new ApplicationException(" ");
if (Id == default)
throw new ApplicationException(" ");
}
}
Nos conectamos a la Tienda de Eventos:
services.AddSingleton(sp =>
{
// TCP .
// . .
var con = EventStoreConnection.Create(new Uri("tcp://admin:changeit@127.0.0.1:1113"), "TodosConnection");
con.ConnectAsync().Wait();
return con;
});
Y así, la parte principal. Almacenamiento y lectura de eventos de la propia tienda de eventos:
public sealed class EventsRepository : IEventsRepository
{
private readonly IEventStoreConnection _connection;
public EventsRepository(IEventStoreConnection connection)
{
_connection = connection;
}
public async Task<long> Add(Guid collectionId, IEnumerable<IDomainEvent> events)
{
var eventPayload = events.Select(e => new EventData(
//Id
e.EventId,
//
e.GetType().Name,
// Json (True|False)
true,
//
Encoding.UTF8.GetBytes(JsonSerializer.Serialize((object)e)),
//
Encoding.UTF8.GetBytes((string)e.GetType().FullName)
));
//
var res = await _connection.AppendToStreamAsync(collectionId.ToString(), ExpectedVersion.Any, eventPayload);
return res.NextExpectedVersion;
}
public async Task<List<IDomainEvent>> Get(Guid collectionId)
{
var results = new List<IDomainEvent>();
long start = 0L;
while (true)
{
var events = await _connection.ReadStreamEventsForwardAsync(collectionId.ToString(), start, 4096, false);
if (events.Status != SliceReadStatus.Success)
return results;
results.AddRange(Deserialize(events.Events));
if (events.IsEndOfStream)
return results;
start = events.NextEventNumber;
}
}
public async Task<List<T>> GetAll<T>() where T : IDomainEvent
{
var results = new List<IDomainEvent>();
Position start = Position.Start;
while (true)
{
var events = await _connection.ReadAllEventsForwardAsync(start, 4096, false);
results.AddRange(Deserialize(events.Events.Where(e => e.Event.EventType == typeof(T).Name)));
if (events.IsEndOfStream)
return results.OfType<T>().ToList();
start = events.NextPosition;
}
}
private List<IDomainEvent> Deserialize(IEnumerable<ResolvedEvent> events) =>
events
.Where(e => IsEvent(e.Event.EventType))
.Select(e =>
{
var result = (IDomainEvent)JsonSerializer.Deserialize(e.Event.Data, ToType(e.Event.EventType));
result.EventNumber = e.Event.EventNumber;
return result;
})
.ToList();
private static bool IsEvent(string eventName)
{
return eventName switch
{
nameof(TodoCreated) => true,
nameof(TodoCompleted) => true,
nameof(TodoRemoved) => true,
_ => false
};
}
private static Type ToType(string eventName)
{
return eventName switch
{
nameof(TodoCreated) => typeof(TodoCreated),
nameof(TodoCompleted) => typeof(TodoCompleted),
nameof(TodoRemoved) => typeof(TodoRemoved),
_ => throw new NotImplementedException(eventName)
};
}
}
La tienda de entidades parece muy simple. Obtenemos los eventos de la entidad del EventStore y los restauramos a partir de ellos, o simplemente guardamos los eventos de la entidad.
public sealed class TodoRepository : ITodoRepository
{
private readonly IEventsRepository _eventsRepository;
public TodoRepository(IEventsRepository eventsRepository)
{
_eventsRepository = eventsRepository;
}
public Task SaveAsync(Todo entity) => _eventsRepository.Add(entity.Id.Value, entity.Events);
public async Task<Todo> GetAsync(TodoId id)
{
var events = await _eventsRepository.Get(id.Value);
return Todo.CreateFrom(events);
}
public async Task<List<Todo>> GetAllAsync()
{
var events = await _eventsRepository.GetAll<TodoCreated>();
var res = await Task.WhenAll(events.Where(t => t != null).Where(e => e.Id != default).Select(e => GetAsync(new TodoId(e.Id))));
return res.Where(t => t != null).ToList();
}
}
El servicio en el que se realiza el trabajo con el repositorio y la entidad:
public sealed class TodoService : ITodoService
{
private readonly ITodoRepository _repository;
public TodoService(ITodoRepository repository)
{
_repository = repository;
}
public async Task<TodoId> Create(TodoCreateDto dto)
{
var todo = Todo.CreateFrom(dto.Name);
await _repository.SaveAsync(todo);
return todo.Id;
}
public async Task Complete(TodoId id)
{
var todo = await _repository.GetAsync(id);
todo.Complete();
await _repository.SaveAsync(todo);
}
public async Task Remove(TodoId id)
{
var todo = await _repository.GetAsync(id);
todo.Delete();
await _repository.SaveAsync(todo);
}
public async Task<List<TodoReadDto>> GetAll()
{
var todos = await _repository.GetAllAsync();
return todos.Select(t => new TodoReadDto()
{
Id = t.Id.Value,
Name = t.Name,
IsComplete = t.IsCompleted
}).ToList();
}
public async Task<List<TodoReadDto>> Get(IEnumerable<TodoId> ids)
{
var todos = await Task.WhenAll(ids.Select(i => _repository.GetAsync(i)));
return todos.Where(t => t != null).Select(t => new TodoReadDto()
{
Id = t.Id.Value,
Name = t.Name,
IsComplete = t.IsCompleted
}).ToList();
}
}
Bueno, en realidad, hasta ahora nada impresionante. En el próximo artículo, cuando agregue una base de datos separada para leer, todo brillará con diferentes colores. Esto inmediatamente nos colgará la consistencia con el tiempo. Event Store y SQL DB en el principio maestro-esclavo. Un ES blanco y muchos MS SQL negros desde los que leer datos.
Digresión lírica. A la luz de los acontecimientos recientes, no pude evitar bromear sobre el amo esclavo y los blancos negros. Ehe, la era se va, les diremos a nuestros nietos que vivimos en una época en que las bases durante la replicación se llamaban amo y esclavo.
En sistemas donde hay mucha lectura y poca escritura de datos (la mayoría), esto aumentará la velocidad de trabajo. En realidad, la replicación del maestro esclavo en sí, tiene como objetivo el hecho de que su escritura se ralentiza (así como con índices), pero a cambio, la lectura se acelera al distribuir la carga entre varias bases de datos.