GH-9416: Extract BaseMessageBuilder for easier message extensions

Fixes: #9416
Issue link: https://github.com/spring-projects/spring-integration/issues/9416

The `MessageBuilderFactory` bean could be used a central place to provide custom `Message`
implementation into the application.
For example, the `GenericMessage.toString()` can be overridden to remove or mask sensitive
information from the payload or headers.

* Extract a `BaseMessageBuilder` from the `MessageBuilder` class to simplify
a custom `MessageBuilderFactory` implementation
* Test and document new feature and its purpose

spring-integration-core/src/test/java/org/springframework/integration/support/MessageBuilderTests.java

@@ -16,16 +16,22 @@
 
 package org.springframework.integration.support;
 
-import org.junit.Test;
+import java.io.Serial;
+import java.util.Map;
+import java.util.stream.Collectors;
 
+import org.junit.jupiter.api.Test;
+
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 4.3.10
- *
  */
 public class MessageBuilderTests {
 
@@ -45,4 +51,71 @@ public class MessageBuilderTests {
 		assertThat(message.getHeaders().get("qux")).isNull();
 	}
 
+	@Test
+	public void personalInfoHeadersAreMaskedWithCustomMessage() {
+		Message<String> message =
+				MessageBuilder.withPayload("some_user")
+						.setHeader("password", "some_password")
+						.build();
+
+		Message<String> piiMessage = new PiiMessageBuilderFactory().fromMessage(message).build();
+
+		assertThat(piiMessage).isInstanceOf(PiiMessage.class);
+		assertThat(piiMessage.getPayload()).isEqualTo("some_user");
+		assertThat(piiMessage.getHeaders().get("password")).isEqualTo("some_password");
+		assertThat(piiMessage.toString())
+				.doesNotContain("some_password")
+				.contains("******");
+	}
+
+	private static class PiiMessageBuilderFactory implements MessageBuilderFactory {
+
+		@Override
+		public <T> PiiMessageBuilder<T> fromMessage(Message<T> message) {
+			return new PiiMessageBuilder<>(message.getPayload(), message);
+		}
+
+		@Override
+		public <T> PiiMessageBuilder<T> withPayload(T payload) {
+			return new PiiMessageBuilder<>(payload, null);
+		}
+
+	}
+
+	private static class PiiMessageBuilder<P> extends BaseMessageBuilder<P, PiiMessageBuilder<P>> {
+
+		public PiiMessageBuilder(P payload, @Nullable Message<P> originalMessage) {
+			super(payload, originalMessage);
+		}
+
+		@Override
+		public Message<P> build() {
+			return new PiiMessage<>(getPayload(), getHeaders());
+		}
+
+	}
+
+	private static class PiiMessage<P> extends GenericMessage<P> {
+
+		@Serial
+		private static final long serialVersionUID = -354503673433669578L;
+
+		public PiiMessage(P payload, Map<String, Object> headers) {
+			super(payload, headers);
+		}
+
+		@Override
+		public String toString() {
+			return "PiiMessage [payload=" + getPayload() + ", headers=" + maskHeaders(getHeaders()) + ']';
+		}
+
+		private static Map<String, Object> maskHeaders(Map<String, Object> headers) {
+			return headers.entrySet()
+					.stream()
+					.map((entry) -> entry.getKey().equals("password") ? Map.entry(entry.getKey(), "******") : entry)
+					.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+		}
+
+	}
+
 }