Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use switch expression; some other clean up

Browse Source
This commit is contained in:
abilan
2023-01-13 16:00:19 -05:00
parent fc3b95ea79
commit 9dd9f08181
13 changed files with 122 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java
View File

@@ -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<Object> {
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();

26
spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java
View File

@@ -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

35
spring-integration-core/src/main/java/org/springframework/integration/handler/LoggingHandler.java
View File

@@ -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
* <code>false</code> 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<CharSequence> 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));
}

30
spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java
View File

@@ -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<F> 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));

14
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/AbstractFtpSessionFactory.java
View File

@@ -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<T extends FTPClient> 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();
}
}

32
spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastDistributedSQLMessageSource.java
View File

@@ -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);
}
}

16
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
View File

@@ -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() {

66
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java
View File

@@ -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()) {

33
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java
View File

@@ -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;
}

18
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/TcpCodecs.java
View File

@@ -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");
};
}
/**

17
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java
View File

@@ -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.
*
* <p>
* By default, the UDP messages will be unreliable (truncation may occur on
* the receiving end; packets may be lost).
*
* <p>
* Reliability can be enhanced by one or both of the following techniques:
* <ul>
* <li>including a binary message length at the beginning of the packet</li>
@@ -192,14 +191,13 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
}
private byte[] getPayloadAsBytes(Message<?> 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<Datagra
throw new IllegalArgumentException("The datagram packet mapper expects " +
"either a byte array or String payload, but received: " + payload.getClass());
}
return bytes;
}
@Override
@@ -283,7 +280,7 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
}
/**
* Peeks at data in the buffer to see if starts with the prefix.
* Peek at data in the buffer to see if starts with the prefix.
*/
private boolean startsWith(ByteBuffer buffer, String prefix) {
int pos = buffer.position();
@@ -305,7 +302,7 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
}
finally {
//reposition the buffer
((Buffer) buffer).position(pos);
buffer.position(pos);
}
}

21
spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2022 the original author or authors.
* Copyright 2020-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.
@@ -158,18 +158,13 @@ public class R2dbcMessageHandler extends AbstractReactiveMessageHandler {
protected Mono<Void> 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();
}

14
spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java
View File

@@ -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<String, RedisLock> 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 {