Skip to content
This repository was archived by the owner on Sep 3, 2024. It is now read-only.

Commit 3d29994

Browse files
thefringeninjathefringeninja
committed
Allow Selection of Store from env
Allows the selection of which backend to use based on the following environment variables: SQLSTREAMSTORE_CONNECTION_STRING SQLSTREAMSTORE_CONNECTION_SCHEMA SQLSTREAMSTORE_CONNECTION_PROVIDER - only postgres, mssql, and inmemory are supported
1 parent ea05d76 commit 3d29994

5 files changed

Lines changed: 173 additions & 20 deletions

File tree

‎NuGet.Config‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
<?xml version="1.0" ?>
2+
<configuration>
3+
<packageSources>
4+
<add key="sql-stream-store" value="https://www.myget.org/F/sqlstreamstore/api/v3/index.json" />
5+
</packageSources>
6+
</configuration>
7+

‎src/SqlStreamStore.HAL.DevServer/Program.cs‎

Lines changed: 16 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,6 @@ internal class Program : IDisposable
1313
{
1414
private static readonly Random s_random = new Random();
1515
private readonly CancellationTokenSource _cts;
16-
private readonly InMemoryStreamStore _streamStore;
17-
private readonly IWebHost _host;
1816
private readonly IConfigurationRoot _configuration;
1917

2018
private bool Interactive => _configuration.GetValue<bool>("interactive");
@@ -36,12 +34,6 @@ public static async Task<int> Main(string[] args)
3634
private Program(string[] args)
3735
{
3836
_cts = new CancellationTokenSource();
39-
_streamStore = new InMemoryStreamStore();
40-
_host = new WebHostBuilder()
41-
.UseKestrel()
42-
.UseStartup(new DevServerStartup(_streamStore))
43-
.UseSerilog()
44-
.Build();
4537
_configuration = new ConfigurationBuilder()
4638
.AddEnvironmentVariables()
4739
.AddCommandLine(args)
@@ -52,17 +44,25 @@ private async Task<int> Run()
5244
{
5345
try
5446
{
55-
var serverTask = _host.RunAsync(_cts.Token);
56-
57-
if(Interactive)
47+
using(var streamStore = await SqlStreamStoreFactory.Create())
48+
using(var host = new WebHostBuilder()
49+
.UseKestrel()
50+
.UseStartup(new DevServerStartup(streamStore))
51+
.UseSerilog()
52+
.Build())
5853
{
59-
Log.Warning("Running interactively.");
60-
DisplayMenu(_streamStore);
61-
}
54+
var serverTask = host.RunAsync(_cts.Token);
6255

63-
await serverTask;
56+
if(Interactive)
57+
{
58+
Log.Warning("Running interactively.");
59+
DisplayMenu(streamStore);
60+
}
6461

65-
return 0;
62+
await serverTask;
63+
64+
return 0;
65+
}
6666
}
6767
catch(Exception ex)
6868
{
@@ -79,7 +79,6 @@ private static void DisplayMenu(IStreamStore streamStore, string url = null)
7979
{
8080
while(true)
8181
{
82-
Console.WriteLine("Using stream store: {0}", streamStore.GetType().Name);
8382
Console.WriteLine("Press w to write 10 messages each to 100 streams");
8483
Console.WriteLine("Press t to write 100 messages each to 10 streams");
8584
Console.WriteLine("Press ESC to exit");
@@ -135,8 +134,6 @@ private static NewStreamMessage[] GenerateMessages(int messageCount)
135134

136135
public void Dispose()
137136
{
138-
_host?.Dispose();
139-
_streamStore?.Dispose();
140137
_cts?.Dispose();
141138
}
142139
}

‎src/SqlStreamStore.HAL.DevServer/SqlStreamStore.HAL.DevServer.csproj‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,8 @@
1313
<PackageReference Include="Microsoft.Extensions.Configuration.CommandLine" Version="2.1.1" />
1414
<PackageReference Include="Serilog.AspNetCore" Version="2.1.1" />
1515
<PackageReference Include="Serilog.Sinks.Console" Version="3.1.1" />
16+
<PackageReference Include="SqlStreamStore.MsSql" Version="1.2.0-*" />
17+
<PackageReference Include="SqlStreamStore.Postgres" Version="1.2.0-*" />
1618
</ItemGroup>
1719
<ItemGroup>
1820
<ProjectReference Include="..\SqlStreamStore.HAL\SqlStreamStore.HAL.csproj" />

‎src/SqlStreamStore.HAL.DevServer/SqlStreamStoreFactory.cs‎

Lines changed: 147 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,147 @@
1+
namespace SqlStreamStore.HAL.DevServer
2+
{
3+
using System;
4+
using System.Collections.Generic;
5+
using System.Data.SqlClient;
6+
using System.Threading;
7+
using System.Threading.Tasks;
8+
using Npgsql;
9+
using Serilog;
10+
using SqlStreamStore.Infrastructure;
11+
12+
internal static class SqlStreamStoreFactory
13+
{
14+
private delegate Task<IStreamStore> CreateStreamStore(
15+
string connectionString,
16+
string schema,
17+
CancellationToken cancellationToken);
18+
19+
private const string SQLSTREAMSTORE_CONNECTION_STRING = nameof(SQLSTREAMSTORE_CONNECTION_STRING);
20+
private const string SQLSTREAMSTORE_SCHEMA = nameof(SQLSTREAMSTORE_SCHEMA);
21+
private const string SQLSTREAMSTORE_PROVIDER = nameof(SQLSTREAMSTORE_PROVIDER);
22+
23+
private const string postgres = nameof(postgres);
24+
private const string mssql = nameof(mssql);
25+
private const string inmemory = nameof(inmemory);
26+
27+
private static readonly IDictionary<string, CreateStreamStore> s_factories
28+
= new Dictionary<string, CreateStreamStore>
29+
{
30+
[inmemory] = CreateInMemoryStreamStore,
31+
[postgres] = CreatePostgresStreamStore,
32+
[mssql] = CreateMssqlStreamStore
33+
};
34+
35+
public static Task<IStreamStore> Create(CancellationToken cancellationToken = default)
36+
{
37+
var provider = Environment.GetEnvironmentVariable(SQLSTREAMSTORE_PROVIDER)?.ToLowerInvariant()
38+
?? inmemory;
39+
40+
Log.Information($"Creating stream store for provider '{provider}'");
41+
42+
if(!s_factories.TryGetValue(provider, out var factory))
43+
{
44+
throw new InvalidOperationException($"No provider factory for provider '{provider}' found.");
45+
}
46+
47+
var connectionString = Environment.GetEnvironmentVariable(SQLSTREAMSTORE_CONNECTION_STRING);
48+
var schema = Environment.GetEnvironmentVariable(SQLSTREAMSTORE_SCHEMA);
49+
50+
return factory(connectionString, schema, cancellationToken);
51+
}
52+
53+
private static Task<IStreamStore> CreateInMemoryStreamStore(
54+
string connectionString,
55+
string schema,
56+
CancellationToken cancellationToken)
57+
=> Task.FromResult<IStreamStore>(new InMemoryStreamStore());
58+
59+
private static async Task<IStreamStore> CreateMssqlStreamStore(
60+
string connectionString,
61+
string schema,
62+
CancellationToken cancellationToken)
63+
{
64+
var connectionStringBuilder = new SqlConnectionStringBuilder(connectionString);
65+
using(var connection = new SqlConnection(new SqlConnectionStringBuilder(connectionString)
66+
{
67+
InitialCatalog = "master"
68+
}.ConnectionString))
69+
{
70+
await connection.OpenAsync(cancellationToken).NotOnCapturedContext();
71+
72+
using(var command = new SqlCommand(
73+
$@"
74+
IF NOT EXISTS (SELECT name FROM sys.databases WHERE name = N'{connectionStringBuilder.InitialCatalog}')
75+
BEGIN
76+
CREATE DATABASE [{connectionStringBuilder.InitialCatalog}]
77+
END;
78+
",
79+
connection))
80+
{
81+
await command.ExecuteNonQueryAsync(cancellationToken).NotOnCapturedContext();
82+
}
83+
}
84+
85+
var settings = new MsSqlStreamStoreV3Settings(connectionString);
86+
87+
if(schema != null)
88+
{
89+
settings.Schema = schema;
90+
}
91+
92+
var streamStore = new MsSqlStreamStoreV3(settings);
93+
94+
await streamStore.CreateSchema(cancellationToken);
95+
96+
return streamStore;
97+
}
98+
99+
private static async Task<IStreamStore> CreatePostgresStreamStore(
100+
string connectionString,
101+
string schema,
102+
CancellationToken cancellationToken)
103+
{
104+
var connectionStringBuilder = new NpgsqlConnectionStringBuilder(connectionString);
105+
106+
using(var connection = new NpgsqlConnection(new NpgsqlConnectionStringBuilder(connectionString)
107+
{
108+
Database = null
109+
}.ConnectionString))
110+
{
111+
bool exists;
112+
await connection.OpenAsync(cancellationToken).NotOnCapturedContext();
113+
114+
using(var command = new NpgsqlCommand(
115+
$"SELECT 1 FROM pg_database WHERE datname = '{connectionStringBuilder.Database}'",
116+
connection))
117+
{
118+
exists = await command.ExecuteScalarAsync(cancellationToken).NotOnCapturedContext()
119+
!= null;
120+
}
121+
122+
if(!exists)
123+
{
124+
using(var command = new NpgsqlCommand(
125+
$"CREATE DATABASE {connectionStringBuilder.Database}",
126+
connection))
127+
{
128+
await command.ExecuteNonQueryAsync(cancellationToken).NotOnCapturedContext();
129+
}
130+
}
131+
132+
var settings = new PostgresStreamStoreSettings(connectionString);
133+
134+
if(schema != null)
135+
{
136+
settings.Schema = schema;
137+
}
138+
139+
var streamStore = new PostgresStreamStore(settings);
140+
141+
await streamStore.CreateSchema(cancellationToken);
142+
143+
return streamStore;
144+
}
145+
}
146+
}
147+
}

‎src/SqlStreamStore.HAL/SqlStreamStore.HAL.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@
1212
<PackageReference Include="LibLog" Version="5.0.2" PrivateAssets="All" />
1313
<PackageReference Include="Microsoft.AspNetCore.Http.Abstractions" Version="2.1.1" />
1414
<PackageReference Include="Newtonsoft.Json" Version="11.0.2" />
15-
<PackageReference Include="SqlStreamStore" Version="1.1.2" />
15+
<PackageReference Include="SqlStreamStore" Version="1.2.0-*" />
1616
</ItemGroup>
1717
<ItemGroup>
1818
<EmbeddedResource Include="Resources\Schema\*.json" />

0 commit comments

Comments
 (0)