Upgrade to Spring Boot 2.4.2

- Upgrade to Spring Cloud Open Service Broker 3.3.0
- Resolve deprecation warnings

Closes #406

Author: royclarkson

gradle.properties

@@ -9,9 +9,9 @@ cfJavaClientVersion = 4.7.0.RELEASE
 checkstyleVersion = 8.37
 commonsTextVersion = 1.8
 immutablesVersion = 2.8.8
-openServiceBrokerVersion = 3.2.0
+openServiceBrokerVersion = 3.3.0
 pmdVersion = 6.29.0
-springBootVersion = 2.3.6.RELEASE
+springBootVersion = 2.4.2
 springCredhubVersion = 2.1.1.RELEASE
-springFrameworkVersion = 5.2.11.RELEASE
+springFrameworkVersion = 5.3.3
 wiremockVersion = 2.27.2

spring-cloud-app-broker-acceptance-tests/src/test/java/org/springframework/cloud/appbroker/acceptance/CloudFoundryAcceptanceTest.java

@@ -421,8 +421,8 @@ protected Mono<String> manageApps(String serviceInstanceName, String serviceName
 					.flatMap(appRoute ->
 						webClient.get()
 							.uri(URI.create(appRoute + "/" + operation + "/" + serviceName + "/" + planName + "/" + serviceInstanceId))
-							.exchange()
-							.flatMap(clientResponse -> clientResponse.toEntity(String.class))
+							.retrieve()
+							.toEntity(String.class)
 							.map(HttpEntity::getBody)));
 	}
 

spring-cloud-app-broker-autoconfigure/src/main/java/org/springframework/cloud/appbroker/autoconfigure/ServiceInstanceLogStreamAutoConfiguration.java

@@ -59,7 +59,7 @@ public WebSocketHandlerAdapter handlerAdapter() {
 	@Bean
 	public HandlerMapping logsHandlerMapping(StreamingLogWebSocketHandler webSocketHandler) {
 		Map<String, WebSocketHandler> map = new HashMap<>();
-		map.put("/logs/**/stream", webSocketHandler);
+		map.put("/logs/**", webSocketHandler);
 
 		SimpleUrlHandlerMapping handlerMapping = new SimpleUrlHandlerMapping();
 		handlerMapping.setOrder(1);

spring-cloud-app-broker-core/src/main/java/org/springframework/cloud/appbroker/service/WorkflowServiceInstanceBindingService.java

@@ -43,6 +43,7 @@
 import org.springframework.cloud.servicebroker.service.ServiceInstanceBindingService;
 import org.springframework.core.annotation.AnnotationAwareOrderComparator;
 import org.springframework.util.CollectionUtils;
+import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;
 
 public class WorkflowServiceInstanceBindingService implements ServiceInstanceBindingService {
@@ -139,7 +140,7 @@ private Flux<CreateServiceInstanceRouteBindingResponseBuilder> invokeRouteBindin
 
 	private boolean isAppBindingRequest(CreateServiceInstanceBindingRequest request) {
 		return request.getBindResource() != null
-			&& StringUtils.isEmpty(request.getBindResource().getRoute());
+			&& ObjectUtils.isEmpty(request.getBindResource().getRoute());
 	}
 
 	private boolean isRouteBindingRequest(CreateServiceInstanceBindingRequest request) {

spring-cloud-app-broker-logging/src/main/java/org/springframework/cloud/appbroker/logging/streaming/endpoint/StreamingLogWebSocketHandler.java

@@ -23,8 +23,8 @@
 import org.cloudfoundry.dropsonde.events.Envelope;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import reactor.core.publisher.EmitterProcessor;
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
 
 import org.springframework.cloud.appbroker.logging.streaming.events.ServiceInstanceLogEvent;
 import org.springframework.cloud.appbroker.logging.streaming.events.StartServiceInstanceLoggingEvent;
@@ -43,7 +43,7 @@ public class StreamingLogWebSocketHandler implements WebSocketHandler, Applicati
 
 	private final ApplicationEventPublisher eventPublisher;
 
-	private final ConcurrentHashMap<String, EmitterProcessor<Envelope>> processors = new ConcurrentHashMap<>();
+	private final ConcurrentHashMap<String, Sinks.One<Envelope>> envelopeSinks = new ConcurrentHashMap<>();
 
 	public StreamingLogWebSocketHandler(ApplicationEventPublisher eventPublisher) {
 		this.eventPublisher = eventPublisher;
@@ -56,14 +56,14 @@ public Mono<Void> handle(WebSocketSession session) {
 			session.getHandshakeInfo().getRemoteAddress(),
 			serviceInstanceId);
 
-		EmitterProcessor<Envelope> processor = processors
-			.computeIfAbsent(serviceInstanceId, s -> EmitterProcessor.create());
+		Sinks.One<Envelope> envelopeSink = envelopeSinks
+			.computeIfAbsent(serviceInstanceId, s -> Sinks.one());
 
 		eventPublisher.publishEvent(new StartServiceInstanceLoggingEvent(this, serviceInstanceId));
 		LOG.info("Published event to start streaming logs for service instance with ID {}", serviceInstanceId);
 
 		return session
-			.send(processor.map(envelope -> session
+			.send(envelopeSink.asMono().map(envelope -> session
 				.binaryMessage(dataBufferFactory -> dataBufferFactory.wrap(Envelope.ADAPTER.encode(envelope)))))
 			.then()
 			.doFinally(signalType -> afterConnectionClosed(session))
@@ -81,10 +81,10 @@ public void broadcastLogMessage(ServiceInstanceLogEvent event) {
 			LOG.debug("Received event to broadcast log message for " + event.getServiceInstanceId());
 		}
 
-		EmitterProcessor<Envelope> processor = this.processors.get(event.getServiceInstanceId());
-		if (processor == null) {
+		Sinks.One<Envelope> envelopeSink = this.envelopeSinks.get(event.getServiceInstanceId());
+		if (envelopeSink == null) {
 			if (LOG.isWarnEnabled()) {
-				LOG.warn("No processor found for {}, stopping log streaming", event.getServiceInstanceId());
+				LOG.warn("No sink found for {}, stopping log streaming", event.getServiceInstanceId());
 			}
 
 			eventPublisher.publishEvent(new StopServiceInstanceLoggingEvent(this, event.getServiceInstanceId()));
@@ -96,7 +96,7 @@ public void broadcastLogMessage(ServiceInstanceLogEvent event) {
 			LOG.debug("Sending message to client for {}", event.getServiceInstanceId());
 		}
 
-		processor.onNext(event.getEnvelope());
+		envelopeSink.tryEmitValue(event.getEnvelope());
 	}
 
 	private void afterConnectionClosed(WebSocketSession webSocketSession) {
@@ -106,7 +106,7 @@ private void afterConnectionClosed(WebSocketSession webSocketSession) {
 
 		final String serviceInstanceId = getServiceInstanceId(webSocketSession);
 		eventPublisher.publishEvent(new StopServiceInstanceLoggingEvent(this, serviceInstanceId));
-		processors.computeIfPresent(serviceInstanceId, (s, envelopeEmitterProcessor) -> null);
+		envelopeSinks.computeIfPresent(serviceInstanceId, (s, envelopeSink) -> null);
 	}
 
 	private String getServiceInstanceId(WebSocketSession webSocketSession) {