Skip to content
Browse files

Implemented write-side of ICapacityRepository

  • Loading branch information...
1 parent 7006e6a commit f1e0259f86cc06d79c4a655d7ba5d6cb01f317e0 @ploeh committed Oct 7, 2011
View
28 BookingDaemon/Program.cs
@@ -28,16 +28,29 @@ static void Main(string[] args)
var extension = "txt";
+ var fileDateStore = new FileDateStore(
+ singleSourceOfTruthDirectory,
+ extension);
+
+ var quickenings = new IQuickening[]
+ {
+ new RequestReservationCommand.Quickening(),
+ new ReservationAcceptedEvent.Quickening(),
+ new ReservationRejectedEvent.Quickening(),
+ new CapacityReservedEvent.Quickening(),
+ new SoldOutEvent.Quickening()
+ };
+
var disposable = new CompositeDisposable();
var messageDispatcher = new Subject<object>();
disposable.Add(
messageDispatcher.Subscribe(
new Dispatcher<RequestReservationCommand>(
new CapacityGate(
new JsonCapacityRepository(
- new FileDateWriter(
- singleSourceOfTruthDirectory,
- extension)),
+ fileDateStore,
+ fileDateStore,
+ quickenings),
new JsonChannel<ReservationAcceptedEvent>(
new FileQueueWriter<ReservationAcceptedEvent>(
queueDirectory,
@@ -56,14 +69,7 @@ static void Main(string[] args)
queueDirectory,
extension),
new JsonStreamObserver(
- new IQuickening[]
- {
- new RequestReservationCommand.Quickening(),
- new ReservationAcceptedEvent.Quickening(),
- new ReservationRejectedEvent.Quickening(),
- new CapacityReservedEvent.Quickening(),
- new SoldOutEvent.Quickening()
- },
+ quickenings,
messageDispatcher));
var tokenSource = new CancellationTokenSource();
View
2 BookingFilePersistence/BookingFilePersistence.csproj
@@ -40,7 +40,7 @@
<Reference Include="System.Xml" />
</ItemGroup>
<ItemGroup>
- <Compile Include="FileDateWriter.cs" />
+ <Compile Include="FileDateStore.cs" />
<Compile Include="FileQueueWriter.cs" />
<Compile Include="FileQueue.cs" />
<Compile Include="Properties\AssemblyInfo.cs" />
View
33 BookingFilePersistence/FileDateWriter.cs → BookingFilePersistence/FileDateStore.cs
@@ -7,26 +7,20 @@
namespace Ploeh.Samples.Booking.Persistence.FileSystem
{
- public class FileDateWriter : IStoreWriter<DateTime>
+ public class FileDateStore : IStoreWriter<DateTime>, IStoreReader<DateTime>
{
private readonly DirectoryInfo directory;
private readonly string extension;
- public FileDateWriter(DirectoryInfo directory, string extension)
+ public FileDateStore(DirectoryInfo directory, string extension)
{
this.directory = directory;
this.extension = extension;
}
public Stream OpenStreamFor(DateTime item)
{
- var dateDirectory = new DirectoryInfo(
- Path.Combine(
- this.directory.FullName,
- item.ToString("yyyyMMdd")));
-
- if (!dateDirectory.Exists)
- dateDirectory.Create();
+ var dateDirectory = this.GetDirectory(item);
var fileName = Guid.NewGuid().ToString();
var path = Path.ChangeExtension(
@@ -36,5 +30,26 @@ public Stream OpenStreamFor(DateTime item)
extension);
return File.Open(path, FileMode.Create);
}
+
+ public IEnumerable<Stream> OpenStreamsFor(DateTime item)
+ {
+ var dateDirectory = this.GetDirectory(item);
+
+ return from f in dateDirectory.EnumerateFiles("*." + extension)
+ orderby f.CreationTime
+ select f.OpenRead();
+ }
+
+ private DirectoryInfo GetDirectory(DateTime item)
+ {
+ var dateDirectory = new DirectoryInfo(
+ Path.Combine(
+ this.directory.FullName,
+ item.ToString("yyyyMMdd")));
+
+ if (!dateDirectory.Exists)
+ dateDirectory.Create();
+ return dateDirectory;
+ }
}
}
View
38 BookingJson/JsonCapacityRepository.cs
@@ -12,17 +12,24 @@ namespace Ploeh.Samples.Booking.JsonAntiCorruption
public class JsonCapacityRepository : ICapacityRepository
{
private readonly IStoreWriter<DateTime> writer;
+ private readonly IStoreReader<DateTime> reader;
+ private readonly IEnumerable<IQuickening> quickenings;
private readonly JsonSerializer serializer;
- public JsonCapacityRepository(IStoreWriter<DateTime> dateWriter)
+ public JsonCapacityRepository(IStoreWriter<DateTime> dateWriter, IStoreReader<DateTime> dateReader, IEnumerable<IQuickening> quickenings)
{
this.writer = dateWriter;
+ this.reader = dateReader;
+ this.quickenings = quickenings;
this.serializer = new JsonSerializer();
}
public IEnumerable<Capacity> Read(DateTime date)
{
- yield break;
+ var capacity = this.GetEventsFor(date)
+ .Aggregate(new Capacity(10), (c, e) => c.Reserve(e));
+
+ return new[] { capacity };
}
public void Write(DateTime date, CapacityReservedEvent capacityReserved)
@@ -31,5 +38,32 @@ public void Write(DateTime date, CapacityReservedEvent capacityReserved)
using (var writer = new StreamWriter(stream))
this.serializer.Serialize(writer, capacityReserved.Envelop());
}
+
+ private IEnumerable<CapacityReservedEvent> GetEventsFor(DateTime date)
+ {
+ var streams = this.reader.OpenStreamsFor(date);
+ foreach (var stream in streams)
+ {
+ try
+ {
+ using (var reader = new StreamReader(stream))
+ using (var jsonReader = new JsonTextReader(reader))
+ {
+ dynamic json = this.serializer.Deserialize(jsonReader);
+ var messages = from q in this.quickenings
+ from m in (IEnumerable<IMessage>)q.Quicken(json)
+ select m;
+ foreach (var m in messages)
+ {
+ yield return (CapacityReservedEvent)m;
+ }
+ }
+ }
+ finally
+ {
+ stream.Dispose();
+ }
+ }
+ }
}
}
View
1 BookingPersistenceModel/BookingPersistenceModel.csproj
@@ -43,6 +43,7 @@
<Compile Include="ExceptionPolicy.cs" />
<Compile Include="IQueue.cs" />
<Compile Include="Dispatcher.cs" />
+ <Compile Include="IStoreReader.cs" />
<Compile Include="QueueConsumer.cs" />
<Compile Include="Properties\AssemblyInfo.cs" />
<Compile Include="IStoreWriter.cs" />
View
13 BookingPersistenceModel/IStoreReader.cs
@@ -0,0 +1,13 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.IO;
+
+namespace Ploeh.Samples.Booking.PersistenceModel
+{
+ public interface IStoreReader<T>
+ {
+ IEnumerable<Stream> OpenStreamsFor(T item);
+ }
+}

0 comments on commit f1e0259

Please sign in to comment.
Something went wrong with that request. Please try again.