From 74da56d2b2e484b0f1559b849b7a4f4503bfc518 Mon Sep 17 00:00:00 2001
From: Robert Schie <54124441+RobotechUSA@users.noreply.github.com>
Date: Wed, 13 Nov 2024 20:52:58 -0800
Subject: [PATCH] Feature: Added Qualified and Custom reference name strategy
 approaches for protobuf references (Schema Registry) (#2345)

* Added QualifiedReferenceSubjectNameStrategy implementation to have feature parity with java version.
Added ability to use any custom ReferenceSubjectNameStrategy by implementing ICustomReferenceSubjectNameStrategy and assigning implemented instance to ProtobufSerializerConfig property CustomReferenceSubjectNameStrategy (Note: ReferenceSubjectNameStrategy enumeration value must be set to Custom)
Code for ProtobufSerializer was reformatted

* Empty-Commit
---
 .../ProtobufSerializer.cs                     |  2 +-
 .../ProtobufSerializerConfig.cs               |  5 +
 .../ReferenceSubjectNameStrategy.cs           | 95 ++++++++++++++++++-
 3 files changed, 96 insertions(+), 6 deletions(-)

diff --git a/src/Confluent.SchemaRegistry.Serdes.Protobuf/ProtobufSerializer.cs b/src/Confluent.SchemaRegistry.Serdes.Protobuf/ProtobufSerializer.cs
index ab0314349..898ec21da 100644
--- a/src/Confluent.SchemaRegistry.Serdes.Protobuf/ProtobufSerializer.cs
+++ b/src/Confluent.SchemaRegistry.Serdes.Protobuf/ProtobufSerializer.cs
@@ -96,7 +96,7 @@ public ProtobufSerializer(ISchemaRegistryClient schemaRegistryClient, ProtobufSe
             if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToDelegate(); }
             this.referenceSubjectNameStrategy = config.ReferenceSubjectNameStrategy == null
                 ? ReferenceSubjectNameStrategy.ReferenceName.ToDelegate()
-                : config.ReferenceSubjectNameStrategy.Value.ToDelegate();
+                : config.ReferenceSubjectNameStrategy.Value.ToDelegate(config.CustomReferenceSubjectNameStrategy);
 
             if (this.useLatestVersion && this.autoRegisterSchema)
             {
diff --git a/src/Confluent.SchemaRegistry.Serdes.Protobuf/ProtobufSerializerConfig.cs b/src/Confluent.SchemaRegistry.Serdes.Protobuf/ProtobufSerializerConfig.cs
index 97770451e..b40a0d9ca 100644
--- a/src/Confluent.SchemaRegistry.Serdes.Protobuf/ProtobufSerializerConfig.cs
+++ b/src/Confluent.SchemaRegistry.Serdes.Protobuf/ProtobufSerializerConfig.cs
@@ -274,5 +274,10 @@ public ReferenceSubjectNameStrategy? ReferenceSubjectNameStrategy
             }
         }
 
+        /// <summary>
+        /// Custom reference subject name strategy resolver.
+        /// Ensure <seealso cref="ReferenceSubjectNameStrategy.Custom"/> is set for this to take effect.
+        /// </summary>
+        public ICustomReferenceSubjectNameStrategy CustomReferenceSubjectNameStrategy { get; set; } = null;
     }
 }
diff --git a/src/Confluent.SchemaRegistry/ReferenceSubjectNameStrategy.cs b/src/Confluent.SchemaRegistry/ReferenceSubjectNameStrategy.cs
index 492f1c11d..17d80a9a8 100644
--- a/src/Confluent.SchemaRegistry/ReferenceSubjectNameStrategy.cs
+++ b/src/Confluent.SchemaRegistry/ReferenceSubjectNameStrategy.cs
@@ -15,6 +15,7 @@
 // Refer to LICENSE for more information.
 
 using Confluent.Kafka;
+using System;
 
 
 namespace Confluent.SchemaRegistry
@@ -38,11 +39,45 @@ namespace Confluent.SchemaRegistry
     public enum ReferenceSubjectNameStrategy
     {
         /// <summary>
-        ///     (default): Use the reference name as the subject name.
+        ///    ReferenceName (default): Use the reference name as the subject name.
         /// </summary>
-        ReferenceName
+        ReferenceName,
+        /// <summary>
+        ///    Qualified: Given a reference name, replace slashes with dots, and remove the .proto suffix to obtain the subject name. 
+        /// For example, mypackage/myfile.proto becomes mypackage.myfile.
+        /// </summary>
+        Qualified,
+        /// <summary>
+        ///    Custom: Use a custom reference subject name strategy resolver to determine the subject name.
+        /// </summary>
+        Custom
+    }
+
+    /// <summary>
+    /// Custom Reference Subject Name Strategy Interface
+    /// </summary>
+    public interface ICustomReferenceSubjectNameStrategy
+    {
+        /// <summary>
+        /// Gets the subject name.
+        /// </summary>
+        /// <param name="context"></param>
+        /// <param name="referenceName"></param>
+        /// <returns></returns>
+        string GetSubjectName(SerializationContext context, string referenceName);
+    }
+
+    /// <summary>
+    ///     Configuration property names specific to the schema registry client.
+    /// </summary>
+    public static partial class PropertyNames
+    {
+        /// <summary>
+        ///     The subject name strategy to use for registration / lookup of referenced schemas
+        ///     Possible values: <see cref="Confluent.SchemaRegistry.ReferenceSubjectNameStrategy" />
+        /// </summary>
+        public const string ReferenceSubjectNameStrategy = "protobuf.serializer.reference.subject.name.strategy";
     }
-    
 
     /// <summary>
     ///     Extension methods for the ReferenceSubjectNameStrategy type.
@@ -52,7 +87,57 @@ public static class ReferenceSubjectNameStrategyExtensions
         /// <summary>
         ///     Provide a functional implementation corresponding to the enum value.
         /// </summary>
-        public static ReferenceSubjectNameStrategyDelegate ToDelegate(this ReferenceSubjectNameStrategy strategy)
-            => (context, referenceName) => referenceName;
+        public static ReferenceSubjectNameStrategyDelegate ToDelegate(this ReferenceSubjectNameStrategy strategy, ICustomReferenceSubjectNameStrategy customReferenceSubjectNameStrategy = null)
+        {
+            return (context, referenceName) =>
+            {
+                switch (strategy)
+                {
+                    case ReferenceSubjectNameStrategy.ReferenceName:
+                        {
+                            return GetReferenceNameSubjectName(context, referenceName);
+                        }
+                    case ReferenceSubjectNameStrategy.Qualified:
+                        {
+                            return GetQualifiedSubjectName(context, referenceName);
+                        }
+                    case ReferenceSubjectNameStrategy.Custom:
+                        {
+                            if (customReferenceSubjectNameStrategy == null)
+                            {
+                                throw new ArgumentException($"Custom strategy requires a custom {nameof(ICustomReferenceSubjectNameStrategy)} implementation to be specified.");
+                            }
+
+                            return customReferenceSubjectNameStrategy.GetSubjectName(context, referenceName);
+                        }
+                    default:
+                        {
+                            throw new ArgumentException($"Unknown ${PropertyNames.ReferenceSubjectNameStrategy} value: {strategy}.");
+                        }
+                }
+            };
+        }
+
+        /// <summary>
+        /// Get Subject Name
+        /// </summary>
+        /// <param name="context"></param>
+        /// <param name="referenceName"></param>
+        /// <returns></returns>
+        public static string GetQualifiedSubjectName(SerializationContext context, string referenceName)
+        {
+            return referenceName.Replace(".proto", string.Empty).Replace("/", ".");
+        }
+
+        /// <summary>
+        /// Get Subject Name
+        /// </summary>
+        /// <param name="context"></param>
+        /// <param name="referenceName"></param>
+        /// <returns></returns>
+        public static string GetReferenceNameSubjectName(SerializationContext context, string referenceName)
+        {
+            return referenceName;
+        }
     }
 }