Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 0 additions & 13 deletions Shared.EventStore/EventStore/EventStoreContextManager.cs
Original file line number Diff line number Diff line change
Expand Up @@ -23,24 +23,11 @@ public EventStoreContextManager(Func<String, IEventStoreContext> eventStoreConte
this.EventStoreContexts = new();
this.EventStoreContextFunc = eventStoreContextFunc;
}

//public EventStoreContextManager(IEventStoreContext eventStoreContext)
//{
//this.Context = eventStoreContext;
//}

public event TraceHandler TraceGenerated;


//public IEventStoreContext GetEventStoreContext(String connectionIdentifier) => this.GetEventStoreContext(connectionIdentifier, "EventStoreConnectionString");

public IEventStoreContext GetEventStoreContext(String connectionStringIdentifier)
{
//if (this.Context != null)
//{
// return this.Context;
//}

this.WriteTrace($"No resolved context found, about to resolve one using connectionIdentifier {connectionStringIdentifier}");

if (this.EventStoreContexts.TryGetValue(connectionStringIdentifier, out IEventStoreContext context))
Expand Down
78 changes: 34 additions & 44 deletions Shared.EventStore/Extensions/IApplicationBuilderExtenstions.cs
Original file line number Diff line number Diff line change
Expand Up @@ -66,20 +66,8 @@ internal static List<SubscriptionWorker> ConfigureSubscriptions(ISubscriptionRep
KeyValuePair<String, IDomainEventHandlerResolver> ehr = eventHandlerResolvers.SingleOrDefault(e => e.Key == "Ordered");

if (ehr.Value != null) {
SubscriptionWorker worker = SubscriptionWorker.CreateOrderedSubscriptionWorker(eventStoreConnectionString,
ehr.Value,
subscriptionRepository,
configuration.PersistentSubscriptionPollingInSeconds);
worker.Trace += (_,
args) => traceHandler(TraceEventType.Information, "ORDERED", args.Message);
worker.Warning += (_,
args) => traceHandler(TraceEventType.Warning, "ORDERED", args.Message);
worker.Error += (_,
args) => traceHandler(TraceEventType.Error, "ORDERED", args.Message);
worker.SetIgnoreGroups(configurationSubscriptionWorker.IgnoreGroups);
worker.SetIgnoreStreams(configurationSubscriptionWorker.IgnoreStreams);
worker.SetIncludeGroups(configurationSubscriptionWorker.IncludeGroups);
worker.SetIncludeStreams(configurationSubscriptionWorker.IncludeStreams);
SubscriptionWorker worker = ConfigureSubscriptionWorker(subscriptionRepository, configuration,
eventStoreConnectionString, traceHandler, ehr, configurationSubscriptionWorker, "ORDERED");
workers.Add(worker);
}
}
Expand All @@ -88,20 +76,9 @@ internal static List<SubscriptionWorker> ConfigureSubscriptions(ISubscriptionRep

if (ehr.Value != null)
{
for (Int32 i = 0; i < configurationSubscriptionWorker.InstanceCount; i++)
{
SubscriptionWorker worker = SubscriptionWorker.CreateSubscriptionWorker(eventStoreConnectionString, ehr.Value, subscriptionRepository, configurationSubscriptionWorker.InflightMessages, configuration.PersistentSubscriptionPollingInSeconds);

worker.Trace += (_,
args) => traceHandler(TraceEventType.Information, "DOMAIN", args.Message);
worker.Warning += (_,
args) => traceHandler(TraceEventType.Warning, "DOMAIN", args.Message);
worker.Error += (_,
args) => traceHandler(TraceEventType.Error, "DOMAIN", args.Message);
worker.SetIgnoreGroups(configurationSubscriptionWorker.IgnoreGroups);
worker.SetIgnoreStreams(configurationSubscriptionWorker.IgnoreStreams);
worker.SetIncludeGroups(configurationSubscriptionWorker.IncludeGroups);
worker.SetIncludeStreams(configurationSubscriptionWorker.IncludeStreams);
for (Int32 i = 0; i < configurationSubscriptionWorker.InstanceCount; i++) {
SubscriptionWorker worker = ConfigureSubscriptionWorker(subscriptionRepository, configuration,
eventStoreConnectionString, traceHandler, ehr, configurationSubscriptionWorker, "DOMAIN");

workers.Add(worker);
}
Expand All @@ -113,22 +90,8 @@ internal static List<SubscriptionWorker> ConfigureSubscriptions(ISubscriptionRep
{
for (Int32 i = 0; i < configurationSubscriptionWorker.InstanceCount; i++)
{
SubscriptionWorker worker = SubscriptionWorker.CreateSubscriptionWorker(eventStoreConnectionString,
ehr.Value,
subscriptionRepository,
configurationSubscriptionWorker.InflightMessages,
configuration.PersistentSubscriptionPollingInSeconds);

worker.Trace += (_,
args) => traceHandler(TraceEventType.Information, "MAIN", args.Message);
worker.Warning += (_,
args) => traceHandler(TraceEventType.Warning, "MAIN", args.Message);
worker.Error += (_,
args) => traceHandler(TraceEventType.Error, "MAIN", args.Message);
worker.SetIgnoreGroups(configurationSubscriptionWorker.IgnoreGroups);
worker.SetIgnoreStreams(configurationSubscriptionWorker.IgnoreStreams);
worker.SetIncludeGroups(configurationSubscriptionWorker.IncludeGroups);
worker.SetIncludeStreams(configurationSubscriptionWorker.IncludeStreams);
SubscriptionWorker worker = ConfigureSubscriptionWorker(subscriptionRepository, configuration,
eventStoreConnectionString, traceHandler, ehr, configurationSubscriptionWorker, "MAIN");

workers.Add(worker);
}
Expand All @@ -138,5 +101,32 @@ internal static List<SubscriptionWorker> ConfigureSubscriptions(ISubscriptionRep

return workers;
}

private static SubscriptionWorker ConfigureSubscriptionWorker(ISubscriptionRepository subscriptionRepository,
SubscriptionWorkersRoot configuration,
String eventStoreConnectionString,
Action<TraceEventType, String, String> traceHandler,
KeyValuePair<String, IDomainEventHandlerResolver> ehr,
SubscriptionWorkerConfig configurationSubscriptionWorker,
String type) {

SubscriptionWorker worker = type switch {
"ORDERED" => SubscriptionWorker.CreateOrderedSubscriptionWorker(eventStoreConnectionString, ehr.Value, subscriptionRepository, configuration.PersistentSubscriptionPollingInSeconds),
_ => SubscriptionWorker.CreateSubscriptionWorker(eventStoreConnectionString, ehr.Value, subscriptionRepository, configurationSubscriptionWorker.InflightMessages, configuration.PersistentSubscriptionPollingInSeconds),
};

worker.Trace += (_,
args) => traceHandler(TraceEventType.Information, type, args.Message);
worker.Warning += (_,
args) => traceHandler(TraceEventType.Warning, type, args.Message);
worker.Error += (_,
args) => traceHandler(TraceEventType.Error, type, args.Message);
worker.SetIgnoreGroups(configurationSubscriptionWorker.IgnoreGroups);
worker.SetIgnoreStreams(configurationSubscriptionWorker.IgnoreStreams);
worker.SetIncludeGroups(configurationSubscriptionWorker.IncludeGroups);
worker.SetIncludeStreams(configurationSubscriptionWorker.IncludeStreams);
return worker;
}

#endregion
}
34 changes: 0 additions & 34 deletions Shared/Middleware/CorrelationIdMiddleware.cs
Original file line number Diff line number Diff line change
Expand Up @@ -139,38 +139,4 @@ public static TenantIdentifiers GetIdentifiersFromRoute(this HttpContext context

return estateId == Guid.Empty ? TenantIdentifiers.Default() : new TenantIdentifiers(estateId, merchantId);
}

/*public static async Task<TenantIdentifiers> GetIdentifiersFromPayload(this HttpContext context)
{
HttpRequest request = context.Request;
String bodyAsText = null;

try
{
request.EnableBuffering();
bodyAsText = await new StreamReader(request.Body).ReadToEndAsync();
request.Body.Position = 0;

if (String.IsNullOrWhiteSpace(bodyAsText))
{
return TenantIdentifiers.Default();
}

JToken rootToken = JToken.Parse(bodyAsText);

JToken organisationIdToken = rootToken.SelectTokens("..organisationId").FirstOrDefault();
JToken storeIdToken = rootToken.SelectTokens("..storeId").FirstOrDefault();

Guid.TryParse(organisationIdToken?.Value<String>(), out Guid organisationId);
Guid.TryParse(storeIdToken?.Value<String>(), out Guid storeId);

return organisationId == Guid.Empty ? TenantIdentifiers.Default() : new TenantIdentifiers(organisationId, storeId);
}
catch (Exception e)
{
EposityLogger.WriteWarning($"Unable to get organisationId from request body [{bodyAsText}]");
EposityLogger.WriteException(e);
return TenantIdentifiers.Default();
}
}*/
}
1 change: 0 additions & 1 deletion Shared/Middleware/ResponseLoggingMiddleware.cs
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,6 @@ public class ResponseLoggingMemoryStream : MemoryStream
public override void Close()
{
// Dont close by default
// base.Close();
}

public void ForceClose()
Expand Down
Loading