first working version

src/Blumchen/Database/Run.cs

@@ -40,7 +40,7 @@ internal static async IAsyncEnumerable<IEnvelope> QueryTransactionSnapshot(this
         string snapshotName,
         TableDescriptorBuilder.MessageTable tableDescriptor,
         ISet<string> registeredTypesKeys,
-        IReplicationDataMapper dataMapper,
+        ReplicationDataMapper dataMapper,
         [EnumeratorCancellation] CancellationToken ct)
     {
         var transaction = await connection.BeginTransactionAsync(IsolationLevel.RepeatableRead, ct).ConfigureAwait(false);

src/Blumchen/Publications/MessageAppender.cs

@@ -1,4 +1,5 @@
 using System.Collections;
+using System.Text.Json.Serialization.Metadata;
 using Blumchen.Serialization;
 using Npgsql;
 
@@ -28,7 +29,7 @@ public static async Task AppendAsync<T>(T @input
 
     private static async Task AppendAsyncOfT<T>(T input
         , TableDescriptorBuilder.MessageTable tableDescriptor
-        , IJsonTypeResolver typeResolver
+        , ITypeResolver<JsonTypeInfo> typeResolver
         , NpgsqlConnection connection
         , NpgsqlTransaction transaction
         , CancellationToken ct) where T : class
@@ -66,7 +67,7 @@ public static async Task AppendAsync<T>(T input
 
     private static async Task AppendBatchAsyncOfT<T>(T inputs
         , TableDescriptorBuilder.MessageTable tableDescriptor
-        , IJsonTypeResolver resolver
+        , ITypeResolver<JsonTypeInfo> resolver
         , NpgsqlConnection connection
         , NpgsqlTransaction transaction
         , CancellationToken ct) where T : class, IEnumerable

src/Blumchen/Publications/PublisherOptions.cs

@@ -1,10 +1,11 @@
+using System.Text.Json.Serialization.Metadata;
 using Blumchen.Database;
 using Blumchen.Serialization;
 using Npgsql;
 
 namespace Blumchen.Publications;
 
-public record PublisherOptions(TableDescriptorBuilder.MessageTable TableDescriptor, IJsonTypeResolver JsonTypeResolver);
+public record PublisherOptions(TableDescriptorBuilder.MessageTable TableDescriptor, ITypeResolver<JsonTypeInfo> JsonTypeResolver);
 
 public static class PublisherOptionsExtensions
 {

src/Blumchen/Publications/PublisherSetupOptionsBuilder.cs

@@ -1,4 +1,5 @@
 using System.Text.Json.Serialization;
+using System.Text.Json.Serialization.Metadata;
 using Blumchen.Serialization;
 using JetBrains.Annotations;
 using static Blumchen.TableDescriptorBuilder;

src/Blumchen/Serialization/ITypeResolver.cs

@@ -7,14 +7,14 @@ namespace Blumchen.Serialization;
 public interface ITypeResolver<T>
 {
     (string, T) Resolve(Type type);
+    Type Resolve(string type);
+    IDictionary<string, Type> RegisteredTypes { get; }
 }
 
-public interface IJsonTypeResolver: ITypeResolver<JsonTypeInfo>;
-
 internal sealed class JsonTypeResolver(
     JsonSerializerContext serializationContext,
     INamingPolicy? namingPolicy = default)
-    : IJsonTypeResolver
+    : ITypeResolver<JsonTypeInfo>
 {
     public JsonSerializerContext SerializationContext { get; } = serializationContext;
     private readonly ConcurrentDictionary<string, Type> _typeDictionary = [];
@@ -31,7 +31,7 @@ internal void WhiteList(Type type)
     public (string, JsonTypeInfo) Resolve(Type type) =>
         (_typeDictionary.Single(kv => kv.Value == type).Key, _typeInfoDictionary[type]);
 
-    internal IDictionary<string,Type> RegisteredTypes { get => _typeDictionary; }
-    internal Type Resolve(string type) => _typeDictionary[type];
+    public IDictionary<string,Type> RegisteredTypes { get => _typeDictionary; }
+    public Type Resolve(string type) => _typeDictionary[type];
 }
 

src/Blumchen/Serialization/JsonTypeResolverExtensions.cs

@@ -3,6 +3,6 @@ namespace Blumchen.Serialization;
 
 public static class JsonTypeResolverExtensions
 {
-    internal static IEnumerable<string> Keys(this JsonTypeResolver? resolver) => resolver?.RegisteredTypes.Keys ?? Enumerable.Empty<string>();
-    internal static IEnumerable<Type> Values(this JsonTypeResolver? resolver) => resolver?.RegisteredTypes.Values ?? Enumerable.Empty<Type>();
+    internal static IEnumerable<string> Keys<T>(this ITypeResolver<T>? resolver) => resolver?.RegisteredTypes.Keys ?? Enumerable.Empty<string>();
+    internal static IEnumerable<Type> Values<T>(this ITypeResolver<T>? resolver) => resolver?.RegisteredTypes.Values ?? Enumerable.Empty<Type>();
 }

src/Blumchen/Serialization/MessageUrnAttribute.cs

@@ -2,7 +2,7 @@
 
 namespace Blumchen.Serialization;
 
-[AttributeUsage(AttributeTargets.Class | AttributeTargets.Interface)]
+[AttributeUsage(AttributeTargets.Class)]
 public class MessageUrnAttribute:
     Attribute
 {
@@ -33,6 +33,48 @@ private static Uri FormatUrn(string urn)
 }
 
 
+public enum RawData
+{
+    String,
+    Object
+}
+
+[AttributeUsage(AttributeTargets.Interface, AllowMultiple = true)]
+public class RawUrnAttribute:
+    Attribute
+{
+    public RawData Data { get; }
+
+    /// <summary>
+    /// </summary>
+    /// <param name="urn">The urn value to use for this message type.</param>
+    /// <param name="data">The <see cref="RawData"/> value to bind to this <paramref name="urn"/></param>
+    public RawUrnAttribute(string urn, RawData data)
+    {
+        Data = data;
+        ArgumentException.ThrowIfNullOrEmpty(urn, nameof(urn));
+
+        if (urn.StartsWith(MessageUrn.Prefix))
+            throw new ArgumentException($"Value should not contain the default prefix '{MessageUrn.Prefix}'.", nameof(urn));
+
+        Urn = FormatUrn(urn);
+    }
+
+    public Uri Urn { get; }
+
+    private static Uri FormatUrn(string urn)
+    {
+        var fullValue = MessageUrn.Prefix + urn;
+
+        if (Uri.TryCreate(fullValue, UriKind.Absolute, out var uri))
+            return uri;
+
+        throw new UriFormatException($"Invalid URN: {fullValue}");
+    }
+}
+
+
+
 public static class MessageUrn
 {
     public const string Prefix = "urn:message:";

src/Blumchen/Subscriptions/ISubscriptionOptions.cs

@@ -10,7 +10,7 @@ public interface ISubscriptionOptions
 {
     [UsedImplicitly] NpgsqlDataSource DataSource { get; }
     [UsedImplicitly] NpgsqlConnectionStringBuilder ConnectionStringBuilder { get; }
-    IReplicationDataMapper DataMapper { get; }
+    IDictionary<string, Tuple<IReplicationJsonBMapper, IMessageHandler>> Registry { get; }
     [UsedImplicitly] PublicationSetupOptions PublicationOptions { get; }
     [UsedImplicitly] ReplicationSlotSetupOptions ReplicationOptions { get; }
     [UsedImplicitly] IErrorProcessor ErrorProcessor { get; }
@@ -21,8 +21,7 @@ void Deconstruct(
         out PublicationSetupOptions publicationSetupOptions,
         out ReplicationSlotSetupOptions replicationSlotSetupOptions,
         out IErrorProcessor errorProcessor,
-        out IReplicationDataMapper dataMapper,
-        out Dictionary<Type, IMessageHandler> registry);
+        out IDictionary<string, Tuple<IReplicationJsonBMapper, IMessageHandler>> registry);
 }
 
 internal record SubscriptionOptions(
@@ -31,5 +30,4 @@ internal record SubscriptionOptions(
     PublicationSetupOptions PublicationOptions,
     ReplicationSlotSetupOptions ReplicationOptions,
     IErrorProcessor ErrorProcessor,
-    IReplicationDataMapper DataMapper,
-    Dictionary<Type, IMessageHandler> Registry): ISubscriptionOptions;
+    IDictionary<string, Tuple<IReplicationJsonBMapper, IMessageHandler>> Registry): ISubscriptionOptions;

src/Blumchen/Subscriptions/Management/PublicationManagement.cs

@@ -1,3 +1,4 @@
+using System.Text.Json.Serialization.Metadata;
 using Blumchen.Database;
 using Blumchen.Serialization;
 using Npgsql;
@@ -16,37 +17,35 @@ public static async Task<SetupPublicationResult> SetupPublication(
         CancellationToken ct
     )
     {
-        var (publicationName, createStyle, shouldReAddTablesIfWereRecreated, typeResolver, tableDescription) = setupOptions;
+        var (publicationName, createStyle, shouldReAddTablesIfWereRecreated, registeredTypes, tableDescription) = setupOptions;
 
         return createStyle switch
         {
             Subscription.CreateStyle.Never => new None(),
-            Subscription.CreateStyle.AlwaysRecreate => await ReCreate(dataSource, publicationName, tableDescription.Name, typeResolver, ct).ConfigureAwait(false),
+            Subscription.CreateStyle.AlwaysRecreate => await ReCreate(dataSource, publicationName, tableDescription.Name, registeredTypes, ct).ConfigureAwait(false),
             Subscription.CreateStyle.WhenNotExists when await dataSource.PublicationExists(publicationName, ct).ConfigureAwait(false) => await Refresh(dataSource, publicationName, tableDescription.Name, shouldReAddTablesIfWereRecreated, ct).ConfigureAwait(false),
-            Subscription.CreateStyle.WhenNotExists => await Create(dataSource, publicationName, tableDescription.Name, typeResolver, ct).ConfigureAwait(false),
+            Subscription.CreateStyle.WhenNotExists => await Create(dataSource, publicationName, tableDescription.Name, registeredTypes, ct).ConfigureAwait(false),
             _ => throw new ArgumentOutOfRangeException(nameof(setupOptions.CreateStyle))
         };
 
         static async Task<SetupPublicationResult> ReCreate(
             NpgsqlDataSource dataSource,
             string publicationName,
             string tableName,
-            JsonTypeResolver? typeResolver,
+            ISet<string> registeredTypes,
             CancellationToken ct
         ) {
             await dataSource.DropPublication(publicationName, ct).ConfigureAwait(false);
-            return await Create(dataSource, publicationName, tableName, typeResolver, ct).ConfigureAwait(false);
+            return await Create(dataSource, publicationName, tableName, registeredTypes, ct).ConfigureAwait(false);
         }
 
         static async Task<SetupPublicationResult> Create(NpgsqlDataSource dataSource,
             string publicationName,
             string tableName,
-            JsonTypeResolver? typeResolver,
+            ISet<string> registeredTypes,
             CancellationToken ct
         ) {
-            await dataSource.CreatePublication(publicationName, tableName,
-                typeResolver.Keys().ToHashSet(), ct).ConfigureAwait(false);
-
+            await dataSource.CreatePublication(publicationName, tableName, registeredTypes, ct).ConfigureAwait(false);
             return new Created();
         }
 
@@ -66,14 +65,14 @@ internal static Task CreatePublication(
         this NpgsqlDataSource dataSource,
         string publicationName,
         string tableName,
-        ISet<string> eventTypes,
+        ISet<string> registeredTypes,
         CancellationToken ct
     ) {
         var sql = $"CREATE PUBLICATION \"{publicationName}\" FOR TABLE {tableName} {{0}} WITH (publish = 'insert');";
-        return eventTypes.Count switch
+        return registeredTypes.Count switch
         {
             0 => Execute(dataSource, string.Format(sql,string.Empty), ct),
-            _ => Execute(dataSource, string.Format(sql, $"WHERE ({PublicationFilter(eventTypes)})"), ct)
+            _ => Execute(dataSource, string.Format(sql, $"WHERE ({PublicationFilter(registeredTypes)})"), ct)
         };
         static string PublicationFilter(ICollection<string> input) => string.Join(" OR ", input.Select(s => $"message_type = '{s}'"));
     }
@@ -143,21 +142,21 @@ public sealed record PublicationSetupOptions(
     )
     {
         internal const string DefaultPublicationName = "pub";
-        internal JsonTypeResolver? TypeResolver { get; init; } = default;
+        internal ISet<string> RegisteredTypes { get; init; } = Enumerable.Empty<string>().ToHashSet();
 
         internal TableDescriptorBuilder.MessageTable TableDescriptor { get; init; } = new TableDescriptorBuilder().Build();
 
         internal void Deconstruct(
             out string publicationName,
             out Subscription.CreateStyle createStyle,
             out bool reAddTablesIfWereRecreated,
-            out JsonTypeResolver? typeResolver,
+            out ISet<string> registeredTypes,
             out TableDescriptorBuilder.MessageTable tableDescription)
         {
             publicationName = PublicationName;
             createStyle = Subscription.CreateStyle.WhenNotExists;
             reAddTablesIfWereRecreated = ShouldReAddTablesIfWereRecreated;
-            typeResolver = TypeResolver;
+            registeredTypes = RegisteredTypes;
             tableDescription = TableDescriptor;
         }
 

src/Blumchen/Subscriptions/Replication/IReplicationDataMapper.cs

@@ -9,4 +9,6 @@ public interface IReplicationDataMapper
     Task<IEnvelope> ReadFromSnapshot(NpgsqlDataReader reader, CancellationToken ct);
 
     Task<IEnvelope> ReadFromReplication(InsertMessage insertMessage, CancellationToken ct);
+
+
 }