From 9dd9f08181a99cf11549cc1d21158f17bd7f2716 Mon Sep 17 00:00:00 2001 From: abilan Date: Fri, 13 Jan 2023 16:00:19 -0500 Subject: [PATCH] Use switch expression; some other clean up --- .../amqp/inbound/AmqpMessageSource.java | 18 ++--- .../channel/FluxMessageChannel.java | 26 +++----- .../integration/handler/LoggingHandler.java | 35 ++++------ .../AbstractRemoteFileOutboundGateway.java | 30 ++++----- .../session/AbstractFtpSessionFactory.java | 14 ++-- .../HazelcastDistributedSQLMessageSource.java | 32 ++++----- .../ip/tcp/connection/TcpNioConnection.java | 16 ++--- .../tcp/connection/TcpNioSSLConnection.java | 66 +++++++------------ .../ByteArrayLengthHeaderSerializer.java | 33 ++++------ .../ip/tcp/serializer/TcpCodecs.java | 18 ++--- .../ip/udp/DatagramPacketMessageMapper.java | 17 ++--- .../r2dbc/outbound/R2dbcMessageHandler.java | 21 +++--- .../redis/util/RedisLockRegistry.java | 14 ++-- 13 files changed, 122 insertions(+), 218 deletions(-) diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java index f84bb575c0..74000f35af 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2022 the original author or authors. + * Copyright 2018-2023 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. @@ -276,17 +276,11 @@ public class AmqpMessageSource extends AbstractMessageSource { try { long deliveryTag = this.ackInfo.getGetResponse().getEnvelope().getDeliveryTag(); switch (status) { - case ACCEPT: - this.ackInfo.getChannel().basicAck(deliveryTag, false); - break; - case REJECT: - this.ackInfo.getChannel().basicReject(deliveryTag, false); - break; - case REQUEUE: - this.ackInfo.getChannel().basicReject(deliveryTag, true); - break; - default: - break; + case ACCEPT -> this.ackInfo.getChannel().basicAck(deliveryTag, false); + case REJECT -> this.ackInfo.getChannel().basicReject(deliveryTag, false); + case REQUEUE -> this.ackInfo.getChannel().basicReject(deliveryTag, true); + default -> { + } } if (this.ackInfo.isTransacted()) { this.ackInfo.getChannel().txCommit(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java index 16316be23f..4bb527d47d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2022 the original author or authors. + * Copyright 2015-2023 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. @@ -78,21 +78,15 @@ public class FluxMessageChannel extends AbstractMessageChannel } private boolean tryEmitMessage(Message message) { - switch (this.sink.tryEmitNext(message)) { - case OK: - return true; - case FAIL_NON_SERIALIZED: - case FAIL_OVERFLOW: - return false; - case FAIL_ZERO_SUBSCRIBER: - throw new IllegalStateException("The [" + this + "] doesn't have subscribers to accept messages"); - case FAIL_TERMINATED: - case FAIL_CANCELLED: - throw new IllegalStateException("Cannot emit messages into the cancelled or terminated sink: " - + this.sink); - default: - throw new UnsupportedOperationException(); - } + return switch (this.sink.tryEmitNext(message)) { + case OK -> true; + case FAIL_NON_SERIALIZED, FAIL_OVERFLOW -> false; + case FAIL_ZERO_SUBSCRIBER -> + throw new IllegalStateException("The [" + this + "] doesn't have subscribers to accept messages"); + case FAIL_TERMINATED, FAIL_CANCELLED -> + throw new IllegalStateException("Cannot emit messages into the cancelled or terminated sink: " + + this.sink); + }; } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/LoggingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/LoggingHandler.java index 7b560c11fa..c3bac33ea6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/LoggingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/LoggingHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2023 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. @@ -151,7 +151,7 @@ public class LoggingHandler extends AbstractMessageHandler { /** * Specify whether to log the full Message. Otherwise, only the payload will be logged. This value is - * false by default. + * {@code false} by default. * @param shouldLogFullMessage true if the complete message should be logged. */ public void setShouldLogFullMessage(boolean shouldLogFullMessage) { @@ -178,26 +178,13 @@ public class LoggingHandler extends AbstractMessageHandler { protected void handleMessageInternal(Message message) { Supplier logMessage = () -> createLogMessage(message); switch (this.level) { - case FATAL: - this.messageLogger.fatal(logMessage); - break; - case ERROR: - this.messageLogger.error(logMessage); - break; - case WARN: - this.messageLogger.warn(logMessage); - break; - case INFO: - this.messageLogger.info(logMessage); - break; - case DEBUG: - this.messageLogger.debug(logMessage); - break; - case TRACE: - this.messageLogger.trace(logMessage); - break; - default: - throw new IllegalStateException("Level '" + this.level + "' is not supported"); + case FATAL -> this.messageLogger.fatal(logMessage); + case ERROR -> this.messageLogger.error(logMessage); + case WARN -> this.messageLogger.warn(logMessage); + case INFO -> this.messageLogger.info(logMessage); + case DEBUG -> this.messageLogger.debug(logMessage); + case TRACE -> this.messageLogger.trace(logMessage); + default -> throw new IllegalStateException("Level '" + this.level + "' is not supported"); } } @@ -209,7 +196,7 @@ public class LoggingHandler extends AbstractMessageHandler { : Objects.toString(logMessage); } - private String createLogMessage(Throwable throwable) { + private static String createLogMessage(Throwable throwable) { StringWriter stringWriter = new StringWriter(); if (throwable instanceof AggregateMessageDeliveryException) { stringWriter.append(throwable.getMessage()); @@ -223,7 +210,7 @@ public class LoggingHandler extends AbstractMessageHandler { return stringWriter.toString(); } - private void printStackTrace(Throwable throwable, Writer writer) { + private static void printStackTrace(Throwable throwable, Writer writer) { throwable.printStackTrace(new PrintWriter(writer, true)); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java index df5e68a2dc..600f228787 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2023 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. @@ -579,24 +579,16 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply @Override protected Object handleRequestMessage(final Message requestMessage) { if (this.command != null) { - switch (this.command) { - case LS: - return doLs(requestMessage); - case NLST: - return doNlst(requestMessage); - case GET: - return doGet(requestMessage); - case MGET: - return doMget(requestMessage); - case RM: - return doRm(requestMessage); - case MV: - return doMv(requestMessage); - case PUT: - return doPut(requestMessage); - case MPUT: - return doMput(requestMessage); - } + return switch (this.command) { + case LS -> doLs(requestMessage); + case NLST -> doNlst(requestMessage); + case GET -> doGet(requestMessage); + case MGET -> doMget(requestMessage); + case RM -> doRm(requestMessage); + case MV -> doMv(requestMessage); + case PUT -> doPut(requestMessage); + case MPUT -> doMput(requestMessage); + }; } return this.remoteFileTemplate.execute(session -> AbstractRemoteFileOutboundGateway.this.messageSessionCallback.doInSession(session, requestMessage)); diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/AbstractFtpSessionFactory.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/AbstractFtpSessionFactory.java index cd355d9c15..60b17a72c1 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/AbstractFtpSessionFactory.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/AbstractFtpSessionFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2021 the original author or authors. + * Copyright 2002-2023 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. @@ -216,18 +216,12 @@ public abstract class AbstractFtpSessionFactory implements } /** - * Sets the mode of the connection. Only local modes are supported. + * Set the mode of the connection. Only local modes are supported. */ private void updateClientMode(FTPClient client) { switch (this.clientMode) { - case FTPClient.ACTIVE_LOCAL_DATA_CONNECTION_MODE: - client.enterLocalActiveMode(); - break; - case FTPClient.PASSIVE_LOCAL_DATA_CONNECTION_MODE: - client.enterLocalPassiveMode(); - break; - default: - break; + case FTPClient.ACTIVE_LOCAL_DATA_CONNECTION_MODE -> client.enterLocalActiveMode(); + case FTPClient.PASSIVE_LOCAL_DATA_CONNECTION_MODE -> client.enterLocalPassiveMode(); } } diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastDistributedSQLMessageSource.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastDistributedSQLMessageSource.java index bb25ad18d7..197a68718d 100644 --- a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastDistributedSQLMessageSource.java +++ b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastDistributedSQLMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2022 the original author or authors. + * Copyright 2015-2023 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. @@ -32,6 +32,8 @@ import org.springframework.util.CollectionUtils; * distributed query in the cluster and returns results in the light of iteration type. * * @author Eren Avsarogullari + * @author Artem Bilan + * * @since 6.0 */ @SuppressWarnings("rawtypes") @@ -63,30 +65,20 @@ public class HazelcastDistributedSQLMessageSource extends AbstractMessageSource @Override @SuppressWarnings("unchecked") protected Collection doReceive() { - switch (this.iterationType) { - case ENTRY: - return getDistributedSQLResultSet(Collections - .unmodifiableCollection(this.distributedMap.entrySet(new SqlPredicate(this.distributedSql)))); + final SqlPredicate predicate = new SqlPredicate(this.distributedSql); + Collection collection = + switch (this.iterationType) { + case ENTRY -> this.distributedMap.entrySet(predicate); + case KEY -> this.distributedMap.keySet(predicate); + case LOCAL_KEY -> this.distributedMap.localKeySet(predicate); + default -> this.distributedMap.values(predicate); + }; - case KEY: - return getDistributedSQLResultSet(Collections - .unmodifiableCollection(this.distributedMap.keySet(new SqlPredicate(this.distributedSql)))); - - case LOCAL_KEY: - return getDistributedSQLResultSet(Collections - .unmodifiableCollection(this.distributedMap.localKeySet(new SqlPredicate(this.distributedSql)))); - - default: - return getDistributedSQLResultSet(this.distributedMap.values(new SqlPredicate(this.distributedSql))); - } - } - - private Collection getDistributedSQLResultSet(Collection collection) { if (CollectionUtils.isEmpty(collection)) { return null; } - return collection; + return Collections.unmodifiableCollection(collection); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java index 55a0022989..fa7c2df42c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2023 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. @@ -22,7 +22,6 @@ import java.io.InputStream; import java.io.OutputStream; import java.io.UncheckedIOException; import java.net.SocketTimeoutException; -import java.nio.Buffer; import java.nio.ByteBuffer; import java.nio.channels.ClosedChannelException; import java.nio.channels.SelectionKey; @@ -224,14 +223,7 @@ public class TcpNioConnection extends TcpConnectionSupport { * @return The buffer. */ protected ByteBuffer allocate(int length) { - ByteBuffer buffer; - if (this.usingDirectBuffers) { - buffer = ByteBuffer.allocateDirect(length); - } - else { - buffer = ByteBuffer.allocate(length); - } - return buffer; + return this.usingDirectBuffers ? ByteBuffer.allocateDirect(length) : ByteBuffer.allocate(length); } /** @@ -447,7 +439,7 @@ public class TcpNioConnection extends TcpConnectionSupport { if (logger.isTraceEnabled()) { logger.trace("After read: " + this.rawBuffer.position() + '/' + this.rawBuffer.limit()); } - ((Buffer) this.rawBuffer).flip(); + this.rawBuffer.flip(); if (logger.isTraceEnabled()) { logger.trace("After flip: " + this.rawBuffer.position() + '/' + this.rawBuffer.limit()); } @@ -472,7 +464,7 @@ public class TcpNioConnection extends TcpConnectionSupport { logger.trace(getConnectionId() + " Sending " + rawBufferToSend.limit() + " to pipe"); } this.channelInputStream.write(rawBufferToSend); - ((Buffer) rawBufferToSend).clear(); + rawBufferToSend.clear(); } private void checkForAssembler() { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java index ac927add90..5c1c61338e 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2023 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. @@ -17,7 +17,6 @@ package org.springframework.integration.ip.tcp.connection; import java.io.IOException; -import java.nio.Buffer; import java.nio.ByteBuffer; import java.nio.channels.SocketChannel; import java.util.concurrent.Semaphore; @@ -27,7 +26,6 @@ import javax.net.ssl.SSLEngine; import javax.net.ssl.SSLEngineResult; import javax.net.ssl.SSLEngineResult.HandshakeStatus; import javax.net.ssl.SSLEngineResult.Status; -import javax.net.ssl.SSLException; import javax.net.ssl.SSLHandshakeException; import javax.net.ssl.SSLSession; @@ -137,7 +135,7 @@ public class TcpNioSSLConnection extends TcpNioConnection { networkBuffer.compact(); } else { - ((Buffer) networkBuffer).clear(); + networkBuffer.clear(); } if (logger.isDebugEnabled()) { logger.debug("sendToPipe.x " + resultToString(result) + ", remaining: " + networkBuffer.remaining()); @@ -145,7 +143,7 @@ public class TcpNioSSLConnection extends TcpNioConnection { } /** - * Performs the actual decryption of a received packet - which may be real + * Perform the actual decryption of a received packet - which may be real * data, or handshaking data. Appropriate action is taken with the data. * If this side did not initiate the handshake, any handshaking data sent out * is handled by the thread running in the {@link SSLChannelOutputStream#doWrite(ByteBuffer)} @@ -156,19 +154,11 @@ public class TcpNioSSLConnection extends TcpNioConnection { HandshakeStatus handshakeStatus = this.sslEngine.getHandshakeStatus(); SSLEngineResult result = new SSLEngineResult(Status.OK, handshakeStatus, 0, 0); switch (handshakeStatus) { - case NEED_TASK: - runTasks(); - break; - case NEED_UNWRAP: - case FINISHED: - case NOT_HANDSHAKING: - result = checkBytesProduced(networkBuffer); - break; - case NEED_WRAP: - result = needWrap(networkBuffer, result); - break; - default: + case NEED_TASK -> runTasks(); + case NEED_UNWRAP, FINISHED, NOT_HANDSHAKING -> result = checkBytesProduced(networkBuffer); + case NEED_WRAP -> result = needWrap(networkBuffer, result); } + switch (result.getHandshakeStatus()) { case FINISHED: resumeWriterIfNeeded(); //NOSONAR - fall-through intended @@ -183,31 +173,28 @@ public class TcpNioSSLConnection extends TcpNioConnection { return result; } - private SSLEngineResult checkBytesProduced(ByteBuffer networkBuffer) throws SSLException, IOException { + private SSLEngineResult checkBytesProduced(ByteBuffer networkBuffer) throws IOException { SSLEngineResult result; - ((Buffer) this.decoded).clear(); + this.decoded.clear(); result = this.sslEngine.unwrap(networkBuffer, this.decoded); if (logger.isDebugEnabled()) { logger.debug("After unwrap: " + resultToString(result)); } Status status = result.getStatus(); if (status == Status.BUFFER_OVERFLOW) { - this.decoded = - this.allocateEncryptionBuffer(this.sslEngine.getSession().getApplicationBufferSize()); + this.decoded = this.allocateEncryptionBuffer(this.sslEngine.getSession().getApplicationBufferSize()); } if (result.bytesProduced() > 0) { - ((Buffer) this.decoded).flip(); + this.decoded.flip(); super.sendToPipe(this.decoded); } return result; } - private SSLEngineResult needWrap(ByteBuffer networkBuffer, SSLEngineResult result) - throws SSLException, IOException { - + private SSLEngineResult needWrap(ByteBuffer networkBuffer, SSLEngineResult result) throws IOException { SSLEngineResult engineResult = result; if (!resumeWriterIfNeeded()) { - ((Buffer) this.encoded).clear(); + this.encoded.clear(); engineResult = this.sslEngine.wrap(networkBuffer, this.encoded); if (logger.isDebugEnabled()) { logger.debug("After wrap: " + resultToString(engineResult)); @@ -216,7 +203,7 @@ public class TcpNioSSLConnection extends TcpNioConnection { this.encoded = this.allocateEncryptionBuffer(this.sslEngine.getSession().getPacketBufferSize()); } else { - ((Buffer) this.encoded).flip(); + this.encoded.flip(); getSSLChannelOutputStream().writeEncoded(this.encoded); } } @@ -302,12 +289,9 @@ public class TcpNioSSLConnection extends TcpNioConnection { } protected SSLChannelOutputStream getSSLChannelOutputStream() { - if (this.sslChannelOutputStream == null) { - return (SSLChannelOutputStream) getChannelOutputStream(); - } - else { - return this.sslChannelOutputStream; - } + return this.sslChannelOutputStream != null + ? this.sslChannelOutputStream + : (SSLChannelOutputStream) getChannelOutputStream(); } private String resultToString(SSLEngineResult result) { @@ -336,14 +320,13 @@ public class TcpNioSSLConnection extends TcpNioConnection { } /** - * Encrypts the plaintText buffer and writes it to the SocketChannel. + * Encrypt the plaintText buffer and writes it to the SocketChannel. * Will participate in SSL handshaking as necessary. For very large * data, the SSL packets will be limited by the engine's buffer sizes * and multiple writes will be necessary. */ @Override - protected synchronized void doWrite(ByteBuffer plainText) - throws IOException { + protected synchronized void doWrite(ByteBuffer plainText) throws IOException { try { TcpNioSSLConnection.this.writerActive = true; int remaining = plainText.remaining(); @@ -373,7 +356,7 @@ public class TcpNioSSLConnection extends TcpNioConnection { } /** - * Handles SSL handshaking; when network data is needed from the peer, suspends + * Handle SSL handshaking; when network data is needed from the peer, suspends * until that data is received. */ private void doClientSideHandshake(ByteBuffer plainText, SSLEngineResult resultArg) throws IOException { @@ -409,16 +392,15 @@ public class TcpNioSSLConnection extends TcpNioConnection { } private void writeEncodedIfAny() throws IOException { - ((Buffer) TcpNioSSLConnection.this.encoded).flip(); + TcpNioSSLConnection.this.encoded.flip(); writeEncoded(TcpNioSSLConnection.this.encoded); - ((Buffer) TcpNioSSLConnection.this.encoded).clear(); + TcpNioSSLConnection.this.encoded.clear(); } /** * Suspend processing until data is received from the peer. */ private HandshakeStatus waitForHandshakeData(SSLEngineResult result) throws IOException { - try { logger.trace("Writer waiting for handshake"); if (!TcpNioSSLConnection.this.semaphore.tryAcquire(TcpNioSSLConnection.this.handshakeTimeout, @@ -443,10 +425,10 @@ public class TcpNioSSLConnection extends TcpNioConnection { } /** - * Encrypts plain text data. The result may indicate handshaking is needed. + * Encrypt plain text data. The result may indicate handshaking is needed. */ private SSLEngineResult encode(ByteBuffer plainText) throws IOException { - ((Buffer) TcpNioSSLConnection.this.encoded).clear(); + TcpNioSSLConnection.this.encoded.clear(); SSLEngineResult result = TcpNioSSLConnection.this.sslEngine.wrap(plainText, TcpNioSSLConnection.this.encoded); if (logger.isDebugEnabled()) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java index ef5ece45c3..941f14e4de 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2021 the original author or authors. + * Copyright 2002-2023 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. @@ -207,27 +207,24 @@ public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer protected void writeHeader(OutputStream outputStream, int length) throws IOException { ByteBuffer lengthPart = ByteBuffer.allocate(this.headerSize); switch (this.headerSize) { - case HEADER_SIZE_INT: - lengthPart.putInt(length); - break; - case HEADER_SIZE_UNSIGNED_BYTE: + case HEADER_SIZE_INT -> lengthPart.putInt(length); + case HEADER_SIZE_UNSIGNED_BYTE -> { if (length > MAX_UNSIGNED_BYTE) { throw new IllegalArgumentException("Length header: " + this.headerSize + " too short to accommodate message length: " + length); } lengthPart.put((byte) length); - break; - case HEADER_SIZE_UNSIGNED_SHORT: + } + case HEADER_SIZE_UNSIGNED_SHORT -> { if (length > MAX_UNSIGNED_SHORT) { throw new IllegalArgumentException("Length header: " + this.headerSize + " too short to accommodate message length: " + length); } lengthPart.putShort((short) length); - break; - default: - throw new IllegalArgumentException("Bad header size: " + this.headerSize); + } + default -> throw new IllegalArgumentException("Bad header size: " + this.headerSize); } outputStream.write(lengthPart.array()); } @@ -249,22 +246,18 @@ public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer } int messageLength; switch (this.headerSize) { - case HEADER_SIZE_INT: + case HEADER_SIZE_INT -> { messageLength = ByteBuffer.wrap(lengthPart).getInt(); if (messageLength < 0) { throw new IllegalArgumentException("Length header: " + messageLength + " is negative"); } - break; - case HEADER_SIZE_UNSIGNED_BYTE: - messageLength = ByteBuffer.wrap(lengthPart).get() & MAX_UNSIGNED_BYTE; - break; - case HEADER_SIZE_UNSIGNED_SHORT: - messageLength = ByteBuffer.wrap(lengthPart).getShort() & MAX_UNSIGNED_SHORT; - break; - default: - throw new IllegalArgumentException("Bad header size: " + this.headerSize); + } + case HEADER_SIZE_UNSIGNED_BYTE -> messageLength = ByteBuffer.wrap(lengthPart).get() & MAX_UNSIGNED_BYTE; + case HEADER_SIZE_UNSIGNED_SHORT -> + messageLength = ByteBuffer.wrap(lengthPart).getShort() & MAX_UNSIGNED_SHORT; + default -> throw new IllegalArgumentException("Bad header size: " + this.headerSize); } return messageLength; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/TcpCodecs.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/TcpCodecs.java index f957420582..4743014660 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/TcpCodecs.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/TcpCodecs.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2022 the original author or authors. + * Copyright 2016-2023 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. @@ -136,16 +136,12 @@ public final class TcpCodecs { * @see AbstractByteArraySerializer#DEFAULT_MAX_MESSAGE_SIZE */ public static ByteArrayLengthHeaderSerializer lengthHeader(int bytes) { - switch (bytes) { - case ByteArrayLengthHeaderSerializer.HEADER_SIZE_UNSIGNED_BYTE: - return lengthHeader1(); - case ByteArrayLengthHeaderSerializer.HEADER_SIZE_UNSIGNED_SHORT: - return lengthHeader2(); - case ByteArrayLengthHeaderSerializer.HEADER_SIZE_INT: - return lengthHeader4(); - default: - throw new IllegalArgumentException("Only 1, 2 or 4 byte headers are supported"); - } + return switch (bytes) { + case ByteArrayLengthHeaderSerializer.HEADER_SIZE_UNSIGNED_BYTE -> lengthHeader1(); + case ByteArrayLengthHeaderSerializer.HEADER_SIZE_UNSIGNED_SHORT -> lengthHeader2(); + case ByteArrayLengthHeaderSerializer.HEADER_SIZE_INT -> lengthHeader4(); + default -> throw new IllegalArgumentException("Only 1, 2 or 4 byte headers are supported"); + }; } /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java index 05dcdc8ca2..c9fc6f0b06 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2023 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. @@ -19,7 +19,6 @@ package org.springframework.integration.ip.udp; import java.io.UncheckedIOException; import java.io.UnsupportedEncodingException; import java.net.DatagramPacket; -import java.nio.Buffer; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.util.Map; @@ -52,10 +51,10 @@ import org.springframework.util.StringUtils; * may be either a byte array or a String. The default charset for converting * a String to a byte array is UTF-8, but that may be changed by invoking the * {@link #setCharset(String)} method. - * + *

* By default, the UDP messages will be unreliable (truncation may occur on * the receiving end; packets may be lost). - * + *

* Reliability can be enhanced by one or both of the following techniques: *

    *
  • including a binary message length at the beginning of the packet
    • @@ -192,14 +191,13 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper message) { - byte[] bytes = null; Object payload = message.getPayload(); if (payload instanceof byte[]) { - bytes = (byte[]) payload; + return (byte[]) payload; } else if (payload instanceof String) { try { - bytes = ((String) payload).getBytes(this.charset); + return ((String) payload).getBytes(this.charset); } catch (UnsupportedEncodingException e) { throw new UncheckedIOException(e); @@ -209,7 +207,6 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper handleMessageInternal(Message message) { Assert.isTrue(this.initialized, "The instance is not yet initialized. Invoke its afterPropertiesSet() method"); return Mono.fromSupplier(() -> this.queryTypeExpression.getValue(this.evaluationContext, message, Type.class)) - .flatMap(mode -> { - switch (mode) { - case INSERT: - return handleInsert(message); - case UPDATE: - return handleUpdate(message); - case DELETE: - return handleDelete(message); - default: - return Mono.error(new IllegalArgumentException()); - } - }).then(); + .flatMap(mode -> + switch (mode) { + case INSERT -> handleInsert(message); + case UPDATE -> handleUpdate(message); + case DELETE -> handleDelete(message); + }) + .then(); } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java index f529a94c55..96c6a39958 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2022 the original author or authors. + * Copyright 2014-2023 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. @@ -274,14 +274,10 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl } private Function getRedisLockConstructor(RedisLockType redisLockType) { - switch (redisLockType) { - case SPIN_LOCK: - return RedisSpinLock::new; - case PUB_SUB_LOCK: - return RedisPubSubLock::new; - default: - throw new IllegalArgumentException(); - } + return switch (redisLockType) { + case SPIN_LOCK -> RedisSpinLock::new; + case PUB_SUB_LOCK -> RedisPubSubLock::new; + }; } private abstract class RedisLock implements Lock {