CompositeCodec constructors require at least one delegate

* Adjust the behavior of the class so that delegates are required.

* In the current implementation, users could pass in an empty list to the composite codec.
* This is an CompositeCodec and if we had no codecs, then why have the class.
* All @nullable's were removed from the delegate attribute.
* Null returns from the findDelegate have been removed.   The default delegate will be returned in the stead.
* Update the tests to require at least one codec in the delegates

* Update code to meet code review requirements

* Add Author tag to affected classes

* Add tests to the CompositeCodecTests to tests delegates

And also test if a user attempts to an invalid class.

* Update deprecated ctor so that it will only establish default codec

* Simplify the initialization of the `codec` attribute

* Update the javadoc for the CompositeCodec class.

Change the  `ClassUtils.findClosestMatch` `failOnTie` to true.
  In the past when finding the appropriate delegate(codec) for a object and
  their were multiple delegates available the first one was returned.
  The problem is it could return a different delegate on each launch of the `findDelegate`.
  Now with `failOnTie` set to true a `IllegalStateException` is returned.

Remove deprecated constructor from CompositeCodec

* Replace sample classes in CompositeCodecTests with records to reduce maintenance

Rename field names of foo and foo2 in CompositeCodecTests to professional descriptive names
In the onlyDefaultCodec implementation in the CompositeCodecTests use a class other than a PojoCodec class to simulate a real world case.
Enhance javadoc for CompositeCodec to include information on ClassUtils.findClosestMatch and its effects on codec selection.
Make sure @see is in the correct location in javadoc

* Add CompositeCodec documentation to reference documentation

Provide example for using CompositeCodec.
Remove @see in CompositeCodec javadocs.   This is handled in the links in the javadoc

* In codec.adoc all sentences must be on same line

Add code snippet into docs for classes
Verify no tabs exist in sample

Author: cppwfs

spring-integration-core/src/main/java/org/springframework/integration/codec/CompositeCodec.java

@@ -23,14 +23,24 @@
 import java.util.HashMap;
 import java.util.Map;
 
-import org.jspecify.annotations.Nullable;
-
 import org.springframework.integration.util.ClassUtils;
 import org.springframework.util.Assert;
 
 /**
- * A Codec that can delegate to one out of many Codecs, each mapped to a class.
+ * An implementation of {@link Codec} that combines multiple codecs into a single codec,
+ * delegating encoding and decoding operations to the appropriate type-specific codec.
+ * This implementation associates object types with their appropriate codecs while providing a fallback default codec
+ * for unregistered types.
+ * This class uses {@code ClassUtils.findClosestMatch} to select the appropriate codec for a given object type.
+ * When multiple codecs match an object type, {@code ClassUtils.findClosestMatch} offers the
+ * {@code failOnTie} option. If {@code failOnTie} is {@code false}, it will return any one of the matching codecs.
+ * If {@code failOnTie} is {@code true} and multiple codecs match, it will throw an {@code IllegalStateException}.
+ * {@link CompositeCodec} sets {@code failOnTie} to {@code true}, so if multiple codecs match, an
+ * {@code IllegalStateException} is thrown.
+ *
  * @author David Turanski
+ * @author Glenn Renfro
+ *
  * @since 4.2
  */
 public class CompositeCodec implements Codec {
@@ -41,63 +51,38 @@ public class CompositeCodec implements Codec {
 
 	public CompositeCodec(Map<Class<?>, Codec> delegates, Codec defaultCodec) {
 		this.defaultCodec = defaultCodec;
-		this.delegates = new HashMap<Class<?>, Codec>(delegates);
-	}
-
-	public CompositeCodec(Codec defaultCodec) {
-		this(Map.of(), defaultCodec);
+		Assert.notEmpty(delegates, "delegates must not be empty");
+		this.delegates = new HashMap<>(delegates);
 	}
 
 	@Override
 	public void encode(Object object, OutputStream outputStream) throws IOException {
 		Assert.notNull(object, "cannot encode a null object");
 		Assert.notNull(outputStream, "'outputStream' cannot be null");
-		Codec codec = findDelegate(object.getClass());
-		if (codec != null) {
-			codec.encode(object, outputStream);
-		}
-		else {
-			this.defaultCodec.encode(object, outputStream);
-		}
+		findDelegate(object.getClass()).encode(object, outputStream);
 	}
 
 	@Override
 	public byte[] encode(Object object) throws IOException {
 		Assert.notNull(object, "cannot encode a null object");
-		Codec codec = findDelegate(object.getClass());
-		if (codec != null) {
-			return codec.encode(object);
-		}
-		else {
-			return this.defaultCodec.encode(object);
-		}
+		return findDelegate(object.getClass()).encode(object);
 	}
 
 	@Override
 	public <T> T decode(InputStream inputStream, Class<T> type) throws IOException {
 		Assert.notNull(inputStream, "'inputStream' cannot be null");
 		Assert.notNull(type, "'type' cannot be null");
-		Codec codec = findDelegate(type);
-		if (codec != null) {
-			return codec.decode(inputStream, type);
-		}
-		else {
-			return this.defaultCodec.decode(inputStream, type);
-		}
+		return findDelegate(type).decode(inputStream, type);
 	}
 
 	@Override
 	public <T> T decode(byte[] bytes, Class<T> type) throws IOException {
 		return decode(new ByteArrayInputStream(bytes), type);
 	}
 
-	private @Nullable Codec findDelegate(Class<?> type) {
-		if (this.delegates.isEmpty()) {
-			return null;
-		}
-
-		Class<?> clazz = ClassUtils.findClosestMatch(type, this.delegates.keySet(), false);
-		return this.delegates.get(clazz);
+	private Codec findDelegate(Class<?> type) {
+		Class<?> clazz = ClassUtils.findClosestMatch(type, this.delegates.keySet(), true);
+		return clazz == null ? this.defaultCodec : this.delegates.getOrDefault(clazz, this.defaultCodec);
 	}
 
 }

spring-integration-core/src/test/java/org/springframework/integration/codec/kryo/CompositeCodecTests.java

@@ -17,68 +17,75 @@
 package org.springframework.integration.codec.kryo;
 
 import java.io.IOException;
-import java.util.HashMap;
 import java.util.Map;
 
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.integration.codec.Codec;
 import org.springframework.integration.codec.CompositeCodec;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
 
 /**
  * @author David Turanski
+ * @author Glenn Renfro
  * @since 4.2
  */
 public class CompositeCodecTests {
 
-	private Codec codec;
-
-	@BeforeEach
-	public void setup() {
-		Map<Class<?>, Codec> codecs = new HashMap<>();
-		this.codec = new CompositeCodec(codecs, new PojoCodec(
-				new KryoClassListRegistrar(SomeClassWithNoDefaultConstructors.class)));
-	}
-
 	@Test
-	public void testPojoSerialization() throws IOException {
-		SomeClassWithNoDefaultConstructors foo = new SomeClassWithNoDefaultConstructors("hello", 123);
-		SomeClassWithNoDefaultConstructors foo2 = this.codec.decode(
-				this.codec.encode(foo),
+	void testWithCodecDelegates() throws IOException {
+		Codec codec = getFullyQualifiedCodec();
+		SomeClassWithNoDefaultConstructors inputInstance = new SomeClassWithNoDefaultConstructors("hello", 123);
+		SomeClassWithNoDefaultConstructors outputInstance = codec.decode(
+				codec.encode(inputInstance),
 				SomeClassWithNoDefaultConstructors.class);
-		assertThat(foo2).isEqualTo(foo);
+		assertThat(outputInstance).isEqualTo(inputInstance);
 	}
 
-	static class SomeClassWithNoDefaultConstructors {
+	@Test
+	void testWithCodecDefault() throws IOException {
+		Codec codec = getFullyQualifiedCodec();
+		AnotherClassWithNoDefaultConstructors inputInstance = new AnotherClassWithNoDefaultConstructors("hello", 123);
+		AnotherClassWithNoDefaultConstructors outputInstance = codec.decode(
+				codec.encode(inputInstance),
+				AnotherClassWithNoDefaultConstructors.class);
+		assertThat(outputInstance).isEqualTo(inputInstance);
+	}
 
-		private String val1;
+	@Test
+	void testWithUnRegisteredClass() throws IOException {
+		// Verify that the default encodes and decodes properly
+		Codec codec = onlyDefaultCodec();
+		SomeClassWithNoDefaultConstructors inputInstance = new SomeClassWithNoDefaultConstructors("hello", 123);
+		SomeClassWithNoDefaultConstructors outputInstance = codec.decode(
+				codec.encode(inputInstance),
+				SomeClassWithNoDefaultConstructors.class);
+		assertThat(outputInstance).isEqualTo(inputInstance);
 
-		private int val2;
+		// Verify that an exception is thrown if an unknown type is to be encoded.
+		assertThatIllegalArgumentException().isThrownBy(() -> codec.decode(
+				codec.encode(inputInstance),
+				AnotherClassWithNoDefaultConstructors.class));
+	}
 
-		SomeClassWithNoDefaultConstructors(String val1, int val2) {
-			this.val1 = val1;
-			this.val2 = val2;
-		}
+	private static Codec getFullyQualifiedCodec() {
+		Map<Class<?>, Codec> codecs = Map.of(SomeClassWithNoDefaultConstructors.class, new PojoCodec(
+				new KryoClassListRegistrar(SomeClassWithNoDefaultConstructors.class)));
+		return new CompositeCodec(codecs, new PojoCodec(
+				new KryoClassListRegistrar(AnotherClassWithNoDefaultConstructors.class)));
+	}
 
-		@Override
-		public boolean equals(Object other) {
-			if (!(other instanceof SomeClassWithNoDefaultConstructors)) {
-				return false;
-			}
-			SomeClassWithNoDefaultConstructors that = (SomeClassWithNoDefaultConstructors) other;
-			return (this.val1.equals(that.val1) && this.val2 == that.val2);
-		}
+	private static Codec onlyDefaultCodec() {
+		PojoCodec pojoCodec = new PojoCodec();
+		Map<Class<?>, Codec> codecs = Map.of(java.util.Date.class, pojoCodec);
+		return new CompositeCodec(codecs, new PojoCodec(
+				new KryoClassListRegistrar(SomeClassWithNoDefaultConstructors.class)));
+	}
 
-		@Override
-		public int hashCode() {
-			int result = this.val1.hashCode();
-			result = 31 * result + this.val2;
-			return result;
-		}
+	private record SomeClassWithNoDefaultConstructors(String val1, int val2) { }
 
-	}
+	private record AnotherClassWithNoDefaultConstructors(String val1, int val2) { }
 
 }

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java

@@ -20,8 +20,8 @@
 import java.io.ByteArrayOutputStream;
 import java.net.InetAddress;
 import java.net.Socket;
+import java.nio.charset.StandardCharsets;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.Map;
 
 import javax.net.SocketFactory;
@@ -35,7 +35,6 @@
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.codec.Codec;
 import org.springframework.integration.codec.CodecMessageConverter;
-import org.springframework.integration.codec.CompositeCodec;
 import org.springframework.integration.codec.kryo.MessageCodec;
 import org.springframework.integration.ip.IpHeaders;
 import org.springframework.integration.ip.tcp.serializer.MapJsonSerializer;
@@ -56,6 +55,7 @@
  * @author Gary Russell
  * @author Artem Bilan
  * @author Gengwu Zhao
+ * @author Glenn Renfro
  * @since 2.0
  *
  */
@@ -67,8 +67,7 @@ public class TcpMessageMapperTests {
 
 	@BeforeEach
 	public void setup() {
-		Map<Class<?>, Codec> codecs = new HashMap<>();
-		this.codec = new CompositeCodec(codecs, new MessageCodec());
+		this.codec = new MessageCodec();
 	}
 
 	@Test
@@ -339,7 +338,7 @@ public void testMapMessageConvertingOutboundJson() throws Exception {
 		MapJsonSerializer serializer = new MapJsonSerializer();
 		ByteArrayOutputStream baos = new ByteArrayOutputStream();
 		serializer.serialize(map, baos);
-		assertThat(new String(baos.toByteArray(), "UTF-8"))
+		assertThat(baos.toString(StandardCharsets.UTF_8))
 				.isEqualTo("{\"headers\":{\"bar\":\"baz\"},\"payload\":\"foo\"}\n");
 	}
 

src/reference/antora/modules/ROOT/pages/codec.adoc

@@ -39,10 +39,11 @@ See the https://docs.spring.io/spring-integration/api/org/springframework/integr
 [[kryo]]
 == Kryo
 
-Currently, this is the only implementation of `Codec`, and it provides two kinds of `Codec`:
+Currently, this is the only implementation of `Codec`, and it provides three kinds of `Codec`:
 
 * `PojoCodec`: Used in the transformers
 * `MessageCodec`: Used in the `CodecMessageConverter`
+* `CompositeCodec`: Used in transformers
 
 The framework provides several custom serializers:
 
@@ -53,6 +54,70 @@ The framework provides several custom serializers:
 The first can be used with the `PojoCodec` by initializing it with the `FileKryoRegistrar`.
 The second and third are used with the `MessageCodec`, which is initialized with the `MessageKryoRegistrar`.
 
+[[composite-codec]]
+=== CompositeCodec
+
+The `CompositeCodec` is a codec that combines multiple codecs into a single codec, delegating encoding and decoding operations to the appropriate type-specific codec.
+This implementation associates object types with their appropriate codecs while providing a fallback default codec for unregistered types.
+
+An example implementation can be seen below:
+```java
+void encodeDecodeSample() {
+        Codec codec = getFullyQualifiedCodec();
+
+        //Encode and Decode a Dog Object
+        Dog dog = new Dog("Wolfy", 3, "woofwoof");
+        dog = codec.decode(
+                codec.encode(dog),
+                Dog.class);
+        System.out.println(dog);
+
+        //Encode and Decode a Cat Object
+        Cat cat = new Cat("Kitty", 2, 8);
+        cat = codec.decode(
+                codec.encode(cat),
+                Cat.class);
+        System.out.println(cat);
+
+        //Use the default code if the type being decoded and encoded is not Cat or dog.
+        Animal animal = new Animal("Badger", 5);
+        Animal animalOut = codec.decode(
+                codec.encode(animal),
+                Animal.class);
+        System.out.println(animalOut);
+}
+
+/**
+ * Create and return a {@link CompositeCodec} that associates {@code Dog} and {@code Cat}
+ * classes with their respective {@link PojoCodec} instances, while providing a default
+ * codec for {@code Animal} types.
+ * <p>
+ * @return a fully qualified {@link CompositeCodec} for {@code Dog}, {@code Cat},
+ *     and fallback for {@code Animal}
+ */
+static Codec getFullyQualifiedCodec() {
+    Map<Class<?>, Codec> codecs = new HashMap<Class<?>, Codec>();
+    codecs.put(Dog.class, new PojoCodec(new KryoClassListRegistrar(Dog.class)));
+    codecs.put(Cat.class, new PojoCodec(new KryoClassListRegistrar(Cat.class)));
+    return new CompositeCodec(codecs, new PojoCodec(
+            new KryoClassListRegistrar(Animal.class)));
+}
+
+// Records that will be encoded and decoded in this sample
+record Dog(String name, int age, String tag) {}
+record Cat(String name, int age, int lives) {}
+record Animal(String name, int age){}
+```
+
+In some cases a single type of object may return multiple codecs.
+In these cases an `IllegalStateException` is thrown.
+
+NOTE: This class uses `ClassUtils.findClosestMatch` to select the appropriate codec for a given object type.
+When multiple codecs match an object type, `ClassUtils.findClosestMatch` offers the `failOnTie` option.
+If `failOnTie` is `false`, it will return any one of the matching codecs.
+If `failOnTie` is `true` and multiple codecs match, it will throw an `IllegalStateException`.
+CompositeCodec` sets `failOnTie` to `true`, so if multiple codecs match, an `IllegalStateException` is thrown.
+
 [[customizing-kryo]]
 === Customizing Kryo