Ability to replace interop strategies for both Kafka and MQTT transpo…

…rts. Closes GH-596. Closes GH-597
JasperFx · Oct 20, 2023 · ee979dc · ee979dc
1 parent 5d9cf21
commit ee979dc
Show file tree

Hide file tree

Showing 4 changed files with 46 additions and 1 deletion.
diff --git a/src/Transports/Kafka/Wolverine.Kafka/KafkaListenerConfiguration.cs b/src/Transports/Kafka/Wolverine.Kafka/KafkaListenerConfiguration.cs
@@ -12,5 +12,15 @@ public KafkaListenerConfiguration(Func<KafkaTopic> source) : base(source)
     {
     }
 
-
+    /// <summary>
+    /// Use a custom interoperability strategy to map Wolverine messages to an upstream
+    /// system's protocol
+    /// </summary>
+    /// <param name="mapper"></param>
+    /// <returns></returns>
+    public KafkaListenerConfiguration UseInterop(IKafkaEnvelopeMapper mapper)
+    {
+        add(e => e.Mapper = mapper);
+        return this;
+    }
 }
diff --git a/src/Transports/Kafka/Wolverine.Kafka/KafkaSubscriberConfiguration.cs b/src/Transports/Kafka/Wolverine.Kafka/KafkaSubscriberConfiguration.cs
@@ -8,4 +8,15 @@ internal KafkaSubscriberConfiguration(KafkaTopic endpoint) : base(endpoint)
     {
     }
 
+    /// <summary>
+    /// Use a custom interoperability strategy to map Wolverine messages to an upstream
+    /// system's protocol
+    /// </summary>
+    /// <param name="mapper"></param>
+    /// <returns></returns>
+    public KafkaSubscriberConfiguration UseInterop(IKafkaEnvelopeMapper mapper)
+    {
+        add(e => e.Mapper = mapper);
+        return this;
+    }
 }
diff --git a/src/Transports/MQTT/Wolverine.MQTT/MqttListenerConfiguration.cs b/src/Transports/MQTT/Wolverine.MQTT/MqttListenerConfiguration.cs
@@ -34,4 +34,16 @@ public MqttListenerConfiguration QualityOfService(MqttQualityOfServiceLevel serv
         add(e => e.QualityOfServiceLevel = serviceLevel);
         return this;
     }
+
+    /// <summary>
+    /// Use a custom interoperability strategy to map Wolverine messages to an upstream
+    /// system's protocol
+    /// </summary>
+    /// <param name="mapper"></param>
+    /// <returns></returns>
+    public MqttListenerConfiguration UseInterop(IMqttEnvelopeMapper mapper)
+    {
+        add(e => e.EnvelopeMapper = mapper);
+        return this;
+    }
 }
diff --git a/src/Transports/MQTT/Wolverine.MQTT/MqttSubscriberConfiguration.cs b/src/Transports/MQTT/Wolverine.MQTT/MqttSubscriberConfiguration.cs
@@ -30,4 +30,16 @@ public MqttSubscriberConfiguration QualityOfService(MqttQualityOfServiceLevel se
         add(e => e.QualityOfServiceLevel = serviceLevel);
         return this;
     }
+
+    /// <summary>
+    /// Use a custom interoperability strategy to map Wolverine messages to an upstream
+    /// system's protocol
+    /// </summary>
+    /// <param name="mapper"></param>
+    /// <returns></returns>
+    public MqttSubscriberConfiguration UseInterop(IMqttEnvelopeMapper mapper)
+    {
+        add(e => e.EnvelopeMapper = mapper);
+        return this;
+    }
 }