Merge branch '5.3.x'

Author: snicoll

build.gradle

@@ -29,7 +29,7 @@ configure(allprojects) { project ->
 		imports {
 			mavenBom "com.fasterxml.jackson:jackson-bom:2.13.1"
 			mavenBom "io.netty:netty-bom:4.1.74.Final"
-			mavenBom "io.projectreactor:reactor-bom:2020.0.16"
+			mavenBom "io.projectreactor:reactor-bom:2020.0.17-SNAPSHOT"
 			mavenBom "io.r2dbc:r2dbc-bom:Borca-RELEASE"
 			mavenBom "io.rsocket:rsocket-bom:1.1.1"
 			mavenBom "org.eclipse.jetty:jetty-bom:11.0.8"
@@ -250,6 +250,7 @@ configure(allprojects) { project ->
 		repositories {
 			mavenCentral()
 			maven { url "https://repo.spring.io/libs-spring-framework-build" }
+			maven { url "https://repo.spring.io/snapshot" } // reactor
 		}
 	}
 	configurations.all {

spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpClient.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -320,7 +320,7 @@ public Publisher<Void> apply(NettyInbound inbound, NettyOutbound outbound) {
 			TcpConnection<P> connection = new ReactorNettyTcpConnection<>(inbound, outbound,  codec, completionSink);
 			scheduler.schedule(() -> this.connectionHandler.afterConnected(connection));
 
-			inbound.withConnection(conn -> conn.addHandler(new StompMessageDecoder<>(codec)));
+			inbound.withConnection(conn -> conn.addHandlerFirst(new StompMessageDecoder<>(codec)));
 
 			inbound.receiveObject()
 					.cast(Message.class)