Merge branch 'feature/helper_subscribeToFeed' into develop

runtime/src/main/java/com/fluxtion/runtime/output/AbstractMessageSink.java

@@ -0,0 +1,25 @@
+/*
+ * SPDX-FileCopyrightText: © 2024 Gregory Higgins <[email protected]>
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package com.fluxtion.runtime.output;
+
+import java.util.function.Function;
+
+public abstract class AbstractMessageSink<T> implements MessageSink<T> {
+
+    private Function<? super T, ?> valueMapper = Function.identity();
+
+    @Override
+    public final void accept(T t) {
+        sendToSink(valueMapper.apply(t));
+    }
+
+    @Override
+    public void setValueMapper(Function<? super T, ?> valueMapper) {
+        this.valueMapper = valueMapper;
+    }
+
+    abstract protected void sendToSink(Object value);
+}

runtime/src/main/java/com/fluxtion/runtime/output/MessageSink.java

@@ -1,12 +1,23 @@
+/*
+ * SPDX-FileCopyrightText: © 2024 Gregory Higgins <[email protected]>
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
 package com.fluxtion.runtime.output;
 
 import java.util.function.Consumer;
+import java.util.function.Function;
 
 /**
  * A sink for an EventProcessor. Implement this interface and register with :
- *
+ * <p>
  * {@link com.fluxtion.runtime.EventProcessor#registerService(Object, Class, String)}
+ *
  * @param <T> the type of message published to the Sink
  */
 public interface MessageSink<T> extends Consumer<T> {
-}
+
+    default void setValueMapper(Function<? super T, ?> valueMapper) {
+        throw new UnsupportedOperationException("setValueMapper not implemented");
+    }
+}